wrpc_transport/frame/
unix.rs

1//! Unix domain socket transport
2
3use std::path::{Path, PathBuf};
4
5use bytes::Bytes;
6use tokio::io::BufReader;
7use tokio::net::UnixStream;
8use tokio::net::unix::{OwnedReadHalf, OwnedWriteHalf};
9use tracing::instrument;
10
11use crate::Invoke;
12use crate::frame::{Incoming, Outgoing, invoke};
13
14/// [Invoke] implementation in terms of a single [`UnixStream`].
15///
16/// [`Invoke::invoke`] can only be called at most once on [Oneshot],
17/// repeated calls will return an error.
18pub type Oneshot = super::Oneshot<OwnedReadHalf, OwnedWriteHalf>;
19
20impl Oneshot {
21    /// Creates a pair of connected [Oneshot] using [`UnixStream::pair`].
22    pub fn unix_pair() -> std::io::Result<(Oneshot, UnixStream)> {
23        let (clt, srv) = UnixStream::pair()?;
24        Ok((clt.into(), srv))
25    }
26}
27
28impl From<UnixStream> for Oneshot {
29    fn from(stream: UnixStream) -> Self {
30        stream.into_split().into()
31    }
32}
33
34/// [Invoke] implementation of a Unix domain socket transport
35#[derive(Clone, Debug)]
36pub struct Client<T>(T);
37
38impl From<PathBuf> for Client<PathBuf> {
39    fn from(path: PathBuf) -> Self {
40        Self(path)
41    }
42}
43
44impl<'a> From<&'a Path> for Client<&'a Path> {
45    fn from(path: &'a Path) -> Self {
46        Self(path)
47    }
48}
49
50impl<'a> From<&'a std::os::unix::net::SocketAddr> for Client<&'a std::os::unix::net::SocketAddr> {
51    fn from(addr: &'a std::os::unix::net::SocketAddr) -> Self {
52        Self(addr)
53    }
54}
55
56impl From<std::os::unix::net::SocketAddr> for Client<std::os::unix::net::SocketAddr> {
57    fn from(addr: std::os::unix::net::SocketAddr) -> Self {
58        Self(addr)
59    }
60}
61
62impl Invoke for Client<PathBuf> {
63    type Context = ();
64
65    #[instrument(level = "trace", skip(self, paths, params), fields(params = format!("{params:02x?}")))]
66    async fn invoke<P>(
67        &self,
68        (): Self::Context,
69        instance: &str,
70        func: &str,
71        params: Bytes,
72        paths: impl AsRef<[P]> + Send,
73    ) -> anyhow::Result<(Outgoing, Incoming)>
74    where
75        P: AsRef<[Option<usize>]> + Send + Sync,
76    {
77        let stream = UnixStream::connect(&self.0).await?;
78        let (rx, tx) = stream.into_split();
79        invoke(tx, BufReader::new(rx), instance, func, params, paths).await
80    }
81}
82
83impl Invoke for Client<&Path> {
84    type Context = ();
85
86    #[instrument(level = "trace", skip(self, paths, params), fields(params = format!("{params:02x?}")))]
87    async fn invoke<P>(
88        &self,
89        (): Self::Context,
90        instance: &str,
91        func: &str,
92        params: Bytes,
93        paths: impl AsRef<[P]> + Send,
94    ) -> anyhow::Result<(Outgoing, Incoming)>
95    where
96        P: AsRef<[Option<usize>]> + Send + Sync,
97    {
98        let stream = UnixStream::connect(self.0).await?;
99        let (rx, tx) = stream.into_split();
100        invoke(tx, BufReader::new(rx), instance, func, params, paths).await
101    }
102}
103
104impl Invoke for Client<&std::os::unix::net::SocketAddr> {
105    type Context = ();
106
107    #[instrument(level = "trace", skip(self, paths, params), fields(params = format!("{params:02x?}")))]
108    async fn invoke<P>(
109        &self,
110        (): Self::Context,
111        instance: &str,
112        func: &str,
113        params: Bytes,
114        paths: impl AsRef<[P]> + Send,
115    ) -> anyhow::Result<(Outgoing, Incoming)>
116    where
117        P: AsRef<[Option<usize>]> + Send + Sync,
118    {
119        let stream = std::os::unix::net::UnixStream::connect_addr(self.0)?;
120        let stream = UnixStream::from_std(stream)?;
121        let (rx, tx) = stream.into_split();
122        invoke(tx, BufReader::new(rx), instance, func, params, paths).await
123    }
124}
125
126impl Invoke for Client<std::os::unix::net::SocketAddr> {
127    type Context = ();
128
129    #[instrument(level = "trace", skip(self, paths, params), fields(params = format!("{params:02x?}")))]
130    async fn invoke<P>(
131        &self,
132        (): Self::Context,
133        instance: &str,
134        func: &str,
135        params: Bytes,
136        paths: impl AsRef<[P]> + Send,
137    ) -> anyhow::Result<(Outgoing, Incoming)>
138    where
139        P: AsRef<[Option<usize>]> + Send + Sync,
140    {
141        let stream = std::os::unix::net::UnixStream::connect_addr(&self.0)?;
142        let stream = UnixStream::from_std(stream)?;
143        let (rx, tx) = stream.into_split();
144        invoke(tx, BufReader::new(rx), instance, func, params, paths).await
145    }
146}
wrpc_transport/frame/unix.rs

wrpc_transport/frame/
unix.rs
