mirrors/gst-plugins-rs
mirrors/gst-plugins-rs
1
1
Fork 0
mirror of https://gitlab.freedesktop.org/gstreamer/gst-plugins-rs.git synced 2024-06-02 13:29:50 +00:00
Code Issues Projects Releases Wiki Activity

Clean code

Browse source
This commit is contained in:
Daniel Vilar 2018-06-27 11:56:11 +02:00
parent 75bba9f0e0
commit 570abeff49
4 changed files with 149 additions and 245 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

229
gst-plugin-ndi/src/lib.rs
View file

@ -26,7 +26,7 @@ pub mod ndilib;
use std::ptr;
use std::{thread, time};
use std::time::{SystemTime, UNIX_EPOCH};
//use std::time::{SystemTime, UNIX_EPOCH};
use std::ffi::{CStr, CString};
use ndilib::*;
use gst_plugin::base_src::*;
@ -41,27 +41,27 @@ fn plugin_init(plugin: &gst::Plugin) -> bool {
}
struct ndi{
struct Ndi{
recv: Option<NdiInstance>,
start_pts: Option<u64>,
//start_pts: Option<u64>,
}
static mut ndi_struct: ndi = ndi{
static mut ndi_struct: Ndi = Ndi{
recv: None,
start_pts: None,
//start_pts: None,
};
fn connect_ndi(element: &BaseSrc, ip: String, stream_name: String) -> bool{
unsafe {
match ndi_struct.recv {
None => {
//gst_element_error!(element, gst::CoreError::Negotiation, ["Have no caps yet"]);
//return true;
}
_ => return true,
};
fn connect_ndi(cat: gst::DebugCategory , element: &BaseSrc, ip: String, stream_name: String) -> bool{
unsafe {
gst_debug!(cat, obj: element, "Starting NDI connection...");
match ndi_struct.recv {
None => {
//gst_element_error!(element, gst::CoreError::Negotiation, ["Have no caps yet"]);
//return true;
}
_ => return true,
};
if !NDIlib_initialize() {
//println!("Cannot run NDI: NDIlib_initialize error.");
gst_element_error!(element, gst::CoreError::Negotiation, ["Cannot run NDI: NDIlib_initialize error"]);
return false;
}
@ -75,26 +75,21 @@ fn connect_ndi(element: &BaseSrc, ip: String, stream_name: String) -> bool{
let ip_ptr = CString::new(ip.clone()).unwrap();
if ip_ptr == CString::new("").unwrap(){
if pNDI_find.is_null() {
//println!("Cannot run NDI: NDIlib_find_create_v2 error.");
gst_element_error!(element, gst::CoreError::Negotiation, ["Cannot run NDI: NDIlib_find_create_v2 error"]);
return false;
}
let mut total_sources: u32 = 0;
let p_sources;
//TODO Delete while. If not, will loop until a source it's available
//while total_sources == 0 {
// TODO Sleep 1s to wait for all sources
thread::sleep(time::Duration::from_millis(2000));
p_sources = NDIlib_find_get_current_sources(pNDI_find, &mut total_sources as *mut u32);
//}
// We need at least one source
if p_sources.is_null() {
//println!("Error getting NDIlib_find_get_current_sources.");
gst_element_error!(element, gst::CoreError::Negotiation, ["Error getting NDIlib_find_get_current_sources"]);
return false;
//::std::process::exit(1);
}
let mut no_source: isize = -1;
@ -110,107 +105,107 @@ fn connect_ndi(element: &BaseSrc, ip: String, stream_name: String) -> bool{
gst_element_error!(element, gst::CoreError::Negotiation, ["Stream name not found"]);
return false;
}
println!(
"Total_sources {}: Name '{}' Address '{}'",
total_sources,
CStr::from_ptr((*p_sources.offset(no_source)).p_ndi_name)
.to_string_lossy()
.into_owned(),
CStr::from_ptr((*p_sources.offset(no_source)).p_ip_address)
.to_string_lossy()
.into_owned()
);
gst_debug!(cat, obj: element, "Total sources in network {}: Connecting to NDI source with name '{}' and address '{}'", total_sources,
CStr::from_ptr((*p_sources.offset(no_source)).p_ndi_name)
.to_string_lossy()
.into_owned(),
CStr::from_ptr((*p_sources.offset(no_source)).p_ip_address)
.to_string_lossy()
.into_owned());
source = *p_sources.offset(no_source).clone();
}
else{
source.p_ip_address = ip_ptr.as_ptr();
println!(
"Address '{}'",
CStr::from_ptr(source.p_ip_address)
.to_string_lossy()
.into_owned()
);
}
// We now have at least one source, so we create a receiver to look at it.
// We tell it that we prefer YCbCr video since it is more efficient for us. If the source has an alpha channel
// it will still be provided in BGRA
let p_ndi_name = CString::new("Galicaster NDI Receiver").unwrap();
let NDI_recv_create_desc = NDIlib_recv_create_v3_t {
source_to_connect_to: source,
p_ndi_name: p_ndi_name.as_ptr(),
..Default::default()
};
let pNDI_recv = NDIlib_recv_create_v3(&NDI_recv_create_desc);
if pNDI_recv.is_null() {
//println!("Cannot run NDI: NDIlib_recv_create_v3 error.");
gst_element_error!(element, gst::CoreError::Negotiation, ["Cannot run NDI: NDIlib_recv_create_v3 error"]);
return false;
//::std::process::exit(1);
}
// Destroy the NDI finder. We needed to have access to the pointers to p_sources[0]
NDIlib_find_destroy(pNDI_find);
// We are now going to mark this source as being on program output for tally purposes (but not on preview)
let tally_state: NDIlib_tally_t = Default::default();
NDIlib_recv_set_tally(pNDI_recv, &tally_state);
// Enable Hardwqre Decompression support if this support has it. Please read the caveats in the documentation
// regarding this. There are times in which it might reduce the performance although on small stream numbers
// it almost always yields the same or better performance.
let data = CString::new("<ndi_hwaccel enabled=\"true\"/>").unwrap();
let enable_hw_accel = NDIlib_metadata_frame_t {
length: data.to_bytes().len() as i32,
timecode: 0,
p_data: data.as_ptr(),
};
NDIlib_recv_send_metadata(pNDI_recv, &enable_hw_accel);
ndi_struct.recv = Some(NdiInstance{recv: pNDI_recv});
let start = SystemTime::now();
let since_the_epoch = start.duration_since(UNIX_EPOCH)
.expect("Time went backwards");
println!("{:?}", since_the_epoch);
ndi_struct.start_pts = Some(since_the_epoch.as_secs() * 1000000000 +
since_the_epoch.subsec_nanos() as u64);
return true;
gst_debug!(cat, obj: element, "Connecting to NDI source with address '{}'", CStr::from_ptr(source.p_ip_address)
.to_string_lossy()
.into_owned()
);
}
}
fn stop_ndi() -> bool{
unsafe{
let recv = match ndi_struct.recv{
None => {
//TODO Update gst_element_error with one more descriptive
//println!("pNDI_recv no encontrado");
//gst_element_error!(element, gst::CoreError::Negotiation, ["No encontramos ndi recv"]);
return true;
}
Some(ref recv) => recv.clone(),
};
let pNDI_recv = recv.recv;
NDIlib_recv_destroy(pNDI_recv);
ndi_struct.recv = None;
//NDIlib_destroy();
return true;
// We now have at least one source, so we create a receiver to look at it.
// We tell it that we prefer YCbCr video since it is more efficient for us. If the source has an alpha channel
// it will still be provided in BGRA
let p_ndi_name = CString::new("Galicaster NDI Receiver").unwrap();
let NDI_recv_create_desc = NDIlib_recv_create_v3_t {
source_to_connect_to: source,
p_ndi_name: p_ndi_name.as_ptr(),
..Default::default()
};
let pNDI_recv = NDIlib_recv_create_v3(&NDI_recv_create_desc);
if pNDI_recv.is_null() {
//println!("Cannot run NDI: NDIlib_recv_create_v3 error.");
gst_element_error!(element, gst::CoreError::Negotiation, ["Cannot run NDI: NDIlib_recv_create_v3 error"]);
return false;
}
}
// Static plugin metdata that is directly stored in the plugin shared object and read by GStreamer
// upon loading.
// Plugin name, plugin description, plugin entry point function, version number of this plugin,
// license of the plugin, source package name, binary package name, origin where it comes from
// and the date/time of release.
plugin_define!(
b"ndi\0",
b"NewTek NDI Plugin\0",
plugin_init,
b"1.0\0",
b"MIT/X11\0",
b"ndi\0",
b"ndi\0",
b"https://gitlab.teltek.es/rubenrua/ndi-rs.git\0",
b"2018-04-09\0"
);
// Destroy the NDI finder. We needed to have access to the pointers to p_sources[0]
NDIlib_find_destroy(pNDI_find);
// We are now going to mark this source as being on program output for tally purposes (but not on preview)
let tally_state: NDIlib_tally_t = Default::default();
NDIlib_recv_set_tally(pNDI_recv, &tally_state);
// Enable Hardwqre Decompression support if this support has it. Please read the caveats in the documentation
// regarding this. There are times in which it might reduce the performance although on small stream numbers
// it almost always yields the same or better performance.
let data = CString::new("<ndi_hwaccel enabled=\"true\"/>").unwrap();
let enable_hw_accel = NDIlib_metadata_frame_t {
length: data.to_bytes().len() as i32,
timecode: 0,
p_data: data.as_ptr(),
};
NDIlib_recv_send_metadata(pNDI_recv, &enable_hw_accel);
ndi_struct.recv = Some(NdiInstance{recv: pNDI_recv});
// let start = SystemTime::now();
// let since_the_epoch = start.duration_since(UNIX_EPOCH)
// .expect("Time went backwards");
// println!("{:?}", since_the_epoch);
// ndi_struct.start_pts = Some(since_the_epoch.as_secs() * 1000000000 +
// since_the_epoch.subsec_nanos() as u64);
gst_debug!(cat, obj: element, "Started NDI connection");
return true;
}
}
fn stop_ndi(cat: gst::DebugCategory , element: &BaseSrc) -> bool{
gst_debug!(cat, obj: element, "Closing NDI connection...");
unsafe{
let recv = match ndi_struct.recv{
None => {
//TODO Update gst_element_error with one more descriptive
//println!("pNDI_recv no encontrado");
//gst_element_error!(element, gst::CoreError::Negotiation, ["No encontramos ndi recv"]);
return true;
}
Some(ref recv) => recv.clone(),
};
let pNDI_recv = recv.recv;
NDIlib_recv_destroy(pNDI_recv);
ndi_struct.recv = None;
NDIlib_destroy();
gst_debug!(cat, obj: element, "Closed NDI connection");
return true;
}
}
// Static plugin metdata that is directly stored in the plugin shared object and read by GStreamer
// upon loading.
// Plugin name, plugin description, plugin entry point function, version number of this plugin,
// license of the plugin, source package name, binary package name, origin where it comes from
// and the date/time of release.
plugin_define!(
b"ndi\0",
b"NewTek NDI Plugin\0",
plugin_init,
b"1.0\0",
b"MIT/X11\0",
b"ndi\0",
b"ndi\0",
b"https://gitlab.teltek.es/rubenrua/ndi-rs.git\0",
b"2018-04-09\0"
);

78
gst-plugin-ndi/src/ndiaudiosrc.rs
View file

@ -69,11 +69,7 @@ impl Default for State {
}
}
struct ClockWait {
clock_id: Option<gst::ClockId>,
flushing: bool,
}
struct Pts{
struct TimestampData{
pts: u64,
offset: u64,
}
@ -83,8 +79,7 @@ struct NdiAudioSrc {
cat: gst::DebugCategory,
settings: Mutex<Settings>,
state: Mutex<State>,
clock_wait: Mutex<ClockWait>,
pts: Mutex<Pts>,
timestamp_data: Mutex<TimestampData>,
}
impl NdiAudioSrc {
@ -103,11 +98,7 @@ impl NdiAudioSrc {
),
settings: Mutex::new(Default::default()),
state: Mutex::new(Default::default()),
clock_wait: Mutex::new(ClockWait {
clock_id: None,
flushing: true,
}),
pts: Mutex::new(Pts{
timestamp_data: Mutex::new(TimestampData{
pts: 0,
offset: 0,
}),
@ -180,7 +171,7 @@ impl NdiAudioSrc {
Property::String("stream-name", ..) => {
let mut settings = self.settings.lock().unwrap();
let stream_name = value.get().unwrap();
gst_warning!(
gst_debug!(
self.cat,
obj: &element,
"Changing stream-name from {} to {}",
@ -196,7 +187,7 @@ impl NdiAudioSrc {
Property::String("ip", ..) => {
let mut settings = self.settings.lock().unwrap();
let ip = value.get().unwrap();
gst_warning!(
gst_debug!(
self.cat,
obj: &element,
"Changing ip from {} to {}",
@ -266,11 +257,9 @@ impl NdiAudioSrc {
fn start(&self, element: &BaseSrc) -> bool {
// Reset state
*self.state.lock().unwrap() = Default::default();
self.unlock_stop(element);
gst_warning!(self.cat, obj: element, "Starting");
let settings = self.settings.lock().unwrap();
return connect_ndi(element, settings.ip.clone(), settings.stream_name.clone());
return connect_ndi(self.cat, element, settings.ip.clone(), settings.stream_name.clone());
}
@ -278,12 +267,9 @@ impl NdiAudioSrc {
fn stop(&self, element: &BaseSrc) -> bool {
// Reset state
*self.state.lock().unwrap() = Default::default();
stop_ndi();
stop_ndi(self.cat, element);
// Commented because when adding ndi destroy stopped in this line
//*self.state.lock().unwrap() = Default::default();
self.unlock(element);
gst_info!(self.cat, obj: element, "Stopped");
true
}
@ -301,14 +287,14 @@ impl NdiAudioSrc {
};
let pNDI_recv = recv.recv;
let mut pts2 = self.pts.lock().unwrap();
let mut timestamp_data = self.timestamp_data.lock().unwrap();
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{
frame_type = NDIlib_recv_capture_v2(pNDI_recv, ptr::null(), &audio_frame, ptr::null(), 1000);
}
pts2.pts = (audio_frame.timecode as u64) * 100;
timestamp_data.pts = (audio_frame.timecode as u64) * 100;
let mut caps = gst::Caps::truncate(caps);
{
@ -337,7 +323,7 @@ impl NdiAudioSrc {
// have to block until this function returns when getting/setting property values
let _settings = &*self.settings.lock().unwrap();
let mut pts2 = self.pts.lock().unwrap();
let mut timestamp_data = self.timestamp_data.lock().unwrap();
// Get a locked reference to our state, i.e. the input and output AudioInfo
let state = self.state.lock().unwrap();
let _info = match state.info {
@ -359,16 +345,11 @@ impl NdiAudioSrc {
};
let pNDI_recv = recv.recv;
let pts: u64;
let audio_frame: NDIlib_audio_frame_v2_t = Default::default();
NDIlib_recv_capture_v2(
pNDI_recv,
ptr::null(),
&audio_frame,
ptr::null(),
1000,
);
pts = ((audio_frame.timecode as u64) * 100) - pts2.pts;
let audio_frame: NDIlib_audio_frame_v2_t = Default::default();
NDIlib_recv_capture_v2(pNDI_recv, ptr::null(), &audio_frame, ptr::null(), 1000,);
pts = ((audio_frame.timecode as u64) * 100) - timestamp_data.pts;
let buff_size = ((audio_frame.channel_stride_in_bytes)) as usize;
let mut buffer = gst::Buffer::with_size(buff_size).unwrap();
{
@ -379,40 +360,15 @@ impl NdiAudioSrc {
let buffer = buffer.get_mut().unwrap();
buffer.set_pts(pts);
buffer.set_duration(duration);
buffer.set_offset(pts2.offset);
buffer.set_offset_end(pts2.offset + 1);
pts2.offset = pts2.offset +1;
buffer.set_offset(timestamp_data.offset);
buffer.set_offset_end(timestamp_data.offset + 1);
timestamp_data.offset = timestamp_data.offset +1;
buffer.copy_from_slice(0, &vec).unwrap();
}
gst_debug!(self.cat, obj: element, "Produced buffer {:?}", buffer);
Ok(buffer)
}
}
fn unlock(&self, element: &BaseSrc) -> bool {
// This should unblock the create() function ASAP, so we
// just unschedule the clock it here, if any.
gst_debug!(self.cat, obj: element, "Unlocking");
let mut clock_wait = self.clock_wait.lock().unwrap();
if let Some(clock_id) = clock_wait.clock_id.take() {
clock_id.unschedule();
}
clock_wait.flushing = true;
true
}
fn unlock_stop(&self, element: &BaseSrc) -> bool {
// This signals that unlocking is done, so we can reset
// all values again.
gst_debug!(self.cat, obj: element, "Unlock stop");
let mut clock_wait = self.clock_wait.lock().unwrap();
clock_wait.flushing = false;
true
}
}

1
gst-plugin-ndi/src/ndilib.rs
View file

@ -17,6 +17,7 @@ extern "C" {
) -> NDIlib_recv_instance_t;
pub fn NDIlib_find_destroy(p_instance: NDIlib_recv_instance_t);
pub fn NDIlib_recv_destroy(p_instance: NDIlib_recv_instance_t);
pub fn NDIlib_destroy();
pub fn NDIlib_recv_set_tally(
p_instance: NDIlib_recv_instance_t,
p_tally: *const NDIlib_tally_t,

86
gst-plugin-ndi/src/ndivideosrc.rs
View file

@ -70,11 +70,7 @@ impl Default for State {
}
}
struct ClockWait {
clock_id: Option<gst::ClockId>,
flushing: bool,
}
struct Pts{
struct TimestampData{
pts: u64,
offset: u64,
}
@ -84,8 +80,7 @@ struct NdiVideoSrc {
cat: gst::DebugCategory,
settings: Mutex<Settings>,
state: Mutex<State>,
clock_wait: Mutex<ClockWait>,
pts: Mutex<Pts>,
timestamp_data: Mutex<TimestampData>,
}
impl NdiVideoSrc {
@ -104,11 +99,7 @@ impl NdiVideoSrc {
),
settings: Mutex::new(Default::default()),
state: Mutex::new(Default::default()),
clock_wait: Mutex::new(ClockWait {
clock_id: None,
flushing: true,
}),
pts: Mutex::new(Pts{
timestamp_data: Mutex::new(TimestampData{
pts: 0,
offset: 0,
}),
@ -188,7 +179,7 @@ impl NdiVideoSrc {
Property::String("stream-name", ..) => {
let mut settings = self.settings.lock().unwrap();
let stream_name = value.get().unwrap();
gst_warning!(
gst_debug!(
self.cat,
obj: &element,
"Changing stream-name from {} to {}",
@ -204,7 +195,7 @@ impl NdiVideoSrc {
Property::String("ip", ..) => {
let mut settings = self.settings.lock().unwrap();
let ip = value.get().unwrap();
gst_warning!(
gst_debug!(
self.cat,
obj: &element,
"Changing ip from {} to {}",
@ -229,12 +220,10 @@ impl NdiVideoSrc {
match *prop {
Property::String("stream-name", ..) => {
let settings = self.settings.lock().unwrap();
//TODO to_value supongo que solo funciona con numeros
Ok(settings.stream_name.to_value())
},
Property::String("ip", ..) => {
let settings = self.settings.lock().unwrap();
//TODO to_value supongo que solo funciona con numeros
Ok(settings.ip.to_value())
}
_ => unimplemented!(),
@ -274,23 +263,18 @@ impl NdiVideoSrc {
fn start(&self, element: &BaseSrc) -> bool {
// Reset state
*self.state.lock().unwrap() = Default::default();
self.unlock_stop(element);
gst_warning!(self.cat, obj: element, "Starting");
let settings = self.settings.lock().unwrap();
return connect_ndi(element, settings.ip.clone(), settings.stream_name.clone());
return connect_ndi(self.cat, element, settings.ip.clone(), settings.stream_name.clone());
}
// Called when shutting down the element so we can release all stream-related state
fn stop(&self, element: &BaseSrc) -> bool {
// Reset state
*self.state.lock().unwrap() = Default::default();
stop_ndi();
stop_ndi(self.cat, element);
// Commented because when adding ndi destroy stopped in this line
//*self.state.lock().unwrap() = Default::default();
self.unlock(element);
gst_info!(self.cat, obj: element, "Stopped");
true
}
@ -309,14 +293,16 @@ impl NdiVideoSrc {
};
let pNDI_recv = recv.recv;
let mut pts2 = self.pts.lock().unwrap();
let mut timestamp_data = self.timestamp_data.lock().unwrap();
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{
frame_type = NDIlib_recv_capture_v2(pNDI_recv, &video_frame, ptr::null(), ptr::null(), 1000);
}
pts2.pts = (video_frame.timecode as u64) * 100;
//FIXME It's possible than timecode not exist
timestamp_data.pts = (video_frame.timecode as u64) * 100;
let mut caps = gst::Caps::truncate(caps);
{
@ -325,9 +311,6 @@ impl NdiVideoSrc {
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));
//s.fixate_field_str("format", &gst_video::VideoFormat::Rgb.to_string());
//caps.set_simple(&[("width", &(1600 as i32))]);
//s.set_value("width", &(1600 as i32));
}
// Let BaseSrc fixate anything else for us. We could've alternatively have
@ -336,7 +319,7 @@ impl NdiVideoSrc {
}
}
//Creates the audio buffers
//Creates the video buffers
fn create(
&self,
element: &BaseSrc,
@ -348,7 +331,7 @@ impl NdiVideoSrc {
// have to block until this function returns when getting/setting property values
let _settings = &*self.settings.lock().unwrap();
let mut pts2 = self.pts.lock().unwrap();
let mut timestamp_data = self.timestamp_data.lock().unwrap();
// Get a locked reference to our state, i.e. the input and output AudioInfo
let state = self.state.lock().unwrap();
let _info = match state.info {
@ -372,15 +355,9 @@ impl NdiVideoSrc {
let pts: u64;
let video_frame: NDIlib_video_frame_v2_t = Default::default();
NDIlib_recv_capture_v2(
pNDI_recv,
&video_frame,
ptr::null(),
ptr::null(),
1000,
);
pts = ((video_frame.timecode as u64) * 100) - pts2.pts;
NDIlib_recv_capture_v2(pNDI_recv, &video_frame, ptr::null(), ptr::null(), 1000,);
//TODO It's possible than timecode not exist
pts = ((video_frame.timecode as u64) * 100) - timestamp_data.pts;
let buff_size = (video_frame.yres * video_frame.line_stride_in_bytes) as usize;
//println!("{:?}", buff_size);
@ -393,9 +370,9 @@ impl NdiVideoSrc {
let buffer = buffer.get_mut().unwrap();
buffer.set_pts(pts);
buffer.set_duration(duration);
buffer.set_offset(pts2.offset);
buffer.set_offset_end(pts2.offset + 1);
pts2.offset = pts2.offset +1;
buffer.set_offset(timestamp_data.offset);
buffer.set_offset_end(timestamp_data.offset + 1);
timestamp_data.offset = timestamp_data.offset +1;
buffer.copy_from_slice(0, &vec).unwrap();
}
@ -404,31 +381,6 @@ impl NdiVideoSrc {
Ok(buffer)
}
}
fn unlock(&self, element: &BaseSrc) -> bool {
// This should unblock the create() function ASAP, so we
// just unschedule the clock it here, if any.
gst_debug!(self.cat, obj: element, "Unlocking");
let mut clock_wait = self.clock_wait.lock().unwrap();
if let Some(clock_id) = clock_wait.clock_id.take() {
clock_id.unschedule();
}
clock_wait.flushing = true;
true
}
fn unlock_stop(&self, element: &BaseSrc) -> bool {
// This signals that unlocking is done, so we can reset
// all values again.
gst_debug!(self.cat, obj: element, "Unlock stop");
let mut clock_wait = self.clock_wait.lock().unwrap();
clock_wait.flushing = false;
true
}
}
// This zero-sized struct is containing the static metadata of our element. It is only necessary to