jsonrpc/
simple_uds.rs

1// SPDX-License-Identifier: CC0-1.0
2
3//! This module implements a synchronous transport over a raw [`std::net::TcpListener`].
4
5use std::os::unix::net::UnixStream;
6use std::{error, fmt, io, path, time};
7
8use crate::client::Transport;
9use crate::{Request, Response};
10
11/// Simple synchronous UDS transport.
12#[derive(Debug, Clone)]
13pub struct UdsTransport {
14    /// The path to the Unix Domain Socket.
15    pub sockpath: path::PathBuf,
16    /// The read and write timeout to use.
17    pub timeout: Option<time::Duration>,
18}
19
20impl UdsTransport {
21    /// Creates a new [`UdsTransport`] without timeouts to use.
22    pub fn new<P: AsRef<path::Path>>(sockpath: P) -> UdsTransport {
23        UdsTransport {
24            sockpath: sockpath.as_ref().to_path_buf(),
25            timeout: None,
26        }
27    }
28
29    fn request<R>(&self, req: impl serde::Serialize) -> Result<R, Error>
30    where
31        R: for<'a> serde::de::Deserialize<'a>,
32    {
33        let mut sock = UnixStream::connect(&self.sockpath)?;
34        sock.set_read_timeout(self.timeout)?;
35        sock.set_write_timeout(self.timeout)?;
36
37        serde_json::to_writer(&mut sock, &req)?;
38
39        // NOTE: we don't check the id there, so it *must* be synchronous
40        let resp: R = serde_json::Deserializer::from_reader(&mut sock)
41            .into_iter()
42            .next()
43            .ok_or(Error::Timeout)??;
44        Ok(resp)
45    }
46}
47
48impl Transport for UdsTransport {
49    fn send_request(&self, req: Request) -> Result<Response, crate::error::Error> {
50        Ok(self.request(req)?)
51    }
52
53    fn send_batch(&self, reqs: &[Request]) -> Result<Vec<Response>, crate::error::Error> {
54        Ok(self.request(reqs)?)
55    }
56
57    fn fmt_target(&self, f: &mut fmt::Formatter) -> fmt::Result {
58        write!(f, "{}", self.sockpath.to_string_lossy())
59    }
60}
61
62/// Error that can occur while using the UDS transport.
63#[derive(Debug)]
64pub enum Error {
65    /// An error occurred on the socket layer.
66    SocketError(io::Error),
67    /// We didn't receive a complete response till the deadline ran out.
68    Timeout,
69    /// JSON parsing error.
70    Json(serde_json::Error),
71}
72
73impl fmt::Display for Error {
74    fn fmt(&self, f: &mut fmt::Formatter) -> Result<(), fmt::Error> {
75        use Error::*;
76
77        match *self {
78            SocketError(ref e) => write!(f, "couldn't connect to host: {}", e),
79            Timeout => f.write_str("didn't receive response data in time, timed out."),
80            Json(ref e) => write!(f, "JSON error: {}", e),
81        }
82    }
83}
84
85impl error::Error for Error {
86    fn source(&self) -> Option<&(dyn error::Error + 'static)> {
87        use self::Error::*;
88
89        match *self {
90            SocketError(ref e) => Some(e),
91            Timeout => None,
92            Json(ref e) => Some(e),
93        }
94    }
95}
96
97impl From<io::Error> for Error {
98    fn from(e: io::Error) -> Self {
99        Error::SocketError(e)
100    }
101}
102
103impl From<serde_json::Error> for Error {
104    fn from(e: serde_json::Error) -> Self {
105        Error::Json(e)
106    }
107}
108
109impl From<Error> for crate::error::Error {
110    fn from(e: Error) -> crate::error::Error {
111        match e {
112            Error::Json(e) => crate::error::Error::Json(e),
113            e => crate::error::Error::Transport(Box::new(e)),
114        }
115    }
116}
117
118#[cfg(test)]
119mod tests {
120    use std::{
121        fs,
122        io::{Read, Write},
123        os::unix::net::UnixListener,
124        process, thread,
125    };
126
127    use super::*;
128    use crate::Client;
129
130    // Test a dummy request / response over an UDS
131    #[test]
132    fn sanity_check_uds_transport() {
133        let socket_path: path::PathBuf = format!("uds_scratch_{}.socket", process::id()).into();
134        // Any leftover?
135        fs::remove_file(&socket_path).unwrap_or(());
136
137        let server = UnixListener::bind(&socket_path).unwrap();
138        let dummy_req = Request {
139            method: "getinfo",
140            params: &[],
141            id: serde_json::Value::Number(111.into()),
142            jsonrpc: Some("2.0"),
143        };
144        let dummy_req_ser = serde_json::to_vec(&dummy_req).unwrap();
145        let dummy_resp = Response {
146            result: None,
147            error: None,
148            id: serde_json::Value::Number(111.into()),
149            jsonrpc: Some("2.0".into()),
150        };
151        let dummy_resp_ser = serde_json::to_vec(&dummy_resp).unwrap();
152
153        let cli_socket_path = socket_path.clone();
154        let client_thread = thread::spawn(move || {
155            let transport = UdsTransport {
156                sockpath: cli_socket_path,
157                timeout: Some(time::Duration::from_secs(5)),
158            };
159            let client = Client::with_transport(transport);
160
161            client.send_request(dummy_req.clone()).unwrap()
162        });
163
164        let (mut stream, _) = server.accept().unwrap();
165        stream.set_read_timeout(Some(time::Duration::from_secs(5))).unwrap();
166        let mut recv_req = vec![0; dummy_req_ser.len()];
167        let mut read = 0;
168        while read < dummy_req_ser.len() {
169            read += stream.read(&mut recv_req[read..]).unwrap();
170        }
171        assert_eq!(recv_req, dummy_req_ser);
172
173        stream.write_all(&dummy_resp_ser).unwrap();
174        stream.flush().unwrap();
175        let recv_resp = client_thread.join().unwrap();
176        assert_eq!(serde_json::to_vec(&recv_resp).unwrap(), dummy_resp_ser);
177
178        // Clean up
179        drop(server);
180        fs::remove_file(&socket_path).unwrap();
181    }
182}
jsonrpc/simple_uds.rs

jsonrpc/
simple_uds.rs
