Add initial version of NDI sink

Fixes teltek#10

src/lib.rs

@@ -3,6 +3,8 @@ use glib::prelude::*;
 mod device_provider;
 pub mod ndi;
 mod ndiaudiosrc;
+mod ndisink;
+pub mod ndisinkmeta;
 pub mod ndisys;
 mod ndivideosrc;
 pub mod receiver;
@@ -35,6 +37,7 @@ fn plugin_init(plugin: &gst::Plugin) -> Result<(), glib::BoolError> {
 
     ndivideosrc::register(plugin)?;
     ndiaudiosrc::register(plugin)?;
+    ndisink::register(plugin)?;
     device_provider::register(plugin)?;
     Ok(())
 }

src/ndisink.rs

@@ -0,0 +1,276 @@
+use glib::subclass;
+use glib::subclass::prelude::*;
+use gst::prelude::*;
+use gst::subclass::prelude::*;
+use gst::{gst_debug, gst_error, gst_error_msg, gst_info, gst_loggable_error, gst_trace};
+use gst_base::subclass::prelude::*;
+
+use std::convert::TryFrom;
+use std::sync::Mutex;
+
+use once_cell::sync::Lazy;
+
+use super::ndi::SendInstance;
+
+static DEFAULT_SENDER_NDI_NAME: Lazy<String> = Lazy::new(|| {
+    format!(
+        "GStreamer NDI Sink {}-{}",
+        env!("CARGO_PKG_VERSION"),
+        env!("COMMIT_ID")
+    )
+});
+
+#[derive(Debug)]
+struct Settings {
+    ndi_name: String,
+}
+
+impl Default for Settings {
+    fn default() -> Self {
+        Settings {
+            ndi_name: DEFAULT_SENDER_NDI_NAME.clone(),
+        }
+    }
+}
+
+static PROPERTIES: [subclass::Property; 1] = [subclass::Property("ndi-name", |name| {
+    glib::ParamSpec::string(
+        name,
+        "NDI Name",
+        "NDI Name to use",
+        Some(DEFAULT_SENDER_NDI_NAME.as_ref()),
+        glib::ParamFlags::READWRITE,
+    )
+})];
+
+struct State {
+    send: SendInstance,
+    info: Option<gst_video::VideoInfo>,
+}
+
+pub struct NdiSink {
+    settings: Mutex<Settings>,
+    state: Mutex<Option<State>>,
+}
+
+static CAT: Lazy<gst::DebugCategory> = Lazy::new(|| {
+    gst::DebugCategory::new("ndisink", gst::DebugColorFlags::empty(), Some("NDI Sink"))
+});
+
+impl ObjectSubclass for NdiSink {
+    const NAME: &'static str = "NdiSink";
+    type ParentType = gst_base::BaseSink;
+    type Instance = gst::subclass::ElementInstanceStruct<Self>;
+    type Class = subclass::simple::ClassStruct<Self>;
+
+    glib::glib_object_subclass!();
+
+    fn new() -> Self {
+        Self {
+            settings: Mutex::new(Default::default()),
+            state: Mutex::new(Default::default()),
+        }
+    }
+
+    fn class_init(klass: &mut subclass::simple::ClassStruct<Self>) {
+        klass.set_metadata(
+            "NDI Sink",
+            "Sink/Audio/Video",
+            "Render as an NDI stream",
+            "Sebastian Dröge <[email protected]>",
+        );
+
+        let caps = gst::Caps::builder("video/x-raw")
+            .field(
+                "format",
+                &gst::List::new(&[
+                    &gst_video::VideoFormat::Uyvy.to_str(),
+                    &gst_video::VideoFormat::I420.to_str(),
+                    &gst_video::VideoFormat::Nv12.to_str(),
+                    &gst_video::VideoFormat::Nv21.to_str(),
+                    &gst_video::VideoFormat::Yv12.to_str(),
+                    &gst_video::VideoFormat::Bgra.to_str(),
+                    &gst_video::VideoFormat::Bgrx.to_str(),
+                    &gst_video::VideoFormat::Rgba.to_str(),
+                    &gst_video::VideoFormat::Rgbx.to_str(),
+                ]),
+            )
+            .field("width", &gst::IntRange::<i32>::new(1, std::i32::MAX))
+            .field("height", &gst::IntRange::<i32>::new(1, std::i32::MAX))
+            .field(
+                "framerate",
+                &gst::FractionRange::new(
+                    gst::Fraction::new(0, 1),
+                    gst::Fraction::new(std::i32::MAX, 1),
+                ),
+            )
+            .build();
+        let sink_pad_template = gst::PadTemplate::new(
+            "sink",
+            gst::PadDirection::Sink,
+            gst::PadPresence::Always,
+            &caps,
+        )
+        .unwrap();
+        klass.add_pad_template(sink_pad_template);
+
+        klass.install_properties(&PROPERTIES);
+    }
+}
+
+impl ObjectImpl for NdiSink {
+    glib::glib_object_impl!();
+
+    fn set_property(&self, _obj: &glib::Object, id: usize, value: &glib::Value) {
+        let prop = &PROPERTIES[id];
+        match *prop {
+            subclass::Property("ndi-name", ..) => {
+                let mut settings = self.settings.lock().unwrap();
+                settings.ndi_name = value
+                    .get::<String>()
+                    .unwrap()
+                    .unwrap_or_else(|| DEFAULT_SENDER_NDI_NAME.clone());
+            }
+            _ => unimplemented!(),
+        };
+    }
+
+    fn get_property(&self, _obj: &glib::Object, id: usize) -> Result<glib::Value, ()> {
+        let prop = &PROPERTIES[id];
+        match *prop {
+            subclass::Property("ndi-name", ..) => {
+                let settings = self.settings.lock().unwrap();
+                Ok(settings.ndi_name.to_value())
+            }
+            _ => unimplemented!(),
+        }
+    }
+}
+
+impl ElementImpl for NdiSink {}
+
+impl BaseSinkImpl for NdiSink {
+    fn start(&self, element: &gst_base::BaseSink) -> Result<(), gst::ErrorMessage> {
+        let mut state_storage = self.state.lock().unwrap();
+        let settings = self.settings.lock().unwrap();
+
+        let send = SendInstance::builder(&settings.ndi_name)
+            .build()
+            .ok_or_else(|| {
+                gst_error_msg!(
+                    gst::ResourceError::OpenWrite,
+                    ["Could not create send instance"]
+                )
+            })?;
+
+        let state = State { send, info: None };
+        *state_storage = Some(state);
+        gst_info!(CAT, obj: element, "Started");
+
+        Ok(())
+    }
+
+    fn stop(&self, element: &gst_base::BaseSink) -> Result<(), gst::ErrorMessage> {
+        let mut state_storage = self.state.lock().unwrap();
+
+        *state_storage = None;
+        gst_info!(CAT, obj: element, "Stopped");
+
+        Ok(())
+    }
+
+    fn unlock(&self, _element: &gst_base::BaseSink) -> Result<(), gst::ErrorMessage> {
+        Ok(())
+    }
+
+    fn unlock_stop(&self, _element: &gst_base::BaseSink) -> Result<(), gst::ErrorMessage> {
+        Ok(())
+    }
+
+    fn set_caps(
+        &self,
+        element: &gst_base::BaseSink,
+        caps: &gst::Caps,
+    ) -> Result<(), gst::LoggableError> {
+        gst_debug!(CAT, obj: element, "Setting caps {}", caps);
+
+        let info = gst_video::VideoInfo::from_caps(caps)
+            .map_err(|_| gst_loggable_error!(CAT, "Couldn't parse caps {}", caps))?;
+
+        let mut state_storage = self.state.lock().unwrap();
+        let state = match &mut *state_storage {
+            None => return Err(gst_loggable_error!(CAT, "Sink not started yet")),
+            Some(ref mut state) => state,
+        };
+        state.info = Some(info);
+
+        Ok(())
+    }
+
+    fn render(
+        &self,
+        element: &gst_base::BaseSink,
+        buffer: &gst::Buffer,
+    ) -> Result<gst::FlowSuccess, gst::FlowError> {
+        let mut state_storage = self.state.lock().unwrap();
+        let state = match &mut *state_storage {
+            None => return Err(gst::FlowError::Error),
+            Some(ref mut state) => state,
+        };
+        let info = match state.info {
+            None => return Err(gst::FlowError::Error),
+            Some(ref mut info) => info,
+        };
+
+        if let Some(audio_meta) = buffer.get_meta::<crate::ndisinkmeta::NdiSinkAudioMeta>() {
+            let frame = crate::ndi::AudioFrame::from_interleaved_16s(
+                audio_meta.info(),
+                audio_meta.buffer(),
+            )
+            .map_err(|_| {
+                gst_error!(CAT, obj: element, "Unsupported audio frame");
+                gst::FlowError::NotNegotiated
+            })?;
+
+            gst_trace!(
+                CAT,
+                obj: element,
+                "Sending audio buffer {:?} with format {:?}",
+                audio_meta.buffer(),
+                audio_meta.info()
+            );
+            state.send.send_audio(&frame);
+        }
+
+        let frame =
+            gst_video::VideoFrameRef::from_buffer_ref_readable(buffer, info).map_err(|_| {
+                gst_error!(CAT, obj: element, "Failed to map buffer");
+                gst::FlowError::Error
+            })?;
+
+        let frame = crate::ndi::VideoFrame::try_from(&frame).map_err(|_| {
+            gst_error!(CAT, obj: element, "Unsupported video frame");
+            gst::FlowError::NotNegotiated
+        })?;
+
+        gst_trace!(
+            CAT,
+            obj: element,
+            "Sending video buffer {:?} with format {:?}",
+            buffer,
+            info
+        );
+        state.send.send_video(&frame);
+
+        Ok(gst::FlowSuccess::Ok)
+    }
+}
+
+pub fn register(plugin: &gst::Plugin) -> Result<(), glib::BoolError> {
+    gst::Element::register(
+        Some(plugin),
+        "ndisink",
+        gst::Rank::None,
+        NdiSink::get_type(),
+    )
+}