Struct NamedPipe

Source

pub struct NamedPipe {
    pub name: String,
    pub file: File,
}

Available on crate feature named_pipes only.

Expand description

Cross-platform abstraction over Windows async named pipes and Unix FIFO.

Fields§

§name: String

The name that the pipe was opened with. It will start with \\.\pipe\ on Windows.

§file: File

Unix-only; a blocking file handle to the FIFO.

Implementations§

Source §

impl NamedPipe

Source

pub fn new<S: AsRef<str>>(pipe_name: S) -> Result<Self>

Examples found in repository ?

examples/named_pipes.rs (line 64)

12fn main() -> anyhow::Result<()> {
13  use anyhow::Result;
14  use ffmpeg_sidecar::command::FfmpegCommand;
15  use ffmpeg_sidecar::event::{FfmpegEvent, LogLevel};
16  use ffmpeg_sidecar::named_pipes::NamedPipe;
17  use ffmpeg_sidecar::pipe_name;
18  use std::io::Read;
19  use std::sync::mpsc;
20  use std::thread;
21
22  const VIDEO_PIPE_NAME: &str = pipe_name!("ffmpeg_video");
23  const AUDIO_PIPE_NAME: &str = pipe_name!("ffmpeg_audio");
24  const SUBTITLES_PIPE_NAME: &str = pipe_name!("ffmpeg_subtitles");
25
26  // Prepare an FFmpeg command with separate outputs for video, audio, and subtitles.
27  let mut command = FfmpegCommand::new();
28  command
29    // Global flags
30    .hide_banner()
31    .overwrite() // <- overwrite required on windows
32    // Generate test video
33    .format("lavfi")
34    .input("testsrc=size=1920x1080:rate=60:duration=10")
35    // Generate test audio
36    .format("lavfi")
37    .input("sine=frequency=1000:duration=10")
38    // Generate test subtitles
39    .format("srt")
40    .input(
41      "data:text/plain;base64,MQ0KMDA6MDA6MDAsMDAwIC0tPiAwMDowMDoxMCw1MDANCkhlbGxvIFdvcmxkIQ==",
42    )
43    // Video output
44    .map("0:v")
45    .format("rawvideo")
46    .pix_fmt("rgb24")
47    .output(VIDEO_PIPE_NAME)
48    // Audio output
49    .map("1:a")
50    .format("s16le")
51    .output(AUDIO_PIPE_NAME)
52    // Subtitles output
53    .map("2:s")
54    .format("srt")
55    .output(SUBTITLES_PIPE_NAME);
56
57  // Create a separate thread for each output pipe
58  let threads = [VIDEO_PIPE_NAME, AUDIO_PIPE_NAME, SUBTITLES_PIPE_NAME]
59    .iter()
60    .cloned()
61    .map(|pipe_name| {
62      // It's important to create the named pipe on the main thread before
63      // sending it elsewhere so that any errors are caught at the top level.
64      let mut pipe = NamedPipe::new(pipe_name)?;
65      println!("[{pipe_name}] pipe created");
66      let (ready_sender, ready_receiver) = mpsc::channel::<()>();
67      let thread = thread::spawn(move || -> Result<()> {
68        // Wait for FFmpeg to start writing
69        // Only needed for Windows, since Unix will block until a writer has connected
70        println!("[{pipe_name}] waiting for ready signal");
71        ready_receiver.recv()?;
72
73        // Read continuously until finished
74        // Note that if the stream of output is interrupted or paused,
75        // you may need additional logic to keep the read loop alive.
76        println!("[{pipe_name}] reading from pipe");
77        let mut buf = vec![0; 1920 * 1080 * 3];
78        let mut total_bytes_read = 0;
79
80        // In the case of subtitles, we'll decode the string contents directly
81        let mut text_content = if pipe_name == SUBTITLES_PIPE_NAME {
82          Some("".to_string())
83        } else {
84          None
85        };
86
87        loop {
88          match pipe.read(&mut buf) {
89            Ok(bytes_read) => {
90              total_bytes_read += bytes_read;
91
92              // read bytes into string
93              if let Some(cur_str) = &mut text_content {
94                let s = std::str::from_utf8(&buf[..bytes_read]).unwrap();
95                text_content = Some(format!("{}{}", cur_str, s));
96              }
97
98              if bytes_read == 0 {
99                break;
100              }
101            }
102            Err(err) => {
103              if err.kind() != std::io::ErrorKind::BrokenPipe {
104                return Err(err.into());
105              } else {
106                break;
107              }
108            }
109          }
110        }
111
112        // Log how many bytes were received over this pipe.
113        // You can visually compare this to the FFmpeg log output to confirm
114        // that all the expected bytes were captured.
115        let size_str = if total_bytes_read < 1024 {
116          format!("{}B", total_bytes_read)
117        } else {
118          format!("{}KiB", total_bytes_read / 1024)
119        };
120
121        if let Some(text_content) = text_content {
122          println!("[{pipe_name}] subtitle text content: ");
123          println!("{}", text_content.trim());
124        }
125
126        println!("[{pipe_name}] done reading ({size_str} total)");
127        Ok(())
128      });
129
130      Ok((thread, ready_sender))
131    })
132    .collect::<Result<Vec<_>>>()?;
133
134  // Start FFmpeg
135  let mut ready_signal_sent = false;
136  command
137    .print_command()
138    .spawn()?
139    .iter()?
140    .for_each(|event| match event {
141      // Signal threads when output is ready
142      FfmpegEvent::Progress(_) if !ready_signal_sent => {
143        threads.iter().for_each(|(_, sender)| {
144          sender.send(()).ok();
145        });
146        ready_signal_sent = true;
147      }
148
149      // Verify output size from FFmpeg logs (video/audio KiB)
150      FfmpegEvent::Log(LogLevel::Info, msg) if msg.starts_with("[out#") => {
151        println!("{msg}");
152      }
153
154      // Log any unexpected errors
155      FfmpegEvent::Log(LogLevel::Warning | LogLevel::Error | LogLevel::Fatal, msg) => {
156        eprintln!("{msg}");
157      }
158
159      _ => {}
160    });
161
162  for (thread, _) in threads {
163    thread.join().unwrap()?;
164  }
165
166  Ok(())
167}