wrpc_transport/frame/
unix.rs

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

wrpc_transport/frame/
unix.rs
