Skip to content

Z
ZLMediaKit
Commit 61fbb635 by xiongziliang
Options

优化http客户端代码

parent aab3a583
正在显示 包含 153 行增加 和 118 行删除
src/Http/HttpClient.cpp
/* /*
* MIT License * MIT License
* *
* Copyright (c) 2016 xiongziliang <771730766@qq.com> * Copyright (c) 2016 xiongziliang <771730766@qq.com>
* *
* This file is part of ZLMediaKit(https://github.com/xiongziliang/ZLMediaKit). * This file is part of ZLMediaKit(https://github.com/xiongziliang/ZLMediaKit).
* *
* Permission is hereby granted, free of charge, to any person obtaining a copy * Permission is hereby granted, free of charge, to any person obtaining a copy
* of this software and associated documentation files (the "Software"), to deal * of this software and associated documentation files (the "Software"), to deal
* in the Software without restriction, including without limitation the rights * in the Software without restriction, including without limitation the rights
* to use, copy, modify, merge, publish, distribute, sublicense, and/or sell * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
* copies of the Software, and to permit persons to whom the Software is * copies of the Software, and to permit persons to whom the Software is
* furnished to do so, subject to the following conditions: * furnished to do so, subject to the following conditions:
* *
* The above copyright notice and this permission notice shall be included in all * The above copyright notice and this permission notice shall be included in all
* copies or substantial portions of the Software. * copies or substantial portions of the Software.
* *
* THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
* IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
* FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
* AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
* LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
* OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
* SOFTWARE. * SOFTWARE.
*/ */
#include "HttpClient.h" #include "HttpClient.h"
#include "Rtsp/Rtsp.h" #include "Rtsp/Rtsp.h"
...@@ -31,22 +31,24 @@ namespace ZL { ...@@ -31,22 +31,24 @@ namespace ZL {
namespace Http { namespace Http {
HttpClient::HttpClient(){ HttpClient::HttpClient() {
} }
HttpClient::~HttpClient(){
HttpClient::~HttpClient() {
} }
void HttpClient::sendRequest(const string &strUrl,float fTimeOutSec){
void HttpClient::sendRequest(const string &strUrl, float fTimeOutSec) {
_aliveTicker.resetTime(); _aliveTicker.resetTime();
auto protocol = FindField(strUrl.data(), NULL , "://"); auto protocol = FindField(strUrl.data(), NULL, "://");
uint16_t defaultPort; uint16_t defaultPort;
bool isHttps; bool isHttps;
if (strcasecmp(protocol.data(), "http") == 0) { if (strcasecmp(protocol.data(), "http") == 0) {
defaultPort = 80; defaultPort = 80;
isHttps = false; isHttps = false;
}else if(strcasecmp(protocol.data(), "https") ==0 ){ } else if (strcasecmp(protocol.data(), "https") == 0) {
defaultPort = 443; defaultPort = 443;
isHttps = true; isHttps = true;
}else{ } else {
auto strErr = StrPrinter << "非法的协议:" << protocol << endl; auto strErr = StrPrinter << "非法的协议:" << protocol << endl;
throw std::invalid_argument(strErr); throw std::invalid_argument(strErr);
} }
...@@ -67,26 +69,27 @@ void HttpClient::sendRequest(const string &strUrl,float fTimeOutSec){ ...@@ -67,26 +69,27 @@ void HttpClient::sendRequest(const string &strUrl,float fTimeOutSec){
//服务器域名 //服务器域名
host = FindField(host.data(), NULL, ":"); host = FindField(host.data(), NULL, ":");
} }
_header.emplace(string("Host"),host); _header.emplace(string("Host"), host);
_header.emplace(string("Tools"),"ZLMediaKit"); _header.emplace(string("Tools"), "ZLMediaKit");
_header.emplace(string("Connection"),"keep-alive"); _header.emplace(string("Connection"), "keep-alive");
_header.emplace(string("Accept"),"*/*"); _header.emplace(string("Accept"), "*/*");
_header.emplace(string("Accept-Language"),"zh-CN,zh;q=0.8"); _header.emplace(string("Accept-Language"), "zh-CN,zh;q=0.8");
_header.emplace(string("User-Agent"),"Mozilla/5.0 (Macintosh; Intel Mac OS X 10_12_1) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/57.0.2987.133 Safari/537.36"); _header.emplace(string("User-Agent"),
"Mozilla/5.0 (Macintosh; Intel Mac OS X 10_12_1) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/57.0.2987.133 Safari/537.36");
if(_body && _body->remainSize()){ if (_body && _body->remainSize()) {
_header.emplace(string("Content-Length"),to_string(_body->remainSize())); _header.emplace(string("Content-Length"), to_string(_body->remainSize()));
_header.emplace(string("Content-Type"),"application/x-www-form-urlencoded; charset=UTF-8"); _header.emplace(string("Content-Type"), "application/x-www-form-urlencoded; charset=UTF-8");
} }
bool bChanged = (_lastHost != host + ":" + to_string(port)) || (_isHttps != isHttps); bool bChanged = (_lastHost != host + ":" + to_string(port)) || (_isHttps != isHttps);
_lastHost = host + ":" + to_string(port); _lastHost = host + ":" + to_string(port);
_isHttps = isHttps; _isHttps = isHttps;
_fTimeOutSec = fTimeOutSec; _fTimeOutSec = fTimeOutSec;
if(!alive() || bChanged){ if (!alive() || bChanged) {
//InfoL << "reconnet:" << _lastHost; //InfoL << "reconnet:" << _lastHost;
startConnect(host, port,fTimeOutSec); startConnect(host, port, fTimeOutSec);
}else{ } else {
SockException ex; SockException ex;
onConnect(ex); onConnect(ex);
} }
...@@ -95,12 +98,15 @@ void HttpClient::sendRequest(const string &strUrl,float fTimeOutSec){ ...@@ -95,12 +98,15 @@ void HttpClient::sendRequest(const string &strUrl,float fTimeOutSec){
void HttpClient::onConnect(const SockException &ex) { void HttpClient::onConnect(const SockException &ex) {
_aliveTicker.resetTime(); _aliveTicker.resetTime();
if(ex){ if (ex) {
onDisconnect(ex); onDisconnect(ex);
return; return;
} }
_recvedBodySize = -1;
_recvedResponse.clear(); _totalBodySize = 0;
_recvedBodySize = 0;
HttpRequestSplitter::reset();
_StrPrinter printer; _StrPrinter printer;
printer << _method + " " << _path + " HTTP/1.1\r\n"; printer << _method + " " << _path + " HTTP/1.1\r\n";
for (auto &pr : _header) { for (auto &pr : _header) {
...@@ -110,91 +116,71 @@ void HttpClient::onConnect(const SockException &ex) { ...@@ -110,91 +116,71 @@ void HttpClient::onConnect(const SockException &ex) {
send(printer << "\r\n"); send(printer << "\r\n");
onSend(); onSend();
} }
void HttpClient::onRecv(const Buffer::Ptr &pBuf) { void HttpClient::onRecv(const Buffer::Ptr &pBuf) {
onRecvBytes(pBuf->data(),pBuf->size()); onRecvBytes(pBuf->data(), pBuf->size());
}
void HttpClient::onRecvBytes(const char *data, int size) {
_aliveTicker.resetTime();
HttpRequestSplitter::input(data, size);
} }
void HttpClient::onErr(const SockException &ex) { void HttpClient::onErr(const SockException &ex) {
if(ex.getErrCode() == Err_eof && _totalBodySize == INT64_MAX){ if (ex.getErrCode() == Err_eof && _totalBodySize == INT64_MAX) {
//如果Content-Length未指定 但服务器断开链接 //如果Content-Length未指定 但服务器断开链接
//则认为本次http请求完成 //则认为本次http请求完成
_totalBodySize = 0; onResponseCompleted_l();
onResponseCompleted();
} }
onDisconnect(ex); onDisconnect(ex);
} }
void HttpClient::onRecvBytes(const char* data, int size) { int64_t HttpClient::onRecvHeader(const char *data, uint64_t len) {
_aliveTicker.resetTime(); _parser.Parse(data);
if(_recvedBodySize == -1){ onResponseHeader(_parser.Url(), _parser.getValues());
//还没有收到http body，这只是http头
auto lastLen = _recvedResponse.size();
_recvedResponse.append(data,size);
auto pos = _recvedResponse.find("\r\n\r\n",lastLen);
if(pos == string::npos){
//http 头还未收到
return;
}
_parser.Parse(_recvedResponse.data());
onResponseHeader(_parser.Url(),_parser.getValues());
_totalBodySize = atoll(((HttpHeader &)_parser.getValues())["Content-Length"].data()); if (_parser["Content-Length"].empty() && !_parser.Content().empty()) {
if(_totalBodySize == 0){ //如果http回复未声明Content-Length字段，但是却有content内容，那说明可能是个不限长度的content
_totalBodySize = INT64_MAX; _totalBodySize = INT64_MAX;
_recvedBodySize = 0;
//返回-1代表不限制content回复大小
return -1;
} }
_recvedBodySize = _recvedResponse.size() - pos - 4; _totalBodySize = atoll(_parser["Content-Length"].data());
if(_totalBodySize < _recvedBodySize){ _recvedBodySize = 0;
//http body 比声明的大 这个不可能的
_StrPrinter printer;
for(auto &pr: _parser.getValues()){
printer << pr.first << ":" << pr.second << "\r\n";
}
ErrorL << _totalBodySize << ":" << _recvedBodySize << "\r\n" << (printer << endl);
shutdown();
return;
}
if (_recvedBodySize) {
//_recvedResponse里面包含body负载
onResponseBody(_recvedResponse.data() + _recvedResponse.size() - _recvedBodySize, _recvedBodySize,_recvedBodySize,_totalBodySize);
}
if(_recvedBodySize >= _totalBodySize){ //虽然我们知道content的确切大小，
_totalBodySize = 0; //但是由于我们没必要等content接收完毕才回调onRecvContent(因为这样浪费内存并且要多次拷贝数据)
onResponseCompleted(); //所以返回-1代表我们接下来分段接收content
} return -1;
_recvedResponse.clear(); }
return;
} void HttpClient::onRecvContent(const char *data, uint64_t len) {
//http body auto recvedBodySize = _recvedBodySize + len;
if(_recvedBodySize < _totalBodySize){ if (recvedBodySize < _totalBodySize) {
_recvedBodySize += size; onResponseBody(data, len, recvedBodySize, _totalBodySize);
onResponseBody(data,size,_recvedBodySize,_totalBodySize); _recvedBodySize = recvedBodySize;
if(_recvedBodySize >= _totalBodySize){ } else {
//如果接收的数据大于Content-Length onResponseBody(data, _totalBodySize - _recvedBodySize, _totalBodySize, _totalBodySize);
//则认为本次http请求完成 bool biggerThanExpected = recvedBodySize > _totalBodySize;
_totalBodySize = 0; onResponseCompleted_l();
onResponseCompleted(); if(biggerThanExpected) {
} //声明的content数据比真实的小，那么我们只截取前面部分的并断开链接
return; shutdown();
onDisconnect(SockException(Err_other, "http response content size bigger than expected"));
} }
//http body 比声明的大 这个不可能的
_StrPrinter printer;
for(auto &pr: _parser.getValues()){
printer << pr.first << ":" << pr.second << "\r\n";
} }
ErrorL << _totalBodySize << ":" << _recvedBodySize << "\r\n" << (printer << endl);
shutdown();
} }
void HttpClient::onSend() { void HttpClient::onSend() {
_aliveTicker.resetTime(); _aliveTicker.resetTime();
while (_body && _body->remainSize() && !isSocketBusy()){ while (_body && _body->remainSize() && !isSocketBusy()) {
auto buffer = _body->readData(); auto buffer = _body->readData();
if (!buffer){ if (!buffer) {
//数据发送结束或读取数据异常 //数据发送结束或读取数据异常
break; break;
} }
if(send(buffer) <= 0){ if (send(buffer) <= 0) {
//发送数据失败，不需要回滚数据，因为发送前已经通过isSocketBusy()判断socket可写 //发送数据失败，不需要回滚数据，因为发送前已经通过isSocketBusy()判断socket可写
//所以发送缓存区肯定未满,该buffer肯定已经写入socket //所以发送缓存区肯定未满,该buffer肯定已经写入socket
break; break;
...@@ -203,20 +189,26 @@ void HttpClient::onSend() { ...@@ -203,20 +189,26 @@ void HttpClient::onSend() {
} }
void HttpClient::onManager() { void HttpClient::onManager() {
if(_aliveTicker.elapsedTime() > 3 * 1000 && _totalBodySize == INT64_MAX){ if (_aliveTicker.elapsedTime() > 3 * 1000 && _totalBodySize == INT64_MAX) {
//如果Content-Length未指定 但接收数据超时 //如果Content-Length未指定 但接收数据超时
//则认为本次http请求完成 //则认为本次http请求完成
_totalBodySize = 0; onResponseCompleted_l();
onResponseCompleted();
} }
if(_fTimeOutSec > 0 && _aliveTicker.elapsedTime() > _fTimeOutSec * 1000){ if (_fTimeOutSec > 0 && _aliveTicker.elapsedTime() > _fTimeOutSec * 1000) {
//超时 //超时
onDisconnect(SockException(Err_timeout,"http request timeout")); onDisconnect(SockException(Err_timeout, "http request timeout"));
shutdown(); shutdown();
} }
} }
void HttpClient::onResponseCompleted_l() {
_totalBodySize = 0;
_recvedBodySize = 0;
HttpRequestSplitter::reset();
onResponseCompleted();
}
} /* namespace Http */ } /* namespace Http */
} /* namespace ZL */ } /* namespace ZL */
......
src/Http/HttpClient.h
...@@ -34,6 +34,7 @@ ...@@ -34,6 +34,7 @@
#include "Rtsp/Rtsp.h" #include "Rtsp/Rtsp.h"
#include "Util/util.h" #include "Util/util.h"
#include "Network/TcpClient.h" #include "Network/TcpClient.h"
#include "HttpRequestSplitter.h"
using namespace std; using namespace std;
using namespace ZL::Util; using namespace ZL::Util;
...@@ -198,7 +199,7 @@ private: ...@@ -198,7 +199,7 @@ private:
class HttpClient : public TcpClient class HttpClient : public TcpClient , public HttpRequestSplitter
{ {
public: public:
typedef StrCaseMap HttpHeader; typedef StrCaseMap HttpHeader;
...@@ -211,7 +212,6 @@ public: ...@@ -211,7 +212,6 @@ public:
_body.reset(); _body.reset();
_method.clear(); _method.clear();
_path.clear(); _path.clear();
_recvedResponse.clear();
_parser.Clear(); _parser.Clear();
} }
void setMethod(const string &method){ void setMethod(const string &method){
...@@ -241,23 +241,57 @@ public: ...@@ -241,23 +241,57 @@ public:
return _parser.getValues(); return _parser.getValues();
} }
protected: protected:
/**
* 收到http回复头
* @param status 状态码，譬如:200 OK
* @param headers http头
*/
virtual void onResponseHeader(const string &status,const HttpHeader &headers){ virtual void onResponseHeader(const string &status,const HttpHeader &headers){
DebugL << status; DebugL << status;
}; };
/**
* 收到http conten数据
* @param buf 数据指针
* @param size 数据大小
* @param recvedSize 已收数据大小(包含本次数据大小),当其等于totalSize时将触发onResponseCompleted回调
* @param totalSize 总数据大小
*/
virtual void onResponseBody(const char *buf,size_t size,size_t recvedSize,size_t totalSize){ virtual void onResponseBody(const char *buf,size_t size,size_t recvedSize,size_t totalSize){
DebugL << size << " " << recvedSize << " " << totalSize; DebugL << size << " " << recvedSize << " " << totalSize;
}; };
/**
* 接收http回复完毕
*/
virtual void onResponseCompleted(){ virtual void onResponseCompleted(){
DebugL; DebugL;
} }
/**
* 收到http回复数据回调
* @param data 数据指针
* @param size 数据大小
*/
virtual void onRecvBytes(const char *data,int size); virtual void onRecvBytes(const char *data,int size);
/**
* http链接断开回调
* @param ex 断开原因
*/
virtual void onDisconnect(const SockException &ex){} virtual void onDisconnect(const SockException &ex){}
//HttpRequestSplitter override
int64_t onRecvHeader(const char *data,uint64_t len) override ;
void onRecvContent(const char *data,uint64_t len) override;
protected: protected:
virtual void onConnect(const SockException &ex) override; virtual void onConnect(const SockException &ex) override;
virtual void onRecv(const Buffer::Ptr &pBuf) override; virtual void onRecv(const Buffer::Ptr &pBuf) override;
virtual void onErr(const SockException &ex) override; virtual void onErr(const SockException &ex) override;
virtual void onSend() override; virtual void onSend() override;
virtual void onManager() override; virtual void onManager() override;
private:
void onResponseCompleted_l();
protected: protected:
bool _isHttps; bool _isHttps;
private: private:
...@@ -267,7 +301,6 @@ private: ...@@ -267,7 +301,6 @@ private:
string _method; string _method;
string _path; string _path;
//recv //recv
string _recvedResponse;
int64_t _recvedBodySize; int64_t _recvedBodySize;
int64_t _totalBodySize; int64_t _totalBodySize;
Parser _parser; Parser _parser;
......
src/Http/HttpRequestSplitter.cpp
...@@ -74,3 +74,8 @@ splitPacket: ...@@ -74,3 +74,8 @@ splitPacket:
void HttpRequestSplitter::setContentLen(int64_t content_len) { void HttpRequestSplitter::setContentLen(int64_t content_len) {
_content_len = content_len; _content_len = content_len;
} }
void HttpRequestSplitter::reset() {
_content_len = 0;
_remain_data.clear();
}
src/Http/HttpRequestSplitter.h
...@@ -26,9 +26,9 @@ protected: ...@@ -26,9 +26,9 @@ protected:
* @param len 请求头长度 * @param len 请求头长度
* *
* @return 请求头后的content长度, * @return 请求头后的content长度,
* <0 : 代表后面所有数据都是content * <0 : 代表后面所有数据都是content，此时后面的content将分段通过onRecvContent函数回调出去
* 0 : 代表为后面数据还是请求头, * 0 : 代表为后面数据还是请求头,
* >0 : 代表后面数据为固定长度content, * >0 : 代表后面数据为固定长度content,此时将缓存content并等到所有content接收完毕一次性通过onRecvContent函数回调出去
*/ */
virtual int64_t onRecvHeader(const char *data,uint64_t len) = 0; virtual int64_t onRecvHeader(const char *data,uint64_t len) = 0;
...@@ -44,6 +44,11 @@ protected: ...@@ -44,6 +44,11 @@ protected:
* 设置content len * 设置content len
*/ */
void setContentLen(int64_t content_len); void setContentLen(int64_t content_len);
/**
* 恢复初始设置
*/
void reset();
private: private:
string _remain_data; string _remain_data;
int64_t _content_len = 0; int64_t _content_len = 0;
......
src/Http/WebSocketSplitter.cpp
...@@ -98,8 +98,8 @@ begin_decode: ...@@ -98,8 +98,8 @@ begin_decode:
if(playload_slice_len + _playload_offset > _playload_len){ if(playload_slice_len + _playload_offset > _playload_len){
playload_slice_len = _playload_len - _playload_offset; playload_slice_len = _playload_len - _playload_offset;
} }
onPlayloadData(ptr,playload_slice_len);
_playload_offset += playload_slice_len; _playload_offset += playload_slice_len;
onPlayloadData(ptr,playload_slice_len);
if(_playload_offset == _playload_len){ if(_playload_offset == _playload_len){
//这是下一个包 //这是下一个包
......
src/Http/WebSocketSplitter.h
...@@ -77,7 +77,7 @@ protected: ...@@ -77,7 +77,7 @@ protected:
* @param packet 数据包包头 * @param packet 数据包包头
* @param ptr 负载数据指针 * @param ptr 负载数据指针
* @param len 负载数据长度 * @param len 负载数据长度
* @param recved 已接收数据长度，等于packet._playload_len时则接受完毕 * @param recved 已接收数据长度(包含本次数据长度)，等于packet._playload_len时则接受完毕
*/ */
virtual void onWebSocketDecodePlayload(const WebSocketHeader &packet, const uint8_t *ptr, uint64_t len, uint64_t recved) {}; virtual void onWebSocketDecodePlayload(const WebSocketHeader &packet, const uint8_t *ptr, uint64_t len, uint64_t recved) {};
......
Markdown 格式
0% 或
您添加了 0 人 到此讨论。请谨慎行事。
请先完成此评论的编辑！
请 注册 或者 后发表评论