async_rs/
runtime.rs

1use crate::{
2    sys::AsSysFd,
3    traits::{Executor, Reactor, RuntimeKit},
4    util::{SocketAddrsResolver, Task},
5};
6use futures_core::Stream;
7use futures_io::{AsyncRead, AsyncWrite};
8use std::{
9    future::Future,
10    io::{self, Read, Write},
11    net::{SocketAddr, ToSocketAddrs},
12    time::{Duration, Instant},
13};
14
15/// A full-featured async runtime that combines an executor and a reactor.
16///
17/// `Runtime<RK>` wraps any [`RuntimeKit`] and adds higher-level helpers such as
18/// domain-name resolution ([`Runtime::to_socket_addrs`]) and runtime-shutdown
19/// error detection ([`Runtime::is_runtime_shutdown_error`]).
20///
21/// Concrete type aliases — `NoopRuntime`, `SmolRuntime`, `TokioRuntime` — are
22/// provided for each built-in backend and are the usual entry points.
23#[derive(Clone, Debug)]
24pub struct Runtime<RK: RuntimeKit> {
25    kit: RK,
26}
27
28impl<RK: RuntimeKit> Runtime<RK> {
29    /// Create a new Runtime from a RuntimeKit
30    pub fn new(kit: RK) -> Self {
31        Self { kit }
32    }
33
34    /// Asynchronously resolve the given domain name
35    pub fn to_socket_addrs<A: ToSocketAddrs + Send + 'static>(
36        &self,
37        addrs: A,
38    ) -> SocketAddrsResolver<'_, RK, A>
39    where
40        <A as ToSocketAddrs>::Iter: Send + 'static,
41    {
42        SocketAddrsResolver {
43            runtime: self,
44            addrs,
45        }
46    }
47
48    /// Check if an `std::io::Error` is a runtime shutdown error
49    ///
50    /// Only tokio's shutdown error is recognised, and it is recognised by its value alone: no
51    /// runtime is consulted, so the answer does not depend on which kit backs this `Runtime` nor
52    /// on where the error came from. What this does *not* do is ask the kit about its own
53    /// shutdown errors, so a `SmolRuntime` gets `false` for anything smol-specific.
54    pub fn is_runtime_shutdown_error(&self, err: &io::Error) -> bool {
55        #[cfg(feature = "tokio")]
56        if tokio::runtime::is_rt_shutdown_err(err) {
57            return true;
58        }
59        #[cfg(not(feature = "tokio"))]
60        let _ = err;
61        false
62    }
63}
64
65impl<RK: RuntimeKit> From<RK> for Runtime<RK> {
66    fn from(kit: RK) -> Self {
67        Self::new(kit)
68    }
69}
70
71impl<RK: RuntimeKit> Executor for Runtime<RK> {
72    type Task<T: Send + 'static> = <RK as Executor>::Task<T>;
73
74    fn block_on<T, F: Future<Output = T>>(&self, f: F) -> T {
75        self.kit.block_on(f)
76    }
77
78    fn spawn<T: Send + 'static, F: Future<Output = T> + Send + 'static>(
79        &self,
80        f: F,
81    ) -> Task<Self::Task<T>> {
82        self.kit.spawn(f)
83    }
84
85    fn spawn_blocking<T: Send + 'static, F: FnOnce() -> T + Send + 'static>(
86        &self,
87        f: F,
88    ) -> Task<Self::Task<T>> {
89        self.kit.spawn_blocking(f)
90    }
91}
92
93impl<RK: RuntimeKit> Reactor for Runtime<RK> {
94    type TcpStream = <RK as Reactor>::TcpStream;
95    type Sleep = <RK as Reactor>::Sleep;
96
97    fn register<H: Read + Write + AsSysFd + Send + 'static>(
98        &self,
99        socket: H,
100    ) -> io::Result<impl AsyncRead + AsyncWrite + Send + Unpin + 'static> {
101        self.kit.register(socket)
102    }
103
104    fn sleep(&self, dur: Duration) -> Self::Sleep {
105        self.kit.sleep(dur)
106    }
107
108    fn interval(&self, dur: Duration) -> impl Stream<Item = Instant> + Send + 'static {
109        self.kit.interval(dur)
110    }
111
112    fn tcp_connect_addr(
113        &self,
114        addr: SocketAddr,
115    ) -> impl Future<Output = io::Result<Self::TcpStream>> + Send + 'static {
116        self.kit.tcp_connect_addr(addr)
117    }
118}
async_rs/runtime.rs

async_rs/
runtime.rs
