async-rs 0.8.2

Async runtime abstration
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133

//! smol implementation of async runtime definition traits

use crate::{
    Runtime,
    sys::AsSysFd,
    traits::{Executor, Reactor, RuntimeKit},
    util::{IOHandle, Task},
};
use futures_core::Stream;
use futures_io::{AsyncRead, AsyncWrite};
use smol::{Async, Timer};
use std::{
    future::Future,
    io::{self, Read, Write},
    net::{SocketAddr, TcpStream},
    time::{Duration, Instant},
};

use task::STask;

/// Type alias for the smol runtime
pub type SmolRuntime = Runtime<Smol>;

impl SmolRuntime {
    /// Create a new SmolRuntime
    pub fn smol() -> Self {
        Self::new(Smol)
    }
}

/// Dummy object implementing async common interfaces on top of smol
#[derive(Debug, Default, Copy, Clone, PartialEq, Eq, PartialOrd, Ord)]
pub struct Smol;

impl RuntimeKit for Smol {}

impl Executor for Smol {
    type Task<T: Send + 'static> = STask<T>;

    fn block_on<T, F: Future<Output = T>>(&self, f: F) -> T {
        smol::block_on(f)
    }

    fn spawn<T: Send + 'static, F: Future<Output = T> + Send + 'static>(
        &self,
        f: F,
    ) -> Task<Self::Task<T>> {
        STask(Some(smol::spawn(f))).into()
    }

    fn spawn_blocking<T: Send + 'static, F: FnOnce() -> T + Send + 'static>(
        &self,
        f: F,
    ) -> Task<Self::Task<T>> {
        STask(Some(smol::unblock(f))).into()
    }
}

impl Reactor for Smol {
    type TcpStream = Async<TcpStream>;
    type Sleep = Timer;

    fn register<H: Read + Write + AsSysFd + Send + 'static>(
        &self,
        socket: H,
    ) -> io::Result<impl AsyncRead + AsyncWrite + Send + Unpin + 'static> {
        Async::new(IOHandle::new(socket))
    }

    fn sleep(&self, dur: Duration) -> Self::Sleep {
        Timer::after(dur)
    }

    fn interval(&self, dur: Duration) -> impl Stream<Item = Instant> + Send + 'static {
        Timer::interval(dur)
    }

    fn tcp_connect_addr(
        &self,
        addr: SocketAddr,
    ) -> impl Future<Output = io::Result<Self::TcpStream>> + Send + 'static {
        Async::<TcpStream>::connect(addr)
    }
}

mod task {
    use crate::util::TaskImpl;
    use async_trait::async_trait;
    use std::{
        future::Future,
        pin::Pin,
        task::{Context, Poll},
    };

    /// A smol task
    #[derive(Debug)]
    pub struct STask<T: Send + 'static>(pub(super) Option<smol::Task<T>>);

    #[async_trait]
    impl<T: Send + 'static> TaskImpl for STask<T> {
        async fn cancel(&mut self) -> Option<T> {
            self.0.take()?.cancel().await
        }

        fn detach(&mut self) {
            if let Some(task) = self.0.take() {
                task.detach();
            }
        }
    }

    impl<T: Send + 'static> Future for STask<T> {
        type Output = T;

        fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
            Pin::new(self.0.as_mut().expect("task canceled")).poll(cx)
        }
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn auto_traits() {
        use crate::util::test::*;
        let runtime = Runtime::smol();
        assert_send(&runtime);
        assert_sync(&runtime);
        assert_clone(&runtime);
    }
}