trunk_recorder_plugin/

sdk.rs

//! Writing a plugin: implement [`Plugin`] and call [`run`] from `main`.

use std::io::{self, BufRead, IsTerminal, Write};
use std::path::PathBuf;
use std::sync::{Arc, Mutex};

use schemars::JsonSchema;
use serde::de::DeserializeOwned;
use serde_json::Value;

use crate::protocol::*;
use crate::schema;

/// A plugin. Every method runs on the one thread that reads the recorder's
/// messages, one message at a time: return quickly, and hand slow work
/// (network, disk, other processes) to a thread of your own.
pub trait Plugin: Sized {
    /// The plugin's settings. `#[serde(default)]` on the struct lets a user
    /// leave any of them empty. The settings form is drawn from its schema
    /// (see [`crate::schema`]). [`NoConfig`] for none. A system can override
    /// any of them (another server for it, say): read them for a system from
    /// [`System::settings`], and from [`Setup::config`] only for what isn't
    /// any one system's.
    type Config: DeserializeOwned + JsonSchema;
    /// Settings for each system, when the plugin needs some (an API key per
    /// system, say). [`NoConfig`] for none.
    type SystemConfig: DeserializeOwned + JsonSchema;

    /// Who the plugin is and what it subscribes to — start from [`crate::manifest!`].
    /// `config` and `system_config` are filled in from the types above.
    fn manifest() -> Manifest;

    /// Start with the user's settings. An `Err` is shown to the user, and the
    /// plugin isn't started again until recording next starts.
    fn start(host: Host, setup: Setup<Self::Config, Self::SystemConfig>) -> Result<Self, String>;

    fn call_start(&mut self, _call: CallInfo) {}
    fn call_end(&mut self, _call: CallInfo) {}
    fn call_concluded(&mut self, _call: ConcludedCall) {}
    fn unit(&mut self, _event: UnitEvent) {}
    fn audio(&mut self, _chunk: AudioChunk) {}
    fn status(&mut self, _status: Status) {}
    /// A call's tones made a pattern: every one heard (subscribe to `tone_pattern`).
    fn tone_pattern(&mut self, _t: TonePatternEvent) {}
    /// One of the plugin's buttons was pressed (`x-actions` in its settings
    /// schema). Answer through `reply`, now or from another thread once the
    /// work is done; a reply dropped unsent answers that it failed.
    fn action(&mut self, _a: Action, reply: Reply) {
        reply.fail("This plugin has no such action");
    }
    /// Its settings changed while it runs (set [`Manifest::live_settings`]
    /// for this to be called rather than the plugin restarted). An `Err` is
    /// shown as its status, and the settings it had stay.
    fn settings(&mut self, _setup: Setup<Self::Config, Self::SystemConfig>) -> Result<(), String> {
        Err("it can't take new settings while it runs".into())
    }

    /// The recorder is stopping (or the plugin's stdin closed): finish or save
    /// what's in flight within `grace` — the process exits when this returns.
    fn shutdown(&mut self, _grace: std::time::Duration) {}
}

pub use crate::schema::NoConfig;

/// What [`Plugin::start`] gets.
#[derive(Clone, Debug)]
pub struct Setup<C, S> {
    /// The plugin's settings for the whole recorder (each system's, with
    /// its overrides: [`System::settings`]).
    pub config: C,
    /// Every system, with the plugin's settings for it.
    pub systems: Vec<System<C, S>>,
    pub host: HostInfo,
    pub capture_dir: PathBuf,
    /// The plugin's own folder, kept across restarts and upgrades.
    pub data_dir: PathBuf,
    /// Audio formats `call.concluded` will carry (always "wav").
    pub audio_formats: Vec<String>,
}

impl<C, S> Setup<C, S> {
    /// The system an event's `system` names. (Its number is for this run
    /// only; keep anything you store by [`System::short_name`].)
    pub fn system(&self, index: u16) -> Option<&System<C, S>> {
        self.systems.iter().find(|s| s.index == index)
    }
    /// The system with this short name — its identity: every event carries
    /// it (`short_name`, or a concluded call's `call.short_name`).
    pub fn system_named(&self, short_name: &str) -> Option<&System<C, S>> {
        self.systems.iter().find(|s| s.short_name == short_name)
    }
    /// Whether `call.concluded` will carry `format` ([`format`](mod@format)).
    pub fn has_format(&self, format: &str) -> bool {
        self.audio_formats.iter().any(|f| f == format)
    }
}

