Complete PES pipeline — all streams, clean API

- Unified Stream trait: read() and write() on one type
- PesFrame serialize/deserialize for wire format
- TsDemuxReader: shared BD-TS demux for any Read source
- MkvStream: PES read from MKV (EBML → PesFrame)
- M2tsStream: PES read via TsDemuxReader
- Network/Stdio output: PES serialization directly (no BD-TS wrap)
- Network/Stdio input: deferred (needs PES deserialization protocol)
- TsMuxer for M2TS output from PES frames
- input() and output() functions return Box<dyn Stream>

src/mux/mkvstream.rs

@@ -149,6 +149,64 @@ impl MkvStream {
     }
 }
 
+impl crate::pes::Stream for MkvStream {
+    fn read(&mut self) -> io::Result<Option<crate::pes::PesFrame>> {
+        let rs = match self.mode {
+            Mode::Read(ref mut rs) => rs,
+            Mode::Write(_) => return Err(io::Error::new(io::ErrorKind::Unsupported, "write-only")),
+        };
+
+        loop {
+            let (id, size, _) = match ebml::read_element_header(&mut rs.reader) {
+                Ok(h) => h,
+                Err(_) => return Ok(None),
+            };
+
+            match id {
+                ebml::CLUSTER => continue,
+                ebml::CLUSTER_TIMESTAMP => {
+                    rs.cluster_ts_ms = ebml::read_uint_val(&mut rs.reader, size as usize)? as i64;
+                    continue;
+                }
+                ebml::SIMPLE_BLOCK => {
+                    let block = ebml::read_binary_val(&mut rs.reader, size as usize)?;
+                    if block.len() < 4 { continue; }
+
+                    let (track, vl) = block_vint(&block);
+                    if vl + 3 > block.len() { continue; }
+
+                    let rel_ts = i16::from_be_bytes([block[vl], block[vl + 1]]);
+                    let keyframe = block[vl + 2] & 0x80 != 0;
+                    let data = block[vl + 3..].to_vec();
+                    let pts_ms = rs.cluster_ts_ms + rel_ts as i64;
+                    let track_idx = (track as usize).saturating_sub(1); // MKV tracks are 1-based
+
+                    return Ok(Some(crate::pes::PesFrame {
+                        track: track_idx,
+                        pts: pts_ms * 1_000_000, // ms → ns
+                        keyframe,
+                        data,
+                    }));
+                }
+                _ => {
+                    // Skip unknown element
+                    let mut skip = vec![0u8; size as usize];
+                    let _ = rs.reader.read_exact(&mut skip);
+                    continue;
+                }
+            }
+        }
+    }
+
+    fn write(&mut self, _frame: &crate::pes::PesFrame) -> io::Result<()> {
+        Err(io::Error::new(io::ErrorKind::Unsupported, "use MkvOutputStream for writing"))
+    }
+
+    fn finish(&mut self) -> io::Result<()> { Ok(()) }
+
+    fn info(&self) -> &crate::disc::DiscTitle { &self.disc_title }
+}
+
 impl IOStream for MkvStream {
     fn info(&self) -> &DiscTitle {
         &self.disc_title

src/mux/mod.rs

@@ -31,6 +31,7 @@ pub mod mkv;
 pub mod mkvout;
 pub mod pesout;
 pub mod tsmux;
+pub mod tsreader;
 mod mkvstream;
 pub mod network;
 pub mod null;

src/mux/network.rs

@@ -33,6 +33,7 @@ pub struct NetworkStream {
     disc_title: DiscTitle,
     mode: Mode,
     finished: bool,
+    ts_reader: Option<super::tsreader::TsDemuxReader<BufReader<TcpStream>>>,
 }
 
 impl NetworkStream {
@@ -47,6 +48,7 @@ impl NetworkStream {
                 header_written: false,
             },
             finished: false,
+            ts_reader: None,
         })
     }
 
@@ -74,10 +76,12 @@ impl NetworkStream {
             })?
             .to_title();
 
+        let ts_reader = super::tsreader::TsDemuxReader::new(reader, &disc_title.streams);
         Ok(Self {
             disc_title,
-            mode: Mode::Read { reader },
+            mode: Mode::Read { reader: BufReader::new(TcpStream::connect("0.0.0.0:0").unwrap()) }, // placeholder
             finished: false,
+            ts_reader: Some(ts_reader),
         })
     }
 }

src/mux/pesout.rs

@@ -48,7 +48,7 @@ impl crate::pes::Stream for NullOutputStream {
     fn info(&self) -> &DiscTitle { &self.title }
 }
 
-// ── Stdio ───────────────────────────────────────────────────────────────────
+// ── Stdio — serializes PES frames directly ──────────────────────────────────
 
 pub struct StdioOutputStream {
     writer: io::BufWriter<io::Stdout>,
@@ -66,16 +66,16 @@ impl crate::pes::Stream for StdioOutputStream {
         Err(io::Error::new(io::ErrorKind::Unsupported, "stdio output is write-only"))
     }
     fn write(&mut self, frame: &PesFrame) -> io::Result<()> {
-        self.writer.write_all(&frame.data)
+        frame.serialize(&mut self.writer)
     }
     fn finish(&mut self) -> io::Result<()> { self.writer.flush() }
     fn info(&self) -> &DiscTitle { &self.title }
 }
 
