Merge branch 'rtp2' into 'main'

Draft: rtp: new rtpbin2 element

See merge request gstreamer/gst-plugins-rs!1426
This commit is contained in:
Matthew Waters 2024-03-19 08:09:49 +00:00
commit b51eeb8ea1
17 changed files with 10014 additions and 29 deletions

16
Cargo.lock generated
View file

@ -2643,15 +2643,23 @@ dependencies = [
"atomic_refcell",
"bitstream-io",
"chrono",
"futures",
"gio",
"gst-plugin-version-helper",
"gstreamer",
"gstreamer-app",
"gstreamer-base",
"gstreamer-check",
"gstreamer-net",
"gstreamer-rtp",
"gstreamer-video",
"log",
"once_cell",
"rand",
"rtcp-types",
"rtp-types",
"smallvec",
"tokio",
]
[[package]]
@ -5397,6 +5405,14 @@ dependencies = [
"winapi",
]
[[package]]
name = "rtcp-types"
version = "0.0.1"
source = "git+https://github.com/ystreet/rtcp-types#c1da8a1a193a0c02d798fea5f16863b69abd9000"
dependencies = [
"thiserror",
]
[[package]]
name = "rtp-types"
version = "0.1.1"

View file

@ -6304,6 +6304,141 @@
},
"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": {
"author": "Thibault Saunier <tsaunier@igalia.com>",
"description": "Estimates current network bandwidth using the Google Congestion Control algorithm notifying about it through the 'bitrate' property",

View file

@ -12,12 +12,21 @@ rust-version.workspace = true
atomic_refcell = "0.1"
bitstream-io = "2.1"
chrono = { version = "0.4", default-features = false }
gst = { workspace = true, features = ["v1_20"] }
gst-rtp = { workspace = true, features = ["v1_20"] }
futures = "0.3"
gio.workspace = true
gst = { workspace = true, features = ["v1_20"] }
gst-base = { workspace = true, features = ["v1_20"] }
gst-net = { workspace = true, features = ["v1_20"] }
gst-rtp = { workspace = true, features = ["v1_20"] }
gst-video = { workspace = true, features = ["v1_20"] }
log = "0.4"
once_cell.workspace = true
rand = { version = "0.8", default-features = false, features = ["std", "std_rng" ] }
rtp-types = { version = "0.1" }
rtcp-types = { git = "https://github.com/ystreet/rtcp-types", version = "0.0" }
smallvec = { version = "1.11", features = ["union", "write", "const_generics", "const_new"] }
# TODO: experiment with other async executors (mio, async-std, etc)
tokio = { version = "1", default-features = false, features = ["rt-multi-thread", "time", "sync"] }
[dev-dependencies]
gst-check = { workspace = true, features = ["v1_20"] }
@ -48,4 +57,4 @@ versioning = false
import_library = false
[package.metadata.capi.pkg_config]
requires_private = "gstreamer-1.0, gstreamer-base-1.0, gstreamer-rtp-1.0, gobject-2.0, glib-2.0, gmodule-2.0"
requires_private = "gstreamer-1.0, gstreamer-base-1.0, gstreamer-rtp-1.0, gstreamer-net-1.0, gstreamer-video-1.0 gobject-2.0, glib-2.0, gmodule-2.0, gio-2.0"

View file

@ -14,9 +14,15 @@
*
* Since: plugins-rs-0.9.0
*/
#[macro_use]
extern crate log;
use gst::glib;
mod gcc;
mod rtpbin2;
mod utils;
mod audio_discont;
mod baseaudiopay;
@ -32,6 +38,7 @@ mod tests;
fn plugin_init(plugin: &gst::Plugin) -> Result<(), glib::BoolError> {
gcc::register(plugin)?;
rtpbin2::register(plugin)?;
#[cfg(feature = "doc")]
{
@ -68,3 +75,14 @@ gst::plugin_define!(
env!("CARGO_PKG_REPOSITORY"),
env!("BUILD_REL_DATE")
);
#[cfg(test)]
pub(crate) fn test_init() {
use std::sync::Once;
static INIT: Once = Once::new();
INIT.call_once(|| {
gst::init().unwrap();
plugin_register_static().expect("rtp plugin test");
});
}

View file

@ -0,0 +1,318 @@
// SPDX-License-Identifier: MPL-2.0
use gst::glib;
use gst::prelude::*;
use gst::subclass::prelude::*;
use once_cell::sync::Lazy;
use std::sync::{Mutex, Weak};
use crate::rtpbin2::internal::SharedSessionInner;
static CAT: Lazy<gst::DebugCategory> = Lazy::new(|| {
gst::DebugCategory::new(
"rtp2-config",
gst::DebugColorFlags::empty(),
Some("Rtp2 config"),
)
});
glib::wrapper! {
pub struct Rtp2Session(ObjectSubclass<imp::Rtp2Session>);
}
impl Rtp2Session {
pub(crate) fn new(weak_session: Weak<Mutex<SharedSessionInner>>) -> Self {
let ret = glib::Object::new::<Self>();
let imp = ret.imp();
imp.set_session(weak_session);
ret
}
}
mod imp {
use std::sync::Arc;
use super::*;
#[derive(Debug, Default)]
struct State {
pub(super) weak_session: Option<Weak<Mutex<SharedSessionInner>>>,
}
#[derive(Debug, Default)]
pub struct Rtp2Session {
state: Mutex<State>,
}
impl Rtp2Session {
pub(super) fn set_session(&self, weak_session: Weak<Mutex<SharedSessionInner>>) {
let mut state = self.state.lock().unwrap();
state.weak_session = Some(weak_session);
}
fn session(&self) -> Option<Arc<Mutex<SharedSessionInner>>> {
self.state
.lock()
.unwrap()
.weak_session
.as_ref()
.unwrap()
.upgrade()
}
pub fn set_pt_map(&self, pt_map: Option<gst::Structure>) {
let Some(session) = self.session() else {
return;
};
let mut session = session.lock().unwrap();
session.clear_pt_map();
let Some(pt_map) = pt_map else {
return;
};
for (key, value) in pt_map.iter() {
let Ok(pt) = key.parse::<u8>() else {
gst::warning!(CAT, "failed to parse key as a pt");
continue;
};
if let Ok(caps) = value.get::<gst::Caps>() {
session.add_caps(caps);
} else {
gst::warning!(CAT, "{pt} does not contain a caps value");
continue;
}
}
}
pub fn pt_map(&self) -> gst::Structure {
let mut ret = gst::Structure::builder("application/x-rtp2-pt-map");
let Some(session) = self.session() else {
return ret.build();
};
let session = session.lock().unwrap();
for (pt, caps) in session.pt_map() {
ret = ret.field(pt.to_string(), caps);
}
ret.build()
}
pub fn stats(&self) -> Option<gst::Structure> {
let Some(session) = self.session() else {
return None;
};
let session = session.lock().unwrap();
Some(session.stats())
}
}
#[glib::object_subclass]
impl ObjectSubclass for Rtp2Session {
const NAME: &'static str = "GstRtp2Session";
type Type = super::Rtp2Session;
type ParentType = glib::Object;
}
impl ObjectImpl for Rtp2Session {
fn properties() -> &'static [glib::ParamSpec] {
static PROPERTIES: Lazy<Vec<glib::ParamSpec>> = Lazy::new(|| {
vec![glib::ParamSpecBoxed::builder::<gst::Structure>("pt-map")
.nick("RTP Payload Type Map")
.blurb("Mapping of RTP payload type to caps")
.build()]
});
PROPERTIES.as_ref()
}
fn property(&self, _id: usize, pspec: &glib::ParamSpec) -> glib::Value {
match pspec.name() {
"pt-map" => self.pt_map().to_value(),
"stats" => self.stats().to_value(),
_ => unreachable!(),
}
}
fn set_property(&self, _id: usize, value: &glib::Value, pspec: &glib::ParamSpec) {
match pspec.name() {
"pt-map" => self.set_pt_map(
value
.get::<Option<gst::Structure>>()
.expect("Type checked upstream"),
),
_ => unreachable!(),
}
}
fn signals() -> &'static [glib::subclass::Signal] {
static SIGNALS: Lazy<Vec<glib::subclass::Signal>> = Lazy::new(|| {
vec![
glib::subclass::Signal::builder("new-ssrc")
.param_types([u32::static_type()])
.build(),
glib::subclass::Signal::builder("bye-ssrc")
.param_types([u32::static_type()])
.build(),
]
});
SIGNALS.as_ref()
}
}
}
#[cfg(test)]
mod tests {
use std::sync::{
atomic::{AtomicBool, AtomicUsize},
Arc,
};
use crate::{rtpbin2::session::tests::generate_rtp_packet, test_init};
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]
fn pt_map_get_empty() {
test_init();
let id = next_element_counter();
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 pt_map = session.property::<gst::Structure>("pt-map");
assert!(pt_map.has_name("application/x-rtp2-pt-map"));
assert_eq!(pt_map.fields().len(), 0);
}
#[test]
fn pt_map_set() {
test_init();
let id = next_element_counter();
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 pt = 96i32;
let pt_caps = gst::Caps::builder("application/x-rtp")
.field("payload", pt)
.field("clock-rate", 90000i32)
.build();
let pt_map = gst::Structure::builder("application/x-rtp2-pt-map")
.field(pt.to_string(), pt_caps.clone())
.build();
session.set_property("pt-map", pt_map);
let prop = session.property::<gst::Structure>("pt-map");
assert!(prop.has_name("application/x-rtp2-pt-map"));
assert_eq!(prop.fields().len(), 1);
let caps = prop.get::<gst::Caps>(pt.to_string()).unwrap();
assert_eq!(pt_caps, caps);
}
#[test]
fn pt_map_set_none() {
test_init();
let id = next_element_counter();
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]);
session.set_property("pt-map", None::<gst::Structure>);
let prop = session.property::<gst::Structure>("pt-map");
assert!(prop.has_name("application/x-rtp2-pt-map"));
}
#[test]
fn new_send_ssrc() {
test_init();
let ssrc = 0x12345678;
let new_ssrc_hit = Arc::new(AtomicBool::new(false));
let id = next_element_counter();
let rtpbin2 = gst::ElementFactory::make("rtpsend")
.property("rtp-id", id.to_string())
.build()
.unwrap();
let mut h =
gst_check::Harness::with_element(&rtpbin2, Some("rtp_sink_0"), Some("rtp_src_0"));
let session = h
.element()
.unwrap()
.emit_by_name::<gst::glib::Object>("get-session", &[&0u32]);
let ssrc_hit = new_ssrc_hit.clone();
session.connect("new-ssrc", false, move |args| {
let new_ssrc = args[1].get::<u32>().unwrap();
ssrc_hit.store(true, std::sync::atomic::Ordering::SeqCst);
assert_eq!(new_ssrc, ssrc);
None
});
h.set_src_caps_str("application/x-rtp,payload=96,clock-rate=90000");
let mut segment = gst::Segment::new();
segment.set_format(gst::Format::Time);
h.push_event(gst::event::Segment::builder(&segment).build());
let buf1 = gst::Buffer::from_mut_slice(generate_rtp_packet(ssrc, 0x34, 0x10, 16));
h.push(buf1.clone()).unwrap();
assert!(new_ssrc_hit.load(std::sync::atomic::Ordering::SeqCst));
let buf2 = gst::Buffer::from_mut_slice(generate_rtp_packet(ssrc, 0x35, 0x10, 16));
h.push(buf2.clone()).unwrap();
let buf3 = h.pull().unwrap();
assert_eq!(buf3, buf1);
let buf4 = h.pull().unwrap();
assert_eq!(buf4, buf2);
}
#[test]
fn bye_send_ssrc() {
test_init();
let ssrc = 0x12345678;
let (bye_ssrc_sender, bye_ssrc_receiver) = std::sync::mpsc::sync_channel(16);
let id = next_element_counter();
let rtpbin2 = gst::ElementFactory::make("rtpsend")
.property("rtp-id", id.to_string())
.build()
.unwrap();
let mut h =
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
.element()
.unwrap()
.emit_by_name::<gst::glib::Object>("get-session", &[&0u32]);
session.connect("bye-ssrc", false, move |args| {
let bye_ssrc = args[1].get::<u32>().unwrap();
assert_eq!(bye_ssrc, ssrc);
bye_ssrc_sender.send(ssrc).unwrap();
None
});
h.set_src_caps_str("application/x-rtp,payload=96,clock-rate=90000");
let mut segment = gst::Segment::new();
segment.set_format(gst::Format::Time);
h.push_event(gst::event::Segment::builder(&segment).build());
let buf1 = gst::Buffer::from_mut_slice(generate_rtp_packet(ssrc, 0x34, 0x10, 16));
h.push(buf1.clone()).unwrap();
let buf2 = gst::Buffer::from_mut_slice(generate_rtp_packet(ssrc, 0x35, 0x10, 16));
h.push(buf2.clone()).unwrap();
let buf3 = h.pull().unwrap();
assert_eq!(buf3, buf1);
let buf4 = h.pull().unwrap();
assert_eq!(buf4, buf2);
h.push_event(gst::event::Eos::builder().build());
let _rtcp = h_rtcp.pull().unwrap();
assert_eq!(bye_ssrc_receiver.recv().unwrap(), ssrc);
}
}

