diff --git a/dart/moq_ffi/lib/src/moq.dart b/dart/moq_ffi/lib/src/moq.dart index 5a218fd041..a6e1af765e 100644 --- a/dart/moq_ffi/lib/src/moq.dart +++ b/dart/moq_ffi/lib/src/moq.dart @@ -7022,7 +7022,7 @@ class LocMoqContainer extends MoqContainer { } } -enum MoqContainerFormat { fmp4, mkv, ts, flv, cmaf } +enum MoqContainerFormat { fmp4, mkv, ts, flv } class FfiConverterMoqContainerFormat { static LiftRetVal read(Uint8List buf) { @@ -7036,8 +7036,6 @@ class FfiConverterMoqContainerFormat { return LiftRetVal(MoqContainerFormat.ts, 4); case 4: return LiftRetVal(MoqContainerFormat.flv, 4); - case 5: - return LiftRetVal(MoqContainerFormat.cmaf, 4); default: throw UniffiInternalError( UniffiInternalError.unexpectedEnumCase, diff --git a/doc/lib/c/index.md b/doc/lib/c/index.md index 63160ab3f1..6921727125 100644 --- a/doc/lib/c/index.md +++ b/doc/lib/c/index.md @@ -65,7 +65,7 @@ inside. The header is the reference: each function carries a doc comment. - **Threading.** Call any function from any thread. Raw publish calls block until the codec takes the frame, which paces a publisher. - **Zeroed config is the default.** A zeroed `moq_client_config` means the default for every field, so new fields don't break callers. A field whose default isn't zero has a `has_*` flag: `backoff_timeout_us = 0` means "retry forever" only with `has_backoff_timeout = true`. - **Compressed tracks.** `moq_track_info.compress` on a raw track sends each group DEFLATE-compressed to opt-in `moq-lite-07-wip` peers; other peers, and every reader here, get the original frames. `@moq/net` cannot read a compressed track yet. -- **fMP4 import.** `MOQ_CONTAINER_FORMAT_FMP4` splits each fragment into one frame per sample; `MOQ_CONTAINER_FORMAT_CMAF` republishes it whole. +- **fMP4 import.** `MOQ_CONTAINER_FORMAT_FMP4` splits each fragment into one frame per sample. - **Audio needs cuts.** Video groups at its keyframes, but audio forms a group only where you call `moq_publish_media_cut`: after every frame, or at a segment cadence to align with video. - **Ending cleanly.** When the input ends, finish every track, then call `moq_publish_finish()`. It consumes the broadcast and returns a finish handle that keeps it announced and serving the finished tracks, writing the linger, their longest max age, to `linger_us`. Keep the handle for that long on your own clock so a subscriber still catching up can drain them, then `moq_publish_finish_close()` it; closing early is the interrupt. `moq_publish_close()` stays the abrupt end. - **Timeline breaks.** On a seek or pause, call `moq_publish_media_discontinuity`, then keep timestamps moving forward and resume video on a keyframe. diff --git a/doc/lib/cpp/index.md b/doc/lib/cpp/index.md index 41fe064fb5..0367c1ad0d 100644 --- a/doc/lib/cpp/index.md +++ b/doc/lib/cpp/index.md @@ -109,7 +109,7 @@ Share `moq::MediaTimebase::init(broadcast)` through `ContainerInit` when several use one source timeline; omitting it keeps an independent offset. A timebase belongs to its broadcast and does not keep it open. `MediaContainerProducer` takes chunks split anywhere, so a raw byte stream needs no other importer. -`moq::ContainerFormat::kFmp4` splits each fragment into one frame per sample; `kCmaf` republishes it whole. +`moq::ContainerFormat::kFmp4` splits each fragment into one frame per sample. ```cpp // Publish encoded frames, or raw pixels with the codec inside the binding. diff --git a/doc/lib/dart/index.md b/doc/lib/dart/index.md index 31519133ba..4e376cf706 100644 --- a/doc/lib/dart/index.md +++ b/doc/lib/dart/index.md @@ -30,7 +30,7 @@ Share `media.Timebase(broadcast: ...)` through `ContainerInit` when several impo one source timeline; omitting it keeps an independent offset. A timebase belongs to its broadcast and does not keep it open. `ContainerProducer` takes chunks split anywhere, so a raw byte stream needs no other importer. -`ContainerFormat.fmp4` splits each fragment into one frame per sample; `ContainerFormat.cmaf` republishes it whole. +`ContainerFormat.fmp4` splits each fragment into one frame per sample. `media.CatalogConsumer.subscribe`, `media.TrackConsumer.subscribe`, and `media.GroupConsumer.fetch` construct the read side from a broadcast; the last two take a catalog `media.Rendition` and follow its broadcast and track themselves. For arrival-order reads, decode each group from diff --git a/doc/lib/go/index.md b/doc/lib/go/index.md index a050c7d0c1..b2a267b120 100644 --- a/doc/lib/go/index.md +++ b/doc/lib/go/index.md @@ -31,7 +31,7 @@ Share `media.NewTimebase(broadcast)` through `ContainerInit` when several import source timeline; omitting it keeps an independent offset. Timebases belong to one broadcast and do not keep it open; release the handle with `Close` on the calling goroutine. `NewContainerProducer` takes chunks split anywhere, so a raw byte stream needs no other importer. -`ContainerFormatFmp4` splits each fragment into one frame per sample; `ContainerFormatCmaf` republishes it whole. +`ContainerFormatFmp4` splits each fragment into one frame per sample. For arrival-order reads, decode each group from `moq.TrackConsumer.RecvGroup` with `media.NewGroupConsumer`; frames carry container media time as they land. `media.FetchGroup` fetches and decodes one past group. diff --git a/doc/lib/kt/index.md b/doc/lib/kt/index.md index 54975d9865..5124e91d3e 100644 --- a/doc/lib/kt/index.md +++ b/doc/lib/kt/index.md @@ -29,7 +29,7 @@ Share `Timebase(broadcast)` through `ContainerInit` when several imports use one timeline; omitting it keeps an independent offset. A timebase belongs to its broadcast and does not keep it open. `ContainerProducer` takes chunks split anywhere, so a raw byte stream needs no other importer. -`ContainerFormat.FMP4` splits each fragment into one frame per sample; `ContainerFormat.CMAF` republishes it whole. +`ContainerFormat.FMP4` splits each fragment into one frame per sample. Construct readers with `CatalogConsumer.subscribe`, `TrackConsumer.subscribe`, or `GroupConsumer.fetch`; the last two take a catalog `Rendition` and follow its broadcast and track themselves. For arrival-order reads, decode each group from diff --git a/doc/lib/py/index.md b/doc/lib/py/index.md index 9e17b03a97..7c6a8d5c99 100644 --- a/doc/lib/py/index.md +++ b/doc/lib/py/index.md @@ -21,7 +21,7 @@ Share `media.Timebase(broadcast)` through `ContainerInit` when several imports u source timeline; omitting it keeps an independent offset. A timebase belongs to its broadcast and does not keep it open. `ContainerProducer` takes chunks split anywhere, so a raw byte stream needs no other importer. -`ContainerFormat.FMP4` splits each fragment into one frame per sample; `ContainerFormat.CMAF` republishes it whole. +`ContainerFormat.FMP4` splits each fragment into one frame per sample. `MediaFrame.timestamp`, media writes, and `flush` use `datetime.timedelta`. A missing data timestamp stays untimed; the library never substitutes zero or now. JSON consumers yield `moq.json.Timed(value, timestamp)`; a snapshot consumer yields each state in order, diff --git a/doc/lib/swift/index.md b/doc/lib/swift/index.md index 5a8f3e3dd5..33929473b3 100644 --- a/doc/lib/swift/index.md +++ b/doc/lib/swift/index.md @@ -30,7 +30,7 @@ Share `Media.Timebase(broadcast:)` through `ContainerInit` when several imports source timeline; omitting it keeps an independent offset. A timebase belongs to its broadcast and does not keep it open. `ContainerProducer` takes chunks split anywhere, so a raw byte stream needs no other importer. -`.fmp4` splits each fragment into one frame per sample; `.cmaf` republishes it whole. +`.fmp4` splits each fragment into one frame per sample. For arrival-order reads, decode each group from `TrackConsumer.recvGroup()` with `Media.GroupConsumer(group:container:)`; frames carry container media time as they land. diff --git a/go/wrapper/media/media.go b/go/wrapper/media/media.go index f823ced385..15f41c52e5 100644 --- a/go/wrapper/media/media.go +++ b/go/wrapper/media/media.go @@ -330,8 +330,6 @@ const ( const ( // ContainerFormatFmp4 is fragmented MP4 / CMAF, split into one frame per sample. ContainerFormatFmp4 = ffi.MoqContainerFormatFmp4 - // ContainerFormatCmaf is fragmented MP4 / CMAF, republished whole as one CMAF frame per fragment. - ContainerFormatCmaf = ffi.MoqContainerFormatCmaf // ContainerFormatMkv is Matroska / WebM. ContainerFormatMkv = ffi.MoqContainerFormatMkv // ContainerFormatTs is an MPEG-2 transport stream. diff --git a/quest/m1/api-audit/README.md b/quest/m1/api-audit/README.md index 603b034673..01769b97d2 100644 --- a/quest/m1/api-audit/README.md +++ b/quest/m1/api-audit/README.md @@ -68,7 +68,6 @@ Rejected in planning: - [Two broadcast resolutions](/quest/m1/api-audit/resolve.md) - Rust, JS, and the bindings resolve a broadcast with `request(path, epoch)` or `follow(path, epoch)`, and every racy two-step caller is gone - [The origin announces](/quest/m1/api-audit/announce-owner.md) - a broadcast producer cannot announce; the origin hands out the announcement, so `Prepared` and standalone broadcasts lose methods that always fail -- [Container formats name inputs](/quest/m1/api-audit/container-format.md) - `ContainerFormat` names what is parsed, never how it is packaged; `ContainerStream` and TS passthrough internals are gone - [moq-net exports with a consumer](/quest/m1/api-audit/net-exports.md) - the dead and test-only moq-net, `@moq/net`, and moq-ffi items the follow-up audit found go private or away - [Bindings read gaps and ends](/quest/m1/api-audit/decode-read-bindings.md) - moq-ffi and every wrapper surface `Gap` and `End` instead of discarding them - [Media surface trim](/quest/m1/api-audit/media-trim.md) - media crate items with no consumer go private, and Rust/JS media names agree diff --git a/quest/m1/api-audit/container-format.md b/quest/m1/api-audit/container-format.md deleted file mode 100644 index 3c1385ee54..0000000000 --- a/quest/m1/api-audit/container-format.md +++ /dev/null @@ -1,42 +0,0 @@ -# [M] Container formats name inputs - -## Goal - -`moq_mux::import::ContainerFormat` names only what is parsed. How the import -is packaged (per-sample frames or preserved fragments) is never chosen by an -input-sounding name, and the format strings mean one thing each. -`ContainerStream` and the TS passthrough internals are not public. - -## Plan - -Decided 2026-10-22 (API audit). #6250 made `Fmp4` unwrap to per-sample frames -and added `Cmaf` for fragment passthrough over the same input -(`rs/moq-mux/src/import/format.rs`, `container.rs`). That flip was -deliberate; the defect is the naming axis. `"cmaf"` means passthrough while -`"m2ts"` means demux to TS, though the catalog's `m2ts` section means TS -passthrough, and `ts::Passthrough` is unreachable from `ContainerFormat`. - -- Delete `ContainerFormat::Cmaf` from Rust, moq-ffi, moq-c - (`MOQ_CONTAINER_FORMAT_CMAF`), and every wrapper enum; none has a caller. - Parse `"cmaf"` as `Fmp4`. Passthrough stays Rust-only through - `fmp4::Import::with_passthrough` and `moq_hls::import::Config::passthrough`, - which the CLI already uses. The Impact bullet supersedes the 8 binding - MIGRATE entries #6250 wrote, for the cut to collapse. Chosen over a `ContainerInit::passthrough` flag because nothing - outside Rust asks for passthrough yet. -- Make the `"m2ts"` string agree with the catalog meaning, or rename it so it - does not collide; verify against callers first. -- Delete `ContainerStream` (`import/container.rs`, inherited, a line-for-line - copy of `Container`); `Container::new` with empty `ContainerInit::data` is - equivalent. Keep `Track` and `TrackStream`. Fix `doc/lib/layers.md`. -- TS passthrough (`container/ts/mod.rs`, `passthrough.rs`): make the module - private and export only `Passthrough`; fold its four error structs into - `moq_mux::Error` instead of `anyhow`; consider - `ts::passthrough::{Import, Export}` for the `Passthrough`/`Release` pair. -- Make `Container::cut()` and `fmp4::Import::discontinuity()` return - `Result` like `Track`'s. - -Public API: breaking in moq-mux, moq-ffi, moq-c, and wrappers. Wire: none. - -## Required - -- [Generated C bindings](/quest/m1/c/README.md) - this changes the C API, so its C half lands on the generated package and C users break once (decided 2026-10-22, maintainer) diff --git a/rs/moq-c/src/api.rs b/rs/moq-c/src/api.rs index 5aaf0f8fe7..7091894c5c 100644 --- a/rs/moq-c/src/api.rs +++ b/rs/moq-c/src/api.rs @@ -155,8 +155,6 @@ pub enum moq_container_format { MOQ_CONTAINER_FORMAT_TS = 2, /// Flash Video, as used by RTMP. MOQ_CONTAINER_FORMAT_FLV = 3, - /// Fragmented MP4 / CMAF, republished whole as one CMAF frame per fragment. - MOQ_CONTAINER_FORMAT_CMAF = 4, } /// Configuration for [moq_publish_audio]. @@ -309,7 +307,6 @@ fn container_format_from_u32(value: u32) -> Result ContainerFormat::Mkv, v if v == moq_container_format::MOQ_CONTAINER_FORMAT_TS as u32 => ContainerFormat::Ts, v if v == moq_container_format::MOQ_CONTAINER_FORMAT_FLV as u32 => ContainerFormat::Flv, - v if v == moq_container_format::MOQ_CONTAINER_FORMAT_CMAF as u32 => ContainerFormat::Cmaf, _ => return Err(Error::InvalidCode), }) } diff --git a/rs/moq-c/src/publish.rs b/rs/moq-c/src/publish.rs index 2df5ad18df..3a8086660c 100644 --- a/rs/moq-c/src/publish.rs +++ b/rs/moq-c/src/publish.rs @@ -452,8 +452,9 @@ impl Publish { pub fn container_cut(&mut self, container: Id) -> Result<(), Error> { let container = self.containers.get_mut(container).ok_or(Error::MediaNotFound)?; container.driver.check()?; - container.inner.cut(); - container.driver.check() + let result = container.inner.cut(); + container.driver.check()?; + Ok(result?) } /// Start a new segment and number its groups `sequence`. diff --git a/rs/moq-ffi/src/media.rs b/rs/moq-ffi/src/media.rs index 7da4049c1b..c5c92b3c0a 100644 --- a/rs/moq-ffi/src/media.rs +++ b/rs/moq-ffi/src/media.rs @@ -296,8 +296,6 @@ pub enum MoqContainerFormat { Ts, /// Flash Video, as used by RTMP. Flv, - /// Fragmented MP4 / CMAF, republished whole as one CMAF frame per fragment. - Cmaf, } /// What an audio publish needs: a format, its codec init bytes, and an optional label. @@ -477,7 +475,6 @@ impl From for moq_mux::import::ContainerFormat { fn from(format: MoqContainerFormat) -> Self { match format { MoqContainerFormat::Fmp4 => Self::Fmp4, - MoqContainerFormat::Cmaf => Self::Cmaf, MoqContainerFormat::Mkv => Self::Mkv, MoqContainerFormat::Ts => Self::Ts, MoqContainerFormat::Flv => Self::Flv, @@ -1076,7 +1073,9 @@ impl MoqMediaContainerProducer { let _guard = crate::ffi::enter(); let mut guard = self.inner.lock().unwrap(); let container = guard.as_mut().ok_or(MoqError::Closed)?; - container.import.cut(); + container.import.cut().map_err(|err| MoqError::Codec { + reason: format!("cut failed: {err}"), + })?; Ok(()) })(); self.driver.check()?; diff --git a/rs/moq-hls/src/import.rs b/rs/moq-hls/src/import.rs index a3c0c9c219..f44af054ea 100644 --- a/rs/moq-hls/src/import.rs +++ b/rs/moq-hls/src/import.rs @@ -640,7 +640,7 @@ impl TrackState { .next_discontinuity .is_some_and(|previous| previous != discontinuity_sequence) { - importer.discontinuity(); + importer.discontinuity()?; } if reanchored { importer.seek(group_sequence)?; @@ -648,7 +648,7 @@ impl TrackState { // Every playlist entry is a source segment boundary, so declare it even when the media // carries no styp. Every rendition declares the same ones; the timeline drops a cut // that would land inside its minimum segment duration, so the duplicates cost nothing. - importer.cut(); + importer.cut()?; importer.decode(&bytes).map_err(|err| match err { // Timestamp resets are unsupported: the published tracks never rewind, and no later // pass can import this segment, so it is refused rather than retried. diff --git a/rs/moq-mux/src/container/fmp4/import.rs b/rs/moq-mux/src/container/fmp4/import.rs index 10d50aced1..2d9f831bfd 100644 --- a/rs/moq-mux/src/container/fmp4/import.rs +++ b/rs/moq-mux/src/container/fmp4/import.rs @@ -262,8 +262,9 @@ impl Import { /// This is where audio groups are drawn too: audio rolls its group at a segment boundary, so /// it follows the same segmentation as video. A passthrough video track rolls there as well, /// while an unwrapped one opens a group at each sync sample. - pub fn cut(&mut self) { + pub fn cut(&mut self) -> Result<()> { self.pending_cut = true; + Ok(()) } /// Restrict which track roles are published. @@ -1389,7 +1390,7 @@ impl Import { /// restarts below where it was still fails with /// [`TimestampRewind`](crate::container::TimestampRewind): a track refuses a group that /// rewinds, before publishing any of it. - pub fn discontinuity(&mut self) { + pub fn discontinuity(&mut self) -> Result<()> { for track in self.text.values_mut() { track.end_decode_time = None; track.discontinuity = true; @@ -1397,5 +1398,6 @@ impl Import { for track in self.tracks.values_mut() { track.last_decode_time = None; } + Ok(()) } } diff --git a/rs/moq-mux/src/container/fmp4/import_test.rs b/rs/moq-mux/src/container/fmp4/import_test.rs index e9193feac3..5dc98c98f1 100644 --- a/rs/moq-mux/src/container/fmp4/import_test.rs +++ b/rs/moq-mux/src/container/fmp4/import_test.rs @@ -2212,7 +2212,7 @@ async fn seek_keeps_the_decode_time_until_a_discontinuity_in(passthrough: bool) "{err:?}" ); - fmp4.discontinuity(); + fmp4.discontinuity().unwrap(); fmp4.seek(200).unwrap(); let err = fmp4.decode(&live_session(0, 5)).unwrap_err(); // Passthrough writes its groups by hand, so it refuses the rewound group itself, the way the diff --git a/rs/moq-mux/src/container/ts/mod.rs b/rs/moq-mux/src/container/ts/mod.rs index 0de791d81b..7a329dc5b6 100644 --- a/rs/moq-mux/src/container/ts/mod.rs +++ b/rs/moq-mux/src/container/ts/mod.rs @@ -31,6 +31,7 @@ mod follower; mod health; mod import; mod mux_rate; +mod passthrough; mod programs; mod psi; mod release; @@ -50,7 +51,6 @@ pub use passthrough::Passthrough; pub use programs::Programs; pub use release::Release; -pub mod passthrough; pub mod stats; #[cfg(test)] diff --git a/rs/moq-mux/src/container/ts/passthrough.rs b/rs/moq-mux/src/container/ts/passthrough.rs index 94645ff3e0..7321bb84fe 100644 --- a/rs/moq-mux/src/container/ts/passthrough.rs +++ b/rs/moq-mux/src/container/ts/passthrough.rs @@ -1,54 +1,4 @@ //! Carrying a transport stream whole, without demultiplexing it. -//! -//! [`Passthrough`] publishes every 188-byte packet of its input, in order, on one track named by -//! the catalog's [`m2ts`](hang::catalog::M2ts) section. It reads the PAT and PMT to find the -//! program clock and the video stream, and reads nothing else past an adaptation field, so a -//! scrambled multiplex groups and paces exactly like its clear twin. -//! -//! # Objects and groups -//! -//! An object runs from one PCR packet on the clock PID to the next, and is also cut wherever a -//! group starts. A group starts at: -//! -//! - a packet with `random_access_indicator` on the video PID, or on the clock PID when the -//! program has no video; -//! - a PCR with `discontinuity_indicator` that jumps forward, since no group may hold a PCR -//! discontinuity; -//! - the first PCR at least [`FALLBACK`] after the group's start, for a source that never marks -//! random access; -//! - the first object that would take the group past [`GROUP_BYTES_MAX`] or -//! [`GROUP_OBJECTS_MAX`], so a long GOP cannot outgrow what a relay caches. Objects are never -//! split to meet the bound. -//! -//! Only the first of these makes a group a random access point, so any other start publishes -//! [`random_access`](hang::catalog::M2ts::random_access) false from then on, as does any group -//! carried while the layout allows no random access (more than one program or video stream). A -//! group the PCR discontinuity starts is behind a break even when it is also a random access -//! point. -//! -//! Bytes before the first group are dropped: until the PMT, a PCR, and a group start have all -//! been seen there is no group to put them in. The track is byte-identical to the input from its -//! first group on, and only from there. Bytes off the 188-byte packet grid are dropped too. -//! -//! # Timestamps -//! -//! An object's timestamp is the PCR time of its first byte, interpolated at the rate between the -//! PCRs either side of it, floored to microseconds. A PCR stamps the byte holding the last bit of -//! its base, ten bytes into its packet, so even an object that starts at a PCR packet sits ten -//! bytes of that rate before its own PCR. -//! -//! Every timestamp is then shifted onto the catalog's [`Clock`](crate::Clock) by one offset, fixed -//! through the reservation's [`Timebase`](crate::catalog::Timebase) by the first object: zero when -//! that object places the clock, and otherwise whatever lands it at now on a clock already in use. -//! -//! PCRs are unwrapped into a 64-bit count of 27 MHz ticks that never goes back. A step forward of -//! at most half the 2^33 * 300 tick period adds to the count, across the wrap included. A PCR with -//! `discontinuity_indicator` whose step is forward by that measure adds the step too, after the -//! bytes before it are timed at the rate before it, and starts a group behind a break marker. Any -//! step back ends the import: [`PcrRewindError`] without the flag, [`PcrRestartError`] with it, -//! since a signalled restart is new content and needs a new broadcast. The timeline therefore -//! depends only on the bytes since the importer's first PCR and its shift, and two importers of one -//! feed started at the same packet each placing its own catalog's clock publish identical objects. use std::collections::{BTreeMap, HashMap, VecDeque}; @@ -62,17 +12,17 @@ use super::psi::{self, Pmt}; /// The PCR clock, in Hz. pub(super) const PCR_HZ: u64 = 27_000_000; -/// The longest a group runs without a random access point before a PCR starts the next. -pub const FALLBACK: std::time::Duration = std::time::Duration::from_secs(1); +/// The longest a group runs without a random access point before a PCR starts the next, in +/// PCR ticks: one second. const FALLBACK_TICKS: u64 = PCR_HZ; /// The most bytes a group holds before its next object starts another: half of what `moq-net` /// caches of one group, so the frame headers and an object of slack never reach the cap. -pub const GROUP_BYTES_MAX: u64 = moq_net::group::MAX_CACHE_BYTES / 2; +pub(super) const GROUP_BYTES_MAX: u64 = moq_net::group::MAX_CACHE_BYTES / 2; /// The most objects a group holds before its next object starts another: half of `moq-net`'s /// frame cap, which a source stamping PCRs far more often than the 40 ms norm could reach. -pub const GROUP_OBJECTS_MAX: usize = moq_net::group::MAX_GROUP_FRAMES / 2; +pub(super) const GROUP_OBJECTS_MAX: usize = moq_net::group::MAX_GROUP_FRAMES / 2; /// The bytes a PCR sits into its packet: four of header, the adaptation field's length and flags, /// and the five bytes the 33-bit base ends in. @@ -104,68 +54,58 @@ const VIDEO_STREAM_TYPES: &[u8] = &[ 0xEA, // VC-1 ]; -/// The input is not a stream of 188-byte packets. +/// Publishes a transport stream verbatim, as one track of whole 188-byte packets. /// -/// A 192-byte source (BDAV M2TS, a four-byte timestamp before each packet) or a 204-byte one -/// (Reed-Solomon parity after each) is refused rather than reframed, since dropping those bytes -/// would break the byte identity passthrough exists for. -#[derive(Clone, Debug, PartialEq, Eq, thiserror::Error)] -#[non_exhaustive] -#[error("transport stream uses {size}-byte packets; passthrough carries 188-byte packets only")] -pub struct PacketSizeError { - /// The packet size the sync bytes recur at. - pub size: usize, -} - -/// The clock PID's PCR went backwards without `discontinuity_indicator`. +/// Every packet goes out in input order, on the track the catalog's [`m2ts`](hang::catalog::M2ts) +/// section names. It reads the PAT and PMT to find the program clock and the video stream, and +/// reads nothing else past an adaptation field, so a scrambled multiplex groups and paces exactly +/// like its clear twin. /// -/// The source broke its own timebase without saying so, which leaves no timestamp that both -/// follows the PCR and stays on one timeline. -#[derive(Clone, Debug, PartialEq, Eq, thiserror::Error)] -#[non_exhaustive] -#[error("PCR on PID {pid} went back from {from} to {to} without a discontinuity_indicator")] -pub struct PcrRewindError { - /// The clock PID. - pub pid: u16, - /// The previous PCR, raw 27 MHz ticks. - pub from: u64, - /// The PCR that stepped back, raw 27 MHz ticks. - pub to: u64, -} - -/// The clock PID's PCR went backwards with `discontinuity_indicator`: the source restarted. +/// # Objects and groups /// -/// What follows is new content, which belongs in a new broadcast rather than on a timeline that -/// steps back, so the import ends here. -#[derive(Clone, Debug, PartialEq, Eq, thiserror::Error)] -#[non_exhaustive] -#[error("PCR on PID {pid} restarted from {from} back to {to}; a restart needs a new broadcast")] -pub struct PcrRestartError { - /// The clock PID. - pub pid: u16, - /// The previous PCR, raw 27 MHz ticks. - pub from: u64, - /// The flagged PCR, raw 27 MHz ticks. - pub to: u64, -} - -/// The clock PID carried no PCR across more bytes than a group may hold. +/// An object runs from one PCR packet on the clock PID to the next, and is also cut wherever a +/// group starts. A group starts at: /// -/// Those bytes cannot be timed, and holding them for a PCR that may never come is unbounded. -#[derive(Clone, Debug, PartialEq, Eq, thiserror::Error)] -#[non_exhaustive] -#[error("no PCR on PID {pid} for {bytes} bytes")] -pub struct PcrGapError { - /// The clock PID. - pub pid: u16, - /// The bytes waiting on a PCR. - pub bytes: u64, -} - -/// Publishes a transport stream verbatim, as one track of whole 188-byte packets. +/// - a packet with `random_access_indicator` on the video PID, or on the clock PID when the +/// program has no video; +/// - a PCR with `discontinuity_indicator` that jumps forward, since no group may hold a PCR +/// discontinuity; +/// - the first PCR at least one second after the group's start, for a source that never marks +/// random access; +/// - the first object that would take the group past half of what `moq-net` caches of one group, +/// in bytes or in frames, so a long GOP cannot outgrow what a relay caches. Objects are never +/// split to meet the bound. +/// +/// Only the first of these makes a group a random access point, so any other start publishes +/// [`random_access`](hang::catalog::M2ts::random_access) false from then on, as does any group +/// carried while the layout allows no random access (more than one program or video stream). A +/// group the PCR discontinuity starts is behind a break even when it is also a random access +/// point. +/// +/// Bytes before the first group are dropped: until the PMT, a PCR, and a group start have all +/// been seen there is no group to put them in. The track is byte-identical to the input from its +/// first group on, and only from there. Bytes off the 188-byte packet grid are dropped too. +/// +/// # Timestamps /// -/// See the [module documentation](self) for where objects and groups start and how they are -/// timestamped. +/// An object's timestamp is the PCR time of its first byte, interpolated at the rate between the +/// PCRs either side of it, floored to microseconds. A PCR stamps the byte holding the last bit of +/// its base, ten bytes into its packet, so even an object that starts at a PCR packet sits ten +/// bytes of that rate before its own PCR. +/// +/// Every timestamp is then shifted onto the catalog's [`Clock`](crate::Clock) by one offset, fixed +/// through the reservation's [`Timebase`](crate::catalog::Timebase) by the first object: zero when +/// that object places the clock, and otherwise whatever lands it at now on a clock already in use. +/// +/// PCRs are unwrapped into a 64-bit count of 27 MHz ticks that never goes back. A step forward of +/// at most half the 2^33 * 300 tick period adds to the count, across the wrap included. A PCR with +/// `discontinuity_indicator` whose step is forward by that measure adds the step too, after the +/// bytes before it are timed at the rate before it, and starts a group behind a break marker. Any +/// step back ends the import: [`PcrRewind`](crate::Error::PcrRewind) without the flag, +/// [`PcrRestart`](crate::Error::PcrRestart) with it, since a signalled restart is new content and +/// needs a new broadcast. The timeline therefore depends only on the bytes since the importer's +/// first PCR and its shift, and two importers of one feed started at the same packet each placing +/// its own catalog's clock publish identical objects. pub struct Passthrough { catalog: crate::catalog::Producer, /// Held until the first group is written, so the first catalog already names the track with @@ -347,7 +287,7 @@ impl Passthrough { /// Append `data` and publish every whole packet it completes. A trailing partial packet is /// kept for the next call. - pub fn decode(&mut self, data: &[u8]) -> anyhow::Result<()> { + pub fn decode(&mut self, data: &[u8]) -> crate::Result<()> { self.scratch.extend_from_slice(data); if !self.sized { self.probe_packet_size(false)?; @@ -359,7 +299,7 @@ impl Passthrough { /// /// No later PCR will arrive, so the objects since the last one are timed at the rate of the /// interval before it. The catalog keeps the `m2ts` section until the importer drops. - pub fn finish(&mut self) -> anyhow::Result<()> { + pub fn finish(&mut self) -> crate::Result<()> { if !self.sized { self.probe_packet_size(true)?; self.packets()?; @@ -380,7 +320,7 @@ impl Passthrough { } /// Publish every whole packet in the scratch buffer, once the packet size is settled. - fn packets(&mut self) -> anyhow::Result<()> { + fn packets(&mut self) -> crate::Result<()> { if !self.sized { return Ok(()); } @@ -400,7 +340,7 @@ impl Passthrough { /// arrived, so where the grid is found does not depend on how the input was chunked. At the /// end of the input a shorter run settles the offsets left, so a stream too short for a full /// run still publishes. - fn probe_packet_size(&mut self, eof: bool) -> anyhow::Result<()> { + fn probe_packet_size(&mut self, eof: bool) -> crate::Result<()> { let scratch = &self.scratch; let syncs = |k: usize, size: usize, run: usize| (0..run).all(|i| scratch.get(k + i * size) == Some(&0x47)); let checkable = (scratch.len() + 1).saturating_sub(SYNC_WINDOW); @@ -431,7 +371,7 @@ impl Passthrough { self.scratch.drain(..k); self.sized = true; } - Some((_, size)) => return Err(PacketSizeError { size }.into()), + Some((_, size)) => return Err(crate::Error::TsPacketSize(size)), // No grid starts at any offset tried: leading junk. None => { self.scratch.drain(..checkable); @@ -440,7 +380,7 @@ impl Passthrough { Ok(()) } - fn packet(&mut self, pkt: &[u8; TsPacket::SIZE]) -> anyhow::Result<()> { + fn packet(&mut self, pkt: &[u8; TsPacket::SIZE]) -> crate::Result<()> { let at = self.offset; self.offset += TsPacket::SIZE as u64; let pid = u16::from_be_bytes([pkt[1] & 0x1f, pkt[2]]); @@ -492,18 +432,17 @@ impl Passthrough { let waiting = self.open.as_ref().map_or(0, |open| open.payload.len() as u64) + self.queue.iter().map(|object| object.payload.len() as u64).sum::(); if waiting > GROUP_BYTES_MAX { - return Err(PcrGapError { + return Err(crate::Error::PcrGap { pid: self.pcr_pid.unwrap_or_default(), bytes: waiting, - } - .into()); + }); } Ok(()) } /// A PCR at `at` on the clock PID: close the object it ends, time everything waiting on it, /// and say whether its packet starts a group. - fn clock(&mut self, pid: u16, raw: u64, at: u64) -> anyhow::Result> { + fn clock(&mut self, pid: u16, raw: u64, at: u64) -> crate::Result> { let at = at + PCR_OFFSET; self.close_object(); let flagged = std::mem::take(&mut self.pending_break); @@ -527,8 +466,8 @@ impl Passthrough { if step > PCR_WRAP / 2 { let from = timebase.raw; return Err(match flagged { - true => PcrRestartError { pid, from, to: raw }.into(), - false => PcrRewindError { pid, from, to: raw }.into(), + true => crate::Error::PcrRestart { pid, from, to: raw }, + false => crate::Error::PcrRewind { pid, from, to: raw }, }); } let last = timebase.samples.back().expect("a timebase has a sample").ticks; @@ -594,7 +533,7 @@ impl Passthrough { } /// Time and write every waiting object against the current timebase. - fn drain_queue(&mut self) -> anyhow::Result<()> { + fn drain_queue(&mut self) -> crate::Result<()> { while let Some(object) = self.queue.pop_front() { let timebase = self.timebase.as_ref().expect("an object follows a PCR"); let ticks = locate(object.at, &timebase.samples); @@ -603,7 +542,7 @@ impl Passthrough { Ok(()) } - fn write(&mut self, object: Object, ticks: u64) -> anyhow::Result<()> { + fn write(&mut self, object: Object, ticks: u64) -> crate::Result<()> { // Extrapolating past the last PCR before a small flagged jump can overshoot the jump. let ticks = ticks.max(self.written); self.written = ticks; @@ -650,7 +589,7 @@ impl Passthrough { } /// Write the section to the catalog if it changed. Nothing is written before the first group. - fn record(&mut self) -> anyhow::Result<()> { + fn record(&mut self) -> crate::Result<()> { if self.group.is_none() && self.section.is_none() { return Ok(()); } @@ -673,7 +612,7 @@ impl Passthrough { Ok(()) } - fn pat_packet(&mut self, pkt: &[u8; TsPacket::SIZE]) -> anyhow::Result<()> { + fn pat_packet(&mut self, pkt: &[u8; TsPacket::SIZE]) -> crate::Result<()> { let Some(pat) = self.pat.push(pkt, &mut self.crc_errors) else { return Ok(()); }; @@ -696,7 +635,7 @@ impl Passthrough { self.remap() } - fn pmt_packet(&mut self, pid: Pid, pkt: &[u8; TsPacket::SIZE]) -> anyhow::Result<()> { + fn pmt_packet(&mut self, pid: Pid, pkt: &[u8; TsPacket::SIZE]) -> crate::Result<()> { let mut sections = Vec::new(); self.pmt_sections.entry(pid).or_default().push(pkt, &mut sections); let mut changed = false; @@ -737,7 +676,7 @@ impl Passthrough { /// The PAT or a PMT changed: find the clock, the PID that marks random access, and whether /// the layout allows random access at all. - fn remap(&mut self) -> anyhow::Result<()> { + fn remap(&mut self) -> crate::Result<()> { let clock = match self.pcr_override { Some(pid) => self .programs diff --git a/rs/moq-mux/src/container/ts/passthrough_test.rs b/rs/moq-mux/src/container/ts/passthrough_test.rs index bade422972..fd4031cbb6 100644 --- a/rs/moq-mux/src/container/ts/passthrough_test.rs +++ b/rs/moq-mux/src/container/ts/passthrough_test.rs @@ -10,9 +10,7 @@ use std::time::Duration; use bytes::Bytes; use super::fixture::{ACCESS, Flags, MS, Mux, WRAP, paced, program}; -use super::passthrough::{ - GROUP_BYTES_MAX, GROUP_OBJECTS_MAX, PacketSizeError, Passthrough, PcrGapError, PcrRestartError, PcrRewindError, -}; +use super::passthrough::{GROUP_BYTES_MAX, GROUP_OBJECTS_MAX, Passthrough}; /// A delay budget no test timeline comes close to, so the reader sees every group. const RECORDING_MAX_DELAY: Duration = Duration::from_secs(30); @@ -634,8 +632,11 @@ async fn a_flagged_step_back_is_a_restart() { Ok(_) => panic!("a restart was carried on one timeline"), Err(err) => err, }; - let restart = err.downcast_ref::().expect("a typed refusal"); - assert_eq!((restart.pid, restart.to), (0x100, 2_000 * MS)); + let restart = err.downcast_ref::(); + assert!( + matches!(restart, Some(crate::Error::PcrRestart { pid: 0x100, to, .. }) if *to == 2_000 * MS), + "{err:?}" + ); } /// A break before any group has started is no group start, so the first group still waits for a @@ -665,8 +666,11 @@ async fn an_unflagged_rewind_is_refused() { Ok(_) => panic!("a rewind was accepted"), Err(err) => err, }; - let rewind = err.downcast_ref::().expect("a typed refusal"); - assert_eq!((rewind.pid, rewind.to), (0x100, 9_000 * MS)); + let rewind = err.downcast_ref::(); + assert!( + matches!(rewind, Some(crate::Error::PcrRewind { pid: 0x100, to, .. }) if *to == 9_000 * MS), + "{err:?}" + ); } #[tokio::test(start_paused = true)] @@ -744,7 +748,13 @@ async fn a_clock_that_stops_is_refused_once_a_group_would_overflow() { Ok(_) => panic!("an untimed run was accepted"), Err(err) => err, }; - assert_eq!(err.downcast_ref::().map(|gap| gap.pid), Some(0x100)); + assert!( + matches!( + err.downcast_ref::(), + Some(crate::Error::PcrGap { pid: 0x100, .. }) + ), + "{err:?}" + ); } #[tokio::test(start_paused = true)] @@ -781,7 +791,10 @@ async fn a_192_or_204_byte_source_is_refused() { Ok(_) => panic!("{size}-byte packets were accepted"), Err(err) => err, }; - assert_eq!(err.downcast_ref::(), Some(&PacketSizeError { size })); + assert!( + matches!(err.downcast_ref::(), Some(crate::Error::TsPacketSize(got)) if *got == size), + "{err:?}" + ); } } diff --git a/rs/moq-mux/src/error.rs b/rs/moq-mux/src/error.rs index 60c42e7e49..d1716f8daf 100644 --- a/rs/moq-mux/src/error.rs +++ b/rs/moq-mux/src/error.rs @@ -216,7 +216,7 @@ pub enum Error { #[error("timeline record {0} is untimed")] TimelineUntimed(u64), - /// Error from a muxer/demuxer that reports via `anyhow` (currently MPEG-TS). + /// Error from a muxer/demuxer that reports via `anyhow` (currently the MPEG-TS demuxer). /// Boxed in an `Arc` so the enum stays `Clone` (`anyhow::Error` is not). #[error("{0}")] Other(std::sync::Arc), @@ -277,6 +277,53 @@ pub enum Error { #[error("capture time is outside the broadcast clock")] InvalidCapture, + /// A transport stream passthrough was fed packets other than 188 bytes. + /// + /// A 192-byte source (BDAV M2TS, a four-byte timestamp before each packet) or a 204-byte one + /// (Reed-Solomon parity after each) is refused rather than reframed, since dropping those bytes + /// would break the byte identity passthrough exists for. + #[error("transport stream uses {0}-byte packets; passthrough carries 188-byte packets only")] + TsPacketSize(usize), + + /// The clock PID's PCR went backwards without `discontinuity_indicator`. + /// + /// The source broke its own timebase without saying so, which leaves no timestamp that both + /// follows the PCR and stays on one timeline. + #[error("PCR on PID {pid} went back from {from} to {to} without a discontinuity_indicator")] + PcrRewind { + /// The clock PID. + pid: u16, + /// The previous PCR, raw 27 MHz ticks. + from: u64, + /// The PCR that stepped back, raw 27 MHz ticks. + to: u64, + }, + + /// The clock PID's PCR went backwards with `discontinuity_indicator`: the source restarted. + /// + /// What follows is new content, which belongs in a new broadcast rather than on a timeline that + /// steps back, so the import ends here. + #[error("PCR on PID {pid} restarted from {from} back to {to}; a restart needs a new broadcast")] + PcrRestart { + /// The clock PID. + pid: u16, + /// The previous PCR, raw 27 MHz ticks. + from: u64, + /// The flagged PCR, raw 27 MHz ticks. + to: u64, + }, + + /// The clock PID carried no PCR across more bytes than a group may hold. + /// + /// Those bytes cannot be timed, and holding them for a PCR that may never come is unbounded. + #[error("no PCR on PID {pid} for {bytes} bytes")] + PcrGap { + /// The clock PID. + pid: u16, + /// The bytes waiting on a PCR. + bytes: u64, + }, + /// Another publisher instance replaced a followed broadcast, and stitching was off. #[error("another publisher instance replaced broadcast `{0}`; enable stitch to follow it as a program switch")] Replaced(String), diff --git a/rs/moq-mux/src/import/container.rs b/rs/moq-mux/src/import/container.rs index 9b0ae4295d..e756f3ecfe 100644 --- a/rs/moq-mux/src/import/container.rs +++ b/rs/moq-mux/src/import/container.rs @@ -1,16 +1,13 @@ -//! Container importers. +//! The container importer. //! -//! [`Container`] decodes a container from whole chunks; [`ContainerStream`] -//! decodes it from a raw byte stream. A container may publish more than one MoQ -//! track, so neither exposes a single-track demand/name handle. Today every -//! container supports both; both wrap the same [`ContainerImpl`] dispatch. +//! [`Container`] decodes a container from chunks split anywhere: every container +//! recovers its own framing. A container may publish more than one MoQ track, so +//! it exposes no single-track demand/name handle. use super::{ContainerFormat, ContainerInit}; use crate::Result; -/// The concrete container importers, shared by [`Container`] and -/// [`ContainerStream`]. Containers parse their own internal framing, so a whole -/// chunk and a stream chunk decode identically. +/// The concrete container importers [`Container`] dispatches to. enum ContainerImpl { // Boxed because it's a large struct and clippy complains about the size. Fmp4(Box>), @@ -26,17 +23,15 @@ impl ContainerImpl { format: ContainerFormat, ) -> Self { match format { - ContainerFormat::Fmp4 => Self::fmp4(broadcast, reserved, false), - ContainerFormat::Cmaf => Self::fmp4(broadcast, reserved, true), + ContainerFormat::Fmp4 => Self::fmp4(broadcast, reserved), ContainerFormat::Mkv => Self::mkv(broadcast, reserved), ContainerFormat::Ts => Self::ts(broadcast, reserved), ContainerFormat::Flv => Self::flv(broadcast, reserved), } } - fn fmp4(broadcast: moq_net::broadcast::Producer, reserved: crate::catalog::Reserved, passthrough: bool) -> Self { - let import = crate::container::fmp4::Import::new(broadcast, reserved).with_passthrough(passthrough); - ContainerImpl::Fmp4(Box::new(import)) + fn fmp4(broadcast: moq_net::broadcast::Producer, reserved: crate::catalog::Reserved) -> Self { + ContainerImpl::Fmp4(Box::new(crate::container::fmp4::Import::new(broadcast, reserved))) } fn mkv(broadcast: moq_net::broadcast::Producer, reserved: crate::catalog::Reserved) -> Self { @@ -78,11 +73,12 @@ impl ContainerImpl { } } - fn cut(&mut self) { + fn cut(&mut self) -> Result<()> { // Only fMP4 has a segment concept to declare. The others recover their own framing and // group on what they find in it, so there is nothing for a caller to draw. - if let ContainerImpl::Fmp4(decoder) = self { - decoder.cut(); + match self { + ContainerImpl::Fmp4(decoder) => decoder.cut(), + _ => Ok(()), } } @@ -96,10 +92,7 @@ impl ContainerImpl { } } -/// A container importer for whole chunks. -/// -/// Use this when the caller hands over discrete buffers (the typical case for -/// files and reassembled network input). May publish more than one track. +/// A container importer, fed chunks split anywhere. May publish more than one track. pub struct Container { inner: ContainerImpl, } @@ -107,6 +100,7 @@ pub struct Container { impl Container { /// Create a new container importer, decoding [`ContainerInit::data`] as the initial chunk. /// + /// The data may be empty, leaving every byte to [`Self::decode`]. pub fn new( broadcast: moq_net::broadcast::Producer, reserved: crate::catalog::Reserved, @@ -138,64 +132,7 @@ impl Container { /// For a caller that knows the source's segmentation out of band (an HLS import following its /// playlist). An fMP4 source that carries `styp` atoms declares its own, so this is only /// needed when it doesn't. Ignored by container formats with no segment concept (MKV, TS, FLV). - pub fn cut(&mut self) { - self.inner.cut() - } - - /// Close the current group and open the next one at `sequence`. - pub fn seek(&mut self, sequence: u64) -> Result<()> { - self.inner.seek(sequence) - } -} - -/// A container importer for a raw byte stream. -/// -/// Use this when the caller pushes arbitrary byte chunks and the container -/// recovers its own framing. May publish more than one track. -pub struct ContainerStream { - inner: ContainerImpl, -} - -impl ContainerStream { - /// Create a new container stream importer. - /// - /// Takes a bare format rather than a [`ContainerInit`]: a stream recovers its own framing, so - /// there are no leading bytes to seed it with. Push everything through [`Self::decode`]. - pub fn new( - broadcast: moq_net::broadcast::Producer, - reserved: crate::catalog::Reserved, - format: ContainerFormat, - ) -> Result { - // A separate list from [`Container::new`]: only containers that can be - // recovered from a raw byte stream belong here. Today that's all of them, - // but a non-streamable container (e.g. RTP) would be added to `Container` - // alone. - let inner = ContainerImpl::new(broadcast, reserved, format); - Ok(Self { inner }) - } - - /// Decode a chunk of the byte stream. - pub fn decode(&mut self, data: &[u8]) -> Result<()> { - self.inner.decode(data) - } - - /// Finish the importer, flushing any buffered data. - pub fn finish(&mut self) -> Result<()> { - self.inner.finish() - } - - /// Abort every published track with `err`, so subscribers see the real cause - /// rather than [`moq_net::Error::Dropped`]. Consumes the importer. - pub fn abort(self, err: moq_net::Error) { - self.inner.abort(err) - } - - /// Declare that the next fragment starts a new segment, rolling a group on every track. - /// - /// For a caller that knows the source's segmentation out of band (an HLS import following its - /// playlist). An fMP4 source that carries `styp` atoms declares its own, so this is only - /// needed when it doesn't. Ignored by container formats with no segment concept (MKV, TS, FLV). - pub fn cut(&mut self) { + pub fn cut(&mut self) -> Result<()> { self.inner.cut() } @@ -209,9 +146,9 @@ impl ContainerStream { mod tests { use super::*; - /// A stream recovers its own framing, which is why [`ContainerStream::new`] takes a bare format - /// with no leading bytes to seed it. Prove it: hand the whole file to `decode` in two chunks - /// split mid-header, and the tracks still land. + /// A container recovers its own framing, so it needs no leading bytes to seed it. Prove it: + /// start empty, hand the whole file to `decode` in two chunks split mid-header, and the tracks + /// still land. #[test] fn a_split_stream_publishes_its_tracks() { let data = include_bytes!("../container/fmp4/test_data/bbb.mp4"); @@ -220,8 +157,8 @@ mod tests { let mut broadcast = moq_net::broadcast::Info::new().produce(); let catalog = crate::catalog::Producer::new(&mut broadcast, crate::catalog::Config::default()).unwrap(); - let mut stream: ContainerStream = - ContainerStream::new(broadcast, catalog.reserve(), ContainerFormat::Fmp4).unwrap(); + let init = ContainerInit::new(ContainerFormat::Fmp4, bytes::Bytes::new()); + let mut stream: Container = Container::new(broadcast, catalog.reserve(), &init).unwrap(); stream.decode(head).unwrap(); // The test file ends on a malformed fragment, so a trailing decode error is expected. diff --git a/rs/moq-mux/src/import/format.rs b/rs/moq-mux/src/import/format.rs index 1f583ed021..90d12529fc 100644 --- a/rs/moq-mux/src/import/format.rs +++ b/rs/moq-mux/src/import/format.rs @@ -48,10 +48,15 @@ pub enum VideoFormat { } /// A container an importer can demux, which may publish more than one track. +/// +/// Names only the input the importer parses. Every container is demuxed into its tracks' frames; +/// republishing the input whole is a choice made on the concrete importer +/// ([`fmp4::Import::with_passthrough`](crate::container::fmp4::Import::with_passthrough), +/// [`ts::Passthrough`](crate::container::ts::Passthrough)), never by the format name. #[derive(Clone, Copy, Debug, PartialEq, Eq, Hash)] #[non_exhaustive] pub enum ContainerFormat { - /// Fragmented MP4 / CMAF, unwrapped into one frame per sample. + /// Fragmented MP4 / CMAF. Fmp4, /// Matroska / WebM. Mkv, @@ -59,8 +64,6 @@ pub enum ContainerFormat { Ts, /// Flash Video, as used by RTMP. Flv, - /// Fragmented MP4 / CMAF, republished whole as one CMAF frame per fragment. - Cmaf, } /// Any format an importer handles, and the kind of import it selects. @@ -131,10 +134,11 @@ impl FromStr for Format { "vp8" | "vp08" => Format::Video(VideoFormat::Vp8), "vp9" | "vp09" => Format::Video(VideoFormat::Vp9), - "fmp4" => Format::Container(ContainerFormat::Fmp4), - "cmaf" => Format::Container(ContainerFormat::Cmaf), + "fmp4" | "cmaf" => Format::Container(ContainerFormat::Fmp4), "mkv" | "webm" | "matroska" => Format::Container(ContainerFormat::Mkv), - "ts" | "mpegts" | "mpeg2ts" | "m2ts" => Format::Container(ContainerFormat::Ts), + // Not `m2ts`: that names the catalog section carrying a multiplex whole, and BDAV's + // 192-byte packets besides. + "ts" | "mpegts" | "mpeg2ts" => Format::Container(ContainerFormat::Ts), "flv" => Format::Container(ContainerFormat::Flv), _ => return Err(crate::Error::UnknownFormat(s.to_string())), @@ -202,7 +206,7 @@ kind_format!(VideoFormat, Video, Av01 => "av01", Vp8 => "vp8", Vp9 => "vp9"); kind_format!(ContainerFormat, Container, - Fmp4 => "fmp4", Cmaf => "cmaf", Mkv => "mkv", Ts => "ts", Flv => "flv"); + Fmp4 => "fmp4", Mkv => "mkv", Ts => "ts", Flv => "flv"); #[cfg(test)] mod tests { @@ -219,10 +223,9 @@ mod tests { (vec!["av01", "av1", "av1c", "av1C"], "av01"), (vec!["vp8", "vp08"], "vp8"), (vec!["vp9", "vp09"], "vp9"), - (vec!["fmp4"], "fmp4"), - (vec!["cmaf"], "cmaf"), + (vec!["fmp4", "cmaf"], "fmp4"), (vec!["mkv", "webm", "matroska"], "mkv"), - (vec!["ts", "mpegts", "mpeg2ts", "m2ts"], "ts"), + (vec!["ts", "mpegts", "mpeg2ts"], "ts"), (vec!["opus"], "opus"), ] { for alias in aliases { diff --git a/rs/moq-mux/src/import/mod.rs b/rs/moq-mux/src/import/mod.rs index 38abd35c83..d137b03e98 100644 --- a/rs/moq-mux/src/import/mod.rs +++ b/rs/moq-mux/src/import/mod.rs @@ -1,17 +1,17 @@ //! Import media into a moq broadcast. //! -//! The importers split along two axes. By multiplicity: [`Track`] / [`TrackStream`] -//! publish a single codec onto one MoQ track, while [`Container`] / -//! [`ContainerStream`] decode a container that may publish more than one track. By -//! frame boundaries: [`Track`] / [`Container`] take whole frames or chunks (the -//! typical case for files and reassembled network input), while [`TrackStream`] / -//! [`ContainerStream`] infer boundaries from a raw byte stream (piped Annex-B -//! H.264, an fMP4 reader, …). +//! The importers split by multiplicity: [`Track`] / [`TrackStream`] publish a +//! single codec onto one MoQ track, while [`Container`] decodes a container +//! that may publish more than one track. A single codec splits again by frame +//! boundaries: [`Track`] takes whole frames (the typical case for files and +//! reassembled network input), while [`TrackStream`] infers them from a raw +//! byte stream (piped Annex-B H.264, ...). A container recovers its own +//! framing, so [`Container`] takes chunks split anywhere. //! -//! Each importer's `new` takes a format string (e.g. `"avc3"`, `"fmp4"`) and -//! errors on a format it doesn't handle — `TrackStream` / `ContainerStream` -//! accept only the self-delimiting formats. The concrete importers live with -//! their format under [`crate::container`] or [`crate::codec`] and own their +//! Each importer takes a typed format (parsed from e.g. `"avc3"` or `"fmp4"`) +//! and errors on a format it doesn't handle; `TrackStream` accepts only the +//! self-delimiting formats. The concrete importers live with their format +//! under [`crate::container`] or [`crate::codec`] and own their //! catalog rendition through their container producer. //! //! A single-codec importer takes a track the caller mints, typically @@ -205,7 +205,15 @@ mod tests { data.resize(data.len() + PAD, 0); assert_released("fmp4", &container(ContainerFormat::Fmp4, &data, CHUNK)); - assert_released("cmaf", &container(ContainerFormat::Cmaf, &data, CHUNK)); + assert_released("fmp4 passthrough", &|broadcast, catalog| { + let mut import = + crate::container::fmp4::Import::new(broadcast.clone(), catalog.reserve()).with_passthrough(true); + for chunk in data.chunks(CHUNK) { + import.decode(chunk).unwrap(); + } + import.finish().unwrap(); + Box::new(import) + }); } #[tokio::test(start_paused = true)]