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        crate::rpc::Method::IssueAppend => {
264            Ok(Response::IssueAppend(serde_json::from_value(value)?))
265        }
266        crate::rpc::Method::IssueReject => {
267            Ok(Response::IssueReject(serde_json::from_value(value)?))
268        }
269        crate::rpc::Method::IssueResolve => {
270            Ok(Response::IssueResolve(serde_json::from_value(value)?))
271        }
272        crate::rpc::Method::IssueVote => Ok(Response::IssueVote(serde_json::from_value(value)?)),
273        crate::rpc::Method::IssueFold => Ok(Response::IssueFold(serde_json::from_value(value)?)),
274        crate::rpc::Method::IssueNormalize => {
275            Ok(Response::IssueNormalize(serde_json::from_value(value)?))
276        }
277        crate::rpc::Method::IssueCheck => Ok(Response::IssueCheck(serde_json::from_value(value)?)),
278        crate::rpc::Method::IssueCount => Ok(Response::IssueCount(serde_json::from_value(value)?)),
279        crate::rpc::Method::IssueCycles => {
280            Ok(Response::IssueCycles(serde_json::from_value(value)?))
281        }
282        crate::rpc::Method::IssueDigest => {
283            Ok(Response::IssueDigest(serde_json::from_value(value)?))
284        }
285        crate::rpc::Method::IssueExport => {
286            Ok(Response::IssueExport(serde_json::from_value(value)?))
287        }
288        crate::rpc::Method::IssueGraph => Ok(Response::IssueGraph(serde_json::from_value(value)?)),
289        crate::rpc::Method::IssueRoadmap => {
290            Ok(Response::IssueRoadmap(serde_json::from_value(value)?))
291        }
292        crate::rpc::Method::IssueStale => Ok(Response::IssueStale(serde_json::from_value(value)?)),
293        crate::rpc::Method::IssueHygiene => {
294            Ok(Response::IssueHygiene(serde_json::from_value(value)?))
295        }
296        crate::rpc::Method::IssueWaitingOn => {
297            Ok(Response::IssueWaitingOn(serde_json::from_value(value)?))
298        }
299        crate::rpc::Method::IssueMirror => {
300            Ok(Response::IssueMirror(serde_json::from_value(value)?))
301        }
302        crate::rpc::Method::EventsPing => Ok(Response::EventsPing(serde_json::from_value(value)?)),
303        crate::rpc::Method::EventsWait => Ok(Response::EventsWait(serde_json::from_value(value)?)),
304    }
305}
306
307#[cfg(test)]
308mod tests {
309    use super::*;
310    use crate::frame::{Framing, read_message, write_message};
311    use crate::rpc::{JsonRpcRequest, NOTIFY_VAULT_CHANGED};
312    use serde_json::json;
313    use std::io::{BufReader, Write};
314    use std::os::unix::net::UnixListener;
315    use std::sync::{Arc, Mutex};
316    use std::thread;
317
318    fn serve_one(
319        path: &Path,
320        framing: Framing,
321        reply: impl FnOnce(JsonRpcRequest) -> Value + Send + 'static,
322    ) {
323        let listener = UnixListener::bind(path).unwrap();
324        thread::spawn(move || {
325            let (stream, _) = listener.accept().unwrap();
326            let mut reader = BufReader::new(stream.try_clone().unwrap());
327            let mut writer = stream;
328            let (payload, got) = read_message(&mut reader).unwrap();
329            assert_eq!(got, framing);
330            let req: JsonRpcRequest = serde_json::from_slice(&payload).unwrap();
331            let body = reply(req);
332            let bytes = serde_json::to_vec(&body).unwrap();
333            write_message(&mut writer, &bytes, framing).unwrap();
334            writer.flush().unwrap();
335        });
336    }
337
338    #[test]
339    fn request_roundtrip_jsonl() {
340        let dir = tempfile::tempdir().unwrap();
341        let sock = dir.path().join("control.sock");
342        serve_one(
343            &sock,
344            Framing::Jsonl,
345            |req| json!({"jsonrpc":"2.0","id":req.id,"result":{"identity":"rg"}}),
346        );
347        let mut client = Client::connect(&sock).unwrap();
348        assert_eq!(client.framing(), Framing::Jsonl);
349        let result = client.request("identity/get", json!({})).unwrap();
350        assert_eq!(result["identity"], "rg");
351    }
352
353    #[test]
354    fn request_roundtrip_headers() {
355        let dir = tempfile::tempdir().unwrap();
356        let sock = dir.path().join("control.sock");
357        serve_one(
358            &sock,
359            Framing::Headers,
360            |req| json!({"jsonrpc":"2.0","id":req.id,"result":{"ok":true}}),
361        );
362        let mut client = Client::connect_with_framing(&sock, Framing::Headers).unwrap();
363        let result = client.request_typed(&Request::IdentityGet).unwrap();
364        assert_eq!(result["ok"], true);
365    }
366
367    #[test]
368    fn notification_callback_fires_before_result() {
369        let dir = tempfile::tempdir().unwrap();
370        let sock = dir.path().join("control.sock");
371        let listener = UnixListener::bind(&sock).unwrap();
372        thread::spawn(move || {
373            let (stream, _) = listener.accept().unwrap();
374            let mut reader = BufReader::new(stream.try_clone().unwrap());
375            let mut writer = stream;
376            let (payload, framing) = read_message(&mut reader).unwrap();
377            let req: JsonRpcRequest = serde_json::from_slice(&payload).unwrap();
378            let note = json!({
379                "jsonrpc":"2.0",
380                "method": NOTIFY_VAULT_CHANGED,
381                "params": {"generation": 9, "revision": 3, "projects": ["atlas"]}
382            });
383            write_message(&mut writer, &serde_json::to_vec(&note).unwrap(), framing).unwrap();
384            let result = json!({"jsonrpc":"2.0","id":req.id,"result":{"ok":true}});
385            write_message(&mut writer, &serde_json::to_vec(&result).unwrap(), framing).unwrap();
386            writer.flush().unwrap();
387        });
388
389        let seen = Arc::new(Mutex::new(Vec::new()));
390        let seen_cb = Arc::clone(&seen);
391        let mut client = Client::connect(&sock).unwrap();
392        client.on_notification(move |n| seen_cb.lock().unwrap().push(n.method().to_string()));
393        let result = client.request("events/gen", json!({})).unwrap();
394        assert_eq!(result["ok"], true);
395        assert_eq!(seen.lock().unwrap().as_slice(), [NOTIFY_VAULT_CHANGED]);
396    }
397
398    #[test]
399    fn null_response_id_is_rpc_error() {
400        let dir = tempfile::tempdir().unwrap();
401        let sock = dir.path().join("control.sock");
402        serve_one(&sock, Framing::Jsonl, |_req| {
403            json!({
404                "jsonrpc":"2.0",
405                "id": null,
406                "error": {"code": -32600, "message": "invalid request"}
407            })
408        });
409        let mut client = Client::connect(&sock).unwrap();
410        let err = client.request("identity/get", json!({})).unwrap_err();
411        match err {
412            Error::Rpc(e) => {
413                assert_eq!(e.code, -32600);
414                assert_eq!(e.message, "invalid request");
415            }
416            other => panic!("{other:?}"),
417        }
418    }
419
420    #[test]
421    fn unmatched_response_id_is_rpc_error() {
422        let dir = tempfile::tempdir().unwrap();
423        let sock = dir.path().join("control.sock");
424        serve_one(
425            &sock,
426            Framing::Jsonl,
427            |_req| json!({"jsonrpc":"2.0","id": 99, "result":{"ok":true}}),
428        );
429        let mut client = Client::connect(&sock).unwrap();
430        let err = client.request("identity/get", json!({})).unwrap_err();
431        match err {
432            Error::Rpc(e) => {
433                assert_eq!(e.code, -32600);
434                assert_eq!(e.message, "response id does not match request");
435            }
436            other => panic!("{other:?}"),
437        }
438    }
439
440    #[test]
441    fn rpc_error_is_returned() {
442        let dir = tempfile::tempdir().unwrap();
443        let sock = dir.path().join("control.sock");
444        serve_one(&sock, Framing::Jsonl, |req| {
445            json!({
446                "jsonrpc":"2.0",
447                "id": req.id,
448                "error": {"code": -32601, "message": "method not found", "data": {"method": "nope"}}
449            })
450        });
451        let mut client = Client::connect(&sock).unwrap();
452        let err = client.request("nope", json!({})).unwrap_err();
453        match err {
454            Error::Rpc(e) => assert_eq!(e.code, -32601),
455            other => panic!("{other:?}"),
456        }
457    }
458
459    #[test]
460    fn wait_notification_reads_a_push() {
461        let dir = tempfile::tempdir().unwrap();
462        let sock = dir.path().join("control.sock");
463        let listener = UnixListener::bind(&sock).unwrap();
464        thread::spawn(move || {
465            let (stream, _) = listener.accept().unwrap();
466            let mut writer = stream;
467            let note = json!({
468                "jsonrpc":"2.0",
469                "method": NOTIFY_VAULT_CHANGED,
470                "params": {"generation": 2, "revision": 4, "projects": []}
471            });
472            write_message(
473                &mut writer,
474                &serde_json::to_vec(&note).unwrap(),
475                Framing::Jsonl,
476            )
477            .unwrap();
478            writer.flush().unwrap();
479            thread::sleep(std::time::Duration::from_millis(50));
480        });
481        let mut client = Client::connect(&sock).unwrap();
482        let note = client
483            .wait_notification(std::time::Duration::from_secs(2))
484            .unwrap();
485        assert_eq!(note.method(), NOTIFY_VAULT_CHANGED);
486    }
487
488    #[test]
489    fn notify_writes_without_id() {
490        let dir = tempfile::tempdir().unwrap();
491        let sock = dir.path().join("control.sock");
492        let listener = UnixListener::bind(&sock).unwrap();
493        let handle = thread::spawn(move || {
494            let (stream, _) = listener.accept().unwrap();
495            let mut reader = BufReader::new(stream);
496            let (payload, _) = read_message(&mut reader).unwrap();
497            let req: JsonRpcRequest = serde_json::from_slice(&payload).unwrap();
498            assert!(req.is_notification());
499            assert_eq!(req.method, "serve/shutting_down");
500        });
501        let mut client = Client::connect(&sock).unwrap();
502        client.notify("serve/shutting_down", json!({})).unwrap();
503        handle.join().unwrap();
504    }
505
506    #[test]
507    fn decode_response_covers_methods() {
508        let value = json!({"protocolVersion":1,"capabilities":[],"root":"/","prefix":"Software","generation":1,"revision":1,"identity":"a"});
509        match decode_response("initialize", value).unwrap() {
510            Response::Initialize(r) => assert_eq!(r.protocol_version, 1),
511            other => panic!("{other:?}"),
512        }
513        let list = json!({"issues":[],"revision":1});
514        assert!(matches!(
515            decode_response("issue/list", list.clone()).unwrap(),
516            Response::IssueList(_)
517        ));
518        assert!(matches!(
519            decode_response("issue/ready", list).unwrap(),
520            Response::IssueReady(_)
521        ));
522        assert!(matches!(
523            decode_response("events/gen", json!({"generation":1,"revision":1})).unwrap(),
524            Response::EventsGen(_)
525        ));
526        assert!(decode_response("issue/fold", json!({})).is_err());
527
528        let detail = json!({
529            "id":"atlas-1a2b","project":"atlas","title":"t","state":"TODO","priority":"B",
530            "properties":{},"org_tags":[],"tags":[],"blocked_by":[],"parent":null,
531            "claimed_by":null,"claimed_at":null,"file":"f","line_start":1,"line_end":2,
532            "revision":1
533        });
534        for method in [
535            "issue/get",
536            "issue/show",
537            "issue/open",
538            "issue/excerpt",
539            "issue/search",
540            "issue/claims",
541            "issue/agenda",
542            "issue/tree",
543            "issue/related",
544            "issue/children",
545            "issue/ancestors",
546            "issue/impact",
547            "issue/backlinks",
548            "issue/create",
549            "issue/update",
550            "issue/claim",
551            "issue/note",
552            "issue/refile",
553            "project/list",
554            "events/since",
555            "identity/get",
556        ] {
557            let value = match method {
558                "issue/get" | "issue/show" | "issue/open" => detail.clone(),
559                "issue/excerpt" => json!({
560                    "id":"atlas-1a2b","file":"f","line_start":1,"line_end":2,
561                    "text":"","suppressed":false
562                }),
563                "issue/search" | "issue/claims" | "issue/agenda" | "issue/related"
564                | "issue/children" | "issue/ancestors" | "issue/impact" | "issue/backlinks" => {
565                    json!([])
566                }
567                "issue/tree" => json!({"text": "* a"}),
568                "issue/create" | "issue/update" | "issue/claim" | "issue/note" | "issue/refile" => {
569                    json!({"ok":true,"report":"","issue":null,"revision":1,"generation":1})
570                }
571                "project/list" => json!({"projects":[],"revision":1}),
572                "events/since" => json!({"events":[],"generation":1}),
573                "identity/get" => {
574                    json!({"identity":"a","root":"/","prefix":"Software","version":"0.2.0"})
575                }
576                _ => json!({}),
577            };
578            decode_response(method, value).expect(method);
579        }
580    }
581}
vissue_control/client.rs

vissue_control/
client.rs
