mirror of
https://gitlab.freedesktop.org/gstreamer/gst-plugins-rs.git
synced 2024-11-11 12:51:25 +00:00
rtpbin2: split send and receive halves into separate elements
There is now two elements, rtpsend and rtprecv that represent the two halves of a rtpsession. This avoids the potential pipeline loop if two peers are sending/receiving data towards each other. The two halves can be connected by setting the rtp-id property on each element to the same value and they will behave like a combined rtpbin-like element. Part-of: <https://gitlab.freedesktop.org/gstreamer/gst-plugins-rs/-/merge_requests/1426>
This commit is contained in:
Matthew Waters
parent
0121d78482
commit
1600d3b055
10 changed files with 1946 additions and 1245 deletions
|
|
@ -7191,141 +7191,6 @@
|
||||||
},
|
},
|
||||||
"rank": "marginal"
|
"rank": "marginal"
|
||||||
},
|
},
|
||||||
"rtpbin2": {
|
|
||||||
"author": "Matthew Waters <matthew@centricular.com>",
|
|
||||||
"description": "RTP sessions management",
|
|
||||||
"hierarchy": [
|
|
||||||
"GstRtpBin2",
|
|
||||||
"GstElement",
|
|
||||||
"GstObject",
|
|
||||||
"GInitiallyUnowned",
|
|
||||||
"GObject"
|
|
||||||
],
|
|
||||||
"klass": "Network/RTP/Filter",
|
|
||||||
"pad-templates": {
|
|
||||||
"rtcp_recv_sink_%%u": {
|
|
||||||
"caps": "application/x-rtcp:\n",
|
|
||||||
"direction": "sink",
|
|
||||||
"presence": "request"
|
|
||||||
},
|
|
||||||
"rtcp_send_src_%%u": {
|
|
||||||
"caps": "application/x-rtcp:\n",
|
|
||||||
"direction": "src",
|
|
||||||
"presence": "request"
|
|
||||||
},
|
|
||||||
"rtp_recv_sink_%%u": {
|
|
||||||
"caps": "application/x-rtp:\n",
|
|
||||||
"direction": "sink",
|
|
||||||
"presence": "request"
|
|
||||||
},
|
|
||||||
"rtp_recv_src_%%u_%%u_%%u": {
|
|
||||||
"caps": "application/x-rtp:\n",
|
|
||||||
"direction": "src",
|
|
||||||
"presence": "sometimes"
|
|
||||||
},
|
|
||||||
"rtp_send_sink_%%u": {
|
|
||||||
"caps": "application/x-rtp:\n",
|
|
||||||
"direction": "sink",
|
|
||||||
"presence": "request"
|
|
||||||
},
|
|
||||||
"rtp_send_src_%%u": {
|
|
||||||
"caps": "application/x-rtp:\n",
|
|
||||||
"direction": "src",
|
|
||||||
"presence": "sometimes"
|
|
||||||
}
|
|
||||||
},
|
|
||||||
"properties": {
|
|
||||||
"latency": {
|
|
||||||
"blurb": "Amount of ms to buffer",
|
|
||||||
"conditionally-available": false,
|
|
||||||
"construct": false,
|
|
||||||
"construct-only": false,
|
|
||||||
"controllable": false,
|
|
||||||
"default": "200",
|
|
||||||
"max": "-1",
|
|
||||||
"min": "0",
|
|
||||||
"mutable": "ready",
|
|
||||||
"readable": true,
|
|
||||||
"type": "guint",
|
|
||||||
"writable": true
|
|
||||||
},
|
|
||||||
"min-rtcp-interval": {
|
|
||||||
"blurb": "Minimum time (in ms) between RTCP reports",
|
|
||||||
"conditionally-available": false,
|
|
||||||
"construct": false,
|
|
||||||
"construct-only": false,
|
|
||||||
"controllable": false,
|
|
||||||
"default": "5000",
|
|
||||||
"max": "-1",
|
|
||||||
"min": "0",
|
|
||||||
"mutable": "ready",
|
|
||||||
"readable": true,
|
|
||||||
"type": "guint",
|
|
||||||
"writable": true
|
|
||||||
},
|
|
||||||
"reduced-size-rtcp": {
|
|
||||||
"blurb": "Use reduced size RTCP. Only has an effect if rtp-profile=avpf",
|
|
||||||
"conditionally-available": false,
|
|
||||||
"construct": false,
|
|
||||||
"construct-only": false,
|
|
||||||
"controllable": false,
|
|
||||||
"default": "false",
|
|
||||||
"mutable": "ready",
|
|
||||||
"readable": true,
|
|
||||||
"type": "gboolean",
|
|
||||||
"writable": true
|
|
||||||
},
|
|
||||||
"rtp-profile": {
|
|
||||||
"blurb": "RTP Profile to use",
|
|
||||||
"conditionally-available": false,
|
|
||||||
"construct": false,
|
|
||||||
"construct-only": false,
|
|
||||||
"controllable": false,
|
|
||||||
"default": "avp (0)",
|
|
||||||
"mutable": "ready",
|
|
||||||
"readable": true,
|
|
||||||
"type": "GstRtpBin2Profile",
|
|
||||||
"writable": true
|
|
||||||
},
|
|
||||||
"stats": {
|
|
||||||
"blurb": "Statistics about the session",
|
|
||||||
"conditionally-available": false,
|
|
||||||
"construct": false,
|
|
||||||
"construct-only": false,
|
|
||||||
"controllable": false,
|
|
||||||
"mutable": "null",
|
|
||||||
"readable": true,
|
|
||||||
"type": "guint",
|
|
||||||
"writable": false
|
|
||||||
},
|
|
||||||
"timestamping-mode": {
|
|
||||||
"blurb": "Govern how to pick presentation timestamps for packets",
|
|
||||||
"conditionally-available": false,
|
|
||||||
"construct": false,
|
|
||||||
"construct-only": false,
|
|
||||||
"controllable": false,
|
|
||||||
"default": "skew (2)",
|
|
||||||
"mutable": "ready",
|
|
||||||
"readable": true,
|
|
||||||
"type": "GstRtpBin2TimestampingMode",
|
|
||||||
"writable": true
|
|
||||||
}
|
|
||||||
},
|
|
||||||
"rank": "none",
|
|
||||||
"signals": {
|
|
||||||
"get-session": {
|
|
||||||
"action": true,
|
|
||||||
"args": [
|
|
||||||
{
|
|
||||||
"name": "arg0",
|
|
||||||
"type": "guint"
|
|
||||||
}
|
|
||||||
],
|
|
||||||
"return-type": "GstRtpBin2Session",
|
|
||||||
"when": "last"
|
|
||||||
}
|
|
||||||
}
|
|
||||||
},
|
|
||||||
"rtpgccbwe": {
|
"rtpgccbwe": {
|
||||||
"author": "Thibault Saunier <tsaunier@igalia.com>",
|
"author": "Thibault Saunier <tsaunier@igalia.com>",
|
||||||
"description": "Estimates current network bandwidth using the Google Congestion Control algorithm notifying about it through the 'bitrate' property",
|
"description": "Estimates current network bandwidth using the Google Congestion Control algorithm notifying about it through the 'bitrate' property",
|
||||||
|
|
@ -7888,6 +7753,206 @@
|
||||||
},
|
},
|
||||||
"rank": "marginal"
|
"rank": "marginal"
|
||||||
},
|
},
|
||||||
|
"rtprecv": {
|
||||||
|
"author": "Matthew Waters <matthew@centricular.com>",
|
||||||
|
"description": "RTP sessions management (receiver)",
|
||||||
|
"hierarchy": [
|
||||||
|
"GstRtpRecv",
|
||||||
|
"GstElement",
|
||||||
|
"GstObject",
|
||||||
|
"GInitiallyUnowned",
|
||||||
|
"GObject"
|
||||||
|
],
|
||||||
|
"klass": "Network/RTP/Filter",
|
||||||
|
"pad-templates": {
|
||||||
|
"rtcp_sink_%%u": {
|
||||||
|
"caps": "application/x-rtcp:\n",
|
||||||
|
"direction": "sink",
|
||||||
|
"presence": "request"
|
||||||
|
},
|
||||||
|
"rtp_sink_%%u": {
|
||||||
|
"caps": "application/x-rtp:\n",
|
||||||
|
"direction": "sink",
|
||||||
|
"presence": "request"
|
||||||
|
},
|
||||||
|
"rtp_src_%%u_%%u_%%u": {
|
||||||
|
"caps": "application/x-rtp:\n",
|
||||||
|
"direction": "src",
|
||||||
|
"presence": "sometimes"
|
||||||
|
}
|
||||||
|
},
|
||||||
|
"properties": {
|
||||||
|
"latency": {
|
||||||
|
"blurb": "Amount of ms to buffer",
|
||||||
|
"conditionally-available": false,
|
||||||
|
"construct": false,
|
||||||
|
"construct-only": false,
|
||||||
|
"controllable": false,
|
||||||
|
"default": "200",
|
||||||
|
"max": "-1",
|
||||||
|
"min": "0",
|
||||||
|
"mutable": "ready",
|
||||||
|
"readable": true,
|
||||||
|
"type": "guint",
|
||||||
|
"writable": true
|
||||||
|
},
|
||||||
|
"rtp-id": {
|
||||||
|
"blurb": "A connection ID shared with a rtpsend element for implementing both sending and receiving using the same RTP context",
|
||||||
|
"conditionally-available": false,
|
||||||
|
"construct": false,
|
||||||
|
"construct-only": false,
|
||||||
|
"controllable": false,
|
||||||
|
"default": "rtp-id",
|
||||||
|
"mutable": "null",
|
||||||
|
"readable": true,
|
||||||
|
"type": "gchararray",
|
||||||
|
"writable": true
|
||||||
|
},
|
||||||
|
"stats": {
|
||||||
|
"blurb": "Statistics about the session",
|
||||||
|
"conditionally-available": false,
|
||||||
|
"construct": false,
|
||||||
|
"construct-only": false,
|
||||||
|
"controllable": false,
|
||||||
|
"mutable": "null",
|
||||||
|
"readable": true,
|
||||||
|
"type": "guint",
|
||||||
|
"writable": false
|
||||||
|
},
|
||||||
|
"timestamping-mode": {
|
||||||
|
"blurb": "Govern how to pick presentation timestamps for packets",
|
||||||
|
"conditionally-available": false,
|
||||||
|
"construct": false,
|
||||||
|
"construct-only": false,
|
||||||
|
"controllable": false,
|
||||||
|
"default": "skew (2)",
|
||||||
|
"mutable": "ready",
|
||||||
|
"readable": true,
|
||||||
|
"type": "GstRtpBin2TimestampingMode",
|
||||||
|
"writable": true
|
||||||
|
}
|
||||||
|
},
|
||||||
|
"rank": "none",
|
||||||
|
"signals": {
|
||||||
|
"get-session": {
|
||||||
|
"action": true,
|
||||||
|
"args": [
|
||||||
|
{
|
||||||
|
"name": "arg0",
|
||||||
|
"type": "guint"
|
||||||
|
}
|
||||||
|
],
|
||||||
|
"return-type": "GstRtp2Session",
|
||||||
|
"when": "last"
|
||||||
|
}
|
||||||
|
}
|
||||||
|
},
|
||||||
|
"rtpsend": {
|
||||||
|
"author": "Matthew Waters <matthew@centricular.com>",
|
||||||
|
"description": "RTP session management (sender)",
|
||||||
|
"hierarchy": [
|
||||||
|
"GstRtpSend",
|
||||||
|
"GstElement",
|
||||||
|
"GstObject",
|
||||||
|
"GInitiallyUnowned",
|
||||||
|
"GObject"
|
||||||
|
],
|
||||||
|
"klass": "Network/RTP/Filter",
|
||||||
|
"pad-templates": {
|
||||||
|
"rtcp_src_%%u": {
|
||||||
|
"caps": "application/x-rtcp:\n",
|
||||||
|
"direction": "src",
|
||||||
|
"presence": "request"
|
||||||
|
},
|
||||||
|
"rtp_sink_%%u": {
|
||||||
|
"caps": "application/x-rtp:\n",
|
||||||
|
"direction": "sink",
|
||||||
|
"presence": "request"
|
||||||
|
},
|
||||||
|
"rtp_src_%%u": {
|
||||||
|
"caps": "application/x-rtp:\n",
|
||||||
|
"direction": "src",
|
||||||
|
"presence": "sometimes"
|
||||||
|
}
|
||||||
|
},
|
||||||
|
"properties": {
|
||||||
|
"min-rtcp-interval": {
|
||||||
|
"blurb": "Minimum time (in ms) between RTCP reports",
|
||||||
|
"conditionally-available": false,
|
||||||
|
"construct": false,
|
||||||
|
"construct-only": false,
|
||||||
|
"controllable": false,
|
||||||
|
"default": "5000",
|
||||||
|
"max": "-1",
|
||||||
|
"min": "0",
|
||||||
|
"mutable": "ready",
|
||||||
|
"readable": true,
|
||||||
|
"type": "guint",
|
||||||
|
"writable": true
|
||||||
|
},
|
||||||
|
"reduced-size-rtcp": {
|
||||||
|
"blurb": "Use reduced size RTCP. Only has an effect if rtp-profile=avpf",
|
||||||
|
"conditionally-available": false,
|
||||||
|
"construct": false,
|
||||||
|
"construct-only": false,
|
||||||
|
"controllable": false,
|
||||||
|
"default": "false",
|
||||||
|
"mutable": "ready",
|
||||||
|
"readable": true,
|
||||||
|
"type": "gboolean",
|
||||||
|
"writable": true
|
||||||
|
},
|
||||||
|
"rtp-id": {
|
||||||
|
"blurb": "A connection ID shared with a rtprecv element for implementing both sending and receiving using the same RTP context",
|
||||||
|
"conditionally-available": false,
|
||||||
|
"construct": false,
|
||||||
|
"construct-only": false,
|
||||||
|
"controllable": false,
|
||||||
|
"default": "rtp-id",
|
||||||
|
"mutable": "null",
|
||||||
|
"readable": true,
|
||||||
|
"type": "gchararray",
|
||||||
|
"writable": true
|
||||||
|
},
|
||||||
|
"rtp-profile": {
|
||||||
|
"blurb": "RTP Profile to use",
|
||||||
|
"conditionally-available": false,
|
||||||
|
"construct": false,
|
||||||
|
"construct-only": false,
|
||||||
|
"controllable": false,
|
||||||
|
"default": "avp (0)",
|
||||||
|
"mutable": "ready",
|
||||||
|
"readable": true,
|
||||||
|
"type": "GstRtpSendProfile",
|
||||||
|
"writable": true
|
||||||
|
},
|
||||||
|
"stats": {
|
||||||
|
"blurb": "Statistics about the session",
|
||||||
|
"conditionally-available": false,
|
||||||
|
"construct": false,
|
||||||
|
"construct-only": false,
|
||||||
|
"controllable": false,
|
||||||
|
"mutable": "null",
|
||||||
|
"readable": true,
|
||||||
|
"type": "guint",
|
||||||
|
"writable": false
|
||||||
|
}
|
||||||
|
},
|
||||||
|
"rank": "none",
|
||||||
|
"signals": {
|
||||||
|
"get-session": {
|
||||||
|
"action": true,
|
||||||
|
"args": [
|
||||||
|
{
|
||||||
|
"name": "arg0",
|
||||||
|
"type": "guint"
|
||||||
|
}
|
||||||
|
],
|
||||||
|
"return-type": "GstRtp2Session",
|
||||||
|
"when": "last"
|
||||||
|
}
|
||||||
|
}
|
||||||
|
},
|
||||||
"rtpvp8depay2": {
|
"rtpvp8depay2": {
|
||||||
"author": "Sebastian Dröge <sebastian@centricular.com>",
|
"author": "Sebastian Dröge <sebastian@centricular.com>",
|
||||||
"description": "Depayload VP8 from RTP packets",
|
"description": "Depayload VP8 from RTP packets",
|
||||||
|
|
|
||||||
|
|
@ -6,22 +6,22 @@ use gst::subclass::prelude::*;
|
||||||
use once_cell::sync::Lazy;
|
use once_cell::sync::Lazy;
|
||||||
use std::sync::{Mutex, Weak};
|
use std::sync::{Mutex, Weak};
|
||||||
|
|
||||||
use crate::rtpbin2::imp::BinSessionInner;
|
use crate::rtpbin2::internal::SharedSessionInner;
|
||||||
|
|
||||||
static CAT: Lazy<gst::DebugCategory> = Lazy::new(|| {
|
static CAT: Lazy<gst::DebugCategory> = Lazy::new(|| {
|
||||||
gst::DebugCategory::new(
|
gst::DebugCategory::new(
|
||||||
"rtpbin2-config",
|
"rtp2-config",
|
||||||
gst::DebugColorFlags::empty(),
|
gst::DebugColorFlags::empty(),
|
||||||
Some("RtpBin2 config"),
|
Some("Rtp2 config"),
|
||||||
)
|
)
|
||||||
});
|
});
|
||||||
|
|
||||||
glib::wrapper! {
|
glib::wrapper! {
|
||||||
pub struct RtpBin2Session(ObjectSubclass<imp::RtpBin2Session>);
|
pub struct Rtp2Session(ObjectSubclass<imp::Rtp2Session>);
|
||||||
}
|
}
|
||||||
|
|
||||||
impl RtpBin2Session {
|
impl Rtp2Session {
|
||||||
pub(crate) fn new(weak_session: Weak<Mutex<BinSessionInner>>) -> Self {
|
pub(crate) fn new(weak_session: Weak<Mutex<SharedSessionInner>>) -> Self {
|
||||||
let ret = glib::Object::new::<Self>();
|
let ret = glib::Object::new::<Self>();
|
||||||
let imp = ret.imp();
|
let imp = ret.imp();
|
||||||
imp.set_session(weak_session);
|
imp.set_session(weak_session);
|
||||||
|
|
@ -36,21 +36,21 @@ mod imp {
|
||||||
|
|
||||||
#[derive(Debug, Default)]
|
#[derive(Debug, Default)]
|
||||||
struct State {
|
struct State {
|
||||||
pub(super) weak_session: Option<Weak<Mutex<BinSessionInner>>>,
|
pub(super) weak_session: Option<Weak<Mutex<SharedSessionInner>>>,
|
||||||
}
|
}
|
||||||
|
|
||||||
#[derive(Debug, Default)]
|
#[derive(Debug, Default)]
|
||||||
pub struct RtpBin2Session {
|
pub struct Rtp2Session {
|
||||||
state: Mutex<State>,
|
state: Mutex<State>,
|
||||||
}
|
}
|
||||||
|
|
||||||
impl RtpBin2Session {
|
impl Rtp2Session {
|
||||||
pub(super) fn set_session(&self, weak_session: Weak<Mutex<BinSessionInner>>) {
|
pub(super) fn set_session(&self, weak_session: Weak<Mutex<SharedSessionInner>>) {
|
||||||
let mut state = self.state.lock().unwrap();
|
let mut state = self.state.lock().unwrap();
|
||||||
state.weak_session = Some(weak_session);
|
state.weak_session = Some(weak_session);
|
||||||
}
|
}
|
||||||
|
|
||||||
fn session(&self) -> Option<Arc<Mutex<BinSessionInner>>> {
|
fn session(&self) -> Option<Arc<Mutex<SharedSessionInner>>> {
|
||||||
self.state
|
self.state
|
||||||
.lock()
|
.lock()
|
||||||
.unwrap()
|
.unwrap()
|
||||||
|
|
@ -84,7 +84,7 @@ mod imp {
|
||||||
}
|
}
|
||||||
|
|
||||||
pub fn pt_map(&self) -> gst::Structure {
|
pub fn pt_map(&self) -> gst::Structure {
|
||||||
let mut ret = gst::Structure::builder("application/x-rtpbin2-pt-map");
|
let mut ret = gst::Structure::builder("application/x-rtp2-pt-map");
|
||||||
let Some(session) = self.session() else {
|
let Some(session) = self.session() else {
|
||||||
return ret.build();
|
return ret.build();
|
||||||
};
|
};
|
||||||
|
|
@ -105,13 +105,13 @@ mod imp {
|
||||||
}
|
}
|
||||||
|
|
||||||
#[glib::object_subclass]
|
#[glib::object_subclass]
|
||||||
impl ObjectSubclass for RtpBin2Session {
|
impl ObjectSubclass for Rtp2Session {
|
||||||
const NAME: &'static str = "GstRtpBin2Session";
|
const NAME: &'static str = "GstRtp2Session";
|
||||||
type Type = super::RtpBin2Session;
|
type Type = super::Rtp2Session;
|
||||||
type ParentType = glib::Object;
|
type ParentType = glib::Object;
|
||||||
}
|
}
|
||||||
|
|
||||||
impl ObjectImpl for RtpBin2Session {
|
impl ObjectImpl for Rtp2Session {
|
||||||
fn properties() -> &'static [glib::ParamSpec] {
|
fn properties() -> &'static [glib::ParamSpec] {
|
||||||
static PROPERTIES: Lazy<Vec<glib::ParamSpec>> = Lazy::new(|| {
|
static PROPERTIES: Lazy<Vec<glib::ParamSpec>> = Lazy::new(|| {
|
||||||
vec![glib::ParamSpecBoxed::builder::<gst::Structure>("pt-map")
|
vec![glib::ParamSpecBoxed::builder::<gst::Structure>("pt-map")
|
||||||
|
|
@ -161,40 +161,57 @@ mod imp {
|
||||||
|
|
||||||
#[cfg(test)]
|
#[cfg(test)]
|
||||||
mod tests {
|
mod tests {
|
||||||
use std::sync::{atomic::AtomicBool, Arc};
|
use std::sync::{
|
||||||
|
atomic::{AtomicBool, AtomicUsize},
|
||||||
|
Arc,
|
||||||
|
};
|
||||||
|
|
||||||
use crate::{rtpbin2::session::tests::generate_rtp_packet, test_init};
|
use crate::{rtpbin2::session::tests::generate_rtp_packet, test_init};
|
||||||
|
|
||||||
use super::*;
|
use super::*;
|
||||||
|
|
||||||
|
static ELEMENT_COUNTER: AtomicUsize = AtomicUsize::new(0);
|
||||||
|
|
||||||
|
fn next_element_counter() -> usize {
|
||||||
|
ELEMENT_COUNTER.fetch_add(1, std::sync::atomic::Ordering::SeqCst)
|
||||||
|
}
|
||||||
|
|
||||||
#[test]
|
#[test]
|
||||||
fn pt_map_get_empty() {
|
fn pt_map_get_empty() {
|
||||||
test_init();
|
test_init();
|
||||||
let rtpbin2 = gst::ElementFactory::make("rtpbin2").build().unwrap();
|
let id = next_element_counter();
|
||||||
let _pad = rtpbin2.request_pad_simple("rtp_send_sink_0").unwrap();
|
let rtpbin2 = gst::ElementFactory::make("rtpsend")
|
||||||
|
.property("rtp-id", id.to_string())
|
||||||
|
.build()
|
||||||
|
.unwrap();
|
||||||
|
let _pad = rtpbin2.request_pad_simple("rtp_sink_0").unwrap();
|
||||||
let session = rtpbin2.emit_by_name::<gst::glib::Object>("get-session", &[&0u32]);
|
let session = rtpbin2.emit_by_name::<gst::glib::Object>("get-session", &[&0u32]);
|
||||||
let pt_map = session.property::<gst::Structure>("pt-map");
|
let pt_map = session.property::<gst::Structure>("pt-map");
|
||||||
assert!(pt_map.has_name("application/x-rtpbin2-pt-map"));
|
assert!(pt_map.has_name("application/x-rtp2-pt-map"));
|
||||||
assert_eq!(pt_map.fields().len(), 0);
|
assert_eq!(pt_map.fields().len(), 0);
|
||||||
}
|
}
|
||||||
|
|
||||||
#[test]
|
#[test]
|
||||||
fn pt_map_set() {
|
fn pt_map_set() {
|
||||||
test_init();
|
test_init();
|
||||||
let rtpbin2 = gst::ElementFactory::make("rtpbin2").build().unwrap();
|
let id = next_element_counter();
|
||||||
let _pad = rtpbin2.request_pad_simple("rtp_send_sink_0").unwrap();
|
let rtpbin2 = gst::ElementFactory::make("rtpsend")
|
||||||
|
.property("rtp-id", id.to_string())
|
||||||
|
.build()
|
||||||
|
.unwrap();
|
||||||
|
let _pad = rtpbin2.request_pad_simple("rtp_sink_0").unwrap();
|
||||||
let session = rtpbin2.emit_by_name::<gst::glib::Object>("get-session", &[&0u32]);
|
let session = rtpbin2.emit_by_name::<gst::glib::Object>("get-session", &[&0u32]);
|
||||||
let pt = 96i32;
|
let pt = 96i32;
|
||||||
let pt_caps = gst::Caps::builder("application/x-rtp")
|
let pt_caps = gst::Caps::builder("application/x-rtp")
|
||||||
.field("payload", pt)
|
.field("payload", pt)
|
||||||
.field("clock-rate", 90000i32)
|
.field("clock-rate", 90000i32)
|
||||||
.build();
|
.build();
|
||||||
let pt_map = gst::Structure::builder("application/x-rtpbin2-pt-map")
|
let pt_map = gst::Structure::builder("application/x-rtp2-pt-map")
|
||||||
.field(pt.to_string(), pt_caps.clone())
|
.field(pt.to_string(), pt_caps.clone())
|
||||||
.build();
|
.build();
|
||||||
session.set_property("pt-map", pt_map);
|
session.set_property("pt-map", pt_map);
|
||||||
let prop = session.property::<gst::Structure>("pt-map");
|
let prop = session.property::<gst::Structure>("pt-map");
|
||||||
assert!(prop.has_name("application/x-rtpbin2-pt-map"));
|
assert!(prop.has_name("application/x-rtp2-pt-map"));
|
||||||
assert_eq!(prop.fields().len(), 1);
|
assert_eq!(prop.fields().len(), 1);
|
||||||
let caps = prop.get::<gst::Caps>(pt.to_string()).unwrap();
|
let caps = prop.get::<gst::Caps>(pt.to_string()).unwrap();
|
||||||
assert_eq!(pt_caps, caps);
|
assert_eq!(pt_caps, caps);
|
||||||
|
|
@ -203,12 +220,16 @@ mod tests {
|
||||||
#[test]
|
#[test]
|
||||||
fn pt_map_set_none() {
|
fn pt_map_set_none() {
|
||||||
test_init();
|
test_init();
|
||||||
let rtpbin2 = gst::ElementFactory::make("rtpbin2").build().unwrap();
|
let id = next_element_counter();
|
||||||
let _pad = rtpbin2.request_pad_simple("rtp_send_sink_0").unwrap();
|
let rtpbin2 = gst::ElementFactory::make("rtpsend")
|
||||||
|
.property("rtp-id", id.to_string())
|
||||||
|
.build()
|
||||||
|
.unwrap();
|
||||||
|
let _pad = rtpbin2.request_pad_simple("rtp_sink_0").unwrap();
|
||||||
let session = rtpbin2.emit_by_name::<gst::glib::Object>("get-session", &[&0u32]);
|
let session = rtpbin2.emit_by_name::<gst::glib::Object>("get-session", &[&0u32]);
|
||||||
session.set_property("pt-map", None::<gst::Structure>);
|
session.set_property("pt-map", None::<gst::Structure>);
|
||||||
let prop = session.property::<gst::Structure>("pt-map");
|
let prop = session.property::<gst::Structure>("pt-map");
|
||||||
assert!(prop.has_name("application/x-rtpbin2-pt-map"));
|
assert!(prop.has_name("application/x-rtp2-pt-map"));
|
||||||
}
|
}
|
||||||
|
|
||||||
#[test]
|
#[test]
|
||||||
|
|
@ -216,12 +237,13 @@ mod tests {
|
||||||
test_init();
|
test_init();
|
||||||
let ssrc = 0x12345678;
|
let ssrc = 0x12345678;
|
||||||
let new_ssrc_hit = Arc::new(AtomicBool::new(false));
|
let new_ssrc_hit = Arc::new(AtomicBool::new(false));
|
||||||
let rtpbin2 = gst::ElementFactory::make("rtpbin2").build().unwrap();
|
let id = next_element_counter();
|
||||||
let mut h = gst_check::Harness::with_element(
|
let rtpbin2 = gst::ElementFactory::make("rtpsend")
|
||||||
&rtpbin2,
|
.property("rtp-id", id.to_string())
|
||||||
Some("rtp_send_sink_0"),
|
.build()
|
||||||
Some("rtp_send_src_0"),
|
.unwrap();
|
||||||
);
|
let mut h =
|
||||||
|
gst_check::Harness::with_element(&rtpbin2, Some("rtp_sink_0"), Some("rtp_src_0"));
|
||||||
let session = h
|
let session = h
|
||||||
.element()
|
.element()
|
||||||
.unwrap()
|
.unwrap()
|
||||||
|
|
@ -254,13 +276,14 @@ mod tests {
|
||||||
test_init();
|
test_init();
|
||||||
let ssrc = 0x12345678;
|
let ssrc = 0x12345678;
|
||||||
let (bye_ssrc_sender, bye_ssrc_receiver) = std::sync::mpsc::sync_channel(16);
|
let (bye_ssrc_sender, bye_ssrc_receiver) = std::sync::mpsc::sync_channel(16);
|
||||||
let rtpbin2 = gst::ElementFactory::make("rtpbin2").build().unwrap();
|
let id = next_element_counter();
|
||||||
let mut h = gst_check::Harness::with_element(
|
let rtpbin2 = gst::ElementFactory::make("rtpsend")
|
||||||
&rtpbin2,
|
.property("rtp-id", id.to_string())
|
||||||
Some("rtp_send_sink_0"),
|
.build()
|
||||||
Some("rtp_send_src_0"),
|
.unwrap();
|
||||||
);
|
let mut h =
|
||||||
let mut h_rtcp = gst_check::Harness::with_element(&rtpbin2, None, Some("rtcp_send_src_0"));
|
gst_check::Harness::with_element(&rtpbin2, Some("rtp_sink_0"), Some("rtp_src_0"));
|
||||||
|
let mut h_rtcp = gst_check::Harness::with_element(&rtpbin2, None, Some("rtcp_src_0"));
|
||||||
let session = h
|
let session = h
|
||||||
.element()
|
.element()
|
||||||
.unwrap()
|
.unwrap()
|
||||||
|
|
|
||||||
451
net/rtp/src/rtpbin2/internal.rs
Normal file
451
net/rtp/src/rtpbin2/internal.rs
Normal file
|
|
@ -0,0 +1,451 @@
|
||||||
|
// SPDX-License-Identifier: MPL-2.0
|
||||||
|
|
||||||
|
use std::{
|
||||||
|
collections::HashMap,
|
||||||
|
sync::{Arc, Mutex},
|
||||||
|
task::Waker,
|
||||||
|
time::Duration,
|
||||||
|
};
|
||||||
|
|
||||||
|
use gst::glib;
|
||||||
|
use once_cell::sync::{Lazy, OnceCell};
|
||||||
|
|
||||||
|
use super::config::Rtp2Session;
|
||||||
|
use super::session::{RtpProfile, Session};
|
||||||
|
use super::source::ReceivedRb;
|
||||||
|
|
||||||
|
static CAT: Lazy<gst::DebugCategory> = Lazy::new(|| {
|
||||||
|
gst::DebugCategory::new(
|
||||||
|
"rtpinternalsession",
|
||||||
|
gst::DebugColorFlags::empty(),
|
||||||
|
Some("RTP Session (internal)"),
|
||||||
|
)
|
||||||
|
});
|
||||||
|
|
||||||
|
static SHARED_RTP_STATE: OnceCell<Mutex<HashMap<String, SharedRtpState>>> = OnceCell::new();
|
||||||
|
|
||||||
|
#[derive(Debug, Clone)]
|
||||||
|
pub struct SharedRtpState {
|
||||||
|
name: String,
|
||||||
|
inner: Arc<Mutex<SharedRtpStateInner>>,
|
||||||
|
}
|
||||||
|
|
||||||
|
#[derive(Debug)]
|
||||||
|
struct SharedRtpStateInner {
|
||||||
|
sessions: HashMap<usize, SharedSession>,
|
||||||
|
send_outstanding: bool,
|
||||||
|
recv_outstanding: bool,
|
||||||
|
}
|
||||||
|
|
||||||
|
impl SharedRtpState {
|
||||||
|
pub fn recv_get_or_init(name: String) -> Self {
|
||||||
|
SHARED_RTP_STATE
|
||||||
|
.get_or_init(|| Mutex::new(HashMap::new()))
|
||||||
|
.lock()
|
||||||
|
.unwrap()
|
||||||
|
.entry(name)
|
||||||
|
.and_modify(|v| {
|
||||||
|
v.inner.lock().unwrap().recv_outstanding = true;
|
||||||
|
})
|
||||||
|
.or_insert_with_key(|name| SharedRtpState {
|
||||||
|
name: name.to_owned(),
|
||||||
|
inner: Arc::new(Mutex::new(SharedRtpStateInner {
|
||||||
|
sessions: HashMap::new(),
|
||||||
|
send_outstanding: false,
|
||||||
|
recv_outstanding: true,
|
||||||
|
})),
|
||||||
|
})
|
||||||
|
.clone()
|
||||||
|
}
|
||||||
|
|
||||||
|
pub fn send_get_or_init(name: String) -> Self {
|
||||||
|
SHARED_RTP_STATE
|
||||||
|
.get_or_init(|| Mutex::new(HashMap::new()))
|
||||||
|
.lock()
|
||||||
|
.unwrap()
|
||||||
|
.entry(name)
|
||||||
|
.and_modify(|v| {
|
||||||
|
v.inner.lock().unwrap().send_outstanding = true;
|
||||||
|
})
|
||||||
|
.or_insert_with_key(|name| SharedRtpState {
|
||||||
|
name: name.to_owned(),
|
||||||
|
inner: Arc::new(Mutex::new(SharedRtpStateInner {
|
||||||
|
sessions: HashMap::new(),
|
||||||
|
send_outstanding: true,
|
||||||
|
recv_outstanding: false,
|
||||||
|
})),
|
||||||
|
})
|
||||||
|
.clone()
|
||||||
|
}
|
||||||
|
|
||||||
|
pub fn name(&self) -> &str {
|
||||||
|
&self.name
|
||||||
|
}
|
||||||
|
|
||||||
|
pub fn unmark_send_outstanding(&self) {
|
||||||
|
let mut inner = self.inner.lock().unwrap();
|
||||||
|
inner.send_outstanding = false;
|
||||||
|
if !inner.recv_outstanding {
|
||||||
|
Self::remove_from_global(&self.name);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
pub fn unmark_recv_outstanding(&self) {
|
||||||
|
let mut inner = self.inner.lock().unwrap();
|
||||||
|
inner.recv_outstanding = false;
|
||||||
|
if !inner.send_outstanding {
|
||||||
|
Self::remove_from_global(&self.name);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
fn remove_from_global(name: &str) {
|
||||||
|
let _shared = SHARED_RTP_STATE.get().unwrap().lock().unwrap().remove(name);
|
||||||
|
}
|
||||||
|
|
||||||
|
pub fn session_get_or_init<F>(&self, id: usize, f: F) -> SharedSession
|
||||||
|
where
|
||||||
|
F: FnOnce() -> SharedSession,
|
||||||
|
{
|
||||||
|
self.inner
|
||||||
|
.lock()
|
||||||
|
.unwrap()
|
||||||
|
.sessions
|
||||||
|
.entry(id)
|
||||||
|
.or_insert_with(f)
|
||||||
|
.clone()
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
#[derive(Debug, Clone)]
|
||||||
|
pub struct SharedSession {
|
||||||
|
pub(crate) id: usize,
|
||||||
|
pub(crate) inner: Arc<Mutex<SharedSessionInner>>,
|
||||||
|
pub(crate) config: Rtp2Session,
|
||||||
|
}
|
||||||
|
|
||||||
|
impl SharedSession {
|
||||||
|
pub fn new(
|
||||||
|
id: usize,
|
||||||
|
profile: RtpProfile,
|
||||||
|
min_rtcp_interval: Duration,
|
||||||
|
reduced_size_rtcp: bool,
|
||||||
|
) -> Self {
|
||||||
|
let mut inner = SharedSessionInner::new(id);
|
||||||
|
inner.session.set_min_rtcp_interval(min_rtcp_interval);
|
||||||
|
inner.session.set_profile(profile);
|
||||||
|
inner.session.set_reduced_size_rtcp(reduced_size_rtcp);
|
||||||
|
let inner = Arc::new(Mutex::new(inner));
|
||||||
|
let weak_inner = Arc::downgrade(&inner);
|
||||||
|
Self {
|
||||||
|
id,
|
||||||
|
inner,
|
||||||
|
config: Rtp2Session::new(weak_inner),
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
#[derive(Debug)]
|
||||||
|
pub(crate) struct SharedSessionInner {
|
||||||
|
id: usize,
|
||||||
|
|
||||||
|
pub(crate) session: Session,
|
||||||
|
|
||||||
|
pub(crate) pt_map: HashMap<u8, gst::Caps>,
|
||||||
|
|
||||||
|
pub(crate) rtcp_waker: Option<Waker>,
|
||||||
|
pub(crate) rtp_send_sinkpad: Option<gst::Pad>,
|
||||||
|
}
|
||||||
|
|
||||||
|
impl SharedSessionInner {
|
||||||
|
fn new(id: usize) -> Self {
|
||||||
|
Self {
|
||||||
|
id,
|
||||||
|
|
||||||
|
session: Session::new(),
|
||||||
|
|
||||||
|
pt_map: HashMap::default(),
|
||||||
|
rtcp_waker: None,
|
||||||
|
rtp_send_sinkpad: None,
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
pub fn clear_pt_map(&mut self) {
|
||||||
|
self.pt_map.clear();
|
||||||
|
}
|
||||||
|
|
||||||
|
pub fn add_caps(&mut self, caps: gst::Caps) {
|
||||||
|
let Some((pt, clock_rate)) = pt_clock_rate_from_caps(&caps) else {
|
||||||
|
return;
|
||||||
|
};
|
||||||
|
let caps_clone = caps.clone();
|
||||||
|
self.pt_map
|
||||||
|
.entry(pt)
|
||||||
|
.and_modify(move |entry| *entry = caps)
|
||||||
|
.or_insert_with(move || caps_clone);
|
||||||
|
self.session.set_pt_clock_rate(pt, clock_rate);
|
||||||
|
}
|
||||||
|
|
||||||
|
pub(crate) fn caps_from_pt(&self, pt: u8) -> gst::Caps {
|
||||||
|
self.pt_map.get(&pt).cloned().unwrap_or(
|
||||||
|
gst::Caps::builder("application/x-rtp")
|
||||||
|
.field("payload", pt as i32)
|
||||||
|
.build(),
|
||||||
|
)
|
||||||
|
}
|
||||||
|
|
||||||
|
pub fn pt_map(&self) -> impl Iterator<Item = (u8, &gst::Caps)> + '_ {
|
||||||
|
self.pt_map.iter().map(|(&k, v)| (k, v))
|
||||||
|
}
|
||||||
|
|
||||||
|
pub fn stats(&self) -> gst::Structure {
|
||||||
|
let mut session_stats = gst::Structure::builder("application/x-rtpbin2-session-stats")
|
||||||
|
.field("id", self.id as u64);
|
||||||
|
for ssrc in self.session.ssrcs() {
|
||||||
|
if let Some(ls) = self.session.local_send_source_by_ssrc(ssrc) {
|
||||||
|
let mut source_stats =
|
||||||
|
gst::Structure::builder("application/x-rtpbin2-source-stats")
|
||||||
|
.field("ssrc", ls.ssrc())
|
||||||
|
.field("sender", true)
|
||||||
|
.field("local", true)
|
||||||
|
.field("packets-sent", ls.packet_count())
|
||||||
|
.field("octets-sent", ls.octet_count())
|
||||||
|
.field("bitrate", ls.bitrate() as u64);
|
||||||
|
if let Some(pt) = ls.payload_type() {
|
||||||
|
if let Some(clock_rate) = self.session.clock_rate_from_pt(pt) {
|
||||||
|
source_stats = source_stats.field("clock-rate", clock_rate);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
if let Some(sr) = ls.last_sent_sr() {
|
||||||
|
source_stats = source_stats
|
||||||
|
.field("sr-ntptime", sr.ntp_timestamp().as_u64())
|
||||||
|
.field("sr-rtptime", sr.rtp_timestamp())
|
||||||
|
.field("sr-octet-count", sr.octet_count())
|
||||||
|
.field("sr-packet-count", sr.packet_count());
|
||||||
|
}
|
||||||
|
let rbs = gst::List::new(ls.received_report_blocks().map(
|
||||||
|
|(sender_ssrc, ReceivedRb { rb, .. })| {
|
||||||
|
gst::Structure::builder("application/x-rtcp-report-block")
|
||||||
|
.field("sender-ssrc", sender_ssrc)
|
||||||
|
.field("rb-fraction-lost", rb.fraction_lost())
|
||||||
|
.field("rb-packets-lost", rb.cumulative_lost())
|
||||||
|
.field("rb-extended_sequence_number", rb.extended_sequence_number())
|
||||||
|
.field("rb-jitter", rb.jitter())
|
||||||
|
.field("rb-last-sr-ntp-time", rb.last_sr_ntp_time())
|
||||||
|
.field("rb-delay_since_last-sr-ntp-time", rb.delay_since_last_sr())
|
||||||
|
.build()
|
||||||
|
},
|
||||||
|
));
|
||||||
|
match rbs.len() {
|
||||||
|
0 => (),
|
||||||
|
1 => {
|
||||||
|
source_stats =
|
||||||
|
source_stats.field("report-blocks", rbs.first().unwrap().clone());
|
||||||
|
}
|
||||||
|
_ => {
|
||||||
|
source_stats = source_stats.field("report-blocks", rbs);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// TODO: add jitter, packets-lost
|
||||||
|
session_stats = session_stats.field(ls.ssrc().to_string(), source_stats.build());
|
||||||
|
} else if let Some(lr) = self.session.local_receive_source_by_ssrc(ssrc) {
|
||||||
|
let mut source_stats =
|
||||||
|
gst::Structure::builder("application/x-rtpbin2-source-stats")
|
||||||
|
.field("ssrc", lr.ssrc())
|
||||||
|
.field("sender", false)
|
||||||
|
.field("local", true);
|
||||||
|
if let Some(pt) = lr.payload_type() {
|
||||||
|
if let Some(clock_rate) = self.session.clock_rate_from_pt(pt) {
|
||||||
|
source_stats = source_stats.field("clock-rate", clock_rate);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
// TODO: add rb stats
|
||||||
|
session_stats = session_stats.field(lr.ssrc().to_string(), source_stats.build());
|
||||||
|
} else if let Some(rs) = self.session.remote_send_source_by_ssrc(ssrc) {
|
||||||
|
let mut source_stats =
|
||||||
|
gst::Structure::builder("application/x-rtpbin2-source-stats")
|
||||||
|
.field("ssrc", rs.ssrc())
|
||||||
|
.field("sender", true)
|
||||||
|
.field("local", false)
|
||||||
|
.field("octets-received", rs.octet_count())
|
||||||
|
.field("packets-received", rs.packet_count())
|
||||||
|
.field("bitrate", rs.bitrate() as u64)
|
||||||
|
.field("jitter", rs.jitter())
|
||||||
|
.field("packets-lost", rs.packets_lost());
|
||||||
|
if let Some(pt) = rs.payload_type() {
|
||||||
|
if let Some(clock_rate) = self.session.clock_rate_from_pt(pt) {
|
||||||
|
source_stats = source_stats.field("clock-rate", clock_rate);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
if let Some(rtp_from) = rs.rtp_from() {
|
||||||
|
source_stats = source_stats.field("rtp-from", rtp_from.to_string());
|
||||||
|
}
|
||||||
|
if let Some(rtcp_from) = rs.rtcp_from() {
|
||||||
|
source_stats = source_stats.field("rtcp-from", rtcp_from.to_string());
|
||||||
|
}
|
||||||
|
if let Some(sr) = rs.last_received_sr() {
|
||||||
|
source_stats = source_stats
|
||||||
|
.field("sr-ntptime", sr.ntp_timestamp().as_u64())
|
||||||
|
.field("sr-rtptime", sr.rtp_timestamp())
|
||||||
|
.field("sr-octet-count", sr.octet_count())
|
||||||
|
.field("sr-packet-count", sr.packet_count());
|
||||||
|
}
|
||||||
|
if let Some(rb) = rs.last_sent_rb() {
|
||||||
|
source_stats = source_stats
|
||||||
|
.field("sent-rb-fraction-lost", rb.fraction_lost())
|
||||||
|
.field("sent-rb-packets-lost", rb.cumulative_lost())
|
||||||
|
.field(
|
||||||
|
"sent-rb-extended-sequence-number",
|
||||||
|
rb.extended_sequence_number(),
|
||||||
|
)
|
||||||
|
.field("sent-rb-jitter", rb.jitter())
|
||||||
|
.field("sent-rb-last-sr-ntp-time", rb.last_sr_ntp_time())
|
||||||
|
.field(
|
||||||
|
"sent-rb-delay-since-last-sr-ntp-time",
|
||||||
|
rb.delay_since_last_sr(),
|
||||||
|
);
|
||||||
|
}
|
||||||
|
let rbs = gst::List::new(rs.received_report_blocks().map(
|
||||||
|
|(sender_ssrc, ReceivedRb { rb, .. })| {
|
||||||
|
gst::Structure::builder("application/x-rtcp-report-block")
|
||||||
|
.field("sender-ssrc", sender_ssrc)
|
||||||
|
.field("rb-fraction-lost", rb.fraction_lost())
|
||||||
|
.field("rb-packets-lost", rb.cumulative_lost())
|
||||||
|
.field("rb-extended_sequence_number", rb.extended_sequence_number())
|
||||||
|
.field("rb-jitter", rb.jitter())
|
||||||
|
.field("rb-last-sr-ntp-time", rb.last_sr_ntp_time())
|
||||||
|
.field("rb-delay_since_last-sr-ntp-time", rb.delay_since_last_sr())
|
||||||
|
.build()
|
||||||
|
},
|
||||||
|
));
|
||||||
|
match rbs.len() {
|
||||||
|
0 => (),
|
||||||
|
1 => {
|
||||||
|
source_stats =
|
||||||
|
source_stats.field("report-blocks", rbs.first().unwrap().clone());
|
||||||
|
}
|
||||||
|
_ => {
|
||||||
|
source_stats = source_stats.field("report-blocks", rbs);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
session_stats = session_stats.field(rs.ssrc().to_string(), source_stats.build());
|
||||||
|
} else if let Some(rr) = self.session.remote_receive_source_by_ssrc(ssrc) {
|
||||||
|
let source_stats = gst::Structure::builder("application/x-rtpbin2-source-stats")
|
||||||
|
.field("ssrc", rr.ssrc())
|
||||||
|
.field("sender", false)
|
||||||
|
.field("local", false)
|
||||||
|
.build();
|
||||||
|
session_stats = session_stats.field(rr.ssrc().to_string(), source_stats);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
session_stats.build()
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
pub fn pt_clock_rate_from_caps(caps: &gst::CapsRef) -> Option<(u8, u32)> {
|
||||||
|
let Some(s) = caps.structure(0) else {
|
||||||
|
gst::debug!(CAT, "no structure!");
|
||||||
|
return None;
|
||||||
|
};
|
||||||
|
let Some((clock_rate, pt)) = Option::zip(
|
||||||
|
s.get::<i32>("clock-rate").ok(),
|
||||||
|
s.get::<i32>("payload").ok(),
|
||||||
|
) else {
|
||||||
|
gst::debug!(
|
||||||
|
CAT,
|
||||||
|
"could not retrieve clock-rate and/or payload from structure"
|
||||||
|
);
|
||||||
|
return None;
|
||||||
|
};
|
||||||
|
if (0..=127).contains(&pt) && clock_rate > 0 {
|
||||||
|
Some((pt as u8, clock_rate as u32))
|
||||||
|
} else {
|
||||||
|
gst::debug!(
|
||||||
|
CAT,
|
||||||
|
"payload value {pt} out of bounds or clock-rate {clock_rate} out of bounds"
|
||||||
|
);
|
||||||
|
None
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
static RUST_CAT: Lazy<gst::DebugCategory> = Lazy::new(|| {
|
||||||
|
gst::DebugCategory::new(
|
||||||
|
"rust-log",
|
||||||
|
gst::DebugColorFlags::empty(),
|
||||||
|
Some("Logs from rust crates"),
|
||||||
|
)
|
||||||
|
});
|
||||||
|
|
||||||
|
static GST_RUST_LOGGER_ONCE: once_cell::sync::OnceCell<()> = once_cell::sync::OnceCell::new();
|
||||||
|
static GST_RUST_LOGGER: GstRustLogger = GstRustLogger {};
|
||||||
|
|
||||||
|
pub(crate) struct GstRustLogger {}
|
||||||
|
|
||||||
|
impl GstRustLogger {
|
||||||
|
pub fn install() {
|
||||||
|
GST_RUST_LOGGER_ONCE.get_or_init(|| {
|
||||||
|
if log::set_logger(&GST_RUST_LOGGER).is_err() {
|
||||||
|
gst::warning!(
|
||||||
|
RUST_CAT,
|
||||||
|
"Cannot install log->gst logger, already installed?"
|
||||||
|
);
|
||||||
|
} else {
|
||||||
|
log::set_max_level(GstRustLogger::debug_level_to_log_level_filter(
|
||||||
|
RUST_CAT.threshold(),
|
||||||
|
));
|
||||||
|
gst::info!(RUST_CAT, "installed log->gst logger");
|
||||||
|
}
|
||||||
|
});
|
||||||
|
}
|
||||||
|
|
||||||
|
fn debug_level_to_log_level_filter(level: gst::DebugLevel) -> log::LevelFilter {
|
||||||
|
match level {
|
||||||
|
gst::DebugLevel::None => log::LevelFilter::Off,
|
||||||
|
gst::DebugLevel::Error => log::LevelFilter::Error,
|
||||||
|
gst::DebugLevel::Warning => log::LevelFilter::Warn,
|
||||||
|
gst::DebugLevel::Fixme | gst::DebugLevel::Info => log::LevelFilter::Info,
|
||||||
|
gst::DebugLevel::Debug => log::LevelFilter::Debug,
|
||||||
|
gst::DebugLevel::Log | gst::DebugLevel::Trace | gst::DebugLevel::Memdump => {
|
||||||
|
log::LevelFilter::Trace
|
||||||
|
}
|
||||||
|
_ => log::LevelFilter::Trace,
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
fn log_level_to_debug_level(level: log::Level) -> gst::DebugLevel {
|
||||||
|
match level {
|
||||||
|
log::Level::Error => gst::DebugLevel::Error,
|
||||||
|
log::Level::Warn => gst::DebugLevel::Warning,
|
||||||
|
log::Level::Info => gst::DebugLevel::Info,
|
||||||
|
log::Level::Debug => gst::DebugLevel::Debug,
|
||||||
|
log::Level::Trace => gst::DebugLevel::Trace,
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
impl log::Log for GstRustLogger {
|
||||||
|
fn enabled(&self, metadata: &log::Metadata) -> bool {
|
||||||
|
RUST_CAT.above_threshold(GstRustLogger::log_level_to_debug_level(metadata.level()))
|
||||||
|
}
|
||||||
|
|
||||||
|
fn log(&self, record: &log::Record) {
|
||||||
|
let gst_level = GstRustLogger::log_level_to_debug_level(record.metadata().level());
|
||||||
|
let file = record
|
||||||
|
.file()
|
||||||
|
.map(glib::GString::from)
|
||||||
|
.unwrap_or_else(|| glib::GString::from("rust-log"));
|
||||||
|
let function = record.target();
|
||||||
|
let line = record.line().unwrap_or(0);
|
||||||
|
RUST_CAT.log(
|
||||||
|
None::<&glib::Object>,
|
||||||
|
gst_level,
|
||||||
|
file.as_gstr(),
|
||||||
|
function,
|
||||||
|
line,
|
||||||
|
*record.args(),
|
||||||
|
);
|
||||||
|
}
|
||||||
|
|
||||||
|
fn flush(&self) {}
|
||||||
|
}
|
||||||
|
|
@ -4,15 +4,20 @@ use gst::glib;
|
||||||
use gst::prelude::*;
|
use gst::prelude::*;
|
||||||
use once_cell::sync::Lazy;
|
use once_cell::sync::Lazy;
|
||||||
mod config;
|
mod config;
|
||||||
mod imp;
|
mod internal;
|
||||||
mod jitterbuffer;
|
mod jitterbuffer;
|
||||||
|
mod rtprecv;
|
||||||
|
mod rtpsend;
|
||||||
mod session;
|
mod session;
|
||||||
mod source;
|
mod source;
|
||||||
mod sync;
|
mod sync;
|
||||||
mod time;
|
mod time;
|
||||||
|
|
||||||
glib::wrapper! {
|
glib::wrapper! {
|
||||||
pub struct RtpBin2(ObjectSubclass<imp::RtpBin2>) @extends gst::Element, gst::Object;
|
pub struct RtpSend(ObjectSubclass<rtpsend::RtpSend>) @extends gst::Element, gst::Object;
|
||||||
|
}
|
||||||
|
glib::wrapper! {
|
||||||
|
pub struct RtpRecv(ObjectSubclass<rtprecv::RtpRecv>) @extends gst::Element, gst::Object;
|
||||||
}
|
}
|
||||||
|
|
||||||
pub fn register(plugin: &gst::Plugin) -> Result<(), glib::BoolError> {
|
pub fn register(plugin: &gst::Plugin) -> Result<(), glib::BoolError> {
|
||||||
|
|
@ -22,12 +27,20 @@ pub fn register(plugin: &gst::Plugin) -> Result<(), glib::BoolError> {
|
||||||
.mark_as_plugin_api(gst::PluginAPIFlags::empty());
|
.mark_as_plugin_api(gst::PluginAPIFlags::empty());
|
||||||
crate::rtpbin2::config::Rtp2Session::static_type()
|
crate::rtpbin2::config::Rtp2Session::static_type()
|
||||||
.mark_as_plugin_api(gst::PluginAPIFlags::empty());
|
.mark_as_plugin_api(gst::PluginAPIFlags::empty());
|
||||||
|
crate::rtpbin2::rtpsend::Profile::static_type()
|
||||||
|
.mark_as_plugin_api(gst::PluginAPIFlags::empty());
|
||||||
}
|
}
|
||||||
gst::Element::register(
|
gst::Element::register(
|
||||||
Some(plugin),
|
Some(plugin),
|
||||||
"rtpbin2",
|
"rtpsend",
|
||||||
gst::Rank::NONE,
|
gst::Rank::NONE,
|
||||||
RtpBin2::static_type(),
|
RtpSend::static_type(),
|
||||||
|
)?;
|
||||||
|
gst::Element::register(
|
||||||
|
Some(plugin),
|
||||||
|
"rtprecv",
|
||||||
|
gst::Rank::NONE,
|
||||||
|
RtpRecv::static_type(),
|
||||||
)
|
)
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|
|
||||||
File diff suppressed because it is too large
Load diff
878
net/rtp/src/rtpbin2/rtpsend.rs
Normal file
878
net/rtp/src/rtpbin2/rtpsend.rs
Normal file
|
|
@ -0,0 +1,878 @@
|
||||||
|
// SPDX-License-Identifier: MPL-2.0
|
||||||
|
|
||||||
|
use std::collections::HashMap;
|
||||||
|
use std::pin::Pin;
|
||||||
|
use std::sync::{Arc, Mutex};
|
||||||
|
use std::task::Poll;
|
||||||
|
use std::time::{Duration, Instant, SystemTime};
|
||||||
|
|
||||||
|
use futures::future::{AbortHandle, Abortable};
|
||||||
|
use futures::StreamExt;
|
||||||
|
use gst::{glib, prelude::*, subclass::prelude::*};
|
||||||
|
use once_cell::sync::Lazy;
|
||||||
|
|
||||||
|
use super::internal::{pt_clock_rate_from_caps, GstRustLogger, SharedRtpState, SharedSession};
|
||||||
|
use super::session::{RtcpSendReply, RtpProfile, SendReply, RTCP_MIN_REPORT_INTERVAL};
|
||||||
|
use super::source::SourceState;
|
||||||
|
|
||||||
|
use crate::rtpbin2::RUNTIME;
|
||||||
|
|
||||||
|
const DEFAULT_MIN_RTCP_INTERVAL: Duration = RTCP_MIN_REPORT_INTERVAL;
|
||||||
|
const DEFAULT_REDUCED_SIZE_RTCP: bool = false;
|
||||||
|
|
||||||
|
static CAT: Lazy<gst::DebugCategory> = Lazy::new(|| {
|
||||||
|
gst::DebugCategory::new(
|
||||||
|
"rtpsend",
|
||||||
|
gst::DebugColorFlags::empty(),
|
||||||
|
Some("RTP Sending"),
|
||||||
|
)
|
||||||
|
});
|
||||||
|
|
||||||
|
#[derive(Debug, Default, Clone, Copy, PartialEq, Eq, glib::Enum)]
|
||||||
|
#[repr(u32)]
|
||||||
|
#[enum_type(name = "GstRtpSendProfile")]
|
||||||
|
pub enum Profile {
|
||||||
|
#[default]
|
||||||
|
#[enum_value(name = "AVP profile as specified in RFC 3550", nick = "avp")]
|
||||||
|
Avp,
|
||||||
|
#[enum_value(name = "AVPF profile as specified in RFC 4585", nick = "avpf")]
|
||||||
|
Avpf,
|
||||||
|
}
|
||||||
|
|
||||||
|
impl From<RtpProfile> for Profile {
|
||||||
|
fn from(value: RtpProfile) -> Self {
|
||||||
|
match value {
|
||||||
|
RtpProfile::Avp => Self::Avp,
|
||||||
|
RtpProfile::Avpf => Self::Avpf,
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
impl From<Profile> for RtpProfile {
|
||||||
|
fn from(value: Profile) -> Self {
|
||||||
|
match value {
|
||||||
|
Profile::Avp => Self::Avp,
|
||||||
|
Profile::Avpf => Self::Avpf,
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
#[derive(Debug, Clone)]
|
||||||
|
struct Settings {
|
||||||
|
rtp_id: String,
|
||||||
|
min_rtcp_interval: Duration,
|
||||||
|
profile: Profile,
|
||||||
|
reduced_size_rtcp: bool,
|
||||||
|
}
|
||||||
|
|
||||||
|
impl Default for Settings {
|
||||||
|
fn default() -> Self {
|
||||||
|
Settings {
|
||||||
|
rtp_id: String::from("rtp-id"),
|
||||||
|
min_rtcp_interval: DEFAULT_MIN_RTCP_INTERVAL,
|
||||||
|
profile: Profile::default(),
|
||||||
|
reduced_size_rtcp: DEFAULT_REDUCED_SIZE_RTCP,
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
#[derive(Debug)]
|
||||||
|
#[must_use = "futures/streams/sinks do nothing unless you `.await` or poll them"]
|
||||||
|
struct RtcpSendStream {
|
||||||
|
state: Arc<Mutex<State>>,
|
||||||
|
session_id: usize,
|
||||||
|
sleep: Pin<Box<tokio::time::Sleep>>,
|
||||||
|
}
|
||||||
|
|
||||||
|
impl RtcpSendStream {
|
||||||
|
fn new(state: Arc<Mutex<State>>, session_id: usize) -> Self {
|
||||||
|
Self {
|
||||||
|
state,
|
||||||
|
session_id,
|
||||||
|
sleep: Box::pin(tokio::time::sleep(Duration::from_secs(1))),
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
impl futures::stream::Stream for RtcpSendStream {
|
||||||
|
type Item = RtcpSendReply;
|
||||||
|
|
||||||
|
fn poll_next(
|
||||||
|
self: Pin<&mut Self>,
|
||||||
|
cx: &mut std::task::Context<'_>,
|
||||||
|
) -> std::task::Poll<Option<Self::Item>> {
|
||||||
|
let mut state = self.state.lock().unwrap();
|
||||||
|
let now = Instant::now();
|
||||||
|
let ntp_now = SystemTime::now();
|
||||||
|
let mut lowest_wait = None;
|
||||||
|
if let Some(session) = state.mut_session_by_id(self.session_id) {
|
||||||
|
let mut session_inner = session.internal_session.inner.lock().unwrap();
|
||||||
|
if let Some(reply) = session_inner.session.poll_rtcp_send(now, ntp_now) {
|
||||||
|
return Poll::Ready(Some(reply));
|
||||||
|
}
|
||||||
|
if let Some(wait) = session_inner.session.poll_rtcp_send_timeout(now) {
|
||||||
|
if lowest_wait.map_or(true, |lowest_wait| wait < lowest_wait) {
|
||||||
|
lowest_wait = Some(wait);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
session_inner.rtcp_waker = Some(cx.waker().clone());
|
||||||
|
}
|
||||||
|
drop(state);
|
||||||
|
|
||||||
|
// default to the minimum initial rtcp delay so we don't busy loop if there are no sessions or no
|
||||||
|
// timeouts available
|
||||||
|
let lowest_wait =
|
||||||
|
lowest_wait.unwrap_or(now + crate::rtpbin2::session::RTCP_MIN_REPORT_INTERVAL / 2);
|
||||||
|
let this = self.get_mut();
|
||||||
|
this.sleep.as_mut().reset(lowest_wait.into());
|
||||||
|
if !std::future::Future::poll(this.sleep.as_mut(), cx).is_pending() {
|
||||||
|
// wake us again if the delay is not pending for another go at finding the next timeout
|
||||||
|
// value
|
||||||
|
cx.waker().wake_by_ref();
|
||||||
|
}
|
||||||
|
Poll::Pending
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
#[derive(Debug)]
|
||||||
|
struct SendSession {
|
||||||
|
internal_session: SharedSession,
|
||||||
|
|
||||||
|
rtcp_task: Mutex<Option<RtcpTask>>,
|
||||||
|
|
||||||
|
// State for sending RTP streams
|
||||||
|
rtp_send_sinkpad: Option<gst::Pad>,
|
||||||
|
rtp_send_srcpad: Option<gst::Pad>,
|
||||||
|
|
||||||
|
rtcp_send_srcpad: Option<gst::Pad>,
|
||||||
|
}
|
||||||
|
|
||||||
|
impl SendSession {
|
||||||
|
fn new(shared_state: &SharedRtpState, id: usize, settings: &Settings) -> Self {
|
||||||
|
let internal_session = shared_state.session_get_or_init(id, || {
|
||||||
|
SharedSession::new(
|
||||||
|
id,
|
||||||
|
settings.profile.into(),
|
||||||
|
settings.min_rtcp_interval,
|
||||||
|
settings.reduced_size_rtcp,
|
||||||
|
)
|
||||||
|
});
|
||||||
|
let mut inner = internal_session.inner.lock().unwrap();
|
||||||
|
inner.session.set_profile(settings.profile.into());
|
||||||
|
inner
|
||||||
|
.session
|
||||||
|
.set_min_rtcp_interval(settings.min_rtcp_interval);
|
||||||
|
inner
|
||||||
|
.session
|
||||||
|
.set_reduced_size_rtcp(settings.reduced_size_rtcp);
|
||||||
|
drop(inner);
|
||||||
|
|
||||||
|
Self {
|
||||||
|
internal_session,
|
||||||
|
|
||||||
|
rtcp_task: Mutex::new(None),
|
||||||
|
rtp_send_sinkpad: None,
|
||||||
|
rtp_send_srcpad: None,
|
||||||
|
rtcp_send_srcpad: None,
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
fn start_rtcp_task(&self, state: Arc<Mutex<State>>) {
|
||||||
|
let mut rtcp_task = self.rtcp_task.lock().unwrap();
|
||||||
|
|
||||||
|
if rtcp_task.is_some() {
|
||||||
|
return;
|
||||||
|
}
|
||||||
|
|
||||||
|
// run the runtime from another task to prevent the "start a runtime from within a runtime" panic
|
||||||
|
// when the plugin is statically linked.
|
||||||
|
let (abort_handle, abort_registration) = AbortHandle::new_pair();
|
||||||
|
let session_id = self.internal_session.id;
|
||||||
|
RUNTIME.spawn(async move {
|
||||||
|
let future = Abortable::new(Self::rtcp_task(state, session_id), abort_registration);
|
||||||
|
future.await
|
||||||
|
});
|
||||||
|
|
||||||
|
rtcp_task.replace(RtcpTask { abort_handle });
|
||||||
|
}
|
||||||
|
|
||||||
|
async fn rtcp_task(state: Arc<Mutex<State>>, session_id: usize) {
|
||||||
|
let mut stream = RtcpSendStream::new(state.clone(), session_id);
|
||||||
|
while let Some(reply) = stream.next().await {
|
||||||
|
let state = state.lock().unwrap();
|
||||||
|
let Some(session) = state.session_by_id(session_id) else {
|
||||||
|
continue;
|
||||||
|
};
|
||||||
|
match reply {
|
||||||
|
RtcpSendReply::Data(data) => {
|
||||||
|
let Some(rtcp_srcpad) = session.rtcp_send_srcpad.clone() else {
|
||||||
|
continue;
|
||||||
|
};
|
||||||
|
drop(state);
|
||||||
|
RUNTIME.spawn_blocking(move || {
|
||||||
|
let buffer = gst::Buffer::from_mut_slice(data);
|
||||||
|
if let Err(e) = rtcp_srcpad.push(buffer) {
|
||||||
|
gst::warning!(CAT, obj: rtcp_srcpad, "Failed to send rtcp data: flow return {e:?}");
|
||||||
|
}
|
||||||
|
});
|
||||||
|
}
|
||||||
|
RtcpSendReply::SsrcBye(ssrc) => session
|
||||||
|
.internal_session
|
||||||
|
.config
|
||||||
|
.emit_by_name::<()>("bye-ssrc", &[&ssrc]),
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
fn stop_rtcp_task(&self) {
|
||||||
|
let mut rtcp_task = self.rtcp_task.lock().unwrap();
|
||||||
|
|
||||||
|
if let Some(rtcp) = rtcp_task.take() {
|
||||||
|
rtcp.abort_handle.abort();
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
#[derive(Debug, Default)]
|
||||||
|
struct State {
|
||||||
|
shared_state: Option<SharedRtpState>,
|
||||||
|
sessions: Vec<SendSession>,
|
||||||
|
max_session_id: usize,
|
||||||
|
pads_session_id_map: HashMap<gst::Pad, usize>,
|
||||||
|
}
|
||||||
|
|
||||||
|
impl State {
|
||||||
|
fn session_by_id(&self, id: usize) -> Option<&SendSession> {
|
||||||
|
self.sessions
|
||||||
|
.iter()
|
||||||
|
.find(|session| session.internal_session.id == id)
|
||||||
|
}
|
||||||
|
|
||||||
|
fn mut_session_by_id(&mut self, id: usize) -> Option<&mut SendSession> {
|
||||||
|
self.sessions
|
||||||
|
.iter_mut()
|
||||||
|
.find(|session| session.internal_session.id == id)
|
||||||
|
}
|
||||||
|
|
||||||
|
fn stats(&self) -> gst::Structure {
|
||||||
|
let mut ret = gst::Structure::builder("application/x-rtp2-stats");
|
||||||
|
for session in self.sessions.iter() {
|
||||||
|
let sess_id = session.internal_session.id;
|
||||||
|
let session = session.internal_session.inner.lock().unwrap();
|
||||||
|
|
||||||
|
ret = ret.field(sess_id.to_string(), session.stats());
|
||||||
|
}
|
||||||
|
ret.build()
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
pub struct RtpSend {
|
||||||
|
settings: Mutex<Settings>,
|
||||||
|
state: Arc<Mutex<State>>,
|
||||||
|
}
|
||||||
|
|
||||||
|
#[derive(Debug)]
|
||||||
|
struct RtcpTask {
|
||||||
|
abort_handle: AbortHandle,
|
||||||
|
}
|
||||||
|
|
||||||
|
impl RtpSend {
|
||||||
|
fn iterate_internal_links(&self, pad: &gst::Pad) -> gst::Iterator<gst::Pad> {
|
||||||
|
let state = self.state.lock().unwrap();
|
||||||
|
if let Some(&id) = state.pads_session_id_map.get(pad) {
|
||||||
|
if let Some(session) = state.session_by_id(id) {
|
||||||
|
if let Some(ref sinkpad) = session.rtp_send_sinkpad {
|
||||||
|
if let Some(ref srcpad) = session.rtp_send_srcpad {
|
||||||
|
if sinkpad == pad {
|
||||||
|
return gst::Iterator::from_vec(vec![srcpad.clone()]);
|
||||||
|
} else if srcpad == pad {
|
||||||
|
return gst::Iterator::from_vec(vec![sinkpad.clone()]);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
// nothing to do for rtcp pads
|
||||||
|
}
|
||||||
|
}
|
||||||
|
gst::Iterator::from_vec(vec![])
|
||||||
|
}
|
||||||
|
|
||||||
|
fn rtp_send_sink_chain(
|
||||||
|
&self,
|
||||||
|
id: usize,
|
||||||
|
buffer: gst::Buffer,
|
||||||
|
) -> Result<gst::FlowSuccess, gst::FlowError> {
|
||||||
|
let state = self.state.lock().unwrap();
|
||||||
|
let Some(session) = state.session_by_id(id) else {
|
||||||
|
gst::error!(CAT, "No session?");
|
||||||
|
return Err(gst::FlowError::Error);
|
||||||
|
};
|
||||||
|
|
||||||
|
let mapped = buffer.map_readable().map_err(|e| {
|
||||||
|
gst::error!(CAT, imp: self, "Failed to map input buffer {e:?}");
|
||||||
|
gst::FlowError::Error
|
||||||
|
})?;
|
||||||
|
let rtp = match rtp_types::RtpPacket::parse(&mapped) {
|
||||||
|
Ok(rtp) => rtp,
|
||||||
|
Err(e) => {
|
||||||
|
gst::error!(CAT, imp: self, "Failed to parse input as valid rtp packet: {e:?}");
|
||||||
|
return Ok(gst::FlowSuccess::Ok);
|
||||||
|
}
|
||||||
|
};
|
||||||
|
|
||||||
|
let srcpad = session.rtp_send_srcpad.clone().unwrap();
|
||||||
|
let session = session.internal_session.clone();
|
||||||
|
let mut session_inner = session.inner.lock().unwrap();
|
||||||
|
drop(state);
|
||||||
|
|
||||||
|
let now = Instant::now();
|
||||||
|
loop {
|
||||||
|
match session_inner.session.handle_send(&rtp, now) {
|
||||||
|
SendReply::SsrcCollision(_ssrc) => (), // TODO: handle ssrc collision
|
||||||
|
SendReply::NewSsrc(ssrc, _pt) => {
|
||||||
|
drop(session_inner);
|
||||||
|
session.config.emit_by_name::<()>("new-ssrc", &[&ssrc]);
|
||||||
|
session_inner = session.inner.lock().unwrap();
|
||||||
|
}
|
||||||
|
SendReply::Passthrough => break,
|
||||||
|
SendReply::Drop => return Ok(gst::FlowSuccess::Ok),
|
||||||
|
}
|
||||||
|
}
|
||||||
|
// TODO: handle other processing
|
||||||
|
drop(mapped);
|
||||||
|
drop(session_inner);
|
||||||
|
srcpad.push(buffer)
|
||||||
|
}
|
||||||
|
|
||||||
|
fn rtp_send_sink_event(&self, pad: &gst::Pad, event: gst::Event, id: usize) -> bool {
|
||||||
|
match event.view() {
|
||||||
|
gst::EventView::Caps(caps) => {
|
||||||
|
if let Some((pt, clock_rate)) = pt_clock_rate_from_caps(caps.caps()) {
|
||||||
|
let state = self.state.lock().unwrap();
|
||||||
|
if let Some(session) = state.session_by_id(id) {
|
||||||
|
let mut session = session.internal_session.inner.lock().unwrap();
|
||||||
|
session.session.set_pt_clock_rate(pt, clock_rate);
|
||||||
|
session.add_caps(caps.caps_owned());
|
||||||
|
}
|
||||||
|
} else {
|
||||||
|
gst::warning!(CAT, obj: pad, "input caps are missing payload or clock-rate fields");
|
||||||
|
}
|
||||||
|
gst::Pad::event_default(pad, Some(&*self.obj()), event)
|
||||||
|
}
|
||||||
|
gst::EventView::Eos(_eos) => {
|
||||||
|
let now = Instant::now();
|
||||||
|
let state = self.state.lock().unwrap();
|
||||||
|
if let Some(session) = state.session_by_id(id) {
|
||||||
|
let mut session = session.internal_session.inner.lock().unwrap();
|
||||||
|
let ssrcs = session.session.ssrcs().collect::<Vec<_>>();
|
||||||
|
// We want to bye all relevant ssrc's here.
|
||||||
|
// Relevant means they will not be used by something else which means that any
|
||||||
|
// local send ssrc that is not being used for Sr/Rr reports (internal_ssrc) can
|
||||||
|
// have the Bye state applied.
|
||||||
|
let mut all_local = true;
|
||||||
|
let internal_ssrc = session.session.internal_ssrc();
|
||||||
|
for ssrc in ssrcs {
|
||||||
|
let Some(local_send) = session.session.mut_local_send_source_by_ssrc(ssrc)
|
||||||
|
else {
|
||||||
|
if let Some(local_recv) =
|
||||||
|
session.session.local_receive_source_by_ssrc(ssrc)
|
||||||
|
{
|
||||||
|
if local_recv.state() != SourceState::Bye
|
||||||
|
&& Some(ssrc) != internal_ssrc
|
||||||
|
{
|
||||||
|
all_local = false;
|
||||||
|
}
|
||||||
|
}
|
||||||
|
continue;
|
||||||
|
};
|
||||||
|
if Some(ssrc) != internal_ssrc {
|
||||||
|
local_send.mark_bye("End of Stream")
|
||||||
|
}
|
||||||
|
}
|
||||||
|
if all_local {
|
||||||
|
// if there are no non-local send ssrc's, then we can Bye the entire
|
||||||
|
// session.
|
||||||
|
session.session.schedule_bye("End of Stream", now);
|
||||||
|
}
|
||||||
|
if let Some(waker) = session.rtcp_waker.take() {
|
||||||
|
waker.wake();
|
||||||
|
}
|
||||||
|
drop(session);
|
||||||
|
}
|
||||||
|
drop(state);
|
||||||
|
gst::Pad::event_default(pad, Some(&*self.obj()), event)
|
||||||
|
}
|
||||||
|
_ => gst::Pad::event_default(pad, Some(&*self.obj()), event),
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
#[glib::object_subclass]
|
||||||
|
impl ObjectSubclass for RtpSend {
|
||||||
|
const NAME: &'static str = "GstRtpSend";
|
||||||
|
type Type = super::RtpSend;
|
||||||
|
type ParentType = gst::Element;
|
||||||
|
|
||||||
|
fn new() -> Self {
|
||||||
|
GstRustLogger::install();
|
||||||
|
Self {
|
||||||
|
settings: Default::default(),
|
||||||
|
state: Default::default(),
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
impl ObjectImpl for RtpSend {
|
||||||
|
fn properties() -> &'static [glib::ParamSpec] {
|
||||||
|
static PROPERTIES: Lazy<Vec<glib::ParamSpec>> = Lazy::new(|| {
|
||||||
|
vec![
|
||||||
|
glib::ParamSpecString::builder("rtp-id")
|
||||||
|
.nick("The RTP Connection ID")
|
||||||
|
.blurb("A connection ID shared with a rtprecv element for implementing both sending and receiving using the same RTP context")
|
||||||
|
.default_value("rtp-id")
|
||||||
|
.build(),
|
||||||
|
glib::ParamSpecUInt::builder("min-rtcp-interval")
|
||||||
|
.nick("Minimum RTCP interval in ms")
|
||||||
|
.blurb("Minimum time (in ms) between RTCP reports")
|
||||||
|
.default_value(DEFAULT_MIN_RTCP_INTERVAL.as_millis() as u32)
|
||||||
|
.mutable_ready()
|
||||||
|
.build(),
|
||||||
|
glib::ParamSpecUInt::builder("stats")
|
||||||
|
.nick("Statistics")
|
||||||
|
.blurb("Statistics about the session")
|
||||||
|
.read_only()
|
||||||
|
.build(),
|
||||||
|
glib::ParamSpecEnum::builder::<Profile>("rtp-profile")
|
||||||
|
.nick("RTP Profile")
|
||||||
|
.blurb("RTP Profile to use")
|
||||||
|
.default_value(Profile::default())
|
||||||
|
.mutable_ready()
|
||||||
|
.build(),
|
||||||
|
glib::ParamSpecBoolean::builder("reduced-size-rtcp")
|
||||||
|
.nick("Reduced Size RTCP")
|
||||||
|
.blurb("Use reduced size RTCP. Only has an effect if rtp-profile=avpf")
|
||||||
|
.default_value(DEFAULT_REDUCED_SIZE_RTCP)
|
||||||
|
.mutable_ready()
|
||||||
|
.build(),
|
||||||
|
]
|
||||||
|
});
|
||||||
|
|
||||||
|
PROPERTIES.as_ref()
|
||||||
|
}
|
||||||
|
|
||||||
|
fn set_property(&self, _id: usize, value: &glib::Value, pspec: &glib::ParamSpec) {
|
||||||
|
match pspec.name() {
|
||||||
|
"rtp-id" => {
|
||||||
|
let mut settings = self.settings.lock().unwrap();
|
||||||
|
settings.rtp_id = value.get::<String>().expect("type checked upstream");
|
||||||
|
}
|
||||||
|
"min-rtcp-interval" => {
|
||||||
|
let mut settings = self.settings.lock().unwrap();
|
||||||
|
settings.min_rtcp_interval = Duration::from_millis(
|
||||||
|
value.get::<u32>().expect("type checked upstream").into(),
|
||||||
|
);
|
||||||
|
}
|
||||||
|
"rtp-profile" => {
|
||||||
|
let mut settings = self.settings.lock().unwrap();
|
||||||
|
settings.profile = value.get::<Profile>().expect("Type checked upstream");
|
||||||
|
}
|
||||||
|
"reduced-size-rtcp" => {
|
||||||
|
let mut settings = self.settings.lock().unwrap();
|
||||||
|
settings.reduced_size_rtcp = value.get::<bool>().expect("Type checked upstream");
|
||||||
|
}
|
||||||
|
_ => unimplemented!(),
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
fn property(&self, _id: usize, pspec: &glib::ParamSpec) -> glib::Value {
|
||||||
|
match pspec.name() {
|
||||||
|
"rtp-id" => {
|
||||||
|
let settings = self.settings.lock().unwrap();
|
||||||
|
settings.rtp_id.to_value()
|
||||||
|
}
|
||||||
|
"min-rtcp-interval" => {
|
||||||
|
let settings = self.settings.lock().unwrap();
|
||||||
|
(settings.min_rtcp_interval.as_millis() as u32).to_value()
|
||||||
|
}
|
||||||
|
"stats" => {
|
||||||
|
let state = self.state.lock().unwrap();
|
||||||
|
state.stats().to_value()
|
||||||
|
}
|
||||||
|
"rtp-profile" => {
|
||||||
|
let settings = self.settings.lock().unwrap();
|
||||||
|
settings.profile.to_value()
|
||||||
|
}
|
||||||
|
"reduced-size-rtcp" => {
|
||||||
|
let settings = self.settings.lock().unwrap();
|
||||||
|
settings.reduced_size_rtcp.to_value()
|
||||||
|
}
|
||||||
|
_ => unimplemented!(),
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
fn signals() -> &'static [glib::subclass::Signal] {
|
||||||
|
static SIGNALS: Lazy<Vec<glib::subclass::Signal>> = Lazy::new(|| {
|
||||||
|
vec![glib::subclass::Signal::builder("get-session")
|
||||||
|
.param_types([u32::static_type()])
|
||||||
|
.return_type::<crate::rtpbin2::config::Rtp2Session>()
|
||||||
|
.action()
|
||||||
|
.class_handler(|_token, args| {
|
||||||
|
let element = args[0].get::<super::RtpSend>().expect("signal arg");
|
||||||
|
let id = args[1].get::<u32>().expect("signal arg");
|
||||||
|
let send = element.imp();
|
||||||
|
let state = send.state.lock().unwrap();
|
||||||
|
state
|
||||||
|
.session_by_id(id as usize)
|
||||||
|
.map(|sess| sess.internal_session.config.to_value())
|
||||||
|
})
|
||||||
|
.build()]
|
||||||
|
});
|
||||||
|
|
||||||
|
SIGNALS.as_ref()
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
impl GstObjectImpl for RtpSend {}
|
||||||
|
|
||||||
|
impl ElementImpl for RtpSend {
|
||||||
|
fn metadata() -> Option<&'static gst::subclass::ElementMetadata> {
|
||||||
|
static ELEMENT_METADATA: Lazy<gst::subclass::ElementMetadata> = Lazy::new(|| {
|
||||||
|
gst::subclass::ElementMetadata::new(
|
||||||
|
"RTP Session Sender",
|
||||||
|
"Network/RTP/Filter",
|
||||||
|
"RTP session management (sender)",
|
||||||
|
"Matthew Waters <matthew@centricular.com>",
|
||||||
|
)
|
||||||
|
});
|
||||||
|
|
||||||
|
Some(&*ELEMENT_METADATA)
|
||||||
|
}
|
||||||
|
|
||||||
|
fn pad_templates() -> &'static [gst::PadTemplate] {
|
||||||
|
static PAD_TEMPLATES: Lazy<Vec<gst::PadTemplate>> = Lazy::new(|| {
|
||||||
|
let rtp_caps = gst::Caps::builder_full()
|
||||||
|
.structure(gst::Structure::builder("application/x-rtp").build())
|
||||||
|
.build();
|
||||||
|
let rtcp_caps = gst::Caps::builder_full()
|
||||||
|
.structure(gst::Structure::builder("application/x-rtcp").build())
|
||||||
|
.build();
|
||||||
|
|
||||||
|
vec![
|
||||||
|
gst::PadTemplate::new(
|
||||||
|
"rtp_sink_%u",
|
||||||
|
gst::PadDirection::Sink,
|
||||||
|
gst::PadPresence::Request,
|
||||||
|
&rtp_caps,
|
||||||
|
)
|
||||||
|
.unwrap(),
|
||||||
|
gst::PadTemplate::new(
|
||||||
|
"rtp_src_%u",
|
||||||
|
gst::PadDirection::Src,
|
||||||
|
gst::PadPresence::Sometimes,
|
||||||
|
&rtp_caps,
|
||||||
|
)
|
||||||
|
.unwrap(),
|
||||||
|
gst::PadTemplate::new(
|
||||||
|
"rtcp_src_%u",
|
||||||
|
gst::PadDirection::Src,
|
||||||
|
gst::PadPresence::Request,
|
||||||
|
&rtcp_caps,
|
||||||
|
)
|
||||||
|
.unwrap(),
|
||||||
|
]
|
||||||
|
});
|
||||||
|
|
||||||
|
PAD_TEMPLATES.as_ref()
|
||||||
|
}
|
||||||
|
|
||||||
|
fn request_new_pad(
|
||||||
|
&self,
|
||||||
|
templ: &gst::PadTemplate,
|
||||||
|
name: Option<&str>,
|
||||||
|
_caps: Option<&gst::Caps>, // XXX: do something with caps?
|
||||||
|
) -> Option<gst::Pad> {
|
||||||
|
let settings = self.settings.lock().unwrap().clone();
|
||||||
|
let state_clone = self.state.clone();
|
||||||
|
let mut state = self.state.lock().unwrap();
|
||||||
|
let max_session_id = state.max_session_id;
|
||||||
|
let rtp_id = settings.rtp_id.clone();
|
||||||
|
|
||||||
|
// parse the possibly provided name into a session id or use the default
|
||||||
|
let sess_parse = move |name: Option<&str>, prefix, default_id| -> Option<usize> {
|
||||||
|
if let Some(name) = name {
|
||||||
|
name.strip_prefix(prefix).and_then(|suffix| {
|
||||||
|
if suffix.starts_with("%u") {
|
||||||
|
Some(default_id)
|
||||||
|
} else {
|
||||||
|
suffix.parse::<usize>().ok()
|
||||||
|
}
|
||||||
|
})
|
||||||
|
} else {
|
||||||
|
Some(default_id)
|
||||||
|
}
|
||||||
|
};
|
||||||
|
|
||||||
|
match templ.name_template() {
|
||||||
|
"rtp_sink_%u" => sess_parse(name, "rtp_sink_", max_session_id).and_then(|id| {
|
||||||
|
let new_pad = move |session: &mut SendSession| -> Option<(
|
||||||
|
gst::Pad,
|
||||||
|
Option<gst::Pad>,
|
||||||
|
usize,
|
||||||
|
Vec<gst::Event>,
|
||||||
|
)> {
|
||||||
|
let sinkpad = gst::Pad::builder_from_template(templ)
|
||||||
|
.chain_function(move |_pad, parent, buffer| {
|
||||||
|
RtpSend::catch_panic_pad_function(
|
||||||
|
parent,
|
||||||
|
|| Err(gst::FlowError::Error),
|
||||||
|
|this| this.rtp_send_sink_chain(id, buffer),
|
||||||
|
)
|
||||||
|
})
|
||||||
|
.iterate_internal_links_function(|pad, parent| {
|
||||||
|
RtpSend::catch_panic_pad_function(
|
||||||
|
parent,
|
||||||
|
|| gst::Iterator::from_vec(vec![]),
|
||||||
|
|this| this.iterate_internal_links(pad),
|
||||||
|
)
|
||||||
|
})
|
||||||
|
.event_function(move |pad, parent, event| {
|
||||||
|
RtpSend::catch_panic_pad_function(
|
||||||
|
parent,
|
||||||
|
|| false,
|
||||||
|
|this| this.rtp_send_sink_event(pad, event, id),
|
||||||
|
)
|
||||||
|
})
|
||||||
|
.flags(gst::PadFlags::PROXY_CAPS)
|
||||||
|
.name(format!("rtp_sink_{}", id))
|
||||||
|
.build();
|
||||||
|
let src_templ = self.obj().pad_template("rtp_src_%u").unwrap();
|
||||||
|
let srcpad = gst::Pad::builder_from_template(&src_templ)
|
||||||
|
.iterate_internal_links_function(|pad, parent| {
|
||||||
|
RtpSend::catch_panic_pad_function(
|
||||||
|
parent,
|
||||||
|
|| gst::Iterator::from_vec(vec![]),
|
||||||
|
|this| this.iterate_internal_links(pad),
|
||||||
|
)
|
||||||
|
})
|
||||||
|
.name(format!("rtp_src_{}", id))
|
||||||
|
.build();
|
||||||
|
session.rtp_send_sinkpad = Some(sinkpad.clone());
|
||||||
|
session.rtp_send_srcpad = Some(srcpad.clone());
|
||||||
|
session
|
||||||
|
.internal_session
|
||||||
|
.inner
|
||||||
|
.lock()
|
||||||
|
.unwrap()
|
||||||
|
.rtp_send_sinkpad = Some(sinkpad.clone());
|
||||||
|
Some((sinkpad, Some(srcpad), id, vec![]))
|
||||||
|
};
|
||||||
|
|
||||||
|
let session = state.mut_session_by_id(id);
|
||||||
|
if let Some(session) = session {
|
||||||
|
if session.rtp_send_sinkpad.is_some() {
|
||||||
|
None
|
||||||
|
} else {
|
||||||
|
new_pad(session)
|
||||||
|
}
|
||||||
|
} else {
|
||||||
|
let shared_state = state
|
||||||
|
.shared_state
|
||||||
|
.get_or_insert_with(|| SharedRtpState::send_get_or_init(rtp_id));
|
||||||
|
let mut session = SendSession::new(shared_state, id, &settings);
|
||||||
|
let ret = new_pad(&mut session);
|
||||||
|
state.sessions.push(session);
|
||||||
|
ret
|
||||||
|
}
|
||||||
|
}),
|
||||||
|
"rtcp_src_%u" => sess_parse(name, "rtcp_src_", max_session_id).and_then(|id| {
|
||||||
|
let new_pad = move |session: &mut SendSession| -> Option<(
|
||||||
|
gst::Pad,
|
||||||
|
Option<gst::Pad>,
|
||||||
|
usize,
|
||||||
|
Vec<gst::Event>,
|
||||||
|
)> {
|
||||||
|
let srcpad = gst::Pad::builder_from_template(templ)
|
||||||
|
.iterate_internal_links_function(|pad, parent| {
|
||||||
|
RtpSend::catch_panic_pad_function(
|
||||||
|
parent,
|
||||||
|
|| gst::Iterator::from_vec(vec![]),
|
||||||
|
|this| this.iterate_internal_links(pad),
|
||||||
|
)
|
||||||
|
})
|
||||||
|
.name(format!("rtcp_src_{}", id))
|
||||||
|
.build();
|
||||||
|
|
||||||
|
let stream_id = format!("{}/rtcp", id);
|
||||||
|
let stream_start = gst::event::StreamStart::builder(&stream_id).build();
|
||||||
|
let seqnum = stream_start.seqnum();
|
||||||
|
|
||||||
|
let caps = gst::Caps::new_empty_simple("application/x-rtcp");
|
||||||
|
let caps = gst::event::Caps::builder(&caps).seqnum(seqnum).build();
|
||||||
|
|
||||||
|
let segment = gst::FormattedSegment::<gst::ClockTime>::new();
|
||||||
|
let segment = gst::event::Segment::new(&segment);
|
||||||
|
|
||||||
|
session.rtcp_send_srcpad = Some(srcpad.clone());
|
||||||
|
session.start_rtcp_task(state_clone);
|
||||||
|
Some((srcpad, None, id, vec![stream_start, caps, segment]))
|
||||||
|
};
|
||||||
|
|
||||||
|
let session = state.mut_session_by_id(id);
|
||||||
|
if let Some(session) = session {
|
||||||
|
if session.rtcp_send_srcpad.is_some() {
|
||||||
|
None
|
||||||
|
} else {
|
||||||
|
new_pad(session)
|
||||||
|
}
|
||||||
|
} else {
|
||||||
|
let shared_state = state
|
||||||
|
.shared_state
|
||||||
|
.get_or_insert_with(|| SharedRtpState::send_get_or_init(rtp_id));
|
||||||
|
let mut session = SendSession::new(shared_state, id, &settings);
|
||||||
|
let ret = new_pad(&mut session);
|
||||||
|
state.sessions.push(session);
|
||||||
|
ret
|
||||||
|
}
|
||||||
|
}),
|
||||||
|
_ => None,
|
||||||
|
}
|
||||||
|
.map(|(pad, otherpad, id, sticky_events)| {
|
||||||
|
state.max_session_id = (id + 1).max(state.max_session_id);
|
||||||
|
state.pads_session_id_map.insert(pad.clone(), id);
|
||||||
|
if let Some(ref pad) = otherpad {
|
||||||
|
state.pads_session_id_map.insert(pad.clone(), id);
|
||||||
|
}
|
||||||
|
|
||||||
|
drop(state);
|
||||||
|
|
||||||
|
pad.set_active(true).unwrap();
|
||||||
|
for event in sticky_events {
|
||||||
|
let _ = pad.store_sticky_event(&event);
|
||||||
|
}
|
||||||
|
self.obj().add_pad(&pad).unwrap();
|
||||||
|
|
||||||
|
if let Some(pad) = otherpad {
|
||||||
|
pad.set_active(true).unwrap();
|
||||||
|
self.obj().add_pad(&pad).unwrap();
|
||||||
|
}
|
||||||
|
|
||||||
|
pad
|
||||||
|
})
|
||||||
|
}
|
||||||
|
|
||||||
|
fn release_pad(&self, pad: &gst::Pad) {
|
||||||
|
let mut state = self.state.lock().unwrap();
|
||||||
|
let mut removed_pads = vec![];
|
||||||
|
let mut removed_session_ids = vec![];
|
||||||
|
if let Some(&id) = state.pads_session_id_map.get(pad) {
|
||||||
|
removed_pads.push(pad.clone());
|
||||||
|
if let Some(session) = state.mut_session_by_id(id) {
|
||||||
|
if Some(pad) == session.rtp_send_sinkpad.as_ref() {
|
||||||
|
session.rtp_send_sinkpad = None;
|
||||||
|
session
|
||||||
|
.internal_session
|
||||||
|
.inner
|
||||||
|
.lock()
|
||||||
|
.unwrap()
|
||||||
|
.rtp_send_sinkpad = None;
|
||||||
|
|
||||||
|
if let Some(srcpad) = session.rtp_send_srcpad.take() {
|
||||||
|
removed_pads.push(srcpad);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
if Some(pad) == session.rtcp_send_srcpad.as_ref() {
|
||||||
|
session.rtcp_send_srcpad = None;
|
||||||
|
}
|
||||||
|
|
||||||
|
if session.rtp_send_sinkpad.is_none() && session.rtcp_send_srcpad.is_none() {
|
||||||
|
removed_session_ids.push(session.internal_session.id);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
drop(state);
|
||||||
|
|
||||||
|
for pad in removed_pads.iter() {
|
||||||
|
let _ = pad.set_active(false);
|
||||||
|
// Pad might not have been added yet if it's a RTP recv srcpad
|
||||||
|
if pad.has_as_parent(&*self.obj()) {
|
||||||
|
let _ = self.obj().remove_pad(pad);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
{
|
||||||
|
let mut state = self.state.lock().unwrap();
|
||||||
|
|
||||||
|
for pad in removed_pads.iter() {
|
||||||
|
state.pads_session_id_map.remove(pad);
|
||||||
|
}
|
||||||
|
for id in removed_session_ids {
|
||||||
|
if let Some(session) = state.session_by_id(id) {
|
||||||
|
if session.rtp_send_sinkpad.is_none() && session.rtcp_send_srcpad.is_none() {
|
||||||
|
session.stop_rtcp_task();
|
||||||
|
state.sessions.retain(|s| s.internal_session.id != id);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
self.parent_release_pad(pad)
|
||||||
|
}
|
||||||
|
|
||||||
|
#[allow(clippy::single_match)]
|
||||||
|
fn change_state(
|
||||||
|
&self,
|
||||||
|
transition: gst::StateChange,
|
||||||
|
) -> Result<gst::StateChangeSuccess, gst::StateChangeError> {
|
||||||
|
match transition {
|
||||||
|
gst::StateChange::NullToReady => {
|
||||||
|
let settings = self.settings.lock().unwrap();
|
||||||
|
let rtp_id = settings.rtp_id.clone();
|
||||||
|
drop(settings);
|
||||||
|
|
||||||
|
let state_clone = self.state.clone();
|
||||||
|
let mut state = self.state.lock().unwrap();
|
||||||
|
let empty_sessions = state.sessions.is_empty();
|
||||||
|
match state.shared_state.as_mut() {
|
||||||
|
Some(shared) => {
|
||||||
|
if !empty_sessions && shared.name() != rtp_id {
|
||||||
|
let other_name = shared.name().to_owned();
|
||||||
|
drop(state);
|
||||||
|
self.post_error_message(gst::error_msg!(gst::LibraryError::Settings, ["rtp-id {rtp_id} does not match the currently set value {other_name}"]));
|
||||||
|
return Err(gst::StateChangeError);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
None => {
|
||||||
|
state.shared_state = Some(SharedRtpState::send_get_or_init(rtp_id.clone()));
|
||||||
|
}
|
||||||
|
}
|
||||||
|
for session in state.sessions.iter_mut() {
|
||||||
|
if session.rtcp_send_srcpad.is_some() {
|
||||||
|
session.start_rtcp_task(state_clone.clone());
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
_ => (),
|
||||||
|
}
|
||||||
|
let success = self.parent_change_state(transition)?;
|
||||||
|
|
||||||
|
match transition {
|
||||||
|
gst::StateChange::ReadyToNull => {
|
||||||
|
let mut state = self.state.lock().unwrap();
|
||||||
|
for session in state.sessions.iter_mut() {
|
||||||
|
session.stop_rtcp_task();
|
||||||
|
}
|
||||||
|
}
|
||||||
|
_ => (),
|
||||||
|
}
|
||||||
|
|
||||||
|
Ok(success)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
impl Drop for RtpSend {
|
||||||
|
fn drop(&mut self) {
|
||||||
|
if let Some(ref shared_state) = self.state.lock().unwrap().shared_state {
|
||||||
|
shared_state.unmark_send_outstanding();
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
@ -1614,7 +1614,7 @@ pub(crate) mod tests {
|
||||||
|
|
||||||
pub(crate) fn init_logs() {
|
pub(crate) fn init_logs() {
|
||||||
let _ = gst::init();
|
let _ = gst::init();
|
||||||
use crate::rtpbin2::imp::GstRustLogger;
|
use crate::rtpbin2::internal::GstRustLogger;
|
||||||
GstRustLogger::install();
|
GstRustLogger::install();
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|
@ -1680,8 +1680,8 @@ pub(crate) mod tests {
|
||||||
.ssrc(ssrc)
|
.ssrc(ssrc)
|
||||||
.sequence_number(seq_no)
|
.sequence_number(seq_no)
|
||||||
.timestamp(rtp_ts)
|
.timestamp(rtp_ts)
|
||||||
.payload(&payload)
|
.payload(payload.as_slice())
|
||||||
.write_into(&mut rtp_data)
|
.write_into(rtp_data.as_mut_slice())
|
||||||
.unwrap();
|
.unwrap();
|
||||||
rtp_data[..len].to_vec()
|
rtp_data[..len].to_vec()
|
||||||
}
|
}
|
||||||
|
|
|
||||||
|
|
@ -734,7 +734,7 @@ impl RemoteSendSource {
|
||||||
};
|
};
|
||||||
self.transit = Some(transit);
|
self.transit = Some(transit);
|
||||||
trace!("jitter {} diff {diff}", self.jitter);
|
trace!("jitter {} diff {diff}", self.jitter);
|
||||||
self.jitter += diff.saturating_sub((self.jitter + 8) >> 4);
|
self.jitter += diff.saturating_sub((self.jitter.saturating_add(8)) >> 4);
|
||||||
}
|
}
|
||||||
self.source.payload_type = Some(payload_type);
|
self.source.payload_type = Some(payload_type);
|
||||||
|
|
||||||
|
|
|
||||||
|
|
@ -118,6 +118,10 @@ impl Context {
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
pub fn clock_rate(&self, ssrc_val: u32) -> Option<u32> {
|
||||||
|
self.ssrcs.get(&ssrc_val).and_then(|ssrc| ssrc.clock_rate)
|
||||||
|
}
|
||||||
|
|
||||||
fn disassociate(&mut self, ssrc_val: u32, cname: &str) {
|
fn disassociate(&mut self, ssrc_val: u32, cname: &str) {
|
||||||
self.cname_to_largest_delays.remove(cname);
|
self.cname_to_largest_delays.remove(cname);
|
||||||
|
|
||||||
|
|
|
||||||
|
|
@ -7,12 +7,18 @@
|
||||||
//
|
//
|
||||||
// SPDX-License-Identifier: MPL-2.0
|
// SPDX-License-Identifier: MPL-2.0
|
||||||
|
|
||||||
use std::sync::{Arc, Mutex};
|
use std::sync::{atomic::AtomicUsize, Arc, Mutex};
|
||||||
|
|
||||||
use gst::{prelude::*, Caps};
|
use gst::{prelude::*, Caps};
|
||||||
use gst_check::Harness;
|
use gst_check::Harness;
|
||||||
use rtp_types::*;
|
use rtp_types::*;
|
||||||
|
|
||||||
|
static ELEMENT_COUNTER: AtomicUsize = AtomicUsize::new(0);
|
||||||
|
|
||||||
|
fn next_element_counter() -> usize {
|
||||||
|
ELEMENT_COUNTER.fetch_add(1, std::sync::atomic::Ordering::SeqCst)
|
||||||
|
}
|
||||||
|
|
||||||
fn init() {
|
fn init() {
|
||||||
use std::sync::Once;
|
use std::sync::Once;
|
||||||
static INIT: Once = Once::new();
|
static INIT: Once = Once::new();
|
||||||
|
|
@ -34,7 +40,7 @@ fn generate_rtp_buffer(seqno: u16, rtpts: u32, payload_len: usize) -> gst::Buffe
|
||||||
.payload_type(TEST_PT)
|
.payload_type(TEST_PT)
|
||||||
.sequence_number(seqno)
|
.sequence_number(seqno)
|
||||||
.timestamp(rtpts)
|
.timestamp(rtpts)
|
||||||
.payload(&payload);
|
.payload(payload.as_slice());
|
||||||
let size = packet.calculate_size().unwrap();
|
let size = packet.calculate_size().unwrap();
|
||||||
let mut data = vec![0; size];
|
let mut data = vec![0; size];
|
||||||
packet.write_into(&mut data).unwrap();
|
packet.write_into(&mut data).unwrap();
|
||||||
|
|
@ -44,8 +50,13 @@ fn generate_rtp_buffer(seqno: u16, rtpts: u32, payload_len: usize) -> gst::Buffe
|
||||||
#[test]
|
#[test]
|
||||||
fn test_send() {
|
fn test_send() {
|
||||||
init();
|
init();
|
||||||
|
let id = next_element_counter();
|
||||||
|
|
||||||
let mut h = Harness::with_padnames("rtpbin2", Some("rtp_send_sink_0"), Some("rtp_send_src_0"));
|
let elem = gst::ElementFactory::make("rtpsend")
|
||||||
|
.property("rtp-id", id.to_string())
|
||||||
|
.build()
|
||||||
|
.unwrap();
|
||||||
|
let mut h = Harness::with_element(&elem, Some("rtp_sink_0"), Some("rtp_src_0"));
|
||||||
h.play();
|
h.play();
|
||||||
|
|
||||||
let caps = Caps::builder("application/x-rtp")
|
let caps = Caps::builder("application/x-rtp")
|
||||||
|
|
@ -89,10 +100,15 @@ fn test_send() {
|
||||||
#[test]
|
#[test]
|
||||||
fn test_receive() {
|
fn test_receive() {
|
||||||
init();
|
init();
|
||||||
|
let id = next_element_counter();
|
||||||
|
|
||||||
let h = Arc::new(Mutex::new(Harness::with_padnames(
|
let elem = gst::ElementFactory::make("rtprecv")
|
||||||
"rtpbin2",
|
.property("rtp-id", id.to_string())
|
||||||
Some("rtp_recv_sink_0"),
|
.build()
|
||||||
|
.unwrap();
|
||||||
|
let h = Arc::new(Mutex::new(Harness::with_element(
|
||||||
|
&elem,
|
||||||
|
Some("rtp_sink_0"),
|
||||||
None,
|
None,
|
||||||
)));
|
)));
|
||||||
let weak_h = Arc::downgrade(&h);
|
let weak_h = Arc::downgrade(&h);
|
||||||
|
|
@ -124,7 +140,7 @@ fn test_receive() {
|
||||||
let push_pad = inner
|
let push_pad = inner
|
||||||
.element()
|
.element()
|
||||||
.unwrap()
|
.unwrap()
|
||||||
.static_pad("rtp_recv_sink_0")
|
.static_pad("rtp_sink_0")
|
||||||
.unwrap()
|
.unwrap()
|
||||||
.peer()
|
.peer()
|
||||||
.unwrap();
|
.unwrap();
|
||||||
|
|
@ -181,10 +197,15 @@ fn test_receive() {
|
||||||
#[test]
|
#[test]
|
||||||
fn test_receive_flush() {
|
fn test_receive_flush() {
|
||||||
init();
|
init();
|
||||||
|
let id = next_element_counter();
|
||||||
|
|
||||||
let h = Arc::new(Mutex::new(Harness::with_padnames(
|
let elem = gst::ElementFactory::make("rtprecv")
|
||||||
"rtpbin2",
|
.property("rtp-id", id.to_string())
|
||||||
Some("rtp_recv_sink_0"),
|
.build()
|
||||||
|
.unwrap();
|
||||||
|
let h = Arc::new(Mutex::new(Harness::with_element(
|
||||||
|
&elem,
|
||||||
|
Some("rtp_sink_0"),
|
||||||
None,
|
None,
|
||||||
)));
|
)));
|
||||||
let weak_h = Arc::downgrade(&h);
|
let weak_h = Arc::downgrade(&h);
|
||||||
|
|
@ -216,7 +237,7 @@ fn test_receive_flush() {
|
||||||
let push_pad = inner
|
let push_pad = inner
|
||||||
.element()
|
.element()
|
||||||
.unwrap()
|
.unwrap()
|
||||||
.static_pad("rtp_recv_sink_0")
|
.static_pad("rtp_sink_0")
|
||||||
.unwrap()
|
.unwrap()
|
||||||
.peer()
|
.peer()
|
||||||
.unwrap();
|
.unwrap();
|
||||||
|
|
|
||||||
Loading…
Reference in a new issue