async_rs/traits/
reactor.rs

1//! A collection of traits to define a common interface across reactors
2
3use crate::{sys::AsSysFd, traits::AsyncToSocketAddrs};
4use futures_core::Stream;
5use futures_io::{AsyncRead, AsyncWrite};
6use std::{
7    io::{self, Read, Write},
8    net::SocketAddr,
9    ops::Deref,
10    time::{Duration, Instant},
11};
12
13/// A common interface for performing actions on a reactor
14pub trait Reactor {
15    /// The type representing a TCP stream (after tcp_connect) for this reactor
16    type TcpStream: AsyncRead + AsyncWrite + Send + Unpin + 'static;
17
18    /// A future that completes after a requested duration has elapsed (see [`Reactor::sleep`])
19    type Sleep: Future + Send + 'static;
20
21    /// Register a synchronous handle, returning an asynchronous one
22    ///
23    /// Whether the handle has to be in non-blocking mode already depends on the reactor: the
24    /// `async-io` based ones set it themselves, while the tokio one requires the caller to have
25    /// done so. Nothing checks, and getting it wrong is quiet: on a readiness notification that
26    /// does not pan out, or a write into a full send buffer, the read or write syscall blocks the
27    /// executor thread instead of reporting `WouldBlock`, stalling every task on it with no error
28    /// to go on. Set it yourself to stay portable across reactors.
29    fn register<H: Read + Write + AsSysFd + Send + 'static>(
30        &self,
31        socket: H,
32    ) -> io::Result<impl AsyncRead + AsyncWrite + Send + Unpin + 'static>
33    where
34        Self: Sized;
35
36    /// Sleep for the given duration
37    fn sleep(&self, dur: Duration) -> Self::Sleep
38    where
39        Self: Sized;
40
41    /// Stream that yields at every given interval
42    fn interval(&self, dur: Duration) -> impl Stream<Item = Instant> + Send + 'static
43    where
44        Self: Sized;
45
46    /// Create a TcpStream by connecting to a remote host
47    fn tcp_connect<A: AsyncToSocketAddrs + Send>(
48        &self,
49        addrs: A,
50    ) -> impl Future<Output = io::Result<Self::TcpStream>> + Send
51    where
52        Self: Sync + Sized,
53    {
54        async move {
55            let mut err = None;
56            for addr in addrs.to_socket_addrs().await? {
57                match self.tcp_connect_addr(addr).await {
58                    Ok(stream) => return Ok(stream),
59                    Err(e) => err = Some(e),
60                }
61            }
62            Err(err.unwrap_or_else(|| {
63                io::Error::new(io::ErrorKind::AddrNotAvailable, "couldn't resolve host")
64            }))
65        }
66    }
67
68    /// Create a TcpStream by connecting to a specific pre-resolved address
69    fn tcp_connect_addr(
70        &self,
71        addr: SocketAddr,
72    ) -> impl Future<Output = io::Result<Self::TcpStream>> + Send + 'static
73    where
74        Self: Sized;
75}
76
77impl<R: Deref> Reactor for R
78where
79    R::Target: Reactor + Sized,
80{
81    type TcpStream = <<R as Deref>::Target as Reactor>::TcpStream;
82    type Sleep = <<R as Deref>::Target as Reactor>::Sleep;
83
84    fn register<H: Read + Write + AsSysFd + Send + 'static>(
85        &self,
86        socket: H,
87    ) -> io::Result<impl AsyncRead + AsyncWrite + Send + Unpin + 'static> {
88        self.deref().register(socket)
89    }
90
91    fn sleep(&self, dur: Duration) -> Self::Sleep {
92        self.deref().sleep(dur)
93    }
94
95    fn interval(&self, dur: Duration) -> impl Stream<Item = Instant> + Send + 'static {
96        self.deref().interval(dur)
97    }
98
99    fn tcp_connect_addr(
100        &self,
101        addr: SocketAddr,
102    ) -> impl Future<Output = io::Result<Self::TcpStream>> + Send + 'static {
103        self.deref().tcp_connect_addr(addr)
104    }
105}
async_rs/traits/reactor.rs

async_rs/traits/
reactor.rs
