refactoring pc closing err

Author: rainliu

Cargo.toml

@@ -21,11 +21,11 @@ ice = { package = "webrtc-ice", version = "0.5.1" }
 dtls = { package = "webrtc-dtls", version = "0.5.0" }
 rtp = "0.6.1"
 rtcp = "0.6.1"
-srtp = { package = "webrtc-srtp", version = "0.8.3" }
+srtp = { package = "webrtc-srtp", version = "0.8.4" }
 sctp = { package = "webrtc-sctp", version = "0.4.1" }
 data = { package = "webrtc-data", version = "0.3.1" }
 media = { package = "webrtc-media", version = "0.4.1" }
-interceptor = "0.7.0"
+interceptor = "0.7.1"
 tokio = { version = "1.12.0", features = ["full"] }
 log = "0.4.14"
 async-trait = "0.1.42"

crates/interceptor

@@ -1,3 +1,4 @@
+use std::collections::HashMap;
 use std::future::Future;
 use std::pin::Pin;
 use std::sync::atomic::{AtomicBool, AtomicU8, Ordering};
@@ -73,7 +74,7 @@ pub struct RTCDtlsTransport {
     pub(crate) srtp_endpoint: Mutex<Option<Arc<Endpoint>>>,
     pub(crate) srtcp_endpoint: Mutex<Option<Arc<Endpoint>>>,
 
-    pub(crate) simulcast_streams: Mutex<Vec<Arc<Stream>>>,
+    pub(crate) simulcast_streams: Mutex<HashMap<SSRC, Arc<Stream>>>,
 
     pub(crate) srtp_ready_signal: Arc<AtomicBool>,
     pub(crate) srtp_ready_tx: Mutex<Option<mpsc::Sender<()>>>,
@@ -488,13 +489,17 @@ impl RTCDtlsTransport {
         {
             let simulcast_streams: Vec<Arc<Stream>> = {
                 let mut simulcast_streams = self.simulcast_streams.lock().await;
-                simulcast_streams.drain(..).collect()
+                simulcast_streams.drain().map(|(_, v)| v).collect()
             };
             for ss in simulcast_streams {
                 match ss.close().await {
                     Ok(_) => {}
                     Err(err) => {
-                        close_errs.push(err.into());
+                        close_errs.push(Error::new(format!(
+                            "simulcast_streams ssrc={}: {}",
+                            ss.get_ssrc(),
+                            err
+                        )));
                     }
                 };
             }
@@ -546,9 +551,14 @@ impl RTCDtlsTransport {
         }
     }
 
-    pub(crate) async fn store_simulcast_stream(&self, stream: Arc<Stream>) {
+    pub(crate) async fn store_simulcast_stream(&self, ssrc: SSRC, stream: Arc<Stream>) {
+        let mut simulcast_streams = self.simulcast_streams.lock().await;
+        simulcast_streams.insert(ssrc, stream);
+    }
+
+    pub(crate) async fn remove_simulcast_stream(&self, ssrc: SSRC) {
         let mut simulcast_streams = self.simulcast_streams.lock().await;
-        simulcast_streams.push(stream)
+        simulcast_streams.remove(&ssrc);
     }
 
     pub(crate) async fn streams_for_ssrc(

crates/srtp

@@ -1857,15 +1857,15 @@ impl RTCPeerConnection {
         let mut close_errs = vec![];
 
         if let Err(err) = self.interceptor.close().await {
-            close_errs.push(err);
+            close_errs.push(Error::new(format!("interceptor: {}", err)));
         }
 
         // https://www.w3.org/TR/webrtc/#dom-rtcpeerconnection-close (step #4)
         {
             let mut rtp_transceivers = self.internal.rtp_transceivers.lock().await;
             for t in &*rtp_transceivers {
                 if let Err(err) = t.stop().await {
-                    close_errs.push(err.into());
+                    close_errs.push(Error::new(format!("rtp_transceivers: {}", err)));
                 }
             }
             rtp_transceivers.clear();
@@ -1876,25 +1876,25 @@ impl RTCPeerConnection {
             let mut data_channels = self.internal.sctp_transport.data_channels.lock().await;
             for d in &*data_channels {
                 if let Err(err) = d.close().await {
-                    close_errs.push(err.into());
+                    close_errs.push(Error::new(format!("data_channels: {}", err)));
                 }
             }
             data_channels.clear();
         }
 
         // https://www.w3.org/TR/webrtc/#dom-rtcpeerconnection-close (step #6)
         if let Err(err) = self.internal.sctp_transport.stop().await {
-            close_errs.push(err.into());
+            close_errs.push(Error::new(format!("sctp_transport: {}", err)));
         }
 
         // https://www.w3.org/TR/webrtc/#dom-rtcpeerconnection-close (step #7)
         if let Err(err) = self.internal.dtls_transport.stop().await {
-            close_errs.push(err.into());
+            close_errs.push(Error::new(format!("dtls_transport: {}", err)));
         }
 
         // https://www.w3.org/TR/webrtc/#dom-rtcpeerconnection-close (step #8, #9, #10)
         if let Err(err) = self.internal.ice_transport.stop().await {
-            close_errs.push(err.into());
+            close_errs.push(Error::new(format!("dtls_transport: {}", err)));
         }
 
         // https://www.w3.org/TR/webrtc/#dom-rtcpeerconnection-close (step #11)
@@ -1908,7 +1908,7 @@ impl RTCPeerConnection {
         .await;
 
         if let Err(err) = self.internal.ops.close().await {
-            close_errs.push(err.into());
+            close_errs.push(Error::new(format!("ops: {}", err)));
         }
 
         flatten_errs(close_errs)

src/dtls_transport/mod.rs

@@ -263,11 +263,12 @@ impl PeerConnectionInternal {
                 let simulcast_routine_count2 = Arc::clone(&simulcast_routine_count);
                 let pci2 = Arc::clone(&pci);
                 tokio::spawn(async move {
+                    let ssrc = stream.get_ssrc();
+
                     dtls_transport2
-                        .store_simulcast_stream(Arc::clone(&stream))
+                        .store_simulcast_stream(ssrc, Arc::clone(&stream))
                         .await;
 
-                    let ssrc = stream.get_ssrc();
                     if let Err(err) = pci2.handle_incoming_ssrc(stream, ssrc).await {
                         log::error!(
                             "Incoming unhandled RTP ssrc({}), on_track will not be fired. {}",
@@ -1182,6 +1183,7 @@ impl PeerConnectionInternal {
                         let _ = rtcp_read_stream.close().await;
                     }
                     icpr.unbind_remote_stream(&stream_info).await;
+                    self.dtls_transport.remove_simulcast_stream(ssrc).await;
                 }
 
                 return Err(Error::ErrPeerConnSimulcastIncomingSSRCFailed);