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