omp_rpc/
uds.rs

1//! Unix-domain-socket transport for local daemon connections.
2
3#[cfg(unix)]
4use std::os::unix::fs::{FileTypeExt, PermissionsExt};
5use std::path::Path;
6
7#[cfg(unix)]
8use hyper_util::rt::TokioIo;
9#[cfg(unix)]
10use tokio::net::{UnixListener, UnixStream};
11#[cfg(unix)]
12use tokio_stream::wrappers::UnixListenerStream;
13use tonic::transport::Channel;
14#[cfg(unix)]
15use tower::service_fn;
16
17use crate::Error;
18
19/// A stream of accepted local RPC connections.
20#[cfg(unix)]
21pub type Incoming = UnixListenerStream;
22
23/// Placeholder incoming stream on platforms without Unix-domain sockets.
24///
25/// Named-pipe support is tracked separately. Until it lands, Windows clients
26/// should use TCP on localhost with token authentication.
27#[cfg(windows)]
28#[derive(Debug)]
29pub struct Incoming;
30
31/// Bind an owner-only Unix-domain socket and return its incoming connection
32/// stream.
33///
34/// Parent directories are created as needed. An existing path is removed only
35/// when it is a socket that cannot be connected to; an active socket or a
36/// non-socket path is left untouched.
37#[cfg(unix)]
38pub async fn listen(path: &Path) -> Result<Incoming, Error> {
39	if let Some(parent) = path.parent()
40		&& !parent.as_os_str().is_empty()
41	{
42		tokio::fs::create_dir_all(parent).await?;
43	}
44
45	match tokio::fs::symlink_metadata(path).await {
46		Ok(metadata) if metadata.file_type().is_socket() => {
47			if UnixStream::connect(path).await.is_ok() {
48				return Err(
49					std::io::Error::new(
50						std::io::ErrorKind::AddrInUse,
51						"Unix socket is already accepting connections",
52					)
53					.into(),
54				);
55			}
56			tracing::debug!(socket = %path.display(), "removing stale Unix socket");
57			tokio::fs::remove_file(path).await?;
58		},
59		Ok(_) => {},
60		Err(error) if error.kind() == std::io::ErrorKind::NotFound => {},
61		Err(error) => return Err(error.into()),
62	}
63
64	let listener = UnixListener::bind(path)?;
65	tokio::fs::set_permissions(path, std::fs::Permissions::from_mode(0o600)).await?;
66	Ok(UnixListenerStream::new(listener))
67}
68
69/// Connect a tonic channel to a Unix-domain socket.
70#[cfg(unix)]
71pub async fn connect(path: &Path) -> Result<Channel, Error> {
72	let path = path.to_owned();
73	let endpoint = tonic::transport::Endpoint::from_static("http://[::]:50051");
74	let channel = endpoint
75		.connect_with_connector(service_fn(move |_| {
76			let path = path.clone();
77			async move { UnixStream::connect(path).await.map(TokioIo::new) }
78		}))
79		.await?;
80	Ok(channel)
81}
82
83/// Return an unsupported error on Windows.
84///
85/// Named-pipe support is tracked separately. Until it lands, use TCP on
86/// localhost with token authentication.
87#[cfg(windows)]
88pub async fn listen(_path: &Path) -> Result<Incoming, Error> {
89	Err(Error::Unsupported("Unix-domain sockets are unavailable on Windows"))
90}
91
92/// Return an unsupported error on Windows.
93///
94/// Named-pipe support is tracked separately. Until it lands, use TCP on
95/// localhost with token authentication.
96#[cfg(windows)]
97pub async fn connect(_path: &Path) -> Result<Channel, Error> {
98	Err(Error::Unsupported("Unix-domain sockets are unavailable on Windows"))
99}
omp_rpc/uds.rs

omp_rpc/
uds.rs
