Something went wrong. Try again.
bayes for days
Something went wrong. Try again.
Rust
12345678910111213141516171819202122232425262728293031323334353637383940414243444546474849505152535455565758596061626364656667686970717273747576//! A node that is a child process, spoken to over its stdio.
use anyhow::{anyhow, Context, Result};use async_trait::async_trait;use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader};use tokio::process::{Child, ChildStdin, ChildStdout, Command};use tokio::sync::Mutex;
use crate::{Node, Request, Response};
pub struct ProcessNode { address: String, // One at a time. The line protocol has no request ids on this hop, so two // concurrent asks would race to read each other's replies. The lock is what // makes a process node safe to share; if a node needs real concurrency, run // several of them. io: Mutex<ProcessIo>,}
struct ProcessIo { _child: Child, stdin: ChildStdin, stdout: BufReader<ChildStdout>,}
impl ProcessNode { /// Spawn `command`, split on whitespace. pub fn spawn(command: &str) -> Result<Self> { let mut parts = command.split_whitespace(); let program = parts.next().context("proc: address has no command")?; let mut child = Command::new(program) .args(parts) .stdin(std::process::Stdio::piped()) .stdout(std::process::Stdio::piped()) // Left attached: a node that dies should say why in the match log // rather than into a pipe nobody reads. .stderr(std::process::Stdio::inherit()) .kill_on_drop(true) .spawn() .with_context(|| format!("could not start node process '{command}'"))?; let stdin = child.stdin.take().context("child has no stdin")?; let stdout = child.stdout.take().context("child has no stdout")?; Ok(Self { address: format!("proc:{command}"), io: Mutex::new(ProcessIo { _child: child, stdin, stdout: BufReader::new(stdout), }), }) }}
#[async_trait]impl Node for ProcessNode { fn address(&self) -> &str { &self.address }
async fn think(&self, request: Request) -> Result<Response> { let mut io = self.io.lock().await; let line = serde_json::to_string(&request)?; io.stdin.write_all(line.as_bytes()).await?; io.stdin.write_all(b"\n").await?; io.stdin.flush().await?;
let mut reply = String::new(); let read = io.stdout.read_line(&mut reply).await?; if read == 0 { return Err(anyhow!("node {} closed its output", self.address)); } serde_json::from_str(reply.trim()) .with_context(|| format!("node {} sent unreadable JSON: {reply}", self.address)) }}