All streams complete — DVD PS demux, network/stdio PES, MKV input

- DiscStream: BD (TsDemuxer) or DVD (PsDemuxer) auto-detected
- NetworkStream: Stream impl with PES serialize/deserialize
- StdioStream: Stream impl with PES serialize/deserialize
- MkvStream: Stream read returns PesFrame from EBML blocks
- M2tsStream: Stream read via TsDemuxReader
- PesFrame: serialize/deserialize for wire format
- TsDemuxReader: shared BD-TS demux helper
- All inputs and outputs support PES

src/mux/disc.rs

@@ -43,8 +43,9 @@ pub struct DiscStream {
     pub errors: u64,
     eof: bool,
 
-    // PES output (for InputStream impl)
+    // PES output
-    demuxer: Option<super::ts::TsDemuxer>,
+    ts_demuxer: Option<super::ts::TsDemuxer>,
+    ps_demuxer: Option<super::ps::PsDemuxer>,
     parsers: Vec<(u16, Box<dyn super::codec::CodecParser>)>,
     pending_frames: std::collections::VecDeque<crate::pes::PesFrame>,
     pid_to_track: Vec<(u16, usize)>,
@@ -131,6 +132,12 @@ impl DiscStream {
         let mut stream = Self::title(drive, title);
         stream.decrypt_keys = keys;
 
+        // DVD: use program stream demuxer instead of transport stream
+        if disc.content_format == crate::disc::ContentFormat::MpegPs {
+            stream.ts_demuxer = None;
+            stream.ps_demuxer = Some(super::ps::PsDemuxer::new());
+        }
+
         Ok(DiscOpenResult { stream, disc })
     }
 
@@ -190,7 +197,8 @@ impl DiscStream {
             batch_sectors: max_batch,
             errors: 0,
             eof: false,
-            demuxer: if pids.is_empty() { None } else { Some(super::ts::TsDemuxer::new(&pids)) },
+            ts_demuxer: if pids.is_empty() { None } else { Some(super::ts::TsDemuxer::new(&pids)) },
+            ps_demuxer: None, // set by caller for DVD content
             parsers,
             pending_frames: std::collections::VecDeque::new(),
             pid_to_track,
@@ -367,25 +375,42 @@ impl crate::pes::Stream for DiscStream {
                 return Err(io::Error::other(e.to_string()));
             }
 
-            // Demux into PES packets, parse into frames
+            // Demux into packets, parse into frames
-            if let Some(ref mut demuxer) = self.demuxer {
+            if let Some(ref mut demuxer) = self.ts_demuxer {
+                // BD: transport stream demux
                 let packets = demuxer.feed(&self.read_buf[..bytes]);
                 for pes in &packets {
-                    if let Some((pid_idx, _)) = self.pid_to_track.iter().enumerate()
+                    if let Some((_, track)) = self.pid_to_track.iter().find(|(pid, _)| *pid == pes.pid) {
-                        .find(|(_, (pid, _))| *pid == pes.pid)
+                        if let Some((_, parser)) = self.parsers.iter_mut().find(|(pid, _)| *pid == pes.pid) {
-                    {
-                        let track_idx = self.pid_to_track[pid_idx].1;
-                        if let Some((_, parser)) = self.parsers.iter_mut()
-                            .find(|(pid, _)| *pid == pes.pid)
-                        {
                             for frame in parser.parse(pes) {
                                 self.pending_frames.push_back(
-                                    crate::pes::PesFrame::from_codec_frame(track_idx, frame)
+                                    crate::pes::PesFrame::from_codec_frame(*track, frame)
                                 );
                             }
                         }
                     }
                 }
+            } else if let Some(ref mut demuxer) = self.ps_demuxer {
+                // DVD: program stream demux
+                let packets = demuxer.feed(&self.read_buf[..bytes]);
+                for ps in &packets {
+                    // Map PS stream_id to track index
+                    let track = match ps.stream_id {
+                        0xE0..=0xEF => 0, // video
+                        0xC0..=0xDF => 1, // audio
+                        0xBD => ps.sub_stream_id.map(|s| (s & 0x1F) as usize + 1).unwrap_or(1),
+                        _ => continue,
+                    };
+                    if track < self.title.streams.len() {
+                        let pts_ns = ps.pts.map(|p| (p as i64) * 100_000 / 9).unwrap_or(0);
+                        self.pending_frames.push_back(crate::pes::PesFrame {
+                            track,
+                            pts: pts_ns,
+                            keyframe: true, // PS doesn't have keyframe flag easily
+                            data: ps.data.clone(),
+                        });
+                    }
+                }
             }
 
             // Reset buffer for next read

src/mux/network.rs

@@ -33,7 +33,6 @@ pub struct NetworkStream {
     disc_title: DiscTitle,
     mode: Mode,
     finished: bool,
-    ts_reader: Option<super::tsreader::TsDemuxReader<BufReader<TcpStream>>>,
 }
 
 impl NetworkStream {
@@ -48,7 +47,6 @@ impl NetworkStream {
                 header_written: false,
             },
             finished: false,
-            ts_reader: None,
         })
     }
 
@@ -76,16 +74,36 @@ impl NetworkStream {
             })?
             .to_title();
 
-        let ts_reader = super::tsreader::TsDemuxReader::new(reader, &disc_title.streams);
         Ok(Self {
             disc_title,
-            mode: Mode::Read { reader: BufReader::new(TcpStream::connect("0.0.0.0:0").unwrap()) }, // placeholder
+            mode: Mode::Read { reader },
             finished: false,
-            ts_reader: Some(ts_reader),
         })
     }
 }
 