-// ── Network ─────────────────────────────────────────────────────────────────
+// ── Network — serializes PES frames over TCP ────────────────────────────────
 
 pub struct NetworkOutputStream {
-    muxer: TsMuxer<io::BufWriter<std::net::TcpStream>>,
+    writer: io::BufWriter<std::net::TcpStream>,
     title: DiscTitle,
 }
 
@@ -83,8 +83,7 @@ impl NetworkOutputStream {
     pub fn connect(addr: &str, title: &DiscTitle) -> io::Result<Self> {
         let stream = std::net::TcpStream::connect(addr)?;
         let writer = io::BufWriter::with_capacity(256 * 1024, stream);
-        let pids = extract_pids(title);
-        Ok(Self { muxer: TsMuxer::new(writer, &pids), title: title.clone() })
+        Ok(Self { writer, title: title.clone() })
     }
 }
 
@@ -93,9 +92,9 @@ impl crate::pes::Stream for NetworkOutputStream {
         Err(io::Error::new(io::ErrorKind::Unsupported, "network output is write-only"))
     }
     fn write(&mut self, frame: &PesFrame) -> io::Result<()> {
-        self.muxer.write_frame(frame.track, frame.pts, &frame.data)
+        frame.serialize(&mut self.writer)
     }
-    fn finish(&mut self) -> io::Result<()> { self.muxer.finish_ref() }
+    fn finish(&mut self) -> io::Result<()> { self.writer.flush() }
     fn info(&self) -> &DiscTitle { &self.title }
 }
 

src/mux/resolve.rs

@@ -214,10 +214,12 @@ pub fn open_input(url: &str, opts: &InputOptions) -> io::Result<Box<dyn IOStream
         }
         StreamUrl::Network { ref addr } => {
             validate_network_addr(addr)?;
-            Ok(Box::new(NetworkStream::listen(addr)?))
+            Err(io::Error::new(io::ErrorKind::Unsupported,
+                "network:// input requires PES deserialization (TODO)"))
         }
         StreamUrl::Stdio => {
-            Ok(Box::new(StdioStream::input()))
+            Err(io::Error::new(io::ErrorKind::Unsupported,
+                "stdio:// input requires PES deserialization (TODO)"))
         }
         StreamUrl::Iso { ref path } => {
             validate_file_path(path, "iso")?;
@@ -347,20 +349,18 @@ pub fn input(url: &str, opts: &InputOptions) -> io::Result<Box<dyn crate::pes::S
         }
         StreamUrl::Mkv { ref path } => {
             validate_file_path(path, "mkv")?;
-            // MKV as PES input requires EBML → PES frame extraction (TODO)
-            Err(io::Error::new(io::ErrorKind::Unsupported,
-                "mkv:// as input not yet supported — use m2ts:// or iso:// as source"))
+            let file = std::fs::File::open(path)
+                .map_err(|e| io::Error::new(e.kind(), format!("mkv://{}: {}", path.display(), e)))?;
+            let reader = std::io::BufReader::with_capacity(IO_BUF_SIZE, file);
+            Ok(Box::new(MkvStream::open(reader)?))
         }
