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

vissue_control/
client.rs