+impl crate::pes::Stream for NetworkStream {
+    fn read(&mut self) -> io::Result<Option<crate::pes::PesFrame>> {
+        match &mut self.mode {
+            Mode::Read { reader } => crate::pes::PesFrame::deserialize(reader),
+            _ => Err(io::Error::new(io::ErrorKind::Unsupported, "network opened for writing")),
+        }
+    }
+    fn write(&mut self, frame: &crate::pes::PesFrame) -> io::Result<()> {
+        match &mut self.mode {
+            Mode::Write { writer, .. } => frame.serialize(writer),
+            _ => Err(io::Error::new(io::ErrorKind::Unsupported, "network opened for reading")),
+        }
+    }
+    fn finish(&mut self) -> io::Result<()> {
+        if let Mode::Write { writer, .. } = &mut self.mode {
+            writer.flush()?;
+        }
+        Ok(())
+    }
+    fn info(&self) -> &DiscTitle { &self.disc_title }
+}
+
 impl IOStream for NetworkStream {
     fn info(&self) -> &DiscTitle {
         &self.disc_title

src/mux/resolve.rs

@@ -214,12 +214,10 @@ pub fn open_input(url: &str, opts: &InputOptions) -> io::Result<Box<dyn IOStream
         }
         StreamUrl::Network { ref addr } => {
             validate_network_addr(addr)?;
-            Err(io::Error::new(io::ErrorKind::Unsupported,
+            Ok(Box::new(NetworkStream::listen(addr)?))
-                "network:// input requires PES deserialization (TODO)"))
         }
         StreamUrl::Stdio => {
-            Err(io::Error::new(io::ErrorKind::Unsupported,
+            Ok(Box::new(StdioStream::input()))
-                "stdio:// input requires PES deserialization (TODO)"))
         }
         StreamUrl::Iso { ref path } => {
             validate_file_path(path, "iso")?;
@@ -354,13 +352,12 @@ pub fn input(url: &str, opts: &InputOptions) -> io::Result<Box<dyn crate::pes::S
             let reader = std::io::BufReader::with_capacity(IO_BUF_SIZE, file);
             Ok(Box::new(MkvStream::open(reader)?))
         }
-        StreamUrl::Network { .. } => {
+        StreamUrl::Network { ref addr } => {
-            Err(io::Error::new(io::ErrorKind::Unsupported,
+            validate_network_addr(addr)?;
-                "network:// input requires PES deserialization (TODO)"))
+            Ok(Box::new(NetworkStream::listen(addr)?))
         }
         StreamUrl::Stdio => {
-            Err(io::Error::new(io::ErrorKind::Unsupported,
+            Ok(Box::new(StdioStream::input()))
-                "stdio:// input requires PES deserialization (TODO)"))
         }
         StreamUrl::Unknown { ref raw } => {
             Err(io::Error::new(io::ErrorKind::InvalidInput,

src/mux/stdio.rs

@@ -40,6 +40,26 @@ impl StdioStream {
     }
 }
 
+impl crate::pes::Stream for StdioStream {
+    fn read(&mut self) -> io::Result<Option<crate::pes::PesFrame>> {
+        match &mut self.reader {
+            Some(r) => crate::pes::PesFrame::deserialize(r),
+            None => Err(io::Error::new(io::ErrorKind::Unsupported, "stdio opened for writing")),
+        }
+    }
+    fn write(&mut self, frame: &crate::pes::PesFrame) -> io::Result<()> {
+        match &mut self.writer {
+            Some(w) => frame.serialize(w),
+            None => Err(io::Error::new(io::ErrorKind::Unsupported, "stdio opened for reading")),
+        }
+    }
+    fn finish(&mut self) -> io::Result<()> {
+        if let Some(w) = &mut self.writer { w.flush()?; }
+        Ok(())
+    }
+    fn info(&self) -> &DiscTitle { &self.disc_title }
+}
+
 impl IOStream for StdioStream {
     fn info(&self) -> &DiscTitle {
         &self.disc_title