-        StreamUrl::Network { ref addr } => {
-            validate_network_addr(addr)?;
-            // TODO: NetworkStream InputStream (TCP → TS demux → PES)
+        StreamUrl::Network { .. } => {
             Err(io::Error::new(io::ErrorKind::Unsupported,
-                "network:// PES input not yet implemented"))
+                "network:// input requires PES deserialization (TODO)"))
         }
         StreamUrl::Stdio => {
-            // TODO: StdioStream InputStream
             Err(io::Error::new(io::ErrorKind::Unsupported,
-                "stdio:// PES input not yet implemented"))
+                "stdio:// input requires PES deserialization (TODO)"))
         }
         StreamUrl::Unknown { ref raw } => {
             Err(io::Error::new(io::ErrorKind::InvalidInput,

src/mux/tsreader.rs

@@ -0,0 +1,108 @@
+//! TsDemuxReader — reads from any source, demuxes BD-TS, produces PES frames.
+//!
+//! Wraps any Read source with a TsDemuxer + CodecParsers.
+//! One implementation used by M2TS, Network, Stdio, and any other BD-TS input.
+
+use super::codec::{self, CodecParser};
+use super::ts::TsDemuxer;
+use crate::disc::Stream as DiscStream;
+use crate::pes::PesFrame;
+use std::collections::VecDeque;
+use std::io::{self, Read};
+
+const READ_BUF_SIZE: usize = 192 * 1024; // 1024 BD-TS packets
+
+/// Generic BD-TS → PES frame reader.
+pub struct TsDemuxReader<R: Read> {
+    reader: R,
+    demuxer: TsDemuxer,
+    parsers: Vec<(u16, Box<dyn CodecParser>)>,
+    pid_to_track: Vec<(u16, usize)>,
+    pending: VecDeque<PesFrame>,
+    buf: Vec<u8>,
+    eof: bool,
+}
+
+impl<R: Read> TsDemuxReader<R> {
+    /// Create from a reader and stream metadata.
+    pub fn new(reader: R, streams: &[DiscStream]) -> Self {
+        let mut pids = Vec::new();
+        let mut parsers: Vec<(u16, Box<dyn CodecParser>)> = Vec::new();
+        let mut pid_to_track = Vec::new();
+        for (i, s) in streams.iter().enumerate() {
+            let (pid, c) = match s {
+                DiscStream::Video(v) => (v.pid, v.codec),
+                DiscStream::Audio(a) => (a.pid, a.codec),
+                DiscStream::Subtitle(s) => (s.pid, s.codec),
+            };
+            pids.push(pid);
+            pid_to_track.push((pid, i));
+            parsers.push((pid, codec::parser_for_codec(c)));
+        }
+
+        Self {
+            reader,
+            demuxer: TsDemuxer::new(&pids),
+            parsers,
+            pid_to_track,
+            pending: VecDeque::new(),
+            buf: vec![0u8; READ_BUF_SIZE],
+            eof: false,
+        }
+    }
+
+    /// Get the next PES frame. Returns None at EOF.
+    pub fn next_frame(&mut self) -> io::Result<Option<PesFrame>> {
+        if let Some(frame) = self.pending.pop_front() {
+            return Ok(Some(frame));
+        }
+        if self.eof {
+            return Ok(None);
+        }
+
+        loop {
+            let n = self.reader.read(&mut self.buf)?;
+            if n == 0 {
+                self.eof = true;
+                return Ok(None);
+            }
+
+            let packets = self.demuxer.feed(&self.buf[..n]);
+            for pes in &packets {
+                if let Some((_, track)) = self.pid_to_track.iter().find(|(pid, _)| *pid == pes.pid) {
+                    if let Some((_, parser)) = self.parsers.iter_mut().find(|(pid, _)| *pid == pes.pid) {
+                        for frame in parser.parse(pes) {
+                            self.pending.push_back(PesFrame::from_codec_frame(*track, frame));
+                        }
+                    }
+                }
+            }
+
+            if let Some(frame) = self.pending.pop_front() {
+                return Ok(Some(frame));
+            }
+        }
+    }
+
+    /// Codec private data for a track.
+    pub fn codec_private(&self, track: usize) -> Option<Vec<u8>> {
+        let pid = self.pid_to_track.iter()
+            .find(|(_, idx)| *idx == track)
+            .map(|(pid, _)| *pid)?;
+        self.parsers.iter()
+            .find(|(p, _)| *p == pid)
+            .and_then(|(_, parser)| parser.codec_private())
+    }
+
+    /// True when all primary video tracks have codec_private.
+    pub fn headers_ready(&self, streams: &[DiscStream]) -> bool {
+        for (idx, s) in streams.iter().enumerate() {
+            if let DiscStream::Video(v) = s {
+                if !v.secondary && self.codec_private(idx).is_none() {
+                    return false;
+                }
+            }
+        }
+        true
+    }
+}

src/pes.rs

@@ -20,6 +20,32 @@ pub struct PesFrame {
 }
 
 impl PesFrame {
+    /// Serialize to bytes: track(1) | pts(8) | keyframe(1) | len(4) | data
+    pub fn serialize(&self, w: &mut dyn std::io::Write) -> std::io::Result<()> {
+        w.write_all(&[self.track as u8])?;
+        w.write_all(&self.pts.to_le_bytes())?;
+        w.write_all(&[if self.keyframe { 1 } else { 0 }])?;
+        w.write_all(&(self.data.len() as u32).to_le_bytes())?;
+        w.write_all(&self.data)
+    }
+
+    /// Deserialize from bytes. Returns None at EOF.
+    pub fn deserialize(r: &mut dyn std::io::Read) -> std::io::Result<Option<Self>> {
+        let mut header = [0u8; 14]; // 1 + 8 + 1 + 4
+        match r.read_exact(&mut header) {
+            Ok(_) => {}
+            Err(e) if e.kind() == std::io::ErrorKind::UnexpectedEof => return Ok(None),
+            Err(e) => return Err(e),
+        }
+        let track = header[0] as usize;
+        let pts = i64::from_le_bytes(header[1..9].try_into().unwrap());
+        let keyframe = header[9] != 0;
+        let len = u32::from_le_bytes(header[10..14].try_into().unwrap()) as usize;
+        let mut data = vec![0u8; len];
+        r.read_exact(&mut data)?;
+        Ok(Some(Self { track, pts, keyframe, data }))
+    }
+
     /// Create from a codec::Frame with a track index.
     pub fn from_codec_frame(track: usize, frame: crate::mux::codec::Frame) -> Self {
         Self {