wallr_core/ipc/
mod.rs

1//! Async IPC transport (Tokio Unix socket server and client).
2//! Command/response vocabulary and validation live in `wallr-common`.
3
4pub use wallr_common::ipc::*;
5
6use std::path::Path;
7use tokio::io::{AsyncBufReadExt, AsyncReadExt, AsyncWriteExt, BufReader};
8use tokio::net::{UnixListener, UnixStream};
9
10#[derive(Debug, thiserror::Error)]
11pub enum IpcError {
12    #[error("daemon not running: {0}")]
13    DaemonNotRunning(String),
14    #[error("IPC I/O error: {0}")]
15    Io(#[from] std::io::Error),
16    #[error("protocol serialization/deserialization error: {0}")]
17    Protocol(#[from] serde_json::Error),
18}
19
20pub async fn send_ipc_command<P: AsRef<Path>>(
21    socket_path: P,
22    command: IpcCommand,
23) -> Result<IpcResponse, IpcError> {
24    let mut stream = UnixStream::connect(socket_path)
25        .await
26        .map_err(|e| IpcError::DaemonNotRunning(e.to_string()))?;
27
28    let req_data = serde_json::to_vec(&command)?;
29    stream.write_all(&req_data).await?;
30    stream.write_all(b"\n").await?;
31    stream.flush().await?;
32
33    let mut reader = BufReader::new(stream);
34    let mut response_line = String::new();
35    reader.read_line(&mut response_line).await?;
36
37    let response: IpcResponse = serde_json::from_str(&response_line)?;
38    Ok(response)
39}
40
41async fn write_response(writer: &mut tokio::io::WriteHalf<UnixStream>, response: &IpcResponse) {
42    if let Ok(res_data) = serde_json::to_vec(response) {
43        let _ = writer.write_all(&res_data).await;
44        let _ = writer.write_all(b"\n").await;
45        let _ = writer.flush().await;
46    }
47}
48
49pub async fn start_ipc_server<P, F, Fut>(socket_path: P, handler: F) -> Result<(), IpcError>
50where
51    P: AsRef<Path>,
52    F: Fn(IpcCommand) -> Fut + Send + Sync + 'static,
53    Fut: std::future::Future<Output = IpcResponse> + Send + 'static,
54{
55    let path = socket_path.as_ref();
56    if path.exists() {
57        let _ = std::fs::remove_file(path);
58    }
59
60    let listener = UnixListener::bind(path)?;
61    let handler = std::sync::Arc::new(handler);
62
63    tokio::spawn(async move {
64        loop {
65            match listener.accept().await {
66                Ok((stream, _)) => {
67                    let handler_clone = handler.clone();
68                    tokio::spawn(async move {
69                        let (reader, mut writer) = tokio::io::split(stream);
70                        let reader = BufReader::new(reader);
71                        // Bound the request: `take` caps memory even when a
72                        // client never sends a newline.
73                        let mut limited = reader.take((MAX_IPC_BYTES + 1) as u64);
74                        let mut buf = Vec::with_capacity(1024);
75                        let response = match limited.read_until(b'\n', &mut buf).await {
76                            Ok(0) => return,
77                            Ok(_) if buf.len() > MAX_IPC_BYTES => {
78                                IpcResponse::err(format!("request exceeds {MAX_IPC_BYTES} bytes"))
79                            }
80                            Ok(_) => match serde_json::from_slice::<IpcCommand>(&buf) {
81                                Ok(cmd) => match validate_command(&cmd) {
82                                    Ok(()) => handler_clone(cmd).await,
83                                    Err(reason) => IpcResponse::err(reason),
84                                },
85                                Err(e) => IpcResponse::err(format!("invalid IPC command: {e}")),
86                            },
87                            Err(e) => IpcResponse::err(format!("IPC read error: {e}")),
88                        };
89                        write_response(&mut writer, &response).await;
90                    });
91                }
92                Err(e) => {
93                    tracing::error!("IPC accept error: {:?}", e);
94                }
95            }
96        }
97    });
98
99    Ok(())
100}
wallr_core/ipc/mod.rs

wallr_core/ipc/
mod.rs
