From 0954af10c7e7c839f3c400eb8a7948f284db3213 Mon Sep 17 00:00:00 2001
From: Mathieu Duponchelle <mathieu@centricular.com>
Date: Thu, 8 Jun 2023 01:51:45 +0200
Subject: [PATCH] webrtc/signalling: fix race condition in message ordering

Spawning one task per message to send out instead of sending them out
sequentially from the one task used to poll the handler sometimes
resulted in peers receiving ICE candidates before SDP offers, triggering
hard to understand errors in the browser.

Part-of: <https://gitlab.freedesktop.org/gstreamer/gst-plugins-rs/-/merge_requests/1254>
---
 net/webrtc/signalling/src/server/mod.rs | 21 ++++++++++++++-------
 1 file changed, 14 insertions(+), 7 deletions(-)

diff --git a/net/webrtc/signalling/src/server/mod.rs b/net/webrtc/signalling/src/server/mod.rs
index 3e774ddc..eaafff62 100644
--- a/net/webrtc/signalling/src/server/mod.rs
+++ b/net/webrtc/signalling/src/server/mod.rs
@@ -38,7 +38,7 @@ impl Server {
     #[instrument(level = "debug", skip(factory))]
     pub fn spawn<
         I: for<'a> Deserialize<'a>,
-        O: Serialize + std::fmt::Debug,
+        O: Serialize + std::fmt::Debug + Send + Sync,
         Factory: FnOnce(Pin<Box<dyn Stream<Item = (String, Option<I>)> + Send>>) -> St,
         St: Stream<Item = (String, O)>,
     >(
@@ -72,12 +72,19 @@ impl Server {
         let _ = task::spawn(async move {
             while let Some((peer_id, msg)) = handler.next().await {
                 match serde_json::to_string(&msg) {
-                    Ok(msg) => {
-                        if let Some(peer) = state_clone.lock().unwrap().peers.get_mut(&peer_id) {
-                            let mut sender = peer.sender.clone();
-                            task::spawn(async move {
-                                let _ = sender.send(msg).await;
-                            });
+                    Ok(msg_str) => {
+                        let sender = {
+                            let mut state = state_clone.lock().unwrap();
+                            if let Some(peer) = state.peers.get_mut(&peer_id) {
+                                Some(peer.sender.clone())
+                            } else {
+                                None
+                            }
+                        };
+
+                        if let Some(mut sender) = sender {
+                            trace!("Sending {}", msg_str);
+                            let _ = sender.send(msg_str).await;
                         }
                     }
                     Err(err) => {