0.18 round 2: FrameSink concrete migrations for mux/* sinks

This commit is contained in:
MattJackson
2026-05-09 09:53:12 -07:00
7 changed files with 188 additions and 34 deletions
+31 -7
View File
@@ -18,10 +18,10 @@ const SCAN_SIZE: usize = 1024 * 1024;
enum Mode {
Write {
muxer: super::tsmux::TsMuxer<Box<dyn Write>>,
muxer: super::tsmux::TsMuxer<Box<dyn Write + Send>>,
},
Read {
reader: Box<dyn Read>,
reader: Box<dyn Read + Send>,
},
}
@@ -57,7 +57,7 @@ pub struct M2tsStream {
impl M2tsStream {
/// Create for writing PES frames → BD-TS output.
/// Writes FMKV metadata header, then muxes PES frames into BD transport stream.
pub fn create(mut writer: impl Write + 'static, title: &DiscTitle) -> io::Result<Self> {
pub fn create(mut writer: impl Write + Send + 'static, title: &DiscTitle) -> io::Result<Self> {
// Write FMKV metadata header
if !title.streams.is_empty() {
let m = meta::M2tsMeta::from_title(title);
@@ -72,7 +72,7 @@ impl M2tsStream {
DiscStream::Subtitle(s) => s.pid,
})
.collect();
let boxed: Box<dyn Write> = Box::new(writer);
let boxed: Box<dyn Write + Send> = Box::new(writer);
let mut muxer = super::tsmux::TsMuxer::new(boxed, &pids);
for (i, cp) in title.codec_privates.iter().enumerate() {
if let Some(data) = cp {
@@ -111,7 +111,7 @@ impl M2tsStream {
/// Open an M2TS stream for reading. Takes any Read source — file, pipe, socket.
///
/// Tries FMKV metadata header first. Falls back to PMT scan of first 1 MB.
pub fn open(mut reader: impl Read + 'static) -> io::Result<Self> {
pub fn open(mut reader: impl Read + Send + 'static) -> io::Result<Self> {
// Read first chunk — enough for FMKV header or PMT scan
let mut head = vec![0u8; SCAN_SIZE];
let head_len = read_fill(&mut reader, &mut head)?;
@@ -125,7 +125,7 @@ impl M2tsStream {
let (pids, parsers, pid_to_track) = Self::setup_pes(&title.streams);
// Chain: remaining head bytes + rest of reader
let remaining_head = &head[header_end..];
let chain: Box<dyn Read> =
let chain: Box<dyn Read + Send> =
Box::new(io::Cursor::new(remaining_head.to_vec()).chain(reader));
return Ok(Self {
disc_title: title.clone(),
@@ -150,7 +150,7 @@ impl M2tsStream {
let (pids, parsers, pid_to_track) = Self::setup_pes(&streams);
// Chain: full head (it's all TS data) + rest of reader
let chain: Box<dyn Read> = Box::new(io::Cursor::new(head).chain(reader));
let chain: Box<dyn Read + Send> = Box::new(io::Cursor::new(head).chain(reader));
Ok(Self {
disc_title: DiscTitle {
@@ -284,3 +284,27 @@ impl crate::pes::Stream for M2tsStream {
true
}
}
/// FrameSink sibling to the deprecated Stream impl; both coexist during the
/// 0.18 deprecation window. Caller may pick either at the trait-object
/// boundary — `Box<dyn Stream>` (deprecated) or `Box<dyn FrameSink>` (new).
/// Use `M2tsStream::create(writer, title)` to construct the write half;
/// calling `FrameSink::write` on an `M2tsStream::open(reader)` instance
/// returns `StreamReadOnly`.
#[allow(deprecated)] // delegating to deprecated Stream during the 0.18 deprecation window so callers don't see the deprecation twice.
impl crate::pes::FrameSink for M2tsStream {
fn write(&mut self, frame: &crate::pes::PesFrame) -> io::Result<()> {
<Self as crate::pes::Stream>::write(self, frame)
}
fn finish(self: Box<Self>) -> io::Result<()> {
// Why: Stream::finish takes &mut self, FrameSink::finish takes Box<Self>.
// Re-borrow inside the box, call Stream::finish, drop the box.
let mut s: Self = *self;
<Self as crate::pes::Stream>::finish(&mut s)
}
fn info(&self) -> &crate::disc::DiscTitle {
<Self as crate::pes::Stream>::info(self)
}
}
+33 -4
View File
@@ -18,7 +18,7 @@ use crate::disc::*;
use std::io::{self, Read};
struct ReadState {
reader: Box<dyn Read>,
reader: Box<dyn Read + Send>,
cluster_ts_ms: i64,
/// Codec private data per track (track_number, hvcC/avcC bytes).
codec_privates: Vec<(u16, Vec<u8>)>,
@@ -26,7 +26,7 @@ struct ReadState {
enum Mode {
Write {
muxer: Option<MkvMuxer<Box<dyn WriteSeek>>>,
muxer: Option<MkvMuxer<Box<dyn WriteSeek + Send>>>,
},
Read(ReadState),
}
@@ -40,7 +40,7 @@ pub struct MkvStream {
impl MkvStream {
/// Create for writing PES frames → MKV container.
/// Codec privates come from title.codec_privates (populated by input stream).
pub fn create(writer: Box<dyn WriteSeek>, title: &DiscTitle) -> io::Result<Self> {
pub fn create(writer: Box<dyn WriteSeek + Send>, title: &DiscTitle) -> io::Result<Self> {
let mut tracks = Vec::new();
let mut has_default_video = false;
let mut has_default_audio = false;
@@ -79,7 +79,7 @@ impl MkvStream {
}
/// Open an MKV file for reading → PES frames.
pub fn open(mut reader: impl Read + 'static) -> io::Result<Self> {
pub fn open(mut reader: impl Read + Send + 'static) -> io::Result<Self> {
let (disc_title, codec_privates) = parse_mkv_header(&mut reader)?;
Ok(Self {
disc_title,
@@ -189,6 +189,35 @@ impl crate::pes::Stream for MkvStream {
}
}
/// FrameSink sibling to the deprecated Stream impl; both coexist during the
/// 0.18 deprecation window. Caller may pick either at the trait-object
/// boundary — `Box<dyn Stream>` (deprecated) or `Box<dyn FrameSink>` (new).
/// Use `MkvStream::create(writer, title)` to construct the write half;
/// calling `FrameSink::write` on a `MkvStream::open(reader)` instance returns
/// `StreamReadOnly`. `finish` is where the Cues index is written, so it must
/// be called for the resulting MKV to be seekable.
#[allow(deprecated)] // delegating to deprecated Stream during the 0.18 deprecation window so callers don't see the deprecation twice.
impl crate::pes::FrameSink for MkvStream {
fn write(&mut self, frame: &crate::pes::PesFrame) -> io::Result<()> {
<Self as crate::pes::Stream>::write(self, frame)
}
fn finish(self: Box<Self>) -> io::Result<()> {
// Why: Stream::finish takes &mut self, FrameSink::finish takes Box<Self>.
// Re-borrow inside the box, call Stream::finish, drop the box.
// The inner `MkvMuxer` is owned via `Option`, and `Stream::finish`
// already takes it via `Option::take()` — moving `*self` out of the
// box hands it the same field by-value, so the muxer's own
// by-value `finish()` runs correctly.
let mut s: Self = *self;
<Self as crate::pes::Stream>::finish(&mut s)
}
fn info(&self) -> &crate::disc::DiscTitle {
<Self as crate::pes::Stream>::info(self)
}
}
// ── MKV header parsing (read side) ────────────────────────────
/// Returns (DiscTitle, codec_privates: Vec<(track_number, codec_private_bytes)>)
+24
View File
@@ -109,6 +109,30 @@ impl crate::pes::Stream for NetworkStream {
}
}
/// FrameSink sibling to the deprecated Stream impl; both coexist during the
/// 0.18 deprecation window. Caller may pick either at the trait-object
/// boundary — `Box<dyn Stream>` (deprecated) or `Box<dyn FrameSink>` (new).
/// Use `NetworkStream::connect(addr).meta(title)` to construct the write
/// half; calling `FrameSink::write` on `NetworkStream::listen(addr)` returns
/// `StreamReadOnly`.
#[allow(deprecated)] // delegating to deprecated Stream during the 0.18 deprecation window so callers don't see the deprecation twice.
impl crate::pes::FrameSink for NetworkStream {
fn write(&mut self, frame: &crate::pes::PesFrame) -> io::Result<()> {
<Self as crate::pes::Stream>::write(self, frame)
}
fn finish(self: Box<Self>) -> io::Result<()> {
// Why: Stream::finish takes &mut self, FrameSink::finish takes Box<Self>.
// Re-borrow inside the box, call Stream::finish, drop the box.
let mut s: Self = *self;
<Self as crate::pes::Stream>::finish(&mut s)
}
fn info(&self) -> &DiscTitle {
<Self as crate::pes::Stream>::info(self)
}
}
// NetworkStream is PES-only — no IOStream/Read/Write byte interface.
#[cfg(test)]
+52
View File
@@ -31,3 +31,55 @@ impl crate::pes::Stream for NullStream {
&self.disc_title
}
}
/// FrameSink sibling to the deprecated Stream impl; both coexist during the
/// 0.18 deprecation window. Caller may pick either at the trait-object
/// boundary — `Box<dyn Stream>` (deprecated) or `Box<dyn FrameSink>` (new).
/// The deprecation-window callers eventually migrate; this impl exists so
/// new callers can target `FrameSink` without waiting for the rest of the
/// migration to complete.
#[allow(deprecated)] // delegating to deprecated Stream during the 0.18 deprecation window so callers don't see the deprecation twice.
impl crate::pes::FrameSink for NullStream {
fn write(&mut self, frame: &crate::pes::PesFrame) -> io::Result<()> {
<Self as crate::pes::Stream>::write(self, frame)
}
fn finish(self: Box<Self>) -> io::Result<()> {
// Why: Stream::finish takes &mut self, FrameSink::finish takes Box<Self>.
// Re-borrow inside the box, call Stream::finish, drop the box.
let mut s: Self = *self;
<Self as crate::pes::Stream>::finish(&mut s)
}
fn info(&self) -> &DiscTitle {
<Self as crate::pes::Stream>::info(self)
}
}
#[cfg(test)]
mod tests {
use super::*;
use crate::pes::FrameSink;
/// Smallest credible witness that the new FrameSink impl on a concrete
/// `mux/*` sink works through the trait object: build a boxed
/// `dyn FrameSink`, write a frame, finish it. The trait-bridge correctness
/// is what's being verified — not NullStream-specific behaviour.
#[test]
fn frame_sink_via_dyn_object_writes_and_finishes() {
let title = DiscTitle::empty();
let mut sink: Box<dyn FrameSink> = Box::new(NullStream::new(&title));
let frame = crate::pes::PesFrame {
track: 0,
pts: 0,
keyframe: true,
data: vec![0x01, 0x02, 0x03],
};
sink.write(&frame).unwrap();
// info() routes through the trait object.
let _ = sink.info();
// finish() consumes the Box<Self> — must compile and run.
sink.finish().unwrap();
}
}
+2 -1
View File
@@ -249,7 +249,8 @@ pub fn output(
// staging doesn't hit the dirty-page burst pathology that
// sweep already side-steps. BufWriter sits on top to coalesce
// mux's many small EBML element writes.
let writer: Box<dyn super::WriteSeek> = Box::new(std::io::BufWriter::with_capacity(
let writer: Box<dyn super::WriteSeek + Send> =
Box::new(std::io::BufWriter::with_capacity(
IO_BUF_SIZE,
crate::io::WritebackFile::create(path)?,
));
+23
View File
@@ -105,3 +105,26 @@ impl crate::pes::Stream for StdioStream {
self.header_read || self.writer.is_some()
}
}
/// FrameSink sibling to the deprecated Stream impl; both coexist during the
/// 0.18 deprecation window. Caller may pick either at the trait-object
/// boundary — `Box<dyn Stream>` (deprecated) or `Box<dyn FrameSink>` (new).
/// Use `StdioStream::output(title)` to construct the write half; calling
/// `FrameSink::write` on a `StdioStream::input()` returns `StreamReadOnly`.
#[allow(deprecated)] // delegating to deprecated Stream during the 0.18 deprecation window so callers don't see the deprecation twice.
impl crate::pes::FrameSink for StdioStream {
fn write(&mut self, frame: &crate::pes::PesFrame) -> io::Result<()> {
<Self as crate::pes::Stream>::write(self, frame)
}
fn finish(self: Box<Self>) -> io::Result<()> {
// Why: Stream::finish takes &mut self, FrameSink::finish takes Box<Self>.
// Re-borrow inside the box, call Stream::finish, drop the box.
let mut s: Self = *self;
<Self as crate::pes::Stream>::finish(&mut s)
}
fn info(&self) -> &DiscTitle {
<Self as crate::pes::Stream>::info(self)
}
}
+20 -19
View File
@@ -307,12 +307,12 @@ fn m2ts_stream_write_read() {
keyframe: i == 0,
data: vec![i; 100],
};
stream.write(&frame).unwrap();
PesStream::write(&mut stream, &frame).unwrap();
}
stream.finish().unwrap();
PesStream::finish(&mut stream).unwrap();
// Verify the info is correct
let info = stream.info();
let info = PesStream::info(&stream);
assert_eq!(info.streams.len(), 4);
assert_eq!(info.duration_secs, 7200.0);
}
@@ -361,7 +361,7 @@ fn m2ts_read_returns_error_on_write_stream() {
let dt = sample_disc_title();
let output = Cursor::new(Vec::new());
let mut stream = M2tsStream::create(output, &dt).unwrap();
assert!(stream.read().is_err());
assert!(PesStream::read(&mut stream).is_err());
}
// ── DiscTitle::empty ──────────────────────────────────────────
@@ -583,7 +583,7 @@ fn meta_all_stream_types() {
#[test]
fn mkvstream_write_finish() {
let dt = sample_disc_title();
let writer: Box<dyn libfreemkv::mux::WriteSeek> = Box::new(Cursor::new(Vec::new()));
let writer: Box<dyn libfreemkv::mux::WriteSeek + Send> = Box::new(Cursor::new(Vec::new()));
let mut stream = MkvStream::create(writer, &dt).unwrap();
// Write some fake PES frames (they won't produce valid MKV content
@@ -595,20 +595,20 @@ fn mkvstream_write_finish() {
keyframe: i == 0,
data: vec![i; 100],
};
stream.write(&frame).unwrap();
PesStream::write(&mut stream, &frame).unwrap();
}
// finish should not panic even without valid codec data
stream.finish().unwrap();
PesStream::finish(&mut stream).unwrap();
}
#[test]
fn mkvstream_meta_sets_title() {
let dt = sample_disc_title();
let writer: Box<dyn libfreemkv::mux::WriteSeek> = Box::new(Cursor::new(Vec::new()));
let writer: Box<dyn libfreemkv::mux::WriteSeek + Send> = Box::new(Cursor::new(Vec::new()));
let stream = MkvStream::create(writer, &dt).unwrap();
let info = stream.info();
let info = PesStream::info(&stream);
assert_eq!(info.playlist, "Test Movie");
assert_eq!(info.duration_secs, 7200.0);
assert_eq!(info.streams.len(), 4);
@@ -645,7 +645,7 @@ fn mkvstream_roundtrip_bdts() {
codec_privates: Vec::new(),
};
let writer: Box<dyn libfreemkv::mux::WriteSeek> = Box::new(Cursor::new(Vec::new()));
let writer: Box<dyn libfreemkv::mux::WriteSeek + Send> = Box::new(Cursor::new(Vec::new()));
let mut stream = MkvStream::create(writer, &dt).unwrap();
// Write PES frames targeting the audio track
@@ -656,13 +656,13 @@ fn mkvstream_roundtrip_bdts() {
keyframe: true,
data: vec![i; 100],
};
stream.write(&frame).unwrap();
PesStream::write(&mut stream, &frame).unwrap();
}
stream.finish().unwrap();
PesStream::finish(&mut stream).unwrap();
// Verify the info is correct
let info = stream.info();
let info = PesStream::info(&stream);
assert_eq!(info.streams.len(), 1);
assert_eq!(info.playlist, "Audio Only");
}
@@ -729,10 +729,10 @@ fn mkvstream_meta_preserves_all_streams() {
codec_privates: Vec::new(),
};
let writer: Box<dyn libfreemkv::mux::WriteSeek> = Box::new(Cursor::new(Vec::new()));
let writer: Box<dyn libfreemkv::mux::WriteSeek + Send> = Box::new(Cursor::new(Vec::new()));
let stream = MkvStream::create(writer, &dt).unwrap();
let info = stream.info();
let info = PesStream::info(&stream);
assert_eq!(info.streams.len(), 5, "all 5 streams should be preserved");
assert_eq!(info.playlist, "Stream Test");
assert_eq!(info.duration_secs, 3600.0);
@@ -837,7 +837,8 @@ fn mkvstream_e2e_h264_produces_valid_mkv() {
}
}
let writer: Box<dyn libfreemkv::mux::WriteSeek> = Box::new(SharedWriter(output2.clone()));
let writer: Box<dyn libfreemkv::mux::WriteSeek + Send> =
Box::new(SharedWriter(output2.clone()));
let mut stream2 = MkvStream::create(writer, &dt).unwrap();
// Write the ES data (SPS+PPS+IDR) as a keyframe PES frame.
@@ -847,7 +848,7 @@ fn mkvstream_e2e_h264_produces_valid_mkv() {
keyframe: true,
data: es_data,
};
stream2.write(&frame1).unwrap();
PesStream::write(&mut stream2, &frame1).unwrap();
// Write a second non-IDR frame
let frame2 = libfreemkv::pes::PesFrame {
@@ -856,8 +857,8 @@ fn mkvstream_e2e_h264_produces_valid_mkv() {
keyframe: false,
data: es_data2,
};
stream2.write(&frame2).unwrap();
stream2.finish().unwrap();
PesStream::write(&mut stream2, &frame2).unwrap();
PesStream::finish(&mut stream2).unwrap();
let data = output2.lock().unwrap().clone().into_inner();