Add state, and fix Gstreamer

gst-warp-sink/Cargo.lock

@@ -645,29 +645,6 @@ dependencies = [
  "system-deps",
 ]
 
-[[package]]
-name = "gst-plugin-fmp4"
-version = "0.12.0-alpha.1"
-source = "git+https://github.com/sdroege/gst-plugin-rs#c7f961cc2216f24e282ed81dbed9be7d29d5ecd0"
-dependencies = [
- "anyhow",
- "gst-plugin-version-helper 0.8.0 (git+https://github.com/sdroege/gst-plugin-rs)",
- "gstreamer",
- "gstreamer-audio",
- "gstreamer-base",
- "gstreamer-pbutils",
- "gstreamer-video",
-]
-
-[[package]]
-name = "gst-plugin-version-helper"
-version = "0.8.0"
-source = "git+https://github.com/sdroege/gst-plugin-rs#c7f961cc2216f24e282ed81dbed9be7d29d5ecd0"
-dependencies = [
- "chrono",
- "toml_edit 0.21.0",
-]
-
 [[package]]
 name = "gst-plugin-version-helper"
 version = "0.8.0"
@@ -800,33 +777,6 @@ dependencies = [
  "system-deps",
 ]
 
-[[package]]
-name = "gstreamer-gl"
-version = "0.22.0"
-source = "git+https://gitlab.freedesktop.org/gstreamer/gstreamer-rs#7f234c88acec5f5233342fa01e0c7ffde51bd049"
-dependencies = [
- "glib",
- "gstreamer",
- "gstreamer-base",
- "gstreamer-gl-sys",
- "gstreamer-video",
- "libc",
-]
-
-[[package]]
-name = "gstreamer-gl-sys"
-version = "0.22.0"
-source = "git+https://gitlab.freedesktop.org/gstreamer/gstreamer-rs#7f234c88acec5f5233342fa01e0c7ffde51bd049"
-dependencies = [
- "glib-sys",
- "gobject-sys",
- "gstreamer-base-sys",
- "gstreamer-sys",
- "gstreamer-video-sys",
- "libc",
- "system-deps",
-]
-
 [[package]]
 name = "gstreamer-pbutils"
 version = "0.22.0"
