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 (internal)/memory/0_18_redesign.md.

Single contributor: MattJackson.

src/mux/m2ts.rs

@@ -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)
+    }
+}

src/mux/mkvstream.rs

@@ -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)>)

src/mux/network.rs

@@ -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)]

src/mux/null.rs

@@ -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();
+    }
+}

src/mux/resolve.rs

@@ -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)?,
                 ));

src/mux/stdio.rs

@@ -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)
+    }
+}

tests/streams.rs

@@ -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();