janusvr_signaller: Do not block in end_session()

Only stop() is allowed to block, wait there.

Fixes #603

Part-of: <https://gitlab.freedesktop.org/gstreamer/gst-plugins-rs/-/merge_requests/1848>

net/webrtc/src/janusvr_signaller/imp.rs

@@ -271,6 +271,7 @@ struct State {
     transaction_id: Option<String>,
     room_id: Option<JanusId>,
     feed_id: Option<JanusId>,
+    leave_room_rx: Option<tokio::sync::oneshot::Receiver<()>>,
 }
 
 #[derive(Clone)]
@@ -568,20 +569,6 @@ impl Signaller {
         }
     }
 
-    // Only used at the end when cleaning up the resources.
-    // So that `SignallableImpl::stop` waits the last message
-    // to be sent properly.
-    fn send_blocking(&self, msg: OutgoingMessage) {
-        let state = self.state.lock().unwrap();
-        if let Some(mut sender) = state.ws_sender.clone() {
-            RUNTIME.block_on(async {
-                if let Err(err) = sender.send(msg).await {
-                    self.raise_error(err.to_string());
-                }
-            });
-        }
-    }
-
     fn set_transaction_id(&self, transaction: String) {
         self.state.lock().unwrap().transaction_id = Some(transaction);
     }
@@ -677,8 +664,8 @@ impl Signaller {
     }
 
     fn leave_room(&self) {
+        let mut state = self.state.lock().unwrap();
         let (transaction, session_id, handle_id, room, feed_id, display, apisecret) = {
-            let state = self.state.lock().unwrap();
             let settings = self.settings.lock().unwrap();
 
             if settings.room_id.is_none() {
@@ -696,7 +683,10 @@ impl Signaller {
                 settings.secret_key.clone(),
             )
         };
-        self.send_blocking(OutgoingMessage::RoomRequest(RoomRequestMsg {
+        if let Some(mut sender) = state.ws_sender.clone() {
+            let (tx, rx) = tokio::sync::oneshot::channel::<()>();
+            state.leave_room_rx = Some(rx);
+            let msg = OutgoingMessage::RoomRequest(RoomRequestMsg {
                 janus: "message".to_string(),
                 transaction,
                 session_id,
@@ -709,7 +699,18 @@ impl Signaller {
                     id: Some(feed_id),
                     display,
                 },
-        }));
+            });
+            RUNTIME.spawn(glib::clone!(
+                #[to_owned(rename_to = this)]
+                self,
+                async move {
+                    if let Err(err) = sender.send(msg).await {
+                        this.raise_error(err.to_string());
+                    }
+                    let _ = tx.send(());
+                }
+            ));
+        }
     }
 
     fn publish(&self, offer: &gst_webrtc::WebRTCSessionDescription) {
@@ -873,6 +874,12 @@ impl SignallableImpl for Signaller {
             });
         }
 
+        if let Some(rx) = state.leave_room_rx.take() {
+            RUNTIME.block_on(async move {
+                let _ = rx.await;
+            });
+        }
+
         state.session_id = None;
         state.handle_id = None;
         state.transaction_id = None;

net/webrtc/src/webrtcsink/imp.rs

@@ -3707,12 +3707,6 @@ impl BaseWebRTCSink {
                             drop(settings);
                             signaller.end_session(session_id);
                         }
-
-                        gst::warning!(
-                            CAT,
-                            imp = self,
-                            "Consumer refused media {session_id}, {media_idx}"
-                        );
                         return;
                     }
                 }