rusty_time_ctl/
lib.rs

1//! The control-plane *client* transport.
2//!
3//! Lives here rather than in the daemon so there is exactly one implementation
4//! of "send an op, read the answer", and so a test or an agent can drive a
5//! running daemon by linking this crate instead of shelling out to the CLI.
6//!
7//! The op types are in `rusty_time-api`, shared with wasm and the future mesh
8//! transport; this transport is native-only. Which transport a `--control`
9//! argument resolves to is decided by `rusty_time_api::control_endpoint`, so
10//! the daemon and this client always agree.
11
12use rusty_time_api::{ControlEndpoint, ControlRequest, ControlResponse};
13use std::io::{BufRead, BufReader, Read, Write};
14use std::time::Duration;
15
16/// Bound on one response line — a daemon has no business sending more, and an
17/// unbounded read would be a memory lever even from a local peer.
18const MAX_RESPONSE_BYTES: u64 = 1024 * 1024;
19const TIMEOUT: Duration = Duration::from_secs(5);
20
21/// Where the control plane lives by default. Defined in `rusty_time-api` so
22/// the daemon and this client cannot disagree.
23pub fn default_path() -> String {
24    rusty_time_api::default_control_spec()
25}
26
27/// Send one op to a running daemon and read its answer.
28pub fn request(spec: &str, req: &ControlRequest) -> Result<ControlResponse, String> {
29    let mut line = serde_json::to_string(req).map_err(|e| e.to_string())?;
30    line.push('\n');
31
32    match rusty_time_api::control_endpoint(spec) {
33        #[cfg(unix)]
34        ControlEndpoint::UnixPath(path) => {
35            use std::os::unix::net::UnixStream;
36            let stream =
37                UnixStream::connect(&path).map_err(|e| format!("connecting {path}: {e}"))?;
38            stream
39                .set_read_timeout(Some(TIMEOUT))
40                .map_err(|e| e.to_string())?;
41            stream
42                .set_write_timeout(Some(TIMEOUT))
43                .map_err(|e| e.to_string())?;
44            exchange(stream, &line)
45        }
46        #[cfg(not(unix))]
47        ControlEndpoint::UnixPath(path) => Err(format!(
48            "unix domain sockets are unavailable on this platform (asked for {path})"
49        )),
50        ControlEndpoint::Loopback(port) => {
51            let addr = format!("127.0.0.1:{port}");
52            let stream = std::net::TcpStream::connect(&addr)
53                .map_err(|e| format!("connecting {addr}: {e}"))?;
54            stream
55                .set_read_timeout(Some(TIMEOUT))
56                .map_err(|e| e.to_string())?;
57            stream
58                .set_write_timeout(Some(TIMEOUT))
59                .map_err(|e| e.to_string())?;
60            exchange(stream, &line)
61        }
62    }
63}
64
65/// One request, one response, on whichever stream the transport produced.
66fn exchange<S: Read + Write>(mut stream: S, line: &str) -> Result<ControlResponse, String> {
67    stream
68        .write_all(line.as_bytes())
69        .map_err(|e| format!("sending request: {e}"))?;
70    stream.flush().map_err(|e| e.to_string())?;
71
72    // Disambiguate: both Read and Write offer `by_ref`.
73    let mut reader = BufReader::new(Read::by_ref(&mut stream).take(MAX_RESPONSE_BYTES));
74    let mut response = String::new();
75    reader
76        .read_line(&mut response)
77        .map_err(|e| format!("reading response: {e}"))?;
78    if response.trim().is_empty() {
79        return Err("daemon closed the connection without answering".into());
80    }
81    serde_json::from_str(response.trim()).map_err(|e| format!("parsing response: {e}"))
82}
83
84#[cfg(test)]
85mod tests {
86    use super::*;
87
88    #[test]
89    fn a_missing_daemon_is_a_clear_error_not_a_hang() {
90        // A control name nothing is serving must fail promptly and say why.
91        let err = request(
92            "rusty_time_definitely_not_running_xyz",
93            &ControlRequest::Ping,
94        )
95        .expect_err("should fail");
96        assert!(
97            err.contains("connecting"),
98            "error should name the connect step, got: {err}"
99        );
100    }
101
102    #[test]
103    fn the_default_control_name_resolves() {
104        // Whatever the platform, the default must map to something usable
105        // rather than panicking or producing an empty spec.
106        let spec = default_path();
107        assert!(!spec.is_empty());
108        let _ = rusty_time_api::control_endpoint(&spec);
109    }
110}
rusty_time_ctl/lib.rs

rusty_time_ctl/
lib.rs
