escouade_plugin_kit/
serve.rs

1//! The loop of a plugin (docs/PLUGINS.md §3, PLG-2), asynchronous: JSON-RPC 2.0, one message a line. Each request is
2//! answered when it ends, several at once (a slow call does not hold back the others); a batch is answered as a batch.
3//! While it answers, the plugin may ask Escouade (`Cx::request`: its secrets, its settings) and write to its journal
4//! (`Cx::log`). `initialize` and `shutdown` are answered here.
5//!
6//! The same [`Plugin`] and the same [`Cx`] on both targets. In process, the loop is on tokio's threads (`process`); in a
7//! module, a reactor on one thread, without tokio (`reactor`, docs/PLUGINS.md §8.3).
8
9use crate::forge::BoxFuture;
10use escouade_plugin_protocol::{code, greeted, initialized, method};
11use parking_lot::Mutex;
12use serde_json::{json, Value};
13use std::collections::{HashMap, HashSet};
14use std::future::poll_fn;
15use std::sync::atomic::{AtomicU64, Ordering};
16use std::sync::Arc;
17use std::task::{Poll, Waker};
18
19#[cfg(not(target_family = "wasm"))]
20mod process;
21#[cfg(any(test, target_family = "wasm"))]
22pub(crate) mod reactor;
23
24#[cfg(not(target_family = "wasm"))]
25pub use process::{run, serve_on};
26#[cfg(target_family = "wasm")]
27pub use reactor::run;
28
29/// A refusal: its code and its message (`escouade_plugin_protocol::code`).
30pub use escouade_plugin_protocol::Refusal;
31
32/// A plugin: what it can do, and its answer to each method but the handshake's.
33pub trait Plugin: Send + Sync + 'static {
34    /// What `initialize` answers it can do.
35    fn capabilities(&self) -> Value {
36        json!({})
37    }
38
39    fn call<'a>(
40        &'a self,
41        cx: &'a Cx,
42        method: &'a str,
43        params: Value,
44    ) -> BoxFuture<'a, Result<Value, Refusal>>;
45}
46
47/// A call to Escouade that waits: its answer when it came, and who to tell.
48#[derive(Default)]
49struct Waiting {
50    answer: Option<Result<Value, Refusal>>,
51    waker: Option<Waker>,
52}
53
54/// How a line goes out; false when it cannot any more.
55type Out = Box<dyn Fn(String) -> bool + Send + Sync>;
56
57/// What a call may ask of Escouade while it runs.
58pub struct Cx {
59    out: Out,
60    pending: Mutex<HashMap<u64, Arc<Mutex<Waiting>>>>,
61    next: AtomicU64,
62    /// The timers that rang in this turn, by what `timer/set` answered (a module's; `crate::time`).
63    fired: Mutex<HashSet<String>>,
64}
65
66impl Cx {
67    fn new(out: impl Fn(String) -> bool + Send + Sync + 'static) -> Self {
68        Self {
69            out: Box::new(out),
70            pending: Mutex::new(HashMap::new()),
71            next: AtomicU64::new(1),
72            fired: Mutex::new(HashSet::new()),
73        }
74    }
75
76    fn send(&self, line: String) -> bool {
77        (self.out)(line)
78    }
79
80    /// Asks Escouade `method`: its result, or its refusal (one that does not come back: Escouade closed).
81    pub async fn request(&self, method: &str, params: Value) -> Result<Value, Refusal> {
82        let id = self.next.fetch_add(1, Ordering::Relaxed);
83        let waiting = Arc::new(Mutex::new(Waiting::default()));
84        self.pending.lock().insert(id, waiting.clone());
85        let line = json!({ "jsonrpc": "2.0", "id": id, "method": method, "params": params });
86        if !self.send(line.to_string()) {
87            self.pending.lock().remove(&id);
88            return Err(Refusal {
89                code: -32000,
90                message: "Escouade is not answering".into(),
91            });
92        }
93        poll_fn(move |task| {
94            let mut w = waiting.lock();
95            match w.answer.take() {
96                Some(answer) => Poll::Ready(answer),
97                None => {
98                    w.waker = Some(task.waker().clone());
99                    Poll::Pending
100                }
101            }
102        })
103        .await
104    }
105
106    /// The secret `key` of the plugin (`permissions.secrets`): none when Escouade keeps none.
107    pub async fn secret(&self, key: &str) -> Result<Option<String>, Refusal> {
108        let v = self.request("secrets/get", json!({ "key": key })).await?;
109        Ok(v.as_str().map(str::to_string))
110    }
111
112    /// A notification to Escouade (`method`, no answer).
113    pub fn notify(&self, method: &str, params: Value) {
114        let line = json!({ "jsonrpc": "2.0", "method": method, "params": params });
115        self.send(line.to_string());
116    }
117
118    /// A line of the plugin's journal.
119    pub fn log(&self, level: &str, message: &str) {
120        self.notify("log", json!({ "level": level, "message": message }));
121    }
122
123    /// Whether the timer `key` rang; said once.
124    #[cfg(any(test, target_family = "wasm"))]
125    pub(crate) fn rang(&self, key: &str) -> bool {
126        self.fired.lock().remove(key)
127    }
128
129    fn answered(&self, msg: &Value) {
130        let Some(id) = msg.get("id").and_then(Value::as_u64) else {
131            return;
132        };
133        let Some(waiting) = self.pending.lock().remove(&id) else {
134            return;
135        };
136        let answer = match msg.get("error") {
137            Some(e) => Err(Refusal {
138                code: e["code"].as_i64().unwrap_or(-32000),
139                message: e["message"].as_str().unwrap_or("").to_string(),
140            }),
141            None => Ok(msg.get("result").cloned().unwrap_or(Value::Null)),
142        };
143        let mut w = waiting.lock();
144        w.answer = Some(answer);
145        if let Some(waker) = w.waker.take() {
146            waker.wake();
147        }
148    }
149}
150
151fn answer(id: &Value, result: Result<Value, Refusal>) -> Value {
152    match result {
153        Ok(result) => json!({ "jsonrpc": "2.0", "id": id, "result": result }),
154        Err(e) => json!({
155            "jsonrpc": "2.0",
156            "id": id,
157            "error": { "code": e.code, "message": e.message },
158        }),
159    }
160}
161
162fn parse_error() -> Refusal {
163    Refusal {
164        code: code::PARSE,
165        message: "not a JSON-RPC request".into(),
166    }
167}
168
169/// An answer of Escouade to one of the plugin's requests.
170fn is_answer(m: &Value) -> bool {
171    m.get("method").is_none() && m.get("id").is_some()
172}
173
174/// A line that came: what the loop has to run, when it is a call. The rest is dealt with here: an answer of Escouade
175/// goes to who waits for it, a timer of a module rings, what is no JSON is refused.
176fn arrived(cx: &Cx, line: &str) -> Option<Value> {
177    if line.trim().is_empty() {
178        return None;
179    }
180    let Ok(msg) = serde_json::from_str::<Value>(line) else {
181        cx.send(answer(&Value::Null, Err(parse_error())).to_string());
182        return None;
183    };
184    if is_answer(&msg) {
185        cx.answered(&msg);
186        return None;
187    }
188    if let Value::Array(items) = &msg {
189        if !items.is_empty() && items.iter().all(is_answer) {
190            items.iter().for_each(|m| cx.answered(m));
191            return None;
192        }
193    }
194    if msg["method"] == "timer/fired" {
195        cx.fired.lock().insert(msg["params"]["timer"].to_string());
196        return None;
197    }
198    Some(msg)
199}
200
201/// What a request is answered (none for a notification), and whether it was `shutdown`.
202async fn one<P: Plugin + ?Sized>(plugin: &P, cx: &Cx, msg: Value) -> (Option<Value>, bool) {
203    let id = msg.get("id").cloned();
204    let Some(name) = msg
205        .get("method")
206        .and_then(Value::as_str)
207        .map(str::to_string)
208    else {
209        return (Some(answer(&Value::Null, Err(parse_error()))), false);
210    };
211    let params = msg.get("params").cloned().unwrap_or(Value::Null);
212    let result = match name.as_str() {
213        method::INITIALIZE => {
214            greeted(&params);
215            crate::i18n::from_greeting();
216            Ok(initialized(plugin.capabilities()))
217        }
218        method::SHUTDOWN => Ok(Value::Null),
219        other => plugin.call(cx, other, params).await,
220    };
221    (id.map(|id| answer(&id, result)), name == method::SHUTDOWN)
222}
223
224/// A call or a batch to its end: what goes out, and whether it was `shutdown`.
225async fn called<P: Plugin + ?Sized>(plugin: &P, cx: &Cx, msg: Value) -> (Option<Value>, bool) {
226    match msg {
227        Value::Array(items) => {
228            let mut out = Vec::new();
229            let mut shut = false;
230            for m in items {
231                let (a, s) = one(plugin, cx, m).await;
232                out.extend(a);
233                shut |= s;
234            }
235            ((!out.is_empty()).then_some(Value::Array(out)), shut)
236        }
237        m => one(plugin, cx, m).await,
238    }
239}
escouade_plugin_kit/serve.rs

escouade_plugin_kit/
serve.rs