#[derive(Clone, Debug)]
pub struct System<C, S> {
    /// The number events carry for it, for this run only.
    pub index: u16,
    /// Its identity: unique among the recorder's systems, and what users know it by.
    pub short_name: String,
    /// "p25" | "smartnet" | "dmr" | "nxdn" | "edacs" | "conventional"
    pub kind: String,
    /// What people call it; empty when the user gave no name.
    pub name: String,
    /// Its FCC licenses by frequency (Hz, as a string), and the call sign
    /// (and licensee) that names it: see [`SystemInfo::fcc`].
    pub fcc: std::collections::BTreeMap<String, FccLicense>,
    pub fcc_call_sign: String,
    pub fcc_name: String,
    /// Its sites.
    pub sites: Vec<SiteInfo>,
    /// The plugin's settings for this system (`None`: left empty).
    pub config: Option<S>,
    /// The plugin's own settings ([`Setup::config`]) as they are for this
    /// system, with any it overrides.
    pub settings: C,
}

/// The way back to the recorder: logs, health, results. Cheap to clone and
/// usable from any thread. (A plugin's stdout belongs to the protocol — never
/// `println!` in a plugin; use this, or `eprintln!` for the raw log.)
#[derive(Clone)]
pub struct Host {
    out: Arc<Mutex<Box<dyn Write + Send>>>,
}

impl Host {
    /// A host writing to `w` instead of stdout (see [`crate::testing`]).
    pub fn to_writer(w: impl Write + Send + 'static) -> Host {
        Host { out: Arc::new(Mutex::new(Box::new(w))) }
    }

    pub fn send(&self, m: &PluginMessage) {
        let mut line = serde_json::to_string(m).unwrap_or_default();
        line.push('\n');
        let mut o = self.out.lock().unwrap_or_else(|e| e.into_inner());
        // The recorder has gone when this fails; stdin's end will stop us.
        let _ = o.write_all(line.as_bytes()).and_then(|_| o.flush());
    }

    pub fn log(&self, level: Level, message: impl Into<String>) {
        self.send(&PluginMessage::Log { level, message: message.into() });
    }
    pub fn error(&self, message: impl Into<String>) {
        self.log(Level::Error, message)
    }
    pub fn warn(&self, message: impl Into<String>) {
        self.log(Level::Warn, message)
    }
    pub fn info(&self, message: impl Into<String>) {
        self.log(Level::Info, message)
    }
    pub fn debug(&self, message: impl Into<String>) {
        self.log(Level::Debug, message)
    }

    /// The plugin's health, shown in the interface until the next.
    pub fn status(&self, state: State, message: impl Into<String>) {
        self.send(&PluginMessage::Status { state, message: message.into() });
    }

    /// How the plugin's work is going (see [`Metrics`]): queue, timing,
    /// the services it talks to. At most every few seconds.
    pub fn metrics(&self, m: &Metrics) {
        self.send(&PluginMessage::Metrics(m.clone()));
    }

    /// What became of a concluded call.
    pub fn call_result(&self, path: &str, outcome: Outcome, message: impl Into<String>, url: impl Into<String>) {
        self.send(&PluginMessage::CallResult { path: path.to_string(), outcome, message: message.into(), url: url.into(), took_ms: None });
    }

    /// The same, and how long the work on it took (an upload's last try):
    /// the recorder tells waiting from uploading by it. [`crate::CallQueue`]
    /// says it for you.
    pub fn call_result_took(&self, path: &str, outcome: Outcome, message: impl Into<String>, url: impl Into<String>, took: std::time::Duration) {
        let took_ms = Some(took.as_millis() as u64);
        self.send(&PluginMessage::CallResult { path: path.to_string(), outcome, message: message.into(), url: url.into(), took_ms });
    }
}

/// The answer to an [`Action`]: send it once, with [`Reply::ok`] or
/// [`Reply::fail`]. It can be moved to another thread.
pub struct Reply {
    host: Host,
    request: u64,
    sent: bool,
}

impl Reply {
    pub fn ok(mut self, message: impl Into<String>) {
        self.send(true, message.into());
    }
    pub fn fail(mut self, message: impl Into<String>) {
        self.send(false, message.into());
    }
    fn send(&mut self, ok: bool, message: String) {
        self.sent = true;
        self.host.send(&PluginMessage::ActionResult { request: self.request, ok, message });
    }
}