@@ -1077,14 +1027,11 @@ dependencies = [
  "anyhow",
  "bytes",
  "chrono",
- "gst-plugin-fmp4",
+ "gst-plugin-version-helper",
- "gst-plugin-version-helper 0.8.0 (git+https://gitlab.freedesktop.org/gstreamer/gst-plugins-rs)",
  "gstreamer",
  "gstreamer-app",
  "gstreamer-check",
- "gstreamer-gl",
  "gstreamer-pbutils",
- "gstreamer-video",
  "isobmff",
  "moq-transport",
  "mp4",

gst-warp-sink/Cargo.toml

@@ -12,9 +12,6 @@ gst = { package = "gstreamer", git = "https://gitlab.freedesktop.org/gstreamer/g
 gst-app = { package = "gstreamer-app", git = "https://gitlab.freedesktop.org/gstreamer/gstreamer-rs"}
 gst-check = { package = "gstreamer-check", git = "https://gitlab.freedesktop.org/gstreamer/gstreamer-rs", features = ["v1_18"] }
 gst-pbutils = { package = "gstreamer-pbutils", git = "https://gitlab.freedesktop.org/gstreamer/gstreamer-rs", features = ["v1_20"] }
-gst-video = { package = "gstreamer-video", git = "https://gitlab.freedesktop.org/gstreamer/gstreamer-rs" }
-gst-gl = { package = "gstreamer-gl", git = "https://gitlab.freedesktop.org/gstreamer/gstreamer-rs" }
-gst-plugin-fmp4 = { git = "https://github.com/sdroege/gst-plugin-rs", version = "0.12.0-alpha.1" }
 chrono = "0.4.31"
 isobmff = { git = "https://github.com/LMinJae/isobmff-rs", version = "0.1.0" }
 bytes = "1.5.0"

gst-warp-sink/src/moqsink/imp.rs

@@ -3,7 +3,7 @@ use gst::glib::once_cell::sync::Lazy;
 use gst::prelude::*;
 use gst::subclass::prelude::*;
 
-use std::sync::Mutex;
+use std::sync::{Arc, Mutex};
 
 use crate::relayurl::*;
 use crate::RUNTIME;
@@ -63,23 +63,13 @@ impl Default for Settings {
     }
 }
 
-#[derive(Debug)]
-struct StartedState {
-    broadcast: broadcast::Publisher,
-}
-
-impl StartedState {
-    pub fn new(broadcast: broadcast::Publisher) -> StartedState {
-        StartedState { broadcast }
-    }
-}
-
 #[derive(Default)]
 enum State {
     #[default]
     Stopped,
-    Completed,
+    Started {
-    Started(StartedState),
+        broadcast: Mutex<broadcast::Publisher>,
+    },
 }
 
 static CAT: Lazy<gst::DebugCategory> = Lazy::new(|| {
@@ -120,34 +110,37 @@ impl MoqSink {
             unreachable!("Element already started");
         }
 
-        // let relay_url = {
+        let relay_url = {
-        //     let url = self.url.lock().unwrap();
+            let url = self.url.lock().unwrap();
-        //     match *url {
+            match *url {
-        //         Some(ref url) => url.clone(),
+                Some(ref url) => url.clone(),
-        //         None => {
+                None => {
-        //             return Err(gst::error_msg!(
+                    return Err(gst::error_msg!(
-        //                 gst::ResourceError::Settings,
+                        gst::ResourceError::Settings,
-        //                 ["Cannot start without a URL being set"]
+                        ["Cannot start without a URL being set"]
-        //             ));
+                    ));
-        //         }
+                }
-        //     }
+            }
-        // };
+        };
-        let relay_url = self
+
-            .url
+        //More complex but with error handling
-            .lock()
+
-            .map_err(|e| {
+        // let relay_url = self
-                gst::error_msg!(
+        //     .url
-                    gst::ResourceError::Settings,
+        //     .lock()
-                    ["Failed to acquire URL lock: {}", e]
+        //     .map_err(|e| {
-                )
+        //         gst::error_msg!(
-            })?
+        //             gst::ResourceError::Settings,
-            .clone()
+        //             ["Failed to acquire URL lock: {}", e]
-            .ok_or_else(|| {
+        //         )
-                gst::error_msg!(
+        //     })?
-                    gst::ResourceError::Settings,
+        //     .clone()
-                    ["Cannot start without a URL being set"]
+        //     .ok_or_else(|| {
-                )
+        //         gst::error_msg!(
-            })?;
+        //             gst::ResourceError::Settings,
+        //             ["Cannot start without a URL being set"]
+        //         )
+        //     })?;
 
         gst::trace!(
             CAT,
@@ -156,9 +149,7 @@ impl MoqSink {
             relay_url
         );
 
-        // Initialize shared state and channels
+        let (publisher, subscriber) = broadcast::new("");
-        let (sender, receiver) = mpsc::channel(32);
-        // self.sender = Some(sender);
 
         // Spawn a new thread to run the Moq server
         RUNTIME.spawn(async move {
@@ -167,8 +158,6 @@ impl MoqSink {
                 .finish();
             tracing::subscriber::set_global_default(tracer).unwrap();
 
-            let (publisher, subscriber) = broadcast::new("");
-
             // Create a list of acceptable root certificates.
             let mut roots = rustls::RootCertStore::empty();
 
@@ -210,9 +199,9 @@ impl MoqSink {
         });
 
         // Update the state to indicate the element has started
-        // *state = State::Started(StartedState {
+        *state = Some(State::Started {
-        //     broadcast: publisher,
+            broadcast: Mutex::new(publisher),
-        // });
+        });
 
         Ok(())
     }
@@ -247,6 +236,7 @@ impl MoqSink {
     }
 }
 
+#[glib::object_subclass]
 impl ObjectSubclass for MoqSink {
     const NAME: &'static str = ELEMENT_CLASS_NAME;
     type Type = super::MoqSink;