pivoted to mediamtx whip webrtc

Cargo.lock

@@ -3250,10 +3250,9 @@ version = "0.1.0"
 dependencies = [
  "anyhow",
  "config",
+ "ctrlc",
  "gst-plugin-webrtc",
  "gstreamer",
- "gstreamer-sdp",
- "gstreamer-webrtc",
  "log",
  "serde",
  "serde_json",

Cargo.toml

@@ -8,10 +8,9 @@ edition = "2021"
 [dependencies]
 anyhow = "1.0.86"
 config = "0.14.0"
-gst-plugin-webrtc = "0.13.0"
+ctrlc = "3.4.4"
+gst-plugin-webrtc = { version = "0.13.0", features = ["v1_22"] }
 gstreamer = { version = "0.23.0", features = ["v1_22"] }
-gstreamer-sdp = { version = "0.23.0", features = ["v1_22"] }
-gstreamer-webrtc = { version = "0.23.0", features = ["v1_22"] }
 log = "0.4.22"
 serde = { version = "1.0.204", features = ["derive"] }
 serde_json = "1.0.122"

flake.nix

@@ -40,8 +40,10 @@ Some utility commands:
           gst_all_1.gstreamer
           gst_all_1.gst-plugins-base
           gst_all_1.gst-plugins-good
-          gst_all_1.gst-plugins-bad # contains gstreamer-webrtc-1.0
+          gst_all_1.gst-plugins-bad # contains gst-webrtc 
+          gst_all_1.gst-plugins-ugly
           gst_all_1.gst-plugins-rs
+          mediamtx
         ];
         cargoHash = nixpkgs.lib.fakeHash;
       };

rtsp-simple-server.yml

@@ -0,0 +1,3 @@
+paths:
+  mystream:
+    source: publisher

src/main.rs

@@ -1,41 +1,68 @@
-mod signaller;
+use std::sync::{atomic::AtomicBool, Arc};
-
-// from outside the plugin repository, one would need to add plugin package as follows:
-// [dependencies]
-// gstrswebrtc = { package = "gst-plugin-webrtc", git = "https://gitlab.freedesktop.org/gstreamer/gst-plugins-rs/" }
-use gstrswebrtc;
 
 use anyhow::Error;
-use gstreamer::prelude::*;
+use gstreamer::{prelude::*, Element, ElementFactory, Pipeline, State};
-use gstrswebrtc::signaller as signaller_interface;
+use tracing::info;
-use gstrswebrtc::webrtcsink;
+use gstrswebrtc::{signaller::Signallable, webrtcsink::WhipWebRTCSink};
+
 
 fn main() -> Result<(), Error> {
+
+    let sub = tracing_subscriber::FmtSubscriber::new();
+    if let Err(e) = tracing::subscriber::set_global_default(sub) {
+        panic!("Could not set tracing global: {e}");
+    }
+
+
+    let to_quit = Arc::new(AtomicBool::new(false));
+    let to_quit_2 = to_quit.clone();
+
+    ctrlc::set_handler(move || {
+        to_quit_2.store(true, std::sync::atomic::Ordering::SeqCst);
+    })?;
+
     gstreamer::init()?;
 
-    let custom_signaller = signaller::MyCustomSignaller::new();
+    let pipeline = Pipeline::with_name("rstp-pipeline");
-    let webrtcsink = webrtcsink::BaseWebRTCSink::with_signaller(
-        signaller_interface::Signallable::from(custom_signaller),
-    );
 
-    let pipeline = gstreamer::Pipeline::new();
+    // let source = ElementFactory::make("videotestsrc")
+    let source = ElementFactory::make("videotestsrc")
+        .build().unwrap();
 
-    let video_src = gstreamer::ElementFactory::make("videotestsrc").build().unwrap();
+    let convert = ElementFactory::make("x264enc")
+        .build().unwrap();
 
-    pipeline
+    let whipsink = gstreamer::ElementFactory::make("whipclientsink")
-        .add_many([&video_src, webrtcsink.upcast_ref()])
+        .name("whip-sink_123059")
-        .unwrap();
+        .build()?;
-    video_src
-        .link(webrtcsink.upcast_ref::<gstreamer::Element>())
-        .unwrap();
 
-    let bus = pipeline.bus().unwrap();
+    // THIS IS THE LINE CAUSING THE ERROR
+    if let Some(whipsink) = whipsink.dynamic_cast_ref::<WhipWebRTCSink>() {
+        //let signaller = whipsink.property::<Signallable>("signaller");
+        //signaller.set_property_from_str(
+        //    "whip-endpoint",
+        //    &format!("http://localhost:{}/whip_sink/{}", 8889, "mystream"),
+        //);
+    }
 
-    pipeline.set_state(gstreamer::State::Playing).unwrap();
+    pipeline.add_many([&source, &convert, &whipsink]).unwrap();
 
-    let _msg = bus.timed_pop_filtered(gstreamer::ClockTime::NONE, &[gstreamer::MessageType::Eos]);
+    Element::link_many([&source, &convert, &whipsink]).unwrap();
 
-    pipeline.set_state(gstreamer::State::Null).unwrap();
+    pipeline.set_state(State::Playing).unwrap();
+
+    loop {
+        if to_quit.load(std::sync::atomic::Ordering::SeqCst) {
+            println!("Recieved Ctrl+C, stopping");
+
+            pipeline.set_state(State::Null).unwrap();
+            break;
+        }
+
+        std::thread::sleep(std::time::Duration::from_millis(200));
+    }
+
+    println!("Success!");
 
     Ok(())
 }

src/signaller/imp.rs

@@ -1,48 +0,0 @@
-use gstreamer::glib;
-use gstreamer::subclass::prelude::*;
-use gstreamer_webrtc::WebRTCSessionDescription;
-
-use gstrswebrtc::signaller::{Signallable, SignallableImpl};
-
-#[derive(Default)]
-pub struct Signaller {}
-
-impl Signaller {}
-
-impl SignallableImpl for Signaller {
-    fn start(&self) {
-        unimplemented!()
-    }
-
-    fn stop(&self) {
-        unimplemented!()
-    }
-
-    fn send_sdp(&self, _session_id: &str, _sdp: &WebRTCSessionDescription) {
-        unimplemented!()
-    }
-
-    fn add_ice(
-        &self,
-        _session_id: &str,
-        _candidate: &str,
-        _sdp_m_line_index: u32,
-        _sdp_mid: Option<String>,
-    ) {
-        unimplemented!()
-    }
-
-    fn end_session(&self, _session_id: &str) {
-        unimplemented!()
-    }
-}
-
-#[glib::object_subclass]
-impl ObjectSubclass for Signaller {
-    const NAME: &'static str = "MyCustomWebRTCSinkSignaller";
-    type Type = super::MyCustomSignaller;
-    type ParentType = glib::Object;
-    type Interfaces = (Signallable,);
-}
-
-impl ObjectImpl for Signaller {}

src/signaller/mod.rs

@@ -1,20 +0,0 @@
-use gstreamer::glib;
-use gstrswebrtc::signaller::Signallable;
-
-mod imp;
-
-glib::wrapper! {
-    pub struct MyCustomSignaller(ObjectSubclass<imp::Signaller>) @implements Signallable;
-}
-
-impl MyCustomSignaller {
-    pub fn new() -> Self {
-        glib::Object::new()
-    }
-}
-
-impl Default for MyCustomSignaller {
-    fn default() -> Self {
-        MyCustomSignaller::new()
-    }
-}