impl Drop for Reply {
    fn drop(&mut self) {
        if !self.sent {
            self.send(false, "The plugin didn't answer".into());
        }
    }
}

/// The manifest with the settings schemas filled in.
pub fn describe<P: Plugin>() -> Manifest {
    let mut m = P::manifest();
    m.api = API_VERSION;
    let c = schema::schema_for::<P::Config>();
    m.config = (!schema::is_empty(&c)).then_some(c);
    let s = schema::schema_for::<P::SystemConfig>();
    m.system_config = (!schema::is_empty(&s)).then_some(s);
    m
}

/// Run the plugin: `--describe` prints its manifest; otherwise it talks to the
/// recorder on stdin/stdout until told to stop. Call from `main`.
pub fn run<P: Plugin>() {
    let arg = std::env::args().nth(1);
    match arg.as_deref() {
        Some("--describe") => {
            println!("{}", serde_json::to_string_pretty(&describe::<P>()).unwrap_or_default());
            return;
        }
        Some("--version" | "-V") => {
            println!("{}", P::manifest().version);
            return;
        }
        Some(a) => {
            eprintln!("unknown argument {a} (a plugin takes --describe or --version; the recorder runs it with none)");
            std::process::exit(2);
        }
        None => {}
    }
    let stdin = io::stdin();
    if stdin.is_terminal() {
        let m = P::manifest();
        eprintln!(
            "{} {} is a Trunk Recorder Pro plugin: the recorder runs it and talks to it over stdin/stdout.\n\
             `--describe` prints its manifest. Try it with `trunk-pro plugin run <this binary>`,\n\
             or paste protocol lines here (a hello first).",
            m.name, m.version
        );
    }
    let code = serve::<P>(stdin.lock(), Host { out: Arc::new(Mutex::new(Box::new(io::stdout()))) });
    std::process::exit(code);
}

/// The message loop over `input`; the process's exit status.
pub fn serve<P: Plugin>(input: impl BufRead, host: Host) -> i32 {
    let mut plugin: Option<P> = None;
    let mut grace = std::time::Duration::from_secs(10);
    for line in input.lines() {
        let Ok(line) = line else { break };
        if line.trim().is_empty() {
            continue;
        }
        let msg: HostMessage = match serde_json::from_str(&line) {
            Ok(m) => m,
            Err(e) => {
                host.warn(format!("unreadable message from the recorder ({e}): {line}"));
                continue;
            }
        };
        let Some(p) = plugin.as_mut() else {
            let HostMessage::Hello(hello) = msg else {
                host.warn("a message before hello, ignored");
                continue;
            };
            match start::<P>(host.clone(), hello) {
                Ok(p) => {
                    plugin = Some(p);
                    host.send(&PluginMessage::Ready);
                }
                Err(e) => {
                    host.status(State::Error, e.clone());
                    host.error(e);
                    return EXIT_CONFIG;
                }
            }
            continue;
        };
        match msg {
            HostMessage::Hello(_) => host.warn("a second hello, ignored"),
            HostMessage::CallStart(c) => p.call_start(c),
            HostMessage::CallEnd(c) => p.call_end(c),
            HostMessage::CallConcluded(c) => p.call_concluded(c),
            HostMessage::Unit(u) => p.unit(u),
            HostMessage::Audio(a) => p.audio(a),
            HostMessage::Status(s) => p.status(s),
            HostMessage::TonePattern(t) => p.tone_pattern(t),
            HostMessage::Action(a) => {
                let reply = Reply { host: host.clone(), request: a.request, sent: false };
                p.action(a, reply)
            }
            HostMessage::Settings(h) => match setup::<P>(h).and_then(|s| p.settings(s)) {
                Ok(()) => host.status(State::Ok, ""),
                Err(e) => host.status(State::Error, format!("new settings not taken: {e}")),
            },
            HostMessage::Shutdown(s) => {
                grace = std::time::Duration::from_secs_f64(s.grace_s.max(0.0));
                break;
            }
            HostMessage::Unknown => {}
        }
    }
    if let Some(p) = plugin.as_mut() {
        p.shutdown(grace);
    }
    0
}

fn start<P: Plugin>(host: Host, hello: Hello) -> Result<P, String> {
    P::start(host, setup::<P>(hello)?)
}

