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 bun_shell;
67pub mod command_builder;
68pub mod command_packages;
69pub mod development;
70pub use command_packages::{CommandEnvironment, CommandPackage, PackageHandler};
71pub use development::{
72 dev, start_repl, DevController, DevEvent, DevOptions, ReplOptions, RustRepl,
73};
74mod detach;
75pub mod events;
76pub mod execa;
77pub mod fy;
78pub mod local_bin;
79#[doc(hidden)]
80pub mod macros;
81pub mod pipeline;
82mod process_argv;
83mod process_stop;
84pub mod quote;
85pub mod result_streams;
86mod run_options;
87pub mod signal;
88pub mod state;
89pub mod stream;
90mod stream_process;
91pub mod stream_processing;
92mod stream_stdin;
93pub mod terminal;
94pub mod trace;
95
96// Core modules
97pub mod commands;
98pub mod shell_parser;
99pub mod shelljs;
100pub mod utils;
101
102// zx-compatible API (google/zx 8.x): `$`-style Shell, ProcessOutput, goods.
103pub mod zx;
104
105use std::path::PathBuf;
106use std::process::Stdio;
107use tokio::io::{AsyncRead, AsyncReadExt, AsyncWriteExt};
108use tokio::process::Child;
109use tokio::sync::mpsc;
110
111pub use command_builder::{fy, CommandBuilder, CommandStdio};
112pub use commands::{CommandContext, StreamChunk};
113pub use execa::{execa, execa_compat, execa_node, execa_sync};
114pub use run_options::{Error, Result, RunOptions, StdinOption};
115pub use shell_parser::{needs_real_shell, parse_shell_command, ParsedCommand};
116pub use utils::{CommandResult, VirtualUtils};
117
118// Re-export modular utilities at crate root for convenient access
119pub use ansi::{AnsiConfig, AnsiUtils};
120pub use events::{EventData, EventType, StreamEmitter};
121pub use local_bin::PreferLocal;
122pub use pipeline::{Pipeline, PipelineBuilder, PipelineExt};
123pub use quote::{
124 escape_for_double_quotes, escape_for_single_quotes, has_shell_escapes,
125 is_pre_quoted_passthrough_enabled, is_quote_context_enabled, quote, quote_for_context,
126 scan_quote_context, QuoteContext,
127};
128pub use signal::{signal_exit_code, signal_number, DEFAULT_KILL_GRACE_MS, DEFAULT_KILL_SIGNAL};
129pub use state::{
130 get_shell_settings, global_state, reset_global_state, set_shell_option, unset_shell_option,
131 GlobalState, ShellSettings,
132};
133pub use stream::{AsyncIterator, IntoStream, OutputChunk, OutputStream, StreamingRunner};
134pub use stream_processing::{DataStream, StreamStats};
135pub use trace::trace;
136
137#[derive(Clone, Copy)]
138enum ChildOutput {
139 Stdout,
140 Stderr,
141}
142
143/// Read child output as byte chunks so capture does not invent a trailing newline.
144///
145/// stdout and stderr use separate futures in `ProcessRunner::run`, preventing
146/// either pipe from filling while the other is being drained. Mirroring keeps
147/// the original bytes too, including output that does not end in a newline.
148async fn collect_child_output<R>(
149 reader: Option<R>,
150 mirror: bool,
151 target: ChildOutput,
152 collected: &mut Vec<u8>,
153) -> std::io::Result<()>
154where
155 R: AsyncRead + Unpin,
156{
157 let Some(mut reader) = reader else {
158 return Ok(());
159 };
160 let mut buffer = [0_u8; 8192];
161
162 loop {
163 let count = reader.read(&mut buffer).await?;
164 if count == 0 {
165 break;
166 }
167
168 let chunk = &buffer[..count];
169 collected.extend_from_slice(chunk);
170 if mirror {
171 match target {
172 ChildOutput::Stdout => {
173 let mut output = std::io::stdout().lock();
174 let _ = std::io::Write::write_all(&mut output, chunk);
175 let _ = std::io::Write::flush(&mut output);
176 }
177 ChildOutput::Stderr => {
178 let mut output = std::io::stderr().lock();
179 let _ = std::io::Write::write_all(&mut output, chunk);
180 let _ = std::io::Write::flush(&mut output);
181 }
182 }
183 }
184 }
185
186 Ok(())
187}
188
189fn fallback_cwd() -> PathBuf {
190 std::env::var_os("HOME")
191 .or_else(|| std::env::var_os("USERPROFILE"))
192 .map(PathBuf::from)
193 .filter(|path| path.is_dir())
194 .unwrap_or_else(std::env::temp_dir)
195}
196
197/// Resolve a working directory that is safe to spawn a child process in.
198///
199/// When no explicit cwd is requested the child normally inherits the parent's
200/// working directory. But if that directory has been deleted or become
201/// inaccessible (the "getcwd() failed" scenario from issue #44), inheriting it
202/// makes the OS-level spawn fail. In that case fall back to a directory that is
203/// known to exist so the command still runs.
204///
205/// Normal behavior is preserved: when an explicit cwd is given, or when the
206/// inherited working directory is valid, this returns the requested value
207/// (`None` meaning "inherit").
208pub(crate) fn resolve_spawn_cwd(cwd: Option<&PathBuf>) -> Option<PathBuf> {
209 // An explicit directory is always honored as-is.
210 if let Some(c) = cwd {
211 return Some(c.clone());
212 }
213
214 // No explicit cwd: we would inherit the parent's working directory. Make
215 // sure that directory is actually usable before relying on inheritance.
216 match std::env::current_dir() {
217 Ok(_) => None,
218 Err(e) => {
219 let fallback = fallback_cwd();
220 trace(
221 "ProcessRunner",
222 &format!(
223 "current_dir() failed ({}); spawning in fallback directory {}",
224 e,
225 fallback.display()
226 ),
227 );
228 Some(fallback)
229 }
230 }
231}
232
233/// A running or completed process
234pub struct ProcessRunner {
235 command: String,
236 /// Exact program and arguments, set by [`ProcessRunner::from_argv`]. When
237 /// present the command bypasses the shell and the virtual commands.
238 argv: Option<stream_process::StreamingCommand>,
239 options: RunOptions,
240 child: Option<Child>,
241 stdin_bytes: Vec<u8>,
242 /// Process id of the spawned child, recorded at spawn time. `run()` takes
243 /// the child in order to await it, so reading the id from it only works
244 /// between `start()` and `run()`; this copy is what makes `pid()` answer
245 /// after the command has finished too (issue #18).
246 pid: Option<u32>,
247 result: Option<CommandResult>,
248 started: bool,
249 finished: bool,
250 cancelled: bool,
251 /// When [`RunOptions::timeout`] stops the command, counted from the start.
252 deadline: Option<tokio::time::Instant>,
253 timed_out: bool,
254 /// Whether the child was spawned into a process group of its own, and so
255 /// can be signalled as a group. Recorded at spawn time because it cannot be
256 /// discovered later: by the time the group is signalled the leader is
257 /// usually a zombie, which macOS refuses to answer `getpgid` for.
258 #[cfg(unix)]
259 own_process_group: bool,
260 output_tx: Option<mpsc::Sender<StreamChunk>>,
261 // Held, never read: dropping the receiver would close the channel, and
262 // streaming virtual commands treat a closed channel as "stop now" (see
263 // `commands::yes`, which loops until `output_tx.send` fails). Keeping it
264 // alive is what gives those commands their run-until-cancelled behaviour.
265 #[allow(dead_code)]
266 output_rx: Option<mpsc::Receiver<StreamChunk>>,
267}
268
269/// Borrowed access to the operating-system child owned by a [`ProcessRunner`].
270///
271/// The wrapper keeps process termination on the runner's signal-aware path:
272/// [`kill`](Self::kill) and [`kill_with`](Self::kill_with) signal the child and
273/// its process group, honor the configured grace period, and then escalate if
274/// necessary. Use [`native`](Self::native) or [`native_mut`](Self::native_mut)
275/// when direct access to Tokio's child object is required.
276pub struct ProcessChild<'a> {
277 runner: &'a mut ProcessRunner,
278}
279
280impl ProcessChild<'_> {
281 /// Process id of the active child.
282 pub fn pid(&self) -> Option<u32> {
283 self.native().id()
284 }
285
286 /// Borrow Tokio's native child process object.
287 pub fn native(&self) -> &Child {
288 self.runner
289 .child
290 .as_ref()
291 .expect("ProcessChild exists only while its native child is present")
292 }
293
294 /// Mutably borrow Tokio's native child process object.
295 pub fn native_mut(&mut self) -> &mut Child {
296 self.runner
297 .child
298 .as_mut()
299 .expect("ProcessChild exists only while its native child is present")
300 }
301
302 /// Stop the child using the runner's configured signal and grace period.
303 pub fn kill(&mut self) -> Result<()> {
304 self.runner.kill()
305 }
306
307 /// Stop the child using an explicit signal and the configured grace period.
308 pub fn kill_with(&mut self, signal: &str) -> Result<()> {
309 self.runner.kill_with(signal)
310 }
311}
312
313impl ProcessRunner {
314 /// Create a new process runner
315 pub fn new(command: impl Into<String>, options: RunOptions) -> Self {
316 let (tx, rx) = mpsc::channel(1024);
317 ProcessRunner {
318 command: command.into(),
319 argv: None,
320 options,
321 child: None,
322 stdin_bytes: Vec::new(),
323 pid: None,
324 result: None,
325 started: false,
326 finished: false,
327 cancelled: false,
328 deadline: None,
329 timed_out: false,
330 #[cfg(unix)]
331 own_process_group: false,
332 output_tx: Some(tx),
333 output_rx: Some(rx),
334 }
335 }
336
337 /// Whether the child will read from the caller's terminal.
338 ///
339 /// Only an *inherited* stdin that is actually a tty counts: a pipe, a null
340 /// stdin, or inherited stdin that has been redirected to a file carries no
341 /// terminal, and neither does output-only inheritance. This is the one case
342 /// where the child must stay in the caller's process group.
343 #[cfg(unix)]
344 fn shares_the_terminal(&self) -> bool {
345 use std::io::IsTerminal;
346
347 self.options.interactive
348 || (matches!(self.options.stdin, StdinOption::Inherit)
349 && std::io::stdin().is_terminal())
350 }
351
352 /// Start the process
353 pub async fn start(&mut self) -> Result<()> {
354 if self.started {
355 return Ok(());
356 }
357 self.started = true;
358 self.deadline = self
359 .options
360 .timeout
361 .map(|limit| tokio::time::Instant::now() + limit);
362
363 utils::trace_lazy("ProcessRunner", || {
364 format!("Starting command: {}", self.command)
365 });
366
367 // Check if this is a virtual command. Backslash escapes are removed by a
368 // real shell but not by the whitespace splitting used for virtual
369 // command args, so such commands always go to the system shell (#49).
370 // The same applies to redirection and expansions: whitespace splitting
371 // would hand `>`, `out.txt` to the virtual command as two ordinary
372 // arguments, so `echo hello > out.txt` printed the redirection instead
373 // of writing the file, and `git push ... 2>&1` reported success while
374 // nothing was pushed (#46).
375 // Removals cannot be expressed to the built-ins, which see `env` as
376 // overrides of the inherited environment, so they go to a real shell.
377 let first_word = if self.argv.is_some()
378 || matches!(self.options.stdin, StdinOption::Pipe)
379 || !self.options.env_remove.is_empty()
380 || has_shell_escapes(&self.command)
381 || needs_real_shell(&self.command)
382 {
383 ""
384 } else {
385 self.command.split_whitespace().next().unwrap_or("")
386 };
387 let virtual_result = match self.deadline {
388 Some(deadline) => {
389 tokio::time::timeout_at(deadline, self.try_virtual_command(first_word))
390 .await
391 .unwrap_or_else(|_| {
392 self.timed_out = true;
393 let code = signal::signal_exit_code(&self.options.kill_signal);
394 Some(CommandResult::new("", "", code))
395 })
396 }
397 None => self.try_virtual_command(first_word).await,
398 };
399 if let Some(mut result) = virtual_result {
400 if let StdinOption::Content(ref content) = self.options.stdin {
401 result.stdin =
402 crate::result_streams::CapturedInput::new(content.as_bytes().to_vec());
403 }
404 self.result = Some(result);
405 self.finished = true;
406 return Ok(());
407 }
408 // Parse command for shell operators (for future use with virtual command pipelines)
409 let _parsed = if self.options.shell_operators && !needs_real_shell(&self.command) {
410 parse_shell_command(&self.command)
411 } else {
412 None
413 };
414
415 // Execute via real shell if needed
416 let mut cmd = match &self.argv {
417 Some(argv) => argv.command(None),
418 None => utils::shell_command(&self.command, self.options.env.as_ref()),
419 };
420
421 // Configure stdin
422 match &self.options.stdin {
423 StdinOption::Inherit => {
424 cmd.stdin(Stdio::inherit());
425 }
426 StdinOption::Pipe => {
427 cmd.stdin(Stdio::piped());
428 }
429 StdinOption::Content(_) => {
430 cmd.stdin(Stdio::piped());
431 }
432 StdinOption::Null => {
433 cmd.stdin(Stdio::null());
434 }
435 }
436
437 // Configure stdout/stderr
438 if self.options.capture || self.options.mirror {
439 cmd.stdout(Stdio::piped());
440 cmd.stderr(Stdio::piped());
441 } else {
442 cmd.stdout(Stdio::inherit());
443 cmd.stderr(Stdio::inherit());
444 }
445
446 // Set working directory. Fall back to a valid directory when the
447 // inherited working directory has been deleted (issue #44).
448 if let Some(cwd) = resolve_spawn_cwd(self.options.cwd.as_ref()) {
449 cmd.current_dir(cwd);
450 }
451
452 // Set environment
453 if let Some(ref env_vars) = self.options.env {
454 for (key, value) in env_vars {
455 cmd.env(key, value);
456 }
457 }
458 let local_cwd = self
459 .options
460 .cwd
461 .clone()
462 .or_else(|| std::env::current_dir().ok())
463 .unwrap_or_else(|| PathBuf::from("."));
464 local_bin::apply_prefer_local(
465 &mut cmd,
466 self.options.env.as_ref(),
467 &local_cwd,
468 &self.options.prefer_local,
469 );
470 for name in &self.options.env_remove {
471 cmd.env_remove(name);
472 }
473
474 // Run the child in its own process group so that killing it can signal
475 // the whole group (parent + grandchildren), matching `StreamingRunner`
476 // and the JavaScript implementation's `detached` spawn.
477 //
478 // A child that shares the terminal is deliberately left in the caller's
479 // group. The tty delivers CTRL+C to its foreground group only, so
480 // moving such a child out would both hide CTRL+C from it and stop it
481 // with SIGTTIN the moment it read from the terminal. JavaScript draws
482 // the same line, spawning interactive commands without `detached`.
483 #[cfg(unix)]
484 {
485 self.own_process_group = !self.shares_the_terminal();
486 if self.own_process_group {
487 cmd.process_group(0);
488 }
489 }
490
491 // Spawn the process
492 let child = cmd.spawn()?;
493 // Record the id while the child is still held. `run()` takes the child
494 // in order to await it, so this copy is what keeps `pid()` readable
495 // afterwards.
496 self.pid = child.id();
497 self.child = Some(child);
498
499 Ok(())
500 }
501
502 /// Borrow the active operating-system child.
503 ///
504 /// Call [`start`](Self::start) first. The result is `None` before startup,
505 /// for built-in commands (which run in-process), and after [`run`](Self::run)
506 /// consumes and reaps the child. Killing through the returned handle keeps
507 /// the runner's process-group and graceful-escalation behavior.
508 ///
509 /// ```no_run
510 /// use command_stream::{ProcessRunner, RunOptions};
511 ///
512 /// # #[tokio::main]
513 /// # async fn main() -> command_stream::Result<()> {
514 /// let mut runner = ProcessRunner::new("sleep 30", RunOptions::default());
515 /// runner.start().await?;
516 /// if let Some(mut child) = runner.child() {
517 /// println!("child pid: {:?}", child.pid());
518 /// child.kill_with("SIGTERM")?;
519 /// }
520 /// let _ = runner.run().await?;
521 /// # Ok(())
522 /// # }
523 /// ```
524 pub fn child(&mut self) -> Option<ProcessChild<'_>> {
525 self.child.as_ref()?;
526 Some(ProcessChild { runner: self })
527 }
528
529 /// Write bytes to the stdin pipe of a running command.
530 ///
531 /// Configure the runner with [`StdinOption::Pipe`], call [`start`](Self::start),
532 /// write as many chunks as needed, and finish with [`close_stdin`](Self::close_stdin).
533 pub async fn write_stdin(&mut self, data: impl AsRef<[u8]>) -> Result<()> {
534 self.start().await?;
535 let stdin = self
536 .child
537 .as_mut()
538 .and_then(|child| child.stdin.as_mut())
539 .ok_or_else(|| {
540 Error::Io(std::io::Error::new(
541 std::io::ErrorKind::BrokenPipe,
542 "command stdin is not available; use StdinOption::Pipe",
543 ))
544 })?;
545 stdin.write_all(data.as_ref()).await?;
546 self.stdin_bytes.extend_from_slice(data.as_ref());
547 Ok(())
548 }
549
550 /// Close a running command's stdin pipe so it can observe end-of-input.
551 pub async fn close_stdin(&mut self) -> Result<()> {
552 self.start().await?;
553 if let Some(mut stdin) = self.child.as_mut().and_then(|child| child.stdin.take()) {
554 stdin.shutdown().await?;
555 }
556 Ok(())
557 }
558
559 /// Run the process to completion
560 pub async fn run(&mut self) -> Result<CommandResult> {
561 self.start().await?;
562
563 if let Some(result) = &self.result {
564 return Ok(result.clone());
565 }
566
567 let mut child = self
568 .child
569 .take()
570 .ok_or_else(|| Error::Io(std::io::Error::other("Process not started")))?;
571
572 // Handle stdin content if provided
573 if let StdinOption::Content(ref content) = self.options.stdin {
574 if let Some(mut stdin) = child.stdin.take() {
575 let content = content.clone();
576 tokio::spawn(async move {
577 let _ = stdin.write_all(content.as_bytes()).await;
578 let _ = stdin.shutdown().await;
579 });
580 }
581 }
582
583 // Drain both pipes concurrently and preserve their newline framing. The
584 // previous line reader appended `\n` to every final line, changing
585 // output from commands such as `printf` that omit a newline (issue #37).
586 let stdout_reader = child.stdout.take();
587 let stderr_reader = child.stderr.take();
588 let mut stdout = Vec::new();
589 let mut stderr = Vec::new();
590 utils::trace_lazy("ProcessRunner", || {
591 format!("Draining child output | pid={:?}", self.pid)
592 });
593 let mirror = self.options.mirror;
594 let drain = async {
595 tokio::try_join!(
596 collect_child_output(stdout_reader, mirror, ChildOutput::Stdout, &mut stdout),
597 collect_child_output(stderr_reader, mirror, ChildOutput::Stderr, &mut stderr),
598 )
599 .map(|_| ())
600 };
601 let (collected, status) = if self.cancelled {
602 // Keep draining while a graceful signal handler runs. After the
603 // child exits, a descendant can still retain either pipe. Like
604 // the streaming runner, bound that remaining drain; the outside
605 // buffers survive cancellation of the reader.
606 tokio::pin!(drain);
607 let mut status = None;
608 let collected = tokio::select! {
609 output = &mut drain => output,
610 child_status = child.wait() => {
611 status = Some(child_status?);
612 match tokio::time::timeout(
613 std::time::Duration::from_millis(stream::DEFAULT_EXIT_PUMP_GRACE_MS),
614 &mut drain,
615 ).await {
616 Ok(output) => output,
617 Err(_) => {
618 utils::trace_lazy("ProcessRunner", || {
619 "Cancelled output drain timed out; retaining buffered bytes".to_string()
620 });
621 Ok(())
622 }
623 }
624 }
625 };
626 (collected, status)
627 } else {
628 // Drain, then reap, unless the timeout elapses first (issue #227).
629 let plan = self.stop_plan();
630 let work = async {
631 let collected = drain.await;
632 let status = match collected {
633 Ok(()) => Some(child.wait().await),
634 Err(_) => None,
635 };
636 (collected, status)
637 };
638 let (outcome, timed_out) = process_stop::bounded(work, self.deadline, plan).await;
639 self.timed_out = timed_out;
640 match outcome {
641 Some((collected, status)) => (collected, status.transpose()?),
642 // Tokio reaps the abandoned child once it lets go of its pipes.
643 None => (Ok(()), None),
644 }
645 };
646 if let Err(error) = collected {
647 // `try_join!` drops the other pipe reader after an error. Stop
648 // and reap the child so it cannot remain blocked on that pipe.
649 let _ = child.start_kill();
650 let _ = child.wait().await;
651 return Err(error.into());
652 }
653
654 utils::trace_lazy("ProcessRunner", || {
655 format!("Child output drained | pid={:?}", self.pid)
656 });
657 let status = match status {
658 Some(status) => Some(status),
659 None if self.timed_out => None,
660 None => Some(child.wait().await?),
661 };
662 utils::trace_lazy("ProcessRunner", || {
663 format!("Child exited | pid={:?} status={status:?}", self.pid)
664 });
665 let code = match status {
666 Some(status) if !self.timed_out => status.code().unwrap_or(-1),
667 _ => signal::signal_exit_code(&self.options.kill_signal),
668 };
669
670 let mut result = CommandResult::new(
671 String::from_utf8_lossy(&stdout).into_owned(),
672 String::from_utf8_lossy(&stderr).into_owned(),
673 code,
674 );
675 if let StdinOption::Content(ref content) = self.options.stdin {
676 result.stdin = crate::result_streams::CapturedInput::new(content.as_bytes().to_vec());
677 } else if !self.stdin_bytes.is_empty() {
678 result.stdin = crate::result_streams::CapturedInput::new(self.stdin_bytes.clone());
679 }
680
681 self.result = Some(result.clone());
682 self.finished = true;
683
684 Ok(result)
685 }
686
687 /// Try to execute as a virtual command
688 async fn try_virtual_command(&self, cmd_name: &str) -> Option<CommandResult> {
689 if !commands::are_virtual_commands_enabled() {
690 return None;
691 }
692
693 // An empty command name means the caller already decided this command
694 // must go to a real shell (redirection, expansions, escapes). Bail out
695 // before tokenizing so we neither waste work nor parse shell syntax we
696 // deliberately delegate.
697 if cmd_name.is_empty() {
698 return None;
699 }
700
701 // Parse args from command string, respecting quotes and performing
702 // POSIX quote removal so `echo label:'help wanted'` reaches the built-in
703 // as the single argument `label:help wanted` (issue #48).
704 let words = shell_parser::split_command_words(&self.command);
705 let args: Vec<String> = words.into_iter().skip(1).collect();
706
707 let ctx = CommandContext {
708 args,
709 stdin: match &self.options.stdin {
710 StdinOption::Content(s) => Some(s.clone()),
711 _ => None,
712 },
713 cwd: self.options.cwd.clone(),
714 env: self.options.env.clone(),
715 output_tx: self.output_tx.clone(),
716 is_cancelled: None,
717 };
718
719 match cmd_name {
720 "echo" => Some(commands::echo(ctx).await),
721 "pwd" => Some(commands::pwd(ctx).await),
722 "cd" => Some(commands::cd::resolve_cd(ctx).await.0),
723 "true" => Some(commands::r#true(ctx).await),
724 "false" => Some(commands::r#false(ctx).await),
725 "sleep" => Some(commands::sleep(ctx).await),
726 "cat" => Some(commands::cat(ctx).await),
727 "ls" => Some(commands::ls(ctx).await),
728 "mkdir" => Some(commands::mkdir(ctx).await),
729 "rm" => Some(commands::rm(ctx).await),
730 "touch" => Some(commands::touch(ctx).await),
731 "cp" => Some(commands::cp(ctx).await),
732 "mv" => Some(commands::mv(ctx).await),
733 "basename" => Some(commands::basename(ctx).await),
734 "dirname" => Some(commands::dirname(ctx).await),
735 "env" => Some(commands::env(ctx).await),
736 "exit" => Some(commands::exit(ctx).await),
737 "which" => Some(commands::which(ctx).await),
738 "yes" => Some(commands::yes(ctx).await),
739 "seq" => Some(commands::seq(ctx).await),
740 "tee" => Some(commands::tee(ctx).await),
741 "test" => Some(commands::test(ctx).await),
742 "$fy" => Some(commands::fy(ctx).await),
743 "head" => Some(commands::head(ctx).await),
744 "tail" => Some(commands::tail(ctx).await),
745 "sort" => Some(commands::sort(ctx).await),
746 "uniq" => Some(commands::uniq(ctx).await),
747
748 _ => None,
749 }
750 }
751
752 /// Check if the process is finished
753 pub fn is_finished(&self) -> bool {
754 self.finished
755 }
756
757 /// Get the result if available
758 pub fn result(&self) -> Option<&CommandResult> {
759 self.result.as_ref()
760 }
761
762 /// Process id of the command, or `None` when there is no operating system
763 /// process to identify.
764 ///
765 /// It is `None` before the command starts, and stays `None` for built-in
766 /// (virtual) commands such as `echo` or `sleep`, which run inside this
767 /// process and never spawn a child. Once a real command has been spawned
768 /// the value is stable: it remains readable after the command finishes,
769 /// unlike the child handle, which [`run`](Self::run) consumes.
770 ///
771 /// Mirrors the JavaScript `runner.pid` property.
772 ///
773 /// ```no_run
774 /// use command_stream::{ProcessRunner, RunOptions};
775 ///
776 /// # #[tokio::main]
777 /// # async fn main() -> command_stream::Result<()> {
778 /// let mut runner = ProcessRunner::new("/bin/sleep 1", RunOptions::default());
779 /// runner.start().await?;
780 /// println!("running as pid {:?}", runner.pid());
781 /// runner.run().await?;
782 /// println!("still readable: {:?}", runner.pid());
783 /// # Ok(())
784 /// # }
785 /// ```
786 pub fn pid(&self) -> Option<u32> {
787 self.pid
788 }
789
790 /// Get the command string
791 pub fn command(&self) -> &str {
792 &self.command
793 }
794
795 /// Get the options
796 pub fn options(&self) -> &RunOptions {
797 &self.options
798 }
799}
800
801/// Execute a command and return the result
802///
803/// This is the main entry point for simple command execution.
804/// Named `run` instead of `$` since `$` is not a valid Rust identifier.
805pub async fn run(command: impl Into<String>) -> Result<CommandResult> {
806 let mut runner = ProcessRunner::new(command, RunOptions::default());
807 runner.run().await
808}
809
810/// Alias for `run` function - for JavaScript-like API feel
811/// Since `$` is not valid in Rust, this provides a similar short name
812pub use run as execute;
813
814/// Execute a command with custom options
815pub async fn exec(command: impl Into<String>, options: RunOptions) -> Result<CommandResult> {
816 let mut runner = ProcessRunner::new(command, options);
817 runner.run().await
818}
819
820/// Create a new process runner without starting it
821pub fn create(command: impl Into<String>, options: RunOptions) -> ProcessRunner {
822 ProcessRunner::new(command, options)
823}
824
825/// Execute a command synchronously (blocking)
826pub fn run_sync(command: impl Into<String>) -> Result<CommandResult> {
827 let rt = tokio::runtime::Runtime::new()?;
828 rt.block_on(run(command))
829}
830
831// Tests are located in tests/ directory for better organization