specfold_core/plugins/
process.rs

1//! Runs a plugin process for one request, with a deadline.
2
3use super::protocol::{PROTOCOL, Request};
4use crate::config::{Config, PluginDef};
5use anyhow::{Context, Result, anyhow, bail};
6use serde::de::DeserializeOwned;
7use serde_json::Value;
8use std::io::{Read, Write};
9use std::path::PathBuf;
10use std::process::{Command, Stdio};
11use std::time::{Duration, Instant};
12
13fn program(cfg: &Config, p: &PluginDef) -> Result<(PathBuf, Vec<String>)> {
14    let (prog, args) = p.cmd.split_first().ok_or_else(|| anyhow!("plugin {} has an empty cmd", p.name))?;
15    let local = cfg.root.join(prog);
16    let resolved = if prog.contains('/') && local.exists() { local } else { PathBuf::from(prog) };
17    Ok((resolved, args.to_vec()))
18}
19
20pub fn call<T: DeserializeOwned>(cfg: &Config, p: &PluginDef, method: &str, params: Value) -> Result<T> {
21    let (prog, args) = program(cfg, p)?;
22    let mut child = Command::new(&prog)
23        .args(&args)
24        .current_dir(&cfg.root)
25        .env("SPECFOLD_PROTOCOL", PROTOCOL)
26        .stdin(Stdio::piped())
27        .stdout(Stdio::piped())
28        .stderr(Stdio::piped())
29        .spawn()
30        .with_context(|| format!("plugin {}: cannot start {}", p.name, prog.display()))?;
31    let body = serde_json::to_vec(&Request { protocol: PROTOCOL, method, params })?;
32    let mut stdin = child.stdin.take().context("no stdin")?;
33    let writer = std::thread::spawn(move || {
34        let _ = stdin.write_all(&body);
35    });
36    let mut stdout = child.stdout.take().context("no stdout")?;
37    let reader = std::thread::spawn(move || {
38        let mut buf = vec![];
39        let _ = stdout.read_to_end(&mut buf);
40        buf
41    });
42    let deadline = Instant::now() + Duration::from_millis(p.timeout_ms);
43    let status = loop {
44        if let Some(s) = child.try_wait()? {
45            break s;
46        }
47        if Instant::now() > deadline {
48            let _ = child.kill();
49            let _ = child.wait();
50            bail!("plugin {} did not answer {method} within {} ms", p.name, p.timeout_ms);
51        }
52        std::thread::sleep(Duration::from_millis(10));
53    };
54    let _ = writer.join();
55    let out = reader.join().map_err(|_| anyhow!("plugin {}: reader failed", p.name))?;
56    if !status.success() {
57        let mut err = String::new();
58        if let Some(mut e) = child.stderr.take() {
59            let _ = e.read_to_string(&mut err);
60        }
61        bail!("plugin {} failed on {method}: {}", p.name, err.lines().last().unwrap_or("exit status non-zero"));
62    }
63    serde_json::from_slice(&out).with_context(|| format!("plugin {} answered {method} with invalid JSON", p.name))
64}
specfold_core/plugins/process.rs

specfold_core/plugins/
process.rs