/// What `hello` sets the plugin up with.
fn setup<P: Plugin>(hello: Hello) -> Result<Setup<P::Config, P::SystemConfig>, String> {
    if hello.api != 0 && hello.api < API_VERSION {
        return Err(format!("this plugin needs a newer recorder (plugin API {API_VERSION}, the recorder has {})", hello.api));
    }
    let config: P::Config = parse(hello.config.clone(), "settings")?;
    let mut systems = Vec::new();
    for s in hello.systems {
        let settings = match s.settings {
            Value::Null => parse::<P::Config>(hello.config.clone(), "settings")?,
            v => parse::<P::Config>(v, &format!("settings for {}", s.short_name))?,
        };
        let config = match s.config {
            Value::Null => None,
            v => Some(parse::<P::SystemConfig>(v, &format!("settings for {}", s.short_name))?),
        };
        systems.push(System { index: s.index, short_name: s.short_name, kind: s.kind, name: s.name, fcc: s.fcc, fcc_call_sign: s.fcc_call_sign, fcc_name: s.fcc_name, sites: s.sites, config, settings });
    }
    Ok(Setup { config, systems, host: hello.host, capture_dir: hello.capture_dir, data_dir: hello.data_dir, audio_formats: hello.audio_formats })
}

/// Settings left out entirely count as `{}`, so `#[serde(default)]` applies.
fn parse<T: DeserializeOwned>(v: Value, what: &str) -> Result<T, String> {
    let v = if v.is_null() { Value::Object(Default::default()) } else { v };
    serde_json::from_value(v).map_err(|e| format!("bad {what}: {e}"))
}

/// A manifest with the id, version, description, homepage, repository,
/// authors and license from Cargo.toml; the name is the id until you set one:
///
/// ```ignore
/// fn manifest() -> Manifest {
///     Manifest { name: "OpenMHz".into(), subscribe: vec![topic::CALL_CONCLUDED.into()], ..trunk_recorder_plugin::manifest!() }
/// }
/// ```
#[macro_export]
macro_rules! manifest {
    () => {{
        let authors: Vec<String> = env!("CARGO_PKG_AUTHORS").split(':').filter(|a| !a.is_empty()).map(String::from).collect();
        $crate::Manifest {
            id: env!("CARGO_PKG_NAME").to_string(),
            name: env!("CARGO_PKG_NAME").to_string(),
            version: env!("CARGO_PKG_VERSION").to_string(),
            description: env!("CARGO_PKG_DESCRIPTION").to_string(),
            api: $crate::API_VERSION,
            homepage: env!("CARGO_PKG_HOMEPAGE").to_string(),
            repository: env!("CARGO_PKG_REPOSITORY").to_string(),
            authors,
            license: env!("CARGO_PKG_LICENSE").to_string(),
            ..Default::default()
        }
    }};
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::testing;
    use serde_json::json;

    #[derive(serde::Deserialize, schemars::JsonSchema, Default)]
    #[serde(default)]
    struct Config {
        server: String,
        retries: u32,
    }

    /// Says each system's server.
    struct Servers;

    impl Plugin for Servers {
        type Config = Config;
        type SystemConfig = NoConfig;
        fn manifest() -> Manifest {
            Manifest { id: "servers".into(), ..Default::default() }
        }
        fn start(host: Host, setup: Setup<Config, NoConfig>) -> Result<Self, String> {
            for s in &setup.systems {
                host.info(format!("{} {} {}", s.short_name, s.settings.server, s.settings.retries));
            }
            Ok(Servers)
        }
    }

    /// A system's settings are the plugin's with what it overrides.
    #[test]
    fn a_system_can_override_the_plugins_settings() {
        let dir = testing::temp_dir("sdk");
        let mut h = testing::hello(&dir, json!({ "server": "https://a", "retries": 3 }));
        h.systems.push(SystemInfo { index: 1, short_name: "sys2".into(), kind: "p25".into(), settings: json!({ "server": "https://b", "retries": 3 }), ..Default::default() });
        let out = testing::run::<Servers>([HostMessage::Hello(h)]);
        assert!(out.ready());
        let said: Vec<String> = out.logs().into_iter().map(|(_, t)| t).collect();
        assert_eq!(said, ["sys1 https://a 3", "sys2 https://b 3"]);
    }
}