src: stream: Post EOS from a new thread

src/stream/mod.rs

@@ -287,9 +287,13 @@ impl Drop for StreamState {
         let pipeline_state = self.pipeline.inner_state_as_ref();
         let pipeline = &pipeline_state.pipeline;
 
-        if let Err(error) = pipeline.post_message(::gst::message::Eos::new()) {
-            error!("Failed posting Eos message into Pipeline bus. Reason: {error:?}");
-        }
+        let pipeline_weak = pipeline.downgrade();
+        std::thread::spawn(move || {
+            let pipeline = pipeline_weak.upgrade().unwrap();
+            if let Err(error) = pipeline.post_message(::gst::message::Eos::new()) {
+                error!("Failed posting Eos message into Pipeline bus. Reason: {error:?}");
+            }
+        });
 
         if let Err(error) = pipeline.set_state(::gst::State::Null) {
             error!("Failed setting Pipeline state to Null. Reason: {error:?}");

src/stream/sink/image_sink.rs

@@ -284,9 +284,13 @@ impl SinkInterface for ImageSink {
 
     #[instrument(level = "debug", skip(self))]
     fn eos(&self) {
-        if let Err(error) = self.pipeline.post_message(gst::message::Eos::new()) {
-            error!("Failed posting Eos message into Sink bus. Reason: {error:?}");
-        }
+        let pipeline_weak = self.pipeline.downgrade();
+        std::thread::spawn(move || {
+            let pipeline = pipeline_weak.upgrade().unwrap();
+            if let Err(error) = pipeline.post_message(gst::message::Eos::new()) {
+                error!("Failed posting Eos message into Sink bus. Reason: {error:?}");
+            }
+        });
     }
 }
 

src/stream/sink/udp_sink.rs

@@ -264,9 +264,13 @@ impl SinkInterface for UdpSink {
 
     #[instrument(level = "debug", skip(self))]
     fn eos(&self) {
-        if let Err(error) = self.pipeline.post_message(gst::message::Eos::new()) {
-            error!("Failed posting Eos message into Sink bus. Reason: {error:?}");
-        }
+        let pipeline_weak = self.pipeline.downgrade();
+        std::thread::spawn(move || {
+            let pipeline = pipeline_weak.upgrade().unwrap();
+            if let Err(error) = pipeline.post_message(gst::message::Eos::new()) {
+                error!("Failed posting Eos message into Sink bus. Reason: {error:?}");
+            }
+        });
     }
 }
 

src/stream/sink/webrtc_sink.rs

@@ -295,9 +295,13 @@ impl SinkInterface for WebRTCSink {
 
     #[instrument(level = "debug", skip(self))]
     fn eos(&self) {
-        if let Err(error) = self.webrtcbin.post_message(gst::message::Eos::new()) {
-            error!("Failed posting Eos message into Sink bus. Reason: {error:?}");
-        }
+        let webrtcbin_weak = self.webrtcbin.downgrade();
+        std::thread::spawn(move || {
+            let webrtcbin = webrtcbin_weak.upgrade().unwrap();
+            if let Err(error) = webrtcbin.post_message(gst::message::Eos::new()) {
+                error!("Failed posting Eos message into Sink bus. Reason: {error:?}");
+            }
+        });
     }
 }