View 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) {}
}

View file

@ -0,0 +1,669 @@
use crate::utils::ExtendedSeqnum;
use rtp_types::RtpPacket;
use std::cmp::Ordering;
use std::collections::BTreeSet;
use std::time::{Duration, Instant};
#[derive(Debug, Clone, Copy)]
struct Stats {
num_late: u64,
num_lost: u64,
num_duplicates: u64,
num_pushed: u64,
}
impl From<Stats> for gst::Structure {
fn from(stats: Stats) -> gst::Structure {
gst::Structure::builder("application/x-rtp-jitterbuffer-stats")
.field("num-late", stats.num_late)
.field("num-duplicates", stats.num_duplicates)
.field("num-lost", stats.num_lost)
.field("num-pushed", stats.num_pushed)
.build()
}
}
#[derive(Debug)]
pub struct JitterBuffer {
packet_counter: usize,
// A set of extended seqnums that we've already seen through,
// intentionally trimmed separately from the items list so that
// we can detect duplicates after the first copy has exited the
// queue
seqnums: BTreeSet<u64>,
items: BTreeSet<Item>,
latency: Duration,
// Arrival time, PTS
base_times: Option<(Instant, u64)>,
last_output_seqnum: Option<u64>,
extended_seqnum: ExtendedSeqnum,
last_input_ts: Option<u64>,
stats: Stats,
flushing: bool,
}
#[derive(Debug, PartialEq, Eq)]
pub enum PollResult {
Forward { id: usize, discont: bool },
Drop(usize),
Timeout(Instant),
Empty,
Flushing,
}
#[derive(Debug, PartialEq, Eq)]
pub enum QueueResult {
Queued(usize),
Late,
Duplicate,
Flushing,
}
#[derive(Eq, Debug)]
struct Item {
id: usize,
// If not set, this is an event / query
pts: Option<u64>,
seqnum: u64,
}
impl Ord for Item {
fn cmp(&self, other: &Self) -> Ordering {
self.seqnum
.cmp(&other.seqnum)
.then(match (self.pts, other.pts) {
(None, Some(_)) => Ordering::Greater,
(Some(_), None) => Ordering::Less,
_ => Ordering::Equal,
})
}
}
impl PartialOrd for Item {
fn partial_cmp(&self, other: &Self) -> Option<Ordering> {
Some(self.cmp(other))
}
}
impl PartialEq for Item {
fn eq(&self, other: &Self) -> bool {
self.cmp(other) == Ordering::Equal
}
}
impl JitterBuffer {
pub fn new(latency: Duration) -> Self {
Self {
packet_counter: 0,
seqnums: BTreeSet::new(),
items: BTreeSet::new(),
latency,
base_times: None,
last_input_ts: None,
last_output_seqnum: None,
extended_seqnum: ExtendedSeqnum::default(),
stats: Stats {
num_late: 0,
num_lost: 0,
num_duplicates: 0,
num_pushed: 0,
},
flushing: true,
}
}
pub fn queue_serialized_item(&mut self) -> QueueResult {
let id = self.packet_counter;
self.packet_counter += 1;
let item = Item {
id,
pts: None,
seqnum: (*self.seqnums.last().unwrap_or(&0)),
};
self.items.insert(item);
trace!("Queued serialized item and assigned ID {id}");
QueueResult::Queued(id)
}
pub fn set_flushing(&mut self, flushing: bool) {
trace!("Flush changed from {} to {flushing}", self.flushing);
self.flushing = flushing;
self.last_output_seqnum = None;
}
pub fn queue_packet(&mut self, rtp: &RtpPacket, mut pts: u64, now: Instant) -> QueueResult {
if self.flushing {
return QueueResult::Flushing;
}
// From this point on we always work with extended sequence numbers
let seqnum = self.extended_seqnum.next(rtp.sequence_number());
if let Some(ts) = self.last_input_ts {
pts = pts.max(ts);
}
self.last_input_ts = Some(pts);
self.base_times.get_or_insert_with(|| {
debug!("Selected base times {:?} {}", now, pts);
(now, pts)
});
// Maintain (and trim) our seqnum list for duplicate detection
while self.seqnums.len() >= std::u16::MAX as usize {
debug!("Trimming");
self.seqnums.pop_first();
}
if self.seqnums.contains(&seqnum) {
trace!(
"Duplicated packet {} (extended {})",
rtp.sequence_number(),
seqnum,
);
self.stats.num_duplicates += 1;
return QueueResult::Duplicate;
}
self.seqnums.insert(seqnum);
if let Some(last_output_seqnum) = self.last_output_seqnum {
if last_output_seqnum >= seqnum {
debug!(
"Late packet {} (extended {})",
rtp.sequence_number(),
seqnum
);
self.stats.num_late += 1;
return QueueResult::Late;
}
}
let id = self.packet_counter;
self.packet_counter += 1;
let item = Item {
id,
pts: Some(pts),
seqnum,
};
if !self.items.insert(item) {
unreachable!()
}
trace!("Queued RTP packet with ts {pts}, assigned ID {id}");
QueueResult::Queued(id)
}
pub fn poll(&mut self, now: Instant) -> PollResult {
if self.flushing {
if let Some(item) = self.items.pop_first() {
return PollResult::Drop(item.id);
} else {
return PollResult::Flushing;
}
}
trace!("Polling at {:?}", now);
let Some((base_instant, base_ts)) = self.base_times else {
return PollResult::Empty;
};
let duration_since_base_instant = now - base_instant;
trace!(
"Duration since base instant {:?}",
duration_since_base_instant
);
let Some(item) = self.items.first() else {
return PollResult::Empty;
};
// If an event / query is at the top of the queue, it can be forwarded immediately
let Some(pts) = item.pts else {
let item = self.items.pop_first().unwrap();
return PollResult::Forward {
id: item.id,
discont: false,
};
};
let ts = pts.checked_sub(base_ts).unwrap();
let deadline = Duration::from_nanos(ts) + self.latency;
trace!(
"Considering packet {} with ts {ts}, deadline is {deadline:?}",
item.id
);
if deadline <= duration_since_base_instant {
debug!("Packet with id {} is ready", item.id);
let discont = match self.last_output_seqnum {
None => true,
Some(last_output_seq_ext) => {
let gap = item.seqnum - last_output_seq_ext;
self.stats.num_lost += gap - 1;
gap != 1
}
};
self.last_output_seqnum = Some(item.seqnum);
// Safe unwrap, we know the queue isn't empty at this point
let packet = self.items.pop_first().unwrap();
self.stats.num_pushed += 1;
PollResult::Forward {
id: packet.id,
discont,
}
} else {
trace!("Packet with id {} is not ready", item.id);
PollResult::Timeout(base_instant + deadline)
}
}
pub fn stats(&self) -> gst::Structure {
self.stats.into()
}
}
#[cfg(test)]
mod tests {
use super::*;
use crate::rtpbin2::session::tests::generate_rtp_packet;
#[test]
fn empty() {
let mut jb = JitterBuffer::new(Duration::from_secs(1));
jb.set_flushing(false);
let now = Instant::now();
assert_eq!(jb.poll(now), PollResult::Empty);
}
#[test]
fn receive_one_packet_no_latency() {
let mut jb = JitterBuffer::new(Duration::from_secs(0));
jb.set_flushing(false);
let rtp_data = generate_rtp_packet(0x12345678, 0, 0, 4);
let packet = RtpPacket::parse(&rtp_data).unwrap();
let now = Instant::now();
let QueueResult::Queued(id) = jb.queue_packet(&packet, 0, now) else {
unreachable!()
};
assert_eq!(jb.poll(now), PollResult::Forward { id, discont: true });
}
#[test]
fn receive_one_packet_with_latency() {
let mut jb = JitterBuffer::new(Duration::from_secs(1));
jb.set_flushing(false);
let rtp_data = generate_rtp_packet(0x12345678, 0, 0, 4);
let packet = RtpPacket::parse(&rtp_data).unwrap();
let mut now = Instant::now();
let QueueResult::Queued(id) = jb.queue_packet(&packet, 0, now) else {
unreachable!()
};
assert_eq!(
jb.poll(now),
PollResult::Timeout(now + Duration::from_secs(1))
);
now += Duration::from_secs(1);
now -= Duration::from_nanos(1);
assert_eq!(
jb.poll(now),
PollResult::Timeout(now + Duration::from_nanos(1))
);
now += Duration::from_nanos(1);
assert_eq!(jb.poll(now), PollResult::Forward { id, discont: true });
}
#[test]
fn ordered_packets_no_latency() {
let mut jb = JitterBuffer::new(Duration::from_secs(0));
jb.set_flushing(false);
let now = Instant::now();
let rtp_data = generate_rtp_packet(0x12345678, 0, 0, 4);
let packet = RtpPacket::parse(&rtp_data).unwrap();
let QueueResult::Queued(id_first) = jb.queue_packet(&packet, 0, now) else {
unreachable!()
};
let rtp_data = generate_rtp_packet(0x12345678, 1, 0, 4);
let packet = RtpPacket::parse(&rtp_data).unwrap();
let QueueResult::Queued(id_second) = jb.queue_packet(&packet, 0, now) else {
unreachable!()
};
assert_eq!(
jb.poll(now),
PollResult::Forward {
id: id_first,
discont: true
}
);
assert_eq!(
jb.poll(now),
PollResult::Forward {
id: id_second,
discont: false
}
);
}
#[test]
fn ordered_packets_no_latency_with_gap() {
let mut jb = JitterBuffer::new(Duration::from_secs(0));
jb.set_flushing(false);
let now = Instant::now();
let rtp_data = generate_rtp_packet(0x12345678, 0, 0, 4);
let packet = RtpPacket::parse(&rtp_data).unwrap();
let QueueResult::Queued(id_first) = jb.queue_packet(&packet, 0, now) else {
unreachable!()
};
let rtp_data = generate_rtp_packet(0x12345678, 2, 0, 4);
let packet = RtpPacket::parse(&rtp_data).unwrap();
let QueueResult::Queued(id_second) = jb.queue_packet(&packet, 0, now) else {
unreachable!()
};
assert_eq!(
jb.poll(now),
PollResult::Forward {
id: id_first,
discont: true
}
);
assert_eq!(
jb.poll(now),
PollResult::Forward {
id: id_second,
discont: true
}
);
}
#[test]
fn misordered_packets_no_latency() {
let mut jb = JitterBuffer::new(Duration::from_secs(0));
jb.set_flushing(false);
let now = Instant::now();
let rtp_data = generate_rtp_packet(0x12345678, 1, 0, 4);
let packet = RtpPacket::parse(&rtp_data).unwrap();
let QueueResult::Queued(id) = jb.queue_packet(&packet, 0, now) else {
unreachable!()
};
assert_eq!(jb.poll(now), PollResult::Forward { id, discont: true });
let rtp_data = generate_rtp_packet(0x12345678, 0, 0, 4);
let packet = RtpPacket::parse(&rtp_data).unwrap();
assert_eq!(jb.queue_packet(&packet, 0, now), QueueResult::Late);
// Try and push a duplicate
let rtp_data = generate_rtp_packet(0x12345678, 1, 0, 4);
let packet = RtpPacket::parse(&rtp_data).unwrap();
assert_eq!(jb.queue_packet(&packet, 0, now), QueueResult::Duplicate);
// We do accept future sequence numbers up to a distance of at least std::i16::MAX
let rtp_data = generate_rtp_packet(0x12345678, std::i16::MAX as u16 + 1, 0, 4);
let packet = RtpPacket::parse(&rtp_data).unwrap();
let QueueResult::Queued(id) = jb.queue_packet(&packet, 0, now) else {
unreachable!()
};
assert_eq!(jb.poll(now), PollResult::Forward { id, discont: true });
// But no further
let rtp_data = generate_rtp_packet(0x12345678, 2, 0, 4);
let packet = RtpPacket::parse(&rtp_data).unwrap();
assert_eq!(jb.queue_packet(&packet, 0, now), QueueResult::Late);
}
#[test]
fn ordered_packets_with_latency() {
let mut jb = JitterBuffer::new(Duration::from_secs(1));
jb.set_flushing(false);
let mut now = Instant::now();
let rtp_data = generate_rtp_packet(0x12345678, 0, 0, 4);
let packet = RtpPacket::parse(&rtp_data).unwrap();
let QueueResult::Queued(id_first) = jb.queue_packet(&packet, 0, now) else {
unreachable!()
};
assert_eq!(
jb.poll(now),
PollResult::Timeout(now + Duration::from_secs(1))
);
let rtp_data = generate_rtp_packet(0x12345678, 1, 180000, 4);
let packet = RtpPacket::parse(&rtp_data).unwrap();
let QueueResult::Queued(id_second) = jb.queue_packet(&packet, 2_000_000_000, now) else {
unreachable!()
};
assert_eq!(
jb.poll(now),
PollResult::Timeout(now + Duration::from_secs(1))
);
now += Duration::from_secs(1);
assert_eq!(
jb.poll(now),
PollResult::Forward {
id: id_first,
discont: true
}
);
assert_eq!(
jb.poll(now),
PollResult::Timeout(now + Duration::from_secs(2))
);
now += Duration::from_secs(2);
assert_eq!(
jb.poll(now),
PollResult::Forward {
id: id_second,
discont: false
}
);
}
fn assert_stats(
jb: &JitterBuffer,
num_late: u64,
num_lost: u64,
num_duplicates: u64,
num_pushed: u64,
) {
let stats = jb.stats();
assert_eq!(stats.get::<u64>("num-late").unwrap(), num_late);
assert_eq!(stats.get::<u64>("num-lost").unwrap(), num_lost);
assert_eq!(stats.get::<u64>("num-duplicates").unwrap(), num_duplicates);
assert_eq!(stats.get::<u64>("num-pushed").unwrap(), num_pushed);
}
#[test]
fn stats() {
let mut jb = JitterBuffer::new(Duration::from_secs(1));
jb.set_flushing(false);
let mut now = Instant::now();
let rtp_data = generate_rtp_packet(0x12345678, 0, 0, 4);
let packet = RtpPacket::parse(&rtp_data).unwrap();
jb.queue_packet(&packet, 0, now);
assert_stats(&jb, 0, 0, 0, 0);
// At this point pushing the same packet in before it gets output
// results in an increment of the duplicate stat
jb.queue_packet(&packet, 0, now);
assert_stats(&jb, 0, 0, 1, 0);
now += Duration::from_secs(1);
let _ = jb.poll(now);
assert_stats(&jb, 0, 0, 1, 1);
// Pushing it after the first version got output also results in
// an increment of the duplicate stat
jb.queue_packet(&packet, 0, now);
assert_stats(&jb, 0, 0, 2, 1);
// Then after a packet with seqnum 2 goes through, the lost
// stat must be incremented by 1 (as packet with seqnum 1 went missing)
let rtp_data = generate_rtp_packet(0x12345678, 2, 9000, 4);
let packet = RtpPacket::parse(&rtp_data).unwrap();
jb.queue_packet(&packet, 100_000_000, now);
now += Duration::from_millis(100);
let _ = jb.poll(now);
assert_stats(&jb, 0, 1, 2, 2);
// If the packet with seqnum 1 does arrive after that, it should be
// considered both late and lost
let rtp_data = generate_rtp_packet(0x12345678, 1, 4500, 4);
let packet = RtpPacket::parse(&rtp_data).unwrap();
jb.queue_packet(&packet, 50_000_000, now);
let _ = jb.poll(now);
assert_stats(&jb, 1, 1, 2, 2);
// Finally if it arrives again it should be considered a duplicate,
// and will have achieved the dubious honor of simultaneously being
// lost, late and duplicated
jb.queue_packet(&packet, 50_000_000, now);
let _ = jb.poll(now);
assert_stats(&jb, 1, 1, 3, 2);
}
#[test]
fn serialized_items() {
let mut jb = JitterBuffer::new(Duration::from_secs(0));
jb.set_flushing(false);
let now = Instant::now();
let rtp_data = generate_rtp_packet(0x12345678, 0, 0, 4);
let packet = RtpPacket::parse(&rtp_data).unwrap();
let QueueResult::Queued(id_first_serialized_item) = jb.queue_serialized_item() else {
unreachable!()
};
let QueueResult::Queued(id_first) = jb.queue_packet(&packet, 0, now) else {
unreachable!()
};
let QueueResult::Queued(id_second_serialized_item) = jb.queue_serialized_item() else {
unreachable!()
};
let rtp_data = generate_rtp_packet(0x12345678, 1, 0, 4);
let packet = RtpPacket::parse(&rtp_data).unwrap();
let QueueResult::Queued(id_second) = jb.queue_packet(&packet, 0, now) else {
unreachable!()
};
assert_eq!(
jb.poll(now),
PollResult::Forward {
id: id_first_serialized_item,
discont: false
}
);
assert_eq!(
jb.poll(now),
PollResult::Forward {
id: id_first,
discont: true
}
);
assert_eq!(
jb.poll(now),
PollResult::Forward {
id: id_second_serialized_item,
discont: false
}
);
assert_eq!(
jb.poll(now),
PollResult::Forward {
id: id_second,
discont: false
}
);
}
#[test]
fn flushing_queue() {
let mut jb = JitterBuffer::new(Duration::from_secs(0));
jb.set_flushing(false);
let now = Instant::now();
let rtp_data = generate_rtp_packet(0x12345678, 0, 0, 4);
let packet = RtpPacket::parse(&rtp_data).unwrap();
let QueueResult::Queued(id_first_serialized_item) = jb.queue_serialized_item() else {
unreachable!()
};
let QueueResult::Queued(id_first) = jb.queue_packet(&packet, 0, now) else {
unreachable!()
};
// Everything after this should eventually return flushing, poll() will instruct to drop
// everything stored and then return flushing indefinitely.
jb.set_flushing(true);
assert_eq!(jb.queue_packet(&packet, 0, now), QueueResult::Flushing);
assert_eq!(jb.poll(now), PollResult::Drop(id_first_serialized_item));
assert_eq!(jb.poll(now), PollResult::Drop(id_first));
assert_eq!(jb.poll(now), PollResult::Flushing);
assert_eq!(jb.poll(now), PollResult::Flushing);
jb.set_flushing(false);
assert_eq!(jb.poll(now), PollResult::Empty);
}
}

