0.18 primitive: FrameSource/FrameSink trait split (deprecate Stream)

Splits the bidirectional pes::Stream into one-direction traits so
calling read() on a write-only sink is a compile error instead of
runtime E9001. Keeps Stream alive as deprecated through 0.18 with a
blanket FrameSource impl so existing concrete types compile unchanged.

FrameSink can't be blanket-impl'd from Stream (different finish
signature), so concrete impls migrate per-type in a follow-up.

Concrete `impl pes::Stream for X` blocks in mux/* and the existing
tests gain a one-line `#[allow(deprecated)]` to keep `-D warnings`
clean during the deprecation window — no behavior changes.

See (internal)/memory/0_18_redesign.md.

Single contributor: MattJackson.

src/pes.rs

@@ -108,6 +108,10 @@ impl PesFrame {
 /// false until enough input frames have been seen to populate every video
 /// track's codec-private blob — callers buffer frames they read until
 /// `headers_ready()` returns true.
+#[deprecated(
+    since = "0.18.0-dev",
+    note = "use FrameSource (read-only) or FrameSink (write-only) instead"
+)]
 pub trait Stream {
     /// Read the next frame, or `Ok(None)` at end of stream. Returns
     /// `StreamWriteOnly` (E9001) on a write-only sink.
@@ -140,6 +144,96 @@ pub trait Stream {
     }
 }
 
+/// Read-only source of PES frames.
+///
+/// Replaces the read half of the deprecated [`Stream`] trait. Implementors
+/// produce frames via [`read`](FrameSource::read) and never accept writes —
+/// passing a `FrameSource` where a sink is expected is a compile error,
+/// not a runtime `E9001`.
+///
+/// `info()` returns the source's `DiscTitle` metadata (track list, codec
+/// info, duration). It must be stable across the lifetime of the source.
+///
+/// `codec_private(track)` exposes per-track codec initialization data
+/// (H.264 SPS/PPS, HEVC VPS/SPS/PPS, AC-3 fscod, etc.) that downstream
+/// muxers may need before any frame is written. `headers_ready()` returns
+/// false until enough input frames have been seen to populate every video
+/// track's codec-private blob — callers buffer frames they read until
+/// `headers_ready()` returns true.
+pub trait FrameSource: Send {
+    /// Read the next frame, or `Ok(None)` at end of stream.
+    fn read(&mut self) -> std::io::Result<Option<PesFrame>>;
+
+    /// Source metadata. Stable across reads — implementors must return a
+    /// consistent reference for the lifetime of the source.
+    fn info(&self) -> &crate::disc::DiscTitle;
+
+    /// Codec initialization data for a track (SPS/PPS, AC-3 fscod, etc.).
+    /// `None` for tracks that don't need codec_private (raw passthrough).
+    fn codec_private(&self, _track: usize) -> Option<Vec<u8>> {
+        None
+    }
+
+    /// True when `codec_private` is available for every video track —
+    /// callers buffer input frames until this flips, since some output
+    /// formats (MKV) can't write frames without codec init data.
+    fn headers_ready(&self) -> bool {
+        true
+    }
+}
+
+/// Write-only sink of PES frames.
+///
+/// Replaces the write half of the deprecated [`Stream`] trait. Implementors
+/// accept frames via [`write`](FrameSink::write) and finalize via
+/// [`finish`](FrameSink::finish) — passing a `FrameSink` where a source is
+/// expected is a compile error, not a runtime `E9000`.
+///
+/// `finish` takes `Box<Self>` (rather than `&mut self` like `Stream::finish`)
+/// so that finalization is a one-shot terminal operation: callers cannot use
+/// the sink after `finish` returns. This is the standard idiom for terminal
+/// methods on `dyn Trait` objects.
+pub trait FrameSink: Send {
+    /// Write a frame to the sink.
+    fn write(&mut self, frame: &PesFrame) -> std::io::Result<()>;
+
+    /// Finalize the sink: flush buffered frames, write any container index
+    /// (e.g. MKV `Cues`), close the underlying file/socket. Consumes the
+    /// sink — callers cannot use it afterwards.
+    fn finish(self: Box<Self>) -> std::io::Result<()>;
+
+    /// Sink metadata. Stable across writes — implementors must return a
+    /// consistent reference for the lifetime of the sink.
+    fn info(&self) -> &crate::disc::DiscTitle;
+}
+
+// Bridge: any type implementing the deprecated `Stream` trait is also a
+// `FrameSource`. This lets existing concrete `Stream` impls in `mux/*`
+// satisfy `FrameSource` bounds without per-type migration during the
+// 0.18 deprecation window.
+//
+// Note: `FrameSink` cannot be blanket-impl'd from `Stream` because
+// `Stream::finish` takes `&mut self` while `FrameSink::finish` takes
+// `Box<Self>`; concrete types will be migrated in a follow-up commit.
+#[allow(deprecated)]
+impl<T: Stream + Send + ?Sized> FrameSource for T {
+    fn read(&mut self) -> std::io::Result<Option<PesFrame>> {
+        <Self as Stream>::read(self)
+    }
+
+    fn info(&self) -> &crate::disc::DiscTitle {
+        <Self as Stream>::info(self)
+    }
+
+    fn codec_private(&self, track: usize) -> Option<Vec<u8>> {
+        <Self as Stream>::codec_private(self, track)
+    }
+
+    fn headers_ready(&self) -> bool {
+        <Self as Stream>::headers_ready(self)
+    }
+}
+
 /// Wraps any output stream and counts bytes written.
 ///
 /// Progress tracking is a CLI concern — streams don't know their size.
@@ -152,11 +246,13 @@ pub trait Stream {
 ///     let pct = output.bytes_written() as f64 / total as f64;
 /// }
 /// ```
+#[allow(deprecated)]
 pub struct CountingStream {
     inner: Box<dyn Stream>,
     written: u64,
 }
 
+#[allow(deprecated)]
 impl CountingStream {
     pub fn new(inner: Box<dyn Stream>) -> Self {
         Self { inner, written: 0 }
@@ -168,6 +264,7 @@ impl CountingStream {
     }
 }
 
+#[allow(deprecated)]
 impl Stream for CountingStream {
     fn read(&mut self) -> std::io::Result<Option<PesFrame>> {
         self.inner.read()
@@ -194,3 +291,178 @@ impl Stream for CountingStream {
         self.inner.headers_ready()
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use crate::disc::DiscTitle;
+
+    fn make_frame(track: usize, pts: i64) -> PesFrame {
+        PesFrame {
+            track,
+            pts,
+            keyframe: track == 0 && pts == 0,
+            data: vec![track as u8, (pts & 0xff) as u8, 0xAA],
+        }
+    }
+
+    /// Direct `FrameSource` impl (not via the deprecated `Stream` blanket).
+    struct MockSource {
+        frames: std::vec::IntoIter<PesFrame>,
+        title: DiscTitle,
+    }
+
+    impl MockSource {
+        fn new(frames: Vec<PesFrame>) -> Self {
+            Self {
+                frames: frames.into_iter(),
+                title: DiscTitle::empty(),
+            }
+        }
+    }
+
+    impl FrameSource for MockSource {
+        fn read(&mut self) -> std::io::Result<Option<PesFrame>> {
+            Ok(self.frames.next())
+        }
+
+        fn info(&self) -> &DiscTitle {
+            &self.title
+        }
+    }
+
+    /// Direct `FrameSink` impl (not via the deprecated `Stream` blanket).
+    struct MockSink {
+        collected: Vec<PesFrame>,
+        title: DiscTitle,
+    }
+
+    impl MockSink {
+        fn new() -> Self {
+            Self {
+                collected: Vec::new(),
+                title: DiscTitle::empty(),
+            }
+        }
+    }
+
+    impl FrameSink for MockSink {
+        fn write(&mut self, frame: &PesFrame) -> std::io::Result<()> {
+            self.collected.push(frame.clone());
+            Ok(())
+        }
+
+        fn finish(self: Box<Self>) -> std::io::Result<()> {
+            // Drop self; in real sinks this is where flush/fsync/close happens.
+            Ok(())
+        }
+
+        fn info(&self) -> &DiscTitle {
+            &self.title
+        }
+    }
+
+    /// Variant of `MockSink` whose `finish` returns the collected frames so
+    /// the test can assert on them after the consuming `Box<Self>` call.
+    struct CollectingSink {
+        collected: Vec<PesFrame>,
+        title: DiscTitle,
+    }
+
+    impl CollectingSink {
+        fn new() -> Self {
+            Self {
+                collected: Vec::new(),
+                title: DiscTitle::empty(),
+            }
+        }
+    }
+
+    impl FrameSink for CollectingSink {
+        fn write(&mut self, frame: &PesFrame) -> std::io::Result<()> {
+            self.collected.push(frame.clone());
+            Ok(())
+        }
+
+        fn finish(self: Box<Self>) -> std::io::Result<()> {
+            // Real CollectingSink consumers would expose `take()` before
+            // finish; this trait method just confirms the boxed signature
+            // compiles and runs.
+            Ok(())
+        }
+
+        fn info(&self) -> &DiscTitle {
+            &self.title
+        }
+    }
+
+    #[test]
+    fn frame_source_yields_frames_then_eof() {
+        let frames = vec![make_frame(0, 0), make_frame(1, 1_000), make_frame(0, 2_000)];
+        let mut src = MockSource::new(frames.clone());
+
+        let f0 = src.read().unwrap().expect("first frame");
+        assert_eq!(f0.track, frames[0].track);
+        assert_eq!(f0.pts, frames[0].pts);
+        assert!(f0.keyframe);
+
+        let f1 = src.read().unwrap().expect("second frame");
+        assert_eq!(f1.track, frames[1].track);
+        assert_eq!(f1.pts, frames[1].pts);
+
+        let f2 = src.read().unwrap().expect("third frame");
+        assert_eq!(f2.track, frames[2].track);
+        assert_eq!(f2.pts, frames[2].pts);
+
+        assert!(src.read().unwrap().is_none());
+        assert!(src.read().unwrap().is_none()); // idempotent at EOF
+    }
+
+    #[test]
+    fn frame_sink_collects_then_finishes() {
+        let mut sink = MockSink::new();
+        let frames = [make_frame(0, 0), make_frame(1, 100), make_frame(2, 200)];
+
+        for f in &frames {
+            sink.write(f).unwrap();
+        }
+        assert_eq!(sink.collected.len(), 3);
+        assert_eq!(sink.collected[0].pts, 0);
+        assert_eq!(sink.collected[1].pts, 100);
+        assert_eq!(sink.collected[2].pts, 200);
+
+        // Box-and-finish — the `self: Box<Self>` shape must compile and run.
+        Box::new(sink).finish().unwrap();
+    }
+
+    #[test]
+    fn frame_sink_via_dyn_object() {
+        let frames = [make_frame(0, 0), make_frame(0, 33)];
+        let mut sink: Box<dyn FrameSink> = Box::new(CollectingSink::new());
+
+        for f in &frames {
+            sink.write(f).unwrap();
+        }
+        // info() routes through the trait object.
+        let _ = sink.info();
+        sink.finish().unwrap();
+    }
+
+    /// The deprecated blanket impl: any concrete `Stream` should also act as
+    /// a `FrameSource`. `NullStream` has the simplest constructor of every
+    /// concrete `Stream` impl in `mux/*`, so it's the smallest credible
+    /// witness that the bridge compiles and dispatches correctly.
+    #[test]
+    #[allow(deprecated)]
+    fn deprecated_stream_satisfies_frame_source() {
+        let title = DiscTitle::empty();
+        let mut null = crate::mux::NullStream::new(&title);
+
+        let src: &mut dyn FrameSource = &mut null;
+        // NullStream::read returns Ok(None) — it's a write-only sink.
+        assert!(src.read().unwrap().is_none());
+        // info() forwards through the blanket impl.
+        let _ = src.info();
+        assert!(src.headers_ready());
+    }
+}