添加踢出推流器功能

src/Common/MediaSource.h

@@ -55,11 +55,17 @@ public:
     virtual ~MediaSourceEvent(){};
 public:
     virtual bool seekTo(uint32_t ui32Stamp){
+        //拖动进度条
         return false;
     }
     virtual uint32_t getStamp() {
+        //获取时间戳
         return 0;
     }
+    virtual bool shutDown() {
+        //通知其停止推流
+        return false;
+    }
 };
 class MediaInfo
 {
@@ -151,9 +157,30 @@ public:
         }
         return listener->getStamp();
     }
+    bool shutDown() {
+        auto listener = m_listener.lock();
+        if(!listener){
+            return false;
+        }
+        return listener->shutDown();
+    }
     void setListener(const std::weak_ptr<MediaSourceEvent> &listener){
         m_listener = listener;
     }
+
+    template <typename FUN>
+    static void for_each_media(FUN && fun){
+        lock_guard<recursive_mutex> lock(g_mtxMediaSrc);
+        for (auto &pr0 : g_mapMediaSrc){
+            for(auto &pr1 : pr0.second){
+                for(auto &pr2 : pr1.second){
+                    for(auto &pr3 : pr2.second){
+                        fun(pr0.first,pr1.first,pr2.first,pr3.first,pr3.second.lock());
+                    }
+                }
+            }
+        }
+    }
 private:
         template <typename FUN>
         static bool searchMedia(const string &schema,
@@ -195,6 +222,7 @@ private:
             }
         }
     };
+
     void unregisted();
 protected:
     std::weak_ptr<MediaSourceEvent> m_listener;

src/MediaFile/MediaReader.cpp

@@ -181,6 +181,10 @@ void MediaReader::startReadMP4() {
 uint32_t MediaReader::getStamp() {
 	 return m_iSeekTime + m_ticker.elapsedTime();
 }
+bool MediaReader::shutDown(){
+    AsyncTaskThread::Instance().CancelTask(reinterpret_cast<uint64_t>(this));
+    return true;
+}
 
 bool MediaReader::readSample(int iTimeInc) {
 	TimeTicker();

src/MediaFile/MediaReader.h

@@ -51,6 +51,7 @@ public:
 public:
 	bool seekTo(uint32_t ui32Stamp) override;
 	uint32_t getStamp() override;
+    bool shutDown() override;
 private:
 
 #ifdef ENABLE_MP4V2

src/Rtmp/RtmpSession.cpp

@@ -145,7 +145,7 @@ void RtmpSession::onCmd_publish(AMFDecoder &dec) {
         sendReply("onStatus", nullptr, status);
         if (!ok) {
             WarnL << "onPublish:"
-                  << (authSuccess ? "Already publishing:" : err.data())
+                  << (authSuccess ? "Already publishing:" : err.data()) << " "
                   << m_mediaInfo.m_vhost << " "
                   << m_mediaInfo.m_app << " "
                   << m_mediaInfo.m_streamid << endl;
@@ -154,6 +154,7 @@ void RtmpSession::onCmd_publish(AMFDecoder &dec) {
         }
         m_bPublisherSrcRegisted = false;
         m_pPublisherSrc.reset(new RtmpToRtspMediaSource(m_mediaInfo.m_vhost,m_mediaInfo.m_app,m_mediaInfo.m_streamid));
+        m_pPublisherSrc->setListener(dynamic_pointer_cast<MediaSourceEvent>(shared_from_this()));
     };
 
     weak_ptr<RtmpSession> weakSelf = dynamic_pointer_cast<RtmpSession>(shared_from_this());
@@ -213,7 +214,7 @@ void  RtmpSession::doPlay(AMFDecoder &dec){
         sendReply("onStatus", nullptr, status);
         if (!ok) {
             WarnL << "onPlayed:"
-                  << (authSuccess ? "No such stream:" : err.data())
+                  << (authSuccess ? "No such stream:" : err.data()) << " "
                   << m_mediaInfo.m_vhost << " "
                   << m_mediaInfo.m_app << " "
                   << m_mediaInfo.m_streamid

src/Rtmp/RtmpSession.h

@@ -44,7 +44,7 @@ using namespace ZL::Network;
 namespace ZL {
 namespace Rtmp {
 
-class RtmpSession: public TcpLimitedSession<MAX_TCP_SESSION> ,public  RtmpProtocol{
+class RtmpSession: public TcpLimitedSession<MAX_TCP_SESSION> ,public  RtmpProtocol , public MediaSourceEvent{
 public:
 	typedef std::shared_ptr<RtmpSession> Ptr;
 	RtmpSession(const std::shared_ptr<ThreadPool> &_th, const Socket::Ptr &_sock);
@@ -95,6 +95,12 @@ private:
 		invoke << str << m_dNowReqID << reply << status;
 		sendResponse(MSG_CMD, invoke.data());
 	}
+
+    bool shutDown() override {
+        InfoL << "kick out:" << m_mediaInfo.m_vhost << " " << m_mediaInfo.m_app << " " << m_mediaInfo.m_streamid;
+        safeShutdown();
+        return true;
+    }
 };
 
 } /* namespace Rtmp */

src/Rtsp/Rtsp.h

@@ -114,7 +114,7 @@ public:
 				auto field = FindField(line.c_str(), NULL, ": ");
 				auto value = FindField(line.c_str(), ": ", NULL);
 				if (field.size() != 0) {
-					m_mapValues[field] = value;
+					m_mapHeaders[field] = value;
 				}
 			}
 			start = start + line.size() + 2;
@@ -142,8 +142,8 @@ public:
 	}
 	const string& operator[](const char *name) const {
 		//rtsp field
-		auto it = m_mapValues.find(name);
-		if (it == m_mapValues.end()) {
+		auto it = m_mapHeaders.find(name);
+		if (it == m_mapHeaders.end()) {
 			return m_strNull;
 		}
 		return it->second;
@@ -157,7 +157,7 @@ public:
         m_strFullUrl.clear();
 		m_strTail.clear();
 		m_strContent.clear();
-		m_mapValues.clear();
+		m_mapHeaders.clear();
 		m_mapUrlArgs.clear();
 	}
 
@@ -169,7 +169,7 @@ public:
 	}
 
 	StrCaseMap& getValues() const {
-		return m_mapValues;
+		return m_mapHeaders;
 	}
 	StrCaseMap& getUrlArgs() const {
 		return m_mapUrlArgs;
@@ -195,7 +195,7 @@ private:
 	string m_strContent;
 	string m_strNull;
     string m_strFullUrl;
-	mutable StrCaseMap m_mapValues;
+	mutable StrCaseMap m_mapHeaders;
 	mutable StrCaseMap m_mapUrlArgs;
 };