change SendMutex to tokio::sync::Mutex

Author: kolserdav

webrtc/src/peer_connection/mod.rs

@@ -469,7 +469,7 @@ impl RTCPeerConnection {
                                 None => return true, // doesn't contain a single a=msid line
                             };
 
-                            let sender = t.sender();
+                            let sender = t.sender().await;
                             // (...)or the number of MSIDs from the a=msid lines in this m= section,
                             // or the MSID values themselves, differ from what is in
                             // transceiver.sender.[[AssociatedMediaStreamIds]], return true.
@@ -1585,7 +1585,7 @@ impl RTCPeerConnection {
     pub(crate) async fn start_rtp_senders(&self) -> Result<()> {
         let current_transceivers = self.internal.rtp_transceivers.lock().await;
         for transceiver in &*current_transceivers {
-            let sender = transceiver.sender();
+            let sender = transceiver.sender().await;
             if sender.is_negotiated() && !sender.has_sent() {
                 sender.send(&sender.get_parameters().await).await?;
             }
@@ -1643,7 +1643,7 @@ impl RTCPeerConnection {
         let mut senders = vec![];
         let rtp_transceivers = self.internal.rtp_transceivers.lock().await;
         for transceiver in &*rtp_transceivers {
-            let sender = transceiver.sender();
+            let sender = transceiver.sender().await;
             senders.push(sender);
         }
         senders
@@ -1654,7 +1654,7 @@ impl RTCPeerConnection {
         let mut receivers = vec![];
         let rtp_transceivers = self.internal.rtp_transceivers.lock().await;
         for transceiver in &*rtp_transceivers {
-            receivers.push(transceiver.receiver());
+            receivers.push(transceiver.receiver().await);
         }
         receivers
     }
@@ -1678,7 +1678,7 @@ impl RTCPeerConnection {
             let rtp_transceivers = self.internal.rtp_transceivers.lock().await;
             for t in &*rtp_transceivers {
                 if !t.stopped.load(Ordering::SeqCst) && t.kind == track.kind() {
-                    let sender = t.sender();
+                    let sender = t.sender().await;
                     if sender.track().await.is_none() {
                         if let Err(err) = sender.replace_track(Some(track)).await {
                             let _ = sender.stop().await;
@@ -1705,7 +1705,7 @@ impl RTCPeerConnection {
             .add_rtp_transceiver(Arc::clone(&transceiver))
             .await;
 
-        Ok(transceiver.sender())
+        Ok(transceiver.sender().await)
     }
 
     /// remove_track removes a Track from the PeerConnection
@@ -1718,7 +1718,7 @@ impl RTCPeerConnection {
         {
             let rtp_transceivers = self.internal.rtp_transceivers.lock().await;
             for t in &*rtp_transceivers {
-                if t.sender().id == sender.id {
+                if t.sender().await.id == sender.id {
                     if sender.track().await.is_none() {
                         return Ok(());
                     }

webrtc/src/peer_connection/peer_connection_internal.rs

@@ -166,7 +166,7 @@ impl PeerConnectionInternal {
             self.undeclared_media_processor();
         } else {
             for t in &current_transceivers {
-                let receiver = t.receiver();
+                let receiver = t.receiver().await;
                 let tracks = receiver.tracks().await;
                 if tracks.is_empty() {
                     continue;
@@ -216,7 +216,7 @@ impl PeerConnectionInternal {
                     Arc::clone(&self.media_engine),
                     interceptor,
                 ));
-                t.set_receiver(receiver);
+                t.set_receiver(receiver).await;
             }
         }
 
@@ -337,7 +337,7 @@ impl PeerConnectionInternal {
         for incoming_track in incoming_tracks {
             // If we already have a TrackRemote for a given SSRC don't handle it again
             for t in local_transceivers {
-                let receiver = t.receiver();
+                let receiver = t.receiver().await;
                 for track in receiver.tracks().await {
                     for ssrc in &incoming_track.ssrcs {
                         if *ssrc == track.ssrc() {
@@ -363,7 +363,7 @@ impl PeerConnectionInternal {
                     continue;
                 }
 
-                let receiver = t.receiver();
+                let receiver = t.receiver().await;
                 if receiver.have_received().await {
                     continue;
                 }
@@ -666,7 +666,7 @@ impl PeerConnectionInternal {
             }
 
             // TODO: This is dubious because of rollbacks.
-            t.sender().set_negotiated();
+            t.sender().await.set_negotiated();
             media_sections.push(MediaSection {
                 id: t.mid().unwrap(),
                 transceivers: vec![Arc::clone(t)],
@@ -755,7 +755,7 @@ impl PeerConnectionInternal {
                         }
 
                         if let Some(t) = find_by_mid(mid_value, &mut local_transceivers).await {
-                            t.sender().set_negotiated();
+                            t.sender().await.set_negotiated();
                             let media_transceivers = vec![t];
 
                             // NB: The below could use `then_some`, but with our current MSRV
@@ -780,7 +780,7 @@ impl PeerConnectionInternal {
         // If we are offering also include unmatched local transceivers
         if include_unmatched {
             for t in &local_transceivers {
-                t.sender().set_negotiated();
+                t.sender().await.set_negotiated();
                 media_sections.push(MediaSection {
                     id: t.mid().unwrap(),
                     transceivers: vec![Arc::clone(t)],
@@ -886,7 +886,7 @@ impl PeerConnectionInternal {
             )
             .await?;
 
-        let receiver = t.receiver();
+        let receiver = t.receiver().await;
         PeerConnectionInternal::start_receiver(
             self.setting_engine.get_receive_mtu(),
             &incoming,
@@ -1004,7 +1004,7 @@ impl PeerConnectionInternal {
                     continue;
                 }
 
-                let receiver = t.receiver();
+                let receiver = t.receiver().await;
 
                 if !rsid.is_empty() {
                     return receiver
@@ -1206,7 +1206,7 @@ impl PeerConnectionInternal {
         }
         let mut track_infos = vec![];
         for transeiver in transceivers {
-            let receiver = transeiver.receiver();
+            let receiver = transeiver.receiver().await;
 
             if let Some(mid) = transeiver.mid() {
                 let tracks = receiver.tracks().await;
@@ -1331,7 +1331,7 @@ impl PeerConnectionInternal {
         }
         let mut track_infos = vec![];
         for transceiver in transceivers {
-            let sender = transceiver.sender();
+            let sender = transceiver.sender().await;
 
             let mid = match transceiver.mid() {
                 Some(mid) => mid,

webrtc/src/peer_connection/sdp/mod.rs

@@ -461,7 +461,7 @@ pub(crate) async fn add_transceiver_sdp(
     }
     if codecs.is_empty() {
         // If we are sender and we have no codecs throw an error early
-        if t.sender().track().await.is_some() {
+        if t.sender().await.track().await.is_some() {
             return Err(Error::ErrSenderWithNoCodecs);
         }
 
@@ -528,7 +528,7 @@ pub(crate) async fn add_transceiver_sdp(
     }
 
     for mt in transceivers {
-        let sender = mt.sender();
+        let sender = mt.sender().await;
         if let Some(track) = sender.track().await {
             media = media.with_media_source(
                 sender.ssrc,

webrtc/src/peer_connection/sdp/sdp_test.rs

@@ -642,17 +642,19 @@ async fn test_media_description_fingerprints() -> Result<()> {
             "video".to_owned(),
             "webrtc-rs".to_owned(),
         ));
-        media[i].transceivers[0].set_sender(Arc::new(
-            RTCRtpSender::new(
-                api.setting_engine.get_receive_mtu(),
-                Some(track),
-                Arc::new(RTCDtlsTransport::default()),
-                Arc::clone(&api.media_engine),
-                Arc::clone(&interceptor),
-                false,
-            )
-            .await,
-        ));
+        media[i].transceivers[0]
+            .set_sender(Arc::new(
+                RTCRtpSender::new(
+                    api.setting_engine.get_receive_mtu(),
+                    Some(track),
+                    Arc::new(RTCDtlsTransport::default()),
+                    Arc::clone(&api.media_engine),
+                    Arc::clone(&interceptor),
+                    false,
+                )
+                .await,
+            ))
+            .await;
         media[i].transceivers[0].set_direction_internal(RTCRtpTransceiverDirection::Sendonly);
     }
 

webrtc/src/rtp_transceiver/mod.rs

@@ -30,7 +30,6 @@ pub mod rtp_receiver;
 pub mod rtp_sender;
 pub mod rtp_transceiver_direction;
 pub(crate) mod srtp_writer_future;
-use util::sync::Mutex as SyncMutex;
 
 /// SSRC represents a synchronization source
 /// A synchronization source is a randomly chosen
@@ -174,9 +173,9 @@ pub type TriggerNegotiationNeededFnOption =
 
 /// RTPTransceiver represents a combination of an RTPSender and an RTPReceiver that share a common mid.
 pub struct RTCRtpTransceiver {
-    mid: OnceCell<String>,                    //atomic.Value
-    sender: SyncMutex<Arc<RTCRtpSender>>,     //atomic.Value
-    receiver: SyncMutex<Arc<RTCRtpReceiver>>, //atomic.Value
+    mid: OnceCell<String>,                //atomic.Value
+    sender: Mutex<Arc<RTCRtpSender>>,     //atomic.Value
+    receiver: Mutex<Arc<RTCRtpReceiver>>, //atomic.Value
 
     direction: AtomicU8,         //RTPTransceiverDirection
     current_direction: AtomicU8, //RTPTransceiverDirection
@@ -206,8 +205,8 @@ impl RTCRtpTransceiver {
 
         let t = Arc::new(RTCRtpTransceiver {
             mid: OnceCell::new(),
-            sender: SyncMutex::new(sender),
-            receiver: SyncMutex::new(receiver),
+            sender: Mutex::new(sender),
+            receiver: Mutex::new(receiver),
 
             direction: AtomicU8::new(direction as u8),
             current_direction: AtomicU8::new(RTCRtpTransceiverDirection::Unspecified as u8),
@@ -218,7 +217,9 @@ impl RTCRtpTransceiver {
             media_engine,
             trigger_negotiation_needed: Mutex::new(trigger_negotiation_needed),
         });
-        t.sender().set_rtp_transceiver(Some(Arc::downgrade(&t)));
+        t.sender()
+            .await
+            .set_rtp_transceiver(Some(Arc::downgrade(&t)));
 
         t
     }
@@ -248,8 +249,8 @@ impl RTCRtpTransceiver {
     }
 
     /// sender returns the RTPTransceiver's RTPSender if it has one
-    pub fn sender(&self) -> Arc<RTCRtpSender> {
-        let sender = self.sender.lock();
+    pub async fn sender(&self) -> Arc<RTCRtpSender> {
+        let sender = self.sender.lock().await;
         sender.clone()
     }
 
@@ -259,33 +260,33 @@ impl RTCRtpTransceiver {
         sender: Arc<RTCRtpSender>,
         track: Option<Arc<dyn TrackLocal + Send + Sync>>,
     ) -> Result<()> {
-        self.set_sender(sender);
+        self.set_sender(sender).await;
         self.set_sending_track(track).await
     }
 
-    pub fn set_sender(self: &Arc<Self>, s: Arc<RTCRtpSender>) {
+    pub async fn set_sender(self: &Arc<Self>, s: Arc<RTCRtpSender>) {
         s.set_rtp_transceiver(Some(Arc::downgrade(self)));
 
-        let prev_sender = self.sender();
+        let prev_sender = self.sender().await;
         prev_sender.set_rtp_transceiver(None);
 
         {
-            let mut sender = self.sender.lock();
+            let mut sender = self.sender.lock().await;
             *sender = s;
         }
     }
 
     /// receiver returns the RTPTransceiver's RTPReceiver if it has one
-    pub fn receiver(&self) -> Arc<RTCRtpReceiver> {
-        let receiver = self.receiver.lock();
+    pub async fn receiver(&self) -> Arc<RTCRtpReceiver> {
+        let receiver = self.receiver.lock().await;
         receiver.clone()
     }
 
-    pub(crate) fn set_receiver(&self, r: Arc<RTCRtpReceiver>) {
+    pub(crate) async fn set_receiver(&self, r: Arc<RTCRtpReceiver>) {
         r.set_transceiver_codecs(Some(Arc::clone(&self.codecs)));
 
         {
-            let mut receiver = self.receiver.lock();
+            let mut receiver = self.receiver.lock().await;
             (*receiver).set_transceiver_codecs(None);
 
             *receiver = r;
@@ -396,7 +397,7 @@ impl RTCRtpTransceiver {
         }
 
         {
-            let receiver = self.receiver.lock().clone();
+            let receiver = self.receiver.lock().await.clone();
             let pause_receiver = !current_direction.has_recv();
 
             if pause_receiver {
@@ -408,7 +409,7 @@ impl RTCRtpTransceiver {
 
         let pause_sender = !current_direction.has_send();
         {
-            let sender = &*self.sender.lock();
+            let sender = &*self.sender.lock().await;
             sender.set_paused(pause_sender);
         }
 
@@ -424,11 +425,11 @@ impl RTCRtpTransceiver {
         self.stopped.store(true, Ordering::SeqCst);
 
         {
-            let sender = self.sender.lock().clone();
+            let sender = self.sender.lock().await;
             sender.stop().await?;
         }
         {
-            let r = self.receiver.lock().clone();
+            let r = self.receiver.lock().await;
             r.stop().await?;
         }
 
@@ -443,7 +444,7 @@ impl RTCRtpTransceiver {
     ) -> Result<()> {
         let track_is_none = track.is_none();
         {
-            let sender = self.sender.lock().clone();
+            let sender = self.sender.lock().await.clone();
             sender.replace_track(track).await?;
         }
 

webrtc/src/rtp_transceiver/rtp_sender/rtp_sender_test.rs

@@ -133,7 +133,7 @@ async fn test_rtp_sender_get_parameters() -> Result<()> {
 
     signal_pair(&mut offerer, &mut answerer).await?;
 
-    let sender = rtp_transceiver.sender();
+    let sender = rtp_transceiver.sender().await;
     let parameters = sender.get_parameters().await;
     assert_ne!(0, parameters.rtp_parameters.codecs.len());
     assert_eq!(1, parameters.encodings.len());