use std::path::{Path, PathBuf};
use anyhow::Context;
use serde::de::DeserializeOwned;
use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader};
use tokio::net::UnixStream;
use super::ipc::{DaemonRequest, DaemonResponse, ErrorKind};
use super::socket::daemon_socket_path;
#[derive(Debug, Clone)]
pub struct DaemonClient {
socket: PathBuf,
}
pub async fn daemon_client() -> anyhow::Result<DaemonClient> {
let socket = daemon_socket_path();
connect_at(&socket).await
}
pub async fn client_for(socket: Option<&Path>) -> anyhow::Result<DaemonClient> {
match socket {
Some(path) => connect_at(path).await,
None => daemon_client().await,
}
}
pub fn daemon_client_unavailable(reason: impl std::fmt::Display) -> anyhow::Error {
anyhow::anyhow!("daemon unavailable: {reason} (start it with `hallouminate daemon`)")
}
pub async fn connect_at(socket: &Path) -> anyhow::Result<DaemonClient> {
UnixStream::connect(socket).await.with_context(|| {
format!(
"daemon unavailable: cannot connect to {} \
(start it with `hallouminate daemon`)",
socket.display()
)
})?;
Ok(DaemonClient {
socket: socket.to_path_buf(),
})
}
impl DaemonClient {
pub fn socket_path(&self) -> &Path {
&self.socket
}
pub async fn call_raw(&self, req: DaemonRequest) -> anyhow::Result<DaemonResponse> {
let mut stream = UnixStream::connect(&self.socket).await.map_err(|e| {
daemon_client_unavailable(format!("connect to {} failed: {e}", self.socket.display()))
})?;
let mut text = serde_json::to_string(&req)?;
text.push('\n');
stream.write_all(text.as_bytes()).await.map_err(|e| {
daemon_client_unavailable(format!("write to {} failed: {e}", self.socket.display()))
})?;
stream.flush().await.map_err(|e| {
daemon_client_unavailable(format!("flush {} failed: {e}", self.socket.display()))
})?;
let (read_half, _) = stream.into_split();
let mut reader = BufReader::new(read_half);
let mut line = String::new();
let n = reader.read_line(&mut line).await.map_err(|e| {
daemon_client_unavailable(format!("read from {} failed: {e}", self.socket.display()))
})?;
if n == 0 {
return Err(daemon_client_unavailable(format!(
"daemon at {} closed the connection before responding",
self.socket.display(),
)));
}
let response: DaemonResponse = serde_json::from_str(line.trim_end()).map_err(|e| {
daemon_client_unavailable(format!(
"invalid daemon response from {}: {e} (response: {line:?})",
self.socket.display(),
))
})?;
Ok(response)
}
pub async fn call<T: DeserializeOwned>(&self, req: DaemonRequest) -> anyhow::Result<T> {
match self.call_raw(req).await? {
DaemonResponse::Ok { result } => serde_json::from_value(result)
.map_err(|e| anyhow::anyhow!("daemon returned unexpected payload: {e}")),
DaemonResponse::Err { kind, message } => match kind {
ErrorKind::InvalidParams => Err(DaemonRpcError::invalid_params(message).into()),
ErrorKind::Internal => Err(DaemonRpcError::internal(message).into()),
},
}
}
}
#[derive(Debug)]
pub struct DaemonRpcError {
pub kind: ErrorKind,
pub message: String,
}
impl DaemonRpcError {
pub fn invalid_params(msg: impl Into<String>) -> Self {
Self {
kind: ErrorKind::InvalidParams,
message: msg.into(),
}
}
pub fn internal(msg: impl Into<String>) -> Self {
Self {
kind: ErrorKind::Internal,
message: msg.into(),
}
}
}
impl std::fmt::Display for DaemonRpcError {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
write!(f, "{}", self.message)
}
}
impl std::error::Error for DaemonRpcError {}