mirrors/gst-plugins-rs
mirrors/gst-plugins-rs
1
1
Fork 0
mirror of https://gitlab.freedesktop.org/gstreamer/gst-plugins-rs.git synced 2025-01-01 06:48:42 +00:00
Code Issues Projects Releases Wiki Activity

Port to new glib subclassing API #13

Browse source
This commit is contained in:
Daniel Vilar 2018-12-11 17:47:03 +01:00
parent c12f216b59
commit d3ba5efff8
4 changed files with 705 additions and 699 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

12
Cargo.toml
View file

@ -6,13 +6,11 @@ repository = "https://github.com/teltek/gst-plugin-ndi"
license = "LGPL" license = "LGPL"
[dependencies] [dependencies]
gobject-subclass = "0.2" glib = { git = "https://github.com/gtk-rs/glib", features = ["subclassing"] }
gst-plugin = "0.3" gstreamer = { git = "https://gitlab.freedesktop.org/gstreamer/gstreamer-rs", features = ["subclassing"] }
glib = "0.6" gstreamer-base = { git = "https://gitlab.freedesktop.org/gstreamer/gstreamer-rs", features = ["subclassing"] }
gstreamer = "0.12" gstreamer-video = { git = "https://gitlab.freedesktop.org/gstreamer/gstreamer-rs" }
gstreamer-base = "0.12" gstreamer-audio = { git = "https://gitlab.freedesktop.org/gstreamer/gstreamer-rs" }
gstreamer-video = "0.12"
gstreamer-audio = "0.12"
lazy_static = "1.1.0" lazy_static = "1.1.0"
[lib] [lib]

37
src/lib.rs
View file

