http客户端-基于boost开发
基于BOOST编写的http客户端，作为BOOST开发学习之用。目前支持功能：
- http协议，单向链接返回http response code 200
- 可content type 为text或image下载到本地
- 仅支持http返回code为200，不支持3XX、4XX等
- 暂不支持chunk传输，chunk代码待调试
- 日志文件，提供类ACE的输出方式，如LOG((LOG_DEBUG,”[%T(%t)] test begin %d %s\n”), int_num_2, string_test.c_str());
- 数据缓冲区，当前为new方式，后续可更改为从boost pool获取
- 。。。
1 类关系图
2 核心代码
2.1 IOServer，提供asio线程环境
#pragma once
#include "boost/serialization/singleton.hpp"
#include "boost/asio.hpp"
#include "boost/thread.hpp"
#include "boost/bind.hpp"
#include "Event.h"
#include "../concurrent/IRunable.h"
#include "SimpleLog.h"
class IO_Server : public IRunable
{
public:
IO_Server(): bExit(false)
{
}
boost::asio::io_service* GetIOS()
{
return &ios;
}
void ActiveIOS()
{
boost::unique_lock<boost::mutex> lock(mu);
if (ios.stopped())
{
cond.notify_one();
}
}
void Exit()
{
ios.stop();
bExit = true;
}
private:
virtual int Run()
{
LOG((LOG_DEBUG,"[%T(%t)] ios server run ,id = %d\n", boost::this_thread::get_id()));
ios.stop();
while ()
{
// 设置退出线程
if (bExit)
{
break;
}
//
{
boost::unique_lock<boost::mutex> lock(mu);
if (ios.stopped())
{
cond.wait(mu);
}
}
if (ios.stopped())
{
ios.reset();
ios.run();
}
}
return ;
}
private:
boost::asio::io_service ios; //所有的asio都要有个ios
boost::mutex mu;
boost::condition_variable_any cond;
bool bExit;
};
typedef boost::serialization::singleton<IO_Server> IOS;
2.2 task及sock资源管理器
#pragma once
#include "ios.h"
#include "handle.h"
#include "ios.h"
#include "SimpleLog.h"
#include "../concurrent/ThreadPool.h"
#include "../concurrent/IRunable.h"
#include "MsgQueue.h"
/************************************************************************/
/* Handle的集合.多线程共享 */
/************************************************************************/
typedef boost::shared_ptr<IRunable> IRunablePtr;
typedef CMsgQueue<IRunablePtr> IRunPtrQueue;
template<class CLIENT>
class HandleSet : public IRunable
{
public:
HandleSet()
{
bExit = false;
nHighWaterMark = ;
nClientUsed = ; m_poThreadPool = boost::shared_ptr<ThreadPool>(new ThreadPool("HandleSet_Pool", , boost::thread::hardware_concurrency()));
//启动ioserver线程
IO_Server& ioserver = IOS::get_mutable_instance();
m_poThreadPool->Start(&ioserver);
//启动
m_poThreadPool->Start(this);
}
~HandleSet(){}
//加入一个task
void AddTask(TaskPtr task)
{
boost::unique_lock<boost::mutex> lock(mu);
ClientHandlePtr client;
if (!GetFreeClient(client))
{
tasks.push_back(task);
return;
}
client->Busy(true);
client->HTTP_Connect(task);
InterlockedIncrement(&nClientUsed);
IOS::get_mutable_instance().ActiveIOS();
} //设置client水位标
void HighWaterMark(int mark)
{
nHighWaterMark = mark;
} void NotifyComplete(ClientHandle* client, const boost::system::error_code& ec,
CMsgBlock& msg)
{
switch (ec.value())
{
case : //正常退出
{
boost::unique_lock<boost::mutex> lock(mu);
client->pTask->Finish(SUCCESS, msg);
client->Busy(false);
client->Reset();
InterlockedDecrement(&nClientUsed);
}
break;
case : //远端关闭连接
{ }
break;
}
}
private:
//找出一个空闲的client
bool GetFreeClient(ClientHandlePtr& ptr)
{
if (clients.size() >= nHighWaterMark && nClientUsed >= nHighWaterMark)
{
return false;
}
//空队列，创建一个新的
if (clients.empty())
{
ptr = CreateNewHandle();
clients.push_back(ptr);
return true;
}
//非空则找出一个空闲
LiClientHandlePtr::iterator iter = clients.begin();
for ( ;iter != clients.end(); iter++)
{
ClientHandlePtr client = *iter;
if (!client->Busy())
{
ptr = client;
return true;
}
}
//无空闲且饱和
if (nClientUsed < nHighWaterMark)
{
ptr = CreateNewHandle();
clients.push_back(ptr);
return true;
}
return false;
}
ClientHandlePtr CreateNewHandle()
{
return ClientHandlePtr(
new CLIENT(IOS::get_mutable_instance().GetIOS(), boost::bind(&HandleSet::NotifyComplete, this, _1, _2, _3)));
}
private:
virtual int Run()
{
LOG((LOG_DEBUG,"[%T(%t)] handleSet %s thread run\n", typeid(this).name()));
while ()
{
// 设置退出线程
if (bExit)
{
break;
}
{
boost::unique_lock<boost::mutex> lock(mu);
if (!tasks.empty())
{
ClientHandlePtr client;
if (GetFreeClient(client))
{
TaskPtr task = tasks.front();
tasks.pop_front();
client->Busy(true);
client->HTTP_Connect(task);
InterlockedIncrement(&nClientUsed);
}
IOS::get_mutable_instance().ActiveIOS();
}
}
boost::this_thread::sleep_for(boost::chrono::milliseconds()); //线程池模式
/*if (queue.empty())
{
boost::this_thread::sleep_for(boost::chrono::milliseconds(20));
continue;
} IRunablePtr ptrTask;
queue.dequeue_head(ptrTask, 10); if (ptrTask)
{
m_poThreadPool->Start(ptrTask);
}*/
}
return ;
}
private:
int nHighWaterMark; //并发执行数量为1--nHighWaterMark
LiClientHandlePtr clients; TaskPtrList tasks;
long nClientUsed;
bool bExit;
boost::mutex mu; boost::shared_ptr<ThreadPool> m_poThreadPool; ///<线程池共享指针
IRunPtrQueue queue;
};
2.3 socket工厂模式虚接口，供资源管理器调用
#pragma once
/************************************************************************/
/* 提供网络行为 */
/************************************************************************/
#include <string>
#include "boost/smart_ptr.hpp"
#include "boost/asio.hpp"
#include "boost/date_time/posix_time/posix_time.hpp"
#include "boost/thread.hpp"
#include "MsgBlock.h"
#include "SimpleLog.h"
#include "userTask.h"
using namespace boost;
using namespace boost::asio;
using namespace std; class ClientHandle;
typedef boost::shared_ptr<ip::tcp::socket> sock_ptr;
typedef boost::shared_ptr<ClientHandle> ClientHandlePtr;
typedef std::list<ClientHandlePtr> LiClientHandlePtr; typedef boost::function<void(ClientHandle* handle,
const boost::system::error_code& ec, CMsgBlock&)> NotifyCompleteFunc; class ClientHandle
{
public:
ClientHandle(boost::asio::io_service* ios, NotifyCompleteFunc completeFun);
~ClientHandle();
public: //主动请求
void HTTP_Connect(TaskPtr task);
//http send请求
void HTTP_Send(CMsgBlock& block);
//http close
void HTTP_Close();
//http recv，接收到数据后回调
void HTTP_Recv();
bool Busy();
void Busy(bool busy);
void Reset();
public: //回调函数
//http send，接收到数据后回调
virtual void CB_Send(const boost::system::error_code& ec, size_t len) = ;
//http recv，接收到数据后回调
virtual void CB_Recv(const boost::system::error_code& ec, size_t len) = ;
//http connect result,连接结果回调
virtual void CB_HTTP_Connect_Result(const boost::system::error_code& ec) = ;
//http result, 返回错误码及第一次数据
virtual int CB_HTTP_CODE(int retCode, CMsgBlock& block) = ;
//连接超时
virtual void CB_HTTP_Timeout() = ;
//对端关闭连接
virtual void CB_Remote_Close() = ;
public:
TaskPtr pTask;
protected:
//写缓冲区和写指针
CMsgBlock _rdBuf;
CMsgBlock _wrBuf;
DWORD lastDataTime; //活动时间
long recvSz; //累计接收数据量
//记录任务
sock_ptr sock;
boost::asio::ip::tcp::endpoint ep;
bool bBusy;
NotifyCompleteFunc _completeFun;
};
2.4 socket管理实例
//http recv，接收到数据后回调
void ITsoftware_Index::CB_Send(const boost::system::error_code& ec, size_t len)
{
if (ec)
{
//错误,调用返回
return;
}
if (len == _wrBuf.Size())
{
_rdBuf.Reset();
sock->async_read_some(boost::asio::buffer(_rdBuf.Base() + _rdBuf.WtPtr(), _rdBuf.Space()),
bind(&ClientHandle::CB_Recv, this, _1, _2));
}
else
{
sock->async_write_some(boost::asio::buffer(_wrBuf.Base() + len, _wrBuf.Size() - len),
bind(&ClientHandle::CB_Send, this, _1, _2));
}
} //http recv，接收到数据后回调
void ITsoftware_Index::CB_Recv(const boost::system::error_code& ec, size_t len)
{
if (ec || !pTask)
{
return;
}
recvSz += len;
_rdBuf.WtPtr(len); bool bLastData = (_rdBuf.Space() > )? true : false; //缓冲区未接收满，说明当前数据已经收完 int ret = ;
if (!pTask->_head.bReady)
{
ret = ParseHead(_rdBuf);
if (ret == ) //头未收完，继续接收
{
sock->async_read_some(boost::asio::buffer(_rdBuf.Base() + _rdBuf.WtPtr(), _rdBuf.Space()),
bind(&ClientHandle::CB_Recv, this, _1, _2));
return;
}
// 返回为http的代码
ret = CB_HTTP_CODE(ret, _rdBuf);
} switch (ret)
{
case -:
{
sock->shutdown(ip::tcp::socket::shutdown_both);
sock->close();
_completeFun(this, ec, parseOnlineBlock);
parseOnlineBlock.Reset();
}
break;
case :
{
int retWriteData = ;
if (_rdBuf.Size() > )
{
retWriteData = WriteData(_rdBuf); //处理数据流
}
_rdBuf.Reset();
if (retWriteData == -)
{
sock->shutdown(ip::tcp::socket::shutdown_both);
sock->close();
_completeFun(this, ec, parseOnlineBlock);
parseOnlineBlock.Reset();
}
else
{
sock->async_read_some(boost::asio::buffer(_rdBuf.Base() + _rdBuf.WtPtr(), _rdBuf.Space()),
bind(&ClientHandle::CB_Recv, this, _1, _2));
}
}
break;
case :
{ }
break;
}
}
//http connect result,连接结果回调
void ITsoftware_Index::CB_HTTP_Connect_Result(const boost::system::error_code& ec)
{
if (ec)
{
//错误,调用返回
return;
}
std::string getContent = "GET " + pTask->_info.URI + " HTTP/1.1\r\n";
getContent += "Host: " + pTask->_info.host + "\r\n";
getContent += "Connection: keep-alive\r\n";
getContent += "User-Agent: Mozilla/5.0 (Windows NT 10.0; WOW64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/61.0.3163.100 Safari/537.36\r\n";
getContent += "Accept-Encoding: gzip, deflate\r\n";
getContent += "Accept-Language: zh-CN,zh;q=0.8\r\n\r\n"; _wrBuf.Reset();
_wrBuf.Write(getContent.c_str(), getContent.length());
HTTP_Send(_wrBuf);
} //http recv,收到全部数据后回调，httpcode=200
int ITsoftware_Index::CB_HTTP_CODE(int retCode, CMsgBlock& block)
{
int ret = ;
//准备文件及缓冲区
if (retCode == )
{
if (pTask->HasFlag(CONTENT_TYPE_DOWNLOAD)) //开启下载
{
if (pTask->_head.nContentType == CONTENT_TYPE_IMAGE ||
pTask->_head.nContentType == CONTENT_TYPE_TEXT)
{
char sss[] = {};
sprintf_s(sss, , "%d", InterlockedIncrement(&filenameprev));
pTask->FilePtr(std::string(sss) + pTask->_info.lastName); //创建文件
if (!pTask->FilePtr()) //文件创建失败，则接收失败
{
ret = -;
//LOG((LOG_DEBUG,"[%T(%t)] create file failed, %s \n", pTask->_url.c_str()));
}
}
}
if (pTask->HasFlag(CONTENT_TYPE_PARSEONLINE)) //调用完成后回传给task
{
if (pTask->_head.bContentLength)
{
parseOnlineBlock.Capacity(pTask->_head.nContentLength);
}
else if (pTask->_head.bChunked)
{
parseOnlineBlock.Capacity(block.Capacity() * );
}
if (parseOnlineBlock.Capacity() == )
{
ret = -;
//LOG((LOG_DEBUG,"[%T(%t)] create parseOnlineBlock buffer failed, %s \n", pTask->_url.c_str()));
}
}
}
//重定向
else if ( < retCode && retCode < )
{
ret = -; //中断当前
}
return ret;
} //连接超时
void ITsoftware_Index::CB_HTTP_Timeout()
{ }
//对端关闭连接
void ITsoftware_Index::CB_Remote_Close()
{ } int ITsoftware_Index::ParseHead(CMsgBlock& _block)
{
int ret = ;
int rnrn = Tools::KMPFind(_block.Base(), _block.Size(), RNRN, RNRN_SIZE);
if (rnrn != -)
{
//头收完全，处理http头
char* cRnrn = new char[rnrn + + ];
ZeroMemory(cRnrn,rnrn+);
memcpy(cRnrn, _block.Base(), rnrn + );
if (pTask->_head.ParseHead(cRnrn ))
{
pTask->_head.bReady = true;
//数据移位，把头去掉，保证缓冲区内都是数据体
int remain = _block.Size() - rnrn -; //剩余长度
char* remain_begin_pos = _block.Base() + rnrn + ;
_block.Reset(); //读写指针置零
_block.Write(remain_begin_pos, remain); //重新写入
ret = pTask->_head.code;
}
else //文件头出错，关闭连接
{
ret = -;
//LOG((LOG_DEBUG,"[%T(%t)] recv http head parse error, %s\n, %s \n", pTask->_url.c_str() ,cRnrn));
}
delete cRnrn;
}
else
{
//没有收到\r\n\r\n，则表明头还没收完全
ret = ;
}
return ret;
} int ITsoftware_Index::WriteData(CMsgBlock& _block)
{
int ret = ;
if(pTask->_head.bContentLength)
{
if (pTask->HasFlag(CONTENT_TYPE_DOWNLOAD) &&
(pTask->_head.nContentType == CONTENT_TYPE_IMAGE || pTask->_head.nContentType == CONTENT_TYPE_TEXT))
{
if (!pTask->FilePtr()->bad())
{
pTask->FilePtr()->write(_block.Base(), _block.Size());
}
}
if (pTask->HasFlag(CONTENT_TYPE_PARSEONLINE))
{
parseOnlineBlock.Write(_block.Base(), _block.Size());
}
if (recvSz >= pTask->_head.nContentLength)
{
if (pTask->HasFlag(CONTENT_TYPE_DOWNLOAD) &&
(pTask->_head.nContentType == CONTENT_TYPE_IMAGE || pTask->_head.nContentType == CONTENT_TYPE_TEXT))
{
pTask->FilePtr()->close();
}
ret = -;
}
}
else if (pTask->_head.bChunked)
{
// //chunk,通过辨识末尾7位是否是HTTP_END，来判断数据是否已接收完整
// if(_block.RdPtr() > HTTP_END_SIZE)
// {
// char cend[HTTP_END_SIZE+1]= {0};
// memcpy(cend, _block.Base() + _block.Size() - 7, HTTP_END_SIZE);
// int cmp = strcmp(cend,HTTP_END);
// if (cmp != 0)
// {
// //未接收完整，继续接收
// return 0;
// }
// }
// //数据接收完全，把chunk的长度字符全部删除
// int beginPos = rnrn + 4;
// while (true)
// {
// int rnBegin = Tools::KMPFind(_block.Base(), _block.Size(), RN, RN_SIZE, beginPos);
// if (rnBegin != -1)
// {
// char* tets = _block.Base() + rnBegin;
// //计算出chunk的长度
// char* cchunk = new char[rnBegin - rnrn - 4 + 1];
// ZeroMemory(cchunk, rnBegin - rnrn - 4 + 1);
// memcpy(cchunk, _block.Base() + rnrn + 4, rnBegin - rnrn - 4);
// int chunk = strtol(cchunk,NULL,16);
// delete cchunk;
// if (chunk ==0)
// break;
// //copy chunck的字节到body
// result.Write(_block.Base() + rnBegin + RN_SIZE, chunk);
// beginPos = rnBegin + chunk+RN_SIZE*2;
// }
// else
// break;
}
return ret;
}
2.5 下发任务
#pragma once
#include <string>
#include "boost/bind.hpp"
#include "boost/function.hpp"
#include "boost/smart_ptr.hpp"
#include "SimpleLog.h"
#include "MsgBlock.h"
#include "httphead.h"
#include "Tools.h"
using namespace std;
class Task; enum NotifyType{
SUCCESS,
FAILED
}; typedef boost::function<void(NotifyType,const std::string&)> NotifyFunc;
typedef boost::shared_ptr<Task> TaskPtr;
typedef std::list<TaskPtr> TaskPtrList; class Task
{
public:
static TaskPtr CreateTask(std::string url, long flag = CONTENT_TYPE_NONE, std::string downPath = "")
{
return boost::shared_ptr<Task>(new Task(url, flag, downPath));
}
public:
virtual ~Task()
{
if (_file)
{
_file->close();
}
}
virtual void Finish(NotifyType type, CMsgBlock& retMsg)
{
LOG((LOG_DEBUG, "[%T(%t)] NotifyComplete %d, %s \n", type, _url.c_str()));
// if (type == SUCCESS)
// {
// std::vector<std::string> vecs;
// Tools::HttpStringOpt::ExtractAllUrl(std::string(result.Base(), result.Size()), vecs);
// }
}
inline ofstreamPtr FilePtr()
{
return _file;
}
inline void FilePtr(std::string filename)
{
if (_file)
{
_file->close();
}
std::string filepath = _downloadPath + std::string("/") + filename;
ofstreamPtr ptr = Tools::GetNewFile(filepath);
_file.swap(ptr);
}
Task(const Task& t)
{
_url = t._url;
}
Task* operator= (const Task& t)
{
_url = t._url;
}
inline bool HasFlag(long flag)
{
return flag & _flag;
}
private:
Task(std::string url, long flag, std::string downPath)
:_url(url),
_flag(flag),
_downloadPath(downPath)
{
_head.Reset();
Tools::HttpStringOpt::SpliterHttpUrl(_url, _info);
}
public:
std::string _url;
long _flag;
HttpHead _head;
std::string _downloadPath; //下载目录
UrlInfo _info;
private:
ofstreamPtr _file;
};
2.5 缓冲数据块
#pragma once
#include "boost/smart_ptr.hpp"
#ifndef DEFAULT_BUF_SIZE
#define DEFAULT_BUF_SIZE 1024
#endif
class CMsgBlock;
typedef boost::shared_ptr<CMsgBlock> MsgBlockPtr; /************************************************************************/
/* 简易数据缓冲区 */
/************************************************************************/
class CMsgBlock
{
public:
CMsgBlock(){
Reset();
_block = NULL;
_capacity = ;
}
CMsgBlock(int sz){
Reset();
_block = NULL;
_capacity = ;
Capacity(sz);
}
~CMsgBlock(){
if (_block != NULL)
{
delete [] _block;
_block = NULL;
}
}
//重置缓冲区,仅移动指针
void Reset()
{
_rdPrt = ;
_wrPtr = ;
}
//获取数据块大小
int Size()
{
return _wrPtr - _rdPrt;
}
//获取缓冲区容量
int Capacity()
{
return _capacity;
}
//获取剩余空间
int Space()
{
return _capacity - _wrPtr;
}
// 获取基址
char* Base()
{
return _block;
}
//读地址
int RdPtr()
{
return _rdPrt;
}
void RdPtr(int ptr)
{
_rdPrt += ptr;
}
//写地址
int WtPtr()
{
return _wrPtr;
}
void WtPtr(int ptr)
{
_wrPtr += ptr;
}
//重置缓冲区大小
bool Capacity(int sz)
{
if (_capacity >= sz)
return true;
else
{
char* temp = new char[sz];
if (temp == NULL)
return false;
if (_block)
{
memcpy(temp,_block,_capacity);
delete [] _block;
}
_block = temp;
_capacity = sz;
}
return true;
}
//写缓冲区
bool Write(const char* buf,int leng)
{
//缓冲区可写区域不足
if (Space() < leng)
{
//重置缓冲区
if (!Capacity(_capacity + *leng + DEFAULT_BUF_SIZE ))
return false;
}
memcpy(_block + _wrPtr,buf,leng);
_wrPtr += leng;
return true;
}
void Copy(CMsgBlock* block)
{
if (block != this)
{
this->Reset();
this->Write(block->Base(),block->Size());
}
}
void Copy(CMsgBlock& block)
{
Copy(&block);
}
private:
char* _block; //数据块
unsigned int _rdPrt; //读指针
unsigned int _wrPtr; //写指针
unsigned int _capacity; //容量
};
2.6 日志输出
//输出日志
void Log(LogType type,const char *format_str, ...)
{ //调整缓冲区
_block.Reset(); va_list argp;
va_start (argp, format_str);
while (*format_str != '\0')
{
if (*format_str != '%')
{
_block.Write(format_str,);
}
else if (format_str[] == '%') // An "escaped" '%' (just print one '%').
{
format_str++; // Store first %
}
else
{
char format[] = {}; // 临时变量，保存%转换的临时结果
int len = ;
format_str++; // Copy in the % switch (*format_str)
{
case '-': case '+': case '': case ' ': case '#':
case '': case '': case '': case '': case '':
case '': case '': case '': case '': case '.':
case 'L': case 'h':
//*fp++ = *format_str;
break;
case 'l': // Source file line number
len = sprintf_s (format,,"%d",__LINE__);
_block.Write(format,len);
break;
case 'N': // Source file name
len = sprintf_s (format,,"%s",__FILE__);
_block.Write(format,len);
break;
case 'n': // Program name
len = sprintf_s (format,,"%s","<unknown>");
_block.Write(format,len);
break;
case 'P': // Process ID
len = sprintf_s (format,,"%d", (int)getpid());
_block.Write(format,len);
break;
case 'T': // Format the timestamp in hour:minute:sec:usec format.
{
std::string strColTime = Tools::GetCurrentTime();
_block.Write(strColTime.c_str(), strColTime.length());
}
break;
case 't': // Format thread id.
len = sprintf_s (format,,"%d", boost::this_thread::get_id());
_block.Write(format,len);
break;
case 's':
{// String
char *str1 = va_arg (argp, char *);
_block.Write(str1,strlen(str1));
break;
}
case 'd':
case 'i':
case 'o':
case 'u':
case 'x':
case 'X':
len = sprintf_s (format,,"%d",va_arg (argp, int));
_block.Write(format,len);
break;
default:
_block.Write(format_str,);
break;
}
}
++format_str;
}
//末尾结束符号
char c('\0');
_block.Write(&c,);
//输出缓冲区
if (BIT_ENABLED (_flags,STDERR)) // This is taken care of by our caller.
{
int const fwrite_result = fprintf (stderr,"%s",_block.Base());
::fflush (stderr);
}
if (BIT_ENABLED (_flags,OSTREAM))
{
if (_ostream != NULL)
{
*_ostream << _block.Base();
_ostream->flush();
}
}
va_end (argp);
}
