diff --git a/src/Common/MediaSource.cpp b/src/Common/MediaSource.cpp index 0b1edd47..9c4b0f40 100644 --- a/src/Common/MediaSource.cpp +++ b/src/Common/MediaSource.cpp @@ -150,8 +150,8 @@ void MediaInfo::parse(const string &url){ auto pos = steamid.find("?"); if(pos != string::npos){ m_streamid = steamid.substr(0,pos); - auto params = steamid.substr(pos + 1); - m_params = Parser::parseArgs(params); + m_param_strs = steamid.substr(pos + 1); + m_params = Parser::parseArgs(m_param_strs); if(m_params.find(VHOST_KEY) != m_params.end()){ m_vhost = m_params[VHOST_KEY]; } diff --git a/src/Common/MediaSource.h b/src/Common/MediaSource.h index 1ca636f2..9f95a0ee 100644 --- a/src/Common/MediaSource.h +++ b/src/Common/MediaSource.h @@ -83,6 +83,8 @@ public: string m_app; string m_streamid; StrCaseMap m_params; + string m_param_strs; + }; diff --git a/src/Common/config.cpp b/src/Common/config.cpp index d8fc45cd..b43a9e86 100644 --- a/src/Common/config.cpp +++ b/src/Common/config.cpp @@ -47,13 +47,13 @@ void loadIniConfig(){ } ////////////广播名称/////////// namespace Broadcast { -const char kBroadcastMediaPlayed[] = "kBroadcastMediaPlayed"; const char kBroadcastMediaChanged[] = "kBroadcastMediaChanged"; const char kBroadcastRecordMP4[] = "kBroadcastRecordMP4"; const char kBroadcastHttpRequest[] = "kBroadcastHttpRequest"; const char kBroadcastOnGetRtspRealm[] = "kBroadcastOnGetRtspRealm"; const char kBroadcastOnRtspAuth[] = "kBroadcastOnRtspAuth"; - +const char kBroadcastMediaPlayed[] = "kBroadcastMediaPlayed"; +const char kBroadcastRtmpPublish[] = "kBroadcastRtmpPublish"; } //namespace Broadcast //代理失败最大重试次数 diff --git a/src/Common/config.h b/src/Common/config.h index aa24e4d0..b07a1548 100644 --- a/src/Common/config.h +++ b/src/Common/config.h @@ -51,6 +51,7 @@ void loadIniConfig(); #define SERVER_NAME "ZLMediaKit" #define VHOST_KEY "vhost" +#define HTTP_SCHEMA "http" #define RTSP_SCHEMA "rtsp" #define RTMP_SCHEMA "rtmp" #define DEFAULT_VHOST "__defaultVhost__" @@ -59,15 +60,16 @@ void loadIniConfig(); ////////////广播名称/////////// namespace Broadcast { -extern const char kBroadcastMediaPlayed[]; -#define BroadcastMediaPlayedArgs const char *schema,const char *vhost,const char *app,const char *stream +//注册或反注册MediaSource事件广播 extern const char kBroadcastMediaChanged[]; #define BroadcastMediaChangedArgs bool bRegist, const char *schema,const char *vhost,const char *app,const char *stream +//录制mp4文件成功后广播 extern const char kBroadcastRecordMP4[]; #define BroadcastRecordMP4Args const Mp4Info &info +//收到http api请求广播 extern const char kBroadcastHttpRequest[]; #define BroadcastHttpRequestArgs const Parser &parser,HttpSession::HttpResponseInvoker &invoker,bool &consumed @@ -80,6 +82,18 @@ extern const char kBroadcastOnGetRtspRealm[]; extern const char kBroadcastOnRtspAuth[]; #define BroadcastOnRtspAuthArgs const char *user_name,bool must_no_encrypt,const RtspSession::onAuth &invoker + +//鉴权结果回调对象 +typedef std::function AuthInvoker; + +//收到rtmp推流事件广播，通过该事件控制推流鉴权 +extern const char kBroadcastRtmpPublish[]; +#define BroadcastRtmpPublishArgs MediaInfo &args,Broadcast::AuthInvoker &invoker + +//播放rtsp或rtmp事件广播，通过该事件控制播放鉴权 +extern const char kBroadcastMediaPlayed[]; +#define BroadcastMediaPlayedArgs MediaInfo &args,Broadcast::AuthInvoker &invoker + } //namespace Broadcast //代理失败最大重试次数 diff --git a/src/Http/HttpSession.cpp b/src/Http/HttpSession.cpp index 59826ce8..a3e8a669 100644 --- a/src/Http/HttpSession.cpp +++ b/src/Http/HttpSession.cpp @@ -171,7 +171,8 @@ void HttpSession::onManager() { shutdown(); } } - +//http-flv 链接格式:http://vhost-url:port/app/streamid.flv?key1=value1&key2=value2 +//如果url(除去?以及后面的参数)后缀是.flv,那么表明该url是一个http-flv直播。 inline bool HttpSession::checkLiveFlvStream(){ auto pos = strrchr(m_parser.Url().data(),'.'); if(!pos){ @@ -182,87 +183,118 @@ inline bool HttpSession::checkLiveFlvStream(){ //未找到".flv"后缀 return false; } - auto fullUrl = string("http://") + m_parser["Host"] + FindField(m_parser.Url().data(),NULL,pos); + //拼接成完整url + auto fullUrl = string(HTTP_SCHEMA) + "://" + m_parser["Host"] + m_parser.FullUrl(); MediaInfo info(fullUrl); - if(!m_parser.getUrlArgs()[VHOST_KEY].empty()){ - info.m_vhost = m_parser.getUrlArgs()[VHOST_KEY]; - } + info.m_streamid.erase(info.m_streamid.size() - 4);//去除.flv后缀 auto mediaSrc = dynamic_pointer_cast(MediaSource::find(RTMP_SCHEMA,info.m_vhost,info.m_app,info.m_streamid)); if(!mediaSrc){ //该rtmp源不存在 - return false; + sendNotFound(true); + shutdown(); + return true; } if(!mediaSrc->ready()){ //未准备好 - return false; + sendNotFound(true); + shutdown(); + return true; } - //找到rtmp源，发送http头，负载后续发送 - sendResponse("200 OK", makeHttpHeader(false,0,get_mime_type(m_parser.Url().data())), ""); - //发送flv文件头 - char flv_file_header[] = "FLV\x1\x5\x0\x0\x0\x9"; // have audio and have video - bool is_have_audio = false,is_have_video = false; - mediaSrc->getConfigFrame([&](const RtmpPacket::Ptr &pkt){ - if(pkt->typeId == MSG_VIDEO){ - is_have_video = true; - } - if(pkt->typeId == MSG_AUDIO){ - is_have_audio = true; - } - }); + auto onRes = [this,mediaSrc](bool authSuccess){ + if(!authSuccess){ + string status = "401 Unauthorized"; + sendResponse(status.data(), makeHttpHeader(true,status.size()),status); + shutdown(); + return ; + } + //找到rtmp源，发送http头，负载后续发送 + sendResponse("200 OK", makeHttpHeader(false,0,get_mime_type(".flv")), ""); + //发送flv文件头 + char flv_file_header[] = "FLV\x1\x5\x0\x0\x0\x9"; // have audio and have video + bool is_have_audio = false,is_have_video = false; - if (is_have_audio && is_have_video) { - flv_file_header[4] = 0x05; - } else if (is_have_audio && !is_have_video) { - flv_file_header[4] = 0x04; - } else if (!is_have_audio && is_have_video) { - flv_file_header[4] = 0x01; - } else { - flv_file_header[4] = 0x00; - } - //send flv header - send(flv_file_header, sizeof(flv_file_header) - 1); - //send metadata - AMFEncoder invoke; - invoke << "onMetaData" << mediaSrc->getMetaData(); - sendRtmp(MSG_DATA, invoke.data(), 0); - //send config frame - mediaSrc->getConfigFrame([&](const RtmpPacket::Ptr &pkt){ - onSendMedia(pkt); - }); + mediaSrc->getConfigFrame([&](const RtmpPacket::Ptr &pkt){ + if(pkt->typeId == MSG_VIDEO){ + is_have_video = true; + } + if(pkt->typeId == MSG_AUDIO){ + is_have_audio = true; + } + }); - //开始发送rtmp负载 - m_pRingReader = mediaSrc->getRing()->attach(); - weak_ptr weakSelf = dynamic_pointer_cast(shared_from_this()); - m_pRingReader->setReadCB([weakSelf](const RtmpPacket::Ptr &pkt){ - auto strongSelf = weakSelf.lock(); - if(!strongSelf) { - return; - } - strongSelf->async([pkt,weakSelf](){ - auto strongSelf = weakSelf.lock(); - if(!strongSelf) { - return; - } - strongSelf->onSendMedia(pkt); - }); - }); - m_pRingReader->setDetachCB([weakSelf](){ - auto strongSelf = weakSelf.lock(); - if(!strongSelf) { - return; - } - strongSelf->async_first([weakSelf](){ - auto strongSelf = weakSelf.lock(); - if(!strongSelf) { - return; - } - strongSelf->shutdown(); - }); - }); - return true; + if (is_have_audio && is_have_video) { + flv_file_header[4] = 0x05; + } else if (is_have_audio && !is_have_video) { + flv_file_header[4] = 0x04; + } else if (!is_have_audio && is_have_video) { + flv_file_header[4] = 0x01; + } else { + flv_file_header[4] = 0x00; + } + //send flv header + send(flv_file_header, sizeof(flv_file_header) - 1); + //send metadata + AMFEncoder invoke; + invoke << "onMetaData" << mediaSrc->getMetaData(); + sendRtmp(MSG_DATA, invoke.data(), 0); + //send config frame + mediaSrc->getConfigFrame([&](const RtmpPacket::Ptr &pkt){ + onSendMedia(pkt); + }); + //开始发送rtmp负载 + m_pRingReader = mediaSrc->getRing()->attach(); + weak_ptr weakSelf = dynamic_pointer_cast(shared_from_this()); + m_pRingReader->setReadCB([weakSelf](const RtmpPacket::Ptr &pkt){ + auto strongSelf = weakSelf.lock(); + if(!strongSelf) { + return; + } + strongSelf->async([pkt,weakSelf](){ + auto strongSelf = weakSelf.lock(); + if(!strongSelf) { + return; + } + strongSelf->onSendMedia(pkt); + }); + }); + m_pRingReader->setDetachCB([weakSelf](){ + auto strongSelf = weakSelf.lock(); + if(!strongSelf) { + return; + } + strongSelf->async_first([weakSelf](){ + auto strongSelf = weakSelf.lock(); + if(!strongSelf) { + return; + } + strongSelf->shutdown(); + }); + }); + }; + + weak_ptr weakSelf = dynamic_pointer_cast(shared_from_this()); + Broadcast::AuthInvoker invoker = [weakSelf,onRes](bool authSuccess){ + auto strongSelf = weakSelf.lock(); + if(!strongSelf){ + return; + } + strongSelf->async([weakSelf,onRes,authSuccess](){ + auto strongSelf = weakSelf.lock(); + if(!strongSelf){ + return; + } + onRes(authSuccess); + }); + }; + auto flag = NoticeCenter::Instance().emitEvent(Broadcast::kBroadcastMediaPlayed,info,invoker); + if(!flag){ + //该事件无人监听,默认不鉴权 + onRes(true); + } + return true; } inline HttpSession::HttpCode HttpSession::Handle_Req_GET() { //先看看该http事件是否被拦截 @@ -274,11 +306,9 @@ inline HttpSession::HttpCode HttpSession::Handle_Req_GET() { } //事件未被拦截，则认为是http下载请求 - auto fullUrl = string("http://") + m_parser["Host"] + m_parser.Url(); + auto fullUrl = string(HTTP_SCHEMA) + "://" + m_parser["Host"] + m_parser.FullUrl(); MediaInfo info(fullUrl); - if(!m_parser.getUrlArgs()[VHOST_KEY].empty()){ - info.m_vhost = m_parser.getUrlArgs()[VHOST_KEY]; - } + string strFile = m_strPath + "/" + info.m_vhost + m_parser.Url(); /////////////HTTP连接是否需要被关闭//////////////// static uint32_t reqCnt = mINI::Instance()[Config::Http::kMaxReqCount].as(); diff --git a/src/Rtmp/RtmpSession.cpp b/src/Rtmp/RtmpSession.cpp index 359880a5..ab75bc0c 100644 --- a/src/Rtmp/RtmpSession.cpp +++ b/src/Rtmp/RtmpSession.cpp @@ -101,7 +101,7 @@ void RtmpSession::onCmd_connect(AMFDecoder &dec) { m_strTcUrl = params["tcUrl"].as_string(); if(m_strTcUrl.empty()){ //defaultVhost:默认vhost - m_strTcUrl = "rtmp://127.0.0.1/" + m_mediaInfo.m_app; + m_strTcUrl = string(RTMP_SCHEMA) + "://" + DEFAULT_VHOST + "/" + m_mediaInfo.m_app; } bool ok = true; //(app == APP_NAME); AMFValue version(AMF_OBJECT); @@ -129,26 +129,54 @@ void RtmpSession::onCmd_createStream(AMFDecoder &dec) { void RtmpSession::onCmd_publish(AMFDecoder &dec) { dec.load();/* NULL */ m_mediaInfo.parse(m_strTcUrl + "/" + dec.load()); - auto src = dynamic_pointer_cast(MediaSource::find(RTMP_SCHEMA, - m_mediaInfo.m_vhost, - m_mediaInfo.m_app, - m_mediaInfo.m_streamid, - false)); - bool ok = (!src && !m_pPublisherSrc); - AMFValue status(AMF_OBJECT); - status.set("level", ok ? "status" : "error"); - status.set("code", ok ? "NetStream.Publish.Start" : "NetStream.Publish.BadName"); - status.set("description", ok ? "Started publishing stream." : "Already publishing."); - status.set("clientid", "0"); - sendReply("onStatus", nullptr, status); - if (!ok) { - throw std::runtime_error( StrPrinter << "Already publishing:" - << m_mediaInfo.m_vhost << " " - << m_mediaInfo.m_app << " " - << m_mediaInfo.m_streamid << endl); - } - m_bPublisherSrcRegisted = false; - m_pPublisherSrc.reset(new RtmpToRtspMediaSource(m_mediaInfo.m_vhost,m_mediaInfo.m_app,m_mediaInfo.m_streamid)); + + auto onRes = [this](bool authSuccess){ + auto src = dynamic_pointer_cast(MediaSource::find(RTMP_SCHEMA, + m_mediaInfo.m_vhost, + m_mediaInfo.m_app, + m_mediaInfo.m_streamid, + false)); + bool ok = (!src && !m_pPublisherSrc && authSuccess); + AMFValue status(AMF_OBJECT); + status.set("level", ok ? "status" : "error"); + status.set("code", ok ? "NetStream.Publish.Start" : (authSuccess ? "NetStream.Publish.BadName" : "NetStream.Publish.BadAuth")); + status.set("description", ok ? "Started publishing stream." : (authSuccess ? "Already publishing." : "Auth failed")); + status.set("clientid", "0"); + sendReply("onStatus", nullptr, status); + if (!ok) { + WarnL << "onPublish:" + << (authSuccess ? "Already publishing:" : "Auth failed:") + << m_mediaInfo.m_vhost << " " + << m_mediaInfo.m_app << " " + << m_mediaInfo.m_streamid << endl; + shutdown(); + return; + } + m_bPublisherSrcRegisted = false; + m_pPublisherSrc.reset(new RtmpToRtspMediaSource(m_mediaInfo.m_vhost,m_mediaInfo.m_app,m_mediaInfo.m_streamid)); + }; + + weak_ptr weakSelf = dynamic_pointer_cast(shared_from_this()); + Broadcast::AuthInvoker invoker = [weakSelf,onRes](bool success){ + auto strongSelf = weakSelf.lock(); + if(!strongSelf){ + return; + } + strongSelf->async([weakSelf,onRes,success](){ + auto strongSelf = weakSelf.lock(); + if(!strongSelf){ + return; + } + onRes(success); + }); + }; + auto flag = NoticeCenter::Instance().emitEvent(Config::Broadcast::kBroadcastRtmpPublish, + m_mediaInfo, + invoker); + if(!flag){ + //该事件无人监听，默认鉴权成功 + onRes(true); + } } void RtmpSession::onCmd_deleteStream(AMFDecoder &dec) { @@ -161,112 +189,134 @@ void RtmpSession::onCmd_deleteStream(AMFDecoder &dec) { } void RtmpSession::doPlay(AMFDecoder &dec){ - m_mediaInfo.parse(m_strTcUrl + "/" + dec.load()); - auto src = dynamic_pointer_cast(MediaSource::find(RTMP_SCHEMA, - m_mediaInfo.m_vhost, - m_mediaInfo.m_app, - m_mediaInfo.m_streamid, - true)); - bool ok = (src.operator bool()); - ok = ok && src->ready(); - //stream begin - sendUserControl(CONTROL_STREAM_BEGIN, STREAM_MEDIA); + auto onRes = [this](bool authSuccess) { + auto src = dynamic_pointer_cast(MediaSource::find(RTMP_SCHEMA, + m_mediaInfo.m_vhost, + m_mediaInfo.m_app, + m_mediaInfo.m_streamid, + true)); + bool ok = (src.operator bool() && authSuccess); + if(ok){ + ok = ok && src->ready(); + } + if (ok) { + //stream begin + sendUserControl(CONTROL_STREAM_BEGIN, STREAM_MEDIA); + } + // onStatus(NetStream.Play.Reset) + AMFValue status(AMF_OBJECT); + status.set("level", ok ? "status" : "error"); + status.set("code", ok ? "NetStream.Play.Reset" : (authSuccess ? "NetStream.Play.StreamNotFound" : "NetStream.Play.BadAuth")); + status.set("description", ok ? "Resetting and playing." : (authSuccess ? "No such stream." : "Auth failed")); + status.set("details", m_mediaInfo.m_streamid); + status.set("clientid", "0"); + sendReply("onStatus", nullptr, status); + if (!ok) { + WarnL << "onPlayed:" + << (authSuccess ? "No such stream:" : "Auth failed:") + << m_mediaInfo.m_vhost << " " + << m_mediaInfo.m_app << " " + << m_mediaInfo.m_streamid + << endl; + shutdown(); + return; + } - // onStatus(NetStream.Play.Reset) - AMFValue status(AMF_OBJECT); - status.set("level", ok ? "status" : "error"); - status.set("code", ok ? "NetStream.Play.Reset" : "NetStream.Play.StreamNotFound"); - status.set("description", ok ? "Resetting and playing." : "No such stream."); - status.set("details", m_mediaInfo.m_streamid); - status.set("clientid", "0"); - sendReply("onStatus", nullptr, status); - if (!ok) { - throw std::runtime_error( StrPrinter << "No such stream:" - << m_mediaInfo.m_vhost << " " - << m_mediaInfo.m_app << " " - << m_mediaInfo.m_streamid - << endl); - } + // onStatus(NetStream.Play.Start) + status.clear(); + status.set("level", "status"); + status.set("code", "NetStream.Play.Start"); + status.set("description", "Started playing."); + status.set("details", m_mediaInfo.m_streamid); + status.set("clientid", "0"); + sendReply("onStatus", nullptr, status); - // onStatus(NetStream.Play.Start) - status.clear(); - status.set("level", "status"); - status.set("code", "NetStream.Play.Start"); - status.set("description", "Started playing."); - status.set("details", m_mediaInfo.m_streamid); - status.set("clientid", "0"); - sendReply("onStatus", nullptr, status); + // |RtmpSampleAccess(true, true) + AMFEncoder invoke; + invoke << "|RtmpSampleAccess" << true << true; + sendResponse(MSG_DATA, invoke.data()); - // |RtmpSampleAccess(true, true) - AMFEncoder invoke; - invoke << "|RtmpSampleAccess" << true << true; - sendResponse(MSG_DATA, invoke.data()); + //onStatus(NetStream.Data.Start) + invoke.clear(); + AMFValue obj(AMF_OBJECT); + obj.set("code", "NetStream.Data.Start"); + invoke << "onStatus" << obj; + sendResponse(MSG_DATA, invoke.data()); - //onStatus(NetStream.Data.Start) - invoke.clear(); - AMFValue obj(AMF_OBJECT); - obj.set("code","NetStream.Data.Start"); - invoke << "onStatus" << obj; - sendResponse(MSG_DATA, invoke.data()); + //onStatus(NetStream.Play.PublishNotify) + status.clear(); + status.set("level", "status"); + status.set("code", "NetStream.Play.PublishNotify"); + status.set("description", "Now published."); + status.set("details", m_mediaInfo.m_streamid); + status.set("clientid", "0"); + sendReply("onStatus", nullptr, status); - //onStatus(NetStream.Play.PublishNotify) - status.clear(); - status.set("level", "status"); - status.set("code", "NetStream.Play.PublishNotify"); - status.set("description", "Now published."); - status.set("details", m_mediaInfo.m_streamid); - status.set("clientid", "0"); - sendReply("onStatus", nullptr, status); + // onMetaData + invoke.clear(); + invoke << "onMetaData" << src->getMetaData(); + sendResponse(MSG_DATA, invoke.data()); - // onMetaData - invoke.clear(); - invoke << "onMetaData" << src->getMetaData(); - sendResponse(MSG_DATA, invoke.data()); + src->getConfigFrame([&](const RtmpPacket::Ptr &pkt) { + //DebugL<<"send initial frame"; + onSendMedia(pkt); + }); - src->getConfigFrame([&](const RtmpPacket::Ptr &pkt) { - //DebugL<<"send initial frame"; - onSendMedia(pkt); - }); + m_pRingReader = src->getRing()->attach(); + weak_ptr weakSelf = dynamic_pointer_cast(shared_from_this()); + SockUtil::setNoDelay(sock->rawFD(), false); + m_pRingReader->setReadCB([weakSelf](const RtmpPacket::Ptr &pkt) { + auto strongSelf = weakSelf.lock(); + if (!strongSelf) { + return; + } + strongSelf->async([weakSelf, pkt]() { + auto strongSelf = weakSelf.lock(); + if (!strongSelf) { + return; + } + strongSelf->onSendMedia(pkt); + }); + }); + m_pRingReader->setDetachCB([weakSelf]() { + auto strongSelf = weakSelf.lock(); + if (!strongSelf) { + return; + } + strongSelf->safeShutdown(); + }); + m_pPlayerSrc = src; + if (src->getRing()->readerCount() == 1) { + src->seekTo(0); + } + }; - m_pRingReader = src->getRing()->attach(); - weak_ptr weakSelf = dynamic_pointer_cast(shared_from_this()); - SockUtil::setNoDelay(sock->rawFD(), false); - m_pRingReader->setReadCB([weakSelf](const RtmpPacket::Ptr &pkt){ - auto strongSelf = weakSelf.lock(); - if(!strongSelf) { - return; - } - strongSelf->async([weakSelf,pkt]() { - auto strongSelf = weakSelf.lock(); - if(!strongSelf) { - return; - } - strongSelf->onSendMedia(pkt); - }); - }); - m_pRingReader->setDetachCB([weakSelf]() { - auto strongSelf = weakSelf.lock(); - if(!strongSelf) { - return; - } - strongSelf->safeShutdown(); - }); - m_pPlayerSrc = src; - if(src->getRing()->readerCount() == 1){ - src->seekTo(0); - } - - NoticeCenter::Instance().emitEvent(Broadcast::kBroadcastMediaPlayed, - RTMP_SCHEMA, - m_mediaInfo.m_vhost.data(), - m_mediaInfo.m_app.data(), - m_mediaInfo.m_streamid.data()); + weak_ptr weakSelf = dynamic_pointer_cast(shared_from_this()); + Broadcast::AuthInvoker invoker = [weakSelf,onRes](bool authSuccess){ + auto strongSelf = weakSelf.lock(); + if(!strongSelf){ + return; + } + strongSelf->async([weakSelf,onRes,authSuccess](){ + auto strongSelf = weakSelf.lock(); + if(!strongSelf){ + return; + } + onRes(authSuccess); + }); + }; + auto flag = NoticeCenter::Instance().emitEvent(Broadcast::kBroadcastMediaPlayed,m_mediaInfo,invoker); + if(!flag){ + //该事件无人监听,默认不鉴权 + onRes(true); + } } void RtmpSession::onCmd_play2(AMFDecoder &dec) { doPlay(dec); } void RtmpSession::onCmd_play(AMFDecoder &dec) { dec.load();/* NULL */ + m_mediaInfo.parse(m_strTcUrl + "/" + dec.load()); doPlay(dec); } diff --git a/src/Rtsp/Rtsp.h b/src/Rtsp/Rtsp.h index e50731ef..9b9f41b7 100644 --- a/src/Rtsp/Rtsp.h +++ b/src/Rtsp/Rtsp.h @@ -101,15 +101,15 @@ public: } if (start == buf) { m_strMethod = FindField(line.c_str(), NULL, " "); - auto full_url = FindField(line.c_str(), " ", " "); - auto args_pos = full_url.find('?'); + m_strFullUrl = FindField(line.c_str(), " ", " "); + auto args_pos = m_strFullUrl.find('?'); if(args_pos != string::npos){ - m_strUrl = full_url.substr(0,args_pos); - m_mapUrlArgs = parseArgs(full_url.substr(args_pos + 1 )); + m_strUrl = m_strFullUrl.substr(0,args_pos); + m_mapUrlArgs = parseArgs(m_strFullUrl.substr(args_pos + 1 )); }else{ - m_strUrl = full_url; + m_strUrl = m_strFullUrl; } - m_strTail = FindField(line.c_str(), (full_url + " ").c_str(), NULL); + m_strTail = FindField(line.c_str(), (m_strFullUrl + " ").c_str(), NULL); } else { auto field = FindField(line.c_str(), NULL, ": "); auto value = FindField(line.c_str(), ": ", NULL); @@ -132,6 +132,10 @@ public: //rtsp url return m_strUrl; } + const string& FullUrl() const { + //rtsp url with args + return m_strFullUrl; + } const string& Tail() const { //RTSP/1.0 return m_strTail; @@ -150,6 +154,7 @@ public: void Clear() { m_strMethod.clear(); m_strUrl.clear(); + m_strFullUrl.clear(); m_strTail.clear(); m_strContent.clear(); m_mapValues.clear(); @@ -189,6 +194,7 @@ private: string m_strTail; string m_strContent; string m_strNull; + string m_strFullUrl; mutable StrCaseMap m_mapValues; mutable StrCaseMap m_mapUrlArgs; }; diff --git a/src/Rtsp/RtspSession.cpp b/src/Rtsp/RtspSession.cpp index 47993845..4a0a9020 100644 --- a/src/Rtsp/RtspSession.cpp +++ b/src/Rtsp/RtspSession.cpp @@ -193,10 +193,7 @@ bool RtspSession::handleReq_Describe() { { //解析url获取媒体名称 m_strUrl = m_parser.Url(); - m_mediaInfo.parse(m_strUrl); - if(!m_parser.getUrlArgs()[VHOST_KEY].empty()){ - m_mediaInfo.m_vhost = m_parser.getUrlArgs()[VHOST_KEY]; - } + m_mediaInfo.parse(m_parser.FullUrl()); } if (!findStream()) { @@ -644,82 +641,113 @@ bool RtspSession::handleReq_Play() { send_SessionNotFound(); return false; } - - if(m_pRtpReader){ - weak_ptr weakSelf = dynamic_pointer_cast(shared_from_this()); - SockUtil::setNoDelay(m_pSender->rawFD(), false); - m_pRtpReader->setReadCB([weakSelf](const RtpPacket::Ptr &pack) { - auto strongSelf = weakSelf.lock(); - if(!strongSelf) { - return; - } - strongSelf->async([weakSelf,pack](){ - auto strongSelf = weakSelf.lock(); - if(!strongSelf) { - return; - } - strongSelf->sendRtpPacket(pack); - }); - - }); - } - - auto pMediaSrc = m_pMediaSrc.lock(); - uint32_t iStamp = 0; - if(pMediaSrc){ - auto strRange = m_parser["Range"]; - if (strRange.size() && !m_bFirstPlay) { - auto strStart = FindField(strRange.data(), "npt=", "-"); - if (strStart == "now") { - strStart = "0"; - } - auto iStartTime = atof(strStart.data()); - InfoL << "rtsp seekTo:" << iStartTime; - pMediaSrc->seekTo(iStartTime * 1000); - iStamp = pMediaSrc->getStamp(); - }else if(pMediaSrc->getRing()->readerCount() == 1){ - //第一个消费者 - pMediaSrc->seekTo(0); - iStamp = 0; - }else{ - iStamp = pMediaSrc->getStamp(); + auto onRes = [this](bool authSuccess){ + char response[2 * 1024]; + m_pcBuf = response; + if(!authSuccess && m_bFirstPlay){ + //第一次play是播放，否则是恢复播放。只对播放鉴权 + int n = sprintf(m_pcBuf, + "RTSP/1.0 401 Unauthorized\r\n" + "CSeq: %d\r\n" + "Server: %s-%0.2f(build in %s)\r\n" + "%s\r\n", + m_iCseq, SERVER_NAME, RTSP_VERSION, RTSP_BUILDTIME, dateHeader().data()); + send(m_pcBuf,n); + shutdown(); + return; } - for (unsigned int i = 0; i < m_uiTrackCnt; i++) { - auto &track = m_aTrackInfo[i]; - track.ssrc = pMediaSrc->getSsrc(track.trackId); - track.seq = pMediaSrc->getSeqence(track.trackId); - track.timeStamp = pMediaSrc->getTimestamp(track.trackId); - } - } - m_bFirstPlay = false; - int iLen = sprintf(m_pcBuf, "RTSP/1.0 200 OK\r\n" - "CSeq: %d\r\n" - "Server: %s-%0.2f(build in %s)\r\n" - "%s" - "Session: %s\r\n" - "Range: npt=%.2f-\r\n" - "RTP-Info: ", m_iCseq, SERVER_NAME, RTSP_VERSION, RTSP_BUILDTIME, - dateHeader().data(), m_strSession.data(),iStamp/1000.0); + if(m_pRtpReader){ + weak_ptr weakSelf = dynamic_pointer_cast(shared_from_this()); + SockUtil::setNoDelay(m_pSender->rawFD(), false); + m_pRtpReader->setReadCB([weakSelf](const RtpPacket::Ptr &pack) { + auto strongSelf = weakSelf.lock(); + if(!strongSelf) { + return; + } + strongSelf->async([weakSelf,pack](){ + auto strongSelf = weakSelf.lock(); + if(!strongSelf) { + return; + } + strongSelf->sendRtpPacket(pack); + }); - for (unsigned int i = 0; i < m_uiTrackCnt; i++) { - auto &track = m_aTrackInfo[i]; - if (track.inited == false) { - //还有track没有setup - return false; - } - iLen += sprintf(m_pcBuf + iLen, "url=%s/%s%d;seq=%d;rtptime=%u,", - m_strUrl.data(), track.trackStyle.data(), - track.trackId, track.seq,track.timeStamp); - } - iLen -= 1; - (m_pcBuf)[iLen] = '\0'; - iLen += sprintf(m_pcBuf + iLen, "\r\n\r\n"); - send(m_pcBuf, iLen); - NoticeCenter::Instance().emitEvent(Broadcast::kBroadcastMediaPlayed, - RTSP_SCHEMA, - m_mediaInfo.m_vhost.data(), - m_mediaInfo.m_app.data(), - m_mediaInfo.m_streamid.data()); + }); + } + + auto pMediaSrc = m_pMediaSrc.lock(); + uint32_t iStamp = 0; + if(pMediaSrc){ + auto strRange = m_parser["Range"]; + if (strRange.size() && !m_bFirstPlay) { + auto strStart = FindField(strRange.data(), "npt=", "-"); + if (strStart == "now") { + strStart = "0"; + } + auto iStartTime = atof(strStart.data()); + InfoL << "rtsp seekTo:" << iStartTime; + pMediaSrc->seekTo(iStartTime * 1000); + iStamp = pMediaSrc->getStamp(); + }else if(pMediaSrc->getRing()->readerCount() == 1){ + //第一个消费者 + pMediaSrc->seekTo(0); + iStamp = 0; + }else{ + iStamp = pMediaSrc->getStamp(); + } + for (unsigned int i = 0; i < m_uiTrackCnt; i++) { + auto &track = m_aTrackInfo[i]; + track.ssrc = pMediaSrc->getSsrc(track.trackId); + track.seq = pMediaSrc->getSeqence(track.trackId); + track.timeStamp = pMediaSrc->getTimestamp(track.trackId); + } + } + m_bFirstPlay = false; + int iLen = sprintf(m_pcBuf, "RTSP/1.0 200 OK\r\n" + "CSeq: %d\r\n" + "Server: %s-%0.2f(build in %s)\r\n" + "%s" + "Session: %s\r\n" + "Range: npt=%.2f-\r\n" + "RTP-Info: ", m_iCseq, SERVER_NAME, RTSP_VERSION, RTSP_BUILDTIME, + dateHeader().data(), m_strSession.data(),iStamp/1000.0); + + for (unsigned int i = 0; i < m_uiTrackCnt; i++) { + auto &track = m_aTrackInfo[i]; + if (track.inited == false) { + //还有track没有setup + shutdown(); + return; + } + iLen += sprintf(m_pcBuf + iLen, "url=%s/%s%d;seq=%d;rtptime=%u,", + m_strUrl.data(), track.trackStyle.data(), + track.trackId, track.seq,track.timeStamp); + } + iLen -= 1; + (m_pcBuf)[iLen] = '\0'; + iLen += sprintf(m_pcBuf + iLen, "\r\n\r\n"); + send(m_pcBuf, iLen); + }; + + weak_ptr weakSelf = dynamic_pointer_cast(shared_from_this()); + Broadcast::AuthInvoker invoker = [weakSelf,onRes](bool authSuccess){ + auto strongSelf = weakSelf.lock(); + if(!strongSelf){ + return; + } + strongSelf->async([weakSelf,onRes,authSuccess](){ + auto strongSelf = weakSelf.lock(); + if(!strongSelf){ + return; + } + onRes(authSuccess); + }); + }; + auto flag = NoticeCenter::Instance().emitEvent(Broadcast::kBroadcastMediaPlayed,m_mediaInfo,invoker); + if(!flag){ + //该事件无人监听,默认不鉴权 + onRes(true); + } return true; } diff --git a/tests/test_server.cpp b/tests/test_server.cpp index 6cc1bbf9..fa769386 100644 --- a/tests/test_server.cpp +++ b/tests/test_server.cpp @@ -107,6 +107,22 @@ static onceToken s_token([](){ invoker(false,user); }); }); + + + NoticeCenter::Instance().addListener(nullptr,Config::Broadcast::kBroadcastRtmpPublish,[](BroadcastRtmpPublishArgs){ + InfoL << args.m_vhost << " " << args.m_app << " " << args.m_streamid << " " << args.m_param_strs ; + EventPoller::Instance().async([invoker](){ + invoker(true); + }); + }); + + NoticeCenter::Instance().addListener(nullptr,Config::Broadcast::kBroadcastMediaPlayed,[](BroadcastMediaPlayedArgs){ + InfoL << args.m_schema << " " << args.m_vhost << " " << args.m_app << " " << args.m_streamid << " " << args.m_param_strs ; + EventPoller::Instance().async([invoker](){ + invoker(true); + }); + }); + }, nullptr); int main(int argc,char *argv[]){