修复rtsp组播相关的bug

src/Rtsp/RtpBroadCaster.cpp

@@ -97,7 +97,7 @@ RtpBroadCaster::~RtpBroadCaster() {
 	_pReader->setDetachCB(nullptr);
 	DebugL;
 }
-RtpBroadCaster::RtpBroadCaster(const string &strLocalIp,const string &strVhost,const string &strApp,const string &strStream) {
+RtpBroadCaster::RtpBroadCaster(const EventPoller::Ptr &poller,const string &strLocalIp,const string &strVhost,const string &strApp,const string &strStream) {
 	auto src = dynamic_pointer_cast<RtspMediaSource>(MediaSource::find(RTSP_SCHEMA,strVhost,strApp, strStream));
 	if(!src){
 		auto strErr = StrPrinter << "未找到媒体源:" << strVhost << " " << strApp << " " << strStream << endl;
@@ -124,7 +124,7 @@ RtpBroadCaster::RtpBroadCaster(const string &strLocalIp,const string &strVhost,c
 		bzero(&(peerAddr.sin_zero), sizeof peerAddr.sin_zero);
 		_apUdpSock[i]->setSendPeerAddr((struct sockaddr *)&peerAddr);
 	}
-	_pReader = src->getRing()->attach(EventPollerPool::Instance().getPoller());
+	_pReader = src->getRing()->attach(poller);
 	_pReader->setReadCB([this](const RtpPacket::Ptr &pkt){
 		int i = (int)(pkt->type);
 		auto &pSock = _apUdpSock[i];
@@ -154,9 +154,9 @@ uint16_t RtpBroadCaster::getPort(TrackType trackType){
 string RtpBroadCaster::getIP(){
 	return inet_ntoa(_aPeerUdpAddr[0].sin_addr);
 }
-RtpBroadCaster::Ptr RtpBroadCaster::make(const string &strLocalIp,const string &strVhost,const string &strApp,const string &strStream){
+RtpBroadCaster::Ptr RtpBroadCaster::make(const EventPoller::Ptr &poller,const string &strLocalIp,const string &strVhost,const string &strApp,const string &strStream){
 	try{
-		auto ret = Ptr(new RtpBroadCaster(strLocalIp,strVhost,strApp,strStream));
+		auto ret = Ptr(new RtpBroadCaster(poller,strLocalIp,strVhost,strApp,strStream));
 		lock_guard<recursive_mutex> lck(g_mtx);
 		string strKey = StrPrinter << strLocalIp << " "  << strVhost << " " << strApp << " " << strStream << endl;
 		weak_ptr<RtpBroadCaster> weakPtr = ret;
@@ -168,17 +168,17 @@ RtpBroadCaster::Ptr RtpBroadCaster::make(const string &strLocalIp,const string &
 	}
 }
 
-RtpBroadCaster::Ptr RtpBroadCaster::get(const string &strLocalIp,const string &strVhost,const string &strApp,const string &strStream) {
+RtpBroadCaster::Ptr RtpBroadCaster::get(const EventPoller::Ptr &poller,const string &strLocalIp,const string &strVhost,const string &strApp,const string &strStream) {
 	string strKey = StrPrinter << strLocalIp << " " << strVhost << " " << strApp << " " << strStream << endl;
 	lock_guard<recursive_mutex> lck(g_mtx);
 	auto it = g_mapBroadCaster.find(strKey);
 	if (it == g_mapBroadCaster.end()) {
-		return make(strLocalIp,strVhost,strApp, strStream);
+		return make(poller,strLocalIp,strVhost,strApp, strStream);
 	}
 	auto ret = it->second.lock();
 	if (!ret) {
 		g_mapBroadCaster.erase(it);
-		return make(strLocalIp,strVhost,strApp, strStream);
+		return make(poller,strLocalIp,strVhost,strApp, strStream);
 	}
 	return ret;
 }

src/Rtsp/RtpBroadCaster.h

@@ -71,14 +71,14 @@ public:
 	typedef std::shared_ptr<RtpBroadCaster> Ptr;
 	typedef function<void()> onDetach;
 	virtual ~RtpBroadCaster();
-	static Ptr get(const string &strLocalIp,const string &strVhost,const string &strApp,const string &strStream);
+	static Ptr get(const EventPoller::Ptr &poller,const string &strLocalIp,const string &strVhost,const string &strApp,const string &strStream);
 	void setDetachCB(void *listener,const onDetach &cb);
 	uint16_t getPort(TrackType trackType);
 	string getIP();
 private:
 	static recursive_mutex g_mtx;
 	static unordered_map<string , weak_ptr<RtpBroadCaster> > g_mapBroadCaster;
-	static Ptr make(const string &strLocalIp,const string &strVhost,const string &strApp,const string &strStream);
+	static Ptr make(const EventPoller::Ptr &poller,const string &strLocalIp,const string &strVhost,const string &strApp,const string &strStream);
 
 	std::shared_ptr<uint32_t> _multiAddr;
 	recursive_mutex _mtx;
@@ -87,7 +87,7 @@ private:
 	Socket::Ptr _apUdpSock[2];
 	struct sockaddr_in _aPeerUdpAddr[2];
 
-	RtpBroadCaster(const string &strLocalIp,const string &strVhost,const string &strApp,const string &strStream);
+	RtpBroadCaster(const EventPoller::Ptr &poller,const string &strLocalIp,const string &strVhost,const string &strApp,const string &strStream);
 
 };
 

src/Rtsp/RtspSession.cpp

@@ -637,7 +637,7 @@ bool RtspSession::handleReq_Setup(const Parser &parser) {
 		break;
 	case Rtsp::RTP_MULTICAST: {
 		if(!_pBrdcaster){
-			_pBrdcaster = RtpBroadCaster::get(get_local_ip(),_mediaInfo._vhost, _mediaInfo._app, _mediaInfo._streamid);
+			_pBrdcaster = RtpBroadCaster::get(getPoller(),get_local_ip(),_mediaInfo._vhost, _mediaInfo._app, _mediaInfo._streamid);
 			if (!_pBrdcaster) {
 				send_NotAcceptable();
 				return false;