@ -1,10 +1,7 @@
#![allow(non_camel_case_types, non_upper_case_globals, non_snake_case)] #![allow(non_camel_case_types, non_upper_case_globals, non_snake_case)]
extern crate glib;
extern crate gobject_subclass;
#[macro_use] #[macro_use]
extern crate gst_plugin; extern crate glib;
#[macro_use] #[macro_use]
extern crate gstreamer as gst; extern crate gstreamer as gst;
use gst::prelude::*; use gst::prelude::*;
@ -19,7 +16,7 @@ mod ndiaudiosrc;
pub mod ndisys; pub mod ndisys;
mod ndivideosrc; mod ndivideosrc;
use gst_plugin::base_src::*; // use gst_plugin::base_src::*;
use ndisys::*; use ndisys::*;
use std::ffi::{CStr, CString}; use std::ffi::{CStr, CString};
use std::{thread, time}; use std::{thread, time};
@ -29,10 +26,10 @@ use std::sync::Mutex;
use gst::GstObjectExt; use gst::GstObjectExt;
fn plugin_init(plugin: &gst::Plugin) -> bool { fn plugin_init(plugin: &gst::Plugin) -> Result<(), glib::BoolError> {
ndivideosrc::register(plugin); ndivideosrc::register(plugin)?;
ndiaudiosrc::register(plugin); ndiaudiosrc::register(plugin)?;
true Ok(())
} }
struct ndi_receiver_info { struct ndi_receiver_info {
@ -62,7 +59,7 @@ lazy_static! {
static mut id_receiver: i8 = 0; 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..."); gst_debug!(cat, obj: element, "Starting NDI connection...");
let mut receivers = hashmap_receivers.lock().unwrap(); 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..."); gst_debug!(cat, obj: element, "Closing NDI connection...");
let mut receivers = hashmap_receivers.lock().unwrap(); let mut receivers = hashmap_receivers.lock().unwrap();
{ {
@ -250,14 +247,14 @@ fn stop_ndi(cat: gst::DebugCategory, element: &BaseSrc, id: i8) -> bool {
true true
} }
plugin_define!( gst_plugin_define!(
b"ndi\0", "ndi",
b"NewTek NDI Plugin\0", "NewTek NDI Plugin",
plugin_init, plugin_init,
b"1.0.0\0", "1.0.0",
b"LGPL\0", "LGPL",
b"ndi\0", "ndi",
b"ndi\0", "ndi",
b"https://github.com/teltek/gst-plugin-ndi\0", "https://github.com/teltek/gst-plugin-ndi",
b"2018-04-09\0" "2018-04-09"
); );

691
src/ndiaudiosrc.rs
View file

@ -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;
use glib::subclass;
use glib::subclass::prelude::*;
use gst; use gst;
use gst::prelude::*; use gst::prelude::*;
use gst::subclass::prelude::*;
use gst_audio; use gst_audio;
use gst_base;
use gst_base::prelude::*; use gst_base::prelude::*;
use gst_base::subclass::prelude::*;
use gobject_subclass::object::*;
use gst_plugin::base_src::*;
use gst_plugin::element::*;
use std::sync::Mutex; use std::sync::Mutex;
use std::{i32, u32}; use std::{i32, u32};
@ -43,29 +44,36 @@ impl Default for Settings {
} }
} }
static PROPERTIES: [Property; 3] = [ static PROPERTIES: [subclass::Property; 3] = [
Property::String( subclass::Property("stream-name", || {
glib::ParamSpec::string(
"stream-name", "stream-name",
"Sream Name", "Sream Name",
"Name of the streaming device", "Name of the streaming device",
None, None,
PropertyMutability::ReadWrite, glib::ParamFlags::READWRITE,
), )
Property::String( }),
subclass::Property("ip", || {
glib::ParamSpec::string(
"ip", "ip",
"Stream IP", "Stream IP",
"IP of the streaming device. Ex: 127.0.0.1:5961", "IP of the streaming device. Ex: 127.0.0.1:5961",
None, None,
PropertyMutability::ReadWrite, glib::ParamFlags::READWRITE,
), )
Property::UInt( }),
subclass::Property("loss-threshold", || {
glib::ParamSpec::uint(
"loss-threshold", "loss-threshold",
"Loss threshold", "Loss threshold",
"Loss threshold. If 0 the stream is never closed by the element", "Loss threshold",
(0, 60), 0,
60,
5, 5,
PropertyMutability::ReadWrite, glib::ParamFlags::READWRITE,
), )
}),
]; ];
struct State { struct State {
@ -89,12 +97,17 @@ struct NdiAudioSrc {
timestamp_data: Mutex<TimestampData>, timestamp_data: Mutex<TimestampData>,
} }
impl NdiAudioSrc { impl ObjectSubclass for NdiAudioSrc {
fn new(element: &BaseSrc) -> Box<BaseSrcImpl<BaseSrc>> {
element.set_live(true);
element.set_format(gst::Format::Time);
Box::new(Self { const NAME: &'static str = "NdiAudioSrc";
type ParentType = gst_base::BaseSrc;
type Instance = gst::subclass::ElementInstanceStruct<Self>;
type Class = subclass::simple::ClassStruct<Self>;
glib_object_subclass!();
fn new() -> Self {
Self {
cat: gst::DebugCategory::new( cat: gst::DebugCategory::new(
"ndiaudiosrc", "ndiaudiosrc",
gst::DebugColorFlags::empty(), gst::DebugColorFlags::empty(),
@ -103,10 +116,10 @@ impl NdiAudioSrc {
settings: Mutex::new(Default::default()), settings: Mutex::new(Default::default()),
state: Mutex::new(Default::default()), state: Mutex::new(Default::default()),
timestamp_data: Mutex::new(TimestampData { offset: 0}), timestamp_data: Mutex::new(TimestampData { offset: 0}),
}) }
} }
fn class_init(klass: &mut BaseSrcClass) { fn class_init(klass: &mut subclass::simple::ClassStruct<Self>) {
klass.set_metadata( klass.set_metadata(
"NewTek NDI Audio Source", "NewTek NDI Audio Source",
"Source", "Source",
@ -117,370 +130,358 @@ impl NdiAudioSrc {
let caps = gst::Caps::new_simple( let caps = gst::Caps::new_simple(
"audio/x-raw", "audio/x-raw",
&[ &[
( (
"format", "format",
&gst::List::new(&[ &gst::List::new(&[
//TODO add more formats? //TODO add more formats?
//&gst_audio::AUDIO_FORMAT_F32.to_string(), //&gst_audio::AUDIO_FORMAT_F32.to_string(),
//&gst_audio::AUDIO_FORMAT_F64.to_string(), //&gst_audio::AUDIO_FORMAT_F64.to_string(),
&gst_audio::AUDIO_FORMAT_S16.to_string(), &gst_audio::AUDIO_FORMAT_S16.to_string(),
]), ]),
), ),
("rate", &gst::IntRange::<i32>::new(1, i32::MAX)), ("rate", &gst::IntRange::<i32>::new(1, i32::MAX)),
("channels", &gst::IntRange::<i32>::new(1, i32::MAX)), ("channels", &gst::IntRange::<i32>::new(1, i32::MAX)),
("layout", &"interleaved"), ("layout", &"interleaved"),
("channel-mask", &gst::Bitmask::new(0)), ("channel-mask", &gst::Bitmask::new(0)),
], ],
); );
let src_pad_template = gst::PadTemplate::new( let src_pad_template = gst::PadTemplate::new(
"src", "src",
gst::PadDirection::Src, gst::PadDirection::Src,
gst::PadPresence::Always, gst::PadPresence::Always,
&caps, &caps,
); );
klass.add_pad_template(src_pad_template); klass.add_pad_template(src_pad_template);
klass.install_properties(&PROPERTIES); klass.install_properties(&PROPERTIES);
}
}
impl ObjectImpl<BaseSrc> 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::<BaseSrc>().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!(),
} }
} }
fn get_property(&self, _obj: &glib::Object, id: u32) -> Result<glib::Value, ()> { impl ObjectImpl for NdiAudioSrc {
let prop = &PROPERTIES[id as usize]; glib_object_impl!();
match *prop { fn constructed(&self, obj: &glib::Object) {
Property::String("stream-name", ..) => { self.parent_constructed(obj);
let settings = self.settings.lock().unwrap();
Ok(settings.stream_name.to_value()) let basesrc = obj.downcast_ref::<gst_base::BaseSrc>().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::<gst_base::BaseSrc>().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<glib::Value, ()> {
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<BaseSrc> for NdiAudioSrc { impl ElementImpl for NdiAudioSrc {
fn change_state( fn change_state(
&self, &self,
element: &BaseSrc, element: &gst::Element,
transition: gst::StateChange, transition: gst::StateChange,
) -> gst::StateChangeReturn { ) -> gst::StateChangeReturn {
if transition == gst::StateChange::PausedToPlaying { if transition == gst::StateChange::PausedToPlaying {
let mut receivers = hashmap_receivers.lock().unwrap(); 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(); 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 recv = &receiver.ndi_instance;
let pNDI_recv = recv.recv; let pNDI_recv = recv.recv;
let audio_frame: NDIlib_audio_frame_v2_t = Default::default(); 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; 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 {
while frame_type != NDIlib_frame_type_e::NDIlib_frame_type_audio { unsafe {
frame_type = NDIlib_recv_capture_v2( frame_type =
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<BaseSrc> 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 =
NDIlib_recv_capture_v2(pNDI_recv, ptr::null(), &audio_frame, ptr::null(), 1000); 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); 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<gst::Buffer, gst::FlowReturn> {
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; let mut caps = gst::Caps::truncate(caps);
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)) { let caps = caps.make_mut();
ndi_struct.start_pts = let s = caps.get_mut_structure(0).unwrap();
element.get_clock().unwrap().get_time() - element.get_base_time(); 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<gst::Buffer, gst::FlowError> {
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 pts = audio_frame.timestamp as u64 - 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) 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)) / f64::from(audio_frame.sample_rate))
* 1_000_000_000.0) as u64) * 1_000_000_000.0) as u64)
.into(); .into();
buffer.set_duration(duration); buffer.set_duration(duration);
buffer.set_offset(timestamp_data.offset); buffer.set_offset(timestamp_data.offset);
timestamp_data.offset += audio_frame.no_samples as u64; timestamp_data.offset += audio_frame.no_samples as u64;
buffer.set_offset_end(timestamp_data.offset); buffer.set_offset_end(timestamp_data.offset);
let mut dst: NDIlib_audio_frame_interleaved_16s_t = Default::default(); let mut dst: NDIlib_audio_frame_interleaved_16s_t = Default::default();
dst.reference_level = 0; dst.reference_level = 0;
dst.p_data = buffer.map_writable().unwrap().as_mut_slice().as_mut_ptr() as *mut i16; 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); 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; pub fn register(plugin: &gst::Plugin) -> Result<(), glib::BoolError> {
gst::Element::register(plugin, "ndiaudiosrc", 0, NdiAudioSrc::get_type())
impl ImplTypeStatic<BaseSrc> for NdiAudioSrcStatic {
fn get_name(&self) -> &str {
"NdiAudioSrc"
} }
fn new(&self, element: &BaseSrc) -> Box<BaseSrcImpl<BaseSrc>> {
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_);
}

664
src/ndivideosrc.rs
View file

@ -1,15 +1,17 @@
#![allow(non_camel_case_types, non_upper_case_globals, non_snake_case)] #![allow(non_camel_case_types, non_upper_case_globals, non_snake_case)]
use glib; use glib;
use glib::subclass;
use glib::subclass::prelude::*;
use gst; use gst;
use gst::prelude::*; use gst::prelude::*;
use gst::Fraction; use gst::subclass::prelude::*;
use gst_base;
use gst_base::prelude::*; use gst_base::prelude::*;
use gst_video; use gst_base::subclass::prelude::*;
use gobject_subclass::object::*; use gst::Fraction;
use gst_plugin::base_src::*; use gst_video;
use gst_plugin::element::*;
use std::sync::Mutex; use std::sync::Mutex;
use std::{i32, u32}; use std::{i32, u32};
@ -44,29 +46,36 @@ impl Default for Settings {
} }
} }
static PROPERTIES: [Property; 3] = [ static PROPERTIES: [subclass::Property; 3] = [
Property::String( subclass::Property("stream-name", || {
glib::ParamSpec::string(
"stream-name", "stream-name",
"Stream Name", "Stream Name",
"Name of the streaming device", "Name of the streaming device",
None, None,
PropertyMutability::ReadWrite, glib::ParamFlags::READWRITE,
), )
Property::String( }),
subclass::Property("ip", || {
glib::ParamSpec::string(
"ip", "ip",
"Stream IP", "Stream IP",
"IP of the streaming device. Ex: 127.0.0.1:5961", "IP of the streaming device. Ex: 127.0.0.1:5961",
None, None,
PropertyMutability::ReadWrite, glib::ParamFlags::READWRITE,
), )
Property::UInt( }),
subclass::Property("loss-threshold", || {
glib::ParamSpec::uint(
"loss-threshold", "loss-threshold",
"Loss threshold", "Loss threshold",
"Loss threshold. If 0 the stream is never closed by the element", "Loss threshold",
(0, 60), 0,
60,
5, 5,
PropertyMutability::ReadWrite, glib::ParamFlags::READWRITE,
), )
}),
]; ];
struct State { struct State {
@ -90,12 +99,17 @@ struct NdiVideoSrc {
timestamp_data: Mutex<TimestampData>, timestamp_data: Mutex<TimestampData>,
} }
impl NdiVideoSrc { impl ObjectSubclass for NdiVideoSrc {
fn new(element: &BaseSrc) -> Box<BaseSrcImpl<BaseSrc>> {
element.set_live(true);
element.set_format(gst::Format::Time);
Box::new(Self { const NAME: &'static str = "NdiVideoSrc";
type ParentType = gst_base::BaseSrc;
type Instance = gst::subclass::ElementInstanceStruct<Self>;
type Class = subclass::simple::ClassStruct<Self>;
glib_object_subclass!();
fn new() -> Self {
Self {
cat: gst::DebugCategory::new( cat: gst::DebugCategory::new(
"ndivideosrc", "ndivideosrc",
gst::DebugColorFlags::empty(), gst::DebugColorFlags::empty(),
@ -104,10 +118,10 @@ impl NdiVideoSrc {
settings: Mutex::new(Default::default()), settings: Mutex::new(Default::default()),
state: Mutex::new(Default::default()), state: Mutex::new(Default::default()),
timestamp_data: Mutex::new(TimestampData { offset: 0 }), timestamp_data: Mutex::new(TimestampData { offset: 0 }),
}) }
} }
fn class_init(klass: &mut BaseSrcClass) { fn class_init(klass: &mut subclass::simple::ClassStruct<Self>) {
klass.set_metadata( klass.set_metadata(
"NewTek NDI Video Source", "NewTek NDI Video Source",
"Source", "Source",
@ -120,13 +134,13 @@ impl NdiVideoSrc {
let caps = gst::Caps::new_simple( let caps = gst::Caps::new_simple(
"video/x-raw", "video/x-raw",
&[ &[
( (
"format", "format",
&gst::List::new(&[ &gst::List::new(&[
//TODO add all formats //TODO add all formats
&gst_video::VideoFormat::Uyvy.to_string(), &gst_video::VideoFormat::Uyvy.to_string(),
//&gst_video::VideoFormat::Rgb.to_string(), //&gst_video::VideoFormat::Rgb.to_string(),
//&gst_video::VideoFormat::Gray8.to_string(), //&gst_video::VideoFormat::Gray8.to_string(),
]), ]),
), ),
("width", &gst::IntRange::<i32>::new(0, i32::MAX)), ("width", &gst::IntRange::<i32>::new(0, i32::MAX)),
@ -138,346 +152,342 @@ impl NdiVideoSrc {
gst::Fraction::new(i32::MAX, 1), gst::Fraction::new(i32::MAX, 1),
), ),
), ),
], ],
); );
let src_pad_template = gst::PadTemplate::new( let src_pad_template = gst::PadTemplate::new(
"src", "src",
gst::PadDirection::Src, gst::PadDirection::Src,
gst::PadPresence::Always, gst::PadPresence::Always,
&caps, &caps,
); );
klass.add_pad_template(src_pad_template); klass.add_pad_template(src_pad_template);
klass.install_properties(&PROPERTIES); klass.install_properties(&PROPERTIES);
}
}
impl ObjectImpl<BaseSrc> 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::<BaseSrc>().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!(),
} }
} }
fn get_property(&self, _obj: &glib::Object, id: u32) -> Result<glib::Value, ()> {
let prop = &PROPERTIES[id as usize];
match *prop { impl ObjectImpl for NdiVideoSrc {
Property::String("stream-name", ..) => { glib_object_impl!();
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::<gst_base::BaseSrc>().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::<gst_base::BaseSrc>().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<glib::Value, ()> {
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<BaseSrc> for NdiVideoSrc { impl ElementImpl for NdiVideoSrc {
fn change_state( fn change_state(
&self, &self,
element: &BaseSrc, element: &gst::Element,
transition: gst::StateChange, transition: gst::StateChange,
) -> gst::StateChangeReturn { ) -> gst::StateChangeReturn {
if transition == gst::StateChange::PausedToPlaying { if transition == gst::StateChange::PausedToPlaying {
let mut receivers = hashmap_receivers.lock().unwrap(); 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(); 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 recv = &receiver.ndi_instance;
let pNDI_recv = recv.recv; let pNDI_recv = recv.recv;
let video_frame: NDIlib_video_frame_v2_t = Default::default(); 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; 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 {
while frame_type != NDIlib_frame_type_e::NDIlib_frame_type_video { unsafe {
frame_type = NDIlib_recv_capture_v2( frame_type =
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<BaseSrc> 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 =
NDIlib_recv_capture_v2(pNDI_recv, &video_frame, ptr::null(), ptr::null(), 1000); 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); gst_debug!(self.cat, obj: element, "NDI video frame received: {:?}", video_frame);
}
} }
}
settings.latency = gst::SECOND.mul_div_floor( settings.latency = gst::SECOND.mul_div_floor(
video_frame.frame_rate_D as u64, video_frame.frame_rate_D as u64,
video_frame.frame_rate_N 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 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()); //Creates the video buffers
element.parent_fixate(caps) fn create(
} &self,
element: &gst_base::BaseSrc,
_offset: u64,
_length: u32,
) -> Result<gst::Buffer, gst::FlowError> {
let _settings = &*self.settings.lock().unwrap();
//Creates the video buffers let mut timestamp_data = self.timestamp_data.lock().unwrap();
fn create( let state = self.state.lock().unwrap();
&self, let _info = match state.info {
element: &BaseSrc, None => {
_offset: u64, gst_element_error!(element, gst::CoreError::Negotiation, ["Have no caps yet"]);
_length: u32, return Err(gst::FlowError::NotNegotiated);
) -> Result<gst::Buffer, gst::FlowReturn> { }
let _settings = &*self.settings.lock().unwrap(); Some(ref info) => info.clone(),
};
let receivers = hashmap_receivers.lock().unwrap();
let mut timestamp_data = self.timestamp_data.lock().unwrap(); let recv = &receivers.get(&_settings.id_receiver).unwrap().ndi_instance;
let state = self.state.lock().unwrap(); let pNDI_recv = recv.recv;
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 pts: u64;
let pNDI_recv = recv.recv; let video_frame: NDIlib_video_frame_v2_t = Default::default();
let pts: u64; unsafe {
let video_frame: NDIlib_video_frame_v2_t = Default::default(); let time = receivers.get(&_settings.id_receiver).unwrap().initial_timestamp;
unsafe { let mut skip_frame = true;
let time = receivers.get(&_settings.id_receiver).unwrap().initial_timestamp; let mut count_frame_none = 0;
while skip_frame {
let mut skip_frame = true; let frame_type =
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); 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 || 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{ let vec = Vec::from_raw_parts(video_frame.p_data as *mut u8, buff_size, buff_size);
count_frame_none += 1; // Newtek NDI yields times in 100ns intervals since the Unix Time
continue; let pts: gst::ClockTime = (pts * 100).into();
}
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;
}
}
gst_log!(self.cat, obj: element, "NDI video frame received: {:?}", (video_frame)); let duration: gst::ClockTime = (((f64::from(video_frame.frame_rate_D)
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)
/ f64::from(video_frame.frame_rate_N)) / f64::from(video_frame.frame_rate_N))
* 1_000_000_000.0) as u64) * 1_000_000_000.0) as u64)
.into(); .into();
let buffer = buffer.get_mut().unwrap(); let buffer = buffer.get_mut().unwrap();
if ndi_struct.start_pts == gst::ClockTime(Some(0)) { if ndi_struct.start_pts == gst::ClockTime(Some(0)) {
ndi_struct.start_pts = ndi_struct.start_pts =
element.get_clock().unwrap().get_time() - element.get_base_time(); 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); gst_log!(self.cat, obj: element, "Produced buffer {:?}", buffer);
buffer.set_duration(duration);
buffer.set_offset(timestamp_data.offset); Ok(buffer)
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)
} }
} }
}
struct NdiVideoSrcStatic; pub fn register(plugin: &gst::Plugin) -> Result<(), glib::BoolError> {
gst::Element::register(plugin, "ndivideosrc", 0, NdiVideoSrc::get_type())
impl ImplTypeStatic<BaseSrc> for NdiVideoSrcStatic {
fn get_name(&self) -> &str {
"NdiVideoSrc"
} }
fn new(&self, element: &BaseSrc) -> Box<BaseSrcImpl<BaseSrc>> {
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_);
}