amass/ZLMediaKit
1
0
Fork 0
mirror of https://github.com/ZLMediaKit/ZLMediaKit.git synced 2024-11-22 19:00:01 +08:00
Code Issues Actions 6 Packages Projects Releases Wiki Activity

格式化HttpSession.cpp代码

Browse Source
This commit is contained in:
xia-chu 2023-06-10 09:45:46 +08:00 committed by 夏楚
parent 7e1078bbcc
commit fff53cf0e2
1 changed files with 166 additions and 177 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

343
src/Http/HttpSession.cpp
View File

@ -24,16 +24,16 @@ using namespace toolkit;
namespace mediakit { namespace mediakit {
HttpSession::HttpSession(const Socket::Ptr &pSock) : Session(pSock) { HttpSession::HttpSession(const Socket::Ptr &pSock) : Session(pSock) {
GET_CONFIG(uint32_t,keep_alive_sec,Http::kKeepAliveSecond); GET_CONFIG(uint32_t, keep_alive_sec, Http::kKeepAliveSecond);
pSock->setSendTimeOutSecond(keep_alive_sec); pSock->setSendTimeOutSecond(keep_alive_sec);
} }
HttpSession::~HttpSession() = default; HttpSession::~HttpSession() = default;
void HttpSession::Handle_Req_HEAD(ssize_t &content_len){ void HttpSession::Handle_Req_HEAD(ssize_t &content_len) {
//暂时全部返回200 OK，因为HTTP GET存在按需生成流的操作，所以不能按照HTTP GET的流程返回 // 暂时全部返回200 OK，因为HTTP GET存在按需生成流的操作，所以不能按照HTTP GET的流程返回
//如果直接返回404，那么又会导致按需生成流的逻辑失效，所以HTTP HEAD在静态文件或者已存在资源时才有效 // 如果直接返回404，那么又会导致按需生成流的逻辑失效，所以HTTP HEAD在静态文件或者已存在资源时才有效
//对于按需生成流的直播场景并不适用 // 对于按需生成流的直播场景并不适用
sendResponse(200, false); sendResponse(200, false);
} }
@ -52,16 +52,16 @@ void HttpSession::Handle_Req_OPTIONS(ssize_t &content_len) {
sendResponse(200, true, nullptr, header); sendResponse(200, true, nullptr, header);
} }
ssize_t HttpSession::onRecvHeader(const char *header,size_t len) { ssize_t HttpSession::onRecvHeader(const char *header, size_t len) {
typedef void (HttpSession::*HttpCMDHandle)(ssize_t &); using func_type = void (HttpSession::*)(ssize_t &);
static unordered_map<string, HttpCMDHandle> s_func_map; static unordered_map<string, func_type> s_func_map;
static onceToken token([]() { static onceToken token([]() {
s_func_map.emplace("GET",&HttpSession::Handle_Req_GET); s_func_map.emplace("GET", &HttpSession::Handle_Req_GET);
s_func_map.emplace("DELETE",&HttpSession::Handle_Req_GET); s_func_map.emplace("DELETE", &HttpSession::Handle_Req_GET);
s_func_map.emplace("POST",&HttpSession::Handle_Req_POST); s_func_map.emplace("POST", &HttpSession::Handle_Req_POST);
s_func_map.emplace("HEAD",&HttpSession::Handle_Req_HEAD); s_func_map.emplace("HEAD", &HttpSession::Handle_Req_HEAD);
s_func_map.emplace("OPTIONS",&HttpSession::Handle_Req_OPTIONS); s_func_map.emplace("OPTIONS", &HttpSession::Handle_Req_OPTIONS);
}, nullptr); });
_parser.Parse(header); _parser.Parse(header);
CHECK(_parser.Url()[0] == '/'); CHECK(_parser.Url()[0] == '/');
@ -75,60 +75,54 @@ ssize_t HttpSession::onRecvHeader(const char *header,size_t len) {
return 0; return 0;
} }
//跨域 // 跨域
_origin = _parser["Origin"]; _origin = _parser["Origin"];
//默认后面数据不是content而是header //默认后面数据不是content而是header
ssize_t content_len = 0; ssize_t content_len = 0;
(this->*(it->second))(content_len); (this->*(it->second))(content_len);
//清空解析器节省内存 // 清空解析器节省内存
_parser.Clear(); _parser.Clear();
//返回content长度 // 返回content长度
return content_len; return content_len;
} }
void HttpSession::onRecvContent(const char *data,size_t len) { void HttpSession::onRecvContent(const char *data, size_t len) {
if(_contentCallBack){ if (_contentCallBack && !_contentCallBack(data, len)) {
if(!_contentCallBack(data,len)){ _contentCallBack = nullptr;
_contentCallBack = nullptr;
}
} }
} }
void HttpSession::onRecv(const Buffer::Ptr &pBuf) { void HttpSession::onRecv(const Buffer::Ptr &pBuf) {
_ticker.resetTime(); _ticker.resetTime();
input(pBuf->data(),pBuf->size()); input(pBuf->data(), pBuf->size());
} }
void HttpSession::onError(const SockException& err) { void HttpSession::onError(const SockException &err) {
if (_is_live_stream) { if (_is_live_stream) {
//flv/ts播放器 // flv/ts播放器
uint64_t duration = _ticker.createdTime() / 1000; uint64_t duration = _ticker.createdTime() / 1000;
WarnP(this) << "FLV/TS/FMP4播放器(" WarnP(this) << "FLV/TS/FMP4播放器(" << _mediaInfo.shortUrl() << ")断开:" << err << ",耗时(s):" << duration;
<< _mediaInfo.shortUrl()
<< ")断开:" << err
<< ",耗时(s):" << duration;
GET_CONFIG(uint32_t, iFlowThreshold, General::kFlowThreshold); GET_CONFIG(uint32_t, iFlowThreshold, General::kFlowThreshold);
if (_total_bytes_usage >= iFlowThreshold * 1024) { if (_total_bytes_usage >= iFlowThreshold * 1024) {
NoticeCenter::Instance().emitEvent(Broadcast::kBroadcastFlowReport, _mediaInfo, _total_bytes_usage, NoticeCenter::Instance().emitEvent(Broadcast::kBroadcastFlowReport, _mediaInfo, _total_bytes_usage, duration, true, static_cast<SockInfo &>(*this));
duration, true, static_cast<SockInfo &>(*this));
} }
return; return;
} }
} }
void HttpSession::onManager() { void HttpSession::onManager() {
GET_CONFIG(uint32_t,keepAliveSec,Http::kKeepAliveSecond); GET_CONFIG(uint32_t, keepAliveSec, Http::kKeepAliveSecond);
if(_ticker.elapsedTime() > keepAliveSec * 1000){ if (_ticker.elapsedTime() > keepAliveSec * 1000) {
//1分钟超时 // 1分钟超时
shutdown(SockException(Err_timeout,"session timeout")); shutdown(SockException(Err_timeout, "session timeout"));
} }
} }
bool HttpSession::checkWebSocket(){ bool HttpSession::checkWebSocket() {
auto Sec_WebSocket_Key = _parser["Sec-WebSocket-Key"]; auto Sec_WebSocket_Key = _parser["Sec-WebSocket-Key"];
if (Sec_WebSocket_Key.empty()) { if (Sec_WebSocket_Key.empty()) {
return false; return false;
@ -147,32 +141,32 @@ bool HttpSession::checkWebSocket(){
_live_over_websocket = true; _live_over_websocket = true;
sendResponse(101, false, nullptr, headerOut, nullptr, true); sendResponse(101, false, nullptr, headerOut, nullptr, true);
}; };
auto res_cb_flv = [this, headerOut]() mutable { auto res_cb_flv = [this, headerOut]() mutable {
_live_over_websocket = true; _live_over_websocket = true;
headerOut.emplace("Cache-Control", "no-store"); headerOut.emplace("Cache-Control", "no-store");
sendResponse(101, false, nullptr, headerOut, nullptr, true); sendResponse(101, false, nullptr, headerOut, nullptr, true);
}; };
//判断是否为websocket-flv // 判断是否为websocket-flv
if (checkLiveStreamFlv(res_cb_flv)) { if (checkLiveStreamFlv(res_cb_flv)) {
//这里是websocket-flv直播请求 // 这里是websocket-flv直播请求
return true; return true;
} }
//判断是否为websocket-ts // 判断是否为websocket-ts
if (checkLiveStreamTS(res_cb)) { if (checkLiveStreamTS(res_cb)) {
//这里是websocket-ts直播请求 // 这里是websocket-ts直播请求
return true; return true;
} }
//判断是否为websocket-fmp4 // 判断是否为websocket-fmp4
if (checkLiveStreamFMP4(res_cb)) { if (checkLiveStreamFMP4(res_cb)) {
//这里是websocket-fmp4直播请求 // 这里是websocket-fmp4直播请求
return true; return true;
} }
//这是普通的websocket连接 // 这是普通的websocket连接
if (!onWebSocketConnect(_parser)) { if (!onWebSocketConnect(_parser)) {
sendResponse(501, true, nullptr, headerOut); sendResponse(501, true, nullptr, headerOut);
return true; return true;
@ -181,7 +175,7 @@ bool HttpSession::checkWebSocket(){
return true; return true;
} }
bool HttpSession::checkLiveStream(const string &schema, const string &url_suffix, const function<void(const MediaSource::Ptr &src)> &cb){ bool HttpSession::checkLiveStream(const string &schema, const string &url_suffix, const function<void(const MediaSource::Ptr &src)> &cb) {
std::string url = _parser.Url(); std::string url = _parser.Url();
auto it = _parser.getUrlArgs().find("schema"); auto it = _parser.getUrlArgs().find("schema");
if (it != _parser.getUrlArgs().end()) { if (it != _parser.getUrlArgs().end()) {
@ -192,57 +186,57 @@ bool HttpSession::checkLiveStream(const string &schema, const string &url_suffi
} else { } else {
auto prefix_size = url_suffix.size(); auto prefix_size = url_suffix.size();
if (url.size() < prefix_size || strcasecmp(url.data() + (url.size() - prefix_size), url_suffix.data())) { if (url.size() < prefix_size || strcasecmp(url.data() + (url.size() - prefix_size), url_suffix.data())) {
//未找到后缀 // 未找到后缀
return false; return false;
} }
// url去除特殊后缀 // url去除特殊后缀
url.resize(url.size() - prefix_size); url.resize(url.size() - prefix_size);
} }
//带参数的url // 带参数的url
if (!_parser.Params().empty()) { if (!_parser.Params().empty()) {
url += "?"; url += "?";
url += _parser.Params(); url += _parser.Params();
} }
//解析带上协议+参数完整的url // 解析带上协议+参数完整的url
_mediaInfo.parse(schema + "://" + _parser["Host"] + url); _mediaInfo.parse(schema + "://" + _parser["Host"] + url);
if (_mediaInfo.app.empty() || _mediaInfo.stream.empty()) { if (_mediaInfo.app.empty() || _mediaInfo.stream.empty()) {
//url不合法 // url不合法
return false; return false;
} }
bool close_flag = !strcasecmp(_parser["Connection"].data(), "close"); bool close_flag = !strcasecmp(_parser["Connection"].data(), "close");
weak_ptr<HttpSession> weak_self = static_pointer_cast<HttpSession>(shared_from_this()); weak_ptr<HttpSession> weak_self = static_pointer_cast<HttpSession>(shared_from_this());
//鉴权结果回调 // 鉴权结果回调
auto onRes = [cb, weak_self, close_flag](const string &err) { auto onRes = [cb, weak_self, close_flag](const string &err) {
auto strong_self = weak_self.lock(); auto strong_self = weak_self.lock();
if (!strong_self) { if (!strong_self) {
//本对象已经销毁 // 本对象已经销毁
return; return;
} }
if (!err.empty()) { if (!err.empty()) {
//播放鉴权失败 // 播放鉴权失败
strong_self->sendResponse(401, close_flag, nullptr, KeyValue(), std::make_shared<HttpStringBody>(err)); strong_self->sendResponse(401, close_flag, nullptr, KeyValue(), std::make_shared<HttpStringBody>(err));
return; return;
} }
//异步查找直播流 // 异步查找直播流
MediaSource::findAsync(strong_self->_mediaInfo, strong_self, [weak_self, close_flag, cb](const MediaSource::Ptr &src) { MediaSource::findAsync(strong_self->_mediaInfo, strong_self, [weak_self, close_flag, cb](const MediaSource::Ptr &src) {
auto strong_self = weak_self.lock(); auto strong_self = weak_self.lock();
if (!strong_self) { if (!strong_self) {
//本对象已经销毁 // 本对象已经销毁
return; return;
} }
if (!src) { if (!src) {
//未找到该流 // 未找到该流
strong_self->sendNotFound(close_flag); strong_self->sendNotFound(close_flag);
} else { } else {
strong_self->_is_live_stream = true; strong_self->_is_live_stream = true;
//触发回调 // 触发回调
cb(src); cb(src);
} }
}); });
@ -256,26 +250,26 @@ bool HttpSession::checkLiveStream(const string &schema, const string &url_suffi
auto flag = NoticeCenter::Instance().emitEvent(Broadcast::kBroadcastMediaPlayed, _mediaInfo, invoker, static_cast<SockInfo &>(*this)); auto flag = NoticeCenter::Instance().emitEvent(Broadcast::kBroadcastMediaPlayed, _mediaInfo, invoker, static_cast<SockInfo &>(*this));
if (!flag) { if (!flag) {
//该事件无人监听,默认不鉴权 // 该事件无人监听,默认不鉴权
onRes(""); onRes("");
} }
return true; return true;
} }
//http-fmp4 链接格式:http://vhost-url:port/app/streamid.live.mp4?key1=value1&key2=value2 // http-fmp4 链接格式:http://vhost-url:port/app/streamid.live.mp4?key1=value1&key2=value2
bool HttpSession::checkLiveStreamFMP4(const function<void()> &cb){ bool HttpSession::checkLiveStreamFMP4(const function<void()> &cb) {
return checkLiveStream(FMP4_SCHEMA, ".live.mp4", [this, cb](const MediaSource::Ptr &src) { return checkLiveStream(FMP4_SCHEMA, ".live.mp4", [this, cb](const MediaSource::Ptr &src) {
auto fmp4_src = dynamic_pointer_cast<FMP4MediaSource>(src); auto fmp4_src = dynamic_pointer_cast<FMP4MediaSource>(src);
assert(fmp4_src); assert(fmp4_src);
if (!cb) { if (!cb) {
//找到源，发送http头，负载后续发送 // 找到源，发送http头，负载后续发送
sendResponse(200, false, HttpFileManager::getContentType(".mp4").data(), KeyValue(), nullptr, true); sendResponse(200, false, HttpFileManager::getContentType(".mp4").data(), KeyValue(), nullptr, true);
} else { } else {
//自定义发送http头 // 自定义发送http头
cb(); cb();
} }
//直播牺牲延时提升发送性能 // 直播牺牲延时提升发送性能
setSocketFlags(); setSocketFlags();
onWrite(std::make_shared<BufferString>(fmp4_src->getInitSegment()), true); onWrite(std::make_shared<BufferString>(fmp4_src->getInitSegment()), true);
weak_ptr<HttpSession> weak_self = static_pointer_cast<HttpSession>(shared_from_this()); weak_ptr<HttpSession> weak_self = static_pointer_cast<HttpSession>(shared_from_this());
@ -285,7 +279,7 @@ bool HttpSession::checkLiveStreamFMP4(const function<void()> &cb){
_fmp4_reader->setDetachCB([weak_self]() { _fmp4_reader->setDetachCB([weak_self]() {
auto strong_self = weak_self.lock(); auto strong_self = weak_self.lock();
if (!strong_self) { if (!strong_self) {
//本对象已经销毁 // 本对象已经销毁
return; return;
} }
strong_self->shutdown(SockException(Err_shutdown, "fmp4 ring buffer detached")); strong_self->shutdown(SockException(Err_shutdown, "fmp4 ring buffer detached"));
@ -293,85 +287,80 @@ bool HttpSession::checkLiveStreamFMP4(const function<void()> &cb){
_fmp4_reader->setReadCB([weak_self](const FMP4MediaSource::RingDataType &fmp4_list) { _fmp4_reader->setReadCB([weak_self](const FMP4MediaSource::RingDataType &fmp4_list) {
auto strong_self = weak_self.lock(); auto strong_self = weak_self.lock();
if (!strong_self) { if (!strong_self) {
//本对象已经销毁 // 本对象已经销毁
return; return;
} }
size_t i = 0; size_t i = 0;
auto size = fmp4_list->size(); auto size = fmp4_list->size();
fmp4_list->for_each([&](const FMP4Packet::Ptr &ts) { fmp4_list->for_each([&](const FMP4Packet::Ptr &ts) { strong_self->onWrite(ts, ++i == size); });
strong_self->onWrite(ts, ++i == size);
});
}); });
}); });
} }
//http-ts 链接格式:http://vhost-url:port/app/streamid.live.ts?key1=value1&key2=value2 // http-ts 链接格式:http://vhost-url:port/app/streamid.live.ts?key1=value1&key2=value2
bool HttpSession::checkLiveStreamTS(const function<void()> &cb){ bool HttpSession::checkLiveStreamTS(const function<void()> &cb) {
return checkLiveStream(TS_SCHEMA, ".live.ts", [this, cb](const MediaSource::Ptr &src) { return checkLiveStream(TS_SCHEMA, ".live.ts", [this, cb](const MediaSource::Ptr &src) {
auto ts_src = dynamic_pointer_cast<TSMediaSource>(src); auto ts_src = dynamic_pointer_cast<TSMediaSource>(src);
assert(ts_src); assert(ts_src);
if (!cb) { if (!cb) {
//找到源，发送http头，负载后续发送 // 找到源，发送http头，负载后续发送
sendResponse(200, false, HttpFileManager::getContentType(".ts").data(), KeyValue(), nullptr, true); sendResponse(200, false, HttpFileManager::getContentType(".ts").data(), KeyValue(), nullptr, true);
} else { } else {
//自定义发送http头 // 自定义发送http头
cb(); cb();
} }
//直播牺牲延时提升发送性能 // 直播牺牲延时提升发送性能
setSocketFlags(); setSocketFlags();
weak_ptr<HttpSession> weak_self = static_pointer_cast<HttpSession>(shared_from_this()); weak_ptr<HttpSession> weak_self = static_pointer_cast<HttpSession>(shared_from_this());
ts_src->pause(false); ts_src->pause(false);
_ts_reader = ts_src->getRing()->attach(getPoller()); _ts_reader = ts_src->getRing()->attach(getPoller());
_ts_reader->setGetInfoCB([weak_self]() { return weak_self.lock(); }); _ts_reader->setGetInfoCB([weak_self]() { return weak_self.lock(); });
_ts_reader->setDetachCB([weak_self](){ _ts_reader->setDetachCB([weak_self]() {
auto strong_self = weak_self.lock(); auto strong_self = weak_self.lock();
if (!strong_self) { if (!strong_self) {
//本对象已经销毁 // 本对象已经销毁
return; return;
} }
strong_self->shutdown(SockException(Err_shutdown,"ts ring buffer detached")); strong_self->shutdown(SockException(Err_shutdown, "ts ring buffer detached"));
}); });
_ts_reader->setReadCB([weak_self](const TSMediaSource::RingDataType &ts_list) { _ts_reader->setReadCB([weak_self](const TSMediaSource::RingDataType &ts_list) {
auto strong_self = weak_self.lock(); auto strong_self = weak_self.lock();
if (!strong_self) { if (!strong_self) {
//本对象已经销毁 // 本对象已经销毁
return; return;
} }
size_t i = 0; size_t i = 0;
auto size = ts_list->size(); auto size = ts_list->size();
ts_list->for_each([&](const TSPacket::Ptr &ts) { ts_list->for_each([&](const TSPacket::Ptr &ts) { strong_self->onWrite(ts, ++i == size); });
strong_self->onWrite(ts, ++i == size);
});
}); });
}); });
} }
//http-flv 链接格式:http://vhost-url:port/app/streamid.live.flv?key1=value1&key2=value2 // http-flv 链接格式:http://vhost-url:port/app/streamid.live.flv?key1=value1&key2=value2
bool HttpSession::checkLiveStreamFlv(const function<void()> &cb){ bool HttpSession::checkLiveStreamFlv(const function<void()> &cb) {
auto start_pts = atoll(_parser.getUrlArgs()["starPts"].data()); auto start_pts = atoll(_parser.getUrlArgs()["starPts"].data());
return checkLiveStream(RTMP_SCHEMA, ".live.flv", [this, cb, start_pts](const MediaSource::Ptr &src) { return checkLiveStream(RTMP_SCHEMA, ".live.flv", [this, cb, start_pts](const MediaSource::Ptr &src) {
auto rtmp_src = dynamic_pointer_cast<RtmpMediaSource>(src); auto rtmp_src = dynamic_pointer_cast<RtmpMediaSource>(src);
assert(rtmp_src); assert(rtmp_src);
if (!cb) { if (!cb) {
//找到源，发送http头，负载后续发送 // 找到源，发送http头，负载后续发送
KeyValue headerOut; KeyValue headerOut;
headerOut["Cache-Control"] = "no-store"; headerOut["Cache-Control"] = "no-store";
sendResponse(200, false, HttpFileManager::getContentType(".flv").data(), headerOut, nullptr, true); sendResponse(200, false, HttpFileManager::getContentType(".flv").data(), headerOut, nullptr, true);
} else { } else {
//自定义发送http头 // 自定义发送http头
cb(); cb();
} }
//直播牺牲延时提升发送性能 // 直播牺牲延时提升发送性能
setSocketFlags(); setSocketFlags();
//非H264/AAC时打印警告日志，防止用户提无效问题 // 非H264/AAC时打印警告日志，防止用户提无效问题
auto tracks = src->getTracks(false); auto tracks = src->getTracks(false);
for (auto &track : tracks) { for (auto &track : tracks) {
switch (track->getCodecId()) { switch (track->getCodecId()) {
case CodecH264: case CodecH264:
case CodecAAC: case CodecAAC: break;
break;
default: { default: {
WarnP(this) << "flv播放器一般只支持H264和AAC编码,该编码格式可能不被播放器支持:" << track->getCodecName(); WarnP(this) << "flv播放器一般只支持H264和AAC编码,该编码格式可能不被播放器支持:" << track->getCodecName();
break; break;
@ -388,11 +377,11 @@ void HttpSession::Handle_Req_GET(ssize_t &content_len) {
} }
void HttpSession::Handle_Req_GET_l(ssize_t &content_len, bool sendBody) { void HttpSession::Handle_Req_GET_l(ssize_t &content_len, bool sendBody) {
//先看看是否为WebSocket请求 // 先看看是否为WebSocket请求
if (checkWebSocket()) { if (checkWebSocket()) {
content_len = -1; content_len = -1;
_contentCallBack = [this](const char *data, size_t len) { _contentCallBack = [this](const char *data, size_t len) {
WebSocketSplitter::decode((uint8_t *) data, len); WebSocketSplitter::decode((uint8_t *)data, len);
//_contentCallBack是可持续的，后面还要处理后续数据 //_contentCallBack是可持续的，后面还要处理后续数据
return true; return true;
}; };
@ -400,29 +389,29 @@ void HttpSession::Handle_Req_GET_l(ssize_t &content_len, bool sendBody) {
} }
if (emitHttpEvent(false)) { if (emitHttpEvent(false)) {
//拦截http api事件 // 拦截http api事件
return; return;
} }
if (checkLiveStreamFlv()) { if (checkLiveStreamFlv()) {
//拦截http-flv播放器 // 拦截http-flv播放器
return; return;
} }
if (checkLiveStreamTS()) { if (checkLiveStreamTS()) {
//拦截http-ts播放器 // 拦截http-ts播放器
return; return;
} }
if (checkLiveStreamFMP4()) { if (checkLiveStreamFMP4()) {
//拦截http-fmp4播放器 // 拦截http-fmp4播放器
return; return;
} }
bool bClose = !strcasecmp(_parser["Connection"].data(),"close"); bool bClose = !strcasecmp(_parser["Connection"].data(), "close");
weak_ptr<HttpSession> weak_self = static_pointer_cast<HttpSession>(shared_from_this()); weak_ptr<HttpSession> weak_self = static_pointer_cast<HttpSession>(shared_from_this());
HttpFileManager::onAccessPath(*this, _parser, [weak_self, bClose](int code, const string &content_type, HttpFileManager::onAccessPath(*this, _parser, [weak_self, bClose](int code, const string &content_type,
const StrCaseMap &responseHeader, const HttpBody::Ptr &body) { const StrCaseMap &responseHeader, const HttpBody::Ptr &body) {
auto strong_self = weak_self.lock(); auto strong_self = weak_self.lock();
if (!strong_self) { if (!strong_self) {
return; return;
@ -468,7 +457,7 @@ public:
static bool onSocketFlushed(const AsyncSenderData::Ptr &data) { static bool onSocketFlushed(const AsyncSenderData::Ptr &data) {
if (data->_read_complete) { if (data->_read_complete) {
if (data->_close_when_complete) { if (data->_close_when_complete) {
//发送完毕需要关闭socket // 发送完毕需要关闭socket
shutdown(data->_session.lock()); shutdown(data->_session.lock());
} }
return false; return false;
@ -478,13 +467,13 @@ public:
data->_body->readDataAsync(sendBufSize, [data](const Buffer::Ptr &sendBuf) { data->_body->readDataAsync(sendBufSize, [data](const Buffer::Ptr &sendBuf) {
auto session = data->_session.lock(); auto session = data->_session.lock();
if (!session) { if (!session) {
//本对象已经销毁 // 本对象已经销毁
return; return;
} }
session->async([data, sendBuf]() { session->async([data, sendBuf]() {
auto session = data->_session.lock(); auto session = data->_session.lock();
if (!session) { if (!session) {
//本对象已经销毁 // 本对象已经销毁
return; return;
} }
onRequestData(data, session, sendBuf); onRequestData(data, session, sendBuf);
@ -497,14 +486,14 @@ private:
static void onRequestData(const AsyncSenderData::Ptr &data, const std::shared_ptr<HttpSession> &session, const Buffer::Ptr &sendBuf) { static void onRequestData(const AsyncSenderData::Ptr &data, const std::shared_ptr<HttpSession> &session, const Buffer::Ptr &sendBuf) {
session->_ticker.resetTime(); session->_ticker.resetTime();
if (sendBuf && session->send(sendBuf) != -1) { if (sendBuf && session->send(sendBuf) != -1) {
//文件还未读完，还需要继续发送 // 文件还未读完，还需要继续发送
if (!session->isSocketBusy()) { if (!session->isSocketBusy()) {
//socket还可写，继续请求数据 // socket还可写，继续请求数据
onSocketFlushed(data); onSocketFlushed(data);
} }
return; return;
} }
//文件写完了 // 文件写完了
data->_read_complete = true; data->_read_complete = true;
if (!session->isSocketBusy() && data->_close_when_complete) { if (!session->isSocketBusy() && data->_close_when_complete) {
shutdown(session); shutdown(session);
@ -512,7 +501,7 @@ private:
} }
static void shutdown(const std::shared_ptr<HttpSession> &session) { static void shutdown(const std::shared_ptr<HttpSession> &session) {
if(session){ if (session) {
session->shutdown(SockException(Err_shutdown, StrPrinter << "close connection after send http body completed.")); session->shutdown(SockException(Err_shutdown, StrPrinter << "close connection after send http body completed."));
} }
} }
@ -532,14 +521,14 @@ void HttpSession::sendResponse(int code,
const char *pcContentType, const char *pcContentType,
const HttpSession::KeyValue &header, const HttpSession::KeyValue &header,
const HttpBody::Ptr &body, const HttpBody::Ptr &body,
bool no_content_length ){ bool no_content_length) {
GET_CONFIG(string,charSet,Http::kCharSet); GET_CONFIG(string, charSet, Http::kCharSet);
GET_CONFIG(uint32_t,keepAliveSec,Http::kKeepAliveSecond); GET_CONFIG(uint32_t, keepAliveSec, Http::kKeepAliveSecond);
//body默认为空 // body默认为空
int64_t size = 0; int64_t size = 0;
if (body && body->remainSize()) { if (body && body->remainSize()) {
//有body，获取body大小 // 有body，获取body大小
size = body->remainSize(); size = body->remainSize();
} }
@ -547,7 +536,7 @@ void HttpSession::sendResponse(int code,
// http-flv直播是Keep-Alive类型 // http-flv直播是Keep-Alive类型
bClose = false; bClose = false;
} else if ((size_t)size >= SIZE_MAX || size < 0) { } else if ((size_t)size >= SIZE_MAX || size < 0) {
//不固定长度的body，那么发送完body后应该关闭socket，以便浏览器做下载完毕的判断 // 不固定长度的body，那么发送完body后应该关闭socket，以便浏览器做下载完毕的判断
bClose = true; bClose = true;
} }
@ -563,30 +552,30 @@ void HttpSession::sendResponse(int code,
} }
if (!_origin.empty()) { if (!_origin.empty()) {
//设置跨域 // 设置跨域
headerOut.emplace(kAccessControlAllowOrigin, _origin); headerOut.emplace(kAccessControlAllowOrigin, _origin);
headerOut.emplace(kAccessControlAllowCredentials, "true"); headerOut.emplace(kAccessControlAllowCredentials, "true");
} }
if (!no_content_length && size >= 0 && (size_t)size < SIZE_MAX) { if (!no_content_length && size >= 0 && (size_t)size < SIZE_MAX) {
//文件长度为固定值,且不是http-flv强制设置Content-Length // 文件长度为固定值,且不是http-flv强制设置Content-Length
headerOut[kContentLength] = to_string(size); headerOut[kContentLength] = to_string(size);
} }
if (size && !pcContentType) { if (size && !pcContentType) {
//有body时，设置缺省类型 // 有body时，设置缺省类型
pcContentType = "text/plain"; pcContentType = "text/plain";
} }
if ((size || no_content_length) && pcContentType) { if ((size || no_content_length) && pcContentType) {
//有body时，设置文件类型 // 有body时，设置文件类型
string strContentType = pcContentType; string strContentType = pcContentType;
strContentType += "; charset="; strContentType += "; charset=";
strContentType += charSet; strContentType += charSet;
headerOut.emplace(kContentType, std::move(strContentType)); headerOut.emplace(kContentType, std::move(strContentType));
} }
//发送http头 // 发送http头
string str; string str;
str.reserve(256); str.reserve(256);
str += "HTTP/1.1 "; str += "HTTP/1.1 ";
@ -605,9 +594,9 @@ void HttpSession::sendResponse(int code,
_ticker.resetTime(); _ticker.resetTime();
if (!size) { if (!size) {
//没有body // 没有body
if (bClose) { if (bClose) {
shutdown(SockException(Err_shutdown,StrPrinter << "close connection after send http header completed with status code:" << code)); shutdown(SockException(Err_shutdown, StrPrinter << "close connection after send http header completed with status code:" << code));
} }
return; return;
} }
@ -622,20 +611,20 @@ void HttpSession::sendResponse(int code,
GET_CONFIG(uint32_t, sendBufSize, Http::kSendBufSize); GET_CONFIG(uint32_t, sendBufSize, Http::kSendBufSize);
if (body->remainSize() > sendBufSize) { if (body->remainSize() > sendBufSize) {
//文件下载提升发送性能 // 文件下载提升发送性能
setSocketFlags(); setSocketFlags();
} }
//发送http body // 发送http body
AsyncSenderData::Ptr data = std::make_shared<AsyncSenderData>(static_pointer_cast<HttpSession>(shared_from_this()), body, bClose); AsyncSenderData::Ptr data = std::make_shared<AsyncSenderData>(static_pointer_cast<HttpSession>(shared_from_this()), body, bClose);
getSock()->setOnFlush([data]() { return AsyncSender::onSocketFlushed(data); }); getSock()->setOnFlush([data]() { return AsyncSender::onSocketFlushed(data); });
AsyncSender::onSocketFlushed(data); AsyncSender::onSocketFlushed(data);
} }
string HttpSession::urlDecode(const string &str){ string HttpSession::urlDecode(const string &str) {
auto ret = strCoding::UrlDecode(str); auto ret = strCoding::UrlDecode(str);
#ifdef _WIN32 #ifdef _WIN32
GET_CONFIG(string,charSet,Http::kCharSet); GET_CONFIG(string, charSet, Http::kCharSet);
bool isGb2312 = !strcasecmp(charSet.data(), "gb2312"); bool isGb2312 = !strcasecmp(charSet.data(), "gb2312");
if (isGb2312) { if (isGb2312) {
ret = strCoding::UTF8ToGB2312(ret); ret = strCoding::UTF8ToGB2312(ret);
@ -644,117 +633,117 @@ string HttpSession::urlDecode(const string &str){
return ret; return ret;
} }
void HttpSession::urlDecode(Parser &parser){ void HttpSession::urlDecode(Parser &parser) {
parser.setUrl(urlDecode(parser.Url())); parser.setUrl(urlDecode(parser.Url()));
for(auto &pr : _parser.getUrlArgs()){ for (auto &pr : _parser.getUrlArgs()) {
const_cast<string &>(pr.second) = urlDecode(pr.second); const_cast<string &>(pr.second) = urlDecode(pr.second);
} }
} }
bool HttpSession::emitHttpEvent(bool doInvoke){ bool HttpSession::emitHttpEvent(bool doInvoke) {
bool bClose = !strcasecmp(_parser["Connection"].data(),"close"); bool bClose = !strcasecmp(_parser["Connection"].data(), "close");
/////////////////////异步回复Invoker/////////////////////////////// /////////////////////异步回复Invoker///////////////////////////////
weak_ptr<HttpSession> weak_self = static_pointer_cast<HttpSession>(shared_from_this()); weak_ptr<HttpSession> weak_self = static_pointer_cast<HttpSession>(shared_from_this());
HttpResponseInvoker invoker = [weak_self,bClose](int code, const KeyValue &headerOut, const HttpBody::Ptr &body){ HttpResponseInvoker invoker = [weak_self, bClose](int code, const KeyValue &headerOut, const HttpBody::Ptr &body) {
auto strong_self = weak_self.lock(); auto strong_self = weak_self.lock();
if(!strong_self) { if (!strong_self) {
return; return;
} }
strong_self->async([weak_self, bClose, code, headerOut, body]() { strong_self->async([weak_self, bClose, code, headerOut, body]() {
auto strong_self = weak_self.lock(); auto strong_self = weak_self.lock();
if (!strong_self) { if (!strong_self) {
//本对象已经销毁 // 本对象已经销毁
return; return;
} }
strong_self->sendResponse(code, bClose, nullptr, headerOut, body); strong_self->sendResponse(code, bClose, nullptr, headerOut, body);
}); });
}; };
///////////////////广播HTTP事件/////////////////////////// ///////////////////广播HTTP事件///////////////////////////
bool consumed = false;//该事件是否被消费 bool consumed = false; // 该事件是否被消费
NoticeCenter::Instance().emitEvent(Broadcast::kBroadcastHttpRequest,_parser,invoker,consumed,static_cast<SockInfo &>(*this)); NoticeCenter::Instance().emitEvent(Broadcast::kBroadcastHttpRequest, _parser, invoker, consumed, static_cast<SockInfo &>(*this));
if(!consumed && doInvoke){ if (!consumed && doInvoke) {
//该事件无人消费，所以返回404 // 该事件无人消费，所以返回404
invoker(404,KeyValue(), HttpBody::Ptr()); invoker(404, KeyValue(), HttpBody::Ptr());
} }
return consumed; return consumed;
} }
std::string HttpSession::get_peer_ip() { std::string HttpSession::get_peer_ip() {
GET_CONFIG(string, forwarded_ip_header, Http::kForwardedIpHeader); GET_CONFIG(string, forwarded_ip_header, Http::kForwardedIpHeader);
if(!forwarded_ip_header.empty() && !_parser.getHeader()[forwarded_ip_header].empty()){ if (!forwarded_ip_header.empty() && !_parser.getHeader()[forwarded_ip_header].empty()) {
return _parser.getHeader()[forwarded_ip_header]; return _parser.getHeader()[forwarded_ip_header];
} }
return Session::get_peer_ip(); return Session::get_peer_ip();
} }
void HttpSession::Handle_Req_POST(ssize_t &content_len) { void HttpSession::Handle_Req_POST(ssize_t &content_len) {
GET_CONFIG(size_t,maxReqSize,Http::kMaxReqSize); GET_CONFIG(size_t, maxReqSize, Http::kMaxReqSize);
ssize_t totalContentLen = _parser["Content-Length"].empty() ? -1 : atoll(_parser["Content-Length"].data()); ssize_t totalContentLen = _parser["Content-Length"].empty() ? -1 : atoll(_parser["Content-Length"].data());
if(totalContentLen == 0){ if (totalContentLen == 0) {
//content为空 // content为空
//emitHttpEvent内部会选择是否关闭连接 // emitHttpEvent内部会选择是否关闭连接
emitHttpEvent(true); emitHttpEvent(true);
return; return;
} }
if(totalContentLen > 0 && (size_t)totalContentLen < maxReqSize ){ if (totalContentLen > 0 && (size_t)totalContentLen < maxReqSize) {
//返回固定长度的content // 返回固定长度的content
content_len = totalContentLen; content_len = totalContentLen;
auto parserCopy = _parser; auto parserCopy = _parser;
_contentCallBack = [this,parserCopy](const char *data,size_t len){ _contentCallBack = [this, parserCopy](const char *data, size_t len) {
//恢复http头 // 恢复http头
_parser = parserCopy; _parser = parserCopy;
//设置content // 设置content
_parser.setContent(string(data,len)); _parser.setContent(string(data, len));
//触发http事件，emitHttpEvent内部会选择是否关闭连接 // 触发http事件，emitHttpEvent内部会选择是否关闭连接
emitHttpEvent(true); emitHttpEvent(true);
//清空数据,节省内存 // 清空数据,节省内存
_parser.Clear(); _parser.Clear();
//content已经接收完毕 // content已经接收完毕
return false; return false;
}; };
}else{ } else {
//返回不固定长度的content或者超过长度限制的content // 返回不固定长度的content或者超过长度限制的content
content_len = -1; content_len = -1;
auto parserCopy = _parser; auto parserCopy = _parser;
std::shared_ptr<size_t> recvedContentLen = std::make_shared<size_t>(0); std::shared_ptr<size_t> recvedContentLen = std::make_shared<size_t>(0);
bool bClose = !strcasecmp(_parser["Connection"].data(),"close"); bool bClose = !strcasecmp(_parser["Connection"].data(), "close");
_contentCallBack = [this,parserCopy,totalContentLen,recvedContentLen,bClose](const char *data,size_t len){ _contentCallBack = [this, parserCopy, totalContentLen, recvedContentLen, bClose](const char *data, size_t len) {
*(recvedContentLen) += len; *(recvedContentLen) += len;
if (totalContentLen < 0) { if (totalContentLen < 0) {
//不固定长度的content,源源不断接收数据 // 不固定长度的content,源源不断接收数据
onRecvUnlimitedContent(parserCopy, data, len, SIZE_MAX, *(recvedContentLen)); onRecvUnlimitedContent(parserCopy, data, len, SIZE_MAX, *(recvedContentLen));
return true; return true;
} }
//长度超过限制的content // 长度超过限制的content
onRecvUnlimitedContent(parserCopy,data,len,totalContentLen,*(recvedContentLen)); onRecvUnlimitedContent(parserCopy, data, len, totalContentLen, *(recvedContentLen));
if(*(recvedContentLen) < (size_t)totalContentLen){ if (*(recvedContentLen) < (size_t)totalContentLen) {
//数据还没接收完毕 // 数据还没接收完毕
//_contentCallBack是可持续的，后面还要处理后续content数据 //_contentCallBack是可持续的，后面还要处理后续content数据
return true; return true;
} }
//数据接收完毕 // 数据接收完毕
if(!bClose){ if (!bClose) {
//keep-alive类型连接 // keep-alive类型连接
//content接收完毕，后续都是http header // content接收完毕，后续都是http header
setContentLen(0); setContentLen(0);
//content已经接收完毕 // content已经接收完毕
return false; return false;
} }
//连接类型是close类型，收完content就关闭连接 // 连接类型是close类型，收完content就关闭连接
shutdown(SockException(Err_shutdown,"recv http content completed")); shutdown(SockException(Err_shutdown, "recv http content completed"));
//content已经接收完毕 // content已经接收完毕
return false ; return false;
}; };
} }
//有后续content数据要处理,暂时不关闭连接 // 有后续content数据要处理,暂时不关闭连接
} }
void HttpSession::sendNotFound(bool bClose) { void HttpSession::sendNotFound(bool bClose) {
@ -762,19 +751,19 @@ void HttpSession::sendNotFound(bool bClose) {
sendResponse(404, bClose, "text/html", KeyValue(), std::make_shared<HttpStringBody>(notFound)); sendResponse(404, bClose, "text/html", KeyValue(), std::make_shared<HttpStringBody>(notFound));
} }
void HttpSession::setSocketFlags(){ void HttpSession::setSocketFlags() {
GET_CONFIG(int, mergeWriteMS, General::kMergeWriteMS); GET_CONFIG(int, mergeWriteMS, General::kMergeWriteMS);
if(mergeWriteMS > 0) { if (mergeWriteMS > 0) {
//推流模式下，关闭TCP_NODELAY会增加推流端的延时，但是服务器性能将提高 // 推流模式下，关闭TCP_NODELAY会增加推流端的延时，但是服务器性能将提高
SockUtil::setNoDelay(getSock()->rawFD(), false); SockUtil::setNoDelay(getSock()->rawFD(), false);
//播放模式下，开启MSG_MORE会增加延时，但是能提高发送性能 // 播放模式下，开启MSG_MORE会增加延时，但是能提高发送性能
setSendFlags(SOCKET_DEFAULE_FLAGS | FLAG_MORE); setSendFlags(SOCKET_DEFAULE_FLAGS | FLAG_MORE);
} }
} }
void HttpSession::onWrite(const Buffer::Ptr &buffer, bool flush) { void HttpSession::onWrite(const Buffer::Ptr &buffer, bool flush) {
if(flush){ if (flush) {
//需要flush那么一次刷新缓存 // 需要flush那么一次刷新缓存
HttpSession::setSendFlushFlag(true); HttpSession::setSendFlushFlag(true);
} }
@ -792,18 +781,18 @@ void HttpSession::onWrite(const Buffer::Ptr &buffer, bool flush) {
} }
if (flush) { if (flush) {
//本次刷新缓存后，下次不用刷新缓存 // 本次刷新缓存后，下次不用刷新缓存
HttpSession::setSendFlushFlag(false); HttpSession::setSendFlushFlag(false);
} }
} }
void HttpSession::onWebSocketEncodeData(Buffer::Ptr buffer){ void HttpSession::onWebSocketEncodeData(Buffer::Ptr buffer) {
_total_bytes_usage += buffer->size(); _total_bytes_usage += buffer->size();
send(std::move(buffer)); send(std::move(buffer));
} }
void HttpSession::onWebSocketDecodeComplete(const WebSocketHeader &header_in){ void HttpSession::onWebSocketDecodeComplete(const WebSocketHeader &header_in) {
WebSocketHeader& header = const_cast<WebSocketHeader&>(header_in); WebSocketHeader &header = const_cast<WebSocketHeader &>(header_in);
header._mask_flag = false; header._mask_flag = false;
switch (header._opcode) { switch (header._opcode) {
@ -813,15 +802,15 @@ void HttpSession::onWebSocketDecodeComplete(const WebSocketHeader &header_in){
break; break;
} }
default : break; default: break;
} }
} }
void HttpSession::onDetach() { void HttpSession::onDetach() {
shutdown(SockException(Err_shutdown,"rtmp ring buffer detached")); shutdown(SockException(Err_shutdown, "rtmp ring buffer detached"));
} }
std::shared_ptr<FlvMuxer> HttpSession::getSharedPtr(){ std::shared_ptr<FlvMuxer> HttpSession::getSharedPtr() {
return dynamic_pointer_cast<FlvMuxer>(shared_from_this()); return dynamic_pointer_cast<FlvMuxer>(shared_from_this());
} }