View file

@ -0,0 +1,44 @@
// SPDX-License-Identifier: MPL-2.0
use gst::glib;
use gst::prelude::*;
use once_cell::sync::Lazy;
mod config;
mod internal;
mod jitterbuffer;
mod rtprecv;
mod rtpsend;
mod session;
mod source;
mod sync;
mod time;
glib::wrapper! {
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> {
gst::Element::register(
Some(plugin),
"rtpsend",
gst::Rank::NONE,
RtpSend::static_type(),
)?;
gst::Element::register(
Some(plugin),
"rtprecv",
gst::Rank::NONE,
RtpRecv::static_type(),
)
}
pub static RUNTIME: Lazy<tokio::runtime::Runtime> = Lazy::new(|| {
tokio::runtime::Builder::new_multi_thread()
.enable_time()
.worker_threads(1)
.build()
.unwrap()
});

File diff suppressed because it is too large Load diff

View 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")]
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();
}
}
}

File diff suppressed because it is too large Load diff

File diff suppressed because it is too large Load diff

826
net/rtp/src/rtpbin2/sync.rs Normal file
View file

@ -0,0 +1,826 @@
use gst::glib;
use gst::prelude::MulDiv;
use std::collections::{HashMap, VecDeque};
use std::sync::Arc;
use std::time::Duration;
use crate::utils::ExtendedTimestamp;
use super::time::NtpTime;
#[derive(Default, Debug)]
struct Ssrc {
cname: Option<Arc<str>>,
clock_rate: Option<u32>,
extended_timestamp: ExtendedTimestamp,
last_sr_ntp_timestamp: Option<NtpTime>,
last_sr_rtp_ext: Option<u64>,
// Arrival, RTP timestamp (extended), PTS (potentially skew-corrected)
base_times: Option<(u64, u64, u64)>,
current_delay: Option<i64>,
observations: Observations,
}
impl Ssrc {
fn new(clock_rate: Option<u32>) -> Self {
Self {
clock_rate,
..Default::default()
}
}
fn reset_times(&mut self) {
self.extended_timestamp = ExtendedTimestamp::default();
self.last_sr_ntp_timestamp = None;
self.last_sr_rtp_ext = None;
self.base_times = None;
self.current_delay = None;
self.observations = Observations::default();
}
/* Returns whether the caller should reset timing associated
* values for this ssrc (eg largest delay) */
fn set_clock_rate(&mut self, clock_rate: u32) -> bool {
if Some(clock_rate) == self.clock_rate {
// No changes
return false;
}
self.clock_rate = Some(clock_rate);
self.reset_times();
true
}
fn add_sender_report(&mut self, rtp_timestamp: u32, ntp_timestamp: u64) {
self.last_sr_rtp_ext = Some(self.extended_timestamp.next(rtp_timestamp));
self.last_sr_ntp_timestamp = Some(ntp_timestamp.into());
// Reset so that the next call to calculate_pts recalculates the NTP / RTP delay
self.current_delay = None;
}
}
#[derive(Debug)]
struct CnameLargestDelay {
largest_delay: i64,
all_sync: bool,
}
/// Govern how to pick presentation timestamps for packets
#[derive(Debug, Default, Clone, Copy, PartialEq, Eq, glib::Enum)]
#[repr(u32)]
#[enum_type(name = "GstRtpBin2TimestampingMode")]
pub enum TimestampingMode {
/// Simply use arrival time as timestamp
#[allow(dead_code)]
#[enum_value(name = "Use arrival time as timestamp", nick = "arrival")]
Arrival,
/// Use RTP timestamps as is
#[allow(dead_code)]
#[enum_value(name = "Use RTP timestamps as is", nick = "rtp")]
Rtp,
/// Correct skew to synchronize sender and receiver clocks
#[default]
#[enum_value(
name = "Correct skew to synchronize sender and receiver clocks",
nick = "skew"
)]
Skew,
}
#[derive(Debug)]
pub struct Context {
ssrcs: HashMap<u32, Ssrc>,
mode: TimestampingMode,
cnames_to_ssrcs: HashMap<Arc<str>, Vec<u32>>,
cname_to_largest_delays: HashMap<Arc<str>, CnameLargestDelay>,
}
impl Context {
pub fn new(mode: TimestampingMode) -> Self {
Self {
ssrcs: HashMap::new(),
mode,
cnames_to_ssrcs: HashMap::new(),
cname_to_largest_delays: HashMap::new(),
}
}
pub fn set_clock_rate(&mut self, ssrc_val: u32, clock_rate: u32) {
if let Some(ssrc) = self.ssrcs.get_mut(&ssrc_val) {
if ssrc.set_clock_rate(clock_rate) {
debug!("{ssrc_val:#08x} times reset after clock rate change");
if let Some(ref cname) = ssrc.cname {
self.cname_to_largest_delays.remove(cname);
}
}
} else {
self.ssrcs.insert(ssrc_val, Ssrc::new(Some(clock_rate)));
}
}
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) {
self.cname_to_largest_delays.remove(cname);
if let Some(ssrcs) = self.cnames_to_ssrcs.get_mut(cname) {
ssrcs.retain(|&other| other != ssrc_val);
}
}
// FIXME: call this on timeouts / BYE (maybe collisions too?)
#[allow(dead_code)]
pub fn remove_ssrc(&mut self, ssrc_val: u32) {
if let Some(ssrc) = self.ssrcs.remove(&ssrc_val) {
debug!("{ssrc_val:#08x} ssrc removed");
if let Some(ref cname) = ssrc.cname {
self.disassociate(ssrc_val, cname)
}
}
}
pub fn associate(&mut self, ssrc_val: u32, cname: &str) {
let ssrc = self
.ssrcs
.entry(ssrc_val)
.or_insert_with(|| Ssrc::new(None));
let cname = Arc::<str>::from(cname);
if let Some(ref old_cname) = ssrc.cname {
if old_cname == &cname {
return;
}
ssrc.cname = Some(cname.clone());
self.disassociate(ssrc_val, cname.as_ref());
} else {
ssrc.cname = Some(cname.clone());
}
let ssrcs = self.cnames_to_ssrcs.entry(cname.clone()).or_default();
ssrcs.push(ssrc_val);
// Recalculate a new largest delay next time calculate_pts is called
self.cname_to_largest_delays.remove(cname.as_ref());
}
pub fn add_sender_report(&mut self, ssrc_val: u32, rtp_timestamp: u32, ntp_timestamp: u64) {
debug!("Adding new sender report for ssrc {ssrc_val:#08x}");
let ssrc = self
.ssrcs
.entry(ssrc_val)
.or_insert_with(|| Ssrc::new(None));
debug!(
"Latest NTP time: {:?}",
NtpTime::from(ntp_timestamp).as_duration().unwrap()
);
ssrc.add_sender_report(rtp_timestamp, ntp_timestamp)
}
pub fn calculate_pts(
&mut self,
ssrc_val: u32,
timestamp: u32,
arrival_time: u64,
) -> (u64, Option<NtpTime>) {
let ssrc = self.ssrcs.get_mut(&ssrc_val).unwrap();
let clock_rate = ssrc.clock_rate.unwrap() as u64;
// Calculate an extended timestamp, calculations only work with extended timestamps
// from that point on
let rtp_ext_ns = ssrc
.extended_timestamp
.next(timestamp)
.mul_div_round(1_000_000_000, clock_rate)
.unwrap();
// Now potentially correct the skew by observing how RTP times and arrival times progress
let mut pts = match self.mode {
TimestampingMode::Skew => {
let (skew_corrected, discont) = ssrc.observations.process(rtp_ext_ns, arrival_time);
trace!(
"{ssrc_val:#08x} using skew corrected RTP ext: {}",
skew_corrected
);
if discont {
ssrc.reset_times();
debug!("{ssrc_val:#08x} times reset after observations discontinuity");
if let Some(ref cname) = ssrc.cname {
self.cname_to_largest_delays.remove(cname);
}
}
skew_corrected
}
TimestampingMode::Rtp => {
trace!("{ssrc_val:#08x} using uncorrected RTP ext: {}", rtp_ext_ns);
rtp_ext_ns
}
TimestampingMode::Arrival => {
trace!("{ssrc_val:#08x} using arrival time: {}", arrival_time);
arrival_time
}
};
// Determine the first arrival time and the first RTP time for that ssrc
if ssrc.base_times.is_none() {
ssrc.base_times = Some((arrival_time, rtp_ext_ns, pts));
}
let (base_arrival_time, base_rtp_ext_ns, base_pts) = ssrc.base_times.unwrap();
// Base the PTS on the first arrival time
pts += base_arrival_time;
trace!("{ssrc_val:#08x} added up base arrival time: {}", pts);
// Now subtract the base PTS we calculated
pts = pts.saturating_sub(base_pts);
trace!("{ssrc_val:#08x} subtracted base PTS: {}", base_pts);
trace!("{ssrc_val:#08x} PTS prior to potential SR offsetting: {pts}");
let mut ntp_time: Option<NtpTime> = None;
// TODO: add property for enabling / disabling offsetting based on
// NTP / RTP mapping, ie inter-stream synchronization
if let Some((last_sr_ntp, last_sr_rtp_ext)) =
ssrc.last_sr_ntp_timestamp.zip(ssrc.last_sr_rtp_ext)
{
let last_sr_rtp_ext_ns = last_sr_rtp_ext
.mul_div_round(1_000_000_000, clock_rate)
.unwrap();
// We have a new SR, we can now figure out an NTP time and calculate how it
// relates to arrival times
if ssrc.current_delay.is_none() {
if let Some(base_ntp_time) = if base_rtp_ext_ns > last_sr_rtp_ext_ns {
let rtp_range_ns = base_rtp_ext_ns - last_sr_rtp_ext_ns;
(last_sr_ntp.as_duration().unwrap().as_nanos() as u64).checked_add(rtp_range_ns)
} else {
let rtp_range_ns = last_sr_rtp_ext_ns - base_rtp_ext_ns;
(last_sr_ntp.as_duration().unwrap().as_nanos() as u64).checked_sub(rtp_range_ns)
} {
trace!(
"{ssrc_val:#08x} Base NTP time on first packet after new SR is {:?} ({:?})",
base_ntp_time,
Duration::from_nanos(base_ntp_time)
);
if base_ntp_time < base_arrival_time {
ssrc.current_delay = Some(base_arrival_time as i64 - base_ntp_time as i64);
} else {
ssrc.current_delay =
Some(-(base_ntp_time as i64 - base_arrival_time as i64));
}
trace!("{ssrc_val:#08x} Current delay is {:?}", ssrc.current_delay);
if let Some(ref cname) = ssrc.cname {
// We should recalculate a new largest delay for this CNAME
self.cname_to_largest_delays.remove(cname);
}
} else {
warn!("{ssrc_val:#08x} Invalid NTP RTP time mapping, waiting for next SR");
ssrc.last_sr_ntp_timestamp = None;
ssrc.last_sr_rtp_ext = None;
}
}
ntp_time = if rtp_ext_ns > last_sr_rtp_ext_ns {
let rtp_range_ns = Duration::from_nanos(rtp_ext_ns - last_sr_rtp_ext_ns);
last_sr_ntp
.as_duration()
.unwrap()
.checked_add(rtp_range_ns)
.map(NtpTime::from_duration)
} else {
let rtp_range_ns = Duration::from_nanos(last_sr_rtp_ext_ns - rtp_ext_ns);
last_sr_ntp
.as_duration()
.unwrap()
.checked_sub(rtp_range_ns)
.map(NtpTime::from_duration)
};
}
// Finally, if we have a CNAME for this SSRC and we have managed to calculate
// a delay for all the other ssrcs for this CNAME, we can calculate by how much
// we need to delay this stream to sync it with the others, if at all.
if let Some(cname) = ssrc.cname.clone() {
let delay = ssrc.current_delay;
let cname_largest_delay = self
.cname_to_largest_delays
.entry(cname.clone())
.or_insert_with(|| {
let mut cname_largest_delay = CnameLargestDelay {
largest_delay: std::i64::MIN,
all_sync: true,
};
trace!("{ssrc_val:#08x} searching for new largest delay");
let ssrc_vals = self.cnames_to_ssrcs.get(&cname).unwrap();
for ssrc_val in ssrc_vals {
let ssrc = self.ssrcs.get(ssrc_val).unwrap();
if let Some(delay) = ssrc.current_delay {
trace!("ssrc {ssrc_val:#08x} has delay {delay}",);
if delay > cname_largest_delay.largest_delay {
cname_largest_delay.largest_delay = delay;
}
} else {
trace!("{ssrc_val:#08x} has no delay calculated yet");
cname_largest_delay.all_sync = false;
}
}
cname_largest_delay
});
trace!("{ssrc_val:#08x} Largest delay is {:?}", cname_largest_delay);
if cname_largest_delay.all_sync {
let offset = (cname_largest_delay.largest_delay - delay.unwrap()) as u64;
trace!("{ssrc_val:#08x} applying offset {}", offset);
pts += offset;
}
}
debug!("{ssrc_val:#08x} calculated PTS {pts}");
(pts, ntp_time)
}
}
const WINDOW_LENGTH: u64 = 512;
const WINDOW_DURATION: u64 = 2_000_000_000;
#[derive(Debug)]
struct Observations {
base_local_time: Option<u64>,
base_remote_time: Option<u64>,
highest_remote_time: Option<u64>,
deltas: VecDeque<i64>,
min_delta: i64,
skew: i64,
filling: bool,
window_size: usize,
}
impl Default for Observations {
fn default() -> Self {
Self {
base_local_time: None,
base_remote_time: None,
highest_remote_time: None,
deltas: VecDeque::new(),
min_delta: 0,
skew: 0,
filling: true,
window_size: 0,
}
}
}
impl Observations {
fn out_time(&self, base_local_time: u64, remote_diff: u64) -> (u64, bool) {
let out_time = base_local_time + remote_diff;
let out_time = if self.skew < 0 {
out_time.saturating_sub((-self.skew) as u64)
} else {
out_time + (self.skew as u64)
};
trace!("Skew {}, min delta {}", self.skew, self.min_delta);
trace!("Outputting {}", out_time);
(out_time, false)
}
// Based on the algorithm used in GStreamer's rtpjitterbuffer, which comes from
// Fober, Orlarey and Letz, 2005, "Real Time Clock Skew Estimation over Network Delays":
// http://citeseerx.ist.psu.edu/viewdoc/summary?doi=10.1.1.102.1546
fn process(&mut self, remote_time: u64, local_time: u64) -> (u64, bool) {
trace!("Local time {}, remote time {}", local_time, remote_time,);
let (base_remote_time, base_local_time) =
match (self.base_remote_time, self.base_local_time) {
(Some(remote), Some(local)) => (remote, local),
_ => {
debug!(
"Initializing base time: local {}, remote {}",
local_time, remote_time,
);
self.base_remote_time = Some(remote_time);
self.base_local_time = Some(local_time);
self.highest_remote_time = Some(remote_time);
return (local_time, false);
}
};
let highest_remote_time = self.highest_remote_time.unwrap();
let remote_diff = remote_time.saturating_sub(base_remote_time);
/* Only update observations when remote times progress forward */
if remote_time <= highest_remote_time {
return self.out_time(base_local_time, remote_diff);
}
self.highest_remote_time = Some(remote_time);
let local_diff = local_time.saturating_sub(base_local_time);
let delta = (local_diff as i64) - (remote_diff as i64);
trace!(
"Local diff {}, remote diff {}, delta {}",
local_diff,
remote_diff,
delta,
);
if remote_diff > 0 && local_diff > 0 {
let slope = (local_diff as f64) / (remote_diff as f64);
if !(0.8..1.2).contains(&slope) {
warn!("Too small/big slope {}, resetting", slope);
let discont = !self.deltas.is_empty();
*self = Observations::default();
debug!(
"Initializing base time: local {}, remote {}",
local_time, remote_time,
);
self.base_remote_time = Some(remote_time);
self.base_local_time = Some(local_time);
self.highest_remote_time = Some(remote_time);
return (local_time, discont);
}
}
if (delta > self.skew && delta - self.skew > 1_000_000_000)
|| (delta < self.skew && self.skew - delta > 1_000_000_000)
{
warn!("Delta {} too far from skew {}, resetting", delta, self.skew);
let discont = !self.deltas.is_empty();
*self = Observations::default();
debug!(
"Initializing base time: local {}, remote {}",
local_time, remote_time,
);
self.base_remote_time = Some(remote_time);
self.base_local_time = Some(local_time);
self.highest_remote_time = Some(remote_time);
return (local_time, discont);
}
if self.filling {
if self.deltas.is_empty() || delta < self.min_delta {
self.min_delta = delta;
}
self.deltas.push_back(delta);
if remote_diff > WINDOW_DURATION || self.deltas.len() as u64 == WINDOW_LENGTH {
self.window_size = self.deltas.len();
self.skew = self.min_delta;
self.filling = false;
} else {
let perc_time = remote_diff.mul_div_floor(100, WINDOW_DURATION).unwrap() as i64;
let perc_window = (self.deltas.len() as u64)
.mul_div_floor(100, WINDOW_LENGTH)
.unwrap() as i64;
let perc = std::cmp::max(perc_time, perc_window);
self.skew = (perc * self.min_delta + ((10_000 - perc) * self.skew)) / 10_000;
}
} else {
let old = self.deltas.pop_front().unwrap();
self.deltas.push_back(delta);
if delta <= self.min_delta {
self.min_delta = delta;
} else if old == self.min_delta {
self.min_delta = self.deltas.iter().copied().min().unwrap();
}
self.skew = (self.min_delta + (124 * self.skew)) / 125;
}
self.out_time(base_local_time, remote_diff)
}
}
#[cfg(test)]
pub(crate) mod tests {
use super::*;
use crate::rtpbin2::session::tests::init_logs;
use crate::rtpbin2::time::system_time_to_ntp_time_u64;
#[test]
fn test_single_stream_no_sr() {
init_logs();
let mut ctx = Context::new(TimestampingMode::Rtp);
let mut now = 0;
ctx.set_clock_rate(0x12345678, 90000);
assert_eq!(ctx.calculate_pts(0x12345678, 0, now), (0, None));
now += 1_000_000_000;
assert_eq!(
ctx.calculate_pts(0x12345678, 90000, now),
(1_000_000_000, None)
);
}
#[test]
fn test_single_stream_with_sr() {
init_logs();
let mut ctx = Context::new(TimestampingMode::Rtp);
let mut now = 0;
ctx.set_clock_rate(0x12345678, 90000);
ctx.add_sender_report(
0x12345678,
0,
system_time_to_ntp_time_u64(std::time::UNIX_EPOCH).as_u64(),
);
assert_eq!(
ctx.calculate_pts(0x12345678, 0, now),
(0, Some(system_time_to_ntp_time_u64(std::time::UNIX_EPOCH)))
);
now += 1_000_000_000;
assert_eq!(
ctx.calculate_pts(0x12345678, 90000, now),
(
1_000_000_000,
Some(system_time_to_ntp_time_u64(
std::time::UNIX_EPOCH + Duration::from_millis(1000)
))
)
);
}
#[test]
fn test_two_streams_with_sr() {
init_logs();
let mut ctx = Context::new(TimestampingMode::Rtp);
let mut now = 0;
ctx.set_clock_rate(0x12345, 90000);
ctx.set_clock_rate(0x67890, 90000);
ctx.associate(0x12345, "foo@bar");
ctx.associate(0x67890, "foo@bar");
ctx.add_sender_report(
0x12345,
0,
system_time_to_ntp_time_u64(std::time::UNIX_EPOCH).as_u64(),
);
ctx.add_sender_report(
0x67890,
0,
system_time_to_ntp_time_u64(std::time::UNIX_EPOCH + Duration::from_millis(500))
.as_u64(),
);
// NTP time 0
assert_eq!(
ctx.calculate_pts(0x12345, 0, now),
(0, Some(system_time_to_ntp_time_u64(std::time::UNIX_EPOCH)))
);
now += 500_000_000;
// NTP time 500, arrival time 500
assert_eq!(
ctx.calculate_pts(0x12345, 45000, now),
(
500_000_000,
Some(system_time_to_ntp_time_u64(
std::time::UNIX_EPOCH + Duration::from_millis(500)
))
)
);
// NTP time 500, arrival time 500
assert_eq!(
ctx.calculate_pts(0x67890, 0, now),
(
500_000_000,
Some(system_time_to_ntp_time_u64(
std::time::UNIX_EPOCH + Duration::from_millis(500)
))
)
);
now += 500_000_000;
// NTP time 1000, arrival time 1000
assert_eq!(
ctx.calculate_pts(0x12345, 90000, now),
(
1_000_000_000,
Some(system_time_to_ntp_time_u64(
std::time::UNIX_EPOCH + Duration::from_millis(1000)
))
)
);
now += 500_000_000;
// NTP time 1500, arrival time 1500
assert_eq!(
ctx.calculate_pts(0x67890, 90000, now),
(
1_500_000_000,
Some(system_time_to_ntp_time_u64(
std::time::UNIX_EPOCH + Duration::from_millis(1500)
))
)
);
}
#[test]
fn test_two_streams_no_sr_and_offset_arrival_times() {
init_logs();
let mut ctx = Context::new(TimestampingMode::Rtp);
let mut now = 0;
ctx.set_clock_rate(0x12345, 90000);
ctx.set_clock_rate(0x67890, 90000);
ctx.associate(0x12345, "foo@bar");
ctx.associate(0x67890, "foo@bar");
assert_eq!(ctx.calculate_pts(0x12345, 0, now), (0, None));
now += 500_000_000;
assert_eq!(ctx.calculate_pts(0x67890, 0, now), (500_000_000, None));
assert_eq!(ctx.calculate_pts(0x12345, 45000, now), (500_000_000, None));
}
#[test]
fn test_two_streams_with_same_sr_and_offset_arrival_times() {
init_logs();
let mut ctx = Context::new(TimestampingMode::Rtp);
let mut now = 0;
ctx.set_clock_rate(0x12345, 90000);
ctx.set_clock_rate(0x67890, 90000);
ctx.associate(0x12345, "foo@bar");
ctx.associate(0x67890, "foo@bar");
ctx.add_sender_report(
0x12345,
0,
system_time_to_ntp_time_u64(std::time::UNIX_EPOCH).as_u64(),
);
ctx.add_sender_report(
0x67890,
0,
system_time_to_ntp_time_u64(std::time::UNIX_EPOCH).as_u64(),
);
assert_eq!(
ctx.calculate_pts(0x12345, 0, now),
(0, Some(system_time_to_ntp_time_u64(std::time::UNIX_EPOCH)))
);
now += 500_000_000;
assert_eq!(
ctx.calculate_pts(0x67890, 0, now),
(
500_000_000,
Some(system_time_to_ntp_time_u64(std::time::UNIX_EPOCH))
)
);
assert_eq!(
ctx.calculate_pts(0x12345, 45000, now),
(
1_000_000_000,
Some(system_time_to_ntp_time_u64(
std::time::UNIX_EPOCH + Duration::from_millis(500)
))
)
);
now += 500_000_000;
assert_eq!(
ctx.calculate_pts(0x67890, 45000, now),
(
1_000_000_000,
Some(system_time_to_ntp_time_u64(
std::time::UNIX_EPOCH + Duration::from_millis(500)
))
)
);
// Now remove the delayed source and observe that the offset is gone
// for the other source
ctx.remove_ssrc(0x67890);
assert_eq!(
ctx.calculate_pts(0x12345, 90000, now),
(
1_000_000_000,
Some(system_time_to_ntp_time_u64(
std::time::UNIX_EPOCH + Duration::from_millis(1000)
))
)
);
}
#[test]
fn test_two_streams_with_sr_different_cnames() {
init_logs();
let mut ctx = Context::new(TimestampingMode::Rtp);
let mut now = 0;
ctx.set_clock_rate(0x12345, 90000);
ctx.set_clock_rate(0x67890, 90000);
ctx.associate(0x12345, "foo@bar");
ctx.associate(0x67890, "foo@baz");
ctx.add_sender_report(
0x12345,
0,
system_time_to_ntp_time_u64(std::time::UNIX_EPOCH).as_u64(),
);
ctx.add_sender_report(
0x67890,
0,
system_time_to_ntp_time_u64(std::time::UNIX_EPOCH).as_u64(),
);
assert_eq!(
ctx.calculate_pts(0x12345, 0, now),
(0, Some(system_time_to_ntp_time_u64(std::time::UNIX_EPOCH)))
);
now += 500_000_000;
assert_eq!(
ctx.calculate_pts(0x67890, 0, now),
(
500_000_000,
Some(system_time_to_ntp_time_u64(std::time::UNIX_EPOCH))
)
);
assert_eq!(
ctx.calculate_pts(0x12345, 45000, now),
(
500_000_000,
Some(system_time_to_ntp_time_u64(
std::time::UNIX_EPOCH + Duration::from_millis(500)
))
)
);
now += 500_000_000;
assert_eq!(
ctx.calculate_pts(0x67890, 45000, now),
(
1_000_000_000,
Some(system_time_to_ntp_time_u64(
std::time::UNIX_EPOCH + Duration::from_millis(500)
))
)
);
}
}

View file

@ -0,0 +1,62 @@
// SPDX-License-Identifier: MPL-2.0
use std::{
ops::{Add, Sub},
time::{Duration, SystemTime},
};
// time between the NTP time at 1900-01-01 and the unix EPOCH (1970-01-01)
const NTP_OFFSET: Duration = Duration::from_secs((365 * 70 + 17) * 24 * 60 * 60);
// 2^32
const F32: f64 = 4_294_967_296.0;
#[derive(Clone, Copy, Debug, PartialEq, Eq, PartialOrd, Ord)]
pub struct NtpTime(u64);
impl NtpTime {
pub fn from_duration(dur: Duration) -> Self {
Self((dur.as_secs_f64() * F32) as u64)
}
pub fn as_duration(&self) -> Result<Duration, std::time::TryFromFloatSecsError> {
Duration::try_from_secs_f64(self.0 as f64 / F32)
}
pub fn as_u32(self) -> u32 {
((self.0 >> 16) & 0xffffffff) as u32
}
pub fn as_u64(self) -> u64 {
self.0
}
}
impl Sub for NtpTime {
type Output = NtpTime;
fn sub(self, rhs: Self) -> Self::Output {
NtpTime(self.0 - rhs.0)
}
}
impl Add for NtpTime {
type Output = NtpTime;
fn add(self, rhs: Self) -> Self::Output {
NtpTime(self.0 + rhs.0)
}
}
pub fn system_time_to_ntp_time_u64(time: SystemTime) -> NtpTime {
let dur = time
.duration_since(SystemTime::UNIX_EPOCH)
.expect("time is before unix epoch?!")
+ NTP_OFFSET;
NtpTime::from_duration(dur)
}
impl From<u64> for NtpTime {
fn from(value: u64) -> Self {
NtpTime(value)
}
}

639
net/rtp/src/utils.rs Normal file
View file

@ -0,0 +1,639 @@
/// Defines a comparable new type `$typ` on a `[std::num::Wrapping]::<u32>`.
///
/// The new type will wrap-around on additions and substractions and it comparison
/// operators take the wrapping in consideration.
///
/// The comparison algorithm uses [serial number arithmetic](serial-number-arithmetic).
/// The limit being that it can't tell whether 0x8000_0000 is greater or less than 0.
///
/// # Examples
///
/// ```rust
/// # use gstrsrtp::define_wrapping_comparable_u32;
///
/// /// Error type to return when comparing 0x8000_0000 to 0.
/// struct RTPTimestampComparisonLimit;
///
/// /// Define the new type comparable and wrapping `u32` `RTPTimestamp`:
/// define_wrapping_comparable_u32!(RTPTimestamp, RTPTimestampComparisonLimit);
///
/// let ts0 = RTPTimestamp::ZERO;
/// assert!(ts0.is_zero());
///
/// let mut ts = ts0;
/// ts += 1;
/// assert_eq!(*ts, 1);
/// assert_eq!(RTPTimestamp::MAX + ts, ts0);
///
/// let ts2: RTPTimestamp = 2.into();
/// assert_eq!(*ts2, 2);
/// assert_eq!(ts - ts2, RTPTimestamp::MAX);
/// ```
///
/// [serial-number-arithmetic]: http://en.wikipedia.org/wiki/Serial_number_arithmetic
/// ```
#[macro_export]
macro_rules! define_wrapping_comparable_u32 {
($typ:ident) => {
#[derive(Clone, Copy, Debug, Default)]
pub struct $typ(std::num::Wrapping<u32>);
impl $typ {
pub const ZERO: $typ = $typ(std::num::Wrapping(0));
pub const MIN: $typ = $typ(std::num::Wrapping(u32::MIN));
pub const MAX: $typ = $typ(std::num::Wrapping(u32::MAX));
pub const NONE: Option<$typ> = None;
#[inline]
pub const fn new(val: u32) -> Self {
Self(std::num::Wrapping((val)))
}
#[inline]
pub fn from_ext(ext_val: u64) -> Self {
Self(std::num::Wrapping((ext_val & 0xffff_ffff) as u32))
}
#[inline]
pub fn is_zero(self) -> bool {
self.0 .0 == 0
}
#[inline]
pub fn distance(self, other: Self) -> Option<i32> {
self.distance_u32(other.0 .0)
}
#[inline]
pub fn distance_u32(self, other: u32) -> Option<i32> {
// See http://en.wikipedia.org/wiki/Serial_number_arithmetic
let this = i32::from_ne_bytes(self.0 .0.to_ne_bytes());
let other = i32::from_ne_bytes(other.to_ne_bytes());
match this.wrapping_sub(other) {
-0x8000_0000 => {
// This is the limit of the algorithm:
// arguments are too far away to determine the result sign,
// i.e. which one is greater than the other
None
}
delta => Some(delta),
}
}
}
impl From<u32> for $typ {
fn from(value: u32) -> Self {
Self(std::num::Wrapping(value))
}
}
impl From<$typ> for u32 {
fn from(value: $typ) -> Self {
value.0 .0
}
}
impl std::ops::Deref for $typ {
type Target = u32;
fn deref(&self) -> &u32 {
&self.0 .0
}
}
impl std::ops::Add for $typ {
type Output = Self;
fn add(self, rhs: Self) -> Self {
Self(self.0.add(rhs.0))
}
}
impl std::ops::Add<u32> for $typ {
type Output = Self;
fn add(self, rhs: u32) -> Self {
Self(self.0.add(std::num::Wrapping(rhs)))
}
}
impl std::ops::Add<i32> for $typ {
type Output = Self;
fn add(self, rhs: i32) -> Self {
// See http://en.wikipedia.org/wiki/Serial_number_arithmetic
let this = i32::from_ne_bytes(self.0 .0.to_ne_bytes());
let res = this.wrapping_add(rhs);
let res = u32::from_ne_bytes(res.to_ne_bytes());
Self(std::num::Wrapping(res))
}
}
impl std::ops::AddAssign for $typ {
fn add_assign(&mut self, rhs: Self) {
self.0.add_assign(rhs.0);
}
}
impl std::ops::AddAssign<u32> for $typ {
fn add_assign(&mut self, rhs: u32) {
self.0.add_assign(std::num::Wrapping(rhs));
}
}
impl std::ops::AddAssign<i32> for $typ {
fn add_assign(&mut self, rhs: i32) {
*self = *self + rhs;
}
}
impl std::ops::Sub for $typ {
type Output = Self;
fn sub(self, rhs: Self) -> Self {
self.sub(rhs.0 .0)
}
}
impl std::ops::Sub<u32> for $typ {
type Output = Self;
fn sub(self, rhs: u32) -> Self {
Self(self.0.sub(std::num::Wrapping(rhs)))
}
}
impl std::ops::SubAssign for $typ {
fn sub_assign(&mut self, rhs: Self) {
self.sub_assign(rhs.0 .0);
}
}
impl std::ops::SubAssign<u32> for $typ {
fn sub_assign(&mut self, rhs: u32) {
self.0.sub_assign(std::num::Wrapping(rhs));
}
}
impl std::cmp::PartialEq for $typ {
fn eq(&self, other: &Self) -> bool {
self.0 .0 == other.0 .0
}
}
impl std::cmp::PartialEq<u32> for $typ {
fn eq(&self, other: &u32) -> bool {
self.0 .0 == *other
}
}
impl std::cmp::Eq for $typ {}
impl std::cmp::PartialOrd for $typ {
fn partial_cmp(&self, other: &Self) -> Option<std::cmp::Ordering> {
self.distance(*other).map(|d| d.cmp(&0))
}
}
impl gst::prelude::OptionOperations for $typ {}
};
($typ:ident, $comp_err_type:ident) => {
define_wrapping_comparable_u32!($typ);
impl $typ {
#[inline]
pub fn try_cmp(&self, other: $typ) -> Result<std::cmp::Ordering, $comp_err_type> {
self.partial_cmp(&other).ok_or($comp_err_type)
}
}
};
($typ:ident, $err_enum:ty, $comp_err_variant:ident) => {
define_wrapping_comparable_u32!($typ);
impl $typ {
#[inline]
pub fn try_cmp(&self, other: $typ) -> Result<std::cmp::Ordering, $err_enum> {
self.partial_cmp(&other)
.ok_or(<$err_enum>::$comp_err_variant)
}
}
};
}
#[macro_export]
macro_rules! define_wrapping_comparable_u32_with_display {
($typ:ident, impl) => {
impl std::fmt::Display for $typ {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.write_fmt(format_args!("{}", self.0 .0))
}
}
};
($typ:ident) => {
define_wrapping_comparable_u32!($typ);
define_wrapping_comparable_u32_with_display!($typ, impl);
};
($typ:ident, $comp_err_type:ty) => {
define_wrapping_comparable_u32!($typ, $comp_err_type);
define_wrapping_comparable_u32_with_display!($typ, impl);
};
($typ:ident, $err_enum:ty, $comp_err_variant:ident,) => {
define_wrapping_comparable_u32!($typ, $err_enum, $comp_err_variant);
define_wrapping_comparable_u32_with_display!($typ, impl);
};
}
/// Stores information necessary to compute a series of extended timestamps
#[derive(Default, Debug)]
pub(crate) struct ExtendedTimestamp {
last_ext: Option<u64>,
}
impl ExtendedTimestamp {
/// Produces the next extended timestamp from a new RTP timestamp
pub(crate) fn next(&mut self, rtp_timestamp: u32) -> u64 {
let ext = match self.last_ext {
None => (1u64 << 32) + rtp_timestamp as u64,
Some(last_ext) => {
// pick wraparound counter from previous timestamp and add to new timestamp
let mut ext = rtp_timestamp as u64 + (last_ext & !0xffffffff);
// check for timestamp wraparound
if ext < last_ext {
let diff = last_ext - ext;
if diff > std::i32::MAX as u64 {
// timestamp went backwards more than allowed, we wrap around and get
// updated extended timestamp.
ext += 1u64 << 32;
}
} else {
let diff = ext - last_ext;
if diff > std::i32::MAX as u64 {
if ext < 1u64 << 32 {
// We can't ever get to such a case as our counter is opaque
unreachable!()
} else {
ext -= 1u64 << 32;
// We don't want the extended timestamp storage to go back, ever
return ext;
}
}
}
ext
}
};
self.last_ext = Some(ext);
ext
}
}
/// Stores information necessary to compute a series of extended seqnums
#[derive(Default, Debug)]
pub(crate) struct ExtendedSeqnum {
last_ext: Option<u64>,
}
impl ExtendedSeqnum {
/// Produces the next extended timestamp from a new RTP timestamp
pub(crate) fn next(&mut self, rtp_seqnum: u16) -> u64 {
let ext = match self.last_ext {
None => (1u64 << 16) + rtp_seqnum as u64,
Some(last_ext) => {
// pick wraparound counter from previous timestamp and add to new timestamp
let mut ext = rtp_seqnum as u64 + (last_ext & !0xffff);
// check for timestamp wraparound
if ext < last_ext {
let diff = last_ext - ext;
if diff > std::i16::MAX as u64 {
// timestamp went backwards more than allowed, we wrap around and get
// updated extended timestamp.
ext += 1u64 << 16;
}
} else {
let diff = ext - last_ext;
if diff > std::i16::MAX as u64 {
if ext < 1u64 << 16 {
// We can't ever get to such a case as our counter is opaque
unreachable!()
} else {
ext -= 1u64 << 16;
// We don't want the extended timestamp storage to go back, ever
return ext;
}
}
}
ext
}
};
self.last_ext = Some(ext);
ext
}
}
#[cfg(test)]
mod tests {
use super::*;
define_wrapping_comparable_u32!(MyWrapper);
#[test]
fn wrapping_u32_basics() {
let zero = MyWrapper::ZERO;
let one = MyWrapper::from(1);
let two = MyWrapper::from(2);
assert_eq!(u32::from(zero), 0);
assert!(zero.is_zero());
assert_eq!(u32::from(one), 1);
assert_eq!(u32::from(two), 2);
let max_plus_1_u64 = MyWrapper::from_ext((u32::MAX as u64) + 1);
assert_eq!(max_plus_1_u64, MyWrapper::ZERO);
}
#[test]
fn add_wrapping_u32() {
let one = MyWrapper::from(1);
let two = MyWrapper::from(2);
assert_eq!(MyWrapper::ZERO + one, one);
assert_eq!(MyWrapper::ZERO + 1u32, one);
assert_eq!(one + one, two);
assert_eq!(one + 1u32, two);
assert_eq!(MyWrapper::MAX + MyWrapper::ZERO, MyWrapper::MAX);
assert_eq!(MyWrapper::MAX + one, MyWrapper::ZERO);
assert_eq!(MyWrapper::MAX + two, one);
let mut var = MyWrapper::ZERO;
assert!(var.is_zero());
var += 1;
assert_eq!(var, one);
var += one;
assert_eq!(var, two);
let mut var = MyWrapper::MAX;
var += 1;
assert!(var.is_zero());
var += one;
assert_eq!(var, one);
}
#[test]
fn add_wrapping_u32_i32() {
let one = MyWrapper::from(1);
assert_eq!(MyWrapper::ZERO + 1i32, one);
assert_eq!(MyWrapper::ZERO + -1i32, MyWrapper::MAX);
assert_eq!(MyWrapper::MAX + 1i32, MyWrapper::ZERO);
assert_eq!(MyWrapper::MAX + 2i32, one);
assert_eq!(
MyWrapper::from(0x8000_0000) + -0i32,
MyWrapper::from(0x8000_0000)
);
assert_eq!(
MyWrapper::from(0x8000_0000) + 1i32,
MyWrapper::from(0x8000_0001)
);
assert_eq!(
MyWrapper::from(0x8000_0000) + -1i32,
MyWrapper::from(0x7fff_ffff)
);
assert_eq!(
MyWrapper::from(0x7fff_ffff) + 1i32,
MyWrapper::from(0x8000_0000)
);
assert_eq!(MyWrapper::ZERO + i32::MIN, MyWrapper::from(0x8000_0000));
let mut var = MyWrapper::ZERO;
var += 1i32;
assert_eq!(var, one);
let mut var = MyWrapper::ZERO;
var += -1i32;
assert_eq!(var, MyWrapper::MAX);
let mut var = MyWrapper::MAX;
var += 1;
assert_eq!(var, MyWrapper::ZERO);
}
#[test]
fn sub_wrapping_u32() {
let one = MyWrapper::from(1);
assert_eq!(MyWrapper::ZERO - MyWrapper::ZERO, MyWrapper::ZERO);
assert_eq!(MyWrapper::MAX - MyWrapper::MAX, MyWrapper::ZERO);
assert_eq!(MyWrapper::ZERO - one, MyWrapper::MAX);
assert_eq!(MyWrapper::ZERO - MyWrapper::MAX, one);
assert_eq!(
MyWrapper::ZERO - MyWrapper::from(0x8000_0000),
MyWrapper::from(0x8000_0000)
);
assert_eq!(
MyWrapper::from(0x8000_0000) - MyWrapper::ZERO,
MyWrapper::from(0x8000_0000)
);
let mut var = MyWrapper::ZERO;
assert!(var.is_zero());
var -= 1;
assert_eq!(var, MyWrapper::MAX);
let mut var = MyWrapper::MAX;
var -= MyWrapper::MAX;
assert!(var.is_zero());
}
#[test]
fn compare_wrapping_u32() {
use std::cmp::Ordering::*;
#[derive(Debug, PartialEq)]
pub struct ComparisonLimit;
define_wrapping_comparable_u32!(MyWrapper, ComparisonLimit);
let cmp = |a: u32, b: u32| MyWrapper::from(a).partial_cmp(&MyWrapper::from(b));
let try_cmp = |a: u32, b: u32| MyWrapper::from(a).try_cmp(MyWrapper::from(b));
assert_eq!(cmp(0, 1).unwrap(), Less);
assert_eq!(try_cmp(0, 1), Ok(Less));
assert_eq!(cmp(1, 1).unwrap(), Equal);
assert_eq!(try_cmp(1, 1), Ok(Equal));
assert_eq!(cmp(1, 0).unwrap(), Greater);
assert_eq!(try_cmp(1, 0), Ok(Greater));
assert_eq!(cmp(0x7fff_ffff, 0).unwrap(), Greater);
assert_eq!(try_cmp(0x7fff_ffff, 0), Ok(Greater));
assert_eq!(cmp(0xffff_ffff, 0).unwrap(), Less);
assert_eq!(try_cmp(0xffff_ffff, 0), Ok(Less));
assert_eq!(cmp(0, 0x7fff_ffff).unwrap(), Less);
assert_eq!(try_cmp(0, 0x7fff_ffff), Ok(Less));
assert_eq!(cmp(0, 0xffff_ffff).unwrap(), Greater);
assert_eq!(try_cmp(0, 0xffff_ffff), Ok(Greater));
// This is the limit of the algorithm:
assert!(cmp(0x8000_0000, 0).is_none());
assert!(cmp(0, 0x8000_0000).is_none());
assert_eq!(try_cmp(0x8000_0000, 0), Err(ComparisonLimit));
assert_eq!(try_cmp(0, 0x8000_0000), Err(ComparisonLimit));
}
#[test]
fn extended_timestamp_basic() {
let mut ext_ts = ExtendedTimestamp::default();
// No wraparound when timestamps are increasing
assert_eq!(ext_ts.next(0), (1 << 32));
assert_eq!(ext_ts.next(10), (1 << 32) + 10);
assert_eq!(ext_ts.next(10), (1 << 32) + 10);
assert_eq!(
ext_ts.next(1 + std::i32::MAX as u32),
(1 << 32) + 1 + std::i32::MAX as u64
);
// Even big bumps under G_MAXINT32 don't result in wrap-around
ext_ts = ExtendedTimestamp::default();
assert_eq!(ext_ts.next(1087500), (1 << 32) + 1087500);
assert_eq!(ext_ts.next(24), (1 << 32) + 24);
}
#[test]
fn extended_timestamp_wraparound() {
let mut ext_ts = ExtendedTimestamp::default();
assert_eq!(
ext_ts.next(std::u32::MAX - 90000 + 1),
(1 << 32) + std::u32::MAX as u64 - 90000 + 1
);
assert_eq!(ext_ts.next(0), (1 << 32) + std::u32::MAX as u64 + 1);
assert_eq!(
ext_ts.next(90000),
(1 << 32) + std::u32::MAX as u64 + 1 + 90000
);
}
#[test]
fn extended_timestamp_wraparound_disordered() {
let mut ext_ts = ExtendedTimestamp::default();
assert_eq!(
ext_ts.next(std::u32::MAX - 90000 + 1),
(1 << 32) + std::u32::MAX as u64 - 90000 + 1
);
assert_eq!(ext_ts.next(0), (1 << 32) + std::u32::MAX as u64 + 1);
// Unwrapping around
assert_eq!(
ext_ts.next(std::u32::MAX - 90000 + 1),
(1 << 32) + std::u32::MAX as u64 - 90000 + 1
);
assert_eq!(
ext_ts.next(90000),
(1 << 32) + std::u32::MAX as u64 + 1 + 90000
);
}
#[test]
fn extended_timestamp_wraparound_disordered_backwards() {
let mut ext_ts = ExtendedTimestamp::default();
assert_eq!(ext_ts.next(90000), (1 << 32) + 90000);
// Wraps backwards
assert_eq!(
ext_ts.next(std::u32::MAX - 90000 + 1),
std::u32::MAX as u64 - 90000 + 1
);
// Wraps again forwards
assert_eq!(ext_ts.next(90000), (1 << 32) + 90000);
}
#[test]
fn extended_seqnum_basic() {
let mut ext_seq = ExtendedSeqnum::default();
// No wraparound when seqnums are increasing
assert_eq!(ext_seq.next(0), (1 << 16));
assert_eq!(ext_seq.next(10), (1 << 16) + 10);
assert_eq!(ext_seq.next(10), (1 << 16) + 10);
assert_eq!(
ext_seq.next(1 + std::i16::MAX as u16),
(1 << 16) + 1 + std::i16::MAX as u64
);
// Even big bumps under MAXINT16 don't result in wrap-around
ext_seq = ExtendedSeqnum::default();
assert_eq!(ext_seq.next(27500), (1 << 16) + 27500);
assert_eq!(ext_seq.next(24), (1 << 16) + 24);
}
#[test]
fn extended_seqnum_wraparound() {
let mut ext_seq = ExtendedSeqnum::default();
assert_eq!(
ext_seq.next(std::u16::MAX - 9000 + 1),
(1 << 16) + std::u16::MAX as u64 - 9000 + 1
);
assert_eq!(ext_seq.next(0), (1 << 16) + std::u16::MAX as u64 + 1);
assert_eq!(
ext_seq.next(9000),
(1 << 16) + std::u16::MAX as u64 + 1 + 9000
);
}
#[test]
fn extended_seqnum_wraparound_disordered() {
let mut ext_seq = ExtendedSeqnum::default();
assert_eq!(
ext_seq.next(std::u16::MAX - 9000 + 1),
(1 << 16) + std::u16::MAX as u64 - 9000 + 1
);
assert_eq!(ext_seq.next(0), (1 << 16) + std::u16::MAX as u64 + 1);
// Unwrapping around
assert_eq!(
ext_seq.next(std::u16::MAX - 9000 + 1),
(1 << 16) + std::u16::MAX as u64 - 9000 + 1
);
assert_eq!(
ext_seq.next(9000),
(1 << 16) + std::u16::MAX as u64 + 1 + 9000
);
}
#[test]
fn extended_seqnum_wraparound_disordered_backwards() {
let mut ext_seq = ExtendedSeqnum::default();
assert_eq!(ext_seq.next(9000), (1 << 16) + 9000);
// Wraps backwards
assert_eq!(
ext_seq.next(std::u16::MAX - 9000 + 1),
std::u16::MAX as u64 - 9000 + 1
);
// Wraps again forwards
assert_eq!(ext_seq.next(9000), (1 << 16) + 9000);
}
}

262
net/rtp/tests/rtpbin2.rs Normal file
View file

@ -0,0 +1,262 @@
//
// Copyright (C) 2023 Matthew Waters <matthew@centricular.com>
//
// This Source Code Form is subject to the terms of the Mozilla Public License, v2.0.
// If a copy of the MPL was not distributed with this file, You can obtain one at
// <https://mozilla.org/MPL/2.0/>.
//
// SPDX-License-Identifier: MPL-2.0
use std::sync::{atomic::AtomicUsize, Arc, Mutex};
use gst::{prelude::*, Caps};
use gst_check::Harness;
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() {
use std::sync::Once;
static INIT: Once = Once::new();
INIT.call_once(|| {
gst::init().unwrap();
gstrsrtp::plugin_register_static().expect("rtpbin2 test");
});
}
const TEST_SSRC: u32 = 0x12345678;
const TEST_PT: u8 = 96;
const TEST_CLOCK_RATE: u32 = 48000;
fn generate_rtp_buffer(seqno: u16, rtpts: u32, payload_len: usize) -> gst::Buffer {
let payload = vec![4; payload_len];
let packet = RtpPacketBuilder::new()
.ssrc(TEST_SSRC)
.payload_type(TEST_PT)
.sequence_number(seqno)
.timestamp(rtpts)
.payload(payload.as_slice());
let size = packet.calculate_size().unwrap();
let mut data = vec![0; size];
packet.write_into(&mut data).unwrap();
gst::Buffer::from_mut_slice(data)
}
#[test]
fn test_send() {
init();
let id = next_element_counter();
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();
let caps = Caps::builder("application/x-rtp")
.field("media", "audio")
.field("payload", TEST_PT as i32)
.field("clock-rate", TEST_CLOCK_RATE as i32)
.field("encoding-name", "custom-test")
.build();
h.set_src_caps(caps);
h.push(generate_rtp_buffer(500, 20, 9)).unwrap();
h.push(generate_rtp_buffer(501, 30, 11)).unwrap();
let buffer = h.pull().unwrap();
let mapped = buffer.map_readable().unwrap();
let rtp = rtp_types::RtpPacket::parse(&mapped).unwrap();
assert_eq!(rtp.sequence_number(), 500);
let buffer = h.pull().unwrap();
let mapped = buffer.map_readable().unwrap();
let rtp = rtp_types::RtpPacket::parse(&mapped).unwrap();
assert_eq!(rtp.sequence_number(), 501);
let stats = h.element().unwrap().property::<gst::Structure>("stats");
let session_stats = stats.get::<gst::Structure>("0").unwrap();
let source_stats = session_stats
.get::<gst::Structure>(TEST_SSRC.to_string())
.unwrap();
assert_eq!(source_stats.get::<u32>("ssrc").unwrap(), TEST_SSRC);
assert_eq!(
source_stats.get::<u32>("clock-rate").unwrap(),
TEST_CLOCK_RATE
);
assert!(source_stats.get::<bool>("sender").unwrap());
assert!(source_stats.get::<bool>("local").unwrap());
assert_eq!(source_stats.get::<u64>("packets-sent").unwrap(), 2);
assert_eq!(source_stats.get::<u64>("octets-sent").unwrap(), 20);
}
#[test]
fn test_receive() {
init();
let id = next_element_counter();
let elem = gst::ElementFactory::make("rtprecv")
.property("rtp-id", id.to_string())
.build()
.unwrap();
let h = Arc::new(Mutex::new(Harness::with_element(
&elem,
Some("rtp_sink_0"),
None,
)));
let weak_h = Arc::downgrade(&h);
let mut inner = h.lock().unwrap();
inner
.element()
.unwrap()
.connect_pad_added(move |_elem, pad| {
weak_h
.upgrade()
.unwrap()
.lock()
.unwrap()
.add_element_src_pad(pad)
});
inner.play();
let caps = Caps::builder("application/x-rtp")
.field("media", "audio")
.field("payload", TEST_PT as i32)
.field("clock-rate", TEST_CLOCK_RATE as i32)
.field("encoding-name", "custom-test")
.build();
inner.set_src_caps(caps);
// Cannot push with harness lock as the 'pad-added' handler needs to add the newly created pad to
// the harness and needs to also take the harness lock. Workaround by pushing from the
// internal harness pad directly.
let push_pad = inner
.element()
.unwrap()
.static_pad("rtp_sink_0")
.unwrap()
.peer()
.unwrap();
drop(inner);
push_pad.push(generate_rtp_buffer(500, 20, 9)).unwrap();
push_pad.push(generate_rtp_buffer(501, 30, 11)).unwrap();
let mut inner = h.lock().unwrap();
let buffer = inner.pull().unwrap();
let mapped = buffer.map_readable().unwrap();
let rtp = rtp_types::RtpPacket::parse(&mapped).unwrap();
assert_eq!(rtp.sequence_number(), 500);
let buffer = inner.pull().unwrap();
let mapped = buffer.map_readable().unwrap();
let rtp = rtp_types::RtpPacket::parse(&mapped).unwrap();
assert_eq!(rtp.sequence_number(), 501);
let stats = inner.element().unwrap().property::<gst::Structure>("stats");
let session_stats = stats.get::<gst::Structure>("0").unwrap();
let source_stats = session_stats
.get::<gst::Structure>(TEST_SSRC.to_string())
.unwrap();
let jitterbuffers_stats = session_stats
.get::<gst::List>("jitterbuffer-stats")
.unwrap();
assert_eq!(jitterbuffers_stats.len(), 1);
let jitterbuffer_stats = jitterbuffers_stats
.first()
.unwrap()
.get::<gst::Structure>()
.unwrap();
assert_eq!(source_stats.get::<u32>("ssrc").unwrap(), TEST_SSRC);
assert_eq!(
source_stats.get::<u32>("clock-rate").unwrap(),
TEST_CLOCK_RATE
);
assert!(source_stats.get::<bool>("sender").unwrap());
assert!(!source_stats.get::<bool>("local").unwrap());
assert_eq!(source_stats.get::<u64>("packets-received").unwrap(), 2);
assert_eq!(source_stats.get::<u64>("octets-received").unwrap(), 20);
assert_eq!(jitterbuffer_stats.get::<u64>("num-late").unwrap(), 0);
assert_eq!(jitterbuffer_stats.get::<u64>("num-lost").unwrap(), 0);
assert_eq!(jitterbuffer_stats.get::<u64>("num-duplicates").unwrap(), 0);
assert_eq!(jitterbuffer_stats.get::<u64>("num-pushed").unwrap(), 2);
assert_eq!(jitterbuffer_stats.get::<i32>("pt").unwrap(), TEST_PT as i32);
assert_eq!(
jitterbuffer_stats.get::<i32>("ssrc").unwrap(),
TEST_SSRC as i32
);
}
#[test]
fn test_receive_flush() {
init();
let id = next_element_counter();
let elem = gst::ElementFactory::make("rtprecv")
.property("rtp-id", id.to_string())
.build()
.unwrap();
let h = Arc::new(Mutex::new(Harness::with_element(
&elem,
Some("rtp_sink_0"),
None,
)));
let weak_h = Arc::downgrade(&h);
let mut inner = h.lock().unwrap();
inner
.element()
.unwrap()
.connect_pad_added(move |_elem, pad| {
weak_h
.upgrade()
.unwrap()
.lock()
.unwrap()
.add_element_src_pad(pad)
});
inner.play();
let caps = Caps::builder("application/x-rtp")
.field("media", "audio")
.field("payload", TEST_PT as i32)
.field("clock-rate", TEST_CLOCK_RATE as i32)
.field("encoding-name", "custom-test")
.build();
inner.set_src_caps(caps);
// Cannot push with harness lock as the 'pad-added' handler needs to add the newly created pad to
// the harness and needs to also take the harness lock. Workaround by pushing from the
// internal harness pad directly.
let push_pad = inner
.element()
.unwrap()
.static_pad("rtp_sink_0")
.unwrap()
.peer()
.unwrap();
drop(inner);
push_pad.push(generate_rtp_buffer(500, 20, 9)).unwrap();
push_pad.push(generate_rtp_buffer(501, 30, 11)).unwrap();
let mut inner = h.lock().unwrap();
let seqnum = gst::Seqnum::next();
inner.push_event(gst::event::FlushStart::builder().seqnum(seqnum).build());
inner.push_event(gst::event::FlushStop::builder(false).seqnum(seqnum).build());
let event = inner.pull_event().unwrap();
let gst::EventView::FlushStart(fs) = event.view() else {
unreachable!();
};
assert_eq!(fs.seqnum(), seqnum);
let event = inner.pull_event().unwrap();
let gst::EventView::FlushStop(fs) = event.view() else {
unreachable!();
};
assert_eq!(fs.seqnum(), seqnum);
}

View file

@ -816,12 +816,11 @@ impl RtspSrc {
)
.await?
};
let manager = RtspManager::new(std::env::var("USE_RTPBIN2").is_ok_and(|s| s == "1"));
let manager = RtspManager::new(std::env::var("USE_RTP2").is_ok_and(|s| s == "1"));
let obj = self.obj();
obj.add(&manager.inner)
manager.add_to(obj.upcast_ref::<gst::Bin>())
.expect("Adding the manager cannot fail");
manager.inner.sync_state_with_parent().unwrap();
let mut tcp_interleave_appsrcs = HashMap::new();
for (rtpsession_n, p) in state.setup_params.iter_mut().enumerate() {
@ -983,7 +982,7 @@ impl RtspSrc {
obj.no_more_pads();
// Expose RTP srcpads
manager.inner.connect_pad_added(|manager, pad| {
manager.recv.connect_pad_added(|manager, pad| {
if pad.direction() != gst::PadDirection::Src {
return;
}
@ -995,9 +994,9 @@ impl RtspSrc {
};
let name = pad.name();
match *name.split('_').collect::<Vec<_>>() {
// rtpbin and rtpbin2
// rtpbin and rtp2
["recv", "rtp", "src", stream_id, ssrc, pt]
| ["rtp", "recv", "src", stream_id, ssrc, pt] => {
| ["rtp", "src", stream_id, ssrc, pt] => {
if stream_id.parse::<u32>().is_err() {
gst::info!(CAT, "Ignoring srcpad with invalid stream id: {name}");
return;
@ -1128,16 +1127,27 @@ impl RtspSrc {
}
struct RtspManager {
inner: gst::Element,
using_rtpbin2: bool,
recv: gst::Element,
send: gst::Element,
using_rtp2: bool,
}
impl RtspManager {
fn new(rtpbin2: bool) -> Self {
let name = if rtpbin2 { "rtpbin2" } else { "rtpbin" };
let manager = gst::ElementFactory::make_with_name(name, None)
.unwrap_or_else(|_| panic!("{name} not found"));
if !rtpbin2 {
fn new(rtp2: bool) -> Self {
let (recv, send) = if rtp2 {
let recv = gst::ElementFactory::make_with_name("rtprecv")
.unwrap_or_else(|_| panic!("rtprecv not found"));
let send = gst::ElementFactory::make("rtpsend")
.property("rtp-id", recv.property::<String>("rtp-id"))
.build()
.unwrap_or_else(|_| panic!("rtpsend not found"));
(recv, send)
} else {
let e = gst::ElementFactory::make_with_name("rtpbin", None)
.unwrap_or_else(|_| panic!("rtpbin not found"));
(e.clone(), e)
};
if !rtp2 {
let on_bye = |args: &[glib::Value]| {
let m = args[0].get::<gst::Element>().unwrap();
let Some(obj) = m.parent() else {
@ -1147,46 +1157,62 @@ impl RtspManager {
bin.send_event(gst::event::Eos::new());
None
};
manager.connect("on-bye-ssrc", true, move |args| {
recv.connect("on-bye-ssrc", true, move |args| {
gst::info!(CAT, "Received BYE packet");
on_bye(args)
});
manager.connect("on-bye-timeout", true, move |args| {
recv.connect("on-bye-timeout", true, move |args| {
gst::info!(CAT, "BYE due to timeout");
on_bye(args)
});
}
RtspManager {
inner: manager,
using_rtpbin2: rtpbin2,
recv,
send,
using_rtp2: rtp2,
}
}
fn rtp_recv_sinkpad(&self, rtpsession: usize) -> Option<gst::Pad> {
let name = if self.using_rtpbin2 {
format!("rtp_recv_sink_{}", rtpsession)
let name = if self.using_rtp2 {
format!("rtp_sink_{}", rtpsession)
} else {
format!("recv_rtp_sink_{}", rtpsession)
};
self.inner.request_pad_simple(&name)
gst::info!(CAT, "requesting {name} for receiving RTP");
self.recv.request_pad_simple(&name)
}
fn rtcp_recv_sinkpad(&self, rtpsession: usize) -> Option<gst::Pad> {
let name = if self.using_rtpbin2 {
format!("rtcp_recv_sink_{}", rtpsession)
let name = if self.using_rtp2 {
format!("rtcp_sink_{}", rtpsession)
} else {
format!("recv_rtcp_sink_{}", rtpsession)
};
self.inner.request_pad_simple(&name)
gst::info!(CAT, "requesting {name} for receiving RTCP");
self.recv.request_pad_simple(&name)
}
fn rtcp_send_srcpad(&self, rtpsession: usize) -> Option<gst::Pad> {
let name = if self.using_rtpbin2 {
format!("rtcp_send_src_{}", rtpsession)
let name = if self.using_rtp2 {
format!("rtcp_src_{}", rtpsession)
} else {
format!("send_rtcp_src_{}", rtpsession)
};
self.inner.request_pad_simple(&name)
gst::info!(CAT, "requesting {name} for sending RTCP");
self.send.request_pad_simple(&name)
}
fn add_to<T: IsA<gst::Bin>>(&self, bin: &T) -> Result<(), glib::BoolError> {
if self.using_rtp2 {
bin.add_many(&[&self.recv, &self.send])?;
self.recv.sync_state_with_parent()?;
self.send.sync_state_with_parent()?;
} else {
bin.add_many(&[&self.recv])?;
self.recv.sync_state_with_parent()?;
}
Ok(())
}
}