0.18 round 2: add FrameSink impls to concrete mux sinks

Per-impl migration of MkvStream / M2tsStream / NetworkStream /
NullStream / StdioStream from the deprecated pes::Stream trait
to the typed pes::FrameSink trait. Both impls coexist during
the 0.18 deprecation window — the existing Stream impls are
unchanged.

The FrameSink::finish signature differs (Box<Self> vs &mut self),
which is why this couldn't be a blanket impl. Each migration
re-borrows the box and delegates to the underlying Stream::finish
body.

FrameSink: Send forced two struct fields (M2tsStream's boxed
Write/Read, MkvStream's boxed WriteSeek/Read) to gain `+ Send`
bounds — minimum surface needed to make the Send-bounded trait
impl-able. mux::resolve::output's local Box<dyn WriteSeek>
construction picks up the same `+ Send`. tests/streams.rs's
shared `stream.write/.finish/.info/.read` calls were
disambiguated to `PesStream::*` to resolve the now-multiple
candidates from coexisting trait impls.

Caller migration (mux::resolve::output return type, autorip,
CLI) is a later slice. This commit only adds new impls; nothing
removed.

See freemkv-private/memory/0_18_redesign.md.

Single contributor: MattJackson.
This commit is contained in:
2026-05-09 09:49:18 -07:00
parent 415293ecd0
commit 760e40b737
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 { enum Mode {
Write { Write {
muxer: super::tsmux::TsMuxer<Box<dyn Write>>, muxer: super::tsmux::TsMuxer<Box<dyn Write + Send>>,
}, },
Read { Read {
reader: Box<dyn Read>, reader: Box<dyn Read + Send>,
}, },
} }
@@ -57,7 +57,7 @@ pub struct M2tsStream {
impl M2tsStream { impl M2tsStream {
/// Create for writing PES frames → BD-TS output. /// Create for writing PES frames → BD-TS output.
/// Writes FMKV metadata header, then muxes PES frames into BD transport stream. /// 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 // Write FMKV metadata header
if !title.streams.is_empty() { if !title.streams.is_empty() {
let m = meta::M2tsMeta::from_title(title); let m = meta::M2tsMeta::from_title(title);
@@ -72,7 +72,7 @@ impl M2tsStream {
DiscStream::Subtitle(s) => s.pid, DiscStream::Subtitle(s) => s.pid,
}) })
.collect(); .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); let mut muxer = super::tsmux::TsMuxer::new(boxed, &pids);
for (i, cp) in title.codec_privates.iter().enumerate() { for (i, cp) in title.codec_privates.iter().enumerate() {
if let Some(data) = cp { if let Some(data) = cp {
@@ -111,7 +111,7 @@ impl M2tsStream {
/// Open an M2TS stream for reading. Takes any Read source — file, pipe, socket. /// 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. /// 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 // Read first chunk — enough for FMKV header or PMT scan
let mut head = vec![0u8; SCAN_SIZE]; let mut head = vec![0u8; SCAN_SIZE];
let head_len = read_fill(&mut reader, &mut head)?; 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); let (pids, parsers, pid_to_track) = Self::setup_pes(&title.streams);
// Chain: remaining head bytes + rest of reader // Chain: remaining head bytes + rest of reader
let remaining_head = &head[header_end..]; 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)); Box::new(io::Cursor::new(remaining_head.to_vec()).chain(reader));
return Ok(Self { return Ok(Self {
disc_title: title.clone(), disc_title: title.clone(),
@@ -150,7 +150,7 @@ impl M2tsStream {
let (pids, parsers, pid_to_track) = Self::setup_pes(&streams); let (pids, parsers, pid_to_track) = Self::setup_pes(&streams);
// Chain: full head (it's all TS data) + rest of reader // 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 { Ok(Self {
disc_title: DiscTitle { disc_title: DiscTitle {
@@ -284,3 +284,27 @@ impl crate::pes::Stream for M2tsStream {
true 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}; use std::io::{self, Read};
struct ReadState { struct ReadState {
reader: Box<dyn Read>, reader: Box<dyn Read + Send>,
cluster_ts_ms: i64, cluster_ts_ms: i64,
/// Codec private data per track (track_number, hvcC/avcC bytes). /// Codec private data per track (track_number, hvcC/avcC bytes).
codec_privates: Vec<(u16, Vec<u8>)>, codec_privates: Vec<(u16, Vec<u8>)>,
@@ -26,7 +26,7 @@ struct ReadState {
enum Mode { enum Mode {
Write { Write {
muxer: Option<MkvMuxer<Box<dyn WriteSeek>>>, muxer: Option<MkvMuxer<Box<dyn WriteSeek + Send>>>,
}, },
Read(ReadState), Read(ReadState),
} }
@@ -40,7 +40,7 @@ pub struct MkvStream {
impl MkvStream { impl MkvStream {
/// Create for writing PES frames → MKV container. /// Create for writing PES frames → MKV container.
/// Codec privates come from title.codec_privates (populated by input stream). /// 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 tracks = Vec::new();
let mut has_default_video = false; let mut has_default_video = false;
let mut has_default_audio = false; let mut has_default_audio = false;
@@ -79,7 +79,7 @@ impl MkvStream {
} }
/// Open an MKV file for reading → PES frames. /// 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)?; let (disc_title, codec_privates) = parse_mkv_header(&mut reader)?;
Ok(Self { Ok(Self {
disc_title, 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) ──────────────────────────── // ── MKV header parsing (read side) ────────────────────────────
/// Returns (DiscTitle, codec_privates: Vec<(track_number, codec_private_bytes)>) /// 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. // NetworkStream is PES-only — no IOStream/Read/Write byte interface.
#[cfg(test)] #[cfg(test)]
+52
View File
@@ -31,3 +31,55 @@ impl crate::pes::Stream for NullStream {
&self.disc_title &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 // staging doesn't hit the dirty-page burst pathology that
// sweep already side-steps. BufWriter sits on top to coalesce // sweep already side-steps. BufWriter sits on top to coalesce
// mux's many small EBML element writes. // 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, IO_BUF_SIZE,
crate::io::WritebackFile::create(path)?, 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() 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, keyframe: i == 0,
data: vec![i; 100], 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 // Verify the info is correct
let info = stream.info(); let info = PesStream::info(&stream);
assert_eq!(info.streams.len(), 4); assert_eq!(info.streams.len(), 4);
assert_eq!(info.duration_secs, 7200.0); 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 dt = sample_disc_title();
let output = Cursor::new(Vec::new()); let output = Cursor::new(Vec::new());
let mut stream = M2tsStream::create(output, &dt).unwrap(); let mut stream = M2tsStream::create(output, &dt).unwrap();
assert!(stream.read().is_err()); assert!(PesStream::read(&mut stream).is_err());
} }
// ── DiscTitle::empty ────────────────────────────────────────── // ── DiscTitle::empty ──────────────────────────────────────────
@@ -583,7 +583,7 @@ fn meta_all_stream_types() {
#[test] #[test]
fn mkvstream_write_finish() { fn mkvstream_write_finish() {
let dt = sample_disc_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 mut stream = MkvStream::create(writer, &dt).unwrap(); let mut stream = MkvStream::create(writer, &dt).unwrap();
// Write some fake PES frames (they won't produce valid MKV content // Write some fake PES frames (they won't produce valid MKV content
@@ -595,20 +595,20 @@ fn mkvstream_write_finish() {
keyframe: i == 0, keyframe: i == 0,
data: vec![i; 100], data: vec![i; 100],
}; };
stream.write(&frame).unwrap(); PesStream::write(&mut stream, &frame).unwrap();
} }
// finish should not panic even without valid codec data // finish should not panic even without valid codec data
stream.finish().unwrap(); PesStream::finish(&mut stream).unwrap();
} }
#[test] #[test]
fn mkvstream_meta_sets_title() { fn mkvstream_meta_sets_title() {
let dt = sample_disc_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 stream = MkvStream::create(writer, &dt).unwrap();
let info = stream.info(); let info = PesStream::info(&stream);
assert_eq!(info.playlist, "Test Movie"); assert_eq!(info.playlist, "Test Movie");
assert_eq!(info.duration_secs, 7200.0); assert_eq!(info.duration_secs, 7200.0);
assert_eq!(info.streams.len(), 4); assert_eq!(info.streams.len(), 4);
@@ -645,7 +645,7 @@ fn mkvstream_roundtrip_bdts() {
codec_privates: Vec::new(), 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(); let mut stream = MkvStream::create(writer, &dt).unwrap();
// Write PES frames targeting the audio track // Write PES frames targeting the audio track
@@ -656,13 +656,13 @@ fn mkvstream_roundtrip_bdts() {
keyframe: true, keyframe: true,
data: vec![i; 100], 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 // Verify the info is correct
let info = stream.info(); let info = PesStream::info(&stream);
assert_eq!(info.streams.len(), 1); assert_eq!(info.streams.len(), 1);
assert_eq!(info.playlist, "Audio Only"); assert_eq!(info.playlist, "Audio Only");
} }
@@ -729,10 +729,10 @@ fn mkvstream_meta_preserves_all_streams() {
codec_privates: Vec::new(), 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 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.streams.len(), 5, "all 5 streams should be preserved");
assert_eq!(info.playlist, "Stream Test"); assert_eq!(info.playlist, "Stream Test");
assert_eq!(info.duration_secs, 3600.0); 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(); let mut stream2 = MkvStream::create(writer, &dt).unwrap();
// Write the ES data (SPS+PPS+IDR) as a keyframe PES frame. // 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, keyframe: true,
data: es_data, data: es_data,
}; };
stream2.write(&frame1).unwrap(); PesStream::write(&mut stream2, &frame1).unwrap();
// Write a second non-IDR frame // Write a second non-IDR frame
let frame2 = libfreemkv::pes::PesFrame { let frame2 = libfreemkv::pes::PesFrame {
@@ -856,8 +857,8 @@ fn mkvstream_e2e_h264_produces_valid_mkv() {
keyframe: false, keyframe: false,
data: es_data2, data: es_data2,
}; };
stream2.write(&frame2).unwrap(); PesStream::write(&mut stream2, &frame2).unwrap();
stream2.finish().unwrap(); PesStream::finish(&mut stream2).unwrap();
let data = output2.lock().unwrap().clone().into_inner(); let data = output2.lock().unwrap().clone().into_inner();