vissue_control/
client.rs

1//! Unix-socket JSON-RPC client. Clients never bind the control socket.
2
3use std::io::{BufReader, Write};
4use std::os::unix::net::UnixStream;
5use std::path::Path;
6use std::time::Duration;
7
8use serde_json::Value;
9
10use crate::frame::{Framing, read_message, write_message};
11use crate::rpc::{
12    Error, JsonRpcError, JsonRpcId, JsonRpcRequest, JsonRpcResponse, Notification, Request,
13    Response, invalid_request,
14};
15
16/// Connected client. One stream; notifications arriving during [`Self::request`]
17/// are handed to the optional callback.
18pub struct Client {
19    writer: UnixStream,
20    reader: BufReader<UnixStream>,
21    framing: Framing,
22    next_id: i64,
23    on_notification: Option<Box<dyn FnMut(Notification) + Send>>,
24}
25
26impl std::fmt::Debug for Client {
27    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
28        f.debug_struct("Client")
29            .field("framing", &self.framing)
30            .field("next_id", &self.next_id)
31            .field(
32                "on_notification",
33                &self.on_notification.as_ref().map(|_| "set"),
34            )
35            .finish_non_exhaustive()
36    }
37}
38
39impl Client {
40    /// Connect to `path`. Uses JSONL framing.
41    ///
42    /// # Errors
43    ///
44    /// Returns an error when the socket cannot be opened or cloned.
45    pub fn connect(path: impl AsRef<Path>) -> Result<Self, Error> {
46        Self::connect_with_framing(path, Framing::Jsonl)
47    }
48
49    /// Connect and send every request in `framing`. The owner replies in kind.
50    ///
51    /// # Errors
52    ///
53    /// Returns an error when the socket cannot be opened or cloned.
54    pub fn connect_with_framing(path: impl AsRef<Path>, framing: Framing) -> Result<Self, Error> {
55        let stream = UnixStream::connect(path.as_ref())?;
56        let reader = BufReader::new(stream.try_clone()?);
57        Ok(Self {
58            writer: stream,
59            reader,
60            framing,
61            next_id: 1,
62            on_notification: None,
63        })
64    }
65
66    /// Framing style this client writes.
67    pub fn framing(&self) -> Framing {
68        self.framing
69    }
70
71    /// Called for each JSON-RPC notification read while waiting on a response.
72    pub fn on_notification<F>(&mut self, callback: F)
73    where
74        F: FnMut(Notification) + Send + 'static,
75    {
76        self.on_notification = Some(Box::new(callback));
77    }
78
79    /// Send `method`/`params` and return the result object.
80    ///
81    /// # Errors
82    ///
83    /// Returns an error when the write or read fails, the payload is not JSON,
84    /// the response id does not match, or the server returns a JSON-RPC error.
85    pub fn request(&mut self, method: &str, params: Value) -> Result<Value, Error> {
86        let id = JsonRpcId::Number(self.next_id);
87        self.next_id += 1;
88        let msg = JsonRpcRequest::call(id.clone(), method, params);
89        self.write_rpc(&msg)?;
90        loop {
91            let (payload, _) = read_message(&mut self.reader)?;
92            let value: Value = serde_json::from_slice(&payload)?;
93            if is_notification(&value) {
94                let method = value
95                    .get("method")
96                    .and_then(Value::as_str)
97                    .unwrap_or("")
98                    .to_string();
99                let params = value.get("params").cloned().unwrap_or(Value::Null);
100                let note = Notification::parse(&method, params);
101                if let Some(cb) = &mut self.on_notification {
102                    cb(note);
103                }
104                continue;
105            }
106            let resp: JsonRpcResponse = serde_json::from_value(value)?;
107            if resp.id.as_ref() != Some(&id) {
108                return Err(Error::Rpc(id_mismatch(resp.error)));
109            }
110            if let Some(err) = resp.error {
111                return Err(Error::Rpc(err));
112            }
113            return Ok(resp.result.unwrap_or(Value::Null));
114        }
115    }
116
117    /// Typed request helper. Result is the raw JSON body (caller decodes).
118    ///
119    /// # Errors
120    ///
121    /// Returns an error when the write or read fails, the payload is not JSON,
122    /// the response id does not match, or the server returns a JSON-RPC error.
123    pub fn request_typed(&mut self, req: &Request) -> Result<Value, Error> {
124        self.request(req.method().as_str(), req.to_params())
125    }
126
127    /// Send a notification (no `id`). Does not wait.
128    ///
129    /// # Errors
130    ///
131    /// Returns an error when the notification cannot be encoded or written.
132    pub fn notify(&mut self, method: &str, params: Value) -> Result<(), Error> {
133        self.write_rpc(&JsonRpcRequest::notification(method, params))
134    }
135
136    /// Block until the next JSON-RPC notification, or `timeout`.
137    ///
138    /// # Errors
139    ///
140    /// Returns an error when the wait times out, the read or decode fails, or
141    /// the next message is a response rather than a notification.
142    pub fn wait_notification(&mut self, timeout: Duration) -> Result<Notification, Error> {
143        self.reader.get_ref().set_read_timeout(Some(timeout))?;
144        let result = read_next_notification(&mut self.reader);
145        let _ = self.reader.get_ref().set_read_timeout(None);
146        result
147    }
148
149    fn write_rpc(&mut self, msg: &JsonRpcRequest) -> Result<(), Error> {
150        let bytes = serde_json::to_vec(msg)?;
151        write_message(&mut self.writer, &bytes, self.framing)?;
152        self.writer.flush()?;
153        Ok(())
154    }
155}
156
157fn read_next_notification(reader: &mut BufReader<UnixStream>) -> Result<Notification, Error> {
158    loop {
159        let (payload, _) = read_message(reader)?;
160        if payload.iter().all(u8::is_ascii_whitespace) {
161            continue;
162        }
163        let value: Value = serde_json::from_slice(&payload)?;
164        if is_notification(&value) {
165            let method = value
166                .get("method")
167                .and_then(Value::as_str)
168                .unwrap_or("")
169                .to_string();
170            let params = value.get("params").cloned().unwrap_or(Value::Null);
171            return Ok(Notification::parse(&method, params));
172        }
173        let mut err = invalid_request();
174        err.message = "expected notification".into();
175        return Err(Error::Rpc(err));
176    }
177}
178
179/// True only when the object has `method` and no `id` member. `"id": null`
180/// is a response, not a notification.
181fn is_notification(value: &Value) -> bool {
182    let obj = match value.as_object() {
183        Some(o) => o,
184        None => return false,
185    };
186    obj.contains_key("method") && !obj.contains_key("id") && !obj.contains_key("result")
187}
188
189fn id_mismatch(server_error: Option<JsonRpcError>) -> JsonRpcError {
190    match server_error {
191        Some(err) => err,
192        None => {
193            let mut err = invalid_request();
194            err.message = "response id does not match request".into();
195            err
196        }
197    }
198}
199
200/// Decode a typed [`Response`] from a raw result when the method is known.
201///
202/// # Errors
203///
204/// Returns an error when `method` is not a v1 method or `value` does not match
205/// that method's result shape.
206pub fn decode_response(method: &str, value: Value) -> Result<Response, Error> {
207    match crate::rpc::Method::parse(method).map_err(Error::Rpc)? {
208        crate::rpc::Method::Initialize => Ok(Response::Initialize(serde_json::from_value(value)?)),
209        crate::rpc::Method::IdentityGet => {
210            Ok(Response::IdentityGet(serde_json::from_value(value)?))
211        }
212        crate::rpc::Method::IssueList => Ok(Response::IssueList(serde_json::from_value(value)?)),
213        crate::rpc::Method::IssueGet => Ok(Response::IssueGet(serde_json::from_value(value)?)),
214        crate::rpc::Method::IssueReady => Ok(Response::IssueReady(serde_json::from_value(value)?)),
215        crate::rpc::Method::IssueSearch => {
216            Ok(Response::IssueSearch(serde_json::from_value(value)?))
217        }
218        crate::rpc::Method::IssueClaims => {
219            Ok(Response::IssueClaims(serde_json::from_value(value)?))
220        }
221        crate::rpc::Method::IssueAgenda => {
222            Ok(Response::IssueAgenda(serde_json::from_value(value)?))
223        }
224        crate::rpc::Method::IssueShow => Ok(Response::IssueShow(serde_json::from_value(value)?)),
225        crate::rpc::Method::IssueExcerpt => {
226            Ok(Response::IssueExcerpt(serde_json::from_value(value)?))
227        }
228        crate::rpc::Method::IssueTree => Ok(Response::IssueTree(serde_json::from_value(value)?)),
229        crate::rpc::Method::IssueRelated => {
230            Ok(Response::IssueRelated(serde_json::from_value(value)?))
231        }
232        crate::rpc::Method::IssueChildren => {
233            Ok(Response::IssueChildren(serde_json::from_value(value)?))
234        }
235        crate::rpc::Method::IssueAncestors => {
236            Ok(Response::IssueAncestors(serde_json::from_value(value)?))
237        }
238        crate::rpc::Method::IssueImpact => {
239            Ok(Response::IssueImpact(serde_json::from_value(value)?))
240        }
241        crate::rpc::Method::IssueBacklinks => {
242            Ok(Response::IssueBacklinks(serde_json::from_value(value)?))
243        }
244        crate::rpc::Method::IssueOpen => Ok(Response::IssueOpen(serde_json::from_value(value)?)),
245        crate::rpc::Method::IssueCreate => {
246            Ok(Response::IssueCreate(serde_json::from_value(value)?))
247        }
248        crate::rpc::Method::IssueUpdate => {
249            Ok(Response::IssueUpdate(serde_json::from_value(value)?))
250        }
251        crate::rpc::Method::IssueClaim => Ok(Response::IssueClaim(serde_json::from_value(value)?)),
252        crate::rpc::Method::IssueNote => Ok(Response::IssueNote(serde_json::from_value(value)?)),
253        crate::rpc::Method::IssueRefile => {
254            Ok(Response::IssueRefile(serde_json::from_value(value)?))
255        }
256        crate::rpc::Method::ProjectList => {
257            Ok(Response::ProjectList(serde_json::from_value(value)?))
258        }
259        crate::rpc::Method::EventsSince => {
260            Ok(Response::EventsSince(serde_json::from_value(value)?))
261        }
262        crate::rpc::Method::EventsGen => Ok(Response::EventsGen(serde_json::from_value(value)?)),
263    }
264}
265
266#[cfg(test)]
267mod tests {
268    use super::*;
269    use crate::frame::{Framing, read_message, write_message};
270    use crate::rpc::{JsonRpcRequest, NOTIFY_VAULT_CHANGED};
271    use serde_json::json;
272    use std::io::{BufReader, Write};
273    use std::os::unix::net::UnixListener;
274    use std::sync::{Arc, Mutex};
275    use std::thread;
276
277    fn serve_one(
278        path: &Path,
279        framing: Framing,
280        reply: impl FnOnce(JsonRpcRequest) -> Value + Send + 'static,
281    ) {
282        let listener = UnixListener::bind(path).unwrap();
283        thread::spawn(move || {
284            let (stream, _) = listener.accept().unwrap();
285            let mut reader = BufReader::new(stream.try_clone().unwrap());
286            let mut writer = stream;
287            let (payload, got) = read_message(&mut reader).unwrap();
288            assert_eq!(got, framing);
289            let req: JsonRpcRequest = serde_json::from_slice(&payload).unwrap();
290            let body = reply(req);
291            let bytes = serde_json::to_vec(&body).unwrap();
292            write_message(&mut writer, &bytes, framing).unwrap();
293            writer.flush().unwrap();
294        });
295    }
296
297    #[test]
298    fn request_roundtrip_jsonl() {
299        let dir = tempfile::tempdir().unwrap();
300        let sock = dir.path().join("control.sock");
301        serve_one(
302            &sock,
303            Framing::Jsonl,
304            |req| json!({"jsonrpc":"2.0","id":req.id,"result":{"identity":"rg"}}),
305        );
306        let mut client = Client::connect(&sock).unwrap();
307        assert_eq!(client.framing(), Framing::Jsonl);
308        let result = client.request("identity/get", json!({})).unwrap();
309        assert_eq!(result["identity"], "rg");
310    }
311
312    #[test]
313    fn request_roundtrip_headers() {
314        let dir = tempfile::tempdir().unwrap();
315        let sock = dir.path().join("control.sock");
316        serve_one(
317            &sock,
318            Framing::Headers,
319            |req| json!({"jsonrpc":"2.0","id":req.id,"result":{"ok":true}}),
320        );
321        let mut client = Client::connect_with_framing(&sock, Framing::Headers).unwrap();
322        let result = client.request_typed(&Request::IdentityGet).unwrap();
323        assert_eq!(result["ok"], true);
324    }
325
326    #[test]
327    fn notification_callback_fires_before_result() {
328        let dir = tempfile::tempdir().unwrap();
329        let sock = dir.path().join("control.sock");
330        let listener = UnixListener::bind(&sock).unwrap();
331        thread::spawn(move || {
332            let (stream, _) = listener.accept().unwrap();
333            let mut reader = BufReader::new(stream.try_clone().unwrap());
334            let mut writer = stream;
335            let (payload, framing) = read_message(&mut reader).unwrap();
336            let req: JsonRpcRequest = serde_json::from_slice(&payload).unwrap();
337            let note = json!({
338                "jsonrpc":"2.0",
339                "method": NOTIFY_VAULT_CHANGED,
340                "params": {"generation": 9, "revision": 3, "projects": ["atlas"]}
341            });
342            write_message(&mut writer, &serde_json::to_vec(&note).unwrap(), framing).unwrap();
343            let result = json!({"jsonrpc":"2.0","id":req.id,"result":{"ok":true}});
344            write_message(&mut writer, &serde_json::to_vec(&result).unwrap(), framing).unwrap();
345            writer.flush().unwrap();
346        });
347
348        let seen = Arc::new(Mutex::new(Vec::new()));
349        let seen_cb = Arc::clone(&seen);
350        let mut client = Client::connect(&sock).unwrap();
351        client.on_notification(move |n| seen_cb.lock().unwrap().push(n.method().to_string()));
352        let result = client.request("events/gen", json!({})).unwrap();
353        assert_eq!(result["ok"], true);
354        assert_eq!(seen.lock().unwrap().as_slice(), [NOTIFY_VAULT_CHANGED]);
355    }
356
357    #[test]
358    fn null_response_id_is_rpc_error() {
359        let dir = tempfile::tempdir().unwrap();
360        let sock = dir.path().join("control.sock");
361        serve_one(&sock, Framing::Jsonl, |_req| {
362            json!({
363                "jsonrpc":"2.0",
364                "id": null,
365                "error": {"code": -32600, "message": "invalid request"}
366            })
367        });
368        let mut client = Client::connect(&sock).unwrap();
369        let err = client.request("identity/get", json!({})).unwrap_err();
370        match err {
371            Error::Rpc(e) => {
372                assert_eq!(e.code, -32600);
373                assert_eq!(e.message, "invalid request");
374            }
375            other => panic!("{other:?}"),
376        }
377    }
378
379    #[test]
380    fn unmatched_response_id_is_rpc_error() {
381        let dir = tempfile::tempdir().unwrap();
382        let sock = dir.path().join("control.sock");
383        serve_one(
384            &sock,
385            Framing::Jsonl,
386            |_req| json!({"jsonrpc":"2.0","id": 99, "result":{"ok":true}}),
387        );
388        let mut client = Client::connect(&sock).unwrap();
389        let err = client.request("identity/get", json!({})).unwrap_err();
390        match err {
391            Error::Rpc(e) => {
392                assert_eq!(e.code, -32600);
393                assert_eq!(e.message, "response id does not match request");
394            }
395            other => panic!("{other:?}"),
396        }
397    }
398
399    #[test]
400    fn rpc_error_is_returned() {
401        let dir = tempfile::tempdir().unwrap();
402        let sock = dir.path().join("control.sock");
403        serve_one(&sock, Framing::Jsonl, |req| {
404            json!({
405                "jsonrpc":"2.0",
406                "id": req.id,
407                "error": {"code": -32601, "message": "method not found", "data": {"method": "nope"}}
408            })
409        });
410        let mut client = Client::connect(&sock).unwrap();
411        let err = client.request("nope", json!({})).unwrap_err();
412        match err {
413            Error::Rpc(e) => assert_eq!(e.code, -32601),
414            other => panic!("{other:?}"),
415        }
416    }
417
418    #[test]
419    fn wait_notification_reads_a_push() {
420        let dir = tempfile::tempdir().unwrap();
421        let sock = dir.path().join("control.sock");
422        let listener = UnixListener::bind(&sock).unwrap();
423        thread::spawn(move || {
424            let (stream, _) = listener.accept().unwrap();
425            let mut writer = stream;
426            let note = json!({
427                "jsonrpc":"2.0",
428                "method": NOTIFY_VAULT_CHANGED,
429                "params": {"generation": 2, "revision": 4, "projects": []}
430            });
431            write_message(
432                &mut writer,
433                &serde_json::to_vec(&note).unwrap(),
434                Framing::Jsonl,
435            )
436            .unwrap();
437            writer.flush().unwrap();
438            thread::sleep(std::time::Duration::from_millis(50));
439        });
440        let mut client = Client::connect(&sock).unwrap();
441        let note = client
442            .wait_notification(std::time::Duration::from_secs(2))
443            .unwrap();
444        assert_eq!(note.method(), NOTIFY_VAULT_CHANGED);
445    }
446
447    #[test]
448    fn notify_writes_without_id() {
449        let dir = tempfile::tempdir().unwrap();
450        let sock = dir.path().join("control.sock");
451        let listener = UnixListener::bind(&sock).unwrap();
452        let handle = thread::spawn(move || {
453            let (stream, _) = listener.accept().unwrap();
454            let mut reader = BufReader::new(stream);
455            let (payload, _) = read_message(&mut reader).unwrap();
456            let req: JsonRpcRequest = serde_json::from_slice(&payload).unwrap();
457            assert!(req.is_notification());
458            assert_eq!(req.method, "serve/shutting_down");
459        });
460        let mut client = Client::connect(&sock).unwrap();
461        client.notify("serve/shutting_down", json!({})).unwrap();
462        handle.join().unwrap();
463    }
464
465    #[test]
466    fn decode_response_covers_methods() {
467        let value = json!({"protocolVersion":1,"capabilities":[],"root":"/","prefix":"Software","generation":1,"revision":1,"identity":"a"});
468        match decode_response("initialize", value).unwrap() {
469            Response::Initialize(r) => assert_eq!(r.protocol_version, 1),
470            other => panic!("{other:?}"),
471        }
472        let list = json!({"issues":[],"revision":1});
473        assert!(matches!(
474            decode_response("issue/list", list.clone()).unwrap(),
475            Response::IssueList(_)
476        ));
477        assert!(matches!(
478            decode_response("issue/ready", list).unwrap(),
479            Response::IssueReady(_)
480        ));
481        assert!(matches!(
482            decode_response("events/gen", json!({"generation":1,"revision":1})).unwrap(),
483            Response::EventsGen(_)
484        ));
485        assert!(decode_response("issue/fold", json!({})).is_err());
486
487        let detail = json!({
488            "id":"atlas-1a2b","project":"atlas","title":"t","state":"TODO","priority":"B",
489            "properties":{},"org_tags":[],"tags":[],"blocked_by":[],"parent":null,
490            "claimed_by":null,"claimed_at":null,"file":"f","line_start":1,"line_end":2,
491            "revision":1
492        });
493        for method in [
494            "issue/get",
495            "issue/show",
496            "issue/open",
497            "issue/excerpt",
498            "issue/search",
499            "issue/claims",
500            "issue/agenda",
501            "issue/tree",
502            "issue/related",
503            "issue/children",
504            "issue/ancestors",
505            "issue/impact",
506            "issue/backlinks",
507            "issue/create",
508            "issue/update",
509            "issue/claim",
510            "issue/note",
511            "issue/refile",
512            "project/list",
513            "events/since",
514            "identity/get",
515        ] {
516            let value = match method {
517                "issue/get" | "issue/show" | "issue/open" => detail.clone(),
518                "issue/excerpt" => json!({
519                    "id":"atlas-1a2b","file":"f","line_start":1,"line_end":2,
520                    "text":"","suppressed":false
521                }),
522                "issue/search" | "issue/claims" | "issue/agenda" | "issue/related"
523                | "issue/children" | "issue/ancestors" | "issue/impact" | "issue/backlinks" => {
524                    json!([])
525                }
526                "issue/tree" => json!({"text": "* a"}),
527                "issue/create" | "issue/update" | "issue/claim" | "issue/note" | "issue/refile" => {
528                    json!({"ok":true,"report":"","issue":null,"revision":1,"generation":1})
529                }
530                "project/list" => json!({"projects":[],"revision":1}),
531                "events/since" => json!({"events":[],"generation":1}),
532                "identity/get" => {
533                    json!({"identity":"a","root":"/","prefix":"Software","version":"0.2.0"})
534                }
535                _ => json!({}),
536            };
537            decode_response(method, value).expect(method);
538        }
539    }
540}
vissue_control/client.rs

vissue_control/
client.rs
