omni_dev/daemon/
client.rs

1//! [`DaemonClient`]: a thin client for the daemon's Unix control socket, used
2//! by `daemon status` / `stop` / `restart` and the single-instance `ping`
3//! probe.
4
5use std::path::{Path, PathBuf};
6
7use anyhow::{bail, Context, Result};
8use futures::{SinkExt, StreamExt};
9use tokio::net::UnixStream;
10use tokio_util::codec::{Framed, LinesCodec};
11
12use super::protocol::{DaemonEnvelope, DaemonReply, StatusReport, MAX_LINE_BYTES};
13
14/// A one-shot client over the daemon's Unix-domain control socket. Each call
15/// opens a fresh connection, sends one [`DaemonEnvelope`], and reads one
16/// [`DaemonReply`].
17#[derive(Debug, Clone)]
18pub struct DaemonClient {
19    socket_path: PathBuf,
20}
21
22impl DaemonClient {
23    /// Creates a client targeting the socket at `socket_path`.
24    pub fn new(socket_path: impl Into<PathBuf>) -> Self {
25        Self {
26            socket_path: socket_path.into(),
27        }
28    }
29
30    /// The socket path this client targets.
31    pub fn socket_path(&self) -> &Path {
32        &self.socket_path
33    }
34
35    /// Sends one envelope and returns the daemon's reply.
36    pub async fn request(&self, envelope: DaemonEnvelope) -> Result<DaemonReply> {
37        let stream = UnixStream::connect(&self.socket_path)
38            .await
39            .with_context(|| {
40                format!(
41                    "failed to connect to daemon socket {} (is the daemon running?)",
42                    self.socket_path.display()
43                )
44            })?;
45        let mut framed = Framed::new(stream, LinesCodec::new_with_max_length(MAX_LINE_BYTES));
46        let line = serde_json::to_string(&envelope).context("failed to encode daemon request")?;
47        framed
48            .send(line)
49            .await
50            .context("failed to send daemon request")?;
51        let response = framed
52            .next()
53            .await
54            .context("daemon closed the connection without replying")?
55            .context("failed to read daemon reply")?;
56        serde_json::from_str(&response).context("failed to decode daemon reply")
57    }
58
59    /// Sends an envelope and returns its payload, turning an `ok: false` reply
60    /// into an `Err`.
61    async fn request_ok(&self, envelope: DaemonEnvelope) -> Result<serde_json::Value> {
62        let reply = self.request(envelope).await?;
63        if reply.ok {
64            Ok(reply.payload)
65        } else {
66            bail!(
67                "daemon returned an error: {}",
68                reply.error.as_deref().unwrap_or("unknown error")
69            )
70        }
71    }
72
73    /// Probes whether a live daemon is answering on the socket.
74    pub async fn ping(&self) -> Result<()> {
75        self.request_ok(DaemonEnvelope::builtin("ping"))
76            .await
77            .map(|_| ())
78    }
79
80    /// Requests aggregated per-service status.
81    pub async fn status(&self) -> Result<StatusReport> {
82        let payload = self.request_ok(DaemonEnvelope::builtin("status")).await?;
83        serde_json::from_value(payload).context("failed to decode daemon status report")
84    }
85
86    /// Asks the daemon to shut down gracefully.
87    pub async fn shutdown(&self) -> Result<()> {
88        self.request_ok(DaemonEnvelope::builtin("shutdown"))
89            .await
90            .map(|_| ())
91    }
92}
omni_dev/daemon/client.rs

omni_dev/daemon/
client.rs
