trunk_recorder_plugin/
sdk.rs

1//! Writing a plugin: implement [`Plugin`] and call [`run`] from `main`.
2
3use std::io::{self, BufRead, IsTerminal, Write};
4use std::path::PathBuf;
5use std::sync::{Arc, Mutex};
6
7use schemars::JsonSchema;
8use serde::de::DeserializeOwned;
9use serde_json::Value;
10
11use crate::protocol::*;
12use crate::schema;
13
14/// A plugin. Every method runs on the one thread that reads the recorder's
15/// messages, one message at a time: return quickly, and hand slow work
16/// (network, disk, other processes) to a thread of your own.
17pub trait Plugin: Sized {
18    /// The plugin's settings. `#[serde(default)]` on the struct lets a user
19    /// leave any of them empty. The settings form is drawn from its schema
20    /// (see [`crate::schema`]). [`NoConfig`] for none. A system can override
21    /// any of them (another server for it, say): read them for a system from
22    /// [`System::settings`], and from [`Setup::config`] only for what isn't
23    /// any one system's.
24    type Config: DeserializeOwned + JsonSchema;
25    /// Settings for each system, when the plugin needs some (an API key per
26    /// system, say). [`NoConfig`] for none.
27    type SystemConfig: DeserializeOwned + JsonSchema;
28
29    /// Who the plugin is and what it subscribes to — start from [`crate::manifest!`].
30    /// `config` and `system_config` are filled in from the types above.
31    fn manifest() -> Manifest;
32
33    /// Start with the user's settings. An `Err` is shown to the user, and the
34    /// plugin isn't started again until recording next starts.
35    fn start(host: Host, setup: Setup<Self::Config, Self::SystemConfig>) -> Result<Self, String>;
36
37    fn call_start(&mut self, _call: CallInfo) {}
38    fn call_end(&mut self, _call: CallInfo) {}
39    fn call_concluded(&mut self, _call: ConcludedCall) {}
40    fn unit(&mut self, _event: UnitEvent) {}
41    fn audio(&mut self, _chunk: AudioChunk) {}
42    fn status(&mut self, _status: Status) {}
43
44    /// The recorder is stopping (or the plugin's stdin closed): finish or save
45    /// what's in flight within `grace` — the process exits when this returns.
46    fn shutdown(&mut self, _grace: std::time::Duration) {}
47}
48
49pub use crate::schema::NoConfig;
50
51/// What [`Plugin::start`] gets.
52#[derive(Clone, Debug)]
53pub struct Setup<C, S> {
54    /// The plugin's settings for the whole recorder (each system's, with
55    /// its overrides: [`System::settings`]).
56    pub config: C,
57    /// Every system, with the plugin's settings for it.
58    pub systems: Vec<System<C, S>>,
59    pub host: HostInfo,
60    pub capture_dir: PathBuf,
61    /// The plugin's own folder, kept across restarts and upgrades.
62    pub data_dir: PathBuf,
63    /// Audio formats `call.concluded` will carry (always "wav").
64    pub audio_formats: Vec<String>,
65}
66
67impl<C, S> Setup<C, S> {
68    /// The system an event's `system` names. (Its number is for this run
69    /// only; keep anything you store by [`System::short_name`].)
70    pub fn system(&self, index: u16) -> Option<&System<C, S>> {
71        self.systems.iter().find(|s| s.index == index)
72    }
73    /// The system with this short name — its identity: every event carries
74    /// it (`short_name`, or a concluded call's `call.short_name`).
75    pub fn system_named(&self, short_name: &str) -> Option<&System<C, S>> {
76        self.systems.iter().find(|s| s.short_name == short_name)
77    }
78    /// Whether `call.concluded` will carry `format` ([`format`](mod@format)).
79    pub fn has_format(&self, format: &str) -> bool {
80        self.audio_formats.iter().any(|f| f == format)
81    }
82}
83
84#[derive(Clone, Debug)]
85pub struct System<C, S> {
86    /// The number events carry for it, for this run only.
87    pub index: u16,
88    /// Its identity: unique among the recorder's systems, and what users know it by.
89    pub short_name: String,
90    /// "p25" | "smartnet" | "dmr" | "nxdn" | "conventional"
91    pub kind: String,
92    /// The plugin's settings for this system (`None`: left empty).
93    pub config: Option<S>,
94    /// The plugin's own settings ([`Setup::config`]) as they are for this
95    /// system, with any it overrides.
96    pub settings: C,
97}
98
99/// The way back to the recorder: logs, health, results. Cheap to clone and
100/// usable from any thread. (A plugin's stdout belongs to the protocol — never
101/// `println!` in a plugin; use this, or `eprintln!` for the raw log.)
102#[derive(Clone)]
103pub struct Host {
104    out: Arc<Mutex<Box<dyn Write + Send>>>,
105}
106
107impl Host {
108    /// A host writing to `w` instead of stdout (see [`crate::testing`]).
109    pub fn to_writer(w: impl Write + Send + 'static) -> Host {
110        Host { out: Arc::new(Mutex::new(Box::new(w))) }
111    }
112
113    pub fn send(&self, m: &PluginMessage) {
114        let mut line = serde_json::to_string(m).unwrap_or_default();
115        line.push('\n');
116        let mut o = self.out.lock().unwrap_or_else(|e| e.into_inner());
117        // The recorder has gone when this fails; stdin's end will stop us.
118        let _ = o.write_all(line.as_bytes()).and_then(|_| o.flush());
119    }
120
121    pub fn log(&self, level: Level, message: impl Into<String>) {
122        self.send(&PluginMessage::Log { level, message: message.into() });
123    }
124    pub fn error(&self, message: impl Into<String>) {
125        self.log(Level::Error, message)
126    }
127    pub fn warn(&self, message: impl Into<String>) {
128        self.log(Level::Warn, message)
129    }
130    pub fn info(&self, message: impl Into<String>) {
131        self.log(Level::Info, message)
132    }
133    pub fn debug(&self, message: impl Into<String>) {
134        self.log(Level::Debug, message)
135    }
136
137    /// The plugin's health, shown in the interface until the next.
138    pub fn status(&self, state: State, message: impl Into<String>) {
139        self.send(&PluginMessage::Status { state, message: message.into() });
140    }
141
142    /// How the plugin's work is going (see [`Metrics`]): queue, timing,
143    /// the services it talks to. At most every few seconds.
144    pub fn metrics(&self, m: &Metrics) {
145        self.send(&PluginMessage::Metrics(m.clone()));
146    }
147
148    /// What became of a concluded call.
149    pub fn call_result(&self, path: &str, outcome: Outcome, message: impl Into<String>, url: impl Into<String>) {
150        self.send(&PluginMessage::CallResult { path: path.to_string(), outcome, message: message.into(), url: url.into() });
151    }
152}
153
154/// The manifest with the settings schemas filled in.
155pub fn describe<P: Plugin>() -> Manifest {
156    let mut m = P::manifest();
157    m.api = API_VERSION;
158    let c = schema::schema_for::<P::Config>();
159    m.config = (!schema::is_empty(&c)).then_some(c);
160    let s = schema::schema_for::<P::SystemConfig>();
161    m.system_config = (!schema::is_empty(&s)).then_some(s);
162    m
163}
164
165/// Run the plugin: `--describe` prints its manifest; otherwise it talks to the
166/// recorder on stdin/stdout until told to stop. Call from `main`.
167pub fn run<P: Plugin>() {
168    let arg = std::env::args().nth(1);
169    match arg.as_deref() {
170        Some("--describe") => {
171            println!("{}", serde_json::to_string_pretty(&describe::<P>()).unwrap_or_default());
172            return;
173        }
174        Some("--version" | "-V") => {
175            println!("{}", P::manifest().version);
176            return;
177        }
178        Some(a) => {
179            eprintln!("unknown argument {a} (a plugin takes --describe or --version; the recorder runs it with none)");
180            std::process::exit(2);
181        }
182        None => {}
183    }
184    let stdin = io::stdin();
185    if stdin.is_terminal() {
186        let m = P::manifest();
187        eprintln!(
188            "{} {} is a Trunk Recorder Pro plugin: the recorder runs it and talks to it over stdin/stdout.\n\
189             `--describe` prints its manifest. Try it with `trunk-pro plugin run <this binary>`,\n\
190             or paste protocol lines here (a hello first).",
191            m.name, m.version
192        );
193    }
194    let code = serve::<P>(stdin.lock(), Host { out: Arc::new(Mutex::new(Box::new(io::stdout()))) });
195    std::process::exit(code);
196}
197
198/// The message loop over `input`; the process's exit status.
199pub fn serve<P: Plugin>(input: impl BufRead, host: Host) -> i32 {
200    let mut plugin: Option<P> = None;
201    let mut grace = std::time::Duration::from_secs(10);
202    for line in input.lines() {
203        let Ok(line) = line else { break };
204        if line.trim().is_empty() {
205            continue;
206        }
207        let msg: HostMessage = match serde_json::from_str(&line) {
208            Ok(m) => m,
209            Err(e) => {
210                host.warn(format!("unreadable message from the recorder ({e}): {line}"));
211                continue;
212            }
213        };
214        let Some(p) = plugin.as_mut() else {
215            let HostMessage::Hello(hello) = msg else {
216                host.warn("a message before hello, ignored");
217                continue;
218            };
219            match start::<P>(host.clone(), hello) {
220                Ok(p) => {
221                    plugin = Some(p);
222                    host.send(&PluginMessage::Ready);
223                }
224                Err(e) => {
225                    host.status(State::Error, e.clone());
226                    host.error(e);
227                    return EXIT_CONFIG;
228                }
229            }
230            continue;
231        };
232        match msg {
233            HostMessage::Hello(_) => host.warn("a second hello, ignored"),
234            HostMessage::CallStart(c) => p.call_start(c),
235            HostMessage::CallEnd(c) => p.call_end(c),
236            HostMessage::CallConcluded(c) => p.call_concluded(c),
237            HostMessage::Unit(u) => p.unit(u),
238            HostMessage::Audio(a) => p.audio(a),
239            HostMessage::Status(s) => p.status(s),
240            HostMessage::Shutdown(s) => {
241                grace = std::time::Duration::from_secs_f64(s.grace_s.max(0.0));
242                break;
243            }
244            HostMessage::Unknown => {}
245        }
246    }
247    if let Some(p) = plugin.as_mut() {
248        p.shutdown(grace);
249    }
250    0
251}
252
253fn start<P: Plugin>(host: Host, hello: Hello) -> Result<P, String> {
254    if hello.api != 0 && hello.api < API_VERSION {
255        return Err(format!("this plugin needs a newer recorder (plugin API {API_VERSION}, the recorder has {})", hello.api));
256    }
257    let config: P::Config = parse(hello.config.clone(), "settings")?;
258    let mut systems = Vec::new();
259    for s in hello.systems {
260        let settings = match s.settings {
261            Value::Null => parse::<P::Config>(hello.config.clone(), "settings")?,
262            v => parse::<P::Config>(v, &format!("settings for {}", s.short_name))?,
263        };
264        let config = match s.config {
265            Value::Null => None,
266            v => Some(parse::<P::SystemConfig>(v, &format!("settings for {}", s.short_name))?),
267        };
268        systems.push(System { index: s.index, short_name: s.short_name, kind: s.kind, config, settings });
269    }
270    let setup = Setup { config, systems, host: hello.host, capture_dir: hello.capture_dir, data_dir: hello.data_dir, audio_formats: hello.audio_formats };
271    P::start(host, setup)
272}
273
274/// Settings left out entirely count as `{}`, so `#[serde(default)]` applies.
275fn parse<T: DeserializeOwned>(v: Value, what: &str) -> Result<T, String> {
276    let v = if v.is_null() { Value::Object(Default::default()) } else { v };
277    serde_json::from_value(v).map_err(|e| format!("bad {what}: {e}"))
278}
279
280/// A manifest with the id, version, description, homepage, repository,
281/// authors and license from Cargo.toml; the name is the id until you set one:
282///
283/// ```ignore
284/// fn manifest() -> Manifest {
285///     Manifest { name: "OpenMHz".into(), subscribe: vec![topic::CALL_CONCLUDED.into()], ..trunk_recorder_plugin::manifest!() }
286/// }
287/// ```
288#[macro_export]
289macro_rules! manifest {
290    () => {{
291        let authors: Vec<String> = env!("CARGO_PKG_AUTHORS").split(':').filter(|a| !a.is_empty()).map(String::from).collect();
292        $crate::Manifest {
293            id: env!("CARGO_PKG_NAME").to_string(),
294            name: env!("CARGO_PKG_NAME").to_string(),
295            version: env!("CARGO_PKG_VERSION").to_string(),
296            description: env!("CARGO_PKG_DESCRIPTION").to_string(),
297            api: $crate::API_VERSION,
298            homepage: env!("CARGO_PKG_HOMEPAGE").to_string(),
299            repository: env!("CARGO_PKG_REPOSITORY").to_string(),
300            authors,
301            license: env!("CARGO_PKG_LICENSE").to_string(),
302            ..Default::default()
303        }
304    }};
305}
306
307#[cfg(test)]
308mod tests {
309    use super::*;
310    use crate::testing;
311    use serde_json::json;
312
313    #[derive(serde::Deserialize, schemars::JsonSchema, Default)]
314    #[serde(default)]
315    struct Config {
316        server: String,
317        retries: u32,
318    }
319
320    /// Says each system's server.
321    struct Servers;
322
323    impl Plugin for Servers {
324        type Config = Config;
325        type SystemConfig = NoConfig;
326        fn manifest() -> Manifest {
327            Manifest { id: "servers".into(), ..Default::default() }
328        }
329        fn start(host: Host, setup: Setup<Config, NoConfig>) -> Result<Self, String> {
330            for s in &setup.systems {
331                host.info(format!("{} {} {}", s.short_name, s.settings.server, s.settings.retries));
332            }
333            Ok(Servers)
334        }
335    }
336
337    /// A system's settings are the plugin's with what it overrides.
338    #[test]
339    fn a_system_can_override_the_plugins_settings() {
340        let dir = testing::temp_dir("sdk");
341        let mut h = testing::hello(&dir, json!({ "server": "https://a", "retries": 3 }));
342        h.systems.push(SystemInfo { index: 1, short_name: "sys2".into(), kind: "p25".into(), settings: json!({ "server": "https://b", "retries": 3 }), ..Default::default() });
343        let out = testing::run::<Servers>([HostMessage::Hello(h)]);
344        assert!(out.ready());
345        let said: Vec<String> = out.logs().into_iter().map(|(_, t)| t).collect();
346        assert_eq!(said, ["sys1 https://a 3", "sys2 https://b 3"]);
347    }
348}
trunk_recorder_plugin/sdk.rs

trunk_recorder_plugin/
sdk.rs
