async_rs/implementors/
smol.rs

1//! smol implementation of async runtime definition traits
2
3use crate::{
4    Runtime,
5    sys::AsSysFd,
6    traits::{Executor, Reactor, RuntimeKit},
7    util::{IOHandle, Task},
8};
9use futures_core::Stream;
10use futures_io::{AsyncRead, AsyncWrite};
11use smol::{Async, Timer};
12use std::{
13    future::Future,
14    io::{self, Read, Write},
15    net::{SocketAddr, TcpStream},
16    time::{Duration, Instant},
17};
18
19use task::STask;
20
21/// Type alias for the smol runtime
22pub type SmolRuntime = Runtime<Smol>;
23
24impl SmolRuntime {
25    /// Create a new SmolRuntime
26    #[must_use]
27    pub fn smol() -> Self {
28        Self::new(Smol)
29    }
30}
31
32/// The [`RuntimeKit`] implementation backed by the smol async runtime
33#[derive(Debug, Default, Copy, Clone, PartialEq, Eq, PartialOrd, Ord)]
34pub struct Smol;
35
36impl RuntimeKit for Smol {}
37
38impl Executor for Smol {
39    type Task<T: Send + 'static> = STask<T>;
40
41    fn block_on<T, F: Future<Output = T>>(&self, f: F) -> T {
42        smol::block_on(f)
43    }
44
45    fn spawn<T: Send + 'static, F: Future<Output = T> + Send + 'static>(
46        &self,
47        f: F,
48    ) -> Task<Self::Task<T>> {
49        STask(Some(smol::spawn(f))).into()
50    }
51
52    fn spawn_blocking<T: Send + 'static, F: FnOnce() -> T + Send + 'static>(
53        &self,
54        f: F,
55    ) -> Task<Self::Task<T>> {
56        STask(Some(smol::unblock(f))).into()
57    }
58}
59
60impl Reactor for Smol {
61    type TcpStream = Async<TcpStream>;
62    type Sleep = Timer;
63
64    fn register<H: Read + Write + AsSysFd + Send + 'static>(
65        &self,
66        socket: H,
67    ) -> io::Result<impl AsyncRead + AsyncWrite + Send + Unpin + 'static> {
68        Async::new(IOHandle::new(socket))
69    }
70
71    fn sleep(&self, dur: Duration) -> Self::Sleep {
72        Timer::after(dur)
73    }
74
75    fn interval(&self, dur: Duration) -> impl Stream<Item = Instant> + Send + 'static {
76        Timer::interval(dur)
77    }
78
79    fn tcp_connect_addr(
80        &self,
81        addr: SocketAddr,
82    ) -> impl Future<Output = io::Result<Self::TcpStream>> + Send + 'static {
83        async move {
84            let stream = Async::<TcpStream>::connect(addr).await?;
85            stream.get_ref().set_nodelay(true)?;
86            Ok(stream)
87        }
88    }
89}
90
91mod task {
92    use crate::util::TaskImpl;
93    use async_trait::async_trait;
94    use std::{
95        future::Future,
96        pin::Pin,
97        task::{Context, Poll},
98    };
99
100    /// A smol task
101    #[derive(Debug)]
102    pub struct STask<T: Send + 'static>(pub(super) Option<smol::Task<T>>);
103
104    #[async_trait]
105    impl<T: Send + 'static> TaskImpl for STask<T> {
106        async fn cancel(&mut self) -> Option<T> {
107            self.0.take()?.cancel().await
108        }
109
110        fn detach(&mut self) {
111            if let Some(task) = self.0.take() {
112                task.detach();
113            }
114        }
115    }
116
117    impl<T: Send + 'static> Future for STask<T> {
118        type Output = T;
119
120        fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
121            // async-task propagates a panicking task on its own; all we have to add is not
122            // stalling forever once the task has been taken away by cancel or detach.
123            let task = self
124                .0
125                .as_mut()
126                .expect("Task polled after it was canceled or completed");
127            Pin::new(task).poll(cx)
128        }
129    }
130}
131
132#[cfg(test)]
133mod tests {
134    use super::*;
135
136    #[test]
137    fn auto_traits() {
138        use crate::util::test::*;
139        let runtime = Runtime::smol();
140        assert_send(&runtime);
141        assert_sync(&runtime);
142        assert_clone(&runtime);
143    }
144}
async_rs/implementors/smol.rs

async_rs/implementors/
smol.rs
