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    /// Next journal record, or `None` at end of stream.
42    pub async fn next_record(&mut self) -> Result<Option<MuseRecord>> {
43        loop {
44            match self.lines.next_line().await? {
45                None => return Ok(None),
46                Some(line) if line.trim().is_empty() => continue,
47                Some(line) => return Ok(Some(serde_json::from_str(&line)?)),
48            }
49        }
50    }
51
52    /// Consume records until the run reaches a terminal state, invoking
53    /// `on_record` for each record seen (including the terminal one), and
54    /// return the terminal payload.
55    ///
56    /// If the stream ends without a `run.terminal.*` record, the child's
57    /// exit code and collected stderr are folded into the error.
58    pub async fn wait_terminal<F>(mut self, mut on_record: F) -> Result<crate::io::RunTerminal>
59    where
60        F: FnMut(&MuseRecord),
61    {
62        while let Some(record) = self.next_record().await? {
63            on_record(&record);
64            if let Ok(MusePayload::RunTerminal(t)) = record.typed_payload() {
65                return Ok(t);
66            }
67        }
68        let status = self.child.wait().await?;
69        let stderr = self.stderr_task.await.unwrap_or_default();
70        Err(Error::Protocol(format!(
71            "stream ended without run.terminal.* (exit {:?}); stderr:\n{}",
72            status.code(),
73            stderr.trim()
74        )))
75    }
76
77    /// Kill the child process.
78    pub async fn kill(&mut self) -> Result<()> {
79        self.child.kill().await?;
80        Ok(())
81    }
82}
83
84fn spawn_stderr_collector(stderr: ChildStderr) -> tokio::task::JoinHandle<String> {
85    tokio::spawn(async move {
86        let mut out = String::new();
87        let mut lines = BufReader::new(stderr).lines();
88        while let Ok(Some(line)) = lines.next_line().await {
89            #[cfg(feature = "async-client")]
90            log::debug!(target: "muse_codes::stderr", "{line}");
91            out.push_str(&line);
92            out.push('\n');
93        }
94        out
95    })
96}
muse_codes/client_async.rs

muse_codes/
client_async.rs
