monocoque_core/
ipc.rs

1//! IPC transport via Unix domain sockets.
2//!
3//! Provides Unix domain socket support for inter-process communication
4//! with zero-copy and low-latency characteristics.
5
6#[cfg(unix)]
7use crate::rt::{UnixListener, UnixStream};
8#[cfg(unix)]
9use std::os::unix::fs::FileTypeExt;
10#[cfg(unix)]
11use std::path::Path;
12
13#[cfg(unix)]
14/// Connect to a Unix domain socket.
15///
16/// # Examples
17///
18/// ```no_run
19/// use monocoque_core::ipc;
20///
21/// # async fn example() -> std::io::Result<()> {
22/// let stream = ipc::connect("/tmp/socket.sock").await?;
23/// # Ok(())
24/// # }
25/// ```
26pub async fn connect<P: AsRef<Path>>(path: P) -> std::io::Result<UnixStream> {
27    UnixStream::connect(path).await
28}
29
30#[cfg(unix)]
31/// Bind a Unix domain socket listener.
32///
33/// Returns the listener ready to accept connections.
34///
35/// # Examples
36///
37/// ```no_run
38/// use monocoque_core::ipc;
39///
40/// # async fn example() -> std::io::Result<()> {
41/// let listener = ipc::bind("/tmp/socket.sock").await?;
42/// # Ok(())
43/// # }
44/// ```
45pub async fn bind<P: AsRef<Path>>(path: P) -> std::io::Result<UnixListener> {
46    let path_ref = path.as_ref();
47    if path_ref.exists() {
48        let metadata = std::fs::symlink_metadata(path_ref)?;
49        if metadata.file_type().is_socket() {
50            std::fs::remove_file(path_ref)?;
51        } else {
52            return Err(std::io::Error::new(
53                std::io::ErrorKind::AlreadyExists,
54                "IPC bind path exists and is not a Unix socket",
55            ));
56        }
57    }
58
59    UnixListener::bind(path).await
60}
61
62#[cfg(unix)]
63/// Accept a connection on a Unix domain socket listener.
64pub async fn accept(listener: &UnixListener) -> std::io::Result<UnixStream> {
65    let (stream, _addr) = listener.accept().await?;
66    Ok(stream)
67}
68
69#[cfg(test)]
70#[cfg(unix)]
71mod tests {
72    use super::*;
73
74    #[test]
75    fn test_ipc_connect_bind() {
76        crate::rt::LocalRuntime::new()
77            .unwrap()
78            .block_on(test_ipc_connect_bind_impl())
79    }
80
81    async fn test_ipc_connect_bind_impl() {
82        let path = "/tmp/monocoque_test_ipc.sock";
83
84        // Clean up any existing socket
85        let _ = std::fs::remove_file(path);
86
87        let listener = bind(path).await.unwrap();
88
89        // Spawn accept task
90        let accept_handle = crate::rt::spawn(async move { accept(&listener).await });
91
92        // Give listener time to start
93        crate::rt::sleep(std::time::Duration::from_millis(10)).await;
94
95        // Connect
96        let client = connect(path).await.unwrap();
97
98        // Wait for accept
99        let server = crate::rt::join(accept_handle).await.unwrap();
100
101        assert!(client.peer_addr().is_ok());
102        assert!(server.local_addr().is_ok());
103
104        // Cleanup
105        drop(client);
106        drop(server);
107        let _ = std::fs::remove_file(path);
108    }
109
110    #[test]
111    fn bind_does_not_unlink_existing_regular_file() {
112        crate::rt::LocalRuntime::new()
113            .unwrap()
114            .block_on(bind_does_not_unlink_existing_regular_file_impl())
115    }
116
117    async fn bind_does_not_unlink_existing_regular_file_impl() {
118        let path = std::env::temp_dir().join(format!(
119            "monocoque-ipc-regular-file-{}.sock",
120            std::process::id()
121        ));
122
123        std::fs::write(&path, b"do not delete").unwrap();
124
125        let result = bind(&path).await;
126
127        assert!(
128            result.is_err(),
129            "binding over an existing regular file should fail instead of unlinking it"
130        );
131        assert!(
132            path.exists(),
133            "ipc::bind unlinked an existing regular file before binding"
134        );
135
136        let _ = std::fs::remove_file(path);
137    }
138}
monocoque_core/ipc.rs

monocoque_core/
ipc.rs
