gst-plugins-rs/generic/threadshare/src/udpsink/imp.rs

1446 lines
48 KiB
Rust
Raw Normal View History

// Copyright (C) 2019 Mathieu Duponchelle <mathieu@centricular.com>
//
// 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 futures::channel::mpsc;
use futures::future::BoxFuture;
use futures::lock::Mutex;
use futures::prelude::*;
use glib::glib_object_subclass;
use glib::prelude::*;
use glib::subclass;
use glib::subclass::prelude::*;
use gst::prelude::*;
use gst::subclass::prelude::*;
use gst::EventView;
use gst::{
gst_debug, gst_element_error, gst_error, gst_error_msg, gst_info, gst_log, gst_trace,
gst_warning,
};
use lazy_static::lazy_static;
use crate::runtime::prelude::*;
2020-03-26 18:07:25 +00:00
use crate::runtime::{self, Context, PadSink, PadSinkRef, Task};
use crate::socket::{wrap_socket, GioSocketWrapper};
2020-03-26 18:07:25 +00:00
use std::convert::TryInto;
use std::mem;
use std::net::{IpAddr, Ipv4Addr, SocketAddr};
2020-03-26 18:07:25 +00:00
use std::string::ToString;
use std::sync::Mutex as StdMutex;
2020-03-26 18:07:25 +00:00
use std::sync::{Arc, RwLock};
use std::time::Duration;
use std::u16;
use std::u8;
const DEFAULT_HOST: Option<&str> = Some("127.0.0.1");
const DEFAULT_PORT: i32 = 5004;
const DEFAULT_SYNC: bool = true;
const DEFAULT_BIND_ADDRESS: &str = "0.0.0.0";
const DEFAULT_BIND_PORT: i32 = 0;
const DEFAULT_BIND_ADDRESS_V6: &str = "::";
const DEFAULT_BIND_PORT_V6: i32 = 0;
const DEFAULT_SOCKET: Option<GioSocketWrapper> = None;
const DEFAULT_USED_SOCKET: Option<GioSocketWrapper> = None;
const DEFAULT_SOCKET_V6: Option<GioSocketWrapper> = None;
const DEFAULT_USED_SOCKET_V6: Option<GioSocketWrapper> = None;
const DEFAULT_AUTO_MULTICAST: bool = true;
const DEFAULT_LOOP: bool = true;
const DEFAULT_TTL: u32 = 64;
const DEFAULT_TTL_MC: u32 = 1;
const DEFAULT_QOS_DSCP: i32 = -1;
const DEFAULT_CLIENTS: &str = "";
const DEFAULT_CONTEXT: &str = "";
const DEFAULT_CONTEXT_WAIT: u32 = 0;
#[derive(Debug, Clone)]
struct Settings {
sync: bool,
bind_address: String,
bind_port: i32,
bind_address_v6: String,
bind_port_v6: i32,
socket: Option<GioSocketWrapper>,
used_socket: Option<GioSocketWrapper>,
socket_v6: Option<GioSocketWrapper>,
used_socket_v6: Option<GioSocketWrapper>,
auto_multicast: bool,
multicast_loop: bool,
ttl: u32,
ttl_mc: u32,
qos_dscp: i32,
context: String,
context_wait: u32,
}
impl Default for Settings {
fn default() -> Self {
Settings {
sync: DEFAULT_SYNC,
bind_address: DEFAULT_BIND_ADDRESS.into(),
bind_port: DEFAULT_BIND_PORT,
bind_address_v6: DEFAULT_BIND_ADDRESS_V6.into(),
bind_port_v6: DEFAULT_BIND_PORT_V6,
socket: DEFAULT_SOCKET,
used_socket: DEFAULT_USED_SOCKET,
socket_v6: DEFAULT_SOCKET_V6,
used_socket_v6: DEFAULT_USED_SOCKET_V6,
auto_multicast: DEFAULT_AUTO_MULTICAST,
multicast_loop: DEFAULT_LOOP,
ttl: DEFAULT_TTL,
ttl_mc: DEFAULT_TTL_MC,
qos_dscp: DEFAULT_QOS_DSCP,
context: DEFAULT_CONTEXT.into(),
context_wait: DEFAULT_CONTEXT_WAIT,
}
}
}
lazy_static! {
static ref CAT: gst::DebugCategory = gst::DebugCategory::new(
"ts-udpsink",
gst::DebugColorFlags::empty(),
Some("Thread-sharing UDP sink"),
);
}
static PROPERTIES: [subclass::Property; 17] = [
subclass::Property("sync", |name| {
glib::ParamSpec::boolean(
name,
"Sync",
"Sync on the clock",
DEFAULT_SYNC,
glib::ParamFlags::READWRITE,
)
}),
subclass::Property("bind-address", |name| {
glib::ParamSpec::string(
name,
"Bind Address",
"Address to bind the socket to",
Some(DEFAULT_BIND_ADDRESS),
glib::ParamFlags::READWRITE,
)
}),
subclass::Property("bind-port", |name| {
glib::ParamSpec::int(
name,
"Bind Port",
"Port to bind the socket to",
0,
u16::MAX as i32,
DEFAULT_BIND_PORT,
glib::ParamFlags::READWRITE,
)
}),
subclass::Property("bind-address-v6", |name| {
glib::ParamSpec::string(
name,
"Bind Address V6",
"Address to bind the V6 socket to",
Some(DEFAULT_BIND_ADDRESS_V6),
glib::ParamFlags::READWRITE,
)
}),
subclass::Property("bind-port-v6", |name| {
glib::ParamSpec::int(
name,
"Bind Port",
"Port to bind the V6 socket to",
0,
u16::MAX as i32,
DEFAULT_BIND_PORT_V6,
glib::ParamFlags::READWRITE,
)
}),
subclass::Property("socket", |name| {
glib::ParamSpec::object(
name,
"Socket",
"Socket to use for UDP transmission. (None == allocate)",
gio::Socket::static_type(),
glib::ParamFlags::READWRITE,
)
}),
subclass::Property("used-socket", |name| {
glib::ParamSpec::object(
name,
"Used Socket",
"Socket currently in use for UDP transmission. (None = no socket)",
gio::Socket::static_type(),
glib::ParamFlags::READABLE,
)
}),
subclass::Property("socket-v6", |name| {
glib::ParamSpec::object(
name,
"Socket V6",
"IPV6 Socket to use for UDP transmission. (None == allocate)",
gio::Socket::static_type(),
glib::ParamFlags::READWRITE,
)
}),
subclass::Property("used-socket-v6", |name| {
glib::ParamSpec::object(
name,
"Used Socket V6",
"V6 Socket currently in use for UDP transmission. (None = no socket)",
gio::Socket::static_type(),
glib::ParamFlags::READABLE,
)
}),
subclass::Property("auto-multicast", |name| {
glib::ParamSpec::boolean(
name,
"Auto multicast",
"Automatically join/leave the multicast groups, FALSE means user has to do it himself",
DEFAULT_AUTO_MULTICAST,
glib::ParamFlags::READWRITE,
)
}),
subclass::Property("loop", |name| {
glib::ParamSpec::boolean(
name,
"Loop",
"Set the multicast loop parameter.",
DEFAULT_LOOP,
glib::ParamFlags::READWRITE,
)
}),
subclass::Property("ttl", |name| {
glib::ParamSpec::uint(
name,
"Time To Live",
"Used for setting the unicast TTL parameter",
0,
u8::MAX as u32,
DEFAULT_TTL,
glib::ParamFlags::READWRITE,
)
}),
subclass::Property("ttl-mc", |name| {
glib::ParamSpec::uint(
name,
"Time To Live Multicast",
"Used for setting the multicast TTL parameter",
0,
u8::MAX as u32,
DEFAULT_TTL_MC,
glib::ParamFlags::READWRITE,
)
}),
subclass::Property("qos-dscp", |name| {
glib::ParamSpec::int(
name,
"QoS DSCP",
"Quality of Service, differentiated services code point (-1 default)",
-1,
63,
DEFAULT_QOS_DSCP,
glib::ParamFlags::READWRITE,
)
}),
subclass::Property("clients", |name| {
glib::ParamSpec::string(
name,
"Clients",
"A comma separated list of host:port pairs with destinations",
Some(DEFAULT_CLIENTS),
glib::ParamFlags::READWRITE,
)
}),
subclass::Property("context", |name| {
glib::ParamSpec::string(
name,
"Context",
"Context name to share threads with",
Some(DEFAULT_CONTEXT),
glib::ParamFlags::READWRITE,
)
}),
subclass::Property("context-wait", |name| {
glib::ParamSpec::uint(
name,
"Context Wait",
"Throttle poll loop to run at most once every this many ms",
0,
1000,
DEFAULT_CONTEXT_WAIT,
glib::ParamFlags::READWRITE,
)
}),
];
#[derive(Debug)]
2020-03-26 18:07:25 +00:00
enum TaskItem {
Buffer(gst::Buffer),
Event(gst::Event),
}
#[derive(Debug)]
struct UdpSinkPadHandlerInner {
sync: bool,
segment: Option<gst::Segment>,
latency: gst::ClockTime,
socket: Arc<Mutex<Option<tokio::net::UdpSocket>>>,
socket_v6: Arc<Mutex<Option<tokio::net::UdpSocket>>>,
clients: Arc<Vec<SocketAddr>>,
clients_to_configure: Vec<SocketAddr>,
clients_to_unconfigure: Vec<SocketAddr>,
sender: Arc<Mutex<Option<mpsc::Sender<TaskItem>>>>,
settings: Arc<StdMutex<Settings>>,
}
2020-03-26 18:07:25 +00:00
impl UdpSinkPadHandlerInner {
fn new(settings: Arc<StdMutex<Settings>>) -> Self {
UdpSinkPadHandlerInner {
sync: DEFAULT_SYNC,
segment: None,
latency: gst::CLOCK_TIME_NONE,
socket: Arc::new(Mutex::new(None)),
socket_v6: Arc::new(Mutex::new(None)),
clients: Arc::new(vec![SocketAddr::new(
DEFAULT_HOST.unwrap().parse().unwrap(),
DEFAULT_PORT as u16,
)]),
clients_to_configure: vec![],
clients_to_unconfigure: vec![],
sender: Arc::new(Mutex::new(None)),
2020-03-15 11:40:45 +00:00
settings,
2020-03-26 18:07:25 +00:00
}
}
fn clear_clients(
&mut self,
gst_pad: &gst::Pad,
clients_to_add: impl Iterator<Item = SocketAddr>,
) {
let old_clients = mem::replace(&mut *Arc::make_mut(&mut self.clients), vec![]);
2020-03-26 18:07:25 +00:00
self.clients_to_configure = vec![];
self.clients_to_unconfigure = vec![];
for addr in clients_to_add {
if !old_clients.contains(&addr) {
self.clients_to_unconfigure.push(addr);
}
2020-03-26 18:07:25 +00:00
self.add_client(gst_pad, addr);
}
}
fn remove_client(&mut self, gst_pad: &gst::Pad, addr: SocketAddr) {
if !self.clients.contains(&addr) {
gst_warning!(CAT, obj: gst_pad, "Not removing unknown client {:?}", &addr);
return;
}
gst_info!(CAT, obj: gst_pad, "Removing client {:?}", addr);
Arc::make_mut(&mut self.clients).retain(|addr2| addr != *addr2);
self.clients_to_unconfigure.push(addr);
self.clients_to_configure.retain(|addr2| addr != *addr2);
}
fn add_client(&mut self, gst_pad: &gst::Pad, addr: SocketAddr) {
if self.clients.contains(&addr) {
gst_warning!(CAT, obj: gst_pad, "Not adding client {:?} again", &addr);
return;
}
gst_info!(CAT, obj: gst_pad, "Adding client {:?}", addr);
Arc::make_mut(&mut self.clients).push(addr);
self.clients_to_configure.push(addr);
self.clients_to_unconfigure.retain(|addr2| addr != *addr2);
}
}
#[derive(Debug)]
enum SocketQualified {
Ipv4(tokio::net::UdpSocket),
Ipv6(tokio::net::UdpSocket),
}
#[derive(Clone, Debug)]
struct UdpSinkPadHandler(Arc<RwLock<UdpSinkPadHandlerInner>>);
impl UdpSinkPadHandler {
fn new(settings: Arc<StdMutex<Settings>>) -> UdpSinkPadHandler {
Self(Arc::new(RwLock::new(UdpSinkPadHandlerInner::new(settings))))
}
fn set_latency(&self, latency: gst::ClockTime) {
self.0.write().unwrap().latency = latency;
}
fn prepare(&self) {
let mut inner = self.0.write().unwrap();
inner.clients_to_configure = inner.clients.to_vec();
}
fn prepare_socket(&self, socket: SocketQualified) {
let mut inner = self.0.write().unwrap();
match socket {
SocketQualified::Ipv4(socket) => inner.socket = Arc::new(Mutex::new(Some(socket))),
SocketQualified::Ipv6(socket) => inner.socket_v6 = Arc::new(Mutex::new(Some(socket))),
}
}
fn unprepare(&self) {
let mut inner = self.0.write().unwrap();
*inner = UdpSinkPadHandlerInner::new(Arc::clone(&inner.settings))
}
fn clear_clients(&self, gst_pad: &gst::Pad, clients_to_add: impl Iterator<Item = SocketAddr>) {
self.0
.write()
.unwrap()
.clear_clients(gst_pad, clients_to_add);
}
fn remove_client(&self, gst_pad: &gst::Pad, addr: SocketAddr) {
self.0.write().unwrap().remove_client(gst_pad, addr);
}
fn add_client(&self, gst_pad: &gst::Pad, addr: SocketAddr) {
self.0.write().unwrap().add_client(gst_pad, addr);
}
fn get_clients(&self) -> Vec<SocketAddr> {
(*self.0.read().unwrap().clients).clone()
}
fn configure_client(
&self,
settings: &Settings,
socket: &mut Option<tokio::net::UdpSocket>,
socket_v6: &mut Option<tokio::net::UdpSocket>,
client: &SocketAddr,
) -> Result<(), gst::ErrorMessage> {
if client.ip().is_multicast() {
match client.ip() {
IpAddr::V4(addr) => {
if let Some(socket) = socket.as_mut() {
if settings.auto_multicast {
socket
.join_multicast_v4(addr, Ipv4Addr::new(0, 0, 0, 0))
.map_err(|err| {
gst_error_msg!(
gst::ResourceError::OpenWrite,
["Failed to join multicast group: {}", err]
)
})?;
}
if settings.multicast_loop {
socket.set_multicast_loop_v4(true).map_err(|err| {
gst_error_msg!(
gst::ResourceError::OpenWrite,
["Failed to set multicast loop: {}", err]
)
})?;
}
socket
.set_multicast_ttl_v4(settings.ttl_mc)
.map_err(|err| {
gst_error_msg!(
gst::ResourceError::OpenWrite,
["Failed to set multicast ttl: {}", err]
)
})?;
}
}
IpAddr::V6(addr) => {
if let Some(socket) = socket_v6.as_mut() {
if settings.auto_multicast {
socket.join_multicast_v6(&addr, 0).map_err(|err| {
gst_error_msg!(
gst::ResourceError::OpenWrite,
["Failed to join multicast group: {}", err]
)
})?;
}
if settings.multicast_loop {
socket.set_multicast_loop_v6(true).map_err(|err| {
gst_error_msg!(
gst::ResourceError::OpenWrite,
["Failed to set multicast loop: {}", err]
)
})?;
}
/* FIXME no API for set_multicast_ttl_v6 ? */
}
}
}
} else {
match client.ip() {
IpAddr::V4(_) => {
if let Some(socket) = socket.as_mut() {
socket.set_ttl(settings.ttl).map_err(|err| {
gst_error_msg!(
gst::ResourceError::OpenWrite,
["Failed to set unicast ttl: {}", err]
)
})?;
}
}
IpAddr::V6(_) => {
if let Some(socket) = socket_v6.as_mut() {
socket.set_ttl(settings.ttl).map_err(|err| {
gst_error_msg!(
gst::ResourceError::OpenWrite,
["Failed to set unicast ttl: {}", err]
)
})?;
}
}
}
}
Ok(())
}
fn unconfigure_client(
&self,
settings: &Settings,
socket: &mut Option<tokio::net::UdpSocket>,
socket_v6: &mut Option<tokio::net::UdpSocket>,
client: &SocketAddr,
) -> Result<(), gst::ErrorMessage> {
if client.ip().is_multicast() {
match client.ip() {
IpAddr::V4(addr) => {
if let Some(socket) = socket.as_mut() {
if settings.auto_multicast {
socket
.leave_multicast_v4(addr, Ipv4Addr::new(0, 0, 0, 0))
.map_err(|err| {
gst_error_msg!(
gst::ResourceError::OpenWrite,
["Failed to join multicast group: {}", err]
)
})?;
}
}
}
IpAddr::V6(addr) => {
if let Some(socket) = socket_v6.as_mut() {
if settings.auto_multicast {
socket.leave_multicast_v6(&addr, 0).map_err(|err| {
gst_error_msg!(
gst::ResourceError::OpenWrite,
["Failed to join multicast group: {}", err]
)
})?;
}
}
}
}
}
Ok(())
}
async fn render(
&self,
element: &super::UdpSink,
buffer: gst::Buffer,
) -> Result<gst::FlowSuccess, gst::FlowError> {
let (
do_sync,
rtime,
clients,
clients_to_configure,
clients_to_unconfigure,
socket,
socket_v6,
settings,
) = {
2020-03-26 18:07:25 +00:00
let mut inner = self.0.write().unwrap();
let do_sync = inner.sync;
let mut rtime: gst::ClockTime = 0.into();
2020-03-26 18:07:25 +00:00
if let Some(segment) = &inner.segment {
if let Some(segment) = segment.downcast_ref::<gst::format::Time>() {
rtime = segment.to_running_time(buffer.get_pts());
2020-03-26 18:07:25 +00:00
if inner.latency.is_some() {
rtime += inner.latency;
}
}
}
2020-03-26 18:07:25 +00:00
let clients_to_configure = mem::replace(&mut inner.clients_to_configure, vec![]);
let clients_to_unconfigure = mem::replace(&mut inner.clients_to_unconfigure, vec![]);
2020-03-26 18:07:25 +00:00
let settings = inner.settings.lock().unwrap().clone();
(
do_sync,
rtime,
2020-03-26 18:07:25 +00:00
Arc::clone(&inner.clients),
clients_to_configure,
clients_to_unconfigure,
2020-03-26 18:07:25 +00:00
Arc::clone(&inner.socket),
Arc::clone(&inner.socket_v6),
settings,
)
};
let mut socket = socket.lock().await;
let mut socket_v6 = socket_v6.lock().await;
if !clients_to_configure.is_empty() {
for client in &clients_to_configure {
self.configure_client(&settings, &mut socket, &mut socket_v6, &client)
.map_err(|err| {
gst_element_error!(
element,
gst::StreamError::Failed,
["Failed to configure client {:?}: {}", client, err]
);
gst::FlowError::Error
})?;
}
}
if !clients_to_unconfigure.is_empty() {
for client in &clients_to_unconfigure {
self.unconfigure_client(&settings, &mut socket, &mut socket_v6, &client)
.map_err(|err| {
gst_element_error!(
element,
gst::StreamError::Failed,
["Failed to unconfigure client {:?}: {}", client, err]
);
gst::FlowError::Error
})?;
}
}
if do_sync {
self.sync(&element, rtime).await;
}
let data = buffer.map_readable().map_err(|_| {
gst_element_error!(
element,
gst::StreamError::Format,
["Failed to map buffer readable"]
);
gst::FlowError::Error
})?;
for client in clients.iter() {
let socket = match client.ip() {
IpAddr::V4(_) => &mut socket,
IpAddr::V6(_) => &mut socket_v6,
};
if let Some(socket) = socket.as_mut() {
gst_log!(CAT, obj: element, "Sending to {:?}", &client);
socket.send_to(&data, client).await.map_err(|err| {
gst_element_error!(
element,
gst::StreamError::Failed,
("I/O error"),
["streaming stopped, I/O error {}", err]
);
gst::FlowError::Error
})?;
} else {
gst_element_error!(
element,
gst::StreamError::Failed,
("I/O error"),
["No socket available for sending to {}", client]
);
return Err(gst::FlowError::Error);
}
}
gst_log!(
CAT,
obj: element,
"Sent buffer {:?} to all clients",
&buffer
);
Ok(gst::FlowSuccess::Ok)
}
/* Wait until specified time */
async fn sync(&self, element: &super::UdpSink, running_time: gst::ClockTime) {
let now = element.get_current_running_time();
if let Some(delay) = running_time
.saturating_sub(now)
.and_then(|delay| delay.nseconds())
{
runtime::time::delay_for(Duration::from_nanos(delay)).await;
}
}
async fn handle_event(&self, element: &super::UdpSink, event: gst::Event) {
match event.view() {
EventView::Eos(_) => {
let _ = element.post_message(gst::message::Eos::builder().src(element).build());
}
EventView::Segment(e) => {
2020-03-26 18:07:25 +00:00
self.0.write().unwrap().segment = Some(e.get_segment().clone());
}
_ => (),
}
}
}
impl PadSinkHandler for UdpSinkPadHandler {
type ElementImpl = UdpSink;
fn sink_chain(
&self,
_pad: &PadSinkRef,
_udpsink: &UdpSink,
element: &gst::Element,
buffer: gst::Buffer,
) -> BoxFuture<'static, Result<gst::FlowSuccess, gst::FlowError>> {
let sender = Arc::clone(&self.0.read().unwrap().sender);
let element = element.clone().downcast::<super::UdpSink>().unwrap();
async move {
if let Some(sender) = sender.lock().await.as_mut() {
if sender.send(TaskItem::Buffer(buffer)).await.is_err() {
gst_debug!(CAT, obj: &element, "Flushing");
return Err(gst::FlowError::Flushing);
}
}
Ok(gst::FlowSuccess::Ok)
}
.boxed()
}
fn sink_chain_list(
&self,
_pad: &PadSinkRef,
_udpsink: &UdpSink,
element: &gst::Element,
list: gst::BufferList,
) -> BoxFuture<'static, Result<gst::FlowSuccess, gst::FlowError>> {
let sender = Arc::clone(&self.0.read().unwrap().sender);
let element = element.clone().downcast::<super::UdpSink>().unwrap();
async move {
if let Some(sender) = sender.lock().await.as_mut() {
for buffer in list.iter_owned() {
if sender.send(TaskItem::Buffer(buffer)).await.is_err() {
gst_debug!(CAT, obj: &element, "Flushing");
return Err(gst::FlowError::Flushing);
}
}
}
Ok(gst::FlowSuccess::Ok)
}
.boxed()
}
fn sink_event_serialized(
&self,
_pad: &PadSinkRef,
_udpsink: &UdpSink,
element: &gst::Element,
event: gst::Event,
) -> BoxFuture<'static, bool> {
let sender = Arc::clone(&self.0.read().unwrap().sender);
let element = element.clone().downcast::<super::UdpSink>().unwrap();
async move {
if let EventView::FlushStop(_) = event.view() {
2020-03-26 18:07:25 +00:00
let udpsink = UdpSink::from_instance(&element);
return udpsink.task.flush_stop().is_ok();
2020-03-15 11:40:45 +00:00
} else if let Some(sender) = sender.lock().await.as_mut() {
if sender.send(TaskItem::Event(event)).await.is_err() {
gst_debug!(CAT, obj: &element, "Flushing");
}
}
2020-03-26 18:07:25 +00:00
true
}
.boxed()
}
fn sink_event(
&self,
_pad: &PadSinkRef,
2020-03-26 18:07:25 +00:00
udpsink: &UdpSink,
_element: &gst::Element,
event: gst::Event,
) -> bool {
2020-03-26 18:07:25 +00:00
if let EventView::FlushStart(..) = event.view() {
return udpsink.task.flush_start().is_ok();
}
true
}
}
#[derive(Debug)]
struct UdpSinkTask {
element: super::UdpSink,
sink_pad_handler: UdpSinkPadHandler,
receiver: Option<mpsc::Receiver<TaskItem>>,
}
impl UdpSinkTask {
fn new(element: &super::UdpSink, sink_pad_handler: &UdpSinkPadHandler) -> Self {
UdpSinkTask {
element: element.clone(),
sink_pad_handler: sink_pad_handler.clone(),
receiver: None,
}
}
}
impl TaskImpl for UdpSinkTask {
fn start(&mut self) -> BoxFuture<'_, Result<(), gst::ErrorMessage>> {
async move {
gst_log!(CAT, obj: &self.element, "Starting task");
let (sender, receiver) = mpsc::channel(0);
let mut sink_pad_handler = self.sink_pad_handler.0.write().unwrap();
sink_pad_handler.sender = Arc::new(Mutex::new(Some(sender)));
self.receiver = Some(receiver);
gst_log!(CAT, obj: &self.element, "Task started");
Ok(())
}
.boxed()
}
fn iterate(&mut self) -> BoxFuture<'_, Result<(), gst::FlowError>> {
async move {
match self.receiver.as_mut().unwrap().next().await {
Some(TaskItem::Buffer(buffer)) => {
match self.sink_pad_handler.render(&self.element, buffer).await {
Err(err) => {
gst_element_error!(
&self.element,
gst::StreamError::Failed,
["Failed to render item, stopping task: {}", err]
);
Err(gst::FlowError::Error)
}
_ => Ok(()),
}
}
Some(TaskItem::Event(event)) => {
self.sink_pad_handler
.handle_event(&self.element, event)
.await;
Ok(())
}
None => Err(gst::FlowError::Flushing),
}
}
.boxed()
}
}
2020-03-26 18:07:25 +00:00
#[derive(Debug)]
enum SocketFamily {
Ipv4,
Ipv6,
}
#[derive(Debug)]
pub struct UdpSink {
2020-03-26 18:07:25 +00:00
sink_pad: PadSink,
sink_pad_handler: UdpSinkPadHandler,
task: Task,
settings: Arc<StdMutex<Settings>>,
}
impl UdpSink {
2020-03-26 18:07:25 +00:00
fn prepare_socket(
&self,
2020-03-26 18:07:25 +00:00
family: SocketFamily,
context: &Context,
element: &super::UdpSink,
) -> Result<(), gst::ErrorMessage> {
let mut settings = self.settings.lock().unwrap();
2020-03-26 18:07:25 +00:00
let wrapped_socket = match family {
SocketFamily::Ipv4 => &settings.socket,
SocketFamily::Ipv6 => &settings.socket_v6,
};
let socket_qualified: SocketQualified;
if let Some(ref wrapped_socket) = wrapped_socket {
use std::net::UdpSocket;
let socket: UdpSocket;
#[cfg(unix)]
{
socket = wrapped_socket.get()
}
#[cfg(windows)]
{
socket = wrapped_socket.get()
}
let socket = context.enter(|| {
tokio::net::UdpSocket::from_std(socket).map_err(|err| {
gst_error_msg!(
gst::ResourceError::OpenWrite,
["Failed to setup socket for tokio: {}", err]
)
})
})?;
2020-03-26 18:07:25 +00:00
match family {
SocketFamily::Ipv4 => {
settings.used_socket = Some(wrapped_socket.clone());
socket_qualified = SocketQualified::Ipv4(socket);
}
SocketFamily::Ipv6 => {
settings.used_socket_v6 = Some(wrapped_socket.clone());
socket_qualified = SocketQualified::Ipv6(socket);
}
}
} else {
2020-03-26 18:07:25 +00:00
let bind_addr = match family {
SocketFamily::Ipv4 => &settings.bind_address,
SocketFamily::Ipv6 => &settings.bind_address_v6,
};
let bind_addr: IpAddr = bind_addr.parse().map_err(|err| {
gst_error_msg!(
gst::ResourceError::Settings,
["Invalid address '{}' set: {}", bind_addr, err]
)
})?;
2020-03-26 18:07:25 +00:00
let bind_port = match family {
SocketFamily::Ipv4 => settings.bind_port,
SocketFamily::Ipv6 => settings.bind_port_v6,
};
let saddr = SocketAddr::new(bind_addr, bind_port as u16);
gst_debug!(CAT, obj: element, "Binding to {:?}", saddr);
let socket = match family {
SocketFamily::Ipv4 => socket2::Socket::new(
socket2::Domain::ipv4(),
socket2::Type::dgram(),
Some(socket2::Protocol::udp()),
),
SocketFamily::Ipv6 => socket2::Socket::new(
socket2::Domain::ipv6(),
socket2::Type::dgram(),
Some(socket2::Protocol::udp()),
),
};
let socket = match socket {
Ok(socket) => socket,
Err(err) => {
gst_warning!(
CAT,
obj: element,
"Failed to create {} socket: {}",
2020-03-26 18:07:25 +00:00
match family {
SocketFamily::Ipv4 => "IPv4",
SocketFamily::Ipv6 => "IPv6",
},
err
);
return Ok(());
}
};
socket.bind(&saddr.into()).map_err(|err| {
gst_error_msg!(
gst::ResourceError::OpenWrite,
["Failed to bind socket: {}", err]
)
})?;
let socket = context.enter(|| {
tokio::net::UdpSocket::from_std(socket.into()).map_err(|err| {
gst_error_msg!(
gst::ResourceError::OpenWrite,
["Failed to setup socket for tokio: {}", err]
)
})
})?;
let wrapper = wrap_socket(&socket)?;
if settings.qos_dscp != -1 {
wrapper.set_tos(settings.qos_dscp).map_err(|err| {
gst_error_msg!(
gst::ResourceError::OpenWrite,
["Failed to set QoS DSCP: {}", err]
)
})?;
}
2020-03-26 18:07:25 +00:00
match family {
SocketFamily::Ipv4 => {
settings.used_socket = Some(wrapper);
socket_qualified = SocketQualified::Ipv4(socket)
}
SocketFamily::Ipv6 => {
settings.used_socket_v6 = Some(wrapper);
socket_qualified = SocketQualified::Ipv6(socket)
}
}
}
2020-03-26 18:07:25 +00:00
self.sink_pad_handler.prepare_socket(socket_qualified);
Ok(())
}
fn prepare(&self, element: &super::UdpSink) -> Result<(), gst::ErrorMessage> {
gst_debug!(CAT, obj: element, "Preparing");
let context = {
let settings = self.settings.lock().unwrap();
Context::acquire(&settings.context, settings.context_wait).map_err(|err| {
gst_error_msg!(
gst::ResourceError::OpenWrite,
["Failed to acquire Context: {}", err]
)
})?
};
2020-03-26 18:07:25 +00:00
self.sink_pad_handler.prepare();
self.prepare_socket(SocketFamily::Ipv4, &context, element)?;
self.prepare_socket(SocketFamily::Ipv6, &context, element)?;
self.task
.prepare(UdpSinkTask::new(&element, &self.sink_pad_handler), context)
.map_err(|err| {
gst_error_msg!(
gst::ResourceError::OpenRead,
["Error preparing Task: {:?}", err]
)
})?;
gst_debug!(CAT, obj: element, "Started preparing");
Ok(())
}
fn unprepare(&self, element: &super::UdpSink) {
gst_debug!(CAT, obj: element, "Unpreparing");
2020-03-26 18:07:25 +00:00
self.task.unprepare().unwrap();
self.sink_pad_handler.unprepare();
gst_debug!(CAT, obj: element, "Unprepared");
}
fn stop(&self, element: &super::UdpSink) -> Result<(), gst::ErrorMessage> {
gst_debug!(CAT, obj: element, "Stopping");
self.task.stop()?;
gst_debug!(CAT, obj: element, "Stopped");
Ok(())
}
fn start(&self, element: &super::UdpSink) -> Result<(), gst::ErrorMessage> {
2020-03-26 18:07:25 +00:00
gst_debug!(CAT, obj: element, "Starting");
self.task.start()?;
gst_debug!(CAT, obj: element, "Started");
Ok(())
2020-03-26 18:07:25 +00:00
}
}
2020-03-26 18:07:25 +00:00
impl UdpSink {
fn clear_clients(&self, clients_to_add: impl Iterator<Item = SocketAddr>) {
self.sink_pad_handler
.clear_clients(&self.sink_pad.gst_pad(), clients_to_add);
}
2020-03-26 18:07:25 +00:00
fn remove_client(&self, addr: SocketAddr) {
self.sink_pad_handler
.remove_client(&self.sink_pad.gst_pad(), addr);
}
2020-03-26 18:07:25 +00:00
fn add_client(&self, addr: SocketAddr) {
self.sink_pad_handler
.add_client(&self.sink_pad.gst_pad(), addr);
}
}
fn try_into_socket_addr(element: &super::UdpSink, host: &str, port: i32) -> Result<SocketAddr, ()> {
2020-03-26 18:07:25 +00:00
let addr: IpAddr = match host.parse() {
Err(err) => {
gst_error!(CAT, obj: element, "Failed to parse host {}: {}", host, err);
return Err(());
}
2020-03-26 18:07:25 +00:00
Ok(addr) => addr,
};
2020-03-26 18:07:25 +00:00
let port: u16 = match port.try_into() {
Err(err) => {
gst_error!(CAT, obj: element, "Invalid port {}: {}", port, err);
return Err(());
}
Ok(port) => port,
};
2020-03-26 18:07:25 +00:00
Ok(SocketAddr::new(addr, port))
}
impl ObjectSubclass for UdpSink {
const NAME: &'static str = "RsTsUdpSink";
type Type = super::UdpSink;
type ParentType = gst::Element;
type Instance = gst::subclass::ElementInstanceStruct<Self>;
type Class = subclass::simple::ClassStruct<Self>;
glib_object_subclass!();
fn class_init(klass: &mut Self::Class) {
klass.set_metadata(
"Thread-sharing UDP sink",
"Sink/Network",
"Thread-sharing UDP sink",
"Mathieu <mathieu@centricular.com>",
);
let caps = gst::Caps::new_any();
let sink_pad_template = gst::PadTemplate::new(
"sink",
gst::PadDirection::Sink,
gst::PadPresence::Always,
&caps,
)
.unwrap();
klass.add_pad_template(sink_pad_template);
klass.add_signal_with_class_handler(
"add",
glib::SignalFlags::RUN_LAST | glib::SignalFlags::ACTION,
&[String::static_type(), i32::static_type()],
glib::types::Type::Unit,
|_, args| {
let element = args[0]
.get::<super::UdpSink>()
.expect("signal arg")
.expect("missing signal arg");
let host = args[1]
.get::<String>()
.expect("signal arg")
.expect("missing signal arg");
let port = args[2]
.get::<i32>()
.expect("signal arg")
.expect("missing signal arg");
2020-03-26 18:07:25 +00:00
if let Ok(addr) = try_into_socket_addr(&element, &host, port) {
let udpsink = Self::from_instance(&element);
udpsink.add_client(addr);
}
None
},
);
klass.add_signal_with_class_handler(
"remove",
glib::SignalFlags::RUN_LAST | glib::SignalFlags::ACTION,
&[String::static_type(), i32::static_type()],
glib::types::Type::Unit,
|_, args| {
let element = args[0]
.get::<super::UdpSink>()
.expect("signal arg")
.expect("missing signal arg");
let host = args[1]
.get::<String>()
.expect("signal arg")
.expect("missing signal arg");
let port = args[2]
.get::<i32>()
.expect("signal arg")
.expect("missing signal arg");
let udpsink = Self::from_instance(&element);
if let Ok(addr) = try_into_socket_addr(&element, &host, port) {
udpsink.remove_client(addr);
}
None
},
);
klass.add_signal_with_class_handler(
"clear",
glib::SignalFlags::RUN_LAST | glib::SignalFlags::ACTION,
&[],
glib::types::Type::Unit,
|_, args| {
let element = args[0]
.get::<super::UdpSink>()
.expect("signal arg")
.expect("missing signal arg");
let udpsink = Self::from_instance(&element);
udpsink.clear_clients(std::iter::empty());
None
},
);
klass.install_properties(&PROPERTIES);
}
fn with_class(klass: &Self::Class) -> Self {
let settings = Arc::new(StdMutex::new(Settings::default()));
let sink_pad_handler = UdpSinkPadHandler::new(Arc::clone(&settings));
Self {
sink_pad: PadSink::new(
2020-06-11 10:45:15 +00:00
gst::Pad::from_template(&klass.get_pad_template("sink").unwrap(), Some("sink")),
sink_pad_handler.clone(),
),
sink_pad_handler,
2020-03-26 18:07:25 +00:00
task: Task::default(),
2020-03-15 11:40:45 +00:00
settings,
}
}
}
impl ObjectImpl for UdpSink {
fn set_property(&self, obj: &Self::Type, id: usize, value: &glib::Value) {
let prop = &PROPERTIES[id];
let mut settings = self.settings.lock().unwrap();
match *prop {
subclass::Property("sync", ..) => {
settings.sync = value.get_some().expect("type checked upstream");
}
subclass::Property("bind-address", ..) => {
settings.bind_address = value
.get()
.expect("type checked upstream")
.unwrap_or_else(|| "".into());
}
subclass::Property("bind-port", ..) => {
settings.bind_port = value.get_some().expect("type checked upstream");
}
subclass::Property("bind-address-v6", ..) => {
settings.bind_address_v6 = value
.get()
.expect("type checked upstream")
.unwrap_or_else(|| "".into());
}
subclass::Property("bind-port-v6", ..) => {
settings.bind_port_v6 = value.get_some().expect("type checked upstream");
}
subclass::Property("socket", ..) => {
settings.socket = value
.get::<gio::Socket>()
.expect("type checked upstream")
.map(|socket| GioSocketWrapper::new(&socket));
}
subclass::Property("used-socket", ..) => {
unreachable!();
}
subclass::Property("socket-v6", ..) => {
settings.socket_v6 = value
.get::<gio::Socket>()
.expect("type checked upstream")
.map(|socket| GioSocketWrapper::new(&socket));
}
subclass::Property("used-socket-v6", ..) => {
unreachable!();
}
subclass::Property("auto-multicast", ..) => {
settings.auto_multicast = value.get_some().expect("type checked upstream");
}
subclass::Property("loop", ..) => {
settings.multicast_loop = value.get_some().expect("type checked upstream");
}
subclass::Property("ttl", ..) => {
settings.ttl = value.get_some().expect("type checked upstream");
}
subclass::Property("ttl-mc", ..) => {
settings.ttl_mc = value.get_some().expect("type checked upstream");
}
subclass::Property("qos-dscp", ..) => {
settings.qos_dscp = value.get_some().expect("type checked upstream");
}
subclass::Property("clients", ..) => {
let clients: String = value
.get()
.expect("type checked upstream")
.unwrap_or_else(|| "".into());
2020-03-26 18:07:25 +00:00
let clients_iter = clients.split(',').filter_map(|client| {
2020-03-26 18:07:25 +00:00
let rsplit: Vec<&str> = client.rsplitn(2, ':').collect();
if rsplit.len() == 2 {
rsplit[0]
.parse::<i32>()
2020-03-26 18:07:25 +00:00
.map_err(|err| {
gst_error!(CAT, obj: obj, "Invalid port {}: {}", rsplit[0], err);
2020-03-26 18:07:25 +00:00
})
.and_then(|port| try_into_socket_addr(&obj, rsplit[1], port))
2020-03-26 18:07:25 +00:00
.ok()
} else {
None
}
});
drop(settings);
2020-03-26 18:07:25 +00:00
self.clear_clients(clients_iter);
}
subclass::Property("context", ..) => {
settings.context = value
.get()
.expect("type checked upstream")
.unwrap_or_else(|| "".into());
}
subclass::Property("context-wait", ..) => {
settings.context_wait = value.get_some().expect("type checked upstream");
}
_ => unimplemented!(),
}
}
fn get_property(&self, _obj: &Self::Type, id: usize) -> Result<glib::Value, ()> {
let prop = &PROPERTIES[id];
let settings = self.settings.lock().unwrap();
match *prop {
subclass::Property("sync", ..) => Ok(settings.sync.to_value()),
subclass::Property("bind-address", ..) => Ok(settings.bind_address.to_value()),
subclass::Property("bind-port", ..) => Ok(settings.bind_port.to_value()),
subclass::Property("bind-address-v6", ..) => Ok(settings.bind_address_v6.to_value()),
subclass::Property("bind-port-v6", ..) => Ok(settings.bind_port_v6.to_value()),
subclass::Property("socket", ..) => Ok(settings
.socket
.as_ref()
.map(GioSocketWrapper::as_socket)
.to_value()),
subclass::Property("used-socket", ..) => Ok(settings
.used_socket
.as_ref()
.map(GioSocketWrapper::as_socket)
.to_value()),
subclass::Property("socket-v6", ..) => Ok(settings
.socket_v6
.as_ref()
.map(GioSocketWrapper::as_socket)
.to_value()),
subclass::Property("used-socket-v6", ..) => Ok(settings
.used_socket_v6
.as_ref()
.map(GioSocketWrapper::as_socket)
.to_value()),
subclass::Property("auto-multicast", ..) => Ok(settings.sync.to_value()),
subclass::Property("loop", ..) => Ok(settings.multicast_loop.to_value()),
subclass::Property("ttl", ..) => Ok(settings.ttl.to_value()),
subclass::Property("ttl-mc", ..) => Ok(settings.ttl_mc.to_value()),
subclass::Property("qos-dscp", ..) => Ok(settings.qos_dscp.to_value()),
subclass::Property("clients", ..) => {
drop(settings);
2020-03-26 18:07:25 +00:00
let clients: Vec<String> = self
.sink_pad_handler
.get_clients()
.iter()
.map(ToString::to_string)
.collect();
Ok(clients.join(",").to_value())
}
subclass::Property("context", ..) => Ok(settings.context.to_value()),
subclass::Property("context-wait", ..) => Ok(settings.context_wait.to_value()),
_ => unimplemented!(),
}
}
fn constructed(&self, obj: &Self::Type) {
self.parent_constructed(obj);
obj.add_pad(self.sink_pad.gst_pad()).unwrap();
crate::set_element_flags(obj, gst::ElementFlags::SINK);
}
}
impl ElementImpl for UdpSink {
fn change_state(
&self,
element: &Self::Type,
transition: gst::StateChange,
) -> Result<gst::StateChangeSuccess, gst::StateChangeError> {
gst_trace!(CAT, obj: element, "Changing state {:?}", transition);
match transition {
gst::StateChange::NullToReady => {
self.prepare(element).map_err(|err| {
element.post_error_message(err);
gst::StateChangeError
})?;
}
gst::StateChange::ReadyToPaused => {
self.start(element).map_err(|_| gst::StateChangeError)?;
}
gst::StateChange::PausedToReady => {
self.stop(element).map_err(|_| gst::StateChangeError)?;
}
gst::StateChange::ReadyToNull => {
self.unprepare(element);
}
_ => (),
}
self.parent_change_state(element, transition)
}
fn send_event(&self, _element: &Self::Type, event: gst::Event) -> bool {
match event.view() {
EventView::Latency(ev) => {
2020-03-26 18:07:25 +00:00
self.sink_pad_handler.set_latency(ev.get_latency());
self.sink_pad.gst_pad().push_event(event)
}
EventView::Step(..) => false,
_ => self.sink_pad.gst_pad().push_event(event),
}
}
}