command_stream/lib.rs
1//! # command-stream
2//!
3//! Modern shell command execution library with streaming, async iteration, and event support.
4//!
5//! This library provides a Rust equivalent to the JavaScript command-stream library,
6//! offering powerful shell command execution with streaming capabilities.
7//!
8//! ## Features
9//!
10//! - Async command execution with tokio
11//! - Streaming output via async iterators
12//! - Event-based output handling (on, once, emit)
13//! - Virtual commands for common operations (cat, ls, mkdir, etc.)
14//! - Shell operator support (&&, ||, ;, |)
15//! - Pipeline support with `.pipe()` method and `Pipeline` builder
16//! - Global state management for shell settings
17//! - `cmd!` macro for ergonomic command creation (similar to JS `$` tagged template literals)
18//! - Cross-platform support
19//!
20//! ## Module Organization
21//!
22//! The codebase follows a modular architecture similar to the JavaScript implementation:
23//!
24//! - `ansi` - ANSI escape code handling utilities
25//! - `commands` - Virtual command implementations
26//! - `events` - Event emitter for stream events
27//! - `macros` - The `cmd!` macro for ergonomic command creation
28//! - `pipeline` - Pipeline execution support
29//! - `quote` - Shell quoting utilities
30//! - `shell_parser` - Shell command parsing
31//! - `state` - Global state management
32//! - `stream` - Async streaming and iteration support
33//! - `trace` - Logging and tracing utilities
34//! - `utils` - Command results and virtual command helpers
35//!
36//! ## Quick Start
37//!
38//! ```rust,no_run
39//! use command_stream::{run, cmd};
40//!
41//! #[tokio::main]
42//! async fn main() -> Result<(), Box<dyn std::error::Error>> {
43//! // Execute a simple command
44//! let result = run("echo hello world").await?;
45//! println!("{}", result.stdout);
46//!
47//! // Using the cmd! macro (similar to JS $ tagged template)
48//! let name = "world";
49//! let result = cmd!("echo hello {}", name).await?;
50//! println!("{}", result.stdout);
51//!
52//! // Using pipelines
53//! use command_stream::Pipeline;
54//! let result = Pipeline::new()
55//! .add("echo hello world")
56//! .add("grep world")
57//! .run()
58//! .await?;
59//!
60//! Ok(())
61//! }
62//! ```
63
64// Modular utility modules (following JavaScript modular pattern)
65pub mod ansi;
66pub mod events;
67#[doc(hidden)]
68pub mod macros;
69pub mod pipeline;
70pub mod quote;
71pub mod signal;
72pub mod state;
73pub mod stream;
74pub mod terminal;
75pub mod trace;
76
77// Core modules
78pub mod commands;
79pub mod shell_parser;
80pub mod utils;
81
82use std::collections::HashMap;
83use std::path::PathBuf;
84use std::process::Stdio;
85use tokio::io::{AsyncRead, AsyncReadExt, AsyncWriteExt};
86use tokio::process::Child;
87use tokio::sync::mpsc;
88
89pub use commands::{CommandContext, StreamChunk};
90pub use shell_parser::{needs_real_shell, parse_shell_command, ParsedCommand};
91pub use utils::{CommandResult, VirtualUtils};
92
93// Re-export modular utilities at crate root for convenient access
94pub use ansi::{AnsiConfig, AnsiUtils};
95pub use events::{EventData, EventType, StreamEmitter};
96pub use pipeline::{Pipeline, PipelineBuilder, PipelineExt};
97pub use quote::{
98 escape_for_double_quotes, escape_for_single_quotes, has_shell_escapes,
99 is_pre_quoted_passthrough_enabled, is_quote_context_enabled, quote, quote_for_context,
100 scan_quote_context, QuoteContext,
101};
102pub use signal::{signal_exit_code, signal_number, DEFAULT_KILL_GRACE_MS, DEFAULT_KILL_SIGNAL};
103pub use state::{
104 get_shell_settings, global_state, reset_global_state, set_shell_option, unset_shell_option,
105 GlobalState, ShellSettings,
106};
107pub use stream::{AsyncIterator, IntoStream, OutputChunk, OutputStream, StreamingRunner};
108pub use trace::trace;
109
110#[derive(Clone, Copy)]
111enum ChildOutput {
112 Stdout,
113 Stderr,
114}
115
116/// Read child output as byte chunks so capture does not invent a trailing newline.
117///
118/// stdout and stderr use separate futures in `ProcessRunner::run`, preventing
119/// either pipe from filling while the other is being drained. Mirroring keeps
120/// the original bytes too, including output that does not end in a newline.
121async fn collect_child_output<R>(
122 reader: Option<R>,
123 mirror: bool,
124 target: ChildOutput,
125) -> std::io::Result<Vec<u8>>
126where
127 R: AsyncRead + Unpin,
128{
129 let Some(mut reader) = reader else {
130 return Ok(Vec::new());
131 };
132 let mut collected = Vec::new();
133 let mut buffer = [0_u8; 8192];
134
135 loop {
136 let count = reader.read(&mut buffer).await?;
137 if count == 0 {
138 break;
139 }
140
141 let chunk = &buffer[..count];
142 collected.extend_from_slice(chunk);
143 if mirror {
144 match target {
145 ChildOutput::Stdout => {
146 let mut output = std::io::stdout().lock();
147 let _ = std::io::Write::write_all(&mut output, chunk);
148 let _ = std::io::Write::flush(&mut output);
149 }
150 ChildOutput::Stderr => {
151 let mut output = std::io::stderr().lock();
152 let _ = std::io::Write::write_all(&mut output, chunk);
153 let _ = std::io::Write::flush(&mut output);
154 }
155 }
156 }
157 }
158
159 Ok(collected)
160}
161
162fn fallback_cwd() -> PathBuf {
163 std::env::var_os("HOME")
164 .or_else(|| std::env::var_os("USERPROFILE"))
165 .map(PathBuf::from)
166 .filter(|path| path.is_dir())
167 .unwrap_or_else(std::env::temp_dir)
168}
169
170/// Resolve a working directory that is safe to spawn a child process in.
171///
172/// When no explicit cwd is requested the child normally inherits the parent's
173/// working directory. But if that directory has been deleted or become
174/// inaccessible (the "getcwd() failed" scenario from issue #44), inheriting it
175/// makes the OS-level spawn fail. In that case fall back to a directory that is
176/// known to exist so the command still runs.
177///
178/// Normal behavior is preserved: when an explicit cwd is given, or when the
179/// inherited working directory is valid, this returns the requested value
180/// (`None` meaning "inherit").
181fn resolve_spawn_cwd(cwd: Option<&PathBuf>) -> Option<PathBuf> {
182 // An explicit directory is always honored as-is.
183 if let Some(c) = cwd {
184 return Some(c.clone());
185 }
186
187 // No explicit cwd: we would inherit the parent's working directory. Make
188 // sure that directory is actually usable before relying on inheritance.
189 match std::env::current_dir() {
190 Ok(_) => None,
191 Err(e) => {
192 let fallback = fallback_cwd();
193 trace(
194 "ProcessRunner",
195 &format!(
196 "current_dir() failed ({}); spawning in fallback directory {}",
197 e,
198 fallback.display()
199 ),
200 );
201 Some(fallback)
202 }
203 }
204}
205
206/// Error type for command-stream operations
207#[derive(Debug, thiserror::Error)]
208pub enum Error {
209 #[error("IO error: {0}")]
210 Io(#[from] std::io::Error),
211
212 #[error("Command failed with exit code {code}: {message}")]
213 CommandFailed { code: i32, message: String },
214
215 #[error("Command not found: {0}")]
216 CommandNotFound(String),
217
218 #[error("Parse error: {0}")]
219 ParseError(String),
220
221 #[error("Cancelled")]
222 Cancelled,
223}
224
225impl Error {
226 /// Build a [`Error::CommandFailed`] for a command that exited with `code`.
227 pub fn command_failed(code: i32, message: impl Into<String>) -> Self {
228 Error::CommandFailed {
229 code,
230 message: message.into(),
231 }
232 }
233
234 /// Exit status carried by the error, when the failure has one.
235 ///
236 /// Mirrors the `error.code` property of the JavaScript implementation
237 /// (issue #38). Failures that never reached a child process, such as parse
238 /// errors, report `None`.
239 pub fn code(&self) -> Option<i32> {
240 match self {
241 Error::CommandFailed { code, .. } => Some(*code),
242 // `command not found` is 127 in POSIX shells, which is also what
243 // the JavaScript implementation reports for a missing executable.
244 Error::CommandNotFound(_) => Some(127),
245 Error::Io(error) => match error.kind() {
246 std::io::ErrorKind::NotFound => Some(127),
247 std::io::ErrorKind::PermissionDenied => Some(126),
248 _ => None,
249 },
250 // A cancelled command is terminated with SIGINT (128 + 2).
251 Error::Cancelled => Some(130),
252 Error::ParseError(_) => None,
253 }
254 }
255
256 /// Alias for [`code`](Self::code).
257 ///
258 /// Node.js `child_process` names this property `code`, while execa, zx,
259 /// nano-spawn, and Bun Shell name it `exitCode`. command-stream exposes
260 /// both spellings in every language (issue #38).
261 pub fn exit_code(&self) -> Option<i32> {
262 self.code()
263 }
264}
265
266/// Result type for command-stream operations
267pub type Result<T> = std::result::Result<T, Error>;
268
269/// Options for command execution
270#[derive(Debug, Clone)]
271pub struct RunOptions {
272 /// Mirror output to parent stdout/stderr
273 pub mirror: bool,
274 /// Capture output in result
275 pub capture: bool,
276 /// Standard input handling
277 pub stdin: StdinOption,
278 /// Working directory
279 pub cwd: Option<PathBuf>,
280 /// Environment variables
281 pub env: Option<HashMap<String, String>>,
282 /// Interactive mode (TTY forwarding)
283 pub interactive: bool,
284 /// Enable shell operator parsing
285 pub shell_operators: bool,
286 /// Enable tracing for this command
287 pub trace: bool,
288 /// Signal used to stop the process when it is killed without an explicit
289 /// signal, i.e. [`ProcessRunner::kill`] (default `SIGTERM`).
290 ///
291 /// Mirrors the JavaScript `killSignal` option. An explicit
292 /// [`ProcessRunner::kill_with`] argument always overrides it.
293 pub kill_signal: String,
294 /// Milliseconds the child is given to handle the kill signal before
295 /// `SIGKILL` is sent (default 100).
296 ///
297 /// Mirrors the JavaScript `killGrace` option. This is the window in which a
298 /// child running its own signal handler can shut down on its own terms.
299 pub kill_grace_ms: u64,
300}
301
302impl Default for RunOptions {
303 fn default() -> Self {
304 RunOptions {
305 mirror: true,
306 capture: true,
307 stdin: StdinOption::Inherit,
308 cwd: None,
309 env: None,
310 interactive: false,
311 shell_operators: true,
312 trace: true,
313 kill_signal: signal::DEFAULT_KILL_SIGNAL.to_string(),
314 kill_grace_ms: signal::DEFAULT_KILL_GRACE_MS,
315 }
316 }
317}
318
319/// Standard input options
320#[derive(Debug, Clone)]
321pub enum StdinOption {
322 /// Inherit from parent process
323 Inherit,
324 /// Pipe (allow writing to stdin)
325 Pipe,
326 /// Provide string content
327 Content(String),
328 /// Null device
329 Null,
330}
331
332/// A running or completed process
333pub struct ProcessRunner {
334 command: String,
335 options: RunOptions,
336 child: Option<Child>,
337 /// Process id of the spawned child, recorded at spawn time. `run()` takes
338 /// the child in order to await it, so reading the id from it only works
339 /// between `start()` and `run()`; this copy is what makes `pid()` answer
340 /// after the command has finished too (issue #18).
341 pid: Option<u32>,
342 result: Option<CommandResult>,
343 started: bool,
344 finished: bool,
345 cancelled: bool,
346 /// Whether the child was spawned into a process group of its own, and so
347 /// can be signalled as a group. Recorded at spawn time because it cannot be
348 /// discovered later: by the time the group is signalled the leader is
349 /// usually a zombie, which macOS refuses to answer `getpgid` for.
350 #[cfg(unix)]
351 own_process_group: bool,
352 output_tx: Option<mpsc::Sender<StreamChunk>>,
353 // Held, never read: dropping the receiver would close the channel, and
354 // streaming virtual commands treat a closed channel as "stop now" (see
355 // `commands::yes`, which loops until `output_tx.send` fails). Keeping it
356 // alive is what gives those commands their run-until-cancelled behaviour.
357 #[allow(dead_code)]
358 output_rx: Option<mpsc::Receiver<StreamChunk>>,
359}
360
361impl ProcessRunner {
362 /// Create a new process runner
363 pub fn new(command: impl Into<String>, options: RunOptions) -> Self {
364 let (tx, rx) = mpsc::channel(1024);
365 ProcessRunner {
366 command: command.into(),
367 options,
368 child: None,
369 pid: None,
370 result: None,
371 started: false,
372 finished: false,
373 cancelled: false,
374 #[cfg(unix)]
375 own_process_group: false,
376 output_tx: Some(tx),
377 output_rx: Some(rx),
378 }
379 }
380
381 /// Whether the child will read from the caller's terminal.
382 ///
383 /// Only an *inherited* stdin that is actually a tty counts: a pipe, a null
384 /// stdin, or inherited stdin that has been redirected to a file carries no
385 /// terminal, and neither does output-only inheritance. This is the one case
386 /// where the child must stay in the caller's process group.
387 #[cfg(unix)]
388 fn shares_the_terminal(&self) -> bool {
389 use std::io::IsTerminal;
390
391 self.options.interactive
392 || (matches!(self.options.stdin, StdinOption::Inherit)
393 && std::io::stdin().is_terminal())
394 }
395
396 /// Start the process
397 pub async fn start(&mut self) -> Result<()> {
398 if self.started {
399 return Ok(());
400 }
401 self.started = true;
402
403 utils::trace_lazy("ProcessRunner", || {
404 format!("Starting command: {}", self.command)
405 });
406
407 // Check if this is a virtual command. Backslash escapes are removed by a
408 // real shell but not by the whitespace splitting used for virtual
409 // command args, so such commands always go to the system shell (#49).
410 // The same applies to redirection and expansions: whitespace splitting
411 // would hand `>`, `out.txt` to the virtual command as two ordinary
412 // arguments, so `echo hello > out.txt` printed the redirection instead
413 // of writing the file, and `git push ... 2>&1` reported success while
414 // nothing was pushed (#46).
415 let first_word = if matches!(self.options.stdin, StdinOption::Pipe)
416 || has_shell_escapes(&self.command)
417 || needs_real_shell(&self.command)
418 {
419 ""
420 } else {
421 self.command.split_whitespace().next().unwrap_or("")
422 };
423 if let Some(result) = self.try_virtual_command(first_word).await {
424 self.result = Some(result);
425 self.finished = true;
426 return Ok(());
427 }
428 // Parse command for shell operators (for future use with virtual command pipelines)
429 let _parsed = if self.options.shell_operators && !needs_real_shell(&self.command) {
430 parse_shell_command(&self.command)
431 } else {
432 None
433 };
434
435 // Execute via real shell if needed
436 let mut cmd = utils::shell_command(&self.command, self.options.env.as_ref());
437
438 // Configure stdin
439 match &self.options.stdin {
440 StdinOption::Inherit => {
441 cmd.stdin(Stdio::inherit());
442 }
443 StdinOption::Pipe => {
444 cmd.stdin(Stdio::piped());
445 }
446 StdinOption::Content(_) => {
447 cmd.stdin(Stdio::piped());
448 }
449 StdinOption::Null => {
450 cmd.stdin(Stdio::null());
451 }
452 }
453
454 // Configure stdout/stderr
455 if self.options.capture || self.options.mirror {
456 cmd.stdout(Stdio::piped());
457 cmd.stderr(Stdio::piped());
458 } else {
459 cmd.stdout(Stdio::inherit());
460 cmd.stderr(Stdio::inherit());
461 }
462
463 // Set working directory. Fall back to a valid directory when the
464 // inherited working directory has been deleted (issue #44).
465 if let Some(cwd) = resolve_spawn_cwd(self.options.cwd.as_ref()) {
466 cmd.current_dir(cwd);
467 }
468
469 // Set environment
470 if let Some(ref env_vars) = self.options.env {
471 for (key, value) in env_vars {
472 cmd.env(key, value);
473 }
474 }
475
476 // Run the child in its own process group so that killing it can signal
477 // the whole group (parent + grandchildren), matching `StreamingRunner`
478 // and the JavaScript implementation's `detached` spawn.
479 //
480 // A child that shares the terminal is deliberately left in the caller's
481 // group. The tty delivers CTRL+C to its foreground group only, so
482 // moving such a child out would both hide CTRL+C from it and stop it
483 // with SIGTTIN the moment it read from the terminal. JavaScript draws
484 // the same line, spawning interactive commands without `detached`.
485 #[cfg(unix)]
486 {
487 self.own_process_group = !self.shares_the_terminal();
488 if self.own_process_group {
489 cmd.process_group(0);
490 }
491 }
492
493 // Spawn the process
494 let child = cmd.spawn()?;
495 // Record the id while the child is still held. `run()` takes the child
496 // in order to await it, so this copy is what keeps `pid()` readable
497 // afterwards.
498 self.pid = child.id();
499 self.child = Some(child);
500
501 Ok(())
502 }
503
504 /// Write bytes to the stdin pipe of a running command.
505 ///
506 /// Configure the runner with [`StdinOption::Pipe`], call [`start`](Self::start),
507 /// write as many chunks as needed, and finish with [`close_stdin`](Self::close_stdin).
508 pub async fn write_stdin(&mut self, data: impl AsRef<[u8]>) -> Result<()> {
509 self.start().await?;
510 let stdin = self
511 .child
512 .as_mut()
513 .and_then(|child| child.stdin.as_mut())
514 .ok_or_else(|| {
515 Error::Io(std::io::Error::new(
516 std::io::ErrorKind::BrokenPipe,
517 "command stdin is not available; use StdinOption::Pipe",
518 ))
519 })?;
520 stdin.write_all(data.as_ref()).await?;
521 Ok(())
522 }
523
524 /// Close a running command's stdin pipe so it can observe end-of-input.
525 pub async fn close_stdin(&mut self) -> Result<()> {
526 self.start().await?;
527 if let Some(mut stdin) = self.child.as_mut().and_then(|child| child.stdin.take()) {
528 stdin.shutdown().await?;
529 }
530 Ok(())
531 }
532
533 /// Run the process to completion
534 pub async fn run(&mut self) -> Result<CommandResult> {
535 self.start().await?;
536
537 if let Some(result) = &self.result {
538 return Ok(result.clone());
539 }
540
541 let mut child = self
542 .child
543 .take()
544 .ok_or_else(|| Error::Io(std::io::Error::other("Process not started")))?;
545
546 // Handle stdin content if provided
547 if let StdinOption::Content(ref content) = self.options.stdin {
548 if let Some(mut stdin) = child.stdin.take() {
549 let content = content.clone();
550 tokio::spawn(async move {
551 let _ = stdin.write_all(content.as_bytes()).await;
552 let _ = stdin.shutdown().await;
553 });
554 }
555 }
556
557 // Drain both pipes concurrently and preserve their newline framing. The
558 // previous line reader appended `\n` to every final line, changing
559 // output from commands such as `printf` that omit a newline (issue #37).
560 let stdout = child.stdout.take();
561 let stderr = child.stderr.take();
562 let collected = tokio::try_join!(
563 collect_child_output(stdout, self.options.mirror, ChildOutput::Stdout),
564 collect_child_output(stderr, self.options.mirror, ChildOutput::Stderr),
565 );
566 let (stdout, stderr) = match collected {
567 Ok(output) => output,
568 Err(error) => {
569 // `try_join!` drops the other pipe reader after an error. Stop
570 // and reap the child so it cannot remain blocked on that pipe.
571 let _ = child.start_kill();
572 let _ = child.wait().await;
573 return Err(error.into());
574 }
575 };
576
577 let status = child.wait().await?;
578 let code = status.code().unwrap_or(-1);
579
580 let result = CommandResult {
581 stdout: String::from_utf8_lossy(&stdout).into_owned(),
582 stderr: String::from_utf8_lossy(&stderr).into_owned(),
583 code,
584 };
585
586 self.result = Some(result.clone());
587 self.finished = true;
588
589 Ok(result)
590 }
591
592 /// Try to execute as a virtual command
593 async fn try_virtual_command(&self, cmd_name: &str) -> Option<CommandResult> {
594 if !commands::are_virtual_commands_enabled() {
595 return None;
596 }
597
598 // An empty command name means the caller already decided this command
599 // must go to a real shell (redirection, expansions, escapes). Bail out
600 // before tokenizing so we neither waste work nor parse shell syntax we
601 // deliberately delegate.
602 if cmd_name.is_empty() {
603 return None;
604 }
605
606 // Parse args from command string, respecting quotes and performing
607 // POSIX quote removal so `echo label:'help wanted'` reaches the built-in
608 // as the single argument `label:help wanted` (issue #48).
609 let words = shell_parser::split_command_words(&self.command);
610 let args: Vec<String> = words.into_iter().skip(1).collect();
611
612 let ctx = CommandContext {
613 args,
614 stdin: match &self.options.stdin {
615 StdinOption::Content(s) => Some(s.clone()),
616 _ => None,
617 },
618 cwd: self.options.cwd.clone(),
619 env: self.options.env.clone(),
620 output_tx: self.output_tx.clone(),
621 is_cancelled: None,
622 };
623
624 match cmd_name {
625 "echo" => Some(commands::echo(ctx).await),
626 "pwd" => Some(commands::pwd(ctx).await),
627 "cd" => Some(commands::cd::resolve_cd(ctx).await.0),
628 "true" => Some(commands::r#true(ctx).await),
629 "false" => Some(commands::r#false(ctx).await),
630 "sleep" => Some(commands::sleep(ctx).await),
631 "cat" => Some(commands::cat(ctx).await),
632 "ls" => Some(commands::ls(ctx).await),
633 "mkdir" => Some(commands::mkdir(ctx).await),
634 "rm" => Some(commands::rm(ctx).await),
635 "touch" => Some(commands::touch(ctx).await),
636 "cp" => Some(commands::cp(ctx).await),
637 "mv" => Some(commands::mv(ctx).await),
638 "basename" => Some(commands::basename(ctx).await),
639 "dirname" => Some(commands::dirname(ctx).await),
640 "env" => Some(commands::env(ctx).await),
641 "exit" => Some(commands::exit(ctx).await),
642 "which" => Some(commands::which(ctx).await),
643 "yes" => Some(commands::yes(ctx).await),
644 "seq" => Some(commands::seq(ctx).await),
645 "tee" => Some(commands::tee(ctx).await),
646 "test" => Some(commands::test(ctx).await),
647 _ => None,
648 }
649 }
650
651 /// Stop the process using the configured kill signal
652 /// ([`RunOptions::kill_signal`], default `SIGTERM`).
653 ///
654 /// Mirrors the JavaScript `kill()` with no argument.
655 pub fn kill(&mut self) -> Result<()> {
656 let signal = self.options.kill_signal.clone();
657 self.kill_with(&signal)
658 }
659
660 /// Stop the process using an explicit signal, overriding
661 /// [`RunOptions::kill_signal`] for this call.
662 ///
663 /// Mirrors the JavaScript `kill(signal)`. The signal is delivered to the
664 /// child and its process group, so grandchildren behind a `sh -c` wrapper
665 /// are stopped too - except for a child sharing the caller's terminal,
666 /// which stays in the caller's group so CTRL+C keeps reaching it. The child
667 /// then has [`RunOptions::kill_grace_ms`] to run its own handler before
668 /// `SIGKILL` follows, so a process that ignores the signal still
669 /// terminates.
670 ///
671 /// ```no_run
672 /// use command_stream::{ProcessRunner, RunOptions};
673 ///
674 /// # #[tokio::main]
675 /// # async fn main() -> command_stream::Result<()> {
676 /// let mut runner = ProcessRunner::new("sleep 30", RunOptions::default());
677 /// runner.start().await?;
678 /// runner.kill_with("SIGINT")?; // the CTRL+C signal
679 /// # Ok(())
680 /// # }
681 /// ```
682 pub fn kill_with(&mut self, signal: &str) -> Result<()> {
683 self.cancelled = true;
684 utils::trace_lazy("ProcessRunner", || format!("kill | signal={signal}"));
685
686 let Some(child) = self.child.as_mut() else {
687 return Ok(());
688 };
689
690 // Windows has no signals to deliver and no handler for the child to
691 // run, so there is nothing to grant a grace period to: the forceful
692 // stop is the only way to end the process.
693 // The `#[cfg(unix)]` block below is stripped on Windows, which leaves
694 // this one as the function's tail expression - hence no `return`.
695 #[cfg(not(unix))]
696 {
697 let _ = signal;
698 child.start_kill()?;
699 Ok(())
700 }
701
702 // Without a pid the process never spawned (or was already reaped);
703 // fall back to the forceful stop so `kill()` still terminates it.
704 #[cfg(unix)]
705 {
706 let Some(pid) = child.id() else {
707 child.start_kill()?;
708 return Ok(());
709 };
710
711 // `SIGKILL` cannot be handled, so there is nothing to wait for.
712 //
713 // A zero grace period means the child is given no opportunity to
714 // handle the signal either, so the requested signal is not
715 // delivered at all. Anything done between it and `SIGKILL` - even a
716 // single syscall - is a window the child can be scheduled in, which
717 // made "no grace" a race the child occasionally won rather than a
718 // guarantee. The reported exit code still comes from the signal
719 // that was requested.
720 let grace = self.options.kill_grace_ms;
721 let delivery = if self.own_process_group {
722 signal::Delivery::ProcessAndGroup
723 } else {
724 signal::Delivery::ProcessOnly
725 };
726 if grace == 0 || signal == "SIGKILL" {
727 signal::send_signal_to_process(pid, "SIGKILL", delivery);
728 let _ = child.start_kill();
729 return Ok(());
730 }
731
732 signal::send_signal_to_process(pid, signal, delivery);
733
734 // Otherwise escalate in the background so the child keeps its grace
735 // period without blocking the caller, which may not be inside an
736 // await point.
737 tokio::spawn(async move {
738 tokio::time::sleep(std::time::Duration::from_millis(grace)).await;
739 // Best effort: if the child already exited on the first signal
740 // this delivery simply fails, and the pid has not been reused
741 // because the `Child` handle above has not reaped it yet. That
742 // unreaped leader is also what keeps the group id alive, so the
743 // group delivery still reaches a grandchild that outlived it.
744 signal::send_signal_to_process(pid, "SIGKILL", delivery);
745 });
746
747 Ok(())
748 }
749 }
750
751 /// Check if the process is finished
752 pub fn is_finished(&self) -> bool {
753 self.finished
754 }
755
756 /// Get the result if available
757 pub fn result(&self) -> Option<&CommandResult> {
758 self.result.as_ref()
759 }
760
761 /// Process id of the command, or `None` when there is no operating system
762 /// process to identify.
763 ///
764 /// It is `None` before the command starts, and stays `None` for built-in
765 /// (virtual) commands such as `echo` or `sleep`, which run inside this
766 /// process and never spawn a child. Once a real command has been spawned
767 /// the value is stable: it remains readable after the command finishes,
768 /// unlike the child handle, which [`run`](Self::run) consumes.
769 ///
770 /// Mirrors the JavaScript `runner.pid` property.
771 ///
772 /// ```no_run
773 /// use command_stream::{ProcessRunner, RunOptions};
774 ///
775 /// # #[tokio::main]
776 /// # async fn main() -> command_stream::Result<()> {
777 /// let mut runner = ProcessRunner::new("/bin/sleep 1", RunOptions::default());
778 /// runner.start().await?;
779 /// println!("running as pid {:?}", runner.pid());
780 /// runner.run().await?;
781 /// println!("still readable: {:?}", runner.pid());
782 /// # Ok(())
783 /// # }
784 /// ```
785 pub fn pid(&self) -> Option<u32> {
786 self.pid
787 }
788
789 /// Get the command string
790 pub fn command(&self) -> &str {
791 &self.command
792 }
793
794 /// Get the options
795 pub fn options(&self) -> &RunOptions {
796 &self.options
797 }
798}
799
800/// Execute a command and return the result
801///
802/// This is the main entry point for simple command execution.
803/// Named `run` instead of `$` since `$` is not a valid Rust identifier.
804pub async fn run(command: impl Into<String>) -> Result<CommandResult> {
805 let mut runner = ProcessRunner::new(command, RunOptions::default());
806 runner.run().await
807}
808
809/// Alias for `run` function - for JavaScript-like API feel
810/// Since `$` is not valid in Rust, this provides a similar short name
811pub use run as execute;
812
813/// Execute a command with custom options
814pub async fn exec(command: impl Into<String>, options: RunOptions) -> Result<CommandResult> {
815 let mut runner = ProcessRunner::new(command, options);
816 runner.run().await
817}
818
819/// Create a new process runner without starting it
820pub fn create(command: impl Into<String>, options: RunOptions) -> ProcessRunner {
821 ProcessRunner::new(command, options)
822}
823
824/// Execute a command synchronously (blocking)
825pub fn run_sync(command: impl Into<String>) -> Result<CommandResult> {
826 let rt = tokio::runtime::Runtime::new()?;
827 rt.block_on(run(command))
828}
829
830// Tests are located in tests/ directory for better organization