acktor_ipc/ipc_method/
pipe.rs

1//! IPC method implementation using Unix domain sockets and Windows named pipes.
2
3use std::io::{Error, ErrorKind};
4use std::sync::atomic::{AtomicU64, Ordering};
5
6use bytes::Bytes;
7use futures_util::{SinkExt, StreamExt};
8use interprocess::local_socket::{
9    GenericNamespaced, ListenerOptions,
10    tokio::{RecvHalf, SendHalf, prelude::*},
11};
12use tokio_util::codec::{FramedRead, FramedWrite, length_delimited::LengthDelimitedCodec};
13use tracing::info;
14
15use super::{IoFuture, IpcConnection, IpcListener};
16
17/// IPC listener implemented with Unix domain sockets and Windows named pipes.
18#[derive(Debug)]
19pub struct PipeListener {
20    name: String,
21    listener: LocalSocketListener,
22    accept_counter: AtomicU64,
23}
24
25impl PipeListener {
26    /// Constructs a new [`PipeListener`] with the given pipe name.
27    pub fn new(name: &str) -> Result<Self, Error> {
28        let name_string = name.to_string();
29        let name = name.to_ns_name::<GenericNamespaced>()?;
30
31        let opts = ListenerOptions::new().name(name);
32        let listener = opts.create_tokio()?;
33
34        Ok(Self {
35            name: name_string,
36            listener,
37            accept_counter: AtomicU64::new(0),
38        })
39    }
40}
41
42impl IpcListener for PipeListener {
43    fn local_endpoint(&self) -> &str {
44        self.name.as_str()
45    }
46
47    fn accept(&self) -> IoFuture<'_, Box<dyn IpcConnection>> {
48        Box::pin(async move {
49            let stream = self.listener.accept().await?;
50
51            let id = self.accept_counter.fetch_add(1, Ordering::Relaxed);
52            let name = format!("{}#{}", self.name, id);
53
54            info!("Accepted a new pipe connection: {}", name);
55
56            Ok(Box::new(PipeConnection::new(stream, name)) as Box<dyn IpcConnection>)
57        })
58    }
59}
60
61/// IPC connection implemented with Unix domain sockets and Windows named pipes.
62#[derive(Debug)]
63pub struct PipeConnection {
64    name: String,
65    tx: FramedWrite<SendHalf, LengthDelimitedCodec>,
66    rx: FramedRead<RecvHalf, LengthDelimitedCodec>,
67}
68
69impl PipeConnection {
70    fn new(stream: LocalSocketStream, name: String) -> Self {
71        let (rx, tx) = stream.split();
72        let codec = LengthDelimitedCodec::new();
73
74        Self {
75            name,
76            tx: FramedWrite::new(tx, codec.clone()),
77            rx: FramedRead::new(rx, codec),
78        }
79    }
80}
81
82impl IpcConnection for PipeConnection {
83    async fn connect(name: &str) -> Result<Self, Error> {
84        let stream = LocalSocketStream::connect(name.to_ns_name::<GenericNamespaced>()?).await?;
85
86        info!("Connected to pipe {}", name);
87
88        Ok(Self::new(stream, name.to_string()))
89    }
90
91    fn peer_endpoint(&self) -> &str {
92        self.name.as_str()
93    }
94
95    fn close(&mut self) -> IoFuture<'_, ()> {
96        Box::pin(async move { Ok(()) })
97    }
98
99    fn send(&mut self, buf: Bytes) -> IoFuture<'_, ()> {
100        Box::pin(self.tx.send(buf))
101    }
102
103    fn recv(&mut self) -> IoFuture<'_, Bytes> {
104        Box::pin(async move {
105            let frame = self
106                .rx
107                .next()
108                .await
109                .ok_or_else(|| Error::from(ErrorKind::ConnectionAborted))??;
110
111            Ok(frame.freeze())
112        })
113    }
114}
115
116#[cfg(test)]
117mod tests {
118    use super::*;
119
120    fn unique_pipe_name() -> String {
121        static COUNTER: AtomicU64 = AtomicU64::new(0);
122        format!(
123            "acktor-ipc-pipe-test-{}-{}",
124            std::process::id(),
125            COUNTER.fetch_add(1, Ordering::Relaxed)
126        )
127    }
128
129    #[tokio::test]
130    async fn test_listener() {
131        // new + local_endpoint
132        let name = unique_pipe_name();
133        let listener = PipeListener::new(&name).unwrap();
134        assert_eq!(listener.local_endpoint(), name);
135
136        // duplicate bind to the same name fails
137        PipeListener::new(&name).expect_err("second bind to the same name must fail");
138
139        // connect to an unbound name fails
140        let missing = unique_pipe_name();
141        PipeConnection::connect(&missing)
142            .await
143            .expect_err("connect to non-existent pipe must fail");
144    }
145
146    #[tokio::test]
147    async fn test_connection() {
148        let name = unique_pipe_name();
149        let listener = PipeListener::new(&name).unwrap();
150        let server_name = name.clone();
151
152        let server = tokio::spawn(async move {
153            // 1st client: echo roundtrip; peer_endpoint uses the `#0` suffix
154            let mut c1 = listener.accept().await.expect("accept c1");
155            assert_eq!(c1.peer_endpoint(), format!("{server_name}#0"));
156            let msg = c1.recv().await.expect("recv c1");
157            c1.send(msg).await.expect("send c1");
158            c1.close().await.expect("close c1");
159
160            // 2nd client: accept counter increments to `#1`; drop without writing to exercise
161            // the client-side ConnectionAborted path
162            let c2 = listener.accept().await.expect("accept c2");
163            assert_eq!(c2.peer_endpoint(), format!("{server_name}#1"));
164            drop(c2);
165        });
166
167        // client 1: full roundtrip
168        let mut client1 = PipeConnection::connect(&name).await.expect("connect c1");
169        assert_eq!(client1.peer_endpoint(), name);
170        let payload = Bytes::from_static(b"hello");
171        client1.send(payload.clone()).await.expect("send");
172        assert_eq!(client1.recv().await.expect("recv"), payload);
173        client1.close().await.expect("close");
174
175        // client 2: server drops its connection, recv returns ConnectionAborted
176        let mut client2 = PipeConnection::connect(&name).await.expect("connect c2");
177        server.await.unwrap();
178        let err = client2
179            .recv()
180            .await
181            .expect_err("recv should fail after peer drop");
182        assert_eq!(err.kind(), ErrorKind::ConnectionAborted);
183    }
184}
acktor_ipc/ipc_method/pipe.rs

acktor_ipc/ipc_method/
pipe.rs
