精简等待监听流异步注册后回复事件相关代码

src/Common/MediaSource.cpp

@@ -30,6 +30,7 @@
 #include "Util/util.h"
 #include "Rtsp/Rtsp.h"
 #include "Network/sockutil.h"
+#include "Network/TcpSession.h"
 
 using namespace toolkit;
 
@@ -38,6 +39,67 @@ namespace mediakit {
 recursive_mutex MediaSource::g_mtxMediaSrc;
 MediaSource::SchemaVhostAppStreamMap MediaSource::g_mapMediaSrc;
 
+
+void MediaSource::findAsync(const MediaInfo &info,
+                            const std::shared_ptr<TcpSession> &session,
+                            bool retry,
+                            int maxWaitMs,
+                            const function<void(const MediaSource::Ptr &src)> &cb){
+
+    auto src = MediaSource::find(info._schema,
+                                 info._vhost,
+                                 info._app,
+                                 info._streamid,
+                                 true);
+    if(src || !retry){
+        cb(src);
+        return;
+    }
+
+    void *listener_tag = session.get();
+    weak_ptr<TcpSession> weakSession = session;
+    //广播未找到流,此时可以立即去拉流，这样还来得及
+    NoticeCenter::Instance().emitEvent(Broadcast::kBroadcastNotFoundStream,info,session);
+
+    //若干秒后执行等待媒体注册超时回调
+    auto onRegistTimeout = session->getPoller()->doDelayTask(maxWaitMs,[cb,listener_tag](){
+        //取消监听该事件
+        NoticeCenter::Instance().delListener(listener_tag,Broadcast::kBroadcastMediaChanged);
+        cb(nullptr);
+        return 0;
+    });
+
+    auto onRegist = [listener_tag,weakSession,info,cb,maxWaitMs,onRegistTimeout](BroadcastMediaChangedArgs) {
+        if(!bRegist || schema != info._schema || vhost != info._vhost || app != info._app ||stream != info._streamid){
+            //不是自己感兴趣的事件，忽略之
+            return;
+        }
+
+        //取消延时任务，防止多次回调
+        onRegistTimeout->cancel();
+
+        //播发器请求的流终于注册上了
+        auto strongSession = weakSession.lock();
+        if(!strongSession) {
+            return;
+        }
+
+        //切换到自己的线程再回复
+        strongSession->async([listener_tag,weakSession,info,cb,maxWaitMs](){
+            auto strongSession = weakSession.lock();
+            if(!strongSession) {
+                return;
+            }
+            DebugL << "收到媒体注册事件,回复播放器:" << info._schema << "/" << info._vhost << "/" << info._app << "/" << info._streamid;
+            //再找一遍媒体源，一般能找到
+            findAsync(info,strongSession,false,maxWaitMs,cb);
+            //取消事件监听
+            NoticeCenter::Instance().delListener(listener_tag,Broadcast::kBroadcastMediaChanged);
+        }, false);
+    };
+    //监听媒体注册事件
+    NoticeCenter::Instance().addListener(listener_tag, Broadcast::kBroadcastMediaChanged, onRegist);
+}
 MediaSource::Ptr MediaSource::find(
         const string &schema,
         const string &vhost_tmp,
@@ -69,7 +131,6 @@ MediaSource::Ptr MediaSource::find(
         ret = MediaReader::onMakeMediaSource(schema, vhost,app,id);
     }
     return ret;
-
 }
 void MediaSource::regist() {
     //注册该源，注册后服务器才能找到该源

src/Common/MediaSource.h

@@ -42,6 +42,10 @@
 using namespace std;
 using namespace toolkit;
 
+namespace toolkit{
+    class TcpSession;
+}//namespace toolkit
+
 namespace mediakit {
 
 class MediaSource;
@@ -90,7 +94,6 @@ public:
     string _param_strs;
 };
 
-
 class MediaSource: public enable_shared_from_this<MediaSource> {
 public:
     typedef std::shared_ptr<MediaSource> Ptr;
@@ -122,6 +125,12 @@ public:
                     const string &id,
                     bool bMake = true) ;
 
+    static void findAsync(const MediaInfo &info,
+                          const std::shared_ptr<TcpSession> &session,
+                          bool retry,
+                          int maxWaitMs,
+                          const function<void(const MediaSource::Ptr &src)> &cb);
+
     const string& getSchema() const {
         return _strSchema;
     }

src/Rtmp/RtmpSession.cpp

@@ -43,10 +43,6 @@ RtmpSession::RtmpSession(const Socket::Ptr &pSock) : TcpSession(pSock) {
 
 RtmpSession::~RtmpSession() {
     DebugL << get_peer_ip();
-    if(_delayTask){
-        _delayTask();
-        _delayTask = nullptr;
-    }
 }
 
 void RtmpSession::onError(const SockException& err) {
@@ -80,12 +76,6 @@ void RtmpSession::onManager() {
 			shutdown();
 		}
 	}
-    if(_delayTask){
-        if(time(NULL) > _iTaskTimeLine){
-            _delayTask();
-            _delayTask = nullptr;
-        }
-    }
 }
 
 void RtmpSession::onRecv(const Buffer::Ptr &pBuf) {
@@ -212,70 +202,6 @@ void RtmpSession::onCmd_deleteStream(AMFDecoder &dec) {
 	throw std::runtime_error(StrPrinter << "Stop publishing." << endl);
 }
 
-void RtmpSession::findStream(const function<void(const RtmpMediaSource::Ptr &src)> &cb,bool retry) {
-    auto src = dynamic_pointer_cast<RtmpMediaSource>(MediaSource::find(RTMP_SCHEMA,
-                                                                       _mediaInfo._vhost,
-                                                                       _mediaInfo._app,
-                                                                       _mediaInfo._streamid,
-                                                                       true));
-    if(src || !retry){
-        cb(src);
-        return;
-    }
-
-    //广播未找到流
-    NoticeCenter::Instance().emitEvent(Broadcast::kBroadcastNotFoundStream,_mediaInfo,*this);
-
-    weak_ptr<RtmpSession> weakSelf = dynamic_pointer_cast<RtmpSession>(shared_from_this());
-    auto task_id = this;
-    auto media_info = _mediaInfo;
-
-    auto onRegist = [task_id, weakSelf, media_info, cb](BroadcastMediaChangedArgs) {
-        if(bRegist &&
-           schema == media_info._schema &&
-           vhost == media_info._vhost &&
-           app == media_info._app &&
-           stream == media_info._streamid){
-            //播发器请求的rtmp流终于注册上了
-            auto strongSelf = weakSelf.lock();
-            if(!strongSelf) {
-                return;
-            }
-            //切换到自己的线程再回复
-            //如果触发 kBroadcastMediaChanged 事件的线程与本RtmpSession绑定的线程相同,
-            //那么strongSelf->async操作可能是同步操作,
-            //通过指定参数may_sync为false确保 NoticeCenter::delListener操作延后执行,
-            //以便防止遍历事件监听对象map时做删除操作
-            strongSelf->async([task_id,weakSelf,media_info,cb](){
-                auto strongSelf = weakSelf.lock();
-                if(!strongSelf) {
-                    return;
-                }
-                DebugL << "收到rtmp注册事件,回复播放器:" << media_info._schema << "/" << media_info._vhost << "/" << media_info._app << "/" << media_info._streamid;
-
-                //再找一遍媒体源，一般能找到
-                strongSelf->findStream(cb,false);
-
-                //取消延时任务，防止多次回复
-                strongSelf->cancelDelyaTask();
-
-                //取消事件监听
-                //在事件触发时不能在当前线程移除事件监听,否则会导致遍历map时做删除操作导致程序崩溃
-                NoticeCenter::Instance().delListener(task_id,Broadcast::kBroadcastMediaChanged);
-            }, false);
-        }
-    };
-
-    NoticeCenter::Instance().addListener(task_id, Broadcast::kBroadcastMediaChanged, onRegist);
-    //5秒后执行失败回调
-    doDelay(5, [cb,task_id]() {
-        //取消监听该事件,该延时任务可以在本对象析构时或到达指定延时后调用
-        //所以该对象在销毁前一定会被取消事件监听
-        NoticeCenter::Instance().delListener(task_id,Broadcast::kBroadcastMediaChanged);
-        cb(nullptr);
-    });
-
-}
 
 void RtmpSession::sendPlayResponse(const string &err,const RtmpMediaSource::Ptr &src){
     bool authSuccess = err.empty();
@@ -377,14 +303,16 @@ void RtmpSession::doPlayResponse(const string &err,const std::function<void(bool
         return;
     }
 
-    weak_ptr<RtmpSession> weakSelf = dynamic_pointer_cast<RtmpSession>(shared_from_this());
     //鉴权成功，查找媒体源并回复
-    findStream([weakSelf,cb](const RtmpMediaSource::Ptr &src){
+    _mediaInfo._schema = RTMP_SCHEMA;
+    weak_ptr<RtmpSession> weakSelf = dynamic_pointer_cast<RtmpSession>(shared_from_this());
+    MediaSource::findAsync(_mediaInfo,weakSelf.lock(), true,5000,[weakSelf,cb](const MediaSource::Ptr &src){
+        auto rtmp_src = dynamic_pointer_cast<RtmpMediaSource>(src);
         auto strongSelf = weakSelf.lock();
         if(strongSelf){
-            strongSelf->sendPlayResponse("", src);
+            strongSelf->sendPlayResponse("", rtmp_src);
         }
-        cb(src.operator bool());
+        cb(rtmp_src.operator bool());
     });
 }
 
@@ -556,17 +484,5 @@ void RtmpSession::onSendMedia(const RtmpPacket::Ptr &pkt) {
 	sendRtmp(pkt->typeId, pkt->streamId, pkt, modifiedStamp, pkt->chunkId);
 }
 
-void RtmpSession::doDelay(int delaySec, const std::function<void()> &fun) {
-    if(_delayTask){
-        _delayTask();
-    }
-    _delayTask = fun;
-    _iTaskTimeLine = time(NULL) + delaySec;
-}
-
-void RtmpSession::cancelDelyaTask(){
-    _delayTask = nullptr;
-}
-
 
 } /* namespace mediakit */

src/Rtmp/RtmpSession.h

@@ -89,10 +89,6 @@ private:
         safeShutdown();
         return true;
     }
-
-    void doDelay(int delaySec,const std::function<void()> &fun);
-	void cancelDelyaTask();
-	void findStream(const function<void(const RtmpMediaSource::Ptr &src)> &cb ,bool retry = true);
 private:
 	std::string _strTcUrl;
 	MediaInfo _mediaInfo;
@@ -105,8 +101,6 @@ private:
 	uint32_t _aui32FirstStamp[2] = {0};
 	//消耗的总流量
 	uint64_t _ui64TotalBytes = 0;
-    std::function<void()> _delayTask;
-    uint32_t _iTaskTimeLine = 0;
 
 };
 

src/Rtsp/RtspSession.cpp

@@ -126,13 +126,6 @@ void RtspSession::onManager() {
 		shutdown();
 		return;
 	}
-
-    if(_delayTask){
-        if(time(NULL) > _iTaskTimeLine){
-            _delayTask();
-            _delayTask = nullptr;
-        }
-    }
 }
 
 void RtspSession::onRecv(const Buffer::Ptr &pBuf) {
@@ -307,43 +300,60 @@ bool RtspSession::handleReq_RECORD(const Parser &parser){
 	return true;
 }
 
-
 bool RtspSession::handleReq_Describe(const Parser &parser) {
-	weak_ptr<RtspSession> weakSelf = dynamic_pointer_cast<RtspSession>(shared_from_this());
+    _mediaInfo._schema = RTSP_SCHEMA;
-	auto authorization = parser["Authorization"];
+    auto authorization = parser["Authorization"];
+    weak_ptr<RtspSession> weakSelf = dynamic_pointer_cast<RtspSession>(shared_from_this());
+    MediaSource::findAsync(_mediaInfo,weakSelf.lock(), true,5000,[weakSelf,authorization](const MediaSource::Ptr &src){
+        auto strongSelf = weakSelf.lock();
+        if(!strongSelf){
+            return;
+        }
+        auto rtsp_src = dynamic_pointer_cast<RtspMediaSource>(src);
+        if (!rtsp_src) {
+            //未找到相应的MediaSource
+            WarnL << "No such stream:" <<  strongSelf->_mediaInfo._vhost << " " <<  strongSelf->_mediaInfo._app << " " << strongSelf->_mediaInfo._streamid;
+            strongSelf->send_StreamNotFound();
+            strongSelf->shutdown();
+            return;
+        }
+        //找到了响应的rtsp流
+        strongSelf->_strSdp = rtsp_src->getSdp();
+        SdpAttr sdpAttr(strongSelf->_strSdp);
+        strongSelf->_aTrackInfo = sdpAttr.getAvailableTrack();
+        if (strongSelf->_aTrackInfo.empty()) {
+            //该流无效
+            strongSelf->send_StreamNotFound();
+            strongSelf->shutdown();
+            return;
+        }
+        strongSelf->_strSession = makeRandStr(12);
+        strongSelf->_pMediaSrc = rtsp_src;
+        for(auto &track : strongSelf->_aTrackInfo){
+            track->_ssrc = rtsp_src->getSsrc(track->_type);
+            track->_seq = rtsp_src->getSeqence(track->_type);
+            track->_time_stamp = rtsp_src->getTimeStamp(track->_type);
+        }
 
-	findStream([weakSelf,authorization](bool success){
+        //该请求中的认证信息
-    	auto strongSelf = weakSelf.lock();
+        onGetRealm invoker = [weakSelf,authorization](const string &realm){
-    	if(!strongSelf){
+            if(realm.empty()){
-			return;
+                //无需认证,回复sdp
-    	}
+                onAuthSuccess(weakSelf);
+                return;
+            }
+            //该流需要认证
+            onAuthUser(weakSelf,realm,authorization);
+        };
 
-    	if(!success){
+        //广播是否需要认证事件
-			//未找到相应的MediaSource
+        if(!NoticeCenter::Instance().emitEvent(Broadcast::kBroadcastOnGetRtspRealm,
-			WarnL << "No such stream:" <<  strongSelf->_mediaInfo._vhost << " " <<  strongSelf->_mediaInfo._app << " " << strongSelf->_mediaInfo._streamid;
+                                               strongSelf->_mediaInfo,
-			strongSelf->send_StreamNotFound();
+                                               invoker,
-			strongSelf->shutdown();
+                                               *strongSelf)){
-			return;
+            //无人监听此事件，说明无需认证
-    	}
+            invoker("");
-		//该请求中的认证信息
+        }
-		onGetRealm invoker = [weakSelf,authorization](const string &realm){
-			if(realm.empty()){
-				//无需认证,回复sdp
-				onAuthSuccess(weakSelf);
-				return;
-			}
-			//该流需要认证
-			onAuthUser(weakSelf,realm,authorization);
-		};
-
-		//广播是否需要认证事件
-		if(!NoticeCenter::Instance().emitEvent(Broadcast::kBroadcastOnGetRtspRealm,
-											   strongSelf->_mediaInfo,
-											   invoker,
-											   *strongSelf)){
-			//无人监听此事件，说明无需认证
-			invoker("");
-		}
     });
     return true;
 }
@@ -885,98 +895,6 @@ inline void RtspSession::send_NotAcceptable() {
 	sendRtspResponse("406 Not Acceptable",{"Connection","Close"});
 }
 
-void RtspSession::doDelay(int delaySec, const std::function<void()> &fun) {
-    if(_delayTask){
-        _delayTask();
-    }
-    _delayTask = fun;
-    _iTaskTimeLine = time(NULL) + delaySec;
-}
-
-void RtspSession::cancelDelyaTask(){
-    _delayTask = nullptr;
-}
-
-void RtspSession::findStream(const function<void(bool)> &cb) {
-	bool success = findStream();
-	if (success) {
-		cb(true);
-		return;
-	}
-
-	//广播未找到流
-	NoticeCenter::Instance().emitEvent(Broadcast::kBroadcastNotFoundStream,_mediaInfo,*this);
-
-	weak_ptr<RtspSession> weakSelf = dynamic_pointer_cast<RtspSession>(shared_from_this());
-	auto task_id = this;
-	auto media_info = _mediaInfo;
-
-	auto onRegist = [task_id, weakSelf, media_info, cb](BroadcastMediaChangedArgs) {
-		if (bRegist &&
-			schema == media_info._schema &&
-			vhost == media_info._vhost &&
-			app == media_info._app &&
-			stream == media_info._streamid) {
-			//播发器请求的rtsp流终于注册上了
-			auto strongSelf = weakSelf.lock();
-			if (!strongSelf) {
-				return;
-			}
-			//切换到自己的线程再回复
-			//如果触发 kBroadcastMediaChanged 事件的线程与本RtspSession绑定的线程相同,
-			//那么strongSelf->async操作可能是同步操作,
-			//通过指定参数may_sync为false确保 NoticeCenter::delListener操作延后执行,
-			//以便防止遍历事件监听对象map时做删除操作
-			strongSelf->async([task_id, weakSelf, media_info, cb]() {
-				auto strongSelf = weakSelf.lock();
-				if (!strongSelf) {
-					return;
-				}
-				DebugL << "收到rtsp注册事件,回复播放器:" << media_info._schema << "/" << media_info._vhost << "/"
-					   << media_info._app << "/" << media_info._streamid;
-				cb(strongSelf->findStream());
-				//取消延时任务，防止多次回复
-				strongSelf->cancelDelyaTask();
-
-				//取消事件监听
-				//在事件触发时不能在当前线程移除事件监听,否则会导致遍历map时做删除操作导致程序崩溃
-				NoticeCenter::Instance().delListener(task_id, Broadcast::kBroadcastMediaChanged);
-			}, false);
-		}
-	};
-
-	NoticeCenter::Instance().addListener(task_id, Broadcast::kBroadcastMediaChanged, onRegist);
-	//5秒后执行失败回调
-	doDelay(5, [cb,task_id]() {
-		NoticeCenter::Instance().delListener(task_id,Broadcast::kBroadcastMediaChanged);
-		cb(false);
-	});
-}
-
-inline bool RtspSession::findStream() {
-	RtspMediaSource::Ptr pMediaSrc =
-    dynamic_pointer_cast<RtspMediaSource>( MediaSource::find(RTSP_SCHEMA,_mediaInfo._vhost, _mediaInfo._app,_mediaInfo._streamid) );
-	if (!pMediaSrc) {
-		return false;
-	}
-	_strSdp = pMediaSrc->getSdp();
-	SdpAttr sdpAttr(_strSdp);
-	_aTrackInfo = sdpAttr.getAvailableTrack();
-
-	if (_aTrackInfo.empty()) {
-		return false;
-	}
-	_strSession = makeRandStr(12);
-	_pMediaSrc = pMediaSrc;
-
-	for(auto &track : _aTrackInfo){
-		track->_ssrc = pMediaSrc->getSsrc(track->_type);
-		track->_seq = pMediaSrc->getSeqence(track->_type);
-		track->_time_stamp = pMediaSrc->getTimeStamp(track->_type);
-	}
-	return true;
-}
-
 
 void RtspSession::onRtpSorted(const RtpPacket::Ptr &rtppt, int trackidx) {
 	_pushSrc->onWrite(rtppt, false);

src/Rtsp/RtspSession.h

@@ -136,9 +136,6 @@ private:
 	void inline send_UnsupportedTransport(); //不支持的传输模式
 	void inline send_SessionNotFound(); //会话id错误
 	void inline send_NotAcceptable(); //rtsp同时播放数限制
-	inline bool findStream(); //根据rtsp url查找 MediaSource实例
-    inline void findStream(const function<void(bool)> &cb); //根据rtsp url查找 MediaSource实例
-
 	inline string printSSRC(uint32_t ui32Ssrc);
 	inline int getTrackIndexByTrackType(TrackType type);
     inline int getTrackIndexByControlSuffix(const string &controlSuffix);
@@ -154,9 +151,6 @@ private:
     static void onAuthBasic(const weak_ptr<RtspSession> &weakSelf,const string &realm,const string &strBase64);
     static void onAuthDigest(const weak_ptr<RtspSession> &weakSelf,const string &realm,const string &strMd5);
 
-    void doDelay(int delaySec,const std::function<void()> &fun);
-    void cancelDelyaTask();
-
 	inline void sendRtpPacket(const RtpPacket::Ptr &pkt);
 	bool sendRtspResponse(const string &res_code,const std::initializer_list<string> &header, const string &sdp = "" , const char *protocol = "RTSP/1.0");
 	bool sendRtspResponse(const string &res_code,const StrCaseMap &header = StrCaseMap(), const string &sdp = "",const char *protocol = "RTSP/1.0");
@@ -190,9 +184,6 @@ private:
 	//一次发送 get 一次发送post，需要通过x-sessioncookie关联起来
 	string _http_x_sessioncookie;
 	function<void(const Buffer::Ptr &pBuf)> _onRecv;
-
-    std::function<void()> _delayTask;
-    uint32_t _iTaskTimeLine = 0;
     atomic<bool> _enableSendRtp;
 
     //rtsp推流相关