Fix deadlock in on_track callback (#131)

Additionally:
- fix `set_send`, `set_recv` race
- fix false track deduplication

Co-authored-by: alexlapa <[email protected]>

Author: rogurotus

.github/workflows/ci.yml

@@ -247,12 +247,6 @@ jobs:
         if: ${{ matrix.platform != 'android'
              && matrix.platform != 'ios' }}
 
-      - name: Set up JDK 11
-        uses: actions/setup-java@v3
-        with:
-          distribution: zulu
-          java-version: 11
-        if: ${{ matrix.platform == 'android' }}
       - name: Test on `${{ matrix.platform }}` platform with emulator
         uses: reactivecircus/android-emulator-runner@v2
         with:

crates/native/src/api.rs

@@ -11,7 +11,9 @@ use libwebrtc_sys as sys;
 
 use crate::{
     devices::{self, DeviceState},
+    pc::PeerConnectionId,
     renderer::FrameHandler,
+    user_media::TrackOrigin,
     Webrtc,
 };
 
@@ -1657,6 +1659,12 @@ pub struct MediaStreamTrack {
     /// Unique identifier (GUID) of this [`MediaStreamTrack`].
     pub id: String,
 
+    /// Unique identifier of the [`PeerConnection`] from which this
+    /// [`MediaStreamTrack`] was received.
+    ///
+    /// Always [`None`] for local [`MediaStreamTrack`]s.
+    pub peer_id: Option<u64>,
+
     /// Label identifying the track source, as in "internal microphone".
     pub device_id: String,
 
@@ -2170,8 +2178,13 @@ pub fn microphone_volume() -> anyhow::Result<u32> {
 }
 
 /// Disposes the specified [`MediaStreamTrack`].
-pub fn dispose_track(track_id: String, kind: MediaType) {
-    WEBRTC.lock().unwrap().dispose_track(track_id, kind);
+pub fn dispose_track(track_id: String, peer_id: Option<u64>, kind: MediaType) {
+    let track_origin = TrackOrigin::from(peer_id.map(PeerConnectionId::from));
+
+    WEBRTC
+        .lock()
+        .unwrap()
+        .dispose_track(track_origin, track_id, kind);
 }
 
 /// Returns the [readyState][0] property of the [`MediaStreamTrack`] by its ID
@@ -2180,9 +2193,15 @@ pub fn dispose_track(track_id: String, kind: MediaType) {
 /// [0]: https://w3.org/TR/mediacapture-streams#dfn-readystate
 pub fn track_state(
     track_id: String,
+    peer_id: Option<u64>,
     kind: MediaType,
 ) -> anyhow::Result<TrackState> {
-    WEBRTC.lock().unwrap().track_state(track_id, kind)
+    let track_origin = TrackOrigin::from(peer_id.map(PeerConnectionId::from));
+
+    WEBRTC
+        .lock()
+        .unwrap()
+        .track_state(track_id, track_origin, kind)
 }
 
 /// Changes the [enabled][1] property of the [`MediaStreamTrack`] by its ID and
@@ -2191,33 +2210,49 @@ pub fn track_state(
 /// [1]: https://w3.org/TR/mediacapture-streams#track-enabled
 pub fn set_track_enabled(
     track_id: String,
+    peer_id: Option<u64>,
     kind: MediaType,
     enabled: bool,
 ) -> anyhow::Result<()> {
-    WEBRTC
-        .lock()
-        .unwrap()
-        .set_track_enabled(track_id, kind, enabled)
+    let track_origin = TrackOrigin::from(peer_id.map(PeerConnectionId::from));
+
+    WEBRTC.lock().unwrap().set_track_enabled(
+        track_id,
+        track_origin,
+        kind,
+        enabled,
+    )
 }
 
 /// Clones the specified [`MediaStreamTrack`].
 pub fn clone_track(
     track_id: String,
+    peer_id: Option<u64>,
     kind: MediaType,
 ) -> anyhow::Result<MediaStreamTrack> {
-    WEBRTC.lock().unwrap().clone_track(track_id, kind)
+    let track_origin = TrackOrigin::from(peer_id.map(PeerConnectionId::from));
+
+    WEBRTC
+        .lock()
+        .unwrap()
+        .clone_track(track_id, track_origin, kind)
 }
 
 /// Registers an observer to the [`MediaStreamTrack`] events.
 pub fn register_track_observer(
     cb: StreamSink<TrackEvent>,
+    peer_id: Option<u64>,
     track_id: String,
     kind: MediaType,
 ) -> anyhow::Result<()> {
-    WEBRTC
-        .lock()
-        .unwrap()
-        .register_track_observer(track_id, kind, cb.into())
+    let track_origin = TrackOrigin::from(peer_id.map(PeerConnectionId::from));
+
+    WEBRTC.lock().unwrap().register_track_observer(
+        track_id,
+        track_origin,
+        kind,
+        cb.into(),
+    )
 }
 
 /// Sets the provided [`OnDeviceChangeCallback`] as the callback to be called
@@ -2241,16 +2276,20 @@ pub fn set_on_device_changed(cb: StreamSink<()>) -> anyhow::Result<()> {
 pub fn create_video_sink(
     cb: StreamSink<TextureEvent>,
     sink_id: i64,
+    peer_id: Option<u64>,
     track_id: String,
     callback_ptr: u64,
     texture_id: i64,
 ) -> anyhow::Result<()> {
     let handler = FrameHandler::new(callback_ptr as _, cb.into(), texture_id);
-
-    WEBRTC
-        .lock()
-        .unwrap()
-        .create_video_sink(sink_id, track_id, handler)
+    let track_origin = TrackOrigin::from(peer_id.map(PeerConnectionId::from));
+
+    WEBRTC.lock().unwrap().create_video_sink(
+        sink_id,
+        track_id,
+        track_origin,
+        handler,
+    )
 }
 
 /// Destroys the [`VideoSink`] by the provided ID.

crates/native/src/bridge_generated.rs

@@ -587,6 +587,7 @@ fn wire_microphone_volume_impl(port_: MessagePort) {
 fn wire_dispose_track_impl(
     port_: MessagePort,
     track_id: impl Wire2Api<String> + UnwindSafe,
+    peer_id: impl Wire2Api<Option<u64>> + UnwindSafe,
     kind: impl Wire2Api<MediaType> + UnwindSafe,
 ) {
     FLUTTER_RUST_BRIDGE_HANDLER.wrap::<_, _, _, ()>(
@@ -597,14 +598,18 @@ fn wire_dispose_track_impl(
         },
         move || {
             let api_track_id = track_id.wire2api();
+            let api_peer_id = peer_id.wire2api();
             let api_kind = kind.wire2api();
-            move |task_callback| Ok(dispose_track(api_track_id, api_kind))
+            move |task_callback| {
+                Ok(dispose_track(api_track_id, api_peer_id, api_kind))
+            }
         },
     )
 }
 fn wire_track_state_impl(
     port_: MessagePort,
     track_id: impl Wire2Api<String> + UnwindSafe,
+    peer_id: impl Wire2Api<Option<u64>> + UnwindSafe,
     kind: impl Wire2Api<MediaType> + UnwindSafe,
 ) {
     FLUTTER_RUST_BRIDGE_HANDLER.wrap::<_, _, _, TrackState>(
@@ -615,14 +620,18 @@ fn wire_track_state_impl(
         },
         move || {
             let api_track_id = track_id.wire2api();
+            let api_peer_id = peer_id.wire2api();
             let api_kind = kind.wire2api();
-            move |task_callback| track_state(api_track_id, api_kind)
+            move |task_callback| {
+                track_state(api_track_id, api_peer_id, api_kind)
+            }
         },
     )
 }
 fn wire_set_track_enabled_impl(
     port_: MessagePort,
     track_id: impl Wire2Api<String> + UnwindSafe,
+    peer_id: impl Wire2Api<Option<u64>> + UnwindSafe,
     kind: impl Wire2Api<MediaType> + UnwindSafe,
     enabled: impl Wire2Api<bool> + UnwindSafe,
 ) {
@@ -634,17 +643,24 @@ fn wire_set_track_enabled_impl(
         },
         move || {
             let api_track_id = track_id.wire2api();
+            let api_peer_id = peer_id.wire2api();
             let api_kind = kind.wire2api();
             let api_enabled = enabled.wire2api();
             move |task_callback| {
-                set_track_enabled(api_track_id, api_kind, api_enabled)
+                set_track_enabled(
+                    api_track_id,
+                    api_peer_id,
+                    api_kind,
+                    api_enabled,
+                )
             }
         },
     )
 }
 fn wire_clone_track_impl(
     port_: MessagePort,
     track_id: impl Wire2Api<String> + UnwindSafe,
+    peer_id: impl Wire2Api<Option<u64>> + UnwindSafe,
     kind: impl Wire2Api<MediaType> + UnwindSafe,
 ) {
     FLUTTER_RUST_BRIDGE_HANDLER.wrap::<_, _, _, MediaStreamTrack>(
@@ -655,13 +671,17 @@ fn wire_clone_track_impl(
         },
         move || {
             let api_track_id = track_id.wire2api();
+            let api_peer_id = peer_id.wire2api();
             let api_kind = kind.wire2api();
-            move |task_callback| clone_track(api_track_id, api_kind)
+            move |task_callback| {
+                clone_track(api_track_id, api_peer_id, api_kind)
+            }
         },
     )
 }
 fn wire_register_track_observer_impl(
     port_: MessagePort,
+    peer_id: impl Wire2Api<Option<u64>> + UnwindSafe,
     track_id: impl Wire2Api<String> + UnwindSafe,
     kind: impl Wire2Api<MediaType> + UnwindSafe,
 ) {
@@ -672,11 +692,13 @@ fn wire_register_track_observer_impl(
             mode: FfiCallMode::Stream,
         },
         move || {
+            let api_peer_id = peer_id.wire2api();
             let api_track_id = track_id.wire2api();
             let api_kind = kind.wire2api();
             move |task_callback| {
                 register_track_observer(
                     task_callback.stream_sink::<_, TrackEvent>(),
+                    api_peer_id,
                     api_track_id,
                     api_kind,
                 )
@@ -701,6 +723,7 @@ fn wire_set_on_device_changed_impl(port_: MessagePort) {
 fn wire_create_video_sink_impl(
     port_: MessagePort,
     sink_id: impl Wire2Api<i64> + UnwindSafe,
+    peer_id: impl Wire2Api<Option<u64>> + UnwindSafe,
     track_id: impl Wire2Api<String> + UnwindSafe,
     callback_ptr: impl Wire2Api<u64> + UnwindSafe,
     texture_id: impl Wire2Api<i64> + UnwindSafe,
@@ -713,13 +736,15 @@ fn wire_create_video_sink_impl(
         },
         move || {
             let api_sink_id = sink_id.wire2api();
+            let api_peer_id = peer_id.wire2api();
             let api_track_id = track_id.wire2api();
             let api_callback_ptr = callback_ptr.wire2api();
             let api_texture_id = texture_id.wire2api();
             move |task_callback| {
                 create_video_sink(
                     task_callback.stream_sink::<_, TextureEvent>(),
                     api_sink_id,
+                    api_peer_id,
                     api_track_id,
                     api_callback_ptr,
                     api_texture_id,
@@ -1055,6 +1080,7 @@ impl support::IntoDart for MediaStreamTrack {
     fn into_dart(self) -> support::DartAbi {
         vec![
             self.id.into_into_dart().into_dart(),
+            self.peer_id.into_dart(),
             self.device_id.into_into_dart().into_dart(),
             self.kind.into_into_dart().into_dart(),
             self.enabled.into_into_dart().into_dart(),
@@ -1950,46 +1976,51 @@ mod io {
     pub extern "C" fn wire_dispose_track(
         port_: i64,
         track_id: *mut wire_uint_8_list,
+        peer_id: *mut u64,
         kind: i32,
     ) {
-        wire_dispose_track_impl(port_, track_id, kind)
+        wire_dispose_track_impl(port_, track_id, peer_id, kind)
     }
 
     #[no_mangle]
     pub extern "C" fn wire_track_state(
         port_: i64,
         track_id: *mut wire_uint_8_list,
+        peer_id: *mut u64,
         kind: i32,
     ) {
-        wire_track_state_impl(port_, track_id, kind)
+        wire_track_state_impl(port_, track_id, peer_id, kind)
     }
 
     #[no_mangle]
     pub extern "C" fn wire_set_track_enabled(
         port_: i64,
         track_id: *mut wire_uint_8_list,
+        peer_id: *mut u64,
         kind: i32,
         enabled: bool,
     ) {
-        wire_set_track_enabled_impl(port_, track_id, kind, enabled)
+        wire_set_track_enabled_impl(port_, track_id, peer_id, kind, enabled)
     }
 
     #[no_mangle]
     pub extern "C" fn wire_clone_track(
         port_: i64,
         track_id: *mut wire_uint_8_list,
+        peer_id: *mut u64,
         kind: i32,
     ) {
-        wire_clone_track_impl(port_, track_id, kind)
+        wire_clone_track_impl(port_, track_id, peer_id, kind)
     }
 
     #[no_mangle]
     pub extern "C" fn wire_register_track_observer(
         port_: i64,
+        peer_id: *mut u64,
         track_id: *mut wire_uint_8_list,
         kind: i32,
     ) {
-        wire_register_track_observer_impl(port_, track_id, kind)
+        wire_register_track_observer_impl(port_, peer_id, track_id, kind)
     }
 
     #[no_mangle]
@@ -2001,13 +2032,15 @@ mod io {
     pub extern "C" fn wire_create_video_sink(
         port_: i64,
         sink_id: i64,
+        peer_id: *mut u64,
         track_id: *mut wire_uint_8_list,
         callback_ptr: u64,
         texture_id: i64,
     ) {
         wire_create_video_sink_impl(
             port_,
             sink_id,
+            peer_id,
             track_id,
             callback_ptr,
             texture_id,
@@ -2085,6 +2118,11 @@ mod io {
         support::new_leak_box_ptr(wire_RtcConfiguration::new_with_null_ptr())
     }
 
+    #[no_mangle]
+    pub extern "C" fn new_box_autoadd_u64_0(value: u64) -> *mut u64 {
+        support::new_leak_box_ptr(value)
+    }
+
     #[no_mangle]
     pub extern "C" fn new_box_autoadd_video_constraints_0(
     ) -> *mut wire_VideoConstraints {
@@ -2261,6 +2299,11 @@ mod io {
             Wire2Api::<RtcConfiguration>::wire2api(*wrap).into()
         }
     }
+    impl Wire2Api<u64> for *mut u64 {
+        fn wire2api(self) -> u64 {
+            unsafe { *support::box_from_leak_ptr(self) }
+        }
+    }
     impl Wire2Api<VideoConstraints> for *mut wire_VideoConstraints {
         fn wire2api(self) -> VideoConstraints {
             let wrap = unsafe { support::box_from_leak_ptr(self) };