walrus-socket 0.0.9

Unix domain socket transport for Walrus daemon communication
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95

//! Unix domain socket client for connecting to a walrus daemon.

use anyhow::Result;
use futures_core::Stream;
use std::path::{Path, PathBuf};
use tokio::net::unix::{OwnedReadHalf, OwnedWriteHalf};
use wcore::protocol::{
    api::Client,
    codec,
    message::{ClientMessage, ErrorMsg, ServerMessage, server_message},
};

/// Client configuration for connecting to a walrus daemon.
#[derive(Debug, Clone)]
pub struct ClientConfig {
    /// Daemon Unix domain socket path.
    pub socket_path: PathBuf,
}

/// Unix domain socket client for the walrus daemon.
///
/// Holds configuration. Call [`WalrusClient::connect`] to establish a
/// connection.
pub struct WalrusClient {
    config: ClientConfig,
}

impl WalrusClient {
    /// Create a new client with the given configuration.
    pub fn new(config: ClientConfig) -> Self {
        Self { config }
    }

    /// Access the client configuration.
    pub fn config(&self) -> &ClientConfig {
        &self.config
    }

    /// Set the daemon socket path.
    pub fn socket_path(mut self, path: impl Into<PathBuf>) -> Self {
        self.config.socket_path = path.into();
        self
    }

    /// Connect to the daemon and return a [`Connection`].
    pub async fn connect(&self) -> Result<Connection> {
        Connection::connect(&self.config.socket_path).await
    }
}

/// An established Unix domain socket connection to a walrus daemon.
///
/// Not Clone — one connection per session. Use [`WalrusClient::connect`]
/// to create a connection.
pub struct Connection {
    reader: OwnedReadHalf,
    writer: OwnedWriteHalf,
}

impl Connection {
    /// Connect to a daemon at the given socket path.
    pub async fn connect(socket_path: &Path) -> Result<Self> {
        let stream = tokio::net::UnixStream::connect(socket_path).await?;
        tracing::debug!("connected to {}", socket_path.display());
        let (reader, writer) = stream.into_split();
        Ok(Self { reader, writer })
    }
}

impl Client for Connection {
    async fn request(&mut self, msg: ClientMessage) -> Result<ServerMessage> {
        codec::write_message(&mut self.writer, &msg).await?;
        Ok(codec::read_message(&mut self.reader).await?)
    }

    fn request_stream(
        &mut self,
        msg: ClientMessage,
    ) -> impl Stream<Item = Result<ServerMessage>> + Send + '_ {
        async_stream::try_stream! {
            codec::write_message(&mut self.writer, &msg).await?;

            loop {
                let server_msg: ServerMessage = codec::read_message(&mut self.reader).await?;

                match &server_msg {
                    ServerMessage { msg: Some(server_message::Msg::Error(ErrorMsg { code, message })) } => {
                        Err(anyhow::anyhow!("server error ({code}): {message}"))?;
                    }
                    _ => yield server_msg,
                }
            }
        }
    }
}