Repository: nathanbabcock/ffmpeg-sidecar (language: rust)

Issue:
`FfmpegIterator::new()` incorrectly blocks until the FFmpeg process ends (!)
Will fix ASAP...


Discussion / hints:

Fix patch:
diff --git a/examples/h265_transcode.rs b/examples/h265_transcode.rs
index 9d296cf..49a322d 100644
--- a/examples/h265_transcode.rs
+++ b/examples/h265_transcode.rs
@@ -29,7 +29,7 @@ fn main() {
 
   // Frames can be transformed by Iterator `.map()`.
   // This example is a no-op, with frames passed through unaltered.
-  let transformed_frames = input.iter().unwrap().filter_frames().map(|f| f);
+  let transformed_frames = input.iter().unwrap().filter_frames();
 
   // You could easily add some "middleware" processing here:
   // - overlay or composite another RGB image (or even another Ffmpeg Iterator)
diff --git a/src/child.rs b/src/child.rs
index 028f14e..916d56e 100644
--- a/src/child.rs
+++ b/src/child.rs
@@ -14,7 +14,7 @@ pub struct FfmpegChild {
 }
 
 impl FfmpegChild {
-  /// Creates an iterator over events emitted by ffmpeg. Functions similarly to
+  /// Creates an iterator over events emitted by FFmpeg. Functions similarly to
   /// `Lines` from [`std::io::BufReader`], but providing a variety of parsed
   /// events:
   /// - Log messages
diff --git a/src/download.rs b/src/download.rs
index f3e3a8e..d1a045c 100644
--- a/src/download.rs
+++ b/src/download.rs
@@ -92,7 +92,7 @@ pub fn parse_macos_version(version: &str) -> Option<String> {
     .split("\"version\":")
     .nth(1)?
     .trim()
-    .split("\"")
+    .split('\"')
     .nth(1)
     .map(|s| s.to_string())
 }
@@ -111,7 +111,6 @@ pub fn parse_linux_version(version: &str) -> Option<String> {
   version
     .split("version:")
     .nth(1)?
-    .trim_start()
     .split_whitespace()
     .next()
     .map(|s| s.to_string())
@@ -208,15 +207,15 @@ pub fn unpack_ffmpeg(from_archive: &PathBuf, binary_folder: &PathBuf) -> Result<
 
   let (ffmpeg, ffplay, ffprobe) = if cfg!(target_os = "windows") {
     (
-      (&inner_folder).path().join("bin/ffmpeg.exe"),
-      (&inner_folder).path().join("bin/ffplay.exe"),
-      (&inner_folder).path().join("bin/ffprobe.exe"),
+      inner_folder.path().join("bin/ffmpeg.exe"),
+      inner_folder.path().join("bin/ffplay.exe"),
+      inner_folder.path().join("bin/ffprobe.exe"),
     )
   } else if cfg!(any(target_os = "linux", target_os = "macos")) {
     (
-      (&inner_folder).path().join("./ffmpeg"),
-      (&inner_folder).path().join("./ffplay"), // <- this typically only exists in Windows builds
-      (&inner_folder).path().join("./ffprobe"),
+      inner_folder.path().join("./ffmpeg"),
+      inner_folder.path().join("./ffplay"), // <- this typically only exists in Windows builds
+      inner_folder.path().join("./ffprobe"),
     )
   } else {
     return Err(Error::msg("Unsupported platform"));
diff --git a/src/event.rs b/src/event.rs
index bb0c428..52f8e95 100644
--- a/src/event.rs
+++ b/src/event.rs
@@ -45,6 +45,8 @@ impl FfmpegOutput {
 
 #[derive(Debug, Clone, PartialEq)]
 pub struct AVStream {
+  /// Typically `video` or `audio`, but might be something else like `data` or `subtitle`.
+  pub stream_type: String,
   /// Corresponds to stream `-f` parameter, e.g. `rawvideo`, `h264`, or `mpegts`
   pub format: String,
   /// Corresponds to stream `-pix_fmt` parameter, e.g. `rgb24`
diff --git a/src/iter.rs b/src/iter.rs
index a8fe0e6..06f57d8 100644
--- a/src/iter.rs
+++ b/src/iter.rs
@@ -1,5 +1,4 @@
 use std::{
-  collections::VecDeque,
   io::{BufReader, ErrorKind, Read},
   process::{ChildStderr, ChildStdout},
   sync::mpsc::{sync_channel, Receiver, SyncSender},
@@ -17,7 +16,19 @@ use crate::{
 /// An iterator over events from an ffmpeg process, including parsed metadata, progress, and raw video frames.
 pub struct FfmpegIterator {
   rx: Receiver<FfmpegEvent>,
-  event_queue: VecDeque<FfmpegEvent>,
+  tx: Option<SyncSender<FfmpegEvent>>,
+  stdout: Option<ChildStdout>,
+  metadata: FfmpegMetadata,
+}
+
+#[derive(Debug, Clone, PartialEq)]
+pub struct FfmpegMetadata {
+  expected_output_streams: usize,
+  output_streams: Vec<AVStream>,
+  outputs: Vec<FfmpegOutput>,
+
+  /// Whether all metadata from the parent process has been gathered into this struct
+  completed: bool,
 }
 
 impl FfmpegIterator {
@@ -25,47 +36,73 @@ impl FfmpegIterator {
     let stderr = child.take_stderr().ok_or_else(|| Error::msg("No stderr channel\n - Did you call `take_stderr` elsewhere?\n - Did you forget to call `.stderr(Stdio::piped)` on the `ChildProcess`?"))?;
     let (tx, rx) = sync_channel::<FfmpegEvent>(0);
     spawn_stderr_thread(stderr, tx.clone());
+    let stdout = child.take_stdout();
 
-    // Await the output metadata
-    let mut output_streams: Vec<AVStream> = Vec::new();
-    let mut outputs: Vec<FfmpegOutput> = Vec::new();
-    let mut expected_output_streams = 0;
-    let mut event_queue: VecDeque<FfmpegEvent> = VecDeque::new();
-    while let Ok(event) = rx.recv() {
-      event_queue.push_back(event.clone());
-      match event {
-        // Every stream mapping corresponds to one output stream
-        // We count these to know when we've received all the output streams
-        FfmpegEvent::ParsedStreamMapping(_) => expected_output_streams += 1,
-        FfmpegEvent::ParsedOutput(output) => outputs.push(output),
-        FfmpegEvent::ParsedOutputStream(stream) => {
-          output_streams.push(stream.clone());
-          if output_streams.len() == expected_output_streams {
-            break;
-          }
-        }
-        FfmpegEvent::LogEOF => {
-          // An unexpected EOF means we bail out here,
-          // but still pass on the events we've already received
-          return Ok(Self { rx, event_queue });
-        }
-        _ => {}
+    Ok(Self {
+      rx,
+      tx: Some(tx),
+      stdout,
+      metadata: FfmpegMetadata {
+        expected_output_streams: 0,
+        output_streams: Vec::new(),
+        outputs: Vec::new(),
+        completed: false,
+      },
+    })
+  }
+
+  fn handle_metadata(&mut self, item: &Option<FfmpegEvent>) -> Result<()> {
+    match item {
+      // Every stream mapping corresponds to one output stream
+      // We count these to know when we've received all the output streams
+      Some(FfmpegEvent::ParsedStreamMapping(_)) => self.metadata.expected_output_streams += 1,
+      Some(FfmpegEvent::ParsedOutput(output)) => self.metadata.outputs.push(output.clone()),
+      Some(FfmpegEvent::ParsedOutputStream(stream)) => {
+        self.metadata.output_streams.push(stream.clone())
       }
+      _ => (),
     }
 
+    if self.metadata.expected_output_streams > 0
+      && self.metadata.output_streams.len() == self.metadata.expected_output_streams
+    {
+      self.finish_metadata()?;
+    }
+
+    Ok(())
+  }
+
+  fn finish_metadata(&mut self) -> Result<()> {
+    // "Seal" the metadata struct
+    self.metadata.completed = true;
+
     // No output detected
-    if output_streams.is_empty() || outputs.is_empty() {
+    if self.metadata.output_streams.is_empty() || self.metadata.outputs.is_empty() {
       let err = Error::msg("No output streams found");
-      child.kill()?;
-      Err(err)? // this is just a cute way of saying `return err`
+      self.tx.take(); // drop the tx so that the channel closes
+      return Err(err);
     }
 
     // Handle stdout
-    if let Some(stdout) = child.take_stdout() {
-      spawn_stdout_thread(stdout, tx, output_streams, outputs);
+    if let Some(stdout) = self.stdout.take() {
+      spawn_stdout_thread(
+        stdout,
+        self.tx.take().ok_or("missing channel tx")?,
+        self.metadata.output_streams.clone(),
+        self.metadata.outputs.clone(),
+      );
+    }
+
+    Ok(())
+  }
+
+  /// Advance the iterator until all metadata has been collected, returning it.
+  pub fn collect_metadata(&mut self) -> FfmpegMetadata {
+    while !self.metadata.completed {
+      self.next();
     }
 
-    Ok(Self { rx, event_queue })
+    self.metadata.clone()
   }
 
   //// Iterator filters
@@ -127,13 +164,21 @@ impl Iterator for FfmpegIterator {
   type Item = FfmpegEvent;
 
   fn next(&mut self) -> Option<Self::Item> {
-    match self.event_queue.pop_front() {
-      // First, re-send the queued events that were used to parse metadata
-      Some(event) => Some(event),
+    let item = self.rx.recv().ok();
 
-      // Then, read from the channel or return `None` when it closes
-      None => self.rx.recv().ok(),
+    if let Some(FfmpegEvent::LogEOF) = item {
+      self.tx.take(); // drop the tx so that the receiver can close
     }
+
+    if !self.metadata.completed {
+      if let Err(e) = self.handle_metadata(&item) {
+        return Some(FfmpegEvent::Error(e.to_string()));
+        // TODO in this case, the preceding `item` is lost;
+        // Probably better to queue it as the next item.
+      }
+    }
+
+    item
   }
 }
 
@@ -153,16 +198,16 @@ pub fn spawn_stdout_thread(
         .unwrap_or(false)
     });
 
-    // Limit to exactly one non-rawvideo stream,
-    // or unlimited rawvideo streams
-    if stdout_output_streams
+    // Error on mixing rawvideo and non-rawvideo streams
+    // TODO: Maybe just revert to chunk mode if this happens?
+    let any_rawvideo = stdout_output_streams
       .clone()
-      .any(|s| s.format != "rawvideo")
-    {
-      assert!(
-        stdout_output_streams.clone().count() == 1,
-        "Only one non-rawvideo output stream can be sent to stdout",
-      );
+      .any(|s| s.format == "rawvideo");
+    let any_non_rawvideo = stdout_output_streams
+      .clone()
+      .any(|s| s.format != "rawvideo");
+    if any_rawvideo && any_non_rawvideo {
+      panic!("Cannot mix rawvideo and non-rawvideo streams");
     }
 
     // Prepare buffers
diff --git a/src/log_parser.rs b/src/log_parser.rs
index f1fa450..4ed20fb 100644
--- a/src/log_parser.rs
+++ b/src/log_parser.rs
@@ -66,6 +66,8 @@ impl<R: Read> FfmpegLogParser<R> {
             configuration,
             raw_log_message: line.to_string(),
           }))
+        } else if self.cur_section == LogSection::StreamMapping && line.contains("  Stream #") {
+          Ok(FfmpegEvent::ParsedStreamMapping(line.to_string()))
         } else if let Some(stream) = try_parse_stream(line) {
           match self.cur_section {
             LogSection::Input(_) => Ok(FfmpegEvent::ParsedInputStream(stream)),
@@ -75,8 +77,6 @@ impl<R: Read> FfmpegLogParser<R> {
               line
             ))),
           }
-        } else if self.cur_section == LogSection::StreamMapping && line.contains("  Stream #") {
-          Ok(FfmpegEvent::ParsedStreamMapping(line.to_string()))
         } else if let Some(progress) = try_parse_progress(line) {
           self.cur_section = LogSection::Other;
           Ok(FfmpegEvent::Progress(progress))
@@ -225,7 +225,7 @@ pub fn try_parse_output(mut string: &str) -> Option<FfmpegOutput> {
 
 /// ## Example
 ///
-/// Input stream:
+/// ### Input stream:
 ///
 /// ```rust
 /// use ffmpeg_sidecar::log_parser::try_parse_stream;
@@ -239,7 +239,7 @@ pub fn try_parse_output(mut string: &str) -> Option<FfmpegOutput> {
 /// assert!(stream.parent_index == 0);
 /// ```
 ///
-/// Output stream:
+/// ### Output stream:
 ///
 /// ```rust
 /// use ffmpeg_sidecar::log_parser::try_parse_stream;
@@ -252,6 +252,16 @@ pub fn try_parse_output(mut string: &str) -> Option<FfmpegOutput> {
 /// assert!(stream.fps == 25.0);
 /// assert!(stream.parent_index == 1);
 /// ```
+///
+/// ### Audio output stream:
+///
+/// ```rust
+/// use ffmpeg_sidecar::log_parser::try_parse_stream;
+/// let line = "[info]   Stream #0:1: Audio: mp2, 44100 Hz, mono, s16, 384 kb/s\n";
+/// let stream = try_parse_stream(line).unwrap();
+/// assert!(stream.stream_type == "Audio");
+/// ```
+///
 pub fn try_parse_stream(mut string: &str) -> Option<AVStream> {
   let raw_log_message = string.to_string();
 
@@ -263,9 +273,19 @@ pub fn try_parse_stream(mut string: &str) -> Option<AVStream> {
 
   let mut colon_parts = string.split(':');
   let parent_index = colon_parts.next()?.parse::<usize>().ok()?;
-  let stream_type = colon_parts.nth(1)?.trim();
+  let stream_type = colon_parts.nth(1)?.trim().to_string();
   if stream_type != "Video" {
-    return None; // Audio not supported yet (PRs welcome)
+    // Audio is not well-supported yet (PRs welcome)
+    return Some(AVStream {
+      stream_type,
+      format: "unknown".into(),
+      pix_fmt: "unknown".into(),
+      width: 0,
+      height: 0,
+      fps: 0.0,
+      parent_index,
+      raw_log_message,
+    });
   }
   let comma_string = colon_parts.next()?.trim();
   let mut comma_iter = CommaIter::new(comma_string);
@@ -292,13 +312,13 @@ pub fn try_parse_stream(mut string: &str) -> Option<AVStream> {
   let fps = string
     .split("fps,")
     .next()?
-    .trim()
     .split_whitespace()
     .last()?
     .parse()
     .ok()?;
 
   Some(AVStream {
+    stream_type,
     parent_index,
     format,
     pix_fmt,
diff --git a/src/read_until_any.rs b/src/read_until_any.rs
index febfefd..df44e17 100644
--- a/src/read_until_any.rs
+++ b/src/read_until_any.rs
@@ -18,7 +18,7 @@ pub fn read_until_any<R: BufRead + ?Sized>(
       // NB: `memchr` crate would be faster, but it's unstable and not worth the dependency.
       let first_delim_index = available
         .iter()
-        .position(|b| delims.iter().find(|d| **d == *b).is_some());
+        .position(|b| delims.iter().any(|d| *d == *b));
 
       match first_delim_index {
         Some(i) => {


Test patch:
diff --git a/src/test.rs b/src/test.rs
index 3941380..2c19c2a 100644
--- a/src/test.rs
+++ b/src/test.rs
@@ -165,6 +165,30 @@ fn test_chunks() {
   assert!(chunks > 0);
 }
 
+#[test]
+fn test_chunks_with_audio() {
+  let mut chunks = 0;
+  let mut frames = 0;
+
+  FfmpegCommand::new()
+    .testsrc()
+    .args("-f lavfi -i sine=frequency=1000 -shortest".split(' '))
+    .codec_video("libx264")
+    .format("mpegts")
+    .pipe_stdout()
+    .spawn()
+    .unwrap()
+    .iter()
+    .unwrap()
+    .for_each(|e| match e {
+      FfmpegEvent::OutputChunk(_) => chunks += 1,
+      FfmpegEvent::OutputFrame(_) => frames += 1,
+      _ => {}
+    });
+
+  assert!(chunks > 0);
+}
+
 #[test]
 fn test_kill_before_iter() {
   let mut child = FfmpegCommand::new().testsrc().rawvideo().spawn().unwrap();