microsandbox_protocol_client/
transport.rs

1//! Exclusively owned byte transports and repeatable dialers.
2
3use std::future::Future;
4use std::path::{Path, PathBuf};
5use std::pin::Pin;
6
7use tokio::io::{AsyncRead, AsyncWrite};
8use tokio::time::Instant;
9
10use crate::{ClientError, ClientResult, ErrorKind};
11
12//--------------------------------------------------------------------------------------------------
13// Types
14//--------------------------------------------------------------------------------------------------
15
16/// Owned transport; reader and writer progress independently after splitting.
17pub trait ByteTransport: AsyncRead + AsyncWrite + Unpin + Send + 'static {}
18
19/// Erased byte transport for external protocol and connector implementations.
20pub type BoxTransport = Box<dyn ByteTransport>;
21
22/// Owned Send future used at the public protocol/connector boundary.
23pub type BoxFuture<'a, T> = Pin<Box<dyn Future<Output = T> + Send + 'a>>;
24
25/// Opens independent transports, without negotiation or application parsing.
26pub trait Connector: Send + Sync {
27    /// Dial before the remaining absolute setup deadline; dropping cancels dial.
28    fn connect(&self, deadline: Instant) -> BoxFuture<'_, ClientResult<BoxTransport>>;
29}
30
31/// Native local endpoint supplied by the caller's existing endpoint helper.
32#[derive(Debug, Clone)]
33pub struct LocalConnector {
34    path: PathBuf,
35}
36
37//--------------------------------------------------------------------------------------------------
38// Methods
39//--------------------------------------------------------------------------------------------------
40
41impl LocalConnector {
42    /// Store an endpoint without altering Unix or Windows naming.
43    pub fn new(path: impl AsRef<Path>) -> Self {
44        Self {
45            path: path.as_ref().to_owned(),
46        }
47    }
48
49    /// The exact endpoint passed by the caller.
50    pub fn path(&self) -> &Path {
51        &self.path
52    }
53}
54
55//--------------------------------------------------------------------------------------------------
56// Trait Implementations
57//--------------------------------------------------------------------------------------------------
58
59impl<T: AsyncRead + AsyncWrite + Unpin + Send + 'static> ByteTransport for T {}
60
61impl Connector for LocalConnector {
62    fn connect(&self, deadline: Instant) -> BoxFuture<'_, ClientResult<BoxTransport>> {
63        Box::pin(async move {
64            tokio::time::timeout_at(deadline, async {
65                #[cfg(all(unix, feature = "uds"))]
66                {
67                    let stream = tokio::net::UnixStream::connect(&self.path).await?;
68                    Ok(Box::new(stream) as BoxTransport)
69                }
70                #[cfg(all(windows, feature = "named-pipe"))]
71                {
72                    loop {
73                        match tokio::net::windows::named_pipe::ClientOptions::new().open(&self.path)
74                        {
75                            Ok(stream) => return Ok(Box::new(stream) as BoxTransport),
76                            Err(error)
77                                if error.kind() == std::io::ErrorKind::NotFound
78                                    || error.raw_os_error() == Some(231) =>
79                            {
80                                tokio::time::sleep(std::time::Duration::from_millis(10)).await;
81                            }
82                            Err(error) => return Err(ClientError::from(error)),
83                        }
84                    }
85                }
86                #[cfg(not(any(all(unix, feature = "uds"), all(windows, feature = "named-pipe"))))]
87                Err(ClientError::new(ErrorKind::UnsupportedOperation))
88            })
89            .await
90            .map_err(|_| ClientError::new(ErrorKind::Timeout))?
91        })
92    }
93}
microsandbox_protocol_client/transport.rs

microsandbox_protocol_client/
transport.rs
