From 271b5838769940c05a52d038d3bb8a533d247bb3 Mon Sep 17 00:00:00 2001
From: Mathieu Duponchelle <mathieu@centricular.com>
Date: Tue, 6 Jun 2023 23:55:31 +0200
Subject: [PATCH] webrtcsink: avoid panic on unprepare from an async tokio
 context

.. and log an error with advice on how to dispose of elements properly
from a tokio runtime.

Part-of: <https://gitlab.freedesktop.org/gstreamer/gst-plugins-rs/-/merge_requests/1218>
---
 net/webrtc/src/webrtcsink/imp.rs | 26 +++++++++++++++++++++++---
 1 file changed, 23 insertions(+), 3 deletions(-)

diff --git a/net/webrtc/src/webrtcsink/imp.rs b/net/webrtc/src/webrtcsink/imp.rs
index c260993a..cdc46131 100644
--- a/net/webrtc/src/webrtcsink/imp.rs
+++ b/net/webrtc/src/webrtcsink/imp.rs
@@ -17,7 +17,7 @@ use once_cell::sync::Lazy;
 use std::collections::HashMap;
 
 use std::ops::Mul;
-use std::sync::{Arc, Condvar, Mutex};
+use std::sync::{mpsc, Arc, Condvar, Mutex};
 
 use super::homegrown_cc::CongestionController;
 use super::{WebRTCSinkCongestionControl, WebRTCSinkError, WebRTCSinkMitigationMode};
@@ -1387,7 +1387,7 @@ impl BaseWebRTCSink {
 
         let codecs_done_receiver = std::mem::take(&mut state.codecs_done_receivers);
         codecs_done_receiver.into_iter().for_each(|receiver| {
-            RUNTIME.spawn(async {
+            RUNTIME.block_on(async {
                 let _ = receiver.await;
             });
         });
@@ -3609,7 +3609,27 @@ impl ElementImpl for BaseWebRTCSink {
 
         match transition {
             gst::StateChange::PausedToReady => {
-                if let Err(err) = self.unprepare(&element) {
+                let unprepare_res = match tokio::runtime::Handle::try_current() {
+                    Ok(_) => {
+                        gst::error!(
+                            CAT,
+                            obj: element,
+                            "Trying to set state to NULL from an async \
+                                    tokio context, working around the panic but \
+                                    you should refactor your code to make use of \
+                                    gst::Element::call_async and set the state to \
+                                    NULL from there, without blocking the runtime"
+                        );
+                        let (tx, rx) = mpsc::channel();
+                        element.call_async(move |element| {
+                            tx.send(element.imp().unprepare(element)).unwrap();
+                        });
+                        rx.recv().unwrap()
+                    }
+                    Err(_) => self.unprepare(&element),
+                };
+
+                if let Err(err) = unprepare_res {
                     gst::element_error!(
                         element,
                         gst::StreamError::Failed,