diff --git a/Cargo.lock b/Cargo.lock index 64dc5f8f5e..c7edf4d01c 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -3897,6 +3897,7 @@ dependencies = [ "scuffle-h265", "serde", "serde_json", + "serde_with", "thiserror 2.0.18", "tokio", "tracing", diff --git a/rs/moq-mux/Cargo.toml b/rs/moq-mux/Cargo.toml index f2d0a87e24..09676b39d1 100644 --- a/rs/moq-mux/Cargo.toml +++ b/rs/moq-mux/Cargo.toml @@ -36,6 +36,7 @@ reqwest = { version = "0.12", default-features = false, features = ["rustls-tls" scuffle-av1 = { version = "0.1.4" } scuffle-h265 = { version = "0.2.2" } serde = { workspace = true } +serde_with = { version = "3", features = ["base64"] } thiserror = "2" tokio = { workspace = true, features = ["macros", "fs"] } tracing = "0.1" diff --git a/rs/moq-mux/src/codec/legacy.rs b/rs/moq-mux/src/codec/legacy.rs index 06957e5a7f..b346ca032a 100644 --- a/rs/moq-mux/src/codec/legacy.rs +++ b/rs/moq-mux/src/codec/legacy.rs @@ -79,6 +79,11 @@ impl Import { }) } + /// The MoQ track name. + pub fn name(&self) -> &str { + &self.track.name + } + /// Finish the track, flushing the current group. pub fn finish(&mut self) -> anyhow::Result<()> { self.track.finish()?; diff --git a/rs/moq-mux/src/container/source.rs b/rs/moq-mux/src/container/source.rs index a00d077f2e..66b5025b71 100644 --- a/rs/moq-mux/src/container/source.rs +++ b/rs/moq-mux/src/container/source.rs @@ -24,7 +24,6 @@ use crate::catalog::hang::Container as HangContainer; use crate::catalog::hang::{Catalog, CatalogExt}; use crate::codec::h264::Avc1; use crate::codec::h265::Hvc1; -use crate::container::ts::scte35; use crate::container::{Consumer, Frame}; /// Source for the catalog stream backing an exporter. @@ -146,18 +145,16 @@ impl ExportSource { }) } - /// Subscribe to a SCTE-35 cue rendition. No codec-shape transform and no - /// description: the frames carry the verbatim `splice_info_section` bytes that - /// the muxer writes back out as private sections. - pub fn for_scte35( + /// Subscribe to a verbatim `mpegts` stream rendition (SCTE-35, private PES, ...). + /// No codec-shape transform and no description: the frames are Legacy-framed + /// verbatim bytes the muxer writes back out as PES or private sections. + pub fn for_stream( broadcast: &moq_net::BroadcastConsumer, name: &str, - config: &scte35::Config, latency: Duration, ) -> Result { - let media: HangContainer = (&config.container).try_into()?; let track = broadcast.subscribe_track(&moq_net::Track::new(name.to_string()))?; - let consumer = Consumer::new(track, media).with_latency(latency); + let consumer = Consumer::new(track, HangContainer::Legacy).with_latency(latency); Ok(Self { consumer, diff --git a/rs/moq-mux/src/container/ts/catalog.rs b/rs/moq-mux/src/container/ts/catalog.rs new file mode 100644 index 0000000000..6e1112b41c --- /dev/null +++ b/rs/moq-mux/src/container/ts/catalog.rs @@ -0,0 +1,221 @@ +//! MPEG-TS catalog extension (the `mpegts` section). +//! +//! The `mpegts` section carries everything needed to faithfully re-mux a broadcast +//! back to MPEG-TS that doesn't belong in the codec-neutral media configs: one +//! entry per track (its original PID and PMT descriptors), a `verbatim` carriage +//! record for every elementary stream we don't decode (SCTE-35, teletext, DVB +//! subtitles, private data, ...), and the program-level PMT descriptors. Demuxed +//! media tracks keep their codec config in the base `video`/`audio` sections; only +//! their MPEG-TS identity lands here. + +use std::collections::BTreeMap; + +use bytes::Bytes; +use serde::{Deserialize, Serialize}; +use serde_with::base64::Base64; +use serde_with::serde_as; + +use crate::catalog::hang::CatalogExt; + +/// The `mpegts` catalog section. +/// +/// Omitted from the catalog when empty, so a broadcast that needs none of it stays +/// byte-identical to one without the extension. +#[derive(Serialize, Deserialize, Debug, Clone, PartialEq, Default)] +#[serde(rename_all = "camelCase")] +#[non_exhaustive] +pub struct Mpegts { + /// Per-track MPEG-TS info, keyed by MoQ track name. Media tracks record their + /// PID and PMT descriptors; undecoded tracks add a [`Verbatim`] carriage record. + #[serde(default, skip_serializing_if = "BTreeMap::is_empty")] + pub tracks: BTreeMap, + + /// PMT program-level descriptors (`program_info`), carried verbatim. Export + /// re-emits these; the SCTE-35 'CUEI' registration is derived when absent. + #[serde(default, skip_serializing_if = "Vec::is_empty")] + pub program_descriptors: Vec, +} + +impl Mpegts { + /// True when the section carries nothing, so it's omitted from the catalog. + pub fn is_empty(&self) -> bool { + self.tracks.is_empty() && self.program_descriptors.is_empty() + } +} + +/// One track's MPEG-TS identity and signaling. +#[derive(Serialize, Deserialize, Debug, Clone, PartialEq, Default)] +#[serde(rename_all = "camelCase")] +#[non_exhaustive] +pub struct Track { + /// Original MPEG-TS PID. Export prefers it so PID cross-references survive; + /// tracks without an entry are renumbered. + pub pid: u16, + + /// PMT ES-level descriptors (ISO-639 language, registration, ...), carried + /// verbatim so they survive the round-trip. + #[serde(default, skip_serializing_if = "Vec::is_empty")] + pub descriptors: Vec, + + /// Present when the stream is carried verbatim (not decoded into `video`/`audio`). + #[serde(default, skip_serializing_if = "Option::is_none")] + pub verbatim: Option, +} + +impl Track { + /// A new media track entry (decoded; no verbatim carriage), recording its PID. + pub fn new(pid: u16) -> Self { + Self { + pid, + descriptors: Vec::new(), + verbatim: None, + } + } +} + +/// Carriage record for an undecoded elementary stream carried byte-for-byte. +#[derive(Serialize, Deserialize, Debug, Clone, PartialEq, Default)] +#[serde(rename_all = "camelCase")] +#[non_exhaustive] +pub struct Verbatim { + /// PMT `stream_type` to re-announce (0x86 SCTE-35, 0x06 private PES, 0x05 + /// private sections, ...). + pub stream_type: u8, + + /// How the verbatim payload is framed, so export knows how to repacketize it. + #[serde(default)] + pub framing: Framing, + + /// Original PES `stream_id` (e.g. 0xBD private_stream_1 for teletext/DVB + /// subtitles/DVB AC-3, 0xC0-0xDF audio). Preserved so export re-emits the PES + /// under its real id rather than relabeling it, which strict broadcast demuxers + /// and TR 101 290 analyzers reject. `None` for section framing or a non-TS + /// source; export then falls back to `private_stream_1`. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub stream_id: Option, +} + +impl Verbatim { + /// A new verbatim carriage record of the given `stream_type` and `framing`. + pub fn new(stream_type: u8, framing: Framing) -> Self { + Self { + stream_type, + framing, + stream_id: None, + } + } +} + +/// How a verbatim stream's payload is framed on the wire, so export can repacketize it. +#[derive(Serialize, Deserialize, Debug, Clone, Copy, PartialEq, Eq, Default)] +#[serde(rename_all = "camelCase")] +#[non_exhaustive] +pub enum Framing { + /// Packetized Elementary Stream: each frame is one PES payload (access unit), + /// timestamped by its PTS. Used by private PES, teletext, DVB subtitles, ... + #[default] + Pes, + /// Private sections (table_id + section_length framing). Each frame is one + /// complete section. Used by SCTE-35 and other private-section signaling. + Section, +} + +/// One PMT descriptor, carried verbatim so language/registration/etc. survive the +/// round-trip without a per-descriptor parser. +#[serde_as] +#[derive(Serialize, Deserialize, Debug, Clone, PartialEq, Eq)] +#[serde(rename_all = "camelCase")] +pub struct Descriptor { + /// The descriptor tag (e.g. 0x05 registration, 0x0A ISO-639 language). + pub tag: u8, + /// The descriptor body, base64-encoded in the catalog. + #[serde_as(as = "Base64")] + pub data: Bytes, +} + +/// The application catalog extension carrying the `mpegts` section. Empty by +/// default, so the section is omitted until an MPEG-TS detail is recorded. +#[derive(Serialize, Deserialize, Debug, Clone, PartialEq, Default)] +#[non_exhaustive] +pub struct Ext { + #[serde(default, skip_serializing_if = "Mpegts::is_empty")] + pub mpegts: Mpegts, +} + +impl CatalogExt for Ext {} + +/// An extension that can carry an `mpegts` catalog section. +/// +/// Implement this for an application extension to compose MPEG-TS carriage with +/// additional sections. +pub trait Catalog: CatalogExt { + /// The section to record MPEG-TS details into, or `None` for an extension that + /// doesn't carry them. + /// + /// Keep this stable per catalog: an importer samples support once at + /// construction, so a result that flips between `Some` and `None` mid-stream + /// would disable verbatim carriage or fail. + fn mpegts_mut(&mut self) -> Option<&mut Mpegts>; +} + +impl Catalog for () { + fn mpegts_mut(&mut self) -> Option<&mut Mpegts> { + None + } +} + +impl Catalog for Ext { + fn mpegts_mut(&mut self) -> Option<&mut Mpegts> { + Some(&mut self.mpegts) + } +} + +#[cfg(test)] +mod test { + use super::*; + + #[test] + fn empty_section_omitted() { + // An empty `mpegts` section serializes to `{}` so a media-only broadcast stays + // byte-identical to one without the extension. + let ext = Ext::default(); + assert_eq!(serde_json::to_string(&ext).unwrap(), "{}"); + } + + #[test] + fn section_roundtrip() { + let mut mpegts = Mpegts::default(); + // A media track: PID + a language descriptor, no verbatim carriage. + mpegts.tracks.insert( + "audio".to_string(), + Track { + pid: 0x101, + descriptors: vec![Descriptor { + tag: 0x0a, + data: Bytes::from_static(b"eng\x00"), + }], + verbatim: None, + }, + ); + // A verbatim SCTE-35 track. + mpegts.tracks.insert( + ".scte35".to_string(), + Track { + pid: 0x102, + descriptors: Vec::new(), + verbatim: Some(Verbatim::new(0x86, Framing::Section)), + }, + ); + mpegts.program_descriptors.push(Descriptor { + tag: 0x05, + data: Bytes::from_static(b"CUEI"), + }); + + let json = serde_json::to_string(&Ext { mpegts: mpegts.clone() }).unwrap(); + // Descriptor bytes are base64 ("CUEI" -> "Q1VFSQ=="). + assert!(json.contains("\"Q1VFSQ==\""), "descriptor data is base64: {json}"); + + let parsed: Ext = serde_json::from_str(&json).unwrap(); + assert_eq!(parsed.mpegts, mpegts, "mpegts section round-trips"); + } +} diff --git a/rs/moq-mux/src/container/ts/export.rs b/rs/moq-mux/src/container/ts/export.rs index 89afc48349..f8a891d3b0 100644 --- a/rs/moq-mux/src/container/ts/export.rs +++ b/rs/moq-mux/src/container/ts/export.rs @@ -34,7 +34,7 @@ use crate::codec::annexb; use crate::container::{CatalogSource, ExportSource, Frame}; use super::adts; -use super::scte35; +use super::catalog; /// PID of the single program's PMT. const PMT_PID: u16 = 0x1000; @@ -48,7 +48,7 @@ const PSI_INTERVAL: Duration = Duration::from_millis(500); /// Use [`next`](Self::next) to pull byte chunks: the first chunk is PAT+PMT, then /// each subsequent chunk is the TS packets for one media frame (preceded by a /// fresh PAT+PMT at video keyframes). Returns `None` when the broadcast ends. -pub struct Export { +pub struct Export { broadcast: moq_net::BroadcastConsumer, catalog: Option>, latency: Duration, @@ -56,6 +56,8 @@ pub struct Export { tracks: HashMap, /// Continuity counter per PID (PAT, PMT, and each elementary stream). counters: HashMap, + /// PMT program-level descriptors captured on import, re-emitted in the PMT. + program_descriptors: Vec, /// Program tables, built once the track layout is known. psi: Option, @@ -69,6 +71,9 @@ struct Track { finished: bool, pid: u16, kind: Kind, + /// PMT ES-level descriptors to re-announce, captured verbatim on import (language, + /// registration, ...). Empty for non-TS sources; AC-3/E-AC-3 then synthesize one. + descriptors: Vec, } #[derive(Clone)] @@ -87,8 +92,15 @@ enum Kind { Ac3, /// E-AC-3 (ATSC stream_type 0x87), carried verbatim. Eac3, - /// SCTE-35: private sections (stream_type 0x86), carried verbatim. - Scte35, + /// An undecoded elementary stream carried verbatim (SCTE-35, private PES, + /// teletext, ...). Re-announced in the PMT with its recorded `stream_type` and + /// repacketized per its `framing`. `stream_id` is the original PES stream_id to + /// re-emit (PES framing only; `None` falls back to `private_stream_1`). + Verbatim { + stream_type: u8, + framing: catalog::Framing, + stream_id: Option, + }, } /// The program tables plus the resolved PID layout. @@ -105,6 +117,8 @@ struct PesUnit { is_video: bool, keyframe: bool, timestamp: crate::container::Timestamp, + /// Explicit PES stream_id (verbatim PES); `None` derives it from `is_video`. + stream_id: Option, } impl Export { @@ -114,7 +128,7 @@ impl Export { } /// Subscribe to `broadcast`, selecting an explicit catalog format. Media only; - /// any catalog extension (e.g. `.scte35` cues) is ignored. + /// any catalog extension (e.g. the `mpegts` verbatim streams) is ignored. pub fn with_catalog_format( broadcast: moq_net::BroadcastConsumer, catalog_format: CatalogFormat, @@ -123,19 +137,17 @@ impl Export { } } -impl Export { - /// Subscribe to `broadcast`, exporting its `.scte35` cue tracks back to MPEG-TS - /// alongside the media. The `Self` type pins the extension, so callers write - /// `Export::with_scte35(..)` with no turbofish (the plain constructors are media-only). - pub fn with_scte35( - broadcast: moq_net::BroadcastConsumer, - catalog_format: CatalogFormat, - ) -> Result { +impl Export { + /// Subscribe to `broadcast`, exporting its `mpegts` verbatim streams (SCTE-35, + /// private data, ...) back to MPEG-TS alongside the media. The `Self` type pins + /// the extension, so callers write `Export::with_ts(..)` with no turbofish (the + /// plain constructors are media-only). + pub fn with_ts(broadcast: moq_net::BroadcastConsumer, catalog_format: CatalogFormat) -> Result { Self::build(broadcast, catalog_format) } } -impl Export { +impl Export { /// Shared constructor. The public entry points each live on a concrete /// `Export` impl that pins `E`, so the extension is chosen by which one you call. fn build(broadcast: moq_net::BroadcastConsumer, catalog_format: CatalogFormat) -> Result { @@ -146,6 +158,7 @@ impl Export { latency: Duration::ZERO, tracks: HashMap::new(), counters: HashMap::new(), + program_descriptors: Vec::new(), psi: None, last_psi: None, }) @@ -247,11 +260,13 @@ impl Export { } fn update_catalog(&mut self, mut catalog: Catalog) -> anyhow::Result<()> { - // The cue tracks live in the extension. The trait only exposes `scte35_mut`, - // and this snapshot is owned, so clone the section out (`()` yields the - // empty default: zero cue tracks). - let scte35 = catalog.scte35_mut().cloned().unwrap_or_default(); + // The MPEG-TS section lives in the extension. The trait only exposes + // `mpegts_mut`, and this snapshot is owned, so clone it out (`()` yields the + // empty default: no verbatim streams, no preserved PIDs/descriptors). + let mpegts = catalog.mpegts_mut().cloned().unwrap_or_default(); + self.program_descriptors = mpegts.program_descriptors.clone(); + // The desired track set: media renditions plus the verbatim streams. let mut active: HashMap = HashMap::new(); for name in catalog.video.renditions.keys() { active.insert(name.clone(), ()); @@ -259,8 +274,10 @@ impl Export { for name in catalog.audio.renditions.keys() { active.insert(name.clone(), ()); } - for name in scte35.renditions.keys() { - active.insert(name.clone(), ()); + for (name, track) in mpegts.tracks.iter() { + if track.verbatim.is_some() { + active.insert(name.clone(), ()); + } } // The program tables are written once; reject layout changes afterwards. @@ -280,75 +297,118 @@ impl Export { return Ok(()); } - let mut next_pid = self - .tracks - .values() - .map(|t| t.pid) - .max() - .map(|p| p + 1) - .unwrap_or(FIRST_ES_PID); + // Assign a PID to every desired track: prefer the original recorded in the + // `mpegts` section, then fill the rest from FIRST_ES_PID. The importer fills + // PIDs, descriptors, and stream_ids across several catalog publishes, so this + // runs every snapshot until the PMT is built and the tracks below are + // *refreshed*, not latched from the first (partial) snapshot. + let mut used: Vec = vec![0x0000, PMT_PID, 0x1FFF]; + let mut pids: HashMap = HashMap::new(); + for name in active.keys() { + if let Some(pid) = mpegts.tracks.get(name).map(|t| t.pid) + && !used.contains(&pid) + { + used.push(pid); + pids.insert(name.clone(), pid); + } + } + for name in active.keys() { + if !pids.contains_key(name) { + let mut pid = FIRST_ES_PID; + while used.contains(&pid) { + pid += 1; + } + used.push(pid); + pids.insert(name.clone(), pid); + } + } + // Reuse each track's existing source (and any pending frame) by name; refresh + // its PID, kind, and descriptors from this snapshot. Drop tracks no longer present. + let mut old = std::mem::take(&mut self.tracks); for (name, config) in catalog.video.renditions.iter() { - if self.tracks.contains_key(name) { - continue; - } let kind = video_kind(config, name)?; - let source = ExportSource::for_video(&self.broadcast, name, config, self.latency)?; - self.tracks.insert( - name.clone(), - Track { - source, - pending: None, - finished: false, - pid: next_pid, - kind, - }, - ); - next_pid += 1; + let descriptors = track_descriptors(&mpegts, name); + let pid = pids[name]; + match old.remove(name) { + Some(mut track) => { + track.pid = pid; + track.kind = kind; + track.descriptors = descriptors; + self.tracks.insert(name.clone(), track); + } + None => { + let source = ExportSource::for_video(&self.broadcast, name, config, self.latency)?; + self.insert_track(name, source, pid, kind, descriptors); + } + } } - for (name, config) in catalog.audio.renditions.iter() { - if self.tracks.contains_key(name) { - continue; - } let kind = audio_kind(config, name)?; - let source = ExportSource::for_audio(&self.broadcast, name, config, self.latency)?; - self.tracks.insert( - name.clone(), - Track { - source, - pending: None, - finished: false, - pid: next_pid, - kind, - }, - ); - next_pid += 1; + let descriptors = track_descriptors(&mpegts, name); + let pid = pids[name]; + match old.remove(name) { + Some(mut track) => { + track.pid = pid; + track.kind = kind; + track.descriptors = descriptors; + self.tracks.insert(name.clone(), track); + } + None => { + let source = ExportSource::for_audio(&self.broadcast, name, config, self.latency)?; + self.insert_track(name, source, pid, kind, descriptors); + } + } } - - for (name, config) in scte35.renditions.iter() { - if self.tracks.contains_key(name) { + for (name, track) in mpegts.tracks.iter() { + let Some(verbatim) = &track.verbatim else { continue; + }; + let kind = Kind::Verbatim { + stream_type: verbatim.stream_type, + framing: verbatim.framing, + stream_id: verbatim.stream_id, + }; + let descriptors = track.descriptors.clone(); + let pid = pids[name]; + match old.remove(name) { + Some(mut existing) => { + existing.pid = pid; + existing.kind = kind; + existing.descriptors = descriptors; + self.tracks.insert(name.clone(), existing); + } + None => { + let source = ExportSource::for_stream(&self.broadcast, name, self.latency)?; + self.insert_track(name, source, pid, kind, descriptors); + } } - let kind = scte35_kind(config, name)?; - let source = ExportSource::for_scte35(&self.broadcast, name, config, self.latency)?; - self.tracks.insert( - name.clone(), - Track { - source, - pending: None, - finished: false, - pid: next_pid, - kind, - }, - ); - next_pid += 1; } - - self.tracks.retain(|name, _| active.contains_key(name)); Ok(()) } + /// Insert a freshly created export track. + fn insert_track( + &mut self, + name: &str, + source: ExportSource, + pid: u16, + kind: Kind, + descriptors: Vec, + ) { + self.tracks.insert( + name.to_string(), + Track { + source, + pending: None, + finished: false, + pid, + kind, + descriptors, + }, + ); + } + /// Header is ready when every track's [`ExportSource`] has resolved its /// codec config (from the catalog `description`, or built by the transform). fn header_ready(&self) -> bool { @@ -361,13 +421,22 @@ impl Export { let mut tracks: Vec<&Track> = self.tracks.values().collect(); tracks.sort_by_key(|t| t.pid); - // SCTE-35 cues are stamped on the video clock (and SCTE carries no PTS for the PCR), - // so a cue program needs a video track; audio alone would leave the cues pinned to zero. - let has_scte = tracks.iter().any(|t| matches!(t.kind, Kind::Scte35)); + // Section-framed verbatim streams (SCTE-35, ...) are stamped on the video clock + // and carry no PTS for the PCR, so they need a video track; audio alone would + // leave them pinned to zero. + let needs_clock = tracks.iter().any(|t| { + matches!( + &t.kind, + Kind::Verbatim { + framing: catalog::Framing::Section, + .. + } + ) + }); let video = tracks.iter().find(|t| matches!(t.kind, Kind::Video(_))); anyhow::ensure!( - !has_scte || video.is_some(), - "TS export of SCTE-35 requires a video track for the program clock" + !needs_clock || video.is_some(), + "TS export of section-framed verbatim streams (e.g. SCTE-35) requires a video track for the program clock" ); let pcr_pid = video .or_else(|| { @@ -381,22 +450,27 @@ impl Export { let es_info = tracks .iter() .map(|t| { - Ok(EsInfo { - stream_type: match t.kind { - Kind::Video(stream_type) => stream_type, - Kind::Aac { .. } => StreamType::AdtsAac, - // Half-rate MPEG-2 BC audio (< 32 kHz) re-announces as 0x04; - // the full rates are MPEG-1 (0x03). The catalog sample rate - // came from the frame header, so the mapping is faithful. - Kind::Mp2 { sample_rate } if sample_rate < 32000 => StreamType::Mpeg2HalvedSampleRateAudio, - Kind::Mp2 { .. } => StreamType::Mpeg1Audio, - Kind::Ac3 => StreamType::DolbyDigitalUpToSixChannelAudio, - Kind::Eac3 => StreamType::DolbyDigitalPlusUpTo16ChannelAudioForAtsc, - Kind::Scte35 => StreamType::Dts8ChannelLosslessAudio, - }, - elementary_pid: Pid::new(t.pid)?, - // ATSC pairs the Dolby stream types with a registration descriptor. - descriptors: match t.kind { + let stream_type = match &t.kind { + Kind::Video(stream_type) => *stream_type, + Kind::Aac { .. } => StreamType::AdtsAac, + // Half-rate MPEG-2 BC audio (< 32 kHz) re-announces as 0x04; the full + // rates are MPEG-1 (0x03). The catalog sample rate came from the frame + // header, so the mapping is faithful. + Kind::Mp2 { sample_rate } if *sample_rate < 32000 => StreamType::Mpeg2HalvedSampleRateAudio, + Kind::Mp2 { .. } => StreamType::Mpeg1Audio, + Kind::Ac3 => StreamType::DolbyDigitalUpToSixChannelAudio, + Kind::Eac3 => StreamType::DolbyDigitalPlusUpTo16ChannelAudioForAtsc, + Kind::Verbatim { stream_type, .. } => { + StreamType::from_u8(*stream_type).map_err(anyhow::Error::msg)? + } + }; + // Prefer the descriptors captured verbatim on import; otherwise synthesize + // the ATSC Dolby registration so a fresh (non-TS) AC-3/E-AC-3 track is + // still announced the way the import path expects. + let descriptors = if !t.descriptors.is_empty() { + to_pmt_descriptors(&t.descriptors) + } else { + match &t.kind { Kind::Ac3 => vec![Descriptor { tag: 0x05, data: b"AC-3".to_vec(), @@ -406,14 +480,32 @@ impl Export { data: b"EAC3".to_vec(), }], _ => Vec::new(), - }, + } + }; + Ok(EsInfo { + stream_type, + elementary_pid: Pid::new(t.pid)?, + descriptors, }) }) .collect::>>()?; - // SCTE-35 is announced by a program-level 'CUEI' registration descriptor; - // the import keys detection off it (stream_type 0x86 alone is ambiguous). - let program_info = if tracks.iter().any(|t| matches!(t.kind, Kind::Scte35)) { + // Re-emit the captured program-level descriptors. With none (a non-TS source), + // derive the SCTE-35 'CUEI' registration when a 0x86 verbatim stream is present. + let program_info = if !self.program_descriptors.is_empty() { + to_pmt_descriptors(&self.program_descriptors) + } else if tracks.iter().any(|t| { + // Only derive CUEI for section-framed 0x86 (SCTE-35); a PES-framed 0x86 + // (e.g. DTS audio) must not advertise SCTE-35 section signaling. + matches!( + &t.kind, + Kind::Verbatim { + stream_type: 0x86, + framing: catalog::Framing::Section, + .. + } + ) + }) { vec![Descriptor { tag: 0x05, data: b"CUEI".to_vec(), @@ -472,8 +564,8 @@ impl Export { let is_video = matches!(kind, Kind::Video(_)); // Build the elementary-stream payload for this frame. Video needs the - // resolved avcC/hvcC to rewrite length-prefixed NALs as Annex-B. SCTE-35 - // carries no PES payload; the section is written separately below. + // resolved avcC/hvcC to rewrite length-prefixed NALs as Annex-B. Section-framed + // verbatim streams carry no PES payload; the section is written separately below. let es_payload = match &kind { Kind::Video(stream_type) => Some(video_es_payload(*stream_type, track.source.description(), &frame)?), Kind::Aac { @@ -488,9 +580,16 @@ impl Export { Some(framed) } // Legacy audio frames were ingested whole (framing header included), so - // they pass through untouched. + // they pass through untouched. PES-framed verbatim payloads likewise. Kind::Mp2 { .. } | Kind::Ac3 | Kind::Eac3 => Some(frame.payload.to_vec()), - Kind::Scte35 => None, + Kind::Verbatim { + framing: catalog::Framing::Pes, + .. + } => Some(frame.payload.to_vec()), + Kind::Verbatim { + framing: catalog::Framing::Section, + .. + } => None, }; let mut out = Vec::with_capacity(TsPacket::SIZE); @@ -510,15 +609,24 @@ impl Export { } match es_payload { - // SCTE-35 rides in private sections, not PES; carry the bytes verbatim. + // Section-framed verbatim (SCTE-35, ...) rides in private sections, not PES; + // carry the bytes verbatim. None => self.write_section(&mut out, pid, &frame.payload)?, Some(es_payload) => { + // Verbatim PES re-emits its original stream_id (falling back to + // private_stream_1 for an undecoded stream with none recorded); media + // derives it from is_video. + let stream_id = match &kind { + Kind::Verbatim { stream_id, .. } => Some(stream_id.unwrap_or(StreamId::PRIVATE_STREAM_1)), + _ => None, + }; let unit = PesUnit { pid, is_pcr, is_video, keyframe: frame.keyframe, timestamp: frame.timestamp, + stream_id, }; self.write_pes(&mut out, &unit, &es_payload)?; } @@ -529,10 +637,10 @@ impl Export { /// Packetize a PES payload into 188-byte TS packets. fn write_pes(&mut self, out: &mut Vec, unit: &PesUnit, payload: &[u8]) -> anyhow::Result<()> { let pts = to_ts_timestamp(unit.timestamp)?; - let stream_id = if unit.is_video { - StreamId::new(StreamId::VIDEO_MIN) - } else { - StreamId::new(StreamId::AUDIO_MIN) + let stream_id = match unit.stream_id { + Some(id) => StreamId::new(id), + None if unit.is_video => StreamId::new(StreamId::VIDEO_MIN), + None => StreamId::new(StreamId::AUDIO_MIN), }; let header = mpeg2ts::pes::PesHeader { stream_id, @@ -599,17 +707,17 @@ impl Export { Ok(()) } - /// Packetize a private section (SCTE-35) verbatim. The first packet carries the - /// pointer_field plus the section start as a `Section` payload (sets the unit- - /// start bit so the receiver finds the pointer_field); continuations are `Raw`. - /// The section bytes are opaque, so this round-trips byte-for-byte. + /// Packetize a private section (SCTE-35 or other) verbatim. The first packet + /// carries the pointer_field plus the section start as a `Section` payload (sets + /// the unit-start bit so the receiver finds the pointer_field); continuations are + /// `Raw`. The section bytes are opaque, so this round-trips byte-for-byte. fn write_section(&mut self, out: &mut Vec, pid: u16, section: &[u8]) -> anyhow::Result<()> { - // The .scte35 track is public; a non-importer producer could publish a frame - // that isn't a complete splice_info_section. Drop it (with a warning) rather - // than emit a malformed section a downstream demuxer would choke on. One bad - // cue must not abort a live export, so this skips instead of erroring. - if !is_complete_scte35_section(section) { - tracing::warn!(pid, len = section.len(), "dropping malformed SCTE-35 section on export"); + // The verbatim track is public; a non-importer producer could publish a frame + // that isn't a complete section. Drop it (with a warning) rather than emit a + // malformed section a downstream demuxer would choke on. One bad section must + // not abort a live export, so this skips instead of erroring. + if !is_complete_section(section) { + tracing::warn!(pid, len = section.len(), "dropping malformed private section on export"); return Ok(()); } @@ -745,18 +853,31 @@ fn audio_kind(config: &AudioConfig, name: &str) -> anyhow::Result { } } -fn scte35_kind(config: &scte35::Config, name: &str) -> anyhow::Result { - ensure_raw(&config.container, "scte35", name)?; - Ok(Kind::Scte35) +/// The PMT descriptors recorded for `name` in the `mpegts` section, if any. +fn track_descriptors(mpegts: &catalog::Mpegts, name: &str) -> Vec { + mpegts + .tracks + .get(name) + .map(|t| t.descriptors.clone()) + .unwrap_or_default() +} + +/// Convert catalog descriptors (base64 bytes) to mpeg2ts PMT descriptors. +fn to_pmt_descriptors(descriptors: &[catalog::Descriptor]) -> Vec { + descriptors + .iter() + .map(|d| Descriptor { + tag: d.tag, + data: d.data.to_vec(), + }) + .collect() } -/// One SCTE-35 frame must be exactly one splice_info_section: table_id 0xFC and a -/// total length matching the declared section_length. Structural only (no splice -/// semantics); the bytes are still carried verbatim. -fn is_complete_scte35_section(section: &[u8]) -> bool { - section.len() >= 3 - && section[0] == 0xfc - && section.len() == 3 + ((((section[1] & 0x0f) as usize) << 8) | section[2] as usize) +/// One section-framed verbatim frame must be exactly one section: at least the +/// 3-byte header and a total length matching the declared section_length. +/// Structural only (no table semantics); the bytes are still carried verbatim. +fn is_complete_section(section: &[u8]) -> bool { + section.len() >= 3 && section.len() == 3 + ((((section[1] & 0x0f) as usize) << 8) | section[2] as usize) } fn ensure_raw(container: &Container, kind: &str, name: &str) -> anyhow::Result<()> { @@ -769,22 +890,22 @@ fn ensure_raw(container: &Container, kind: &str, name: &str) -> anyhow::Result<( #[cfg(test)] mod tests { - use super::is_complete_scte35_section; + use super::is_complete_section; #[test] - fn scte35_section_validation() { - // table_id 0xFC, section_length 27 (0x1b) -> 30 bytes total. + fn section_validation() { + // section_length 27 (0x1b) -> 30 bytes total. let mut ok = vec![0xfc, 0x30, 0x1b]; ok.resize(30, 0x00); - assert!(is_complete_scte35_section(&ok)); + assert!(is_complete_section(&ok)); // minimal: section_length 0 -> exactly the 3-byte header. - assert!(is_complete_scte35_section(&[0xfc, 0x00, 0x00])); + assert!(is_complete_section(&[0xfc, 0x00, 0x00])); + // any table_id is accepted (verbatim carriage isn't SCTE-specific). + assert!(is_complete_section(&[0x00, 0x00, 0x00])); // shorter than the 3-byte header. - assert!(!is_complete_scte35_section(&[0xfc, 0x00])); - // wrong table_id (not a splice_info_section). - assert!(!is_complete_scte35_section(&[0x00, 0x00, 0x00])); + assert!(!is_complete_section(&[0xfc, 0x00])); // declared section_length (27) does not match the actual length (3). - assert!(!is_complete_scte35_section(&[0xfc, 0x30, 0x1b])); + assert!(!is_complete_section(&[0xfc, 0x30, 0x1b])); } } diff --git a/rs/moq-mux/src/container/ts/export_test.rs b/rs/moq-mux/src/container/ts/export_test.rs index 88f98d325b..d8f16a53bb 100644 --- a/rs/moq-mux/src/container/ts/export_test.rs +++ b/rs/moq-mux/src/container/ts/export_test.rs @@ -15,7 +15,7 @@ use mpeg2ts::pes::{PesPacketReader, ReadPesPacket}; use mpeg2ts::ts::{ReadTsPacket, TsPacketReader, TsPayload}; use crate::catalog::hang::Container as HangContainer; -use crate::container::ts::{Export, scte35}; +use crate::container::ts::{Export, catalog as tscat}; use crate::container::{Frame, Producer, Timestamp}; const SC: &[u8] = &[0, 0, 0, 1]; @@ -63,7 +63,7 @@ async fn drain(consumer: moq_net::BroadcastConsumer) -> BytesMut { } /// `drain` for an exporter built with an explicit catalog extension. -async fn drain_with(mut exporter: Export) -> BytesMut { +async fn drain_with(mut exporter: Export) -> BytesMut { let mut out = BytesMut::new(); // `while let Ok` stops on the first timeout (`Pending`: no more output). while let Ok(res) = tokio::time::timeout(std::time::Duration::from_secs(1), exporter.next()).await { @@ -345,18 +345,21 @@ async fn export_scte35_roundtrip() { let mut broadcast = moq_net::Broadcast::new().produce(); let consumer = broadcast.consume(); let mut catalog = - crate::catalog::Producer::with_catalog(&mut broadcast, crate::catalog::hang::Catalog::::default()) + crate::catalog::Producer::with_catalog(&mut broadcast, crate::catalog::hang::Catalog::::default()) .unwrap(); - // Create and write the .scte35 cue track BEFORE moving `broadcast` into + // Create and write the SCTE-35 cue track BEFORE moving `broadcast` into // `Import` (which consumes it); the producer stays alive so the exporter can // subscribe to the retained track. let scte = broadcast.unique_track(".scte35").unwrap(); let scte_name = scte.name.clone(); { - let mut cfg = scte35::Config::new(); - cfg.container = Container::Legacy; - catalog.lock().scte35.renditions.insert(scte_name.clone(), cfg); + let track = tscat::Track { + pid: 0x102, + descriptors: Vec::new(), + verbatim: Some(tscat::Verbatim::new(0x86, tscat::Framing::Section)), + }; + catalog.lock().mpegts.tracks.insert(scte_name.clone(), track); } let mut scte_producer = Producer::new(scte, HangContainer::Legacy); scte_producer @@ -375,8 +378,8 @@ async fn export_scte35_roundtrip() { import.finish().unwrap(); // `import`, `catalog`, and `scte_producer` stay alive: retained tracks. The - // exporter must carry the extension to see the scte35 section. - let ts = drain_with(Export::with_scte35(consumer, crate::catalog::CatalogFormat::Hang).unwrap()).await; + // exporter must carry the extension to see the ts section. + let ts = drain_with(Export::with_ts(consumer, crate::catalog::CatalogFormat::Hang).unwrap()).await; assert_packet_aligned(&ts); // The first PMT advertises the SCTE-35 ES (0x86) and the CUEI descriptor. @@ -404,18 +407,17 @@ async fn export_scte35_roundtrip() { // Re-import the exported TS and read the .scte35 frame back. let mut broadcast2 = moq_net::Broadcast::new().produce(); let consumer2 = broadcast2.consume(); - let catalog2 = crate::catalog::Producer::with_catalog( - &mut broadcast2, - crate::catalog::hang::Catalog::::default(), - ) - .unwrap(); + let catalog2 = + crate::catalog::Producer::with_catalog(&mut broadcast2, crate::catalog::hang::Catalog::::default()) + .unwrap(); let mut import2 = crate::container::ts::Import::new(broadcast2, catalog2.clone()); import2.decode(&mut BytesMut::from(ts.as_ref())).unwrap(); import2.finish().unwrap(); let snapshot = catalog2.snapshot(); - assert_eq!(snapshot.scte35.renditions.len(), 1, "round-trip lost the SCTE-35 track"); - let name = snapshot.scte35.renditions.keys().next().unwrap(); + let verbatim = snapshot.mpegts.tracks.values().filter(|t| t.verbatim.is_some()).count(); + assert_eq!(verbatim, 1, "round-trip lost the SCTE-35 track"); + let name = scte_track(&snapshot).expect("a scte35 track"); let track = consumer2.subscribe_track(&moq_net::Track::new(name.clone())).unwrap(); let mut scte_reader = crate::container::Consumer::new(track, HangContainer::Legacy); @@ -438,16 +440,19 @@ async fn scte35_without_video_export_is_rejected() { let mut broadcast = moq_net::Broadcast::new().produce(); let consumer = broadcast.consume(); let mut catalog = - crate::catalog::Producer::with_catalog(&mut broadcast, crate::catalog::hang::Catalog::::default()) + crate::catalog::Producer::with_catalog(&mut broadcast, crate::catalog::hang::Catalog::::default()) .unwrap(); - // A scte35 cue track and nothing else. + // A SCTE-35 cue track and nothing else. let scte = broadcast.unique_track(".scte35").unwrap(); let scte_name = scte.name.clone(); { - let mut cfg = scte35::Config::new(); - cfg.container = Container::Legacy; - catalog.lock().scte35.renditions.insert(scte_name, cfg); + let track = tscat::Track { + pid: 0x102, + descriptors: Vec::new(), + verbatim: Some(tscat::Verbatim::new(0x86, tscat::Framing::Section)), + }; + catalog.lock().mpegts.tracks.insert(scte_name, track); } let mut producer = Producer::new(scte, HangContainer::Legacy); producer @@ -460,7 +465,7 @@ async fn scte35_without_video_export_is_rejected() { producer.finish_group().unwrap(); producer.finish().unwrap(); - let mut exporter = Export::with_scte35(consumer, crate::catalog::CatalogFormat::Hang).unwrap(); + let mut exporter = Export::with_ts(consumer, crate::catalog::CatalogFormat::Hang).unwrap(); let err = loop { match tokio::time::timeout(std::time::Duration::from_secs(1), exporter.next()).await { Ok(Ok(Some(_))) => continue, @@ -775,6 +780,16 @@ async fn kyrion_ac3_mp2_roundtrip_byte_exact() { assert_eq!(roundtripped, ingested, "both audio streams survive byte-for-byte"); } +/// Find the SCTE-35 verbatim stream (stream_type 0x86) in a catalog snapshot. A +/// clip may carry other undecoded streams verbatim, so select by type, not order. +fn scte_track(snap: &crate::catalog::hang::Catalog) -> Option { + snap.mpegts + .tracks + .iter() + .find(|(_, t)| t.verbatim.as_ref().is_some_and(|v| v.stream_type == 0x86)) + .map(|(name, _)| name.clone()) +} + /// Subscribe to a cue track and read every retained `splice_info_section` it holds. async fn read_cues(consumer: &moq_net::BroadcastConsumer, name: &str) -> Vec<(Vec, Timestamp)> { let track = consumer @@ -849,7 +864,7 @@ async fn scte35_fixtures_survive_roundtrip() { let consumer = broadcast.consume(); let catalog = crate::catalog::Producer::with_catalog( &mut broadcast, - crate::catalog::hang::Catalog::::default(), + crate::catalog::hang::Catalog::::default(), ) .unwrap(); let mut import = crate::container::ts::Import::new(broadcast, catalog.clone()); @@ -858,7 +873,9 @@ async fn scte35_fixtures_survive_roundtrip() { let snap = catalog.snapshot(); assert!(!snap.video.renditions.is_empty(), "{source}: video track from the clip"); - let name = snap.scte35.renditions.keys().next().expect("a scte35 track").clone(); + // Select the SCTE-35 stream by stream_type (0x86); a clip may also carry other + // undecoded streams verbatim (e.g. Opus as private PES in bbb5s). + let name = scte_track(&snap).expect("a scte35 track"); let ingested = read_cues(&consumer, &name).await; assert_eq!(ingested.len(), *total, "{source}: {total} cues on ingest"); assert!( @@ -891,27 +908,20 @@ async fn scte35_fixtures_survive_roundtrip() { ); // Export and re-ingest. - let ts = drain_with(Export::with_scte35(consumer, crate::catalog::CatalogFormat::Hang).unwrap()).await; + let ts = drain_with(Export::with_ts(consumer, crate::catalog::CatalogFormat::Hang).unwrap()).await; assert_packet_aligned(&ts); let mut broadcast2 = moq_net::Broadcast::new().produce(); let consumer2 = broadcast2.consume(); let catalog2 = crate::catalog::Producer::with_catalog( &mut broadcast2, - crate::catalog::hang::Catalog::::default(), + crate::catalog::hang::Catalog::::default(), ) .unwrap(); let mut import2 = crate::container::ts::Import::new(broadcast2, catalog2.clone()); import2.decode(&mut BytesMut::from(ts.as_ref())).unwrap(); import2.finish().unwrap(); - let name2 = catalog2 - .snapshot() - .scte35 - .renditions - .keys() - .next() - .expect("a scte35 track") - .clone(); + let name2 = scte_track(&catalog2.snapshot()).expect("a scte35 track"); let roundtripped = read_cues(&consumer2, &name2).await; let before: Vec<&Vec> = ingested.iter().map(|(b, _)| b).collect(); diff --git a/rs/moq-mux/src/container/ts/import.rs b/rs/moq-mux/src/container/ts/import.rs index 4c744c56fc..94f3bda312 100644 --- a/rs/moq-mux/src/container/ts/import.rs +++ b/rs/moq-mux/src/container/ts/import.rs @@ -2,12 +2,13 @@ //! //! [`Import`] reads a TS byte stream, reassembles PES packets per PID, and //! routes their payloads to the codec importers (H.264/H.265/AAC, plus the -//! legacy MP2/AC-3/E-AC-3 verbatim path), -//! which own their broadcast tracks and catalog entries. SCTE-35 rides in private -//! sections (not PES), so those PIDs are intercepted before the mpeg2ts reader -//! and reassembled onto a typed scte35 catalog section. TS adds PAT/PMT -//! discovery, PES reassembly, the SCTE-35 section path, and the 90 kHz -> -//! microsecond PTS conversion. +//! legacy MP2/AC-3/E-AC-3 verbatim path), which own their broadcast tracks and +//! catalog entries. Elementary streams we don't decode are carried verbatim, one +//! MoQ track per PID, described in the `mpegts` catalog section: PES-framed streams +//! ride the normal PES reassembly, while section-framed streams (SCTE-35 and +//! other private sections, which are not PES) are intercepted before the mpeg2ts +//! reader and reassembled. TS adds PAT/PMT discovery, PES reassembly, the +//! private-section path, and the 90 kHz -> microsecond PTS conversion. use std::collections::{HashMap, HashSet}; use std::io::Read; @@ -21,7 +22,7 @@ use mpeg2ts::ts::{Pid, ReadTsPacket, TsPacket, TsPacketReader, TsPayload}; use tokio::io::{AsyncRead, AsyncReadExt}; use super::adts; -use super::scte35; +use super::catalog; use crate::catalog::hang::CatalogExt; use crate::codec::{aac, ac3, eac3, h264, h265, legacy, mp2}; use crate::container::Timestamp; @@ -30,12 +31,16 @@ use crate::container::Timestamp; /// /// Supports H.264 (stream type 0x1B), H.265 (0x24), ADTS AAC (0x0F), MP2 /// (0x03/0x04), AC-3 (0x81), and E-AC-3 (0x87). LATM/LOAS AAC (0x11) is not -/// ADTS-framed and is dropped. SCTE-35 (private sections marked -/// by a program-level 'CUEI' registration descriptor) is intercepted before the -/// reader and reassembled. Other elementary streams are logged and dropped. Each -/// codec stream is fed to its importer, which manages the track, catalog config, -/// and keyframe-based group boundaries. -pub struct Import { +/// ADTS-framed and is dropped. Each codec stream is fed to its importer, which +/// manages the track, catalog config, and keyframe-based group boundaries. +/// +/// Elementary streams we don't decode are carried verbatim, one MoQ track per +/// PID, when the catalog `E` carries the [`mpegts`](catalog) section: PES-framed +/// streams ride the normal PES reassembly, section-framed streams (SCTE-35, marked +/// by a program-level 'CUEI' registration descriptor, and other private sections) +/// are intercepted before the reader and reassembled. With a base `Catalog<()>` +/// they're logged and dropped instead. +pub struct Import { broadcast: moq_net::BroadcastProducer, catalog: crate::catalog::Producer, @@ -59,9 +64,9 @@ pub struct Import { /// the audio catalog jitter (see [`AacStream::write`]). Reset on a video frame. audio_burst: Option, - /// Whole-packet accumulator. Bytes are routed one TS packet at a time (SCTE - /// PIDs diverted, the rest fed to the reader); a trailing partial packet is - /// kept here for the next call. + /// Whole-packet accumulator. Bytes are routed one TS packet at a time + /// (section-framed verbatim PIDs diverted, the rest fed to the reader); a + /// trailing partial packet is kept here for the next call. scratch: Vec, /// Sync lock. 0x47 is the packet sync byte but also occurs freely in payload (TS /// has no byte stuffing), so a lone 0x47 isn't a boundary. False until a candidate @@ -69,14 +74,24 @@ pub struct Import { /// and trust the per-packet check. Persists across `decode` calls so a candidate /// pending confirmation at a buffer tail is re-confirmed, not trusted blindly. synced: bool, - /// SCTE-35 PIDs, intercepted before the reader. SCTE-35 is carried as private - /// sections (table_id 0xFC), not PES, so the reader would `Pes::read_from` and - /// abort. Keyed by PID. Detected via the PMT 'CUEI' registration descriptor. - scte: HashMap>, - /// Whether the catalog can carry the scte35 section, sampled once at construction. - /// A base `Catalog<()>` can't, so its SCTE PIDs route to `Stream::Ignored`. - supports_scte35: bool, - /// Latest video PTS: the media clock used to timestamp SCTE-35 sections, which + /// Section-framed verbatim PIDs, intercepted before the reader. Private sections + /// (SCTE-35 table_id 0xFC and others) are not PES, so the reader would + /// `Pes::read_from` and abort. Keyed by PID. SCTE-35 is detected via the PMT + /// 'CUEI' registration descriptor. + sections: HashMap>, + /// Whether the catalog can carry the `mpegts` section, sampled once at construction. + /// A base `Catalog<()>` can't, so its undecoded PIDs route to `Stream::Ignored`. + supports_mpegts: bool, + /// PMT ES-level descriptors per PID, stashed when a PMT is parsed so a decoded + /// media track can record them (language, registration, ...) once its track exists. + es_descriptors: HashMap>, + /// Decoded media PIDs already recorded into `mpegts.tracks`, so the reconcile in + /// [`Self::flush`] runs once per track rather than on every frame. + recorded_media: HashSet, + /// Whether the PMT program-level descriptors have been recorded yet (set once; + /// PMT `program_info` is stable for the program's life). + program_recorded: bool, + /// Latest video PTS: the media clock used to timestamp private sections, which /// carry no PES PTS of their own. Unwrapped independently of the video stream. /// SPTS scope: one clock for the whole input. Under MPTS every program's video /// advances it, so a cue could be stamped with another program's PTS. @@ -84,13 +99,13 @@ pub struct Import { media_unwrap: PtsUnwrap, } -impl Import { +impl Import { pub fn new(broadcast: moq_net::BroadcastProducer, catalog: crate::catalog::Producer) -> Self { let feed = Feed::default(); // Sample the real catalog once at construction, not E::default(): an extension // may carry the section by value, and a snapshot clones under the mutex (no publish). let mut snapshot = catalog.snapshot(); - let supports_scte35 = snapshot.scte35_mut().is_some(); + let supports_mpegts = snapshot.mpegts_mut().is_some(); Self { broadcast, catalog, @@ -103,8 +118,11 @@ impl Import { audio_burst: None, scratch: Vec::new(), synced: false, - scte: HashMap::new(), - supports_scte35, + sections: HashMap::new(), + supports_mpegts, + es_descriptors: HashMap::new(), + recorded_media: HashSet::new(), + program_recorded: false, last_pts: None, media_unwrap: PtsUnwrap::default(), } @@ -140,10 +158,11 @@ impl Import { buf.advance(len); } - // Route one whole packet at a time. SCTE-35 PIDs are intercepted here (the - // reader would PES-parse their sections and abort); every other packet is - // fed to the reader. Per-packet so a PMT is parsed (and any SCTE PID - // registered) before the packets that follow it in the same chunk route. + // Route one whole packet at a time. Section-framed verbatim PIDs are + // intercepted here (the reader would PES-parse their sections and abort); + // every other packet is fed to the reader. Per-packet so a PMT is parsed + // (and any section PID registered) before the packets that follow it in the + // same chunk route. let mut off = 0; while off + TsPacket::SIZE <= self.scratch.len() { // A TS packet starts with the 0x47 sync byte. Once synced we trust it and stride @@ -186,13 +205,14 @@ impl Import { off += TsPacket::SIZE; let pid = (((pkt[1] & 0x1f) as u16) << 8) | pkt[2] as u16; let pts = self.last_pts.unwrap_or(Timestamp::ZERO); - if let Some(scte) = self.scte.get_mut(&pid) { - scte.packet(&pkt, pts)?; + if let Some(section) = self.sections.get_mut(&pid) { + section.packet(&pkt, pts)?; continue; } - // PIDs we don't decode (`Stream::Ignored`: unsupported codecs, or a 0x86 - // section PID without CUEI) are dropped here, not fed to the PES reader, - // which aborts on private sections (spec section 7: never fatal). + // PIDs we don't decode and don't carry (`Stream::Ignored`: a base catalog's + // undecoded streams, or an ambiguous 0x86 PID without CUEI) are dropped here, + // not fed to the PES reader, which aborts on private sections (spec section 7: + // never fatal). if let Ok(p) = Pid::new(pid) && matches!(self.streams.get(&p), Some(Stream::Ignored)) { @@ -231,15 +251,37 @@ impl Import { // format_identifier 'CUEI' (ITU-T J.181). The stream itself uses // stream_type 0x86, which mpeg2ts maps to a DTS audio variant, so // detection keys off the CUEI descriptor, not the stream type alone. - let scte = pmt + let cuei = pmt .program_info .iter() .any(|d| d.tag == 0x05 && d.data.len() >= 4 && &d.data[0..4] == b"CUEI"); + + // Record the program-level descriptors once (PMT program_info is stable); + // export re-emits them verbatim, including the original CUEI. + if self.supports_mpegts && !self.program_recorded && !pmt.program_info.is_empty() { + let program = to_descriptors(&pmt.program_info); + if let Some(mpegts) = self.catalog.lock().mpegts_mut() { + mpegts.program_descriptors = program; + } + self.program_recorded = true; + } + for es in &pmt.es_info { - if scte && matches!(es.stream_type, StreamType::Dts8ChannelLosslessAudio) { - self.ensure_scte(es.elementary_pid)?; + let stream_type = es.stream_type as u8; + // Stash ES descriptors so a decoded media track can record them once its + // (lazily created) track exists; verbatim streams record their own. + if self.supports_mpegts { + self.es_descriptors + .insert(es.elementary_pid.as_u16(), to_descriptors(&es.descriptors)); + } + // Section-framed private data is intercepted before the reader (which + // aborts on private sections): private sections (0x05) and CUEI-marked + // SCTE-35 (0x86). Everything else routes through ensure_stream (a decoded + // codec, PES-framed verbatim, or dropped). + if stream_type == 0x05 || (cuei && matches!(es.stream_type, StreamType::Dts8ChannelLosslessAudio)) { + self.ensure_section(es.elementary_pid, stream_type, &es.descriptors)?; } else { - self.ensure_stream(es.elementary_pid, es.stream_type)?; + self.ensure_stream(es.elementary_pid, es.stream_type, &es.descriptors)?; } } } @@ -255,7 +297,21 @@ impl Import { Ok(()) } - fn ensure_stream(&mut self, pid: Pid, stream_type: StreamType) -> anyhow::Result<()> { + fn ensure_stream( + &mut self, + pid: Pid, + stream_type: StreamType, + descriptors: &[mpeg2ts::ts::Descriptor], + ) -> anyhow::Result<()> { + // A later PMT can remap a PID that was section-framed (intercepted in + // `decode`) to a PES codec/verbatim stream. Drop the stale section route first, + // or it would keep intercepting the PID and the new stream would never get data. + // This only fires on a genuine remap: section PIDs otherwise route to + // `ensure_section`, never here. + if let Some(mut section) = self.sections.remove(&pid.as_u16()) { + section.finish()?; + self.pending.remove(&pid); + } if self.streams.contains_key(&pid) { return Ok(()); } @@ -289,9 +345,30 @@ impl Import { StreamType::DolbyDigitalUpToSixChannelAudio => self.legacy_stream(&ac3::DESCRIPTOR), StreamType::DolbyDigitalPlusUpTo16ChannelAudioForAtsc => self.legacy_stream(&eac3::DESCRIPTOR), StreamType::Mpeg1Video | StreamType::Mpeg2Video => Stream::Clock, + // A codec we don't decode. Carry it verbatim as PES when the catalog supports + // the `mpegts` section. 0x86 is excluded: it's ambiguous (DTS audio, or a + // non-conformant SCTE-35 mux without CUEI, which is sections the PES reader + // would abort on), so drop it rather than risk feeding sections to the reader. other => { - tracing::warn!(?other, pid = pid.as_u16(), "unsupported TS stream type, dropping"); - Stream::Ignored + if self.supports_mpegts && !matches!(other, StreamType::Dts8ChannelLosslessAudio) { + let descriptors = to_descriptors(descriptors); + match VerbatimStream::new( + self.broadcast.clone(), + self.catalog.clone(), + pid.as_u16(), + stream_type as u8, + descriptors, + ) { + Ok(stream) => Stream::Verbatim(Box::new(stream)), + Err(err) => { + tracing::warn!(?err, pid = pid.as_u16(), "failed to create verbatim stream, dropping"); + Stream::Ignored + } + } + } else { + tracing::warn!(?other, pid = pid.as_u16(), "unsupported TS stream type, dropping"); + Stream::Ignored + } } }; @@ -315,33 +392,47 @@ impl Import { })) } - /// Register a SCTE-35 PID: intercepted (see [`Self::decode`]) with a cue track when - /// the catalog carries the section, dropped as `Ignored` when it can't. - fn ensure_scte(&mut self, pid: Pid) -> anyhow::Result<()> { - if self.scte.contains_key(&pid.as_u16()) { + /// Register a section-framed verbatim PID (SCTE-35 or other private sections): + /// intercepted (see [`Self::decode`]) with a verbatim track when the catalog + /// carries the `mpegts` section, dropped as `Ignored` when it can't. + fn ensure_section( + &mut self, + pid: Pid, + stream_type: u8, + descriptors: &[mpeg2ts::ts::Descriptor], + ) -> anyhow::Result<()> { + if self.sections.contains_key(&pid.as_u16()) { return Ok(()); } - // This PID is becoming SCTE; drop any partial PES a prior codec left pending. + // This PID is becoming section-framed; drop any partial PES a prior codec left pending. self.pending.remove(&pid); - if !self.supports_scte35 { + if !self.supports_mpegts { // Always route to Ignored, replacing any prior codec on this PID (a later PMT // can reassign it), so a private section never reaches the PES reader. Warn once. if !matches!(self.streams.insert(pid, Stream::Ignored), Some(Stream::Ignored)) { tracing::warn!( pid = pid.as_u16(), - "SCTE-35 detected without catalog support; dropping cues" + "private section stream detected without `mpegts` catalog support; dropping" ); } return Ok(()); } - // A pre-CUEI PMT may have routed this PID to Ignored; drop it so the PID has one route. + // A prior PMT may have routed this PID to Ignored; drop it so the PID has one route. self.streams.remove(&pid); - let stream = ScteStream::new(self.broadcast.clone(), self.catalog.clone())?; - self.scte.insert(pid.as_u16(), stream); + let descriptors = to_descriptors(descriptors); + let stream = SectionStream::new( + self.broadcast.clone(), + self.catalog.clone(), + pid.as_u16(), + stream_type, + descriptors, + )?; + self.sections.insert(pid.as_u16(), stream); self.initialized = true; tracing::debug!( pid = pid.as_u16(), - "SCTE-35 stream detected (CUEI); intercepting before the reader" + stream_type, + "private section stream detected; intercepting before the reader" ); Ok(()) } @@ -381,6 +472,7 @@ impl Import { let data_len = pes_data_len(&pes.header, pes.pes_packet_len); let mut pending = Pending { pts: pes.header.pts.map(|t| t.as_u64()), + stream_id: pes.header.stream_id.as_u8(), data: Vec::with_capacity(pes.data.len()), data_len, }; @@ -428,7 +520,39 @@ impl Import { let Some(stream) = self.streams.get_mut(&pid) else { return Ok(()); }; - stream.write(pending, run_start) + stream.write(pending, run_start)?; + + // Record the decoded media track's PID + PMT descriptors (language, ...) once + // its lazily created track exists, so export can preserve them. + self.record_media_track(pid); + Ok(()) + } + + /// Record a decoded media stream's PID and ES descriptors into `mpegts.tracks`, + /// once per track. No-op without the `mpegts` section, before the track exists, + /// or for verbatim streams (which self-register). + fn record_media_track(&mut self, pid: Pid) { + if !self.supports_mpegts || self.recorded_media.contains(&pid) { + return; + } + let (name, descriptors) = { + let Some(name) = self.streams.get(&pid).and_then(|s| s.media_track_name()) else { + return; + }; + ( + name, + self.es_descriptors.get(&pid.as_u16()).cloned().unwrap_or_default(), + ) + }; + if let Some(mpegts) = self.catalog.lock().mpegts_mut() { + let entry = mpegts + .tracks + .entry(name) + .or_insert_with(|| catalog::Track::new(pid.as_u16())); + entry.pid = pid.as_u16(); + entry.descriptors = descriptors; + } + self.recorded_media.insert(pid); } /// Close the current group on every track and reopen at `sequence`. @@ -436,8 +560,8 @@ impl Import { for stream in self.streams.values_mut() { stream.seek(sequence)?; } - for scte in self.scte.values_mut() { - scte.seek(sequence)?; + for section in self.sections.values_mut() { + section.seek(sequence)?; } Ok(()) } @@ -451,8 +575,8 @@ impl Import { for stream in self.streams.values_mut() { stream.finish()?; } - for scte in self.scte.values_mut() { - scte.finish()?; + for section in self.sections.values_mut() { + section.finish()?; } Ok(()) } @@ -462,46 +586,99 @@ impl Import { struct Pending { /// Raw 90 kHz PTS, before wrap-unwrapping. pts: Option, + /// PES stream_id, preserved for verbatim PES carriage. + stream_id: u8, data: Vec, /// Expected payload length for bounded PES, else `None` (unbounded video). data_len: Option, } -/// Publishes reassembled SCTE-35 `splice_info_section`s as frames on a track in -/// the catalog's typed scte35 section. +/// Convert mpeg2ts PMT descriptors to the catalog's verbatim form. +fn to_descriptors(descriptors: &[mpeg2ts::ts::Descriptor]) -> Vec { + descriptors + .iter() + .map(|d| catalog::Descriptor { + tag: d.tag, + data: bytes::Bytes::copy_from_slice(&d.data), + }) + .collect() +} + +/// Create a verbatim track and record it in the `mpegts` catalog section as a +/// [`Track`](catalog::Track) with a `verbatim` carriage record. Shared by the +/// section- and PES-framed paths. +fn register_verbatim( + broadcast: &mut moq_net::BroadcastProducer, + catalog: &mut crate::catalog::Producer, + pid: u16, + stream_type: u8, + framing: catalog::Framing, + descriptors: Vec, +) -> anyhow::Result> { + let track = broadcast.unique_track(".ts")?; + + let mut guard = catalog.lock(); + let Some(mpegts) = guard.mpegts_mut() else { + // supports_mpegts was true when sampled at construction; None here means the + // catalog dropped the section since. + anyhow::bail!("catalog extension no longer carries an mpegts section"); + }; + mpegts.tracks.insert( + track.name.clone(), + catalog::Track { + pid, + descriptors, + verbatim: Some(catalog::Verbatim::new(stream_type, framing)), + }, + ); + drop(guard); + + Ok(crate::container::Producer::new( + track, + crate::catalog::hang::Container::Legacy, + )) +} + +/// Remove a verbatim track's entry from the `mpegts` catalog section on drop. +fn unregister_verbatim(catalog: &mut crate::catalog::Producer, name: &str) { + if let Some(mpegts) = catalog.lock().mpegts_mut() { + mpegts.tracks.remove(name); + } +} + +/// Publishes reassembled private sections (SCTE-35 and others) as verbatim frames +/// on a track described in the `mpegts` catalog section. /// -/// SCTE-35 rides in private sections (table_id 0xFC), not PES, so this PID is +/// Private sections (e.g. SCTE-35 table_id 0xFC) are not PES, so this PID is /// intercepted before the mpeg2ts reader (which would PES-parse it and abort). -/// The byte-level reassembly lives in [`ScteReassembler`]; this type owns the +/// The byte-level reassembly lives in [`SectionReassembler`]; this type owns the /// track and catalog entry and stamps each section with the media clock. -struct ScteStream { +struct SectionStream { track: crate::container::Producer, catalog: crate::catalog::Producer, - reassembler: ScteReassembler, + reassembler: SectionReassembler, } -impl ScteStream { +impl SectionStream { fn new( mut broadcast: moq_net::BroadcastProducer, mut catalog: crate::catalog::Producer, + pid: u16, + stream_type: u8, + descriptors: Vec, ) -> anyhow::Result { - let mut guard = catalog.lock(); - let Some(scte35) = guard.scte35_mut() else { - // supports_scte35 was true when sampled at construction; None here means - // the catalog dropped the section since. - anyhow::bail!("catalog extension no longer carries a scte35 section"); - }; - - let track = broadcast.unique_track(".scte35")?; - let mut config = scte35::Config::new(); - config.container = hang::catalog::Container::Legacy; - scte35.renditions.insert(track.name.clone(), config); - drop(guard); - + let track = register_verbatim( + &mut broadcast, + &mut catalog, + pid, + stream_type, + catalog::Framing::Section, + descriptors, + )?; Ok(Self { - track: crate::container::Producer::new(track, crate::catalog::hang::Container::Legacy), + track, catalog, - reassembler: ScteReassembler::default(), + reassembler: SectionReassembler::default(), }) } @@ -540,23 +717,104 @@ impl ScteStream { } } -impl Drop for ScteStream { +impl Drop for SectionStream { fn drop(&mut self) { - if let Some(scte35) = self.catalog.lock().scte35_mut() { - scte35.renditions.remove(&self.track.name); + let name = self.track.name.clone(); + unregister_verbatim(&mut self.catalog, &name); + } +} + +/// Publishes whole reassembled PES payloads verbatim as frames on a track +/// described in the `mpegts` catalog section, for elementary streams we don't decode +/// (DTS audio, private PES, teletext, ...). +/// +/// Unlike [`SectionStream`], these ride the normal PES reassembly path, so this +/// type only stamps each PES payload with its (unwrapped) PTS and writes it. +struct VerbatimStream { + track: crate::container::Producer, + catalog: crate::catalog::Producer, + unwrap: PtsUnwrap, + /// Whether the PES stream_id has been recorded into the catalog yet (once). + stream_id_recorded: bool, +} + +impl VerbatimStream { + fn new( + mut broadcast: moq_net::BroadcastProducer, + mut catalog: crate::catalog::Producer, + pid: u16, + stream_type: u8, + descriptors: Vec, + ) -> anyhow::Result { + let track = register_verbatim( + &mut broadcast, + &mut catalog, + pid, + stream_type, + catalog::Framing::Pes, + descriptors, + )?; + Ok(Self { + track, + catalog, + unwrap: PtsUnwrap::default(), + stream_id_recorded: false, + }) + } + + /// Publish one reassembled PES payload verbatim, in its own group, stamped with + /// its PTS (or zero when the PES carried none). + fn write(&mut self, pending: Pending) -> anyhow::Result<()> { + // Record the original PES stream_id once, from the first PES, so export + // re-emits the stream under its real id (e.g. 0xBD for teletext/DVB AC-3). + if !self.stream_id_recorded { + let name = self.track.name.clone(); + if let Some(mpegts) = self.catalog.lock().mpegts_mut() + && let Some(verbatim) = mpegts.tracks.get_mut(&name).and_then(|t| t.verbatim.as_mut()) + { + verbatim.stream_id = Some(pending.stream_id); + } + self.stream_id_recorded = true; } + + let pts = unwrap_pts(&mut self.unwrap, pending.pts)?.unwrap_or(Timestamp::ZERO); + let frame = crate::container::Frame { + timestamp: pts, + payload: bytes::Bytes::from(pending.data), + keyframe: true, + }; + self.track.write(frame)?; + self.track.finish_group()?; + Ok(()) + } + + fn seek(&mut self, sequence: u64) -> anyhow::Result<()> { + self.track.seek(sequence)?; + Ok(()) + } + + fn finish(&mut self) -> anyhow::Result<()> { + self.track.finish()?; + Ok(()) + } +} + +impl Drop for VerbatimStream { + fn drop(&mut self) { + let name = self.track.name.clone(); + unregister_verbatim(&mut self.catalog, &name); } } /// Byte-level reassembler for MPEG-TS private sections on one PID. /// -/// SCTE-35 rides in private sections (table_id 0xFC), not PES. This handles +/// Private sections (SCTE-35 table_id 0xFC and others) are not PES. This handles /// pointer_field alignment, sections split across packets (including a 3-byte /// header split, where section_length is not yet known), continuity-counter /// gaps, and adaptation-field discontinuities. Deliberately private and minimal: -/// just enough to recover whole splice_info_sections. +/// just enough to recover whole sections verbatim. #[derive(Default)] -struct ScteReassembler { +struct SectionReassembler { /// Bytes of the section currently being reassembled. Its 3-byte header (and /// thus section_length) may not all be present yet, so completeness is /// re-checked as bytes arrive; empty means no section in progress. @@ -567,9 +825,8 @@ struct ScteReassembler { last_pkt: Option<[u8; 188]>, } -impl ScteReassembler { - /// Consume one 188-byte TS packet, appending every completed - /// splice_info_section (table_id 0xFC) to `out`. +impl SectionReassembler { + /// Consume one 188-byte TS packet, appending every completed section to `out`. fn push(&mut self, pkt: &[u8], out: &mut Vec>) { // transport_error_indicator: the demodulator flagged this packet as corrupt, // so its payload can't be trusted (and we don't validate CRC-32). Drop it and @@ -667,8 +924,8 @@ impl ScteReassembler { /// Move every complete section out of `acc` into `out`, stopping at the first /// partial. The 3-byte header (which holds section_length) can itself be split /// across TS packets, so a short buffer waits for more bytes rather than being - /// dropped. Only splice_info_sections (table_id 0xFC) are kept; anything else - /// that slipped through PID detection is consumed and discarded. + /// dropped. Every complete section is carried verbatim (SCTE-35 and any other + /// private-section table on the PID); only 0xff stuffing is dropped. fn drain(&mut self, out: &mut Vec>) { loop { match self.acc.first() { @@ -684,9 +941,9 @@ impl ScteReassembler { return; } let section_length = (((self.acc[1] & 0x0f) as usize) << 8) | self.acc[2] as usize; - // SCTE-35 sections are tiny; section_length tops out at 4093 per spec. A - // larger value means we are misparsing garbage, so drop and resync at the - // next pointer_field rather than buffering up to ~4 KB of junk. + // section_length tops out at 4093 per spec (12-bit field, top 2 bits zero). A + // larger value means we are misparsing garbage, so drop and resync at the next + // pointer_field rather than buffering up to ~4 KB of junk. if section_length > 4093 { self.acc.clear(); return; @@ -695,16 +952,13 @@ impl ScteReassembler { if self.acc.len() < full { return; } - let section: Vec = self.acc.drain(..full).collect(); - if section.first() == Some(&0xfc) { - out.push(section); - } + out.push(self.acc.drain(..full).collect()); } } } /// One elementary stream's codec importer plus PTS-unwrap state. -enum Stream { +enum Stream { H264 { import: Box>, unwrap: PtsUnwrap, @@ -715,13 +969,15 @@ enum Stream { }, Aac(Box>), Legacy(Box>), - /// MPEG-1/2 video we don't decode, kept only to advance the SCTE-35 media clock. + /// A codec we don't decode, carried verbatim as PES (DTS audio, private PES, ...). + Verbatim(Box>), + /// MPEG-1/2 video we don't decode, kept only to advance the media clock. /// `is_video` counts it, so never reuse this variant for audio or data. Clock, Ignored, } -impl Stream { +impl Stream { fn write(&mut self, pending: Pending, burst: Option) -> anyhow::Result<()> { match self { Stream::H264 { import, unwrap } => { @@ -734,6 +990,7 @@ impl Stream { } Stream::Aac(stream) => stream.write(pending, burst), Stream::Legacy(stream) => stream.write(pending), + Stream::Verbatim(stream) => stream.write(pending), Stream::Clock | Stream::Ignored => Ok(()), } } @@ -744,6 +1001,7 @@ impl Stream { Stream::H265 { import, .. } => import.seek(sequence), Stream::Aac(stream) => stream.seek(sequence), Stream::Legacy(stream) => stream.seek(sequence), + Stream::Verbatim(stream) => stream.seek(sequence), Stream::Clock | Stream::Ignored => Ok(()), } } @@ -754,9 +1012,22 @@ impl Stream { Stream::H265 { import, .. } => import.finish(), Stream::Aac(stream) => stream.finish(), Stream::Legacy(stream) => stream.finish(), + Stream::Verbatim(stream) => stream.finish(), Stream::Clock | Stream::Ignored => Ok(()), } } + + /// The MoQ track name of a decoded media stream, once its (lazily created) track + /// exists. `None` for verbatim/clock/ignored streams (verbatim self-registers). + fn media_track_name(&self) -> Option { + match self { + Stream::H264 { import, .. } => import.track().map(|t| t.name.clone()), + Stream::H265 { import, .. } => import.track().ok().map(|t| t.name.clone()), + Stream::Aac(stream) => stream.import.as_ref().map(|i| i.track().name.clone()), + Stream::Legacy(stream) => stream.import.as_ref().map(|i| i.name().to_string()), + Stream::Verbatim(_) | Stream::Clock | Stream::Ignored => None, + } + } } /// AAC needs the first ADTS header before it can build a [`aac::Import`] @@ -1088,7 +1359,7 @@ impl Read for Feed { mod test { use mpeg2ts::es::StreamType; - use super::ScteReassembler; + use super::SectionReassembler; use crate::container::Timestamp; // libklvanc public-sample cue: table_id 0xFC, section_length 0x1b (27), 30 bytes total. @@ -1132,7 +1403,7 @@ mod test { } fn run(pkts: &[Vec]) -> Vec> { - let mut r = ScteReassembler::default(); + let mut r = SectionReassembler::default(); let mut out = Vec::new(); for p in pkts { r.push(p, &mut out); @@ -1146,12 +1417,13 @@ mod test { } #[test] - fn filters_non_scte() { - // A table_id 0x00 section ahead of the cue: only the 0xFC cue is emitted, and - // the filtered section doesn't desync parsing of what follows it. - let mut body = fake_section(0x00, 5); + fn carries_all_sections_verbatim() { + // A non-SCTE table_id 0x00 section ahead of the cue: both are carried verbatim + // (we no longer filter by table_id), and back-to-back sections parse cleanly. + let other = fake_section(0x00, 5); + let mut body = other.clone(); body.extend_from_slice(&CUE); - assert_eq!(run(&[packet(true, 0, 0, &body)]), vec![CUE.to_vec()]); + assert_eq!(run(&[packet(true, 0, 0, &body)]), vec![other, CUE.to_vec()]); } #[test] @@ -1312,11 +1584,10 @@ mod test { #[test] fn scte35_extension_catalogs_the_cue_track() { use crate::catalog::hang::Catalog; - use crate::container::ts::scte35; + use crate::container::ts::catalog::Ext; let mut broadcast = moq_net::Broadcast::new().produce(); - let catalog = - crate::catalog::Producer::with_catalog(&mut broadcast, Catalog::::default()).unwrap(); + let catalog = crate::catalog::Producer::with_catalog(&mut broadcast, Catalog::::default()).unwrap(); let mut import = super::Import::new(broadcast, catalog.clone()); let mut bytes = bytes::BytesMut::new(); @@ -1326,7 +1597,7 @@ mod test { import.finish().unwrap(); assert_eq!( - catalog.snapshot().scte35.renditions.len(), + catalog.snapshot().mpegts.tracks.len(), 1, "expected one scte35 rendition" ); @@ -1348,7 +1619,10 @@ mod test { import.decode(&mut bytes).unwrap(); // must not abort on the private section import.finish().unwrap(); - assert!(import.scte.is_empty(), "no cue stream is created for a base catalog"); + assert!( + import.sections.is_empty(), + "no cue stream is created for a base catalog" + ); assert!( matches!( import.streams.get(&mpeg2ts::ts::Pid::new(0x21).unwrap()), @@ -1373,15 +1647,14 @@ mod test { async fn pmt_without_cuei_then_with_cuei_upgrades() { use crate::catalog::hang::{Catalog, Container}; use crate::container::Consumer; - use crate::container::ts::scte35; + use crate::container::ts::catalog::Ext; const SECTION_PID: u16 = 0x0021; let pid = mpeg2ts::ts::Pid::new(SECTION_PID).unwrap(); let mut broadcast = moq_net::Broadcast::new().produce(); let consumer = broadcast.consume(); - let catalog = - crate::catalog::Producer::with_catalog(&mut broadcast, Catalog::::default()).unwrap(); + let catalog = crate::catalog::Producer::with_catalog(&mut broadcast, Catalog::::default()).unwrap(); let mut import = super::Import::new(broadcast, catalog.clone()); // First PMT lacks CUEI: the 0x86 PID is ambiguous and routes to Ignored. @@ -1408,12 +1681,12 @@ mod test { "upgrade drops the stale Ignored route" ); assert_eq!( - catalog.snapshot().scte35.renditions.len(), + catalog.snapshot().mpegts.tracks.len(), 1, "upgrade advertises the cue track" ); - let name = catalog.snapshot().scte35.renditions.keys().next().unwrap().clone(); + let name = catalog.snapshot().mpegts.tracks.keys().next().unwrap().clone(); let track = consumer.subscribe_track(&moq_net::Track::new(name)).unwrap(); let mut reader = Consumer::new(track, Container::Legacy).with_latency(std::time::Duration::ZERO); let frame = tokio::time::timeout(std::time::Duration::from_secs(1), reader.read()) @@ -1694,15 +1967,14 @@ mod test { #[tokio::test(start_paused = true)] async fn scte35_cue_stamped_with_video_pts() { use crate::catalog::hang::{Catalog, Container}; - use crate::container::ts::scte35; + use crate::container::ts::catalog::Ext; use crate::container::{Consumer, Timestamp}; const VIDEO_PID: u16 = 0x0050; let mut broadcast = moq_net::Broadcast::new().produce(); let consumer = broadcast.consume(); - let catalog = - crate::catalog::Producer::with_catalog(&mut broadcast, Catalog::::default()).unwrap(); + let catalog = crate::catalog::Producer::with_catalog(&mut broadcast, Catalog::::default()).unwrap(); let mut import = super::Import::new(broadcast, catalog.clone()); let mut bytes = bytes::BytesMut::new(); @@ -1719,7 +1991,7 @@ mod test { let clock = import.last_pts.expect("video set the media clock"); import.finish().unwrap(); - let name = catalog.snapshot().scte35.renditions.keys().next().unwrap().clone(); + let name = catalog.snapshot().mpegts.tracks.keys().next().unwrap().clone(); let track = consumer.subscribe_track(&moq_net::Track::new(name)).unwrap(); let mut reader = Consumer::new(track, Container::Legacy).with_latency(std::time::Duration::ZERO); let frame = tokio::time::timeout(std::time::Duration::from_secs(1), reader.read()) @@ -1739,7 +2011,7 @@ mod test { #[test] fn section_pid_without_cuei_is_dropped_not_cataloged() { use crate::catalog::hang::Catalog; - use crate::container::ts::scte35; + use crate::container::ts::catalog::Ext; const VIDEO_PID: u16 = 0x0050; const SECTION_PID: u16 = 0x0021; @@ -1747,8 +2019,7 @@ mod test { let mut broadcast = moq_net::Broadcast::new().produce(); // scte35::Ext (not the base catalog) makes a wrong ensure_scte() observable: it // would create a rendition, which the base catalog silently drops. - let catalog = - crate::catalog::Producer::with_catalog(&mut broadcast, Catalog::::default()).unwrap(); + let catalog = crate::catalog::Producer::with_catalog(&mut broadcast, Catalog::::default()).unwrap(); let mut import = super::Import::new(broadcast, catalog.clone()); let mut bytes = bytes::BytesMut::new(); @@ -1769,7 +2040,7 @@ mod test { "video kept importing past the dropped section PID" ); assert!( - catalog.snapshot().scte35.renditions.is_empty(), + catalog.snapshot().mpegts.tracks.is_empty(), "a 0x86 PID without CUEI must not be cataloged" ); } @@ -1804,4 +2075,55 @@ mod test { let p4 = packet(true, 3, 0, &CUE); // clean PUSI: resync and emit assert_eq!(run(&[p1, p2, p3, p4]), vec![CUE.to_vec()]); } + + // A PES-framed elementary stream we don't decode (private data, stream_type 0x06) + // is carried verbatim: cataloged in the `mpegts` section with its PID and framing, and + // its PES payload published byte-for-byte. + #[tokio::test(start_paused = true)] + async fn private_pes_carried_verbatim() { + use crate::catalog::hang::{Catalog, Container}; + use crate::container::Consumer; + use crate::container::ts::catalog::{Ext, Framing}; + + const VIDEO_PID: u16 = 0x0050; + const DATA_PID: u16 = 0x0052; + + let mut broadcast = moq_net::Broadcast::new().produce(); + let consumer = broadcast.consume(); + let catalog = crate::catalog::Producer::with_catalog(&mut broadcast, Catalog::::default()).unwrap(); + let mut import = super::Import::new(broadcast, catalog.clone()); + + let mut bytes = bytes::BytesMut::new(); + bytes.extend_from_slice(&synth_pmt( + &[ + (StreamType::Mpeg2Video, VIDEO_PID), + (StreamType::Mpeg2PacketizedData, DATA_PID), + ], + false, + )); + bytes.extend_from_slice(&pes_packet(VIDEO_PID, 90_000)); // video sets the media clock + let payload = [0xDE, 0xAD, 0xBE, 0xEF, 0x01, 0x02]; + bytes.extend_from_slice(&audio_pes_packet(DATA_PID, 0, 90_000, &payload)); + import.decode(&mut bytes).unwrap(); + import.finish().unwrap(); + + let snap = catalog.snapshot(); + assert_eq!(snap.mpegts.tracks.len(), 1, "the private PES PID is carried verbatim"); + let (name, track) = snap.mpegts.tracks.iter().next().unwrap(); + let verbatim = track.verbatim.as_ref().expect("a verbatim carriage record"); + assert_eq!(verbatim.stream_type, 0x06, "recorded the PMT stream_type"); + assert_eq!(verbatim.framing, Framing::Pes, "private PES is PES-framed"); + // `audio_pes_packet` uses stream_id 0xC0; it must be captured for faithful re-emit. + assert_eq!(verbatim.stream_id, Some(0xC0), "recorded the PES stream_id"); + assert_eq!(track.pid, DATA_PID, "recorded the original PID"); + + let track = consumer.subscribe_track(&moq_net::Track::new(name.clone())).unwrap(); + let mut reader = Consumer::new(track, Container::Legacy).with_latency(std::time::Duration::ZERO); + let frame = tokio::time::timeout(std::time::Duration::from_secs(1), reader.read()) + .await + .expect("verbatim read timed out") + .unwrap() + .expect("a published verbatim frame"); + assert_eq!(&frame.payload[..], &payload[..], "verbatim PES payload round-trips"); + } } diff --git a/rs/moq-mux/src/container/ts/import_test.rs b/rs/moq-mux/src/container/ts/import_test.rs index 6db4c28ce0..4db9343584 100644 --- a/rs/moq-mux/src/container/ts/import_test.rs +++ b/rs/moq-mux/src/container/ts/import_test.rs @@ -284,7 +284,7 @@ async fn kyrion_dirtystart_extracts_real_cues() { let consumer = broadcast.consume(); let catalog = crate::catalog::Producer::with_catalog( &mut broadcast, - crate::catalog::hang::Catalog::::default(), + crate::catalog::hang::Catalog::::default(), ) .unwrap(); let mut import = crate::container::ts::Import::new(broadcast, catalog.clone()); @@ -295,7 +295,15 @@ async fn kyrion_dirtystart_extracts_real_cues() { let snap = catalog.snapshot(); assert_eq!(snap.video.renditions.len(), 1, "video track lost across the dirty join"); - let name = snap.scte35.renditions.keys().next().expect("scte35 track").clone(); + // Select the SCTE-35 stream by its verbatim stream_type; media tracks also appear + // in mpegts.tracks now (with their PID + descriptors). + let name = snap + .mpegts + .tracks + .iter() + .find(|(_, t)| t.verbatim.as_ref().is_some_and(|v| v.stream_type == 0x86)) + .map(|(name, _)| name.clone()) + .expect("scte35 track"); let track = consumer.subscribe_track(&moq_net::Track::new(name)).unwrap(); let mut reader = crate::container::Consumer::new(track, crate::catalog::hang::Container::Legacy); let mut cues = Vec::new(); diff --git a/rs/moq-mux/src/container/ts/mod.rs b/rs/moq-mux/src/container/ts/mod.rs index 1785289e31..35820070f6 100644 --- a/rs/moq-mux/src/container/ts/mod.rs +++ b/rs/moq-mux/src/container/ts/mod.rs @@ -5,11 +5,18 @@ //! codec layer (H.264/H.265/AAC, plus the legacy MP2/AC-3/E-AC-3 parsers) does //! the elementary-stream parsing; this module only handles PAT/PMT/PES framing, //! PTS, and ADTS framing for AAC. +//! +//! Elementary streams we don't decode (SCTE-35, teletext, DVB subtitles, private +//! data, ...) are carried verbatim, one MoQ track per PID, described in the +//! [`catalog`] (`mpegts`) section. SCTE-35 is just one such stream (`stream_type` 0x86). mod adts; mod export; mod import; -pub mod scte35; + +/// The `mpegts` catalog section: per-track PID + descriptors plus verbatim +/// carriage of undecoded elementary streams. +pub mod catalog; pub use export::*; pub use import::*; diff --git a/rs/moq-mux/src/container/ts/scte35.rs b/rs/moq-mux/src/container/ts/scte35.rs deleted file mode 100644 index 4caca50ae8..0000000000 --- a/rs/moq-mux/src/container/ts/scte35.rs +++ /dev/null @@ -1,72 +0,0 @@ -//! SCTE-35 application catalog extension for ingesting MPEG-TS splice cues. - -use std::collections::BTreeMap; - -use serde::{Deserialize, Serialize}; - -use crate::catalog::hang::CatalogExt; - -/// SCTE-35 splice cue tracks: a map of renditions (one per MPEG-TS PID), each -/// carried as the verbatim `splice_info_section` bytes. -#[derive(Serialize, Deserialize, Debug, Clone, PartialEq, Default)] -#[serde(rename_all = "camelCase")] -pub struct Cues { - pub renditions: BTreeMap, -} - -impl Cues { - /// Omitted from the catalog when empty, so a broadcast without cues stays byte-identical. - pub fn is_empty(&self) -> bool { - self.renditions.is_empty() - } -} - -/// One SCTE-35 cue track. Records how the verbatim section was framed; the -/// stream_type (0x86) and CUEI signaling are implicit to SCTE-35. -#[derive(Serialize, Deserialize, Debug, Clone, PartialEq, Default)] -#[serde(rename_all = "camelCase")] -#[non_exhaustive] -pub struct Config { - #[serde(default)] - pub container: hang::catalog::Container, -} - -impl Config { - pub fn new() -> Self { - Self::default() - } -} - -/// The application catalog extension carrying the `scte35` section. Empty by -/// default, so the section is omitted until a cue track is added. -#[derive(Serialize, Deserialize, Debug, Clone, PartialEq, Default)] -pub struct Ext { - #[serde(default, skip_serializing_if = "Cues::is_empty")] - pub scte35: Cues, -} - -impl CatalogExt for Ext {} - -/// An extension that can carry an SCTE-35 catalog section. -/// -/// Implement this for an application extension to compose SCTE-35 with -/// additional sections. -pub trait Catalog: CatalogExt { - /// The section to write cues into, or `None` for an extension that doesn't carry them. - /// - /// Keep this stable per catalog: an importer samples support once at construction, so a - /// result that flips between `Some` and `None` mid-stream would disable cues or fail. - fn scte35_mut(&mut self) -> Option<&mut Cues>; -} - -impl Catalog for () { - fn scte35_mut(&mut self) -> Option<&mut Cues> { - None - } -} - -impl Catalog for Ext { - fn scte35_mut(&mut self) -> Option<&mut Cues> { - Some(&mut self.scte35) - } -}