p4cli_20251/
lib.rs

1use std::io::{BufReader, Read, Write};
2use std::path::PathBuf;
3use std::process::{Child, Command, ExitStatus, Stdio};
4use std::thread;
5use std::time::Duration;
6
7/// Per-instance temp dir via `tempfile`, cleaned up on Drop.
8///
9/// ```rust
10/// use p4cli_20251::P4Cli;
11/// fn main() -> std::io::Result<()> {
12///     let p4: P4Cli = P4Cli::new()?;
13///     let output: p4cli_20251::P4Output = p4.run(&["--help"])?;
14///     println!("exit: {}", output.exit_code());
15///     println!("stdout: {}", output.stdout_str()?);
16///     Ok(())
17/// }
18/// ```
19pub struct P4Cli {
20    bin_path: PathBuf,
21    _temp_dir: tempfile::TempDir,
22}
23
24/// Raw stdout/stderr bytes, exit code and timeout flag from a p4 invocation.
25///
26/// For large outputs (e.g. `p4 sync`, `p4 fstat`), use [`P4Cli::stream`]
27/// instead of [`P4Cli::run`] to avoid buffering all output in memory.
28pub struct P4Output {
29    exit_code: i32,
30    timed_out: bool,
31    stdout: Vec<u8>,
32    stderr: Vec<u8>,
33}
34
35impl P4Output {
36    pub fn exit_code(&self) -> i32 {
37        self.exit_code
38    }
39
40    /// Returns `true` when the process was killed by a timeout.
41    pub fn timed_out(&self) -> bool {
42        self.timed_out
43    }
44
45    /// Returns `true` when the process exited with code 0 and was not timed out.
46    pub fn success(&self) -> bool {
47        !self.timed_out && self.exit_code == 0
48    }
49
50    pub fn stdout(&self) -> &[u8] {
51        &self.stdout
52    }
53
54    pub fn stderr(&self) -> &[u8] {
55        &self.stderr
56    }
57
58    pub fn stdout_str(&self) -> std::io::Result<&str> {
59        std::str::from_utf8(&self.stdout).map_err(std::io::Error::other)
60    }
61
62    pub fn stderr_str(&self) -> std::io::Result<&str> {
63        std::str::from_utf8(&self.stderr).map_err(std::io::Error::other)
64    }
65
66    pub fn stdout_lines(&self) -> std::io::Result<Vec<&str>> {
67        let s = self.stdout_str()?;
68        if s.is_empty() {
69            Ok(Vec::new())
70        } else {
71            Ok(s.lines().collect())
72        }
73    }
74
75    pub fn stderr_lines(&self) -> std::io::Result<Vec<&str>> {
76        let s = self.stderr_str()?;
77        if s.is_empty() {
78            Ok(Vec::new())
79        } else {
80            Ok(s.lines().collect())
81        }
82    }
83}
84
85impl std::fmt::Debug for P4Output {
86    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
87        f.debug_struct("P4Output")
88            .field("exit_code", &self.exit_code)
89            .field("timed_out", &self.timed_out)
90            .field("stdout_len", &self.stdout.len())
91            .field("stderr_len", &self.stderr.len())
92            .finish()
93    }
94}
95
96// ---------------------------------------------------------------------------
97// Streaming API
98// ---------------------------------------------------------------------------
99
100/// A single event yielded by [`P4Stream`].
101pub enum P4StreamEvent {
102    Stdout(Vec<u8>),
103    Stderr(Vec<u8>),
104    Exit(i32),
105}
106
107impl P4StreamEvent {
108    /// Try to decode this event's payload as UTF-8.
109    pub fn as_utf8(&self) -> Option<&str> {
110        match self {
111            P4StreamEvent::Stdout(data) | P4StreamEvent::Stderr(data) => {
112                std::str::from_utf8(data).ok()
113            }
114            P4StreamEvent::Exit(_) => None,
115        }
116    }
117}
118
119impl std::fmt::Display for P4StreamEvent {
120    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
121        match self {
122            P4StreamEvent::Stdout(data) | P4StreamEvent::Stderr(data) => {
123                if let Ok(text) = std::str::from_utf8(data) {
124                    write!(f, "{text}")
125                } else {
126                    write!(f, "<{} bytes>", data.len())
127                }
128            }
129            P4StreamEvent::Exit(code) => write!(f, "(exit {code})"),
130        }
131    }
132}
133
134/// Merged stdout/stderr byte chunks (~64 KB each) as a single iterator.
135///
136/// Uses a bounded channel (64 slots) for backpressure. The final item is
137/// always [`P4StreamEvent::Exit`]. Drop mid-way to kill.
138///
139/// ```rust
140/// use p4cli_20251::{P4Cli, P4StreamEvent};
141/// fn main() -> std::io::Result<()> {
142///     let p4: P4Cli = P4Cli::new()?;
143///     for event in p4.stream(&["--help"])? {
144///         match event? {
145///             P4StreamEvent::Stdout(chunk) => {
146///                 if let Ok(text) = std::str::from_utf8(&chunk) {
147///                     print!("{text}");
148///                 }
149///             }
150///             P4StreamEvent::Stderr(chunk) => {
151///                 if let Ok(text) = std::str::from_utf8(&chunk) {
152///                     eprint!("{text}");
153///                 }
154///             }
155///             P4StreamEvent::Exit(code) => println!("exit {code}"),
156///         }
157///     }
158///     Ok(())
159/// }
160/// ```
161pub struct P4Stream {
162    rx: std::sync::mpsc::Receiver<std::io::Result<P4StreamEvent>>,
163    child: Option<Child>,
164    exhausted: bool,
165}
166
167impl std::fmt::Debug for P4Stream {
168    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
169        f.debug_struct("P4Stream")
170            .field("exhausted", &self.exhausted)
171            .finish()
172    }
173}
174
175impl Iterator for P4Stream {
176    type Item = std::io::Result<P4StreamEvent>;
177
178    fn next(&mut self) -> Option<Self::Item> {
179        if self.exhausted {
180            return None;
181        }
182        match self.rx.recv() {
183            Ok(item) => Some(item),
184            Err(_) => {
185                self.exhausted = true;
186                // Wait for the child to be reaped.
187                match self.child.take() {
188                    Some(mut c) => match c.wait() {
189                        Ok(status) => Some(Ok(P4StreamEvent::Exit(status.code().unwrap_or(-1)))),
190                        Err(e) => Some(Err(e)),
191                    },
192                    None => None,
193                }
194            }
195        }
196    }
197}
198
199impl Drop for P4Stream {
200    fn drop(&mut self) {
201        // Drop rx so sender threads break out of send().
202        // The handles field is consumed by join, but dropping JoinHandle
203        // simply detaches the thread — the threads exit on their own once
204        // the child is killed and pipes close.
205        if let Some(ref mut child) = self.child {
206            let _ = child.kill();
207            let _ = child.wait();
208        }
209    }
210}
211
212// ---------------------------------------------------------------------------
213// Binary extraction
214// ---------------------------------------------------------------------------
215
216fn write_p4_cli_to_disk() -> std::io::Result<(PathBuf, tempfile::TempDir)> {
217    let zst_data = get_p4_cli_zst();
218    let binary_data = decompress_zst(&zst_data)?;
219
220    let temp_dir = create_temp_dir()?;
221    let bin_path = temp_dir.path().join("p4_binary");
222    let tmp_path = temp_dir.path().join(".tmp");
223
224    {
225        let mut file = std::fs::File::create(&tmp_path)?;
226        file.write_all(&binary_data)?;
227        file.sync_all()?;
228    }
229    std::fs::rename(&tmp_path, &bin_path)?;
230    set_executable_perms(&bin_path)?;
231
232    Ok((bin_path, temp_dir))
233}
234
235fn create_temp_dir() -> std::io::Result<tempfile::TempDir> {
236    let mut builder = tempfile::Builder::new();
237    builder.prefix("p4cli-20251");
238    #[cfg(unix)]
239    {
240        use std::os::unix::fs::PermissionsExt;
241        builder.permissions(std::fs::Permissions::from_mode(0o700));
242    }
243    builder.tempdir()
244}
245
246fn decompress_zst(zst_data: &[u8]) -> std::io::Result<Vec<u8>> {
247    let mut decoder = zstd::stream::Decoder::new(zst_data)?;
248    let mut buf = Vec::new();
249    std::io::copy(&mut decoder, &mut buf)?;
250    Ok(buf)
251}
252
253#[cfg(unix)]
254fn set_executable_perms(path: &std::path::Path) -> std::io::Result<()> {
255    use std::os::unix::fs::PermissionsExt;
256    std::fs::set_permissions(path, std::fs::Permissions::from_mode(0o700))
257}
258
259#[cfg(not(unix))]
260fn set_executable_perms(_path: &std::path::Path) -> std::io::Result<()> {
261    Ok(())
262}
263
264fn get_p4_cli_zst() -> Vec<u8> {
265    #[cfg(all(target_os = "windows", target_arch = "x86_64"))]
266    {
267        use p4cli_20251_win_x64::get_p4_cli_zst;
268        get_p4_cli_zst()
269    }
270
271    #[cfg(all(target_os = "macos", target_arch = "aarch64"))]
272    {
273        use p4cli_20251_mac_arm64::get_p4_cli_zst;
274        get_p4_cli_zst()
275    }
276
277    #[cfg(all(target_os = "macos", target_arch = "x86_64"))]
278    {
279        use p4cli_20251_mac_x64::get_p4_cli_zst;
280        get_p4_cli_zst()
281    }
282
283    #[cfg(all(
284        target_os = "linux",
285        target_arch = "x86_64",
286        any(target_env = "gnu", target_env = "musl")
287    ))]
288    {
289        use p4cli_20251_linux_x64::get_p4_cli_zst;
290        get_p4_cli_zst()
291    }
292
293    #[cfg(all(
294        target_os = "linux",
295        target_arch = "aarch64",
296        any(target_env = "gnu", target_env = "musl")
297    ))]
298    {
299        use p4cli_20251_linux_arm64::get_p4_cli_zst;
300        get_p4_cli_zst()
301    }
302
303    #[cfg(not(any(
304        all(target_os = "windows", target_arch = "x86_64"),
305        all(target_os = "macos", target_arch = "aarch64"),
306        all(target_os = "macos", target_arch = "x86_64"),
307        all(
308            target_os = "linux",
309            target_arch = "x86_64",
310            any(target_env = "gnu", target_env = "musl")
311        ),
312        all(
313            target_os = "linux",
314            target_arch = "aarch64",
315            any(target_env = "gnu", target_env = "musl")
316        ),
317    )))]
318    {
319        compile_error!(format!(
320            "Unsupported platform: {}-{}",
321            std::env::consts::OS,
322            std::env::consts::ARCH
323        ));
324        Vec::new()
325    }
326}
327
328// ---------------------------------------------------------------------------
329// P4Command builder
330// ---------------------------------------------------------------------------
331
332/// Builder for a single p4 invocation (timeout, cwd, env, stdin).
333pub struct P4Command<'a> {
334    cli: &'a P4Cli,
335    args: Vec<std::ffi::OsString>,
336    timeout: Option<Duration>,
337    cwd: Option<PathBuf>,
338    envs: Vec<(std::ffi::OsString, std::ffi::OsString)>,
339    stdin_data: Option<Vec<u8>>,
340}
341
342impl<'a> P4Command<'a> {
343    fn new(cli: &'a P4Cli) -> Self {
344        Self {
345            cli,
346            args: Vec::new(),
347            timeout: None,
348            cwd: None,
349            envs: Vec::new(),
350            stdin_data: None,
351        }
352    }
353
354    pub fn arg(&mut self, arg: impl AsRef<std::ffi::OsStr>) -> &mut Self {
355        self.args.push(arg.as_ref().to_os_string());
356        self
357    }
358
359    pub fn args(&mut self, args: &[impl AsRef<std::ffi::OsStr>]) -> &mut Self {
360        self.args
361            .extend(args.iter().map(|a| a.as_ref().to_os_string()));
362        self
363    }
364
365    /// Maximum wall-clock time. Kills the direct child on timeout (not process tree).
366    ///
367    /// When the process is killed by timeout, [`P4Output::timed_out`] returns `true`.
368    /// Not supported with [`stream()`](Self::stream) — use [`run()`](Self::run) instead.
369    pub fn timeout(&mut self, timeout: Duration) -> &mut Self {
370        self.timeout = Some(timeout);
371        self
372    }
373
374    pub fn cwd(&mut self, path: impl Into<PathBuf>) -> &mut Self {
375        self.cwd = Some(path.into());
376        self
377    }
378
379    pub fn env(
380        &mut self,
381        key: impl Into<std::ffi::OsString>,
382        val: impl Into<std::ffi::OsString>,
383    ) -> &mut Self {
384        self.envs.push((key.into(), val.into()));
385        self
386    }
387
388    pub fn stdin(&mut self, data: impl Into<Vec<u8>>) -> &mut Self {
389        self.stdin_data = Some(data.into());
390        self
391    }
392
393    /// Block until the process exits, returning collected output.
394    ///
395    /// For large output, consider [`stream()`](Self::stream) instead to
396    /// avoid buffering everything in memory.
397    pub fn run(&mut self) -> std::io::Result<P4Output> {
398        let mut cmd = Command::new(&self.cli.bin_path);
399        cmd.args(&self.args)
400            .stdout(Stdio::piped())
401            .stderr(Stdio::piped());
402
403        if self.stdin_data.is_some() {
404            cmd.stdin(Stdio::piped());
405        } else {
406            cmd.stdin(Stdio::null());
407        }
408
409        if let Some(ref cwd) = self.cwd {
410            cmd.current_dir(cwd);
411        }
412        for (k, v) in &self.envs {
413            cmd.env(k, v);
414        }
415
416        let mut child = cmd.spawn()?;
417
418        let stdin_handle = self.stdin_data.take().and_then(|data| {
419            child.stdin.take().map(|mut stdin| {
420                thread::spawn(move || {
421                    if let Err(e) = stdin.write_all(&data) {
422                        Err(e)
423                    } else {
424                        Ok(())
425                    }
426                })
427            })
428        });
429
430        let stdout = child
431            .stdout
432            .take()
433            .ok_or_else(|| std::io::Error::other("stdout was not captured"))?;
434        let stderr = child
435            .stderr
436            .take()
437            .ok_or_else(|| std::io::Error::other("stderr was not captured"))?;
438
439        let stdout_handle = thread::spawn(move || {
440            let mut buf = Vec::new();
441            BufReader::new(stdout).read_to_end(&mut buf)?;
442            Ok::<_, std::io::Error>(buf)
443        });
444
445        let stderr_handle = thread::spawn(move || {
446            let mut buf = Vec::new();
447            BufReader::new(stderr).read_to_end(&mut buf)?;
448            Ok::<_, std::io::Error>(buf)
449        });
450
451        let (exit_status, timed_out) = wait_process(&mut child, self.timeout)?;
452
453        // Surface stdin write errors.
454        if let Some(handle) = stdin_handle {
455            handle
456                .join()
457                .map_err(|_| std::io::Error::other("stdin thread panicked"))?
458                .map_err(|e| std::io::Error::other(format!("stdin write failed: {e}")))?;
459        }
460
461        let stdout_buf = stdout_handle
462            .join()
463            .map_err(|_| std::io::Error::other("stdout reader thread panicked"))?
464            .map_err(|e| std::io::Error::other(format!("stdout read failed: {e}")))?;
465        let stderr_buf = stderr_handle
466            .join()
467            .map_err(|_| std::io::Error::other("stderr reader thread panicked"))?
468            .map_err(|e| std::io::Error::other(format!("stderr read failed: {e}")))?;
469
470        Ok(P4Output {
471            exit_code: exit_status.code().unwrap_or(-1),
472            timed_out,
473            stdout: stdout_buf,
474            stderr: stderr_buf,
475        })
476    }
477
478    /// Streaming iterator over stdout/stderr byte chunks.
479    ///
480    /// Uses a bounded channel (64 slots) for backpressure. The final event
481    /// is [`P4StreamEvent::Exit`]. Drop mid-way to cancel.
482    ///
483    /// **Note**: [`timeout`](Self::timeout) is not supported — use
484    /// [`run()`](Self::run) instead.
485    pub fn stream(&mut self) -> std::io::Result<P4Stream> {
486        if self.timeout.is_some() {
487            return Err(std::io::Error::other(
488                "timeout is not supported on stream(); use run() instead",
489            ));
490        }
491
492        let mut cmd = Command::new(&self.cli.bin_path);
493        cmd.args(&self.args)
494            .stdout(Stdio::piped())
495            .stderr(Stdio::piped());
496
497        if self.stdin_data.is_some() {
498            cmd.stdin(Stdio::piped());
499        } else {
500            cmd.stdin(Stdio::null());
501        }
502        if let Some(ref cwd) = self.cwd {
503            cmd.current_dir(cwd);
504        }
505        for (k, v) in &self.envs {
506            cmd.env(k, v);
507        }
508
509        let mut child = cmd.spawn()?;
510
511        if let Some(data) = self.stdin_data.take()
512            && let Some(mut stdin) = child.stdin.take()
513        {
514            thread::spawn(move || {
515                let _ = stdin.write_all(&data);
516            });
517        }
518
519        let mut stdout = child
520            .stdout
521            .take()
522            .ok_or_else(|| std::io::Error::other("stdout was not captured"))?;
523        let mut stderr = child
524            .stderr
525            .take()
526            .ok_or_else(|| std::io::Error::other("stderr was not captured"))?;
527
528        // Bounded channel for backpressure (64 slots).
529        let (tx, rx) = std::sync::mpsc::sync_channel(64);
530
531        let tx_out = tx.clone();
532        thread::spawn(move || {
533            let mut buf = vec![0u8; 65536];
534            loop {
535                let n = match stdout.read(&mut buf) {
536                    Ok(0) => break,
537                    Ok(n) => n,
538                    Err(e) => {
539                        let _ = tx_out.send(Err(e));
540                        break;
541                    }
542                };
543                if tx_out
544                    .send(Ok(P4StreamEvent::Stdout(buf[..n].to_vec())))
545                    .is_err()
546                {
547                    break;
548                }
549            }
550        });
551
552        let tx_err = tx.clone();
553        thread::spawn(move || {
554            let mut buf = vec![0u8; 65536];
555            loop {
556                let n = match stderr.read(&mut buf) {
557                    Ok(0) => break,
558                    Ok(n) => n,
559                    Err(e) => {
560                        let _ = tx_err.send(Err(e));
561                        break;
562                    }
563                };
564                if tx_err
565                    .send(Ok(P4StreamEvent::Stderr(buf[..n].to_vec())))
566                    .is_err()
567                {
568                    break;
569                }
570            }
571        });
572
573        Ok(P4Stream {
574            rx,
575            child: Some(child),
576            exhausted: false,
577        })
578    }
579}
580
581// ---------------------------------------------------------------------------
582// Process helpers
583// ---------------------------------------------------------------------------
584
585fn wait_process(
586    child: &mut Child,
587    timeout: Option<Duration>,
588) -> std::io::Result<(ExitStatus, bool)> {
589    match timeout {
590        None => Ok((child.wait()?, false)),
591        Some(t) => wait_with_timeout(child, t),
592    }
593}
594
595fn wait_with_timeout(child: &mut Child, timeout: Duration) -> std::io::Result<(ExitStatus, bool)> {
596    let start = std::time::Instant::now();
597    loop {
598        if let Some(status) = child.try_wait()? {
599            return Ok((status, false));
600        }
601        if start.elapsed() >= timeout {
602            child.kill()?;
603            return Ok((child.wait()?, true));
604        }
605        thread::sleep(Duration::from_millis(50));
606    }
607}
608
609// ---------------------------------------------------------------------------
610// Public API
611// ---------------------------------------------------------------------------
612
613impl P4Cli {
614    /// Decompress embedded p4 binary to a `tempfile`-managed temp directory.
615    pub fn new() -> std::io::Result<Self> {
616        let (bin_path, temp_dir) = write_p4_cli_to_disk()?;
617        Ok(Self {
618            bin_path,
619            _temp_dir: temp_dir,
620        })
621    }
622
623    /// Equivalent to `self.command().args(args).run()`.
624    pub fn run<S: AsRef<std::ffi::OsStr>>(&self, args: &[S]) -> std::io::Result<P4Output> {
625        self.command().args(args).run()
626    }
627
628    /// Equivalent to `self.command().args(args).stream()`.
629    pub fn stream<S: AsRef<std::ffi::OsStr>>(&self, args: &[S]) -> std::io::Result<P4Stream> {
630        self.command().args(args).stream()
631    }
632
633    /// Obtain a [`P4Command`] builder.
634    ///
635    /// ```rust
636    /// use p4cli_20251::P4Cli;
637    /// use std::time::Duration;
638    /// fn main() -> std::io::Result<()> {
639    ///     let p4: P4Cli = P4Cli::new()?;
640    ///     let output: p4cli_20251::P4Output = p4
641    ///         .command()
642    ///         .arg("--help")
643    ///         .timeout(Duration::from_secs(10))
644    ///         .run()?;
645    ///     if output.success() {
646    ///         println!("{}", output.stdout_str()?);
647    ///     }
648    ///     Ok(())
649    /// }
650    /// ```
651    pub fn command(&self) -> P4Command<'_> {
652        P4Command::new(self)
653    }
654}
p4cli_20251/lib.rs

p4cli_20251/
lib.rs
