mirror of
https://gitlab.freedesktop.org/gstreamer/gst-plugins-rs.git
synced 2024-09-27 06:20:03 +00:00
352 lines
11 KiB
Rust
352 lines
11 KiB
Rust
// GStreamer RTP MPEG-TS Depayloader
|
|
//
|
|
// Copyright (C) 2023-2024 Tim-Philipp Müller <tim 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
|
|
|
|
/**
|
|
* SECTION:element-rtpmp2tdepay2
|
|
* @see_also: rtpmp2tpay2, rtpmp2tdepay, rtpmp2tpay, tsdemux, mpegtsmux
|
|
*
|
|
* Depayload an MPEG Transport Stream from RTP packets as per [RFC 2250][rfc-2250].
|
|
*
|
|
* [rfc-2250]: https://www.rfc-editor.org/rfc/rfc2250.html
|
|
*
|
|
* ## Example pipeline
|
|
*
|
|
* |[
|
|
* gst-launch-1.0 udpsrc address=127.0.0.1 port=5555 caps='application/x-rtp,media=video,clock-rate=90000,encoding-name=MP2T' ! rtpjitterbuffer latency=100 ! rtpmp2tdepay2 ! decodebin3 ! videoconvertscale ! autovideosink
|
|
* ]| This will depayload an incoming RTP MPEG-TS stream. You can use the #rtpmp2tpay2 or #rtpmp2tpay
|
|
* element to create such an RTP stream.
|
|
*
|
|
* Since: plugins-rs-0.13.0
|
|
*/
|
|
use atomic_refcell::AtomicRefCell;
|
|
|
|
use gst::{glib, prelude::*, subclass::prelude::*};
|
|
|
|
use once_cell::sync::Lazy;
|
|
|
|
use std::num::NonZeroUsize;
|
|
use std::sync::Mutex;
|
|
|
|
use crate::basedepay::RtpBaseDepay2Ext;
|
|
|
|
const TS_PACKET_SYNC: u8 = 0x47;
|
|
|
|
#[derive(Default)]
|
|
pub struct RtpMP2TDepay {
|
|
state: AtomicRefCell<State>,
|
|
settings: Mutex<Settings>,
|
|
}
|
|
|
|
#[derive(Default)]
|
|
struct State {
|
|
packet_size: Option<NonZeroUsize>,
|
|
bytes_to_skip: usize,
|
|
}
|
|
|
|
#[derive(Debug, Clone)]
|
|
struct Settings {
|
|
skip_first_bytes: u32,
|
|
}
|
|
|
|
const DEFAULT_SKIP_FIRST_BYTES: u32 = 0;
|
|
|
|
impl Default for Settings {
|
|
fn default() -> Self {
|
|
Settings {
|
|
skip_first_bytes: DEFAULT_SKIP_FIRST_BYTES,
|
|
}
|
|
}
|
|
}
|
|
|
|
static CAT: Lazy<gst::DebugCategory> = Lazy::new(|| {
|
|
gst::DebugCategory::new(
|
|
"rtpmp2tdepay2",
|
|
gst::DebugColorFlags::empty(),
|
|
Some("RTP MPEG-TS Depayloader"),
|
|
)
|
|
});
|
|
|
|
#[glib::object_subclass]
|
|
impl ObjectSubclass for RtpMP2TDepay {
|
|
const NAME: &'static str = "GstRtpMP2TDepay2";
|
|
type Type = super::RtpMP2TDepay;
|
|
type ParentType = crate::basedepay::RtpBaseDepay2;
|
|
}
|
|
|
|
impl ObjectImpl for RtpMP2TDepay {
|
|
fn properties() -> &'static [glib::ParamSpec] {
|
|
static PROPERTIES: Lazy<Vec<glib::ParamSpec>> = Lazy::new(|| {
|
|
vec![glib::ParamSpecUInt::builder("skip-first-bytes")
|
|
.nick("Skip first bytes")
|
|
.blurb("Number of bytes to skip at the beginning of the payload")
|
|
.default_value(DEFAULT_SKIP_FIRST_BYTES)
|
|
.mutable_ready()
|
|
.build()]
|
|
});
|
|
|
|
PROPERTIES.as_ref()
|
|
}
|
|
fn set_property(&self, _id: usize, value: &glib::Value, pspec: &glib::ParamSpec) {
|
|
match pspec.name() {
|
|
"skip-first-bytes" => {
|
|
let mut settings = self.settings.lock().unwrap();
|
|
settings.skip_first_bytes = value.get().expect("type checked upstream");
|
|
}
|
|
name => unimplemented!("Property '{name}'"),
|
|
};
|
|
}
|
|
|
|
fn property(&self, _id: usize, pspec: &glib::ParamSpec) -> glib::Value {
|
|
match pspec.name() {
|
|
"skip-first-bytes" => {
|
|
let settings = self.settings.lock().unwrap();
|
|
settings.skip_first_bytes.to_value()
|
|
}
|
|
name => unimplemented!("Property '{name}'"),
|
|
}
|
|
}
|
|
}
|
|
|
|
impl GstObjectImpl for RtpMP2TDepay {}
|
|
|
|
impl ElementImpl for RtpMP2TDepay {
|
|
fn metadata() -> Option<&'static gst::subclass::ElementMetadata> {
|
|
static ELEMENT_METADATA: Lazy<gst::subclass::ElementMetadata> = Lazy::new(|| {
|
|
gst::subclass::ElementMetadata::new(
|
|
"RTP MPEG-TS Depayloader",
|
|
"Codec/Depayloader/Network/RTP",
|
|
"Depayload an MPEG Transport Stream from RTP packets (RFC 2250)",
|
|
"Tim-Philipp Müller <tim centricular com>",
|
|
)
|
|
});
|
|
|
|
Some(&*ELEMENT_METADATA)
|
|
}
|
|
|
|
fn pad_templates() -> &'static [gst::PadTemplate] {
|
|
static PAD_TEMPLATES: Lazy<Vec<gst::PadTemplate>> = Lazy::new(|| {
|
|
let sink_pad_template = gst::PadTemplate::new(
|
|
"sink",
|
|
gst::PadDirection::Sink,
|
|
gst::PadPresence::Always,
|
|
&gst::Caps::builder_full()
|
|
.structure(
|
|
// Note: C depayloader accepts MP2T-ES as well but that was just for
|
|
// backward compatibility because the GStreamer 0.10 payloader used
|
|
// to (wrongly) produce that at some point a long time ago.
|
|
// Also spec (and common sense) say clock-rate should always be 90000
|
|
// (C depayloader accepts any clock rate in caps), see
|
|
// https://gitlab.freedesktop.org/gstreamer/gst-plugins-good/-/issues/691
|
|
gst::Structure::builder("application/x-rtp")
|
|
.field("media", "video")
|
|
.field("clock-rate", 90000i32)
|
|
.field("encoding-name", "MP2T")
|
|
.build(),
|
|
)
|
|
.structure(
|
|
gst::Structure::builder("application/x-rtp")
|
|
.field("media", "video")
|
|
.field("payload", 33i32)
|
|
.field("clock-rate", 90000i32)
|
|
.build(),
|
|
)
|
|
.build(),
|
|
)
|
|
.unwrap();
|
|
|
|
let src_pad_template = gst::PadTemplate::new(
|
|
"src",
|
|
gst::PadDirection::Src,
|
|
gst::PadPresence::Always,
|
|
&gst::Caps::builder("video/mpegts")
|
|
.field("packetsize", gst::List::new([188i32, 192, 204, 208]))
|
|
.field("systemstream", true)
|
|
.build(),
|
|
)
|
|
.unwrap();
|
|
|
|
vec![src_pad_template, sink_pad_template]
|
|
});
|
|
|
|
PAD_TEMPLATES.as_ref()
|
|
}
|
|
}
|
|
|
|
impl crate::basedepay::RtpBaseDepay2Impl for RtpMP2TDepay {
|
|
fn start(&self) -> Result<(), gst::ErrorMessage> {
|
|
let settings = self.settings.lock().unwrap();
|
|
// Copy skip bytes into state so we don't have to take the settings lock all the time
|
|
*self.state.borrow_mut() = State {
|
|
packet_size: None,
|
|
bytes_to_skip: settings.skip_first_bytes as usize,
|
|
};
|
|
|
|
Ok(())
|
|
}
|
|
|
|
fn stop(&self) -> Result<(), gst::ErrorMessage> {
|
|
*self.state.borrow_mut() = State::default();
|
|
|
|
Ok(())
|
|
}
|
|
|
|
// Encapsulation of MPEG System and Transport Streams:
|
|
// https://www.rfc-editor.org/rfc/rfc2250.html#section-2
|
|
//
|
|
fn handle_packet(
|
|
&self,
|
|
packet: &crate::basedepay::Packet,
|
|
) -> Result<gst::FlowSuccess, gst::FlowError> {
|
|
let mut state = self.state.borrow_mut();
|
|
let bytes_to_skip = state.bytes_to_skip;
|
|
|
|
let payload = packet.payload();
|
|
|
|
if payload.len() < 188 + bytes_to_skip {
|
|
gst::warning!(
|
|
CAT,
|
|
imp = self,
|
|
"Payload too small: {} bytes, but need at least {} bytes",
|
|
payload.len(),
|
|
188 + bytes_to_skip
|
|
);
|
|
|
|
self.obj().drop_packet(packet);
|
|
|
|
return Ok(gst::FlowSuccess::Ok);
|
|
}
|
|
|
|
let (_, payload) = payload.split_at(bytes_to_skip);
|
|
|
|
if state.packet_size.is_none() {
|
|
state.packet_size = self.detect_packet_size(payload);
|
|
|
|
if let Some(packet_size) = state.packet_size {
|
|
let src_caps = gst::Caps::builder("video/mpegts")
|
|
.field("packetsize", packet_size.get() as i32)
|
|
.field("systemstream", true)
|
|
.build();
|
|
|
|
self.obj().set_src_caps(&src_caps);
|
|
}
|
|
}
|
|
|
|
let Some(packet_size) = state.packet_size else {
|
|
gst::debug!(
|
|
CAT,
|
|
imp = self,
|
|
"Could not determine packet size, dropping packet {packet:?}"
|
|
);
|
|
self.obj().drop_packet(packet);
|
|
return Ok(gst::FlowSuccess::Ok);
|
|
};
|
|
|
|
let packet_size = packet_size.get();
|
|
|
|
// For MPEG2 Transport Streams the RTP payload will contain an integral
|
|
// number of MPEG transport packets.
|
|
let n_packets = payload.len() / packet_size;
|
|
|
|
if payload.len() % packet_size != 0 {
|
|
gst::warning!(
|
|
CAT,
|
|
imp = self,
|
|
"Payload does not contain an integral number of MPEG-TS packets! ({} left over)",
|
|
payload.len() % packet_size
|
|
);
|
|
}
|
|
|
|
let output_size = n_packets * packet_size;
|
|
|
|
gst::trace!(
|
|
CAT,
|
|
imp = self,
|
|
"Packet with {n_packets} MPEG-TS packets of size {packet_size}"
|
|
);
|
|
|
|
let mut buffer =
|
|
packet.payload_subbuffer_from_offset_with_length(bytes_to_skip, output_size);
|
|
|
|
// Marker flag indicates MPEG-TS timestamping discontinuity
|
|
if packet.marker_bit() {
|
|
let buffer_ref = buffer.get_mut().unwrap();
|
|
buffer_ref.set_flags(gst::BufferFlags::RESYNC);
|
|
}
|
|
|
|
gst::trace!(CAT, imp = self, "Finishing buffer {buffer:?}");
|
|
|
|
self.obj().queue_buffer(packet.into(), buffer)
|
|
}
|
|
}
|
|
|
|
impl RtpMP2TDepay {
|
|
fn detect_packet_size(&self, payload: &[u8]) -> Option<NonZeroUsize> {
|
|
const PACKET_SIZES: [(usize, usize); 4] = [(188, 0), (192, 4), (204, 0), (208, 0)];
|
|
|
|
for (size, offset) in PACKET_SIZES {
|
|
gst::debug!(
|
|
CAT,
|
|
imp = self,
|
|
"Trying MPEG-TS packet size of {size} bytes.."
|
|
);
|
|
|
|
// Try exact size match for the payload first
|
|
if payload.len() >= size
|
|
&& payload.len() % size == 0
|
|
&& payload
|
|
.chunks_exact(size)
|
|
.all(|packet| packet[offset] == TS_PACKET_SYNC)
|
|
{
|
|
gst::info!(
|
|
CAT,
|
|
imp = self,
|
|
"Detected MPEG-TS packet size of {size} bytes, {} packets",
|
|
payload.len() / size
|
|
);
|
|
return NonZeroUsize::new(size);
|
|
}
|
|
}
|
|
|
|
gst::warning!(
|
|
CAT,
|
|
imp = self,
|
|
"Could not detect MPEG-TS packet size using full payload"
|
|
);
|
|
|
|
// No match? Try if we find a size if we ignore any leftover bytes
|
|
for (size, offset) in PACKET_SIZES {
|
|
gst::debug!(
|
|
CAT,
|
|
imp = self,
|
|
"Trying MPEG-TS packet size of {size} bytes with remainder.."
|
|
);
|
|
|
|
if payload.len() >= size
|
|
&& payload.len() % size != 0
|
|
&& payload
|
|
.chunks_exact(size)
|
|
.all(|packet| packet[offset] == TS_PACKET_SYNC)
|
|
{
|
|
gst::info!(
|
|
CAT,
|
|
imp = self,
|
|
"Detected MPEG-TS packet size of {size} bytes, {} packets, {} bytes leftover",
|
|
payload.len() / size,
|
|
payload.len() % size
|
|
);
|
|
return NonZeroUsize::new(size);
|
|
}
|
|
}
|
|
|
|
gst::warning!(CAT, imp = self, "Could not detect MPEG-TS packet size");
|
|
|
|
None
|
|
}
|
|
}
|