From d3ba5efff87810c9839ff9df01e3d3ae1c6fc1b8 Mon Sep 17 00:00:00 2001 From: Daniel Vilar Date: Tue, 11 Dec 2018 17:47:03 +0100 Subject: [PATCH] Port to new glib subclassing API #13 --- Cargo.toml | 12 +- src/lib.rs | 37 ++- src/ndiaudiosrc.rs | 691 +++++++++++++++++++++++---------------------- src/ndivideosrc.rs | 664 ++++++++++++++++++++++--------------------- 4 files changed, 705 insertions(+), 699 deletions(-) diff --git a/Cargo.toml b/Cargo.toml index 89f43cc0..f7342312 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -6,13 +6,11 @@ repository = "https://github.com/teltek/gst-plugin-ndi" license = "LGPL" [dependencies] -gobject-subclass = "0.2" -gst-plugin = "0.3" -glib = "0.6" -gstreamer = "0.12" -gstreamer-base = "0.12" -gstreamer-video = "0.12" -gstreamer-audio = "0.12" +glib = { git = "https://github.com/gtk-rs/glib", features = ["subclassing"] } +gstreamer = { git = "https://gitlab.freedesktop.org/gstreamer/gstreamer-rs", features = ["subclassing"] } +gstreamer-base = { git = "https://gitlab.freedesktop.org/gstreamer/gstreamer-rs", features = ["subclassing"] } +gstreamer-video = { git = "https://gitlab.freedesktop.org/gstreamer/gstreamer-rs" } +gstreamer-audio = { git = "https://gitlab.freedesktop.org/gstreamer/gstreamer-rs" } lazy_static = "1.1.0" [lib] diff --git a/src/lib.rs b/src/lib.rs index fab45fa1..d3b2dd27 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -1,10 +1,7 @@ #![allow(non_camel_case_types, non_upper_case_globals, non_snake_case)] -extern crate glib; -extern crate gobject_subclass; - #[macro_use] -extern crate gst_plugin; +extern crate glib; #[macro_use] extern crate gstreamer as gst; use gst::prelude::*; @@ -19,7 +16,7 @@ mod ndiaudiosrc; pub mod ndisys; mod ndivideosrc; -use gst_plugin::base_src::*; +// use gst_plugin::base_src::*; use ndisys::*; use std::ffi::{CStr, CString}; use std::{thread, time}; @@ -29,10 +26,10 @@ use std::sync::Mutex; use gst::GstObjectExt; -fn plugin_init(plugin: &gst::Plugin) -> bool { - ndivideosrc::register(plugin); - ndiaudiosrc::register(plugin); - true +fn plugin_init(plugin: &gst::Plugin) -> Result<(), glib::BoolError> { + ndivideosrc::register(plugin)?; + ndiaudiosrc::register(plugin)?; + Ok(()) } struct ndi_receiver_info { @@ -62,7 +59,7 @@ lazy_static! { static mut id_receiver: i8 = 0; -fn connect_ndi(cat: gst::DebugCategory, element: &BaseSrc, ip: &str, stream_name: &str) -> i8 { +fn connect_ndi(cat: gst::DebugCategory, element: &gst_base::BaseSrc, ip: &str, stream_name: &str) -> i8 { gst_debug!(cat, obj: element, "Starting NDI connection..."); let mut receivers = hashmap_receivers.lock().unwrap(); @@ -223,7 +220,7 @@ fn connect_ndi(cat: gst::DebugCategory, element: &BaseSrc, ip: &str, stream_name } } -fn stop_ndi(cat: gst::DebugCategory, element: &BaseSrc, id: i8) -> bool { +fn stop_ndi(cat: gst::DebugCategory, element: &gst_base::BaseSrc, id: i8) -> bool { gst_debug!(cat, obj: element, "Closing NDI connection..."); let mut receivers = hashmap_receivers.lock().unwrap(); { @@ -250,14 +247,14 @@ fn stop_ndi(cat: gst::DebugCategory, element: &BaseSrc, id: i8) -> bool { true } -plugin_define!( - b"ndi\0", - b"NewTek NDI Plugin\0", +gst_plugin_define!( + "ndi", + "NewTek NDI Plugin", plugin_init, - b"1.0.0\0", - b"LGPL\0", - b"ndi\0", - b"ndi\0", - b"https://github.com/teltek/gst-plugin-ndi\0", - b"2018-04-09\0" + "1.0.0", + "LGPL", + "ndi", + "ndi", + "https://github.com/teltek/gst-plugin-ndi", + "2018-04-09" ); diff --git a/src/ndiaudiosrc.rs b/src/ndiaudiosrc.rs index fe57dd50..ea172d9e 100644 --- a/src/ndiaudiosrc.rs +++ b/src/ndiaudiosrc.rs @@ -1,14 +1,15 @@ -#![allow(non_camel_case_types, non_upper_case_globals, non_snake_case)] +#![allow(non_camel_case_types, non_upper_case_globals, non_snake_case, cast_ptr_alignment)] use glib; +use glib::subclass; +use glib::subclass::prelude::*; use gst; use gst::prelude::*; +use gst::subclass::prelude::*; use gst_audio; +use gst_base; use gst_base::prelude::*; - -use gobject_subclass::object::*; -use gst_plugin::base_src::*; -use gst_plugin::element::*; +use gst_base::subclass::prelude::*; use std::sync::Mutex; use std::{i32, u32}; @@ -43,29 +44,36 @@ impl Default for Settings { } } -static PROPERTIES: [Property; 3] = [ - Property::String( +static PROPERTIES: [subclass::Property; 3] = [ +subclass::Property("stream-name", || { + glib::ParamSpec::string( "stream-name", "Sream Name", "Name of the streaming device", None, - PropertyMutability::ReadWrite, - ), - Property::String( + glib::ParamFlags::READWRITE, + ) +}), +subclass::Property("ip", || { + glib::ParamSpec::string( "ip", "Stream IP", "IP of the streaming device. Ex: 127.0.0.1:5961", None, - PropertyMutability::ReadWrite, - ), - Property::UInt( + glib::ParamFlags::READWRITE, + ) +}), +subclass::Property("loss-threshold", || { + glib::ParamSpec::uint( "loss-threshold", "Loss threshold", - "Loss threshold. If 0 the stream is never closed by the element", - (0, 60), + "Loss threshold", + 0, + 60, 5, - PropertyMutability::ReadWrite, - ), + glib::ParamFlags::READWRITE, + ) +}), ]; struct State { @@ -89,12 +97,17 @@ struct NdiAudioSrc { timestamp_data: Mutex, } -impl NdiAudioSrc { - fn new(element: &BaseSrc) -> Box> { - element.set_live(true); - element.set_format(gst::Format::Time); +impl ObjectSubclass for NdiAudioSrc { - Box::new(Self { + const NAME: &'static str = "NdiAudioSrc"; + type ParentType = gst_base::BaseSrc; + type Instance = gst::subclass::ElementInstanceStruct; + type Class = subclass::simple::ClassStruct; + + glib_object_subclass!(); + + fn new() -> Self { + Self { cat: gst::DebugCategory::new( "ndiaudiosrc", gst::DebugColorFlags::empty(), @@ -103,10 +116,10 @@ impl NdiAudioSrc { settings: Mutex::new(Default::default()), state: Mutex::new(Default::default()), timestamp_data: Mutex::new(TimestampData { offset: 0}), - }) + } } - fn class_init(klass: &mut BaseSrcClass) { + fn class_init(klass: &mut subclass::simple::ClassStruct) { klass.set_metadata( "NewTek NDI Audio Source", "Source", @@ -117,370 +130,358 @@ impl NdiAudioSrc { let caps = gst::Caps::new_simple( "audio/x-raw", &[ - ( - "format", - &gst::List::new(&[ - //TODO add more formats? - //&gst_audio::AUDIO_FORMAT_F32.to_string(), - //&gst_audio::AUDIO_FORMAT_F64.to_string(), - &gst_audio::AUDIO_FORMAT_S16.to_string(), + ( + "format", + &gst::List::new(&[ + //TODO add more formats? + //&gst_audio::AUDIO_FORMAT_F32.to_string(), + //&gst_audio::AUDIO_FORMAT_F64.to_string(), + &gst_audio::AUDIO_FORMAT_S16.to_string(), ]), ), ("rate", &gst::IntRange::::new(1, i32::MAX)), ("channels", &gst::IntRange::::new(1, i32::MAX)), ("layout", &"interleaved"), ("channel-mask", &gst::Bitmask::new(0)), - ], - ); + ], + ); - let src_pad_template = gst::PadTemplate::new( - "src", - gst::PadDirection::Src, - gst::PadPresence::Always, - &caps, - ); - klass.add_pad_template(src_pad_template); + let src_pad_template = gst::PadTemplate::new( + "src", + gst::PadDirection::Src, + gst::PadPresence::Always, + &caps, + ); + klass.add_pad_template(src_pad_template); - klass.install_properties(&PROPERTIES); - } -} - -impl ObjectImpl for NdiAudioSrc { - fn set_property(&self, obj: &glib::Object, id: u32, value: &glib::Value) { - let prop = &PROPERTIES[id as usize]; - let element = obj.clone().downcast::().unwrap(); - - match *prop { - Property::String("stream-name", ..) => { - let mut settings = self.settings.lock().unwrap(); - let stream_name = value.get().unwrap(); - gst_debug!( - self.cat, - obj: &element, - "Changing stream-name from {} to {}", - settings.stream_name, - stream_name - ); - settings.stream_name = stream_name; - drop(settings); - - let _ = - element.post_message(&gst::Message::new_latency().src(Some(&element)).build()); - } - Property::String("ip", ..) => { - let mut settings = self.settings.lock().unwrap(); - let ip = value.get().unwrap(); - gst_debug!( - self.cat, - obj: &element, - "Changing ip from {} to {}", - settings.ip, - ip - ); - settings.ip = ip; - drop(settings); - - let _ = - element.post_message(&gst::Message::new_latency().src(Some(&element)).build()); - } - Property::UInt("loss-threshold", ..) => { - let mut settings = self.settings.lock().unwrap(); - let loss_threshold = value.get().unwrap(); - gst_debug!( - self.cat, - obj: &element, - "Changing loss threshold from {} to {}", - settings.loss_threshold, - loss_threshold - ); - settings.loss_threshold = loss_threshold; - drop(settings); - } - _ => unimplemented!(), + klass.install_properties(&PROPERTIES); } } - fn get_property(&self, _obj: &glib::Object, id: u32) -> Result { - let prop = &PROPERTIES[id as usize]; + impl ObjectImpl for NdiAudioSrc { + glib_object_impl!(); - match *prop { - Property::String("stream-name", ..) => { - let settings = self.settings.lock().unwrap(); - Ok(settings.stream_name.to_value()) + fn constructed(&self, obj: &glib::Object) { + self.parent_constructed(obj); + + let basesrc = obj.downcast_ref::().unwrap(); + // Initialize live-ness and notify the base class that + // we'd like to operate in Time format + basesrc.set_live(true); + basesrc.set_format(gst::Format::Time); + } + + fn set_property(&self, obj: &glib::Object, id: usize, value: &glib::Value) { + let prop = &PROPERTIES[id]; + let basesrc = obj.downcast_ref::().unwrap(); + + match *prop { + subclass::Property("stream-name", ..) => { + let mut settings = self.settings.lock().unwrap(); + let stream_name = value.get().unwrap(); + gst_debug!( + self.cat, + obj: basesrc, + "Changing stream-name from {} to {}", + settings.stream_name, + stream_name + ); + settings.stream_name = stream_name; + drop(settings); + } + subclass::Property("ip", ..) => { + let mut settings = self.settings.lock().unwrap(); + let ip = value.get().unwrap(); + gst_debug!( + self.cat, + obj: basesrc, + "Changing ip from {} to {}", + settings.ip, + ip + ); + settings.ip = ip; + drop(settings); + } + subclass::Property("loss-threshold", ..) => { + let mut settings = self.settings.lock().unwrap(); + let loss_threshold = value.get().unwrap(); + gst_debug!( + self.cat, + obj: basesrc, + "Changing loss threshold from {} to {}", + settings.loss_threshold, + loss_threshold + ); + settings.loss_threshold = loss_threshold; + drop(settings); + } + _ => unimplemented!(), } - Property::String("ip", ..) => { - let settings = self.settings.lock().unwrap(); - Ok(settings.ip.to_value()) + } + + fn get_property(&self, _obj: &glib::Object, id: usize) -> Result { + let prop = &PROPERTIES[id]; + + match *prop { + subclass::Property("stream-name", ..) => { + let settings = self.settings.lock().unwrap(); + Ok(settings.stream_name.to_value()) + } + subclass::Property("ip", ..) => { + let settings = self.settings.lock().unwrap(); + Ok(settings.ip.to_value()) + } + subclass::Property("loss-threshold", ..) => { + let settings = self.settings.lock().unwrap(); + Ok(settings.loss_threshold.to_value()) + } + _ => unimplemented!(), } - Property::UInt("loss-threshold", ..) => { - let settings = self.settings.lock().unwrap(); - Ok(settings.loss_threshold.to_value()) - } - _ => unimplemented!(), } } -} -impl ElementImpl for NdiAudioSrc { - fn change_state( - &self, - element: &BaseSrc, - transition: gst::StateChange, - ) -> gst::StateChangeReturn { - if transition == gst::StateChange::PausedToPlaying { - let mut receivers = hashmap_receivers.lock().unwrap(); + impl ElementImpl for NdiAudioSrc { + fn change_state( + &self, + element: &gst::Element, + transition: gst::StateChange, + ) -> gst::StateChangeReturn { + if transition == gst::StateChange::PausedToPlaying { + let mut receivers = hashmap_receivers.lock().unwrap(); + let settings = self.settings.lock().unwrap(); + + let receiver = receivers.get_mut(&settings.id_receiver).unwrap(); + let recv = &receiver.ndi_instance; + let pNDI_recv = recv.recv; + + let audio_frame: NDIlib_audio_frame_v2_t = Default::default(); + + let mut frame_type: NDIlib_frame_type_e = NDIlib_frame_type_e::NDIlib_frame_type_none; + unsafe { + while frame_type != NDIlib_frame_type_e::NDIlib_frame_type_audio { + frame_type = NDIlib_recv_capture_v2( + pNDI_recv, + ptr::null(), + &audio_frame, + ptr::null(), + 1000, + ); + gst_debug!(self.cat, obj: element, "NDI audio frame received: {:?}", audio_frame); + } + + if receiver.initial_timestamp <= audio_frame.timestamp as u64 + || receiver.initial_timestamp == 0 + { + receiver.initial_timestamp = audio_frame.timestamp as u64; + } + gst_debug!(self.cat, obj: element, "Setting initial timestamp to {}", receiver.initial_timestamp); + } + } + self.parent_change_state(element, transition) + } + } + + impl BaseSrcImpl for NdiAudioSrc { + fn set_caps(&self, element: &gst_base::BaseSrc, caps: &gst::CapsRef) -> bool { + let info = match gst_audio::AudioInfo::from_caps(caps) { + None => return false, + Some(info) => info, + }; + + gst_debug!(self.cat, obj: element, "Configuring for caps {}", caps); + + let mut state = self.state.lock().unwrap(); + state.info = Some(info); + + true + } + + fn start(&self, element: &gst_base::BaseSrc) -> bool { + *self.state.lock().unwrap() = Default::default(); + + let mut settings = self.settings.lock().unwrap(); + settings.id_receiver = connect_ndi( + self.cat, + element, + &settings.ip.clone(), + &settings.stream_name.clone(), + ); + + settings.id_receiver != 0 + } + + fn stop(&self, element: &gst_base::BaseSrc) -> bool { + *self.state.lock().unwrap() = Default::default(); + let settings = self.settings.lock().unwrap(); + stop_ndi(self.cat, element, settings.id_receiver); + // Commented because when adding ndi destroy stopped in this line + //*self.state.lock().unwrap() = Default::default(); + true + } + + fn query(&self, element: &gst_base::BaseSrc, query: &mut gst::QueryRef) -> bool { + use gst::QueryView; + if let QueryView::Scheduling(ref mut q) = query.view_mut() { + q.set(gst::SchedulingFlags::SEQUENTIAL, 1, -1, 0); + q.add_scheduling_modes(&[gst::PadMode::Push]); + return true; + } + if let QueryView::Latency(ref mut q) = query.view_mut() { + let settings = &*self.settings.lock().unwrap(); + let state = self.state.lock().unwrap(); + + if let Some(ref _info) = state.info { + let latency = settings.latency.unwrap(); + gst_debug!(self.cat, obj: element, "Returning latency {}", latency); + q.set(true, latency, gst::CLOCK_TIME_NONE); + return true; + } else { + return false; + } + } + BaseSrcImpl::parent_query(self, element, query) + } + + fn fixate(&self, element: &gst_base::BaseSrc, caps: gst::Caps) -> gst::Caps { + let receivers = hashmap_receivers.lock().unwrap(); + let mut settings = self.settings.lock().unwrap(); + + let receiver = receivers.get(&settings.id_receiver).unwrap(); - let receiver = receivers.get_mut(&settings.id_receiver).unwrap(); let recv = &receiver.ndi_instance; let pNDI_recv = recv.recv; let audio_frame: NDIlib_audio_frame_v2_t = Default::default(); let mut frame_type: NDIlib_frame_type_e = NDIlib_frame_type_e::NDIlib_frame_type_none; - unsafe { - while frame_type != NDIlib_frame_type_e::NDIlib_frame_type_audio { - frame_type = NDIlib_recv_capture_v2( - pNDI_recv, - ptr::null(), - &audio_frame, - ptr::null(), - 1000, - ); - gst_debug!(self.cat, obj: element, "NDI audio frame received: {:?}", audio_frame); - } - - if receiver.initial_timestamp <= audio_frame.timestamp as u64 - || receiver.initial_timestamp == 0 - { - receiver.initial_timestamp = audio_frame.timestamp as u64; - } - gst_debug!(self.cat, obj: element, "Setting initial timestamp to {}", receiver.initial_timestamp); - } - } - element.parent_change_state(transition) - } -} - -impl BaseSrcImpl for NdiAudioSrc { - fn set_caps(&self, element: &BaseSrc, caps: &gst::CapsRef) -> bool { - let info = match gst_audio::AudioInfo::from_caps(caps) { - None => return false, - Some(info) => info, - }; - - gst_debug!(self.cat, obj: element, "Configuring for caps {}", caps); - - let mut state = self.state.lock().unwrap(); - state.info = Some(info); - - true - } - - fn start(&self, element: &BaseSrc) -> bool { - *self.state.lock().unwrap() = Default::default(); - - let mut settings = self.settings.lock().unwrap(); - settings.id_receiver = connect_ndi( - self.cat, - element, - &settings.ip.clone(), - &settings.stream_name.clone(), - ); - - settings.id_receiver != 0 - } - - fn stop(&self, element: &BaseSrc) -> bool { - *self.state.lock().unwrap() = Default::default(); - - let settings = self.settings.lock().unwrap(); - stop_ndi(self.cat, element, settings.id_receiver); - // Commented because when adding ndi destroy stopped in this line - //*self.state.lock().unwrap() = Default::default(); - true - } - - fn query(&self, element: &BaseSrc, query: &mut gst::QueryRef) -> bool { - use gst::QueryView; - if let QueryView::Scheduling(ref mut q) = query.view_mut() { - q.set(gst::SchedulingFlags::SEQUENTIAL, 1, -1, 0); - q.add_scheduling_modes(&[gst::PadMode::Push]); - return true; - } - if let QueryView::Latency(ref mut q) = query.view_mut() { - let settings = &*self.settings.lock().unwrap(); - let state = self.state.lock().unwrap(); - - if let Some(ref _info) = state.info { - let latency = settings.latency.unwrap(); - gst_debug!(self.cat, obj: element, "Returning latency {}", latency); - q.set(true, latency, gst::CLOCK_TIME_NONE); - return true; - } else { - return false; - } - } - BaseSrcBase::parent_query(element, query) - } - - fn fixate(&self, element: &BaseSrc, caps: gst::Caps) -> gst::Caps { - let receivers = hashmap_receivers.lock().unwrap(); - let mut settings = self.settings.lock().unwrap(); - - let receiver = receivers.get(&settings.id_receiver).unwrap(); - - let recv = &receiver.ndi_instance; - let pNDI_recv = recv.recv; - - let audio_frame: NDIlib_audio_frame_v2_t = Default::default(); - - let mut frame_type: NDIlib_frame_type_e = NDIlib_frame_type_e::NDIlib_frame_type_none; - while frame_type != NDIlib_frame_type_e::NDIlib_frame_type_audio { - unsafe { - frame_type = + while frame_type != NDIlib_frame_type_e::NDIlib_frame_type_audio { + unsafe { + frame_type = NDIlib_recv_capture_v2(pNDI_recv, ptr::null(), &audio_frame, ptr::null(), 1000); gst_debug!(self.cat, obj: element, "NDI audio frame received: {:?}", audio_frame); - } - } - - let no_samples = audio_frame.no_samples as u64; - let audio_rate = audio_frame.sample_rate; - settings.latency = gst::SECOND.mul_div_floor(no_samples, audio_rate as u64); - - let mut caps = gst::Caps::truncate(caps); - { - let caps = caps.make_mut(); - let s = caps.get_mut_structure(0).unwrap(); - s.fixate_field_nearest_int("rate", audio_rate); - s.fixate_field_nearest_int("channels", audio_frame.no_channels); - s.fixate_field_str("layout", "interleaved"); - s.set_value("channel-mask", gst::Bitmask::new(gst_audio::AudioChannelPosition::get_fallback_mask(audio_frame.no_channels as u32)).to_send_value()); - } - - let _ = element.post_message(&gst::Message::new_latency().src(Some(element)).build()); - element.parent_fixate(caps) - } - - fn create( - &self, - element: &BaseSrc, - _offset: u64, - _length: u32, - ) -> Result { - let _settings = &*self.settings.lock().unwrap(); - - let mut timestamp_data = self.timestamp_data.lock().unwrap(); - - let state = self.state.lock().unwrap(); - let _info = match state.info { - None => { - gst_element_error!(element, gst::CoreError::Negotiation, ["Have no caps yet"]); - return Err(gst::FlowReturn::NotNegotiated); - } - Some(ref info) => info.clone(), - }; - let receivers = hashmap_receivers.lock().unwrap(); - - let recv = &receivers.get(&_settings.id_receiver).unwrap().ndi_instance; - let pNDI_recv = recv.recv; - - let pts: u64; - let audio_frame: NDIlib_audio_frame_v2_t = Default::default(); - - unsafe { - let time = receivers.get(&_settings.id_receiver).unwrap().initial_timestamp; - - let mut skip_frame = true; - let mut count_frame_none = 0; - while skip_frame { - let frame_type = - NDIlib_recv_capture_v2(pNDI_recv, ptr::null(), &audio_frame, ptr::null(), 1000); - if (frame_type == NDIlib_frame_type_e::NDIlib_frame_type_none && _settings.loss_threshold != 0) - || frame_type == NDIlib_frame_type_e::NDIlib_frame_type_error - { - if count_frame_none < _settings.loss_threshold{ - count_frame_none += 1; - continue; - } - gst_element_error!(element, gst::ResourceError::Read, ["NDI frame type none or error received, assuming that the source closed the stream...."]); - return Err(gst::FlowReturn::CustomError); - } - else{ - if frame_type == NDIlib_frame_type_e::NDIlib_frame_type_none && _settings.loss_threshold == 0{ - gst_debug!(self.cat, obj: element, "No audio frame received, sending empty buffer"); - let buffer = gst::Buffer::with_size(0).unwrap(); - return Ok(buffer) - } - } - if time >= (audio_frame.timestamp as u64) { - gst_debug!(self.cat, obj: element, "Frame timestamp ({:?}) is lower than received in the first frame from NDI ({:?}), so skiping...", (audio_frame.timestamp as u64), time); - } else { - skip_frame = false; } } - gst_log!(self.cat, obj: element, "NDI audio frame received: {:?}", (audio_frame)); + let no_samples = audio_frame.no_samples as u64; + let audio_rate = audio_frame.sample_rate; + settings.latency = gst::SECOND.mul_div_floor(no_samples, audio_rate as u64); - pts = audio_frame.timestamp as u64 - time; - - gst_log!(self.cat, obj: element, "Calculated pts for audio frame: {:?}", (pts)); - - // We multiply by 2 because is the size in bytes of an i16 variable - let buff_size = (audio_frame.no_samples * 2 * audio_frame.no_channels) as usize; - let mut buffer = gst::Buffer::with_size(buff_size).unwrap(); + let mut caps = gst::Caps::truncate(caps); { - if ndi_struct.start_pts == gst::ClockTime(Some(0)) { - ndi_struct.start_pts = - element.get_clock().unwrap().get_time() - element.get_base_time(); + let caps = caps.make_mut(); + let s = caps.get_mut_structure(0).unwrap(); + s.fixate_field_nearest_int("rate", audio_rate); + s.fixate_field_nearest_int("channels", audio_frame.no_channels); + s.fixate_field_str("layout", "interleaved"); + s.set_value("channel-mask", gst::Bitmask::new(gst_audio::AudioChannelPosition::get_fallback_mask(audio_frame.no_channels as u32)).to_send_value()); + } + + let _ = element.post_message(&gst::Message::new_latency().src(Some(element)).build()); + self.parent_fixate(element, caps) + } + + fn create( + &self, + element: &gst_base::BaseSrc, + _offset: u64, + _length: u32, + ) -> Result { + let _settings = &*self.settings.lock().unwrap(); + + let mut timestamp_data = self.timestamp_data.lock().unwrap(); + + let state = self.state.lock().unwrap(); + let _info = match state.info { + None => { + gst_element_error!(element, gst::CoreError::Negotiation, ["Have no caps yet"]); + return Err(gst::FlowError::NotNegotiated); + } + Some(ref info) => info.clone(), + }; + let receivers = hashmap_receivers.lock().unwrap(); + + let recv = &receivers.get(&_settings.id_receiver).unwrap().ndi_instance; + let pNDI_recv = recv.recv; + + let pts: u64; + let audio_frame: NDIlib_audio_frame_v2_t = Default::default(); + + unsafe { + let time = receivers.get(&_settings.id_receiver).unwrap().initial_timestamp; + + let mut skip_frame = true; + let mut count_frame_none = 0; + while skip_frame { + let frame_type = + NDIlib_recv_capture_v2(pNDI_recv, ptr::null(), &audio_frame, ptr::null(), 1000); + if (frame_type == NDIlib_frame_type_e::NDIlib_frame_type_none && _settings.loss_threshold != 0) + || frame_type == NDIlib_frame_type_e::NDIlib_frame_type_error + { + if count_frame_none < _settings.loss_threshold{ + count_frame_none += 1; + continue; + } + gst_element_error!(element, gst::ResourceError::Read, ["NDI frame type none or error received, assuming that the source closed the stream...."]); + return Err(gst::FlowError::CustomError); + } + else if frame_type == NDIlib_frame_type_e::NDIlib_frame_type_none && _settings.loss_threshold == 0{ + gst_debug!(self.cat, obj: element, "No audio frame received, sending empty buffer"); + let buffer = gst::Buffer::with_size(0).unwrap(); + return Ok(buffer) + } + + if time >= (audio_frame.timestamp as u64) { + gst_debug!(self.cat, obj: element, "Frame timestamp ({:?}) is lower than received in the first frame from NDI ({:?}), so skiping...", (audio_frame.timestamp as u64), time); + } else { + skip_frame = false; + } } - let buffer = buffer.get_mut().unwrap(); + gst_log!(self.cat, obj: element, "NDI audio frame received: {:?}", (audio_frame)); - // Newtek NDI yields times in 100ns intervals since the Unix Time - let pts: gst::ClockTime = (pts * 100).into(); - buffer.set_pts(pts + ndi_struct.start_pts); + pts = audio_frame.timestamp as u64 - time; - let duration: gst::ClockTime = (((f64::from(audio_frame.no_samples) + gst_log!(self.cat, obj: element, "Calculated pts for audio frame: {:?}", (pts)); + + // We multiply by 2 because is the size in bytes of an i16 variable + let buff_size = (audio_frame.no_samples * 2 * audio_frame.no_channels) as usize; + let mut buffer = gst::Buffer::with_size(buff_size).unwrap(); + { + if ndi_struct.start_pts == gst::ClockTime(Some(0)) { + ndi_struct.start_pts = + element.get_clock().unwrap().get_time() - element.get_base_time(); + } + + let buffer = buffer.get_mut().unwrap(); + + // Newtek NDI yields times in 100ns intervals since the Unix Time + let pts: gst::ClockTime = (pts * 100).into(); + buffer.set_pts(pts + ndi_struct.start_pts); + + let duration: gst::ClockTime = (((f64::from(audio_frame.no_samples) / f64::from(audio_frame.sample_rate)) * 1_000_000_000.0) as u64) .into(); - buffer.set_duration(duration); + buffer.set_duration(duration); - buffer.set_offset(timestamp_data.offset); - timestamp_data.offset += audio_frame.no_samples as u64; - buffer.set_offset_end(timestamp_data.offset); + buffer.set_offset(timestamp_data.offset); + timestamp_data.offset += audio_frame.no_samples as u64; + buffer.set_offset_end(timestamp_data.offset); - let mut dst: NDIlib_audio_frame_interleaved_16s_t = Default::default(); - dst.reference_level = 0; - dst.p_data = buffer.map_writable().unwrap().as_mut_slice().as_mut_ptr() as *mut i16; - NDIlib_util_audio_to_interleaved_16s_v2(&audio_frame, &mut dst); + let mut dst: NDIlib_audio_frame_interleaved_16s_t = Default::default(); + dst.reference_level = 0; + dst.p_data = buffer.map_writable().unwrap().as_mut_slice().as_mut_ptr() as *mut i16; + NDIlib_util_audio_to_interleaved_16s_v2(&audio_frame, &mut dst); + } + + gst_log!(self.cat, obj: element, "Produced buffer {:?}", buffer); + + Ok(buffer) } - - gst_log!(self.cat, obj: element, "Produced buffer {:?}", buffer); - - Ok(buffer) } } -} -struct NdiAudioSrcStatic; - -impl ImplTypeStatic for NdiAudioSrcStatic { - fn get_name(&self) -> &str { - "NdiAudioSrc" + pub fn register(plugin: &gst::Plugin) -> Result<(), glib::BoolError> { + gst::Element::register(plugin, "ndiaudiosrc", 0, NdiAudioSrc::get_type()) } - - fn new(&self, element: &BaseSrc) -> Box> { - NdiAudioSrc::new(element) - } - - fn class_init(&self, klass: &mut BaseSrcClass) { - NdiAudioSrc::class_init(klass); - } -} - -pub fn register(plugin: &gst::Plugin) { - let type_ = register_type(NdiAudioSrcStatic); - gst::Element::register(plugin, "ndiaudiosrc", 0, type_); -} diff --git a/src/ndivideosrc.rs b/src/ndivideosrc.rs index 698ca60d..e641bb45 100644 --- a/src/ndivideosrc.rs +++ b/src/ndivideosrc.rs @@ -1,15 +1,17 @@ #![allow(non_camel_case_types, non_upper_case_globals, non_snake_case)] use glib; +use glib::subclass; +use glib::subclass::prelude::*; use gst; use gst::prelude::*; -use gst::Fraction; +use gst::subclass::prelude::*; +use gst_base; use gst_base::prelude::*; -use gst_video; +use gst_base::subclass::prelude::*; -use gobject_subclass::object::*; -use gst_plugin::base_src::*; -use gst_plugin::element::*; +use gst::Fraction; +use gst_video; use std::sync::Mutex; use std::{i32, u32}; @@ -44,29 +46,36 @@ impl Default for Settings { } } -static PROPERTIES: [Property; 3] = [ - Property::String( +static PROPERTIES: [subclass::Property; 3] = [ +subclass::Property("stream-name", || { + glib::ParamSpec::string( "stream-name", "Stream Name", "Name of the streaming device", None, - PropertyMutability::ReadWrite, - ), - Property::String( + glib::ParamFlags::READWRITE, + ) +}), +subclass::Property("ip", || { + glib::ParamSpec::string( "ip", "Stream IP", "IP of the streaming device. Ex: 127.0.0.1:5961", None, - PropertyMutability::ReadWrite, - ), - Property::UInt( + glib::ParamFlags::READWRITE, + ) +}), +subclass::Property("loss-threshold", || { + glib::ParamSpec::uint( "loss-threshold", "Loss threshold", - "Loss threshold. If 0 the stream is never closed by the element", - (0, 60), + "Loss threshold", + 0, + 60, 5, - PropertyMutability::ReadWrite, - ), + glib::ParamFlags::READWRITE, + ) +}), ]; struct State { @@ -90,12 +99,17 @@ struct NdiVideoSrc { timestamp_data: Mutex, } -impl NdiVideoSrc { - fn new(element: &BaseSrc) -> Box> { - element.set_live(true); - element.set_format(gst::Format::Time); +impl ObjectSubclass for NdiVideoSrc { - Box::new(Self { + const NAME: &'static str = "NdiVideoSrc"; + type ParentType = gst_base::BaseSrc; + type Instance = gst::subclass::ElementInstanceStruct; + type Class = subclass::simple::ClassStruct; + + glib_object_subclass!(); + + fn new() -> Self { + Self { cat: gst::DebugCategory::new( "ndivideosrc", gst::DebugColorFlags::empty(), @@ -104,10 +118,10 @@ impl NdiVideoSrc { settings: Mutex::new(Default::default()), state: Mutex::new(Default::default()), timestamp_data: Mutex::new(TimestampData { offset: 0 }), - }) + } } - fn class_init(klass: &mut BaseSrcClass) { + fn class_init(klass: &mut subclass::simple::ClassStruct) { klass.set_metadata( "NewTek NDI Video Source", "Source", @@ -120,13 +134,13 @@ impl NdiVideoSrc { let caps = gst::Caps::new_simple( "video/x-raw", &[ - ( - "format", - &gst::List::new(&[ - //TODO add all formats - &gst_video::VideoFormat::Uyvy.to_string(), - //&gst_video::VideoFormat::Rgb.to_string(), - //&gst_video::VideoFormat::Gray8.to_string(), + ( + "format", + &gst::List::new(&[ + //TODO add all formats + &gst_video::VideoFormat::Uyvy.to_string(), + //&gst_video::VideoFormat::Rgb.to_string(), + //&gst_video::VideoFormat::Gray8.to_string(), ]), ), ("width", &gst::IntRange::::new(0, i32::MAX)), @@ -138,346 +152,342 @@ impl NdiVideoSrc { gst::Fraction::new(i32::MAX, 1), ), ), - ], - ); + ], + ); - let src_pad_template = gst::PadTemplate::new( - "src", - gst::PadDirection::Src, - gst::PadPresence::Always, - &caps, - ); - klass.add_pad_template(src_pad_template); + let src_pad_template = gst::PadTemplate::new( + "src", + gst::PadDirection::Src, + gst::PadPresence::Always, + &caps, + ); + klass.add_pad_template(src_pad_template); - klass.install_properties(&PROPERTIES); - } -} - -impl ObjectImpl for NdiVideoSrc { - fn set_property(&self, obj: &glib::Object, id: u32, value: &glib::Value) { - let prop = &PROPERTIES[id as usize]; - let element = obj.clone().downcast::().unwrap(); - - match *prop { - Property::String("stream-name", ..) => { - let mut settings = self.settings.lock().unwrap(); - let stream_name = value.get().unwrap(); - gst_debug!( - self.cat, - obj: &element, - "Changing stream-name from {} to {}", - settings.stream_name, - stream_name - ); - settings.stream_name = stream_name; - drop(settings); - } - Property::String("ip", ..) => { - let mut settings = self.settings.lock().unwrap(); - let ip = value.get().unwrap(); - gst_debug!( - self.cat, - obj: &element, - "Changing ip from {} to {}", - settings.ip, - ip - ); - settings.ip = ip; - drop(settings); - } - Property::UInt("loss-threshold", ..) => { - let mut settings = self.settings.lock().unwrap(); - let loss_threshold = value.get().unwrap(); - gst_debug!( - self.cat, - obj: &element, - "Changing loss threshold from {} to {}", - settings.loss_threshold, - loss_threshold - ); - settings.loss_threshold = loss_threshold; - drop(settings); - } - _ => unimplemented!(), + klass.install_properties(&PROPERTIES); } } - fn get_property(&self, _obj: &glib::Object, id: u32) -> Result { - let prop = &PROPERTIES[id as usize]; - match *prop { - Property::String("stream-name", ..) => { - let settings = self.settings.lock().unwrap(); - Ok(settings.stream_name.to_value()) + impl ObjectImpl for NdiVideoSrc { + glib_object_impl!(); + + fn constructed(&self, obj: &glib::Object) { + self.parent_constructed(obj); + + let basesrc = obj.downcast_ref::().unwrap(); + // Initialize live-ness and notify the base class that + // we'd like to operate in Time format + basesrc.set_live(true); + basesrc.set_format(gst::Format::Time); + } + + + fn set_property(&self, obj: &glib::Object, id: usize, value: &glib::Value) { + let prop = &PROPERTIES[id]; + let basesrc = obj.downcast_ref::().unwrap(); + + match *prop { + subclass::Property("stream-name", ..) => { + let mut settings = self.settings.lock().unwrap(); + let stream_name = value.get().unwrap(); + gst_debug!( + self.cat, + obj: basesrc, + "Changing stream-name from {} to {}", + settings.stream_name, + stream_name + ); + settings.stream_name = stream_name; + drop(settings); + } + subclass::Property("ip", ..) => { + let mut settings = self.settings.lock().unwrap(); + let ip = value.get().unwrap(); + gst_debug!( + self.cat, + obj: basesrc, + "Changing ip from {} to {}", + settings.ip, + ip + ); + settings.ip = ip; + drop(settings); + } + subclass::Property("loss-threshold", ..) => { + let mut settings = self.settings.lock().unwrap(); + let loss_threshold = value.get().unwrap(); + gst_debug!( + self.cat, + obj: basesrc, + "Changing loss threshold from {} to {}", + settings.loss_threshold, + loss_threshold + ); + settings.loss_threshold = loss_threshold; + drop(settings); + } + _ => unimplemented!(), } - Property::String("ip", ..) => { - let settings = self.settings.lock().unwrap(); - Ok(settings.ip.to_value()) + } + + fn get_property(&self, _obj: &glib::Object, id: usize) -> Result { + let prop = &PROPERTIES[id]; + + match *prop { + subclass::Property("stream-name", ..) => { + let settings = self.settings.lock().unwrap(); + Ok(settings.stream_name.to_value()) + } + subclass::Property("ip", ..) => { + let settings = self.settings.lock().unwrap(); + Ok(settings.ip.to_value()) + } + subclass::Property("loss-threshold", ..) => { + let settings = self.settings.lock().unwrap(); + Ok(settings.loss_threshold.to_value()) + } + _ => unimplemented!(), } - Property::UInt("loss-threshold", ..) => { - let settings = self.settings.lock().unwrap(); - Ok(settings.loss_threshold.to_value()) - } - _ => unimplemented!(), } } -} -impl ElementImpl for NdiVideoSrc { - fn change_state( - &self, - element: &BaseSrc, - transition: gst::StateChange, - ) -> gst::StateChangeReturn { - if transition == gst::StateChange::PausedToPlaying { - let mut receivers = hashmap_receivers.lock().unwrap(); + impl ElementImpl for NdiVideoSrc { + fn change_state( + &self, + element: &gst::Element, + transition: gst::StateChange, + ) -> gst::StateChangeReturn { + if transition == gst::StateChange::PausedToPlaying { + let mut receivers = hashmap_receivers.lock().unwrap(); + let settings = self.settings.lock().unwrap(); + + let receiver = receivers.get_mut(&settings.id_receiver).unwrap(); + let recv = &receiver.ndi_instance; + let pNDI_recv = recv.recv; + + let video_frame: NDIlib_video_frame_v2_t = Default::default(); + + let mut frame_type: NDIlib_frame_type_e = NDIlib_frame_type_e::NDIlib_frame_type_none; + unsafe { + while frame_type != NDIlib_frame_type_e::NDIlib_frame_type_video { + frame_type = NDIlib_recv_capture_v2( + pNDI_recv, + &video_frame, + ptr::null(), + ptr::null(), + 1000, + ); + gst_debug!(self.cat, obj: element, "NDI video frame received: {:?}", video_frame); + } + + if receiver.initial_timestamp <= video_frame.timestamp as u64 + || receiver.initial_timestamp == 0 + { + receiver.initial_timestamp = video_frame.timestamp as u64; + } + gst_debug!(self.cat, obj: element, "Setting initial timestamp to {}", receiver.initial_timestamp); + } + } + self.parent_change_state(element, transition) + } + } + + impl BaseSrcImpl for NdiVideoSrc { + fn set_caps(&self, element: &gst_base::BaseSrc, caps: &gst::CapsRef) -> bool { + let info = match gst_video::VideoInfo::from_caps(caps) { + None => return false, + Some(info) => info, + }; + gst_debug!(self.cat, obj: element, "Configuring for caps {}", caps); + + let mut state = self.state.lock().unwrap(); + state.info = Some(info); + let _ = element.post_message(&gst::Message::new_latency().src(Some(element)).build()); + true + } + + fn start(&self, element: &gst_base::BaseSrc) -> bool { + *self.state.lock().unwrap() = Default::default(); + let mut settings = self.settings.lock().unwrap(); + settings.id_receiver = connect_ndi( + self.cat, + element, + &settings.ip.clone(), + &settings.stream_name.clone(), + ); + + settings.id_receiver != 0 + } + + fn stop(&self, element: &gst_base::BaseSrc) -> bool { + *self.state.lock().unwrap() = Default::default(); + let settings = self.settings.lock().unwrap(); + stop_ndi(self.cat, element, settings.id_receiver); + // Commented because when adding ndi destroy stopped in this line + //*self.state.lock().unwrap() = Default::default(); + true + } - let receiver = receivers.get_mut(&settings.id_receiver).unwrap(); + fn query(&self, element: &gst_base::BaseSrc, query: &mut gst::QueryRef) -> bool { + use gst::QueryView; + if let QueryView::Scheduling(ref mut q) = query.view_mut() { + q.set(gst::SchedulingFlags::SEQUENTIAL, 1, -1, 0); + q.add_scheduling_modes(&[gst::PadMode::Push]); + return true; + } + if let QueryView::Latency(ref mut q) = query.view_mut() { + let settings = &*self.settings.lock().unwrap(); + let state = self.state.lock().unwrap(); + + if let Some(ref _info) = state.info { + let latency = settings.latency.unwrap(); + gst_debug!(self.cat, obj: element, "Returning latency {}", latency); + q.set(true, latency, gst::CLOCK_TIME_NONE); + return true; + } else { + return false; + } + } + BaseSrcImpl::parent_query(self, element, query) + } + + fn fixate(&self, element: &gst_base::BaseSrc, caps: gst::Caps) -> gst::Caps { + let receivers = hashmap_receivers.lock().unwrap(); + let mut settings = self.settings.lock().unwrap(); + + let receiver = receivers.get(&settings.id_receiver).unwrap(); let recv = &receiver.ndi_instance; let pNDI_recv = recv.recv; let video_frame: NDIlib_video_frame_v2_t = Default::default(); let mut frame_type: NDIlib_frame_type_e = NDIlib_frame_type_e::NDIlib_frame_type_none; - unsafe { - while frame_type != NDIlib_frame_type_e::NDIlib_frame_type_video { - frame_type = NDIlib_recv_capture_v2( - pNDI_recv, - &video_frame, - ptr::null(), - ptr::null(), - 1000, - ); - gst_debug!(self.cat, obj: element, "NDI video frame received: {:?}", video_frame); - } - - if receiver.initial_timestamp <= video_frame.timestamp as u64 - || receiver.initial_timestamp == 0 - { - receiver.initial_timestamp = video_frame.timestamp as u64; - } - gst_debug!(self.cat, obj: element, "Setting initial timestamp to {}", receiver.initial_timestamp); - } - } - element.parent_change_state(transition) - } -} - -impl BaseSrcImpl for NdiVideoSrc { - fn set_caps(&self, element: &BaseSrc, caps: &gst::CapsRef) -> bool { - let info = match gst_video::VideoInfo::from_caps(caps) { - None => return false, - Some(info) => info, - }; - gst_debug!(self.cat, obj: element, "Configuring for caps {}", caps); - - let mut state = self.state.lock().unwrap(); - state.info = Some(info); - let _ = element.post_message(&gst::Message::new_latency().src(Some(element)).build()); - true - } - - fn start(&self, element: &BaseSrc) -> bool { - *self.state.lock().unwrap() = Default::default(); - let mut settings = self.settings.lock().unwrap(); - settings.id_receiver = connect_ndi( - self.cat, - element, - &settings.ip.clone(), - &settings.stream_name.clone(), - ); - - settings.id_receiver != 0 - } - - fn stop(&self, element: &BaseSrc) -> bool { - *self.state.lock().unwrap() = Default::default(); - - let settings = self.settings.lock().unwrap(); - stop_ndi(self.cat, element, settings.id_receiver); - // Commented because when adding ndi destroy stopped in this line - //*self.state.lock().unwrap() = Default::default(); - true - } - - fn query(&self, element: &BaseSrc, query: &mut gst::QueryRef) -> bool { - use gst::QueryView; - if let QueryView::Scheduling(ref mut q) = query.view_mut() { - q.set(gst::SchedulingFlags::SEQUENTIAL, 1, -1, 0); - q.add_scheduling_modes(&[gst::PadMode::Push]); - return true; - } - if let QueryView::Latency(ref mut q) = query.view_mut() { - let settings = &*self.settings.lock().unwrap(); - let state = self.state.lock().unwrap(); - - if let Some(ref _info) = state.info { - let latency = settings.latency.unwrap(); - gst_debug!(self.cat, obj: element, "Returning latency {}", latency); - q.set(true, latency, gst::CLOCK_TIME_NONE); - return true; - } else { - return false; - } - } - BaseSrcBase::parent_query(element, query) - } - - fn fixate(&self, element: &BaseSrc, caps: gst::Caps) -> gst::Caps { - let receivers = hashmap_receivers.lock().unwrap(); - let mut settings = self.settings.lock().unwrap(); - - let receiver = receivers.get(&settings.id_receiver).unwrap(); - let recv = &receiver.ndi_instance; - let pNDI_recv = recv.recv; - - let video_frame: NDIlib_video_frame_v2_t = Default::default(); - - let mut frame_type: NDIlib_frame_type_e = NDIlib_frame_type_e::NDIlib_frame_type_none; - while frame_type != NDIlib_frame_type_e::NDIlib_frame_type_video { - unsafe { - frame_type = + while frame_type != NDIlib_frame_type_e::NDIlib_frame_type_video { + unsafe { + frame_type = NDIlib_recv_capture_v2(pNDI_recv, &video_frame, ptr::null(), ptr::null(), 1000); gst_debug!(self.cat, obj: element, "NDI video frame received: {:?}", video_frame); + } } - } - settings.latency = gst::SECOND.mul_div_floor( - video_frame.frame_rate_D as u64, - video_frame.frame_rate_N as u64, - ); - - let mut caps = gst::Caps::truncate(caps); - { - let caps = caps.make_mut(); - let s = caps.get_mut_structure(0).unwrap(); - s.fixate_field_nearest_int("width", video_frame.xres); - s.fixate_field_nearest_int("height", video_frame.yres); - s.fixate_field_nearest_fraction( - "framerate", - Fraction::new(video_frame.frame_rate_N, video_frame.frame_rate_D), + settings.latency = gst::SECOND.mul_div_floor( + video_frame.frame_rate_D as u64, + video_frame.frame_rate_N as u64, ); + + let mut caps = gst::Caps::truncate(caps); + { + let caps = caps.make_mut(); + let s = caps.get_mut_structure(0).unwrap(); + s.fixate_field_nearest_int("width", video_frame.xres); + s.fixate_field_nearest_int("height", video_frame.yres); + s.fixate_field_nearest_fraction( + "framerate", + Fraction::new(video_frame.frame_rate_N, video_frame.frame_rate_D), + ); + } + + let _ = element.post_message(&gst::Message::new_latency().src(Some(element)).build()); + self.parent_fixate(element, caps) } - let _ = element.post_message(&gst::Message::new_latency().src(Some(element)).build()); - element.parent_fixate(caps) - } + //Creates the video buffers + fn create( + &self, + element: &gst_base::BaseSrc, + _offset: u64, + _length: u32, + ) -> Result { + let _settings = &*self.settings.lock().unwrap(); - //Creates the video buffers - fn create( - &self, - element: &BaseSrc, - _offset: u64, - _length: u32, - ) -> Result { - let _settings = &*self.settings.lock().unwrap(); + let mut timestamp_data = self.timestamp_data.lock().unwrap(); + let state = self.state.lock().unwrap(); + let _info = match state.info { + None => { + gst_element_error!(element, gst::CoreError::Negotiation, ["Have no caps yet"]); + return Err(gst::FlowError::NotNegotiated); + } + Some(ref info) => info.clone(), + }; + let receivers = hashmap_receivers.lock().unwrap(); - let mut timestamp_data = self.timestamp_data.lock().unwrap(); - let state = self.state.lock().unwrap(); - let _info = match state.info { - None => { - gst_element_error!(element, gst::CoreError::Negotiation, ["Have no caps yet"]); - return Err(gst::FlowReturn::NotNegotiated); - } - Some(ref info) => info.clone(), - }; - let receivers = hashmap_receivers.lock().unwrap(); + let recv = &receivers.get(&_settings.id_receiver).unwrap().ndi_instance; + let pNDI_recv = recv.recv; - let recv = &receivers.get(&_settings.id_receiver).unwrap().ndi_instance; - let pNDI_recv = recv.recv; + let pts: u64; + let video_frame: NDIlib_video_frame_v2_t = Default::default(); - let pts: u64; - let video_frame: NDIlib_video_frame_v2_t = Default::default(); + unsafe { + let time = receivers.get(&_settings.id_receiver).unwrap().initial_timestamp; - unsafe { - let time = receivers.get(&_settings.id_receiver).unwrap().initial_timestamp; - - let mut skip_frame = true; - let mut count_frame_none = 0; - while skip_frame { - let frame_type = + let mut skip_frame = true; + let mut count_frame_none = 0; + while skip_frame { + let frame_type = NDIlib_recv_capture_v2(pNDI_recv, &video_frame, ptr::null(), ptr::null(), 1000); - if (frame_type == NDIlib_frame_type_e::NDIlib_frame_type_none && _settings.loss_threshold != 0) + if (frame_type == NDIlib_frame_type_e::NDIlib_frame_type_none && _settings.loss_threshold != 0) || frame_type == NDIlib_frame_type_e::NDIlib_frame_type_error + { + if count_frame_none < _settings.loss_threshold{ + count_frame_none += 1; + continue; + } + gst_element_error!(element, gst::ResourceError::Read, ["NDI frame type none or error received, assuming that the source closed the stream...."]); + return Err(gst::FlowError::CustomError); + } + else if frame_type == NDIlib_frame_type_e::NDIlib_frame_type_none && _settings.loss_threshold == 0{ + gst_debug!(self.cat, obj: element, "No video frame received, sending empty buffer"); + let buffer = gst::Buffer::with_size(0).unwrap(); + return Ok(buffer) + } + + if time >= (video_frame.timestamp as u64) { + gst_debug!(self.cat, obj: element, "Frame timestamp ({:?}) is lower than received in the first frame from NDI ({:?}), so skiping...", (video_frame.timestamp as u64), time); + } else { + skip_frame = false; + } + } + + gst_log!(self.cat, obj: element, "NDI video frame received: {:?}", (video_frame)); + + pts = video_frame.timestamp as u64 - time; + + gst_log!(self.cat, obj: element, "Calculated pts for video frame: {:?}", (pts)); + + let buff_size = (video_frame.yres * video_frame.line_stride_in_bytes) as usize; + let mut buffer = gst::Buffer::with_size(buff_size).unwrap(); { - if count_frame_none < _settings.loss_threshold{ - count_frame_none += 1; - continue; - } - gst_element_error!(element, gst::ResourceError::Read, ["NDI frame type none or error received, assuming that the source closed the stream...."]); - return Err(gst::FlowReturn::CustomError); - } - else{ - if frame_type == NDIlib_frame_type_e::NDIlib_frame_type_none && _settings.loss_threshold == 0{ - gst_debug!(self.cat, obj: element, "No video frame received, sending empty buffer"); - let buffer = gst::Buffer::with_size(0).unwrap(); - return Ok(buffer) - } - } - if time >= (video_frame.timestamp as u64) { - gst_debug!(self.cat, obj: element, "Frame timestamp ({:?}) is lower than received in the first frame from NDI ({:?}), so skiping...", (video_frame.timestamp as u64), time); - } else { - skip_frame = false; - } - } + let vec = Vec::from_raw_parts(video_frame.p_data as *mut u8, buff_size, buff_size); + // Newtek NDI yields times in 100ns intervals since the Unix Time + let pts: gst::ClockTime = (pts * 100).into(); - gst_log!(self.cat, obj: element, "NDI video frame received: {:?}", (video_frame)); - - pts = video_frame.timestamp as u64 - time; - - gst_log!(self.cat, obj: element, "Calculated pts for video frame: {:?}", (pts)); - - let buff_size = (video_frame.yres * video_frame.line_stride_in_bytes) as usize; - let mut buffer = gst::Buffer::with_size(buff_size).unwrap(); - { - let vec = Vec::from_raw_parts(video_frame.p_data as *mut u8, buff_size, buff_size); - // Newtek NDI yields times in 100ns intervals since the Unix Time - let pts: gst::ClockTime = (pts * 100).into(); - - let duration: gst::ClockTime = (((f64::from(video_frame.frame_rate_D) + let duration: gst::ClockTime = (((f64::from(video_frame.frame_rate_D) / f64::from(video_frame.frame_rate_N)) * 1_000_000_000.0) as u64) .into(); - let buffer = buffer.get_mut().unwrap(); + let buffer = buffer.get_mut().unwrap(); - if ndi_struct.start_pts == gst::ClockTime(Some(0)) { - ndi_struct.start_pts = + if ndi_struct.start_pts == gst::ClockTime(Some(0)) { + ndi_struct.start_pts = element.get_clock().unwrap().get_time() - element.get_base_time(); + } + + buffer.set_pts(pts + ndi_struct.start_pts); + buffer.set_duration(duration); + buffer.set_offset(timestamp_data.offset); + timestamp_data.offset += 1; + buffer.set_offset_end(timestamp_data.offset); + buffer.copy_from_slice(0, &vec).unwrap(); } - buffer.set_pts(pts + ndi_struct.start_pts); - buffer.set_duration(duration); - buffer.set_offset(timestamp_data.offset); - timestamp_data.offset += 1; - buffer.set_offset_end(timestamp_data.offset); - buffer.copy_from_slice(0, &vec).unwrap(); + gst_log!(self.cat, obj: element, "Produced buffer {:?}", buffer); + + Ok(buffer) } - - gst_log!(self.cat, obj: element, "Produced buffer {:?}", buffer); - - Ok(buffer) } } -} -struct NdiVideoSrcStatic; - -impl ImplTypeStatic for NdiVideoSrcStatic { - fn get_name(&self) -> &str { - "NdiVideoSrc" + pub fn register(plugin: &gst::Plugin) -> Result<(), glib::BoolError> { + gst::Element::register(plugin, "ndivideosrc", 0, NdiVideoSrc::get_type()) } - - fn new(&self, element: &BaseSrc) -> Box> { - NdiVideoSrc::new(element) - } - - fn class_init(&self, klass: &mut BaseSrcClass) { - NdiVideoSrc::class_init(klass); - } -} - -pub fn register(plugin: &gst::Plugin) { - let type_ = register_type(NdiVideoSrcStatic); - gst::Element::register(plugin, "ndivideosrc", 0, type_); -}