提高webrtc代码健壮性

webrtc/WebRtcSession.cpp

@@ -50,26 +50,24 @@ EventPoller::Ptr WebRtcSession::getPoller(const Buffer::Ptr &buffer) {
 }
 }
 
 
 void WebRtcSession::onRecv(const Buffer::Ptr &buffer) {
 void WebRtcSession::onRecv(const Buffer::Ptr &buffer) {
-    auto buf = buffer->data();
-    auto len = buffer->size();
-
-    if (!_transport) {
-        auto user_name = getUserName(buffer);
-        if (user_name.empty()) {
-            //逻辑分支不太可能走到这里
-            WarnL << user_name;
-            return;
-        }
-        _transport = WebRtcTransportImp::getRtcTransport(user_name, true);
-        if (!_transport) {
-            //逻辑分支不太可能走到这里
-            WarnL << user_name;
-            return;
+    try {
+        onRecv_l(buffer);
+    } catch (std::exception &ex) {
+        shutdown(SockException(Err_shutdown, ex.what()));
     }
     }
+}
+
+void WebRtcSession::onRecv_l(const Buffer::Ptr &buffer) {
+    if (_find_transport) {
+        //只允许寻找一次transport
+        _find_transport = false;
+        _transport = WebRtcTransportImp::getRtcTransport(getUserName(buffer), true);
+        CHECK(_transport && _transport->getPoller()->isCurrentThread());
         _transport->setSession(shared_from_this());
         _transport->setSession(shared_from_this());
     }
     }
     _ticker.resetTime();
     _ticker.resetTime();
-    _transport->inputSockData(buf, len, &_peer_addr);
+    CHECK(_transport);
+    _transport->inputSockData(buffer->data(), buffer->size(), &_peer_addr);
 }
 }
 
 
 void WebRtcSession::onError(const SockException &err) {
 void WebRtcSession::onError(const SockException &err) {

webrtc/WebRtcSession.h

@@ -30,6 +30,10 @@ public:
     void onManager() override;
     void onManager() override;
 
 
 private:
 private:
+    void onRecv_l(const Buffer::Ptr &);
+
+private:
+    bool _find_transport = true;
     Ticker _ticker;
     Ticker _ticker;
     struct sockaddr _peer_addr;
     struct sockaddr _peer_addr;
     std::shared_ptr<WebRtcTransportImp> _transport;
     std::shared_ptr<WebRtcTransportImp> _transport;

webrtc/WebRtcTransport.cpp

@@ -1018,7 +1018,10 @@ void WebRtcTransportImp::unregisterSelf() {
     s_rtc_map.erase(getKey());
     s_rtc_map.erase(getKey());
 }
 }
 
 
-WebRtcTransportImp::Ptr WebRtcTransportImp::getRtcTransport(const string &key, bool unref_self){
+WebRtcTransportImp::Ptr WebRtcTransportImp::getRtcTransport(const string &key, bool unref_self) {
+    if (key.empty()) {
+        return nullptr;
+    }
     lock_guard<mutex> lck(s_rtc_mtx);
     lock_guard<mutex> lck(s_rtc_mtx);
     auto it = s_rtc_map.find(key);
     auto it = s_rtc_map.find(key);
     if (it == s_rtc_map.end()) {
     if (it == s_rtc_map.end()) {