muse_codes/
client_async.rs

1//! Async client for streaming a headless `muse exec --json` run.
2//!
3//! One [`ExecRun`] wraps one child process; [`next_record`](ExecRun::next_record)
4//! yields typed journal records as they arrive, and
5//! [`wait_terminal`](ExecRun::wait_terminal) drives the run to its
6//! `run.terminal.*` record.
7
8use crate::cli::MuseExecBuilder;
9use crate::error::{Error, Result};
10use crate::io::{MusePayload, MuseRecord};
11use tokio::io::{AsyncBufReadExt, BufReader, Lines};
12use tokio::process::{Child, ChildStderr, ChildStdout};
13
14/// A running `muse exec --json` invocation.
15pub struct ExecRun {
16    child: Child,
17    lines: Lines<BufReader<ChildStdout>>,
18    /// stderr is collected in the background for error context.
19    stderr_task: tokio::task::JoinHandle<String>,
20}
21
22impl ExecRun {
23    /// Spawn a run from a builder.
24    pub async fn spawn(builder: &MuseExecBuilder) -> Result<Self> {
25        let mut child = builder.spawn().await?;
26        let stdout = child
27            .stdout
28            .take()
29            .ok_or_else(|| Error::Protocol("failed to get stdout".to_string()))?;
30        let stderr = child
31            .stderr
32            .take()
33            .ok_or_else(|| Error::Protocol("failed to get stderr".to_string()))?;
34        Ok(Self {
35            child,
36            lines: BufReader::new(stdout).lines(),
37            stderr_task: spawn_stderr_collector(stderr),
38        })
39    }
40
41    /// OS process id of the running child, when the platform exposes one.
42    ///
43    /// Useful for supervisors that signal the process group directly rather
44    /// than relying on `kill_on_drop`.
45    pub fn pid(&self) -> Option<u32> {
46        self.child.id()
47    }
48
49    /// Next journal record, or `None` at end of stream.
50    pub async fn next_record(&mut self) -> Result<Option<MuseRecord>> {
51        loop {
52            match self.lines.next_line().await? {
53                None => return Ok(None),
54                Some(line) if line.trim().is_empty() => continue,
55                Some(line) => return Ok(Some(serde_json::from_str(&line)?)),
56            }
57        }
58    }
59
60    /// Consume records until the run reaches a terminal state, invoking
61    /// `on_record` for each record seen (including the terminal one), and
62    /// return the terminal payload.
63    ///
64    /// If the stream ends without a `run.terminal.*` record, the child's
65    /// exit code and collected stderr are folded into the error.
66    pub async fn wait_terminal<F>(mut self, mut on_record: F) -> Result<crate::io::RunTerminal>
67    where
68        F: FnMut(&MuseRecord),
69    {
70        while let Some(record) = self.next_record().await? {
71            on_record(&record);
72            if let Ok(MusePayload::RunTerminal(t)) = record.typed_payload() {
73                return Ok(t);
74            }
75        }
76        let status = self.child.wait().await?;
77        let stderr = self.stderr_task.await.unwrap_or_default();
78        Err(Error::Protocol(format!(
79            "stream ended without run.terminal.* (exit {:?}); stderr:\n{}",
80            status.code(),
81            stderr.trim()
82        )))
83    }
84
85    /// Kill the child process.
86    pub async fn kill(&mut self) -> Result<()> {
87        self.child.kill().await?;
88        Ok(())
89    }
90}
91
92fn spawn_stderr_collector(stderr: ChildStderr) -> tokio::task::JoinHandle<String> {
93    tokio::spawn(async move {
94        let mut out = String::new();
95        let mut lines = BufReader::new(stderr).lines();
96        while let Ok(Some(line)) = lines.next_line().await {
97            #[cfg(feature = "async-client")]
98            log::debug!(target: "muse_codes::stderr", "{line}");
99            out.push_str(&line);
100            out.push('\n');
101        }
102        out
103    })
104}
muse_codes/client_async.rs

muse_codes/
client_async.rs
