防止线程竞争的bug

src/Rtmp/RtmpPusher.cpp

@@ -41,9 +41,12 @@ void RtmpPusher::teardown() {
 		m_strApp.clear();
 		m_strApp.clear();
 		m_strStream.clear();
 		m_strStream.clear();
 		m_strTcUrl.clear();
 		m_strTcUrl.clear();
-		m_mapOnResultCB.clear();
+		{
+			lock_guard<recursive_mutex> lck(m_mtxOnResultCB);
+			m_mapOnResultCB.clear();
+		}
         {
         {
-            lock_guard<recursive_mutex> lck(m_mtxDeque);
+            lock_guard<recursive_mutex> lck(m_mtxOnStatusCB);
             m_dqOnStatusCB.clear();
             m_dqOnStatusCB.clear();
         }
         }
 		m_pPlayTimer.reset();
 		m_pPlayTimer.reset();
@@ -198,6 +201,7 @@ inline void RtmpPusher::send_metaData(){
 }
 }
 void RtmpPusher::onCmd_result(AMFDecoder &dec){
 void RtmpPusher::onCmd_result(AMFDecoder &dec){
 	auto iReqId = dec.load<int>();
 	auto iReqId = dec.load<int>();
+	lock_guard<recursive_mutex> lck(m_mtxOnResultCB);
 	auto it = m_mapOnResultCB.find(iReqId);
 	auto it = m_mapOnResultCB.find(iReqId);
 	if(it != m_mapOnResultCB.end()){
 	if(it != m_mapOnResultCB.end()){
 		it->second(dec);
 		it->second(dec);
@@ -218,7 +222,7 @@ void RtmpPusher::onCmd_onStatus(AMFDecoder &dec) {
 		throw std::runtime_error("onStatus: 未找到结果对象");
 		throw std::runtime_error("onStatus: 未找到结果对象");
 	}
 	}
 
 
-    lock_guard<recursive_mutex> lck(m_mtxDeque);
+    lock_guard<recursive_mutex> lck(m_mtxOnStatusCB);
 	if(m_dqOnStatusCB.size()){
 	if(m_dqOnStatusCB.size()){
 		m_dqOnStatusCB.front()(val);
 		m_dqOnStatusCB.front()(val);
 		m_dqOnStatusCB.pop_front();
 		m_dqOnStatusCB.pop_front();

src/Rtmp/RtmpPusher.h

@@ -53,11 +53,12 @@ private:
 
 
 	template<typename FUN>
 	template<typename FUN>
 	inline void addOnResultCB(const FUN &fun) {
 	inline void addOnResultCB(const FUN &fun) {
+		lock_guard<recursive_mutex> lck(m_mtxOnResultCB);
 		m_mapOnResultCB.emplace(m_iReqID, fun);
 		m_mapOnResultCB.emplace(m_iReqID, fun);
 	}
 	}
 	template<typename FUN>
 	template<typename FUN>
 	inline void addOnStatusCB(const FUN &fun) {
 	inline void addOnStatusCB(const FUN &fun) {
-		lock_guard<recursive_mutex> lck(m_mtxDeque);
+		lock_guard<recursive_mutex> lck(m_mtxOnStatusCB);
 		m_dqOnStatusCB.emplace_back(fun);
 		m_dqOnStatusCB.emplace_back(fun);
 	}
 	}
 
 
@@ -75,8 +76,9 @@ private:
 	string m_strTcUrl;
 	string m_strTcUrl;
 
 
 	unordered_map<int, function<void(AMFDecoder &dec)> > m_mapOnResultCB;
 	unordered_map<int, function<void(AMFDecoder &dec)> > m_mapOnResultCB;
+	recursive_mutex m_mtxOnResultCB;
 	deque<function<void(AMFValue &dec)> > m_dqOnStatusCB;
 	deque<function<void(AMFValue &dec)> > m_dqOnStatusCB;
-	recursive_mutex m_mtxDeque;
+	recursive_mutex m_mtxOnStatusCB;
 
 
 	typedef void (RtmpPusher::*rtmpCMDHandle)(AMFDecoder &dec);
 	typedef void (RtmpPusher::*rtmpCMDHandle)(AMFDecoder &dec);
 	static unordered_map<string, rtmpCMDHandle> g_mapCmd;
 	static unordered_map<string, rtmpCMDHandle> g_mapCmd;