diff --git a/gst-plugin-threadshare/Cargo.toml b/gst-plugin-threadshare/Cargo.toml index b55d98b0..37cb4247 100644 --- a/gst-plugin-threadshare/Cargo.toml +++ b/gst-plugin-threadshare/Cargo.toml @@ -41,3 +41,7 @@ path = "examples/udpsrc_benchmark.rs" [[example]] name = "udpsrc-benchmark-sender" path = "examples/udpsrc_benchmark_sender.rs" + +[[example]] +name = "tcpclientsrc-benchmark-sender" +path = "examples/tcpclientsrc_benchmark_sender.rs" diff --git a/gst-plugin-threadshare/examples/tcpclientsrc_benchmark_sender.rs b/gst-plugin-threadshare/examples/tcpclientsrc_benchmark_sender.rs new file mode 100644 index 00000000..ec86b8b9 --- /dev/null +++ b/gst-plugin-threadshare/examples/tcpclientsrc_benchmark_sender.rs @@ -0,0 +1,49 @@ +// Copyright (C) 2018 LEE Dongjun +// +// This library is free software; you can redistribute it and/or +// modify it under the terms of the GNU Library General Public +// License as published by the Free Software Foundation; either +// version 2 of the License, or (at your option) any later version. +// +// This library is distributed in the hope that it will be useful, +// but WITHOUT ANY WARRANTY; without even the implied warranty of +// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU +// Library General Public License for more details. +// +// You should have received a copy of the GNU Library General Public +// License along with this library; if not, write to the +// Free Software Foundation, Inc., 51 Franklin Street, Suite 500, +// Boston, MA 02110-1335, USA. + +use std::io::Write; +use std::sync::{Arc, Mutex}; +use std::{net, thread, time}; + +fn main() { + let sockets: Arc>> = Arc::new(Mutex::new(vec![])); + + let streams = sockets.clone(); + let _handler = thread::spawn(move || { + let listener = net::TcpListener::bind("0.0.0.0:40000").unwrap(); + for stream in listener.incoming() { + streams.lock().unwrap().push(stream.unwrap()); + } + }); + + let buffer = [0; 160]; + let wait = time::Duration::from_millis(20); + + let streams = sockets.clone(); + loop { + let now = time::Instant::now(); + + for mut stream in streams.lock().unwrap().iter() { + let _ = stream.write(&buffer); + } + + let elapsed = now.elapsed(); + if elapsed < wait { + thread::sleep(wait - elapsed); + } + } +} diff --git a/gst-plugin-threadshare/examples/udpsrc_benchmark.rs b/gst-plugin-threadshare/examples/udpsrc_benchmark.rs index ba97b94b..35c50398 100644 --- a/gst-plugin-threadshare/examples/udpsrc_benchmark.rs +++ b/gst-plugin-threadshare/examples/udpsrc_benchmark.rs @@ -95,6 +95,20 @@ fn main() { source } + "ts-tcpclientsrc" => { + let source = gst::ElementFactory::make( + "ts-tcpclientsrc", + Some(format!("source-{}", i).as_str()), + ).unwrap(); + source.set_property("port", &(40000u32)).unwrap(); + source + .set_property("context", &format!("context-{}", (i as u32) % n_groups)) + .unwrap(); + source.set_property("context-threads", &n_threads).unwrap(); + source.set_property("context-wait", &wait).unwrap(); + + source + } "tonegeneratesrc" => { let source = gst::ElementFactory::make( "tonegeneratesrc", diff --git a/gst-plugin-threadshare/src/lib.rs b/gst-plugin-threadshare/src/lib.rs index 00c39525..95308c0a 100644 --- a/gst-plugin-threadshare/src/lib.rs +++ b/gst-plugin-threadshare/src/lib.rs @@ -47,6 +47,7 @@ extern crate net2; mod iocontext; mod socket; +mod tcpclientsrc; mod udpsrc; mod appsrc; @@ -56,6 +57,7 @@ mod queue; fn plugin_init(plugin: &gst::Plugin) -> bool { udpsrc::register(plugin); + tcpclientsrc::register(plugin); queue::register(plugin); proxy::register(plugin); appsrc::register(plugin); diff --git a/gst-plugin-threadshare/src/tcpclientsrc.rs b/gst-plugin-threadshare/src/tcpclientsrc.rs new file mode 100644 index 00000000..7e9015cc --- /dev/null +++ b/gst-plugin-threadshare/src/tcpclientsrc.rs @@ -0,0 +1,767 @@ +// Copyright (C) 2018 Sebastian Dröge +// Copyright (C) 2018 LEE Dongjun +// +// This library is free software; you can redistribute it and/or +// modify it under the terms of the GNU Library General Public +// License as published by the Free Software Foundation; either +// version 2 of the License, or (at your option) any later version. +// +// This library is distributed in the hope that it will be useful, +// but WITHOUT ANY WARRANTY; without even the implied warranty of +// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU +// Library General Public License for more details. +// +// You should have received a copy of the GNU Library General Public +// License along with this library; if not, write to the +// Free Software Foundation, Inc., 51 Franklin Street, Suite 500, +// Boston, MA 02110-1335, USA. + +use glib; +use glib::prelude::*; +use gst; +use gst::prelude::*; + +use gobject_subclass::object::*; +use gst_plugin::element::*; + +use std::io; +use std::sync::Mutex; +use std::u16; + +use futures; +use futures::future; +use futures::{Async, Future, Poll}; +use tokio::io::AsyncRead; +use tokio::net; + +use either::Either; + +use rand; + +use iocontext::*; +use socket::*; + +const DEFAULT_ADDRESS: Option<&'static str> = Some("127.0.0.1"); +const DEFAULT_PORT: u32 = 5000; +const DEFAULT_CAPS: Option = None; +const DEFAULT_CHUNK_SIZE: u32 = 4096; +const DEFAULT_CONTEXT: &'static str = ""; +const DEFAULT_CONTEXT_THREADS: i32 = 0; +const DEFAULT_CONTEXT_WAIT: u32 = 0; + +#[derive(Debug, Clone)] +struct Settings { + address: Option, + port: u32, + caps: Option, + chunk_size: u32, + context: String, + context_threads: i32, + context_wait: u32, +} + +impl Default for Settings { + fn default() -> Self { + Settings { + address: DEFAULT_ADDRESS.map(Into::into), + port: DEFAULT_PORT, + caps: DEFAULT_CAPS, + chunk_size: DEFAULT_CHUNK_SIZE, + context: DEFAULT_CONTEXT.into(), + context_threads: DEFAULT_CONTEXT_THREADS, + context_wait: DEFAULT_CONTEXT_WAIT, + } + } +} + +static PROPERTIES: [Property; 7] = [ + Property::String( + "address", + "Address", + "Address to receive packets from", + DEFAULT_ADDRESS, + PropertyMutability::ReadWrite, + ), + Property::UInt( + "port", + "Port", + "Port to receive packets from", + (0, u16::MAX as u32), + DEFAULT_PORT, + PropertyMutability::ReadWrite, + ), + Property::Boxed( + "caps", + "Caps", + "Caps to use", + gst::Caps::static_type, + PropertyMutability::ReadWrite, + ), + Property::UInt( + "chunk-size", + "Chunk Size", + "Chunk Size", + (0, u16::MAX as u32), + DEFAULT_CHUNK_SIZE, + PropertyMutability::ReadWrite, + ), + Property::String( + "context", + "Context", + "Context name to share threads with", + Some(DEFAULT_CONTEXT), + PropertyMutability::ReadWrite, + ), + Property::Int( + "context-threads", + "Context Threads", + "Number of threads for the context thread-pool if we create it", + (-1, u16::MAX as i32), + DEFAULT_CONTEXT_THREADS, + PropertyMutability::ReadWrite, + ), + Property::UInt( + "context-wait", + "Context Wait", + "Throttle poll loop to run at most once every this many ms", + (0, 1000), + DEFAULT_CONTEXT_WAIT, + PropertyMutability::ReadWrite, + ), +]; + +pub struct TcpClientReader { + connect_future: net::ConnectFuture, + socket: Option, +} + +impl TcpClientReader { + pub fn new(connect_future: net::ConnectFuture) -> Self { + Self { + connect_future: connect_future, + socket: None, + } + } +} + +impl SocketRead for TcpClientReader { + const DO_TIMESTAMP: bool = false; + + fn poll_read(&mut self, buf: &mut [u8]) -> Poll { + let socket = match self.socket { + Some(ref mut socket) => socket, + None => match self.connect_future.poll() { + Ok(Async::Ready(stream)) => { + self.socket = Some(stream); + self.socket.as_mut().unwrap() + } + Err(err) => { + return Err(err); + } + _ => return Ok(Async::NotReady), + }, + }; + + socket.poll_read(buf) + } +} + +struct State { + io_context: Option, + pending_future_id: Option, + socket: Option>, + need_initial_events: bool, + configured_caps: Option, + pending_future_cancel: Option>, +} + +impl Default for State { + fn default() -> State { + State { + io_context: None, + pending_future_id: None, + socket: None, + need_initial_events: true, + configured_caps: None, + pending_future_cancel: None, + } + } +} + +struct TcpClientSrc { + cat: gst::DebugCategory, + src_pad: gst::Pad, + state: Mutex, + settings: Mutex, +} + +impl TcpClientSrc { + fn class_init(klass: &mut ElementClass) { + klass.set_metadata( + "Thread-sharing TCP client source", + "Source/Network", + "Receives data over the network via TCP", + "Sebastian Dröge , LEE Dongjun ", + ); + + let caps = gst::Caps::new_any(); + 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); + } + + fn init(element: &Element) -> Box> { + let templ = element.get_pad_template("src").unwrap(); + let src_pad = gst::Pad::new_from_template(&templ, "src"); + + src_pad.set_event_function(|pad, parent, event| { + TcpClientSrc::catch_panic_pad_function( + parent, + || false, + |tcpclientsrc, element| tcpclientsrc.src_event(pad, element, event), + ) + }); + src_pad.set_query_function(|pad, parent, query| { + TcpClientSrc::catch_panic_pad_function( + parent, + || false, + |tcpclientsrc, element| tcpclientsrc.src_query(pad, element, query), + ) + }); + element.add_pad(&src_pad).unwrap(); + + ::set_element_flags(element, gst::ElementFlags::SOURCE); + + Box::new(Self { + cat: gst::DebugCategory::new( + "ts-tcpclientsrc", + gst::DebugColorFlags::empty(), + "Thread-sharing TCP Client source", + ), + src_pad: src_pad, + state: Mutex::new(State::default()), + settings: Mutex::new(Settings::default()), + }) + } + + fn src_event(&self, pad: &gst::Pad, element: &Element, event: gst::Event) -> bool { + use gst::EventView; + + gst_log!(self.cat, obj: pad, "Handling event {:?}", event); + + let ret = match event.view() { + EventView::FlushStart(..) => { + let _ = self.stop(element); + true + } + EventView::FlushStop(..) => { + let (ret, state, pending) = element.get_state(0.into()); + if ret == gst::StateChangeReturn::Success && state == gst::State::Playing + || ret == gst::StateChangeReturn::Async && pending == gst::State::Playing + { + let _ = self.start(element); + } + true + } + EventView::Reconfigure(..) => true, + EventView::Latency(..) => true, + _ => false, + }; + + if ret { + gst_log!(self.cat, obj: pad, "Handled event {:?}", event); + } else { + gst_log!(self.cat, obj: pad, "Didn't handle event {:?}", event); + } + ret + } + + fn src_query(&self, pad: &gst::Pad, _element: &Element, query: &mut gst::QueryRef) -> bool { + use gst::QueryView; + + gst_log!(self.cat, obj: pad, "Handling query {:?}", query); + let ret = match query.view_mut() { + QueryView::Latency(ref mut q) => { + q.set(false, 0.into(), 0.into()); + true + } + QueryView::Scheduling(ref mut q) => { + q.set(gst::SchedulingFlags::SEQUENTIAL, 1, -1, 0); + q.add_scheduling_modes(&[gst::PadMode::Push]); + true + } + QueryView::Caps(ref mut q) => { + let state = self.state.lock().unwrap(); + let caps = if let Some(ref caps) = state.configured_caps { + q.get_filter() + .map(|f| f.intersect_with_mode(caps, gst::CapsIntersectMode::First)) + .unwrap_or(caps.clone()) + } else { + q.get_filter() + .map(|f| f.to_owned()) + .unwrap_or(gst::Caps::new_any()) + }; + + q.set_result(&caps); + + true + } + _ => false, + }; + + if ret { + gst_log!(self.cat, obj: pad, "Handled query {:?}", query); + } else { + gst_log!(self.cat, obj: pad, "Didn't handle query {:?}", query); + } + ret + } + + fn create_io_context_event(state: &State) -> Option { + if let (&Some(ref pending_future_id), &Some(ref io_context)) = + (&state.pending_future_id, &state.io_context) + { + let s = gst::Structure::new( + "ts-io-context", + &[ + ("io-context", &glib::AnySendValue::new(io_context.clone())), + ( + "pending-future-id", + &glib::AnySendValue::new(*pending_future_id), + ), + ], + ); + Some(gst::Event::new_custom_downstream_sticky(s).build()) + } else { + None + } + } + + fn push_buffer( + &self, + element: &Element, + buffer: gst::Buffer, + ) -> future::Either< + Box + Send + 'static>, + future::FutureResult<(), gst::FlowError>, + > { + let mut events = Vec::new(); + let mut state = self.state.lock().unwrap(); + if state.need_initial_events { + gst_debug!(self.cat, obj: element, "Pushing initial events"); + + let stream_id = format!("{:08x}{:08x}", rand::random::(), rand::random::()); + events.push(gst::Event::new_stream_start(&stream_id).build()); + if let Some(ref caps) = self.settings.lock().unwrap().caps { + events.push(gst::Event::new_caps(&caps).build()); + state.configured_caps = Some(caps.clone()); + } + events.push( + gst::Event::new_segment(&gst::FormattedSegment::::new()).build(), + ); + + if let Some(event) = Self::create_io_context_event(&state) { + events.push(event); + + // Get rid of reconfigure flag + self.src_pad.check_reconfigure(); + } + state.need_initial_events = false; + } else if self.src_pad.check_reconfigure() { + if let Some(event) = Self::create_io_context_event(&state) { + events.push(event); + } + } + + if buffer.get_size() == 0 { + events.push(gst::Event::new_eos().build()); + } + + drop(state); + + for event in events { + self.src_pad.push_event(event); + } + + let res = match self.src_pad.push(buffer).into_result() { + Ok(_) => { + gst_log!(self.cat, obj: element, "Successfully pushed buffer"); + Ok(()) + } + Err(gst::FlowError::Flushing) => { + gst_debug!(self.cat, obj: element, "Flushing"); + let state = self.state.lock().unwrap(); + if let Some(ref socket) = state.socket { + socket.pause(); + } + Ok(()) + } + Err(gst::FlowError::Eos) => { + gst_debug!(self.cat, obj: element, "EOS"); + let state = self.state.lock().unwrap(); + if let Some(ref socket) = state.socket { + socket.pause(); + } + Ok(()) + } + Err(err) => { + gst_error!(self.cat, obj: element, "Got error {}", err); + gst_element_error!( + element, + gst::StreamError::Failed, + ("Internal data stream error"), + ["streaming stopped, reason {}", err] + ); + Err(gst::FlowError::CustomError) + } + }; + + match res { + Ok(()) => { + let mut state = self.state.lock().unwrap(); + + if let State { + io_context: Some(ref io_context), + pending_future_id: Some(ref pending_future_id), + ref mut pending_future_cancel, + .. + } = *state + { + let (cancel, future) = io_context.drain_pending_futures(*pending_future_id); + *pending_future_cancel = cancel; + + future + } else { + future::Either::B(future::ok(())) + } + } + Err(err) => future::Either::B(future::err(err)), + } + } + + fn prepare(&self, element: &Element) -> Result<(), gst::ErrorMessage> { + use std::net::{IpAddr, SocketAddr}; + + gst_debug!(self.cat, obj: element, "Preparing"); + + let settings = self.settings.lock().unwrap().clone(); + + let mut state = self.state.lock().unwrap(); + + let io_context = IOContext::new( + &settings.context, + settings.context_threads as isize, + settings.context_wait, + ).map_err(|err| { + gst_error_msg!( + gst::ResourceError::OpenRead, + ["Failed to create IO context: {}", err] + ) + })?; + + let addr: IpAddr = match settings.address { + None => { + return Err(gst_error_msg!( + gst::ResourceError::Settings, + ["No address set"] + )) + } + Some(ref addr) => match addr.parse() { + Err(err) => { + return Err(gst_error_msg!( + gst::ResourceError::Settings, + ["Invalid address '{}' set: {}", addr, err] + )) + } + Ok(addr) => addr, + }, + }; + let port = settings.port; + + let saddr = SocketAddr::new(addr, port as u16); + gst_debug!(self.cat, obj: element, "Connecting to {:?}", saddr); + let socket = net::TcpStream::connect(&saddr); + + let buffer_pool = gst::BufferPool::new(); + let mut config = buffer_pool.get_config(); + config.set_params(None, settings.chunk_size, 0, 0); + buffer_pool.set_config(config).map_err(|_| { + gst_error_msg!( + gst::ResourceError::Settings, + ["Failed to configure buffer pool"] + ) + })?; + + let socket = Socket::new( + element.upcast_ref(), + TcpClientReader::new(socket), + buffer_pool, + ); + + let element_clone = element.clone(); + let element_clone2 = element.clone(); + socket + .schedule( + &io_context, + move |buffer| { + let tcpclientsrc = element_clone + .get_impl() + .downcast_ref::() + .unwrap(); + tcpclientsrc.push_buffer(&element_clone, buffer) + }, + move |err| { + let tcpclientsrc = element_clone2 + .get_impl() + .downcast_ref::() + .unwrap(); + gst_error!(tcpclientsrc.cat, obj: &element_clone2, "Got error {}", err); + match err { + Either::Left(gst::FlowError::CustomError) => (), + Either::Left(err) => { + gst_element_error!( + element_clone2, + gst::StreamError::Failed, + ("Internal data stream error"), + ["streaming stopped, reason {}", err] + ); + } + Either::Right(err) => { + gst_element_error!( + element_clone2, + gst::StreamError::Failed, + ("I/O error"), + ["streaming stopped, I/O error {}", err] + ); + } + } + }, + ) + .map_err(|_| { + gst_error_msg!(gst::ResourceError::OpenRead, ["Failed to schedule socket"]) + })?; + + let pending_future_id = io_context.acquire_pending_future_id(); + gst_debug!( + self.cat, + obj: element, + "Got pending future id {:?}", + pending_future_id + ); + + state.socket = Some(socket); + state.io_context = Some(io_context); + state.pending_future_id = Some(pending_future_id); + + gst_debug!(self.cat, obj: element, "Prepared"); + + Ok(()) + } + + fn unprepare(&self, element: &Element) -> Result<(), ()> { + gst_debug!(self.cat, obj: element, "Unpreparing"); + + // FIXME: The IO Context has to be alive longer than the queue, + // otherwise the queue can't finish any remaining work + let (mut socket, io_context) = { + let mut state = self.state.lock().unwrap(); + + if let (&Some(ref pending_future_id), &Some(ref io_context)) = + (&state.pending_future_id, &state.io_context) + { + io_context.release_pending_future_id(*pending_future_id); + } + + let socket = state.socket.take(); + let io_context = state.io_context.take(); + *state = State::default(); + (socket, io_context) + }; + + if let Some(ref socket) = socket.take() { + socket.shutdown(); + } + drop(io_context); + + gst_debug!(self.cat, obj: element, "Unprepared"); + Ok(()) + } + + fn start(&self, element: &Element) -> Result<(), ()> { + gst_debug!(self.cat, obj: element, "Starting"); + let state = self.state.lock().unwrap(); + + if let Some(ref socket) = state.socket { + socket.unpause(None, None); + } + + gst_debug!(self.cat, obj: element, "Started"); + + Ok(()) + } + + fn stop(&self, element: &Element) -> Result<(), ()> { + gst_debug!(self.cat, obj: element, "Stopping"); + let mut state = self.state.lock().unwrap(); + + if let Some(ref socket) = state.socket { + socket.pause(); + } + let _ = state.pending_future_cancel.take(); + + gst_debug!(self.cat, obj: element, "Stopped"); + + Ok(()) + } +} + +impl ObjectImpl for TcpClientSrc { + fn set_property(&self, _obj: &glib::Object, id: u32, value: &glib::Value) { + let prop = &PROPERTIES[id as usize]; + + match *prop { + Property::String("address", ..) => { + let mut settings = self.settings.lock().unwrap(); + settings.address = value.get(); + } + Property::UInt("port", ..) => { + let mut settings = self.settings.lock().unwrap(); + settings.port = value.get().unwrap(); + } + Property::Boxed("caps", ..) => { + let mut settings = self.settings.lock().unwrap(); + settings.caps = value.get(); + } + Property::UInt("chunk-size", ..) => { + let mut settings = self.settings.lock().unwrap(); + settings.chunk_size = value.get().unwrap(); + } + Property::String("context", ..) => { + let mut settings = self.settings.lock().unwrap(); + settings.context = value.get().unwrap_or_else(|| "".into()); + } + Property::Int("context-threads", ..) => { + let mut settings = self.settings.lock().unwrap(); + settings.context_threads = value.get().unwrap(); + } + Property::UInt("context-wait", ..) => { + let mut settings = self.settings.lock().unwrap(); + settings.context_wait = value.get().unwrap(); + } + _ => unimplemented!(), + } + } + + fn get_property(&self, _obj: &glib::Object, id: u32) -> Result { + let prop = &PROPERTIES[id as usize]; + + match *prop { + Property::String("address", ..) => { + let mut settings = self.settings.lock().unwrap(); + Ok(settings.address.to_value()) + } + Property::UInt("port", ..) => { + let mut settings = self.settings.lock().unwrap(); + Ok(settings.port.to_value()) + } + Property::Boxed("caps", ..) => { + let mut settings = self.settings.lock().unwrap(); + Ok(settings.caps.to_value()) + } + Property::UInt("chunk-size", ..) => { + let mut settings = self.settings.lock().unwrap(); + Ok(settings.chunk_size.to_value()) + } + Property::String("context", ..) => { + let mut settings = self.settings.lock().unwrap(); + Ok(settings.context.to_value()) + } + Property::Int("context-threads", ..) => { + let mut settings = self.settings.lock().unwrap(); + Ok(settings.context_threads.to_value()) + } + Property::UInt("context-wait", ..) => { + let mut settings = self.settings.lock().unwrap(); + Ok(settings.context_wait.to_value()) + } + _ => unimplemented!(), + } + } +} + +impl ElementImpl for TcpClientSrc { + fn change_state( + &self, + element: &Element, + transition: gst::StateChange, + ) -> gst::StateChangeReturn { + gst_trace!(self.cat, obj: element, "Changing state {:?}", transition); + + match transition { + gst::StateChange::NullToReady => match self.prepare(element) { + Err(err) => { + element.post_error_message(&err); + return gst::StateChangeReturn::Failure; + } + Ok(_) => match self.start(element) { + Err(_) => return gst::StateChangeReturn::Failure, + Ok(_) => (), + }, + }, + gst::StateChange::PlayingToPaused => match self.stop(element) { + Err(_) => return gst::StateChangeReturn::Failure, + Ok(_) => match self.unprepare(element) { + Err(_) => return gst::StateChangeReturn::Failure, + Ok(_) => (), + }, + }, + _ => (), + } + + let mut ret = element.parent_change_state(transition); + if ret == gst::StateChangeReturn::Failure { + return ret; + } + + match transition { + gst::StateChange::ReadyToPaused => { + ret = gst::StateChangeReturn::Success; + } + gst::StateChange::PausedToReady => { + let mut state = self.state.lock().unwrap(); + state.need_initial_events = true; + } + _ => (), + } + + ret + } +} + +struct TcpClientSrcStatic; + +impl ImplTypeStatic for TcpClientSrcStatic { + fn get_name(&self) -> &str { + "TcpClientSrc" + } + + fn new(&self, element: &Element) -> Box> { + TcpClientSrc::init(element) + } + + fn class_init(&self, klass: &mut ElementClass) { + TcpClientSrc::class_init(klass); + } +} + +pub fn register(plugin: &gst::Plugin) { + let tcpclientsrc_static = TcpClientSrcStatic; + let type_ = register_type(tcpclientsrc_static); + gst::Element::register(plugin, "ts-tcpclientsrc", 0, type_); +} diff --git a/gst-plugin-threadshare/tests/tcpclientsrc.rs b/gst-plugin-threadshare/tests/tcpclientsrc.rs new file mode 100644 index 00000000..8f0d19dd --- /dev/null +++ b/gst-plugin-threadshare/tests/tcpclientsrc.rs @@ -0,0 +1,149 @@ +// Copyright (C) 2018 Sebastian Dröge +// Copyright (C) 2018 LEE Dongjun +// +// This library is free software; you can redistribute it and/or +// modify it under the terms of the GNU Library General Public +// License as published by the Free Software Foundation; either +// version 2 of the License, or (at your option) any later version. +// +// This library is distributed in the hope that it will be useful, +// but WITHOUT ANY WARRANTY; without even the implied warranty of +// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU +// Library General Public License for more details. +// +// You should have received a copy of the GNU Library General Public +// License along with this library; if not, write to the +// Free Software Foundation, Inc., 51 Franklin Street, Suite 500, +// Boston, MA 02110-1335, USA. + +extern crate glib; +use glib::prelude::*; + +extern crate gstreamer as gst; +use gst::prelude::*; + +use std::io::Write; +use std::sync::{Arc, Mutex}; +use std::{thread, time}; + +fn init() { + use std::sync::{Once, ONCE_INIT}; + static INIT: Once = ONCE_INIT; + + INIT.call_once(|| { + gst::init().unwrap(); + + #[cfg(debug_assertions)] + { + use std::path::Path; + + let mut path = Path::new("target/debug"); + if !path.exists() { + path = Path::new("../target/debug"); + } + + gst::Registry::get().scan_path(path); + } + #[cfg(not(debug_assertions))] + { + use std::path::Path; + + let mut path = Path::new("target/release"); + if !path.exists() { + path = Path::new("../target/release"); + } + + gst::Registry::get().scan_path(path); + } + }); +} + +#[test] +fn test_push() { + init(); + + let handler = thread::spawn(move || { + use std::net; + + let listener = net::TcpListener::bind("0.0.0.0:5000").unwrap(); + for stream in listener.incoming() { + let buffer = [0; 160]; + let mut socket = stream.unwrap(); + for _ in 0..3 { + let _ = socket.write(&buffer); + thread::sleep(time::Duration::from_millis(20)); + } + break; + } + }); + + let pipeline = gst::Pipeline::new(None); + + let tcpclientsrc = gst::ElementFactory::make("ts-tcpclientsrc", None).unwrap(); + let appsink = gst::ElementFactory::make("appsink", None).unwrap(); + appsink.set_property("sync", &false).unwrap(); + appsink.set_property("async", &false).unwrap(); + + pipeline.add_many(&[&tcpclientsrc, &appsink]).unwrap(); + tcpclientsrc.link(&appsink).unwrap(); + + let caps = gst::Caps::new_simple("foo/bar", &[]); + tcpclientsrc.set_property("caps", &caps).unwrap(); + tcpclientsrc.set_property("port", &(5000u32)).unwrap(); + + appsink.set_property("emit-signals", &true).unwrap(); + + let samples = Arc::new(Mutex::new(Vec::new())); + + let samples_clone = samples.clone(); + appsink + .connect("new-sample", true, move |args| { + let appsink = args[0].get::().unwrap(); + + let sample = appsink + .emit("pull-sample", &[]) + .unwrap() + .unwrap() + .get::() + .unwrap(); + + let mut samples = samples_clone.lock().unwrap(); + samples.push(sample); + Some(gst::FlowReturn::Ok.to_value()) + }) + .unwrap(); + + pipeline + .set_state(gst::State::Playing) + .into_result() + .unwrap(); + + let mut eos = false; + let bus = pipeline.get_bus().unwrap(); + while let Some(msg) = bus.timed_pop(5 * gst::SECOND) { + use gst::MessageView; + match msg.view() { + MessageView::Eos(..) => { + eos = true; + break; + } + MessageView::Error(..) => unreachable!(), + _ => (), + } + } + + assert!(eos); + let samples = samples.lock().unwrap(); + for sample in samples.iter() { + assert_eq!(Some(&caps), sample.get_caps().as_ref()); + } + + let total_received_size = samples.iter().fold(0, |acc, sample| { + acc + sample.get_buffer().unwrap().get_size() + }); + assert_eq!(total_received_size, 3 * 160); + + pipeline.set_state(gst::State::Null).into_result().unwrap(); + + handler.join().unwrap(); +}