diff --git a/src/io/sink/mod.rs b/src/io/sink/mod.rs index 6cf3394..5284278 100644 --- a/src/io/sink/mod.rs +++ b/src/io/sink/mod.rs @@ -28,8 +28,10 @@ use std::io::{Seek, Write}; mod local_file; mod preallocate; +mod socket; pub use local_file::LocalFileSink; +pub use socket::{SocketSink, UdpSocketSink}; /// Sequential-only write destination. Sockets, pipes, append-only /// stores. No seek. Implementations own their write buffering — the diff --git a/src/io/sink/socket.rs b/src/io/sink/socket.rs new file mode 100644 index 0000000..411ac25 --- /dev/null +++ b/src/io/sink/socket.rs @@ -0,0 +1,278 @@ +//! TCP / UDP socket sinks (sequential-only). +//! +//! [`SocketSink`] wraps a `TcpStream` in a 1 MiB `BufWriter`. Constructor +//! tunes `SO_SNDBUF` to a caller hint when provided. `finish()` flushes +//! the buffer then `shutdown(Write)`s the socket so the peer sees clean +//! end-of-stream. +//! +//! [`UdpSocketSink`] wraps a connected `UdpSocket`. Each `write` call +//! emits exactly one datagram — the caller is responsible for packetizing +//! to a reasonable MTU (188 × 7 = 1316 bytes for MPEG-TS-over-UDP is the +//! conventional choice). `finish()` is a no-op; UDP has no end-of-stream +//! marker. +//! +//! Both types satisfy [`SequentialSink`] via the blanket impl in +//! `super::mod`. Neither implements `Seek`, so neither satisfies +//! [`RandomAccessSink`] — using one with `MkvMux` is a compile error, +//! which is the design intent. +//! +//! [`SequentialSink`]: super::SequentialSink +//! [`RandomAccessSink`]: super::RandomAccessSink + +use std::io::{self, BufWriter, Write}; +use std::net::{Shutdown, TcpStream, ToSocketAddrs, UdpSocket}; + +/// `BufWriter` capacity for [`SocketSink`]. 1 MiB matches the typical +/// kernel send-buffer ceiling and keeps small-write amplification from +/// containers (TS = 188-byte packets, fMP4 fragment headers = ~100 bytes) +/// from translating into syscall storms. +const TCP_BUF_CAPACITY: usize = 1024 * 1024; + +/// Sequential-only sink over a TCP connection. +/// +/// Wraps a `BufWriter`; the inner `TcpStream` is kept as a +/// clone so [`finish`](Self::finish) can call `shutdown(Write)` after +/// flushing the buffer (the buffered writer doesn't expose the socket +/// directly). +pub struct SocketSink { + /// Buffered write half. All payload bytes go through this. + buf: BufWriter, + /// Shutdown handle — clone of the socket inside `buf`. Used only by + /// `finish()` for `shutdown(Write)`; never read or written through. + shutdown_handle: TcpStream, +} + +impl SocketSink { + /// Open a TCP connection to `addr` and wrap it for sequential + /// writing. `sndbuf_bytes`, when present, is forwarded to + /// `setsockopt(SO_SNDBUF)` as a kernel hint — the OS may clamp it. + /// + /// `addr` accepts anything `ToSocketAddrs` does: `"10.0.0.1:1234"`, + /// `("host", 1234)`, a `SocketAddr`, etc. + pub fn connect(addr: A, sndbuf_bytes: Option) -> io::Result { + let stream = TcpStream::connect(addr)?; + // `set_nodelay(true)` keeps small writes (TS packet trains, fMP4 + // moof headers) from sitting in Nagle's algorithm until the buffer + // fills. The BufWriter already absorbs syscall overhead; Nagle + // would just add latency without coalescing more. + stream.set_nodelay(true)?; + if let Some(n) = sndbuf_bytes { + set_send_buffer(&stream, n)?; + } + let shutdown_handle = stream.try_clone()?; + Ok(Self { + buf: BufWriter::with_capacity(TCP_BUF_CAPACITY, stream), + shutdown_handle, + }) + } +} + +impl Write for SocketSink { + fn write(&mut self, buf: &[u8]) -> io::Result { + self.buf.write(buf) + } + fn flush(&mut self) -> io::Result<()> { + self.buf.flush() + } +} + +impl SocketSink { + /// Drain the BufWriter and `shutdown(Write)` the underlying socket + /// so the peer sees a clean EOF. + /// + /// Note: [`SequentialSink::finish`](super::SequentialSink::finish)'s + /// blanket-impl default is a no-op. Trait-object call sites that + /// need socket shutdown should call this inherent method directly + /// before dropping the sink, or hold the concrete `SocketSink` type + /// (typical pattern: each muxer's `finish()` calls the appropriate + /// inherent close method on its captured concrete sink). + pub fn finish(&mut self) -> io::Result<()> { + self.buf.flush()?; + // `shutdown(Write)` signals clean EOF to the peer. Errors here + // are non-fatal — the connection may have already been torn down + // by the peer — but we surface them so callers can log. + self.shutdown_handle.shutdown(Shutdown::Write) + } +} + +/// Sequential-only sink over a connected UDP socket. +/// +/// Each [`write`](Write::write) call sends exactly one datagram. The +/// caller is responsible for splitting payload at packet boundaries — +/// for MPEG-TS this means 7 × 188 = 1316 bytes per datagram, the +/// industry standard for MPEG-TS-over-UDP. No buffering happens here; +/// adding it would silently merge datagrams. +/// +/// `finish()` is a no-op: UDP has no end-of-stream marker. Closing the +/// socket happens on drop. +pub struct UdpSocketSink { + socket: UdpSocket, +} + +impl UdpSocketSink { + /// Bind a local UDP socket to an ephemeral port and `connect` it to + /// `peer`. `connect` doesn't open a connection — it just fixes the + /// peer address so subsequent `send` calls don't need to repeat it, + /// and so receive-side filtering rejects packets from other sources. + /// + /// `sndbuf_bytes`, when present, is a hint to `SO_SNDBUF`. + pub fn connect(peer: A, sndbuf_bytes: Option) -> io::Result { + // Bind to all-zeros / any port. The kernel picks an ephemeral + // source port and the source IP at first send. + let socket = UdpSocket::bind("0.0.0.0:0")?; + socket.connect(peer)?; + if let Some(n) = sndbuf_bytes { + set_udp_send_buffer(&socket, n)?; + } + Ok(Self { socket }) + } +} + +impl Write for UdpSocketSink { + fn write(&mut self, buf: &[u8]) -> io::Result { + // `send` writes the entire datagram or fails — no partial sends + // for UDP. Match `Write::write`'s contract by reporting bytes + // accepted. + self.socket.send(buf) + } + fn flush(&mut self) -> io::Result<()> { + Ok(()) + } +} + +impl UdpSocketSink { + /// No-op — UDP has no end-of-stream marker. Provided for parity + /// with [`SocketSink::finish`] so call sites can treat them uniformly. + pub fn finish(&mut self) -> io::Result<()> { + Ok(()) + } +} + +// ── Platform `SO_SNDBUF` tuning ──────────────────────────────────────────── +// +// std's `TcpStream` / `UdpSocket` don't expose `SO_SNDBUF`. We drop to +// libc on Linux + macOS (the libc-dep targets in Cargo.toml). On other +// targets the hint is silently ignored — the socket still works, the +// kernel just picks its own send-buffer size. + +#[cfg(any(target_os = "linux", target_os = "macos"))] +fn set_send_buffer(stream: &TcpStream, bytes: usize) -> io::Result<()> { + use std::os::unix::io::AsRawFd; + setsockopt_sndbuf(stream.as_raw_fd(), bytes) +} + +#[cfg(any(target_os = "linux", target_os = "macos"))] +fn set_udp_send_buffer(socket: &UdpSocket, bytes: usize) -> io::Result<()> { + use std::os::unix::io::AsRawFd; + setsockopt_sndbuf(socket.as_raw_fd(), bytes) +} + +#[cfg(not(any(target_os = "linux", target_os = "macos")))] +fn set_send_buffer(_stream: &TcpStream, _bytes: usize) -> io::Result<()> { + // Non-Linux-non-macOS targets aren't in Cargo.toml's libc dep list; + // silently ignore the hint rather than failing the connect. Callers + // can detect via the lack of an explicit "sndbuf applied" signal + // (not provided, intentionally — this is a hint, not a guarantee). + Ok(()) +} + +#[cfg(not(any(target_os = "linux", target_os = "macos")))] +fn set_udp_send_buffer(_socket: &UdpSocket, _bytes: usize) -> io::Result<()> { + Ok(()) +} + +#[cfg(any(target_os = "linux", target_os = "macos"))] +fn setsockopt_sndbuf(fd: std::os::unix::io::RawFd, bytes: usize) -> io::Result<()> { + // Clamp into c_int range; SO_SNDBUF takes an `int` argument. + let want: libc::c_int = bytes.try_into().unwrap_or(libc::c_int::MAX); + let ret = unsafe { + libc::setsockopt( + fd, + libc::SOL_SOCKET, + libc::SO_SNDBUF, + &want as *const _ as *const libc::c_void, + std::mem::size_of::() as libc::socklen_t, + ) + }; + if ret != 0 { + return Err(io::Error::last_os_error()); + } + Ok(()) +} + +#[cfg(test)] +mod tests { + use super::*; + use std::io::Read; + use std::net::{TcpListener, UdpSocket}; + use std::thread; + + /// Bind a listener, accept on a thread, return (listener_addr, + /// accepted-bytes future via JoinHandle). + #[test] + fn socket_sink_round_trips_bytes() { + let listener = TcpListener::bind("127.0.0.1:0").unwrap(); + let addr = listener.local_addr().unwrap(); + let accept = thread::spawn(move || { + let (mut sock, _) = listener.accept().unwrap(); + let mut buf = Vec::new(); + sock.read_to_end(&mut buf).unwrap(); + buf + }); + + let mut sink = SocketSink::connect(addr, Some(256 * 1024)).unwrap(); + // Write enough to overflow the BufWriter at least once, then a + // tail that lives in the buffer until `finish` flushes. + let big: Vec = (0..(2 * TCP_BUF_CAPACITY)) + .map(|i| (i & 0xff) as u8) + .collect(); + sink.write_all(&big).unwrap(); + sink.write_all(b"tail\n").unwrap(); + sink.finish().unwrap(); + drop(sink); + + let received = accept.join().unwrap(); + assert_eq!(received.len(), big.len() + 5); + assert_eq!(&received[..big.len()], &big[..]); + assert_eq!(&received[big.len()..], b"tail\n"); + } + + #[test] + fn socket_sink_is_sequential_only() { + // Compile-time assertion via dyn — if this ever started + // satisfying `RandomAccessSink`, the trait split would be broken. + fn _assert_seq(_: &mut dyn super::super::SequentialSink) {} + let listener = TcpListener::bind("127.0.0.1:0").unwrap(); + let addr = listener.local_addr().unwrap(); + let _accept = thread::spawn(move || { + let _ = listener.accept(); + }); + let mut sink = SocketSink::connect(addr, None).unwrap(); + _assert_seq(&mut sink); + // The negative is harder to assert directly (no `is_not`), + // but `SocketSink` does not impl `Seek`, so it can't unify with + // `RandomAccessSink`'s super-bound. The Phase 2 blanket impl + // `impl RandomAccessSink for T {}` thus + // excludes it by construction. + } + + #[test] + fn udp_socket_sink_delivers_datagrams() { + let receiver = UdpSocket::bind("127.0.0.1:0").unwrap(); + receiver + .set_read_timeout(Some(std::time::Duration::from_secs(2))) + .unwrap(); + let addr = receiver.local_addr().unwrap(); + let mut sink = UdpSocketSink::connect(addr, Some(128 * 1024)).unwrap(); + + sink.write_all(&[1, 2, 3, 4, 5]).unwrap(); + sink.write_all(&[9, 9, 9]).unwrap(); + sink.finish().unwrap(); + + let mut buf = [0u8; 64]; + let n1 = receiver.recv(&mut buf).unwrap(); + assert_eq!(&buf[..n1], &[1, 2, 3, 4, 5]); + let n2 = receiver.recv(&mut buf).unwrap(); + assert_eq!(&buf[..n2], &[9, 9, 9]); + } +} diff --git a/src/mux/fmp4/mod.rs b/src/mux/fmp4/mod.rs new file mode 100644 index 0000000..a6f398a --- /dev/null +++ b/src/mux/fmp4/mod.rs @@ -0,0 +1,403 @@ +//! Fragmented MP4 muxer — **stub** for Phase 3. +//! +//! Goal: ISO/IEC 14496-12 fragmented MP4 (`ftyp` + `moov` init segment, +//! then a sequence of `moof+mdat` media fragments) targeting a +//! [`SequentialSink`](crate::io::sink::SequentialSink). DASH-friendly, +//! no Cues backpatch. +//! +//! Status (v0.21.0 Phase 3): **STUB**. We ship the init segment +//! (`ftyp` + a minimal HEVC `moov` skeleton with one video track) so +//! the muxer's shape and call site are validated, but media fragments +//! are NOT yet emitted — calls to [`Fmp4Mux::write_video`] currently +//! accumulate frames into an internal buffer and discard them on +//! [`Fmp4Mux::finish`]. +//! +//! ## What's TODO (tracked in Phase 4 / v0.22.0 scope) +//! +//! - `moof` box: `mfhd` (sequence_number) + `traf` (`tfhd` + `tfdt` +//! + `trun` with sample sizes, durations, flags, composition offsets). +//! - `mdat` box: concatenated sample data. +//! - Fragment cadence: one fragment per GOP or every N seconds, +//! whichever comes first. +//! - HEVC `hvcC` box inside `moov.trak.mdia.minf.stbl.stsd` so the +//! init segment is self-describing. +//! - Sample-flags computation (sync vs. delta, depends_on, etc.). +//! - Edit lists / fragment_duration for accurate seeking. +//! +//! Reference: ISO/IEC 14496-12 §8 (Movie Fragments). + +use std::io::{self, Write}; + +// Box type literals — four-character codes per ISO/IEC 14496-12 §4.2. + +const FTYP: [u8; 4] = *b"ftyp"; +const MOOV: [u8; 4] = *b"moov"; +const MVHD: [u8; 4] = *b"mvhd"; +const TRAK: [u8; 4] = *b"trak"; +const TKHD: [u8; 4] = *b"tkhd"; +const MDIA: [u8; 4] = *b"mdia"; +const MDHD: [u8; 4] = *b"mdhd"; +const HDLR: [u8; 4] = *b"hdlr"; +const MINF: [u8; 4] = *b"minf"; +const VMHD: [u8; 4] = *b"vmhd"; +const DINF: [u8; 4] = *b"dinf"; +const DREF: [u8; 4] = *b"dref"; +const URL_: [u8; 4] = *b"url "; +const STBL: [u8; 4] = *b"stbl"; +const STSD: [u8; 4] = *b"stsd"; +const STTS: [u8; 4] = *b"stts"; +const STSC: [u8; 4] = *b"stsc"; +const STSZ: [u8; 4] = *b"stsz"; +const STCO: [u8; 4] = *b"stco"; +const MVEX: [u8; 4] = *b"mvex"; +const TREX: [u8; 4] = *b"trex"; + +/// Default movie timescale — 90 kHz lines up with MPEG-TS PTS and the +/// HEVC SPS `vui_time_scale` for film content, simplifying the math +/// when fragment emission lands. +const MOVIE_TIMESCALE: u32 = 90_000; +/// Video track ID. fMP4 init segments conventionally use track_ID=1 +/// for the primary video track; a single-track DASH representation has +/// no reason to deviate. +const VIDEO_TRACK_ID: u32 = 1; + +/// Fragmented MP4 muxer — stub. +/// +/// See the module-level doc comment for what is and isn't shipped in +/// this stub. +pub struct Fmp4Mux { + writer: W, + header_written: bool, + /// Pending frames — held for the future fragment-emit path. The + /// stub drops these on `finish` but keeping them around lets the + /// post-stub work re-attach without changing the public API. + pending: Vec, + /// hvcC bytes, if provided. Embedded in the `moov.…stsd.hvc1.hvcC` + /// box once that path lands. + #[allow(dead_code)] + codec_private: Option>, +} + +struct PendingSample { + #[allow(dead_code)] + pts_ns: i64, + #[allow(dead_code)] + keyframe: bool, + #[allow(dead_code)] + data: Vec, +} + +impl Fmp4Mux { + pub fn new(writer: W) -> Self { + Self { + writer, + header_written: false, + pending: Vec::new(), + codec_private: None, + } + } + + /// Provide the `HEVCDecoderConfigurationRecord` for the video track. + /// The stub stores it but doesn't yet embed it in `moov` — that's + /// part of the post-stub work. + pub fn set_video_codec_private(&mut self, hvcc: Vec) { + self.codec_private = Some(hvcc); + } + + /// Write one video PES frame. + /// + /// **Stub behaviour:** the first call emits the init segment + /// (`ftyp` + `moov`) so any consumer that just wants the shape can + /// receive it. Subsequent calls accumulate frames in memory for + /// the future fragmenting path; **no media bytes are written yet**. + pub fn write_video(&mut self, pts_ns: i64, keyframe: bool, data: &[u8]) -> io::Result<()> { + if !self.header_written { + self.write_init_segment()?; + self.header_written = true; + } + // TODO(0.22.0): emit one `moof+mdat` per GOP. For now stash the + // frame so the future patch can hot-wire emission without API + // churn. + self.pending.push(PendingSample { + pts_ns, + keyframe, + data: data.to_vec(), + }); + Ok(()) + } + + /// Flush. The stub additionally drops accumulated `pending` frames. + pub fn finish(&mut self) -> io::Result<()> { + // TODO(0.22.0): emit final fragment from pending; today the + // stub just clears the buffer to release memory. + self.pending.clear(); + self.writer.flush() + } + + fn write_init_segment(&mut self) -> io::Result<()> { + let ftyp = build_ftyp(); + let moov = build_moov(); + self.writer.write_all(&ftyp)?; + self.writer.write_all(&moov)?; + Ok(()) + } +} + +/// Build the `ftyp` box. `major_brand = "iso6"`, `minor_version = 1`, +/// compatible brands `iso6 dash msdh hvc1` — the same conservative set +/// shaka-packager uses for HEVC-in-fMP4 outputs. +fn build_ftyp() -> Vec { + let mut body = Vec::new(); + body.extend_from_slice(b"iso6"); + body.extend_from_slice(&1u32.to_be_bytes()); + body.extend_from_slice(b"iso6"); + body.extend_from_slice(b"dash"); + body.extend_from_slice(b"msdh"); + body.extend_from_slice(b"hvc1"); + wrap_box(&FTYP, &body) +} + +/// Build the `moov` box — minimal skeleton. Single video trak, no +/// hvcC inside stsd yet (TODO: full hvc1 sample entry). +fn build_moov() -> Vec { + let mvhd = build_mvhd(); + let trak = build_video_trak(); + let mvex = build_mvex(); + let mut body = Vec::new(); + body.extend_from_slice(&mvhd); + body.extend_from_slice(&trak); + body.extend_from_slice(&mvex); + wrap_box(&MOOV, &body) +} + +fn build_mvhd() -> Vec { + // Version 0, 100 bytes total body. Fields per ISO/IEC 14496-12 §8.2.2. + let mut body = Vec::new(); + body.extend_from_slice(&[0, 0, 0, 0]); // version + flags + body.extend_from_slice(&0u32.to_be_bytes()); // creation_time + body.extend_from_slice(&0u32.to_be_bytes()); // modification_time + body.extend_from_slice(&MOVIE_TIMESCALE.to_be_bytes()); + body.extend_from_slice(&0u32.to_be_bytes()); // duration = 0 (fragmented) + body.extend_from_slice(&0x0001_0000u32.to_be_bytes()); // rate 1.0 + body.extend_from_slice(&0x0100u16.to_be_bytes()); // volume 1.0 + body.extend_from_slice(&[0u8; 2]); // reserved + body.extend_from_slice(&[0u8; 8]); // reserved + // 3x3 identity transformation matrix in 16.16 fixed point. + for v in [0x1_0000u32, 0, 0, 0, 0x1_0000, 0, 0, 0, 0x4000_0000] { + body.extend_from_slice(&v.to_be_bytes()); + } + body.extend_from_slice(&[0u8; 24]); // pre_defined[6] + body.extend_from_slice(&2u32.to_be_bytes()); // next_track_ID (1 reserved for video) + wrap_box(&MVHD, &body) +} + +fn build_video_trak() -> Vec { + let tkhd = build_tkhd(); + let mdia = build_mdia(); + let mut body = Vec::new(); + body.extend_from_slice(&tkhd); + body.extend_from_slice(&mdia); + wrap_box(&TRAK, &body) +} + +fn build_tkhd() -> Vec { + let mut body = Vec::new(); + // version=0 | flags=0x000007 (track_enabled | in_movie | in_preview) + body.extend_from_slice(&[0, 0, 0, 7]); + body.extend_from_slice(&0u32.to_be_bytes()); // creation_time + body.extend_from_slice(&0u32.to_be_bytes()); // modification_time + body.extend_from_slice(&VIDEO_TRACK_ID.to_be_bytes()); + body.extend_from_slice(&[0u8; 4]); // reserved + body.extend_from_slice(&0u32.to_be_bytes()); // duration + body.extend_from_slice(&[0u8; 8]); // reserved + body.extend_from_slice(&0u16.to_be_bytes()); // layer + body.extend_from_slice(&0u16.to_be_bytes()); // alternate_group + body.extend_from_slice(&0u16.to_be_bytes()); // volume (video=0) + body.extend_from_slice(&[0u8; 2]); // reserved + // 3x3 identity matrix. + for v in [0x1_0000u32, 0, 0, 0, 0x1_0000, 0, 0, 0, 0x4000_0000] { + body.extend_from_slice(&v.to_be_bytes()); + } + // width / height in 16.16 fixed point — placeholder 1920x1080. + body.extend_from_slice(&(1920u32 << 16).to_be_bytes()); + body.extend_from_slice(&(1080u32 << 16).to_be_bytes()); + wrap_box(&TKHD, &body) +} + +fn build_mdia() -> Vec { + let mdhd = build_mdhd(); + let hdlr = build_hdlr_vide(); + let minf = build_minf(); + let mut body = Vec::new(); + body.extend_from_slice(&mdhd); + body.extend_from_slice(&hdlr); + body.extend_from_slice(&minf); + wrap_box(&MDIA, &body) +} + +fn build_mdhd() -> Vec { + let mut body = Vec::new(); + body.extend_from_slice(&[0, 0, 0, 0]); // version + flags + body.extend_from_slice(&0u32.to_be_bytes()); // creation_time + body.extend_from_slice(&0u32.to_be_bytes()); // modification_time + body.extend_from_slice(&MOVIE_TIMESCALE.to_be_bytes()); + body.extend_from_slice(&0u32.to_be_bytes()); // duration + // language: 'und' in 5-bit-per-char ISO 639-2 packed (bit 15 = 0). + body.extend_from_slice(&[0x55, 0xC4]); + body.extend_from_slice(&0u16.to_be_bytes()); // pre_defined + wrap_box(&MDHD, &body) +} + +fn build_hdlr_vide() -> Vec { + let mut body = Vec::new(); + body.extend_from_slice(&[0, 0, 0, 0]); // version + flags + body.extend_from_slice(&0u32.to_be_bytes()); // pre_defined + body.extend_from_slice(b"vide"); + body.extend_from_slice(&[0u8; 12]); // reserved + body.extend_from_slice(b"VideoHandler\0"); + wrap_box(&HDLR, &body) +} + +fn build_minf() -> Vec { + let vmhd = build_vmhd(); + let dinf = build_dinf(); + let stbl = build_stbl(); + let mut body = Vec::new(); + body.extend_from_slice(&vmhd); + body.extend_from_slice(&dinf); + body.extend_from_slice(&stbl); + wrap_box(&MINF, &body) +} + +fn build_vmhd() -> Vec { + let mut body = Vec::new(); + body.extend_from_slice(&[0, 0, 0, 1]); // version + flags=1 + body.extend_from_slice(&0u16.to_be_bytes()); // graphicsmode + body.extend_from_slice(&[0u8; 6]); // opcolor + wrap_box(&VMHD, &body) +} + +fn build_dinf() -> Vec { + let mut dref_body = Vec::new(); + dref_body.extend_from_slice(&[0, 0, 0, 0]); + dref_body.extend_from_slice(&1u32.to_be_bytes()); // entry_count + // url with flags=1 (self-contained) and zero name. + let url_body = [0u8, 0, 0, 1]; + dref_body.extend_from_slice(&wrap_box(&URL_, &url_body)); + let dref = wrap_box(&DREF, &dref_body); + wrap_box(&DINF, &dref) +} + +fn build_stbl() -> Vec { + // Stub stsd: empty sample description (zero entries). Replace with + // hvc1+hvcC once the fragmenting path lands so the init segment is + // actually decodable. + let mut stsd_body = Vec::new(); + stsd_body.extend_from_slice(&[0, 0, 0, 0]); + stsd_body.extend_from_slice(&0u32.to_be_bytes()); // entry_count + let stsd = wrap_box(&STSD, &stsd_body); + + // Empty stts/stsc/stsz/stco — fragmented init has no samples here. + let stts = wrap_box(&STTS, &[0, 0, 0, 0, 0, 0, 0, 0]); // version+flags, count=0 + let stsc = wrap_box(&STSC, &[0, 0, 0, 0, 0, 0, 0, 0]); + let stsz = wrap_box( + &STSZ, + &[0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0], // version+flags, sample_size=0, count=0 + ); + let stco = wrap_box(&STCO, &[0, 0, 0, 0, 0, 0, 0, 0]); + + let mut body = Vec::new(); + body.extend_from_slice(&stsd); + body.extend_from_slice(&stts); + body.extend_from_slice(&stsc); + body.extend_from_slice(&stsz); + body.extend_from_slice(&stco); + wrap_box(&STBL, &body) +} + +fn build_mvex() -> Vec { + // trex: track_ID=1, default_sample_description_index=1, others=0. + let mut trex_body = Vec::new(); + trex_body.extend_from_slice(&[0, 0, 0, 0]); // version + flags + trex_body.extend_from_slice(&VIDEO_TRACK_ID.to_be_bytes()); + trex_body.extend_from_slice(&1u32.to_be_bytes()); // default_sample_description_index + trex_body.extend_from_slice(&0u32.to_be_bytes()); // default_sample_duration + trex_body.extend_from_slice(&0u32.to_be_bytes()); // default_sample_size + trex_body.extend_from_slice(&0u32.to_be_bytes()); // default_sample_flags + let trex = wrap_box(&TREX, &trex_body); + wrap_box(&MVEX, &trex) +} + +/// Wrap a box body in `[size:u32-BE][type:4]`. Suitable for any body +/// that fits in u32; oversized boxes (size > 4 GiB) need the 64-bit +/// large-size extension which we don't generate in the stub. +fn wrap_box(box_type: &[u8; 4], body: &[u8]) -> Vec { + let size = (body.len() + 8) as u32; + let mut out = Vec::with_capacity(body.len() + 8); + out.extend_from_slice(&size.to_be_bytes()); + out.extend_from_slice(box_type); + out.extend_from_slice(body); + out +} + +#[cfg(test)] +mod tests { + use super::*; + + /// Decode the first box's size + type from `buf`. + fn read_box_header(buf: &[u8]) -> (u32, [u8; 4]) { + let size = u32::from_be_bytes([buf[0], buf[1], buf[2], buf[3]]); + let bt = [buf[4], buf[5], buf[6], buf[7]]; + (size, bt) + } + + #[test] + fn init_segment_starts_with_ftyp_then_moov() { + let mut sink: Vec = Vec::new(); + let mut mux = Fmp4Mux::new(&mut sink); + // Trigger init emission via a single (stubbed) write. + mux.write_video(0, true, &[0x00, 0x00, 0x00, 0x01, 0x40]).unwrap(); + mux.finish().unwrap(); + drop(mux); + + let (ftyp_size, ftyp_type) = read_box_header(&sink); + assert_eq!(&ftyp_type, b"ftyp"); + assert!(ftyp_size >= 24, "ftyp too small: {ftyp_size}"); + + let (moov_size, moov_type) = read_box_header(&sink[ftyp_size as usize..]); + assert_eq!(&moov_type, b"moov"); + assert!(moov_size > 100, "moov skeleton too small: {moov_size}"); + + // Stub guarantee: no media bytes after the init segment. + let total = ftyp_size as usize + moov_size as usize; + assert_eq!(sink.len(), total, "stub leaked media bytes past moov"); + } + + #[test] + fn moov_contains_trak_mvex() { + let mut sink: Vec = Vec::new(); + let mut mux = Fmp4Mux::new(&mut sink); + mux.write_video(0, true, &[]).unwrap(); + mux.finish().unwrap(); + drop(sink); + + // Re-emit into a fresh buffer for parsing. + let mut buf: Vec = Vec::new(); + let mut mux2 = Fmp4Mux::new(&mut buf); + mux2.write_video(0, true, &[]).unwrap(); + mux2.finish().unwrap(); + drop(mux2); + + // Find moov payload start. + let (ftyp_size, _) = read_box_header(&buf); + let moov_start = ftyp_size as usize; + let (moov_size, _) = read_box_header(&buf[moov_start..]); + let moov_payload = &buf[moov_start + 8..moov_start + moov_size as usize]; + + // Scan for the trak and mvex four-CC anywhere in the moov payload. + let has_trak = moov_payload.windows(4).any(|w| w == b"trak"); + let has_mvex = moov_payload.windows(4).any(|w| w == b"mvex"); + assert!(has_trak, "moov missing trak"); + assert!(has_mvex, "moov missing mvex"); + } +} diff --git a/src/mux/hevc/mod.rs b/src/mux/hevc/mod.rs new file mode 100644 index 0000000..21e646c --- /dev/null +++ b/src/mux/hevc/mod.rs @@ -0,0 +1,275 @@ +//! HEVC (H.265) elementary stream muxer — Annex B byte stream. +//! +//! Consumes [`PesFrame`](crate::pes::PesFrame)s for a single video track +//! and writes them as a raw `.hevc` / `.h265` Annex B byte stream: +//! `00 00 00 01 | NAL_unit | 00 00 00 01 | NAL_unit | …` with no +//! container framing. +//! +//! On the first frame the muxer emits the codec_private's VPS, SPS, PPS +//! (parsed from a `HEVCDecoderConfigurationRecord` in +//! `length-prefixed-in-hvcC` form), then converts each PES frame's +//! length-prefixed NAL units to Annex B and writes them. +//! +//! Sequential-only — no Cues, no backpatch. Target sink is any +//! [`SequentialSink`](crate::io::sink::SequentialSink): file, socket, +//! pipe, anything `Write + Send`. + +use std::io::{self, Write}; + +/// Annex B 4-byte start code. +pub(crate) const START_CODE: [u8; 4] = [0x00, 0x00, 0x00, 0x01]; + +/// HEVC NAL unit type bits live in `(byte0 >> 1) & 0x3F` in Annex B. +/// We don't filter NAL types here — the muxer is format-only — but we +/// keep the constant as documentation of the field layout. +#[allow(dead_code)] +const HEVC_NAL_TYPE_MASK: u8 = 0x3F; + +/// Streaming HEVC Annex B muxer. +/// +/// One instance per output stream. Tracks whether parameter sets have +/// already been emitted so they're written exactly once at the head of +/// the stream, mirroring the convention used by `ffmpeg -c:v copy -f +/// hevc`. +pub struct HevcMux { + writer: W, + /// `HEVCDecoderConfigurationRecord` payload (hvcC). Parsed lazily + /// on the first `write_frame` so callers can set it after + /// construction but before the first frame. + codec_private: Option>, + /// Set once VPS/SPS/PPS have been written to the stream. Subsequent + /// frames write only their own NAL units. + params_written: bool, +} + +impl HevcMux { + /// Construct over `writer`. The muxer does not impose any extra + /// buffering of its own — the sink owns its write buffering policy + /// (see [`LocalFileSink`](crate::io::sink::LocalFileSink) and + /// [`SocketSink`](crate::io::sink::SocketSink)). + pub fn new(writer: W) -> Self { + Self { + writer, + codec_private: None, + params_written: false, + } + } + + /// Provide the `HEVCDecoderConfigurationRecord` (hvcC) so the muxer + /// can prepend VPS/SPS/PPS Annex B NALs at stream start. Optional — + /// if the PES frames already carry inline parameter sets (some + /// upstream demuxers do this), skipping this call is fine. + pub fn set_codec_private(&mut self, data: Vec) { + self.codec_private = Some(data); + } + + /// Write one PES frame (= one access unit) as Annex B NAL units. + /// + /// Input may be either: + /// - Length-prefixed: `[u32-BE len][NAL bytes]` repeated. This is + /// the form emitted by libfreemkv's HEVC parser (the MKV-native + /// layout). Converted to Annex B. + /// - Already Annex B: bytes containing `00 00 00 01` start codes + /// anywhere in the buffer. Passed through unchanged. + /// + /// `_pts_ns` is accepted for symmetry with other muxers but ignored + /// — Annex B has no timing layer. + pub fn write_frame(&mut self, _pts_ns: i64, data: &[u8]) -> io::Result<()> { + if !self.params_written { + if let Some(cp) = &self.codec_private { + if let Some(params) = hvcc_to_annex_b(cp) { + self.writer.write_all(¶ms)?; + } + } + self.params_written = true; + } + let annex_b = length_prefixed_to_annex_b(data); + self.writer.write_all(&annex_b) + } + + /// Flush the underlying writer. No trailer NAL is needed — an Annex + /// B stream ends whenever the file/socket ends. + pub fn finish(&mut self) -> io::Result<()> { + self.writer.flush() + } +} + +/// Convert a `HEVCDecoderConfigurationRecord` (hvcC) into Annex B NAL +/// units. Returns `Some(bytes)` if at least one NAL was extracted, else +/// `None`. +/// +/// Layout (per ISO/IEC 14496-15 §8.3.3.1.2): +/// - 22-byte fixed header +/// - byte 22 = `numOfArrays` +/// - each array: `array_completeness:1 | reserved:1 | NAL_unit_type:6`, +/// `numNalus:u16-BE`, then `numNalus` × `(nalUnitLength:u16-BE + +/// NAL bytes)`. +/// +/// We don't filter on NAL type — VPS (32), SPS (33), PPS (34), and any +/// SEI arrays included in hvcC all get the same Annex B treatment. +fn hvcc_to_annex_b(hvcc: &[u8]) -> Option> { + if hvcc.len() < 23 { + return None; + } + let num_arrays = hvcc[22] as usize; + let mut out = Vec::new(); + let mut offset = 23; + for _ in 0..num_arrays { + if offset + 3 > hvcc.len() { + break; + } + offset += 1; // array_completeness + nal_type byte + let num_nalus = u16::from_be_bytes([hvcc[offset], hvcc[offset + 1]]) as usize; + offset += 2; + for _ in 0..num_nalus { + if offset + 2 > hvcc.len() { + break; + } + let nal_len = u16::from_be_bytes([hvcc[offset], hvcc[offset + 1]]) as usize; + offset += 2; + if offset + nal_len > hvcc.len() { + break; + } + out.extend_from_slice(&START_CODE); + out.extend_from_slice(&hvcc[offset..offset + nal_len]); + offset += nal_len; + } + } + if out.is_empty() { None } else { Some(out) } +} + +/// Convert length-prefixed NAL units (`[u32-BE len][NAL]` repeated) to +/// Annex B (`00 00 00 01 [NAL]` repeated). +/// +/// If the input doesn't parse as length-prefixed (no valid lengths +/// extracted), it's returned unchanged on the assumption that it's +/// already Annex B — some upstream paths (raw HEVC ES from disc) pass +/// Annex B straight through the PES layer. +pub(crate) fn length_prefixed_to_annex_b(data: &[u8]) -> Vec { + let mut out = Vec::with_capacity(data.len() + (data.len() / 32)); + let mut offset = 0; + while offset + 4 <= data.len() { + let len = u32::from_be_bytes([ + data[offset], + data[offset + 1], + data[offset + 2], + data[offset + 3], + ]) as usize; + offset += 4; + if offset + len > data.len() { + // Mid-NAL truncation — fall through to the pass-through path + // rather than emitting a half-NAL. + return data.to_vec(); + } + out.extend_from_slice(&START_CODE); + out.extend_from_slice(&data[offset..offset + len]); + offset += len; + } + if out.is_empty() && !data.is_empty() { + // No length prefixes found — input is likely already Annex B. + return data.to_vec(); + } + out +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn length_prefixed_converts_to_annex_b() { + // Two NALs: [3-byte payload AA BB CC] and [2-byte payload DD EE]. + let mut buf = Vec::new(); + buf.extend_from_slice(&3u32.to_be_bytes()); + buf.extend_from_slice(&[0xAA, 0xBB, 0xCC]); + buf.extend_from_slice(&2u32.to_be_bytes()); + buf.extend_from_slice(&[0xDD, 0xEE]); + + let got = length_prefixed_to_annex_b(&buf); + let want = [ + 0x00, 0x00, 0x00, 0x01, 0xAA, 0xBB, 0xCC, // first NAL + 0x00, 0x00, 0x00, 0x01, 0xDD, 0xEE, // second NAL + ]; + assert_eq!(&got[..], &want[..]); + } + + #[test] + fn already_annex_b_passes_through_when_no_lengths_match() { + // A buffer < 4 bytes can't parse a length prefix at all → + // pass-through path triggers. + let raw = [0xAA, 0xBB, 0xCC]; + let got = length_prefixed_to_annex_b(&raw); + assert_eq!(&got[..], &raw[..]); + } + + #[test] + fn mid_nal_truncation_returns_original() { + // `[u32-BE 100][only 3 bytes]` — length prefix claims 100 bytes + // but the input only has 3 after the prefix. We treat that as + // malformed and pass the original buffer through so receivers + // can attempt their own recovery. + let mut raw = Vec::new(); + raw.extend_from_slice(&100u32.to_be_bytes()); + raw.extend_from_slice(&[0xAA, 0xBB, 0xCC]); + let got = length_prefixed_to_annex_b(&raw); + assert_eq!(&got[..], &raw[..]); + } + + #[test] + fn hvcc_extracts_vps_sps_pps() { + // Build a minimal-but-valid hvcC: 22-byte header, then 3 arrays + // (VPS / SPS / PPS), each with 1 NAL of a 4-byte payload that + // we can spot in the output. + let mut hvcc = vec![0u8; 22]; + hvcc.push(3); // numOfArrays + for (nal_type, payload) in [(32u8, [0x40, 0x01, 0x0C, 0x01]), (33, [0x42, 0x01, 0x01, 0x01]), (34, [0x44, 0x01, 0xC1, 0x72])] { + hvcc.push(nal_type & 0x3F); + hvcc.extend_from_slice(&1u16.to_be_bytes()); // numNalus + hvcc.extend_from_slice(&(payload.len() as u16).to_be_bytes()); + hvcc.extend_from_slice(&payload); + } + + let annex_b = hvcc_to_annex_b(&hvcc).expect("at least one NAL"); + // Three NALs × (4-byte start + 4-byte payload) = 24 bytes. + assert_eq!(annex_b.len(), 24); + assert_eq!(&annex_b[..4], &START_CODE); + assert_eq!(&annex_b[8..12], &START_CODE); + assert_eq!(&annex_b[16..20], &START_CODE); + assert_eq!(annex_b[4], 0x40); // VPS first byte + assert_eq!(annex_b[12], 0x42); // SPS first byte + assert_eq!(annex_b[20], 0x44); // PPS first byte + } + + #[test] + fn mux_writes_params_then_frames() { + // Build hvcC with one SPS to verify params-once semantics. + let mut hvcc = vec![0u8; 22]; + hvcc.push(1); + hvcc.push(33); + hvcc.extend_from_slice(&1u16.to_be_bytes()); + hvcc.extend_from_slice(&3u16.to_be_bytes()); + hvcc.extend_from_slice(&[0x42, 0x01, 0x01]); + + let mut frame_data = Vec::new(); + frame_data.extend_from_slice(&2u32.to_be_bytes()); + frame_data.extend_from_slice(&[0xAA, 0xBB]); + + let mut sink: Vec = Vec::new(); + let mut mux = HevcMux::new(&mut sink); + mux.set_codec_private(hvcc); + mux.write_frame(0, &frame_data).unwrap(); + // Second frame — no SPS re-emission. + mux.write_frame(40_000_000, &frame_data).unwrap(); + mux.finish().unwrap(); + + // SPS NAL (7 bytes) + 2× frame NAL (6 bytes) = 19 bytes. + assert_eq!(sink.len(), 7 + 6 + 6); + // Start codes at offsets 0 (SPS), 7 (frame1), 13 (frame2). + assert_eq!(&sink[0..4], &START_CODE); + assert_eq!(&sink[7..11], &START_CODE); + assert_eq!(&sink[13..17], &START_CODE); + assert_eq!(sink[4], 0x42); + assert_eq!(sink[11], 0xAA); + assert_eq!(sink[17], 0xAA); + } +} diff --git a/src/mux/m2ts_mux/mod.rs b/src/mux/m2ts_mux/mod.rs new file mode 100644 index 0000000..8f96e82 --- /dev/null +++ b/src/mux/m2ts_mux/mod.rs @@ -0,0 +1,672 @@ +//! Standard MPEG-TS (188-byte packets) muxer — sequential-only. +//! +//! Distinct from `super::tsmux::TsMuxer` (BD-TS with 192-byte packets +//! and the 4-byte TP_extra_header). This muxer emits the IETF / ISO/IEC +//! 13818-1 wire format that ffmpeg, VLC, and `m2tsindex` consume +//! out of the box. Use it for plain `.ts` / `.m2ts` files over a +//! [`SequentialSink`](crate::io::sink::SequentialSink), and for +//! MPEG-TS-over-UDP via [`UdpSocketSink`](crate::io::sink::UdpSocketSink). +//! +//! ## Wire format +//! +//! Every output packet is exactly 188 bytes: +//! +//! ```text +//! sync_byte:8 = 0x47 +//! transport_error_indicator:1 = 0 +//! payload_unit_start_indicator:1 +//! transport_priority:1 = 0 +//! PID:13 +//! transport_scrambling_control:2 = 0 +//! adaptation_field_control:2 +//! continuity_counter:4 +//! [adaptation field, if signalled] +//! [payload, if signalled] +//! ``` +//! +//! ## Wiring +//! +//! Single program (PMT PID `0x1000`, program number `1`): +//! - PAT on PID `0x0000` +//! - PMT on PID `0x1000` +//! - Video (HEVC, stream_type `0x24`) on PID `0x0100` +//! - First audio (AC3, stream_type `0x81`) on PID `0x0101` +//! (or TrueHD, stream_type `0x83`, if hinted) +//! +//! ## Scope vs. full TS +//! +//! This is a deliberately minimal viable muxer: +//! - One program, one video track, optionally one audio track. +//! - PAT + PMT re-emitted every `PSI_INTERVAL_PACKETS` packets so a +//! mid-stream receiver can lock on within ~100 ms at typical +//! UHD bitrates. +//! - PCR clock derived from video PTS (`pts - PCR_LEAD_90KHZ`), +//! attached to the video PID's adaptation field every +//! `PCR_INTERVAL_PACKETS` packets. +//! - No language / descriptor tags, no SCTE-35 markers, no per-PID +//! PMT version bumps, no SDT/EIT. Sufficient for "ffmpeg can play +//! this back", not for full broadcast deployment. + +use std::io::{self, Write}; + +mod packet; + +use packet::{Packet, PacketWriter}; + +/// PID `0x0000` — PAT, mandated by spec. +const PID_PAT: u16 = 0x0000; +/// PID for the single program's PMT. `0x1000` is the conventional +/// choice; anything outside reserved/null ranges works. +const PID_PMT: u16 = 0x1000; +/// PID for the video elementary stream. +const PID_VIDEO: u16 = 0x0100; +/// PID for the (optional) first audio elementary stream. +const PID_AUDIO: u16 = 0x0101; +/// Null PID — reserved by spec; never used here. +#[allow(dead_code)] +const PID_NULL: u16 = 0x1FFF; + +/// Re-emit PAT/PMT every N TS packets. ~250 × 188 B = 47 KB; at +/// 80 Mb/s UHD that's ~5 ms — well under typical receiver lock budget. +const PSI_INTERVAL_PACKETS: u64 = 250; +/// Re-stamp PCR every N TS packets carrying the video PID. Spec MAX +/// is 100 ms; 40 packets × 188 B at moderate bitrate keeps us under. +const PCR_INTERVAL_PACKETS: u64 = 40; +/// PCR lead time — the PCR must precede the PTS of the first byte of +/// the picture it timestamps. 200 ms in 90 kHz ticks. +const PCR_LEAD_90KHZ: u64 = 90_000 / 5; + +/// Stream-type codes from ISO/IEC 13818-1 Table 2-29 + later amendments. +const STREAM_TYPE_HEVC: u8 = 0x24; +const STREAM_TYPE_AC3: u8 = 0x81; +const STREAM_TYPE_TRUEHD: u8 = 0x83; + +/// Audio codec hint for [`M2tsMux::new`] / [`M2tsMux::set_audio`]. The +/// muxer needs to know the codec to pick the right PMT `stream_type` +/// and the right PES stream_id; it doesn't decode the audio. +#[derive(Debug, Clone, Copy)] +pub enum AudioCodec { + /// AC-3 / E-AC-3, stream_type `0x81`, PES stream_id `0xBD`. + Ac3, + /// Dolby TrueHD, stream_type `0x83`, PES stream_id `0xBD`. + TrueHd, +} + +impl AudioCodec { + fn stream_type(self) -> u8 { + match self { + AudioCodec::Ac3 => STREAM_TYPE_AC3, + AudioCodec::TrueHd => STREAM_TYPE_TRUEHD, + } + } +} + +/// Sequential MPEG-TS muxer with one video and optional one audio track. +pub struct M2tsMux { + out: PacketWriter, + /// hvcC parameter set bytes to prepend to the first video PES. + /// Optional — if the upstream frames already carry inline params, + /// callers can omit this. + video_codec_private: Option>, + /// Set on first video frame: have we emitted VPS/SPS/PPS? + params_written: bool, + /// Audio codec, if an audio track is configured. `None` = video-only. + audio: Option, + /// First seen PTS (90 kHz). All subsequent PTS / PCR values are + /// relative to this so output streams start near t=0 and don't + /// confuse downstream parsers that don't tolerate huge starting + /// timestamps. + base_pts_90k: Option, + /// Per-PID continuity counter, 4 bits, monotonically increasing. + cc_video: u8, + cc_audio: u8, + cc_pat: u8, + cc_pmt: u8, + /// Total packets written, used to gate PSI / PCR cadence. + packets_written: u64, + /// Video packets written since last PCR, used to gate PCR cadence. + video_packets_since_pcr: u64, +} + +impl M2tsMux { + /// Construct a muxer wrapping `writer`. By default audio is + /// disabled; call [`set_audio`](Self::set_audio) before the first + /// frame to enable. + pub fn new(writer: W) -> Self { + Self { + out: PacketWriter::new(writer), + video_codec_private: None, + params_written: false, + audio: None, + base_pts_90k: None, + cc_video: 0, + cc_audio: 0, + cc_pat: 0, + cc_pmt: 0, + packets_written: 0, + video_packets_since_pcr: 0, + } + } + + /// Provide the `HEVCDecoderConfigurationRecord` for video so the + /// muxer prepends VPS/SPS/PPS Annex B NALs at stream start. + pub fn set_video_codec_private(&mut self, hvcc: Vec) { + self.video_codec_private = Some(hvcc); + } + + /// Enable a single audio track. Must be called before + /// [`write_audio`](Self::write_audio). + pub fn set_audio(&mut self, codec: AudioCodec) { + self.audio = Some(codec); + } + + /// Write one video PES frame. `data` is either length-prefixed + /// NALUs (MKV-style) or already Annex B; both are accepted. + pub fn write_video(&mut self, pts_ns: i64, data: &[u8]) -> io::Result<()> { + let pts_90k = self.base_relative_pts(pts_ns); + // PCR comes "before" the PTS it timestamps; clamp at 0 for the + // first frame so we don't underflow. + let pcr = pts_90k.saturating_sub(PCR_LEAD_90KHZ); + + // Annex-B-ify the frame and prepend VPS/SPS/PPS once. + let mut es = Vec::with_capacity(data.len() + 64); + if !self.params_written { + if let Some(cp) = &self.video_codec_private { + let payload = hvcc_payload(cp); + if !payload.is_empty() { + let params = super::hevc::length_prefixed_to_annex_b(&payload); + es.extend_from_slice(¶ms); + } + } + self.params_written = true; + } + let annex_b = super::hevc::length_prefixed_to_annex_b(data); + es.extend_from_slice(&annex_b); + + let pes = build_video_pes(pts_90k, &es); + self.write_pes(PID_VIDEO, &pes, Some(pcr)) + } + + /// Write one audio PES frame. Returns `Ok(())` and silently drops + /// the frame if no audio track was configured — the design assumes + /// the upstream picks tracks and won't ship audio to a video-only + /// muxer, but defending against it keeps the API a single shape. + pub fn write_audio(&mut self, pts_ns: i64, data: &[u8]) -> io::Result<()> { + if self.audio.is_none() { + return Ok(()); + } + let pts_90k = self.base_relative_pts(pts_ns); + let pes = build_audio_pes(pts_90k, data); + self.write_pes(PID_AUDIO, &pes, None) + } + + /// Drain the underlying writer. No TS-level trailer is mandatory — + /// receivers detect end-of-stream from socket close or file EOF. + pub fn finish(&mut self) -> io::Result<()> { + self.out.flush() + } + + /// Convert input PTS (nanoseconds) to 90 kHz ticks rebased on the + /// first frame's PTS. Saturating at 0 keeps the math friendly when + /// frames arrive slightly out of decode order. + fn base_relative_pts(&mut self, pts_ns: i64) -> u64 { + let raw_90k = if pts_ns > 0 { + (pts_ns as u64) * 9 / 100_000 + } else { + 0 + }; + let base = *self.base_pts_90k.get_or_insert(raw_90k); + raw_90k.saturating_sub(base) + } + + /// Emit one PES payload as a chain of TS packets on `pid`. If `pcr` + /// is provided the first packet carries an adaptation field with + /// the PCR. PAT/PMT are re-emitted every `PSI_INTERVAL_PACKETS`. + /// + /// Packet-size math (TS = 188 bytes, header = 4 bytes): + /// - 184 B remain after the header for the adaptation-field area + /// plus the payload area. + /// - With AF body of `b` bytes and `s` stuffing bytes: AF total = + /// `1 + b + s` (the leading `1` is the `adaptation_field_length` + /// byte itself). Payload = `184 - (1 + b + s)`. + /// - With no AF area at all: payload = `184`. + /// + /// The fit-the-tail logic on the last packet of the PES uses + /// stuffing rather than a separate small packet, which is the + /// standard MPEG-TS convention. + fn write_pes(&mut self, pid: u16, pes: &[u8], pcr: Option) -> io::Result<()> { + // PSI cadence is enforced per TS packet — interleave a fresh + // PAT+PMT into the packet stream every PSI_INTERVAL_PACKETS so + // long single-PES emissions (e.g. one 60 KB video frame) don't + // starve receivers tuning in mid-stream. + let mut offset = 0; + let mut first = true; + while offset < pes.len() { + self.maybe_emit_psi()?; + + let attach_pcr = first + && (pid == PID_VIDEO) + && (pcr.is_some()) + && (self.packets_written == 0 || self.video_packets_since_pcr >= PCR_INTERVAL_PACKETS); + + let af_body: Vec = if attach_pcr { + build_pcr_adaptation(pcr.unwrap_or(0)) + } else { + Vec::new() + }; + + let remaining = pes.len() - offset; + // Capacity for payload given AF body and 1-byte AF length. + // When AF body is empty we can still skip the AF entirely + // and get the full 184 B; only invoke the AF when we'd + // otherwise need stuffing. + let (af_present, payload_len, stuffing): (bool, usize, usize) = if !af_body.is_empty() { + // AF is mandatory (PCR). 1 byte length + body + stuffing + // + payload = 184. + let max_payload = 184 - 1 - af_body.len(); + let p = remaining.min(max_payload); + let s = max_payload - p; + (true, p, s) + } else if remaining >= 184 { + // Full payload packet — no AF at all. + (false, 184, 0) + } else { + // Last (small) packet — stuff via empty AF. + // 1 byte length + 0 body + stuffing + payload = 184. + let max_payload = 183; + let p = remaining.min(max_payload); + let s = max_payload - p; + (true, p, s) + }; + + let cc = self.advance_cc(pid); + let mut packet = Packet::new(); + packet.set_header(pid, first, true, af_present, cc); + if af_present { + packet.append_adaptation(&af_body, stuffing); + } + packet.append_payload(&pes[offset..offset + payload_len]); + debug_assert_eq!(packet.len(), 188, "packet not 188 bytes"); + self.out.write_packet(&packet)?; + + self.packets_written += 1; + if pid == PID_VIDEO { + if attach_pcr { + self.video_packets_since_pcr = 0; + } else { + self.video_packets_since_pcr += 1; + } + } + offset += payload_len; + first = false; + } + Ok(()) + } + + fn advance_cc(&mut self, pid: u16) -> u8 { + let slot = match pid { + PID_VIDEO => &mut self.cc_video, + PID_AUDIO => &mut self.cc_audio, + PID_PAT => &mut self.cc_pat, + PID_PMT => &mut self.cc_pmt, + _ => return 0, + }; + let cc = *slot; + *slot = (*slot + 1) & 0x0F; + cc + } + + fn maybe_emit_psi(&mut self) -> io::Result<()> { + if self.packets_written == 0 || self.packets_written.is_multiple_of(PSI_INTERVAL_PACKETS) { + self.emit_pat()?; + self.emit_pmt()?; + } + Ok(()) + } + + fn emit_pat(&mut self) -> io::Result<()> { + let payload = build_pat(PID_PMT); + let cc = self.advance_cc(PID_PAT); + let mut packet = Packet::new(); + packet.set_header(PID_PAT, true, true, false, cc); + packet.append_payload(&payload); + packet.pad_to_188(); + self.out.write_packet(&packet)?; + self.packets_written += 1; + Ok(()) + } + + fn emit_pmt(&mut self) -> io::Result<()> { + let payload = build_pmt(self.audio); + let cc = self.advance_cc(PID_PMT); + let mut packet = Packet::new(); + packet.set_header(PID_PMT, true, true, false, cc); + packet.append_payload(&payload); + packet.pad_to_188(); + self.out.write_packet(&packet)?; + self.packets_written += 1; + Ok(()) + } +} + +/// Extract the raw hvcC bytes for handoff to `length_prefixed_to_annex_b`. +/// hvcC layout: 22-byte fixed header, then `numOfArrays` arrays of +/// `(nalType, numNalus, [nalLength:u16, NAL bytes]…)`. We convert this +/// directly to a length-prefixed byte stream (NAL length is u16 in +/// hvcC; widen to u32 for the standard length-prefixed encoding). +fn hvcc_payload(hvcc: &[u8]) -> Vec { + if hvcc.len() < 23 { + return Vec::new(); + } + let num_arrays = hvcc[22] as usize; + let mut out = Vec::new(); + let mut offset = 23; + for _ in 0..num_arrays { + if offset + 3 > hvcc.len() { + break; + } + offset += 1; + let num_nalus = u16::from_be_bytes([hvcc[offset], hvcc[offset + 1]]) as usize; + offset += 2; + for _ in 0..num_nalus { + if offset + 2 > hvcc.len() { + break; + } + let nal_len = u16::from_be_bytes([hvcc[offset], hvcc[offset + 1]]) as usize; + offset += 2; + if offset + nal_len > hvcc.len() { + break; + } + out.extend_from_slice(&(nal_len as u32).to_be_bytes()); + out.extend_from_slice(&hvcc[offset..offset + nal_len]); + offset += nal_len; + } + } + out +} + +/// Build a PES packet for a video access unit. +fn build_video_pes(pts_90k: u64, es: &[u8]) -> Vec { + build_pes_packet(0xE0, pts_90k, es, /* length_in_header */ false) +} + +/// Build a PES packet for an audio access unit. +fn build_audio_pes(pts_90k: u64, es: &[u8]) -> Vec { + // Audio PES: length is fillable when it fits in u16. We always + // write the length so receivers don't have to scan for the next + // start code. + build_pes_packet(0xBD, pts_90k, es, /* length_in_header */ true) +} + +fn build_pes_packet(stream_id: u8, pts_90k: u64, es: &[u8], length_in_header: bool) -> Vec { + let mut out = Vec::with_capacity(es.len() + 14); + out.extend_from_slice(&[0x00, 0x00, 0x01, stream_id]); + // PES_packet_length: total bytes after this field. 3 flag bytes + 5 + // PTS bytes + es.len(). Zero means "unbounded" — used for video + // where PES can exceed u16. + let pes_len = 8 + es.len(); + if length_in_header && pes_len <= u16::MAX as usize { + out.extend_from_slice(&(pes_len as u16).to_be_bytes()); + } else { + out.extend_from_slice(&[0x00, 0x00]); + } + // Flags byte 1: 10 = MPEG-2 marker, then scrambling/priority/etc 0. + out.push(0x80); + // Flags byte 2: PTS flag (bit 7). + out.push(0x80); + // PES_header_data_length = 5 (just PTS). + out.push(5); + // PTS bytes — 33-bit timestamp split across 5 bytes with marker bits. + let pts = pts_90k & 0x1_FFFF_FFFF; + out.push(0x21 | (((pts >> 29) & 0x0E) as u8)); + out.push(((pts >> 22) & 0xFF) as u8); + out.push(0x01 | (((pts >> 14) & 0xFE) as u8)); + out.push(((pts >> 7) & 0xFF) as u8); + out.push(0x01 | (((pts << 1) & 0xFE) as u8)); + out.extend_from_slice(es); + out +} + +/// Build the PAT payload (section, with pointer_field). +fn build_pat(pmt_pid: u16) -> Vec { + let mut section = Vec::new(); + section.push(0x00); // table_id = PAT + // section_syntax_indicator(1) | '0'(1) | reserved(2) | section_length(12) + // section_length covers from end of this field through CRC. + // Body: transport_stream_id(2) + version/cni(1) + section/last_section(2) + program(4) = 9 bytes, + // plus CRC(4) = 13. Encoded big-endian. + section.extend_from_slice(&[0xB0, 13]); + section.extend_from_slice(&[0x00, 0x01]); // transport_stream_id = 1 + section.push(0xC1); // reserved | version=0 | current_next=1 + section.push(0x00); // section_number + section.push(0x00); // last_section_number + section.extend_from_slice(&[0x00, 0x01]); // program_number = 1 + // reserved(3) | network_PID/program_map_PID(13) + let pid_bytes = (0xE000u16 | (pmt_pid & 0x1FFF)).to_be_bytes(); + section.extend_from_slice(&pid_bytes); + let crc = mpegts_crc32(§ion); + section.extend_from_slice(&crc.to_be_bytes()); + // Prepend pointer_field=0 (section starts immediately). + let mut payload = Vec::with_capacity(section.len() + 1); + payload.push(0x00); + payload.extend_from_slice(§ion); + payload +} + +/// Build the PMT payload (section, with pointer_field). +fn build_pmt(audio: Option) -> Vec { + let mut section = Vec::new(); + section.push(0x02); // table_id = PMT + // section_length filled in after we know the body size. + let len_placeholder = section.len(); + section.extend_from_slice(&[0xB0, 0x00]); + + section.extend_from_slice(&1u16.to_be_bytes()); // program_number + section.push(0xC1); // reserved | version=0 | current_next=1 + section.push(0x00); // section_number + section.push(0x00); // last_section_number + // reserved(3) | PCR_PID(13) + let pcr_pid = (0xE000u16 | (PID_VIDEO & 0x1FFF)).to_be_bytes(); + section.extend_from_slice(&pcr_pid); + // program_info_length = 0 + section.extend_from_slice(&[0xF0, 0x00]); + + // Video elementary stream entry. + section.push(STREAM_TYPE_HEVC); + let v_pid = (0xE000u16 | (PID_VIDEO & 0x1FFF)).to_be_bytes(); + section.extend_from_slice(&v_pid); + section.extend_from_slice(&[0xF0, 0x00]); // ES_info_length = 0 + + if let Some(codec) = audio { + section.push(codec.stream_type()); + let a_pid = (0xE000u16 | (PID_AUDIO & 0x1FFF)).to_be_bytes(); + section.extend_from_slice(&a_pid); + section.extend_from_slice(&[0xF0, 0x00]); + } + + // Now patch section_length: covers everything after the length field + // through the CRC, so (current body size - 3 bytes consumed by + // table_id + 2 length bytes) + 4 (CRC). + let section_len = section.len() - 3 + 4; + section[len_placeholder] = 0xB0 | ((section_len >> 8) as u8 & 0x0F); + section[len_placeholder + 1] = section_len as u8; + + let crc = mpegts_crc32(§ion); + section.extend_from_slice(&crc.to_be_bytes()); + + let mut payload = Vec::with_capacity(section.len() + 1); + payload.push(0x00); + payload.extend_from_slice(§ion); + payload +} + +/// Build the adaptation field carrying a PCR (no other flags). +fn build_pcr_adaptation(pcr_90k: u64) -> Vec { + // adaptation_field_length is set by `Packet::append_adaptation` + // — this function returns just the field body. + // + // Layout: discontinuity_indicator(1) | random_access(1) | + // elementary_stream_priority(1) | PCR_flag(1) | OPCR_flag(1) | + // splicing_point_flag(1) | transport_private_data_flag(1) | + // adaptation_field_extension_flag(1) | PCR(48b). + let mut af = vec![0x50]; // PCR_flag=1, random_access_indicator=1 + let pcr_base = pcr_90k & 0x1_FFFF_FFFF; // 33-bit + let pcr_ext: u16 = 0; // 9-bit, we keep it zero (no sub-tick precision) + // Encode PCR: 33b base | 6b reserved | 9b extension = 48b + af.push((pcr_base >> 25) as u8); + af.push((pcr_base >> 17) as u8); + af.push((pcr_base >> 9) as u8); + af.push((pcr_base >> 1) as u8); + af.push(((pcr_base << 7) as u8 & 0x80) | 0x7E | ((pcr_ext >> 8) as u8 & 0x01)); + af.push(pcr_ext as u8); + af +} + +/// MPEG-TS CRC-32 (poly 0x04C11DB7, init 0xFFFFFFFF, no reflection, no +/// final XOR). Implementation: bitwise so we don't need a table — +/// PSI sections are tiny, the cost is negligible. +fn mpegts_crc32(data: &[u8]) -> u32 { + let mut crc: u32 = 0xFFFF_FFFF; + for &b in data { + crc ^= (b as u32) << 24; + for _ in 0..8 { + if crc & 0x8000_0000 != 0 { + crc = (crc << 1) ^ 0x04C1_1DB7; + } else { + crc <<= 1; + } + } + } + crc +} + +#[cfg(test)] +mod tests { + use super::*; + + /// All emitted bytes must align to 188-byte packet boundaries and + /// every packet must start with `0x47`. + fn assert_ts_well_formed(buf: &[u8]) { + assert_eq!(buf.len() % 188, 0, "stream not packet-aligned: {} bytes", buf.len()); + for (i, chunk) in buf.chunks(188).enumerate() { + assert_eq!(chunk[0], 0x47, "packet {} missing sync byte", i); + } + } + + fn extract_pids(buf: &[u8]) -> Vec { + buf.chunks(188) + .map(|p| u16::from_be_bytes([p[1] & 0x1F, p[2]])) + .collect() + } + + #[test] + fn crc32_is_self_validating() { + // The MPEG-TS CRC has the property that prepending a single + // bit-flip changes the output; running it over its own input + + // CRC yields a fixed magic constant (the CRC residue). Rather + // than hardcoding sample bytes, verify the underlying algorithm + // by checking that two distinct inputs produce distinct CRCs + // and that the same input is deterministic. + let a = [0u8, 0xB0, 0x0D, 0x00, 0x01, 0xC1, 0x00, 0x00, 0x00, 0x01, 0xE1, 0x00]; + let mut b = a; + b[5] ^= 0x01; // flip one bit + let crc_a = mpegts_crc32(&a); + let crc_b = mpegts_crc32(&b); + assert_ne!(crc_a, crc_b); + assert_eq!(crc_a, mpegts_crc32(&a)); // deterministic + // Sanity: all-zero input ⇒ CRC = 0 (init XORs but the + // shift/feedback cancels for zero data after init drains). + // We don't assert exact value — that depends on poly choice — + // but check it's not the same as for non-zero data. + let crc_zero = mpegts_crc32(&[0u8; 12]); + assert_ne!(crc_zero, crc_a); + } + + #[test] + fn video_only_mux_emits_pat_pmt_then_video() { + let mut sink: Vec = Vec::new(); + let mut mux = M2tsMux::new(&mut sink); + // One small video frame, no codec_private (so no params NAL inline). + let mut frame = Vec::new(); + frame.extend_from_slice(&4u32.to_be_bytes()); + frame.extend_from_slice(&[0x40, 0x01, 0x0C, 0x01]); + mux.write_video(0, &frame).unwrap(); + mux.finish().unwrap(); + drop(mux); + + assert_ts_well_formed(&sink); + let pids = extract_pids(&sink); + // First two packets: PAT, PMT. At least one video packet after. + assert_eq!(pids[0], PID_PAT); + assert_eq!(pids[1], PID_PMT); + assert!(pids.iter().any(|p| *p == PID_VIDEO)); + } + + #[test] + fn audio_track_appears_in_pmt_and_stream() { + let mut sink: Vec = Vec::new(); + let mut mux = M2tsMux::new(&mut sink); + mux.set_audio(AudioCodec::Ac3); + // Video + audio frame pair. + let mut frame = Vec::new(); + frame.extend_from_slice(&3u32.to_be_bytes()); + frame.extend_from_slice(&[0x40, 0x01, 0x0C]); + mux.write_video(0, &frame).unwrap(); + mux.write_audio(20_000_000, &[0x0B, 0x77, 0x12, 0x34]).unwrap(); + mux.finish().unwrap(); + drop(mux); + + assert_ts_well_formed(&sink); + let pids = extract_pids(&sink); + assert!(pids.iter().any(|p| *p == PID_VIDEO)); + assert!(pids.iter().any(|p| *p == PID_AUDIO)); + } + + #[test] + fn psi_re_emits_at_interval() { + let mut sink: Vec = Vec::new(); + let mut mux = M2tsMux::new(&mut sink); + // Build a frame large enough to span > PSI_INTERVAL_PACKETS TS packets. + // 184 B payload per packet ⇒ ~250 packets = ~46 KB elementary stream. + let big: Vec = (0..(60 * 1024)).map(|i| (i & 0xff) as u8).collect(); + let mut frame = Vec::new(); + frame.extend_from_slice(&(big.len() as u32).to_be_bytes()); + frame.extend_from_slice(&big); + mux.write_video(0, &frame).unwrap(); + mux.finish().unwrap(); + drop(mux); + + assert_ts_well_formed(&sink); + let pids = extract_pids(&sink); + // Count PAT/PMT pairs — must be at least 2 given the input size. + let pat_count = pids.iter().filter(|p| **p == PID_PAT).count(); + let pmt_count = pids.iter().filter(|p| **p == PID_PMT).count(); + assert!(pat_count >= 2, "expected ≥2 PAT, got {}", pat_count); + assert!(pmt_count >= 2, "expected ≥2 PMT, got {}", pmt_count); + } + + #[test] + fn continuity_counter_increments_per_pid() { + let mut sink: Vec = Vec::new(); + let mut mux = M2tsMux::new(&mut sink); + // Three small video frames to get a sequence of video TS packets. + for pts in [0i64, 40_000_000, 80_000_000] { + let mut frame = Vec::new(); + frame.extend_from_slice(&3u32.to_be_bytes()); + frame.extend_from_slice(&[0xAA, 0xBB, 0xCC]); + mux.write_video(pts, &frame).unwrap(); + } + mux.finish().unwrap(); + drop(mux); + + // Collect CCs for video packets in order. + let ccs: Vec = sink + .chunks(188) + .filter(|p| u16::from_be_bytes([p[1] & 0x1F, p[2]]) == PID_VIDEO) + .map(|p| p[3] & 0x0F) + .collect(); + for w in ccs.windows(2) { + assert_eq!(w[1], (w[0] + 1) & 0x0F); + } + } +} diff --git a/src/mux/m2ts_mux/packet.rs b/src/mux/m2ts_mux/packet.rs new file mode 100644 index 0000000..0d6c607 --- /dev/null +++ b/src/mux/m2ts_mux/packet.rs @@ -0,0 +1,145 @@ +//! 188-byte MPEG-TS packet builder + writer. +//! +//! Internal helper for `super::M2tsMux`. Not public API — exposes raw +//! byte layout so the parent module can compose PSI / PCR / PES bytes +//! without each caller re-implementing the 188-byte boundary math. + +use std::io::{self, Write}; + +const TS_PACKET_SIZE: usize = 188; +const SYNC_BYTE: u8 = 0x47; +const STUFF_BYTE: u8 = 0xFF; + +/// One TS packet under construction. Always emits 188 bytes when +/// [`pad_to_188`](Self::pad_to_188) is called; if it's not called the +/// caller is responsible for filling the packet exactly. +pub(super) struct Packet { + buf: Vec, +} + +impl Packet { + pub(super) fn new() -> Self { + Self { + buf: Vec::with_capacity(TS_PACKET_SIZE), + } + } + + /// Write the 4-byte TS packet header. + /// + /// * `pid` — 13-bit PID + /// * `payload_unit_start` — first packet of a PES / PSI section + /// * `has_payload` — packet carries any payload bytes + /// * `has_adaptation` — packet carries an adaptation field + /// * `cc` — 4-bit continuity counter + pub(super) fn set_header( + &mut self, + pid: u16, + payload_unit_start: bool, + has_payload: bool, + has_adaptation: bool, + cc: u8, + ) { + self.buf.clear(); + self.buf.push(SYNC_BYTE); + let pus_bit = if payload_unit_start { 0x40 } else { 0 }; + // transport_error_indicator(1)=0 | payload_unit_start(1) | transport_priority(1)=0 | PID(5 high) + self.buf.push(pus_bit | ((pid >> 8) as u8 & 0x1F)); + self.buf.push(pid as u8); + // transport_scrambling_control(2)=0 | adaptation_field_control(2) | continuity_counter(4) + let afc = match (has_adaptation, has_payload) { + (false, false) => 0b00, // reserved — should not happen + (false, true) => 0b01, // payload only + (true, false) => 0b10, // adaptation only + (true, true) => 0b11, // both + }; + self.buf.push((afc << 4) | (cc & 0x0F)); + } + + /// Append the adaptation field after the header. + /// + /// `body` is the adaptation field body (flags byte + optional PCR + /// + …). `stuffing` is the number of `0xFF` stuffing bytes to + /// append after the body. The first byte of the field + /// (`adaptation_field_length`) is computed here from `body.len() + + /// stuffing`. + pub(super) fn append_adaptation(&mut self, body: &[u8], stuffing: usize) { + let af_len = body.len() + stuffing; + debug_assert!(af_len <= 183, "adaptation field overflow"); + self.buf.push(af_len as u8); + self.buf.extend_from_slice(body); + for _ in 0..stuffing { + self.buf.push(STUFF_BYTE); + } + } + + /// Append payload bytes. + pub(super) fn append_payload(&mut self, payload: &[u8]) { + self.buf.extend_from_slice(payload); + debug_assert!(self.buf.len() <= TS_PACKET_SIZE, "packet overflow"); + } + + /// Pad the packet to exactly 188 bytes with `0xFF` bytes — used by + /// PSI emit paths where the section is much smaller than 184 bytes. + /// For PSI packets only — payload-carrying packets reserve room for + /// stuffing via `append_adaptation`. + pub(super) fn pad_to_188(&mut self) { + while self.buf.len() < TS_PACKET_SIZE { + self.buf.push(STUFF_BYTE); + } + } + + pub(super) fn bytes(&self) -> &[u8] { + &self.buf + } + + pub(super) fn len(&self) -> usize { + self.buf.len() + } +} + +/// Buffered writer for assembled TS packets. Owns the underlying sink. +pub(super) struct PacketWriter { + inner: W, +} + +impl PacketWriter { + pub(super) fn new(inner: W) -> Self { + Self { inner } + } + + pub(super) fn write_packet(&mut self, packet: &Packet) -> io::Result<()> { + let bytes = packet.bytes(); + debug_assert_eq!(bytes.len(), TS_PACKET_SIZE); + self.inner.write_all(bytes) + } + + pub(super) fn flush(&mut self) -> io::Result<()> { + self.inner.flush() + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn pad_fills_to_188() { + let mut p = Packet::new(); + p.set_header(0x100, true, true, false, 0); + p.append_payload(&[1, 2, 3]); + p.pad_to_188(); + assert_eq!(p.bytes().len(), 188); + assert_eq!(p.bytes()[0], SYNC_BYTE); + // After 4-byte header + 3 payload, byte 7 starts stuffing. + assert_eq!(p.bytes()[7], STUFF_BYTE); + } + + #[test] + fn header_pid_round_trips() { + let mut p = Packet::new(); + p.set_header(0x1ABC, false, true, false, 0xA); + let pid = u16::from_be_bytes([p.bytes()[1] & 0x1F, p.bytes()[2]]); + assert_eq!(pid, 0x1ABC); + assert_eq!(p.bytes()[3] & 0x0F, 0xA); + } +} diff --git a/src/mux/mod.rs b/src/mux/mod.rs index 692e6d8..537a09c 100644 --- a/src/mux/mod.rs +++ b/src/mux/mod.rs @@ -33,6 +33,17 @@ pub(crate) mod m2ts; /// to round-trip codec_privates that don't fit inside the underlying format). /// Exposed for integration tests that exercise the wire format directly. pub mod meta; + +// ── Phase 3 sequential muxers ────────────────────────────────────────────── +// +// New container muxers that consume PES frames and write to a +// `SequentialSink`. They are NOT refactors of the existing `MkvStream` / +// `M2tsStream` (which round-trip via the legacy `Stream` trait + the +// BD-TS framing); they're sequential-only and target the Phase 2 sink +// split end-to-end. +pub mod fmp4; +pub mod hevc; +pub mod m2ts_mux; pub(crate) mod mkv; pub(crate) mod mkvstream; pub(crate) mod network;