dewit 0.0.1

Define scheduling and execution of code separately from data flow
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
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159

use alloc::boxed::Box;
use core::pin::Pin;

use futures::future::join_all;

use crate::{
    mode::Mode,
    task::{
        Task,
        TaskOutput,
    },
};

/// Executes code as a series of futures, otherwise blocking in a single thread
/// for non-IO operations.
pub struct Async;

impl<'src> Mode<'src> for Async {
    type Output<O>
        = Pin<Box<dyn Future<Output = O> + Send + 'src>>
    where
        O: Send + 'src;

    #[inline(always)]
    fn invoke<I, O, T>(&'src self, task: T, input: I) -> TaskOutput<'src, Self, O>
    where
        T: Task<'src, I, O>,
        O: Send + 'src,
    {
        task.go(self, input)
    }

    #[inline(always)]
    fn spawn_blocking<O>(&self, f: impl FnOnce() -> O + Send + 'src) -> Self::Output<O>
    where
        O: Send + 'src,
    {
        // We immediately evaluate the function here to avoid putting blocking work into
        // an async runtime that may not be prepared for it.
        let res = f();
        Box::pin(async move { res })
    }
    #[inline(always)]
    fn spawn_cpu<O>(&self, f: impl FnOnce() -> O + Send + 'src) -> Self::Output<O>
    where
        O: Send + 'src,
    {
        // We immediately evaluate the function here to avoid putting CPU work into
        // an async runtime that may not be prepared for it.
        let res = f();
        Box::pin(async move { res })
    }
    #[inline(always)]
    fn spawn_io<O, Fut>(&self, f: impl FnOnce() -> Fut + Send + 'src) -> Self::Output<O>
    where
        O: Send + 'src,
        Fut: Future<Output = O> + Send + 'src,
    {
        Box::pin(async move { f().await })
    }

    #[inline(always)]
    fn join<AO, BO>(
        &'src self,
        a: impl FnOnce(&'src Self) -> Self::Output<AO> + Send + 'src,
        b: impl FnOnce(&'src Self) -> Self::Output<BO> + Send + 'src,
    ) -> Self::Output<(AO, BO)>
    where
        AO: Send + 'src,
        BO: Send + 'src,
    {
        // FIXME: don't depend on this nightly thing
        Box::pin(core::future::join!(a(self), b(self),))
    }

    #[inline(always)]
    fn then<I, O>(
        &'src self,
        input: Self::Output<I>,
        f: impl FnOnce(&'src Self, I) -> Self::Output<O> + Send + 'src,
    ) -> Self::Output<O>
    where
        I: Send + 'src,
        O: Send + 'src,
    {
        Box::pin(async move { f(self, input.await).await })
    }

    #[inline(always)]
    fn map_result<I, E, O>(
        &'src self,
        input: Self::Output<Result<I, E>>,
        f: impl FnOnce(&'src Self, I) -> Self::Output<O> + Send + 'src,
    ) -> Self::Output<Result<O, E>>
    where
        I: Send + 'src,
        E: Send + 'src,
        O: Send + 'src,
    {
        Box::pin(async move { Ok(f(self, input.await?).await) })
    }
    #[inline(always)]
    fn map_option<I, O>(
        &'src self,
        input: Self::Output<Option<I>>,
        f: impl FnOnce(&'src Self, I) -> Self::Output<O> + Send + 'src,
    ) -> Self::Output<Option<O>>
    where
        I: Send + 'src,
        O: Send + 'src,
    {
        Box::pin(async move { Some(f(self, input.await?).await) })
    }
    #[inline(always)]
    fn map_iterator<I, O, II>(
        &'src self,
        input: Self::Output<II>,
        f: impl Fn(&'src Self, I) -> Self::Output<O> + Send + 'src,
    ) -> Self::Output<impl Iterator<Item = O> + Send + 'src>
    where
        O: Send + 'src,
        II: Iterator<Item = I> + Send + 'src,
    {
        // FIXME: this is ugly as sin and also requires synchronization after each map,
        // ideally there'd be some magical world where we can just stream iterators into
        // one another.
        Box::pin(async move {
            join_all(input.await.map(move |input| f(self, input)))
                .await
                .into_iter()
        })
    }

    #[inline(always)]
    fn and_then_result<I, E, O>(
        &'src self,
        input: Self::Output<Result<I, E>>,
        f: impl FnOnce(&'src Self, I) -> Self::Output<Result<O, E>> + Send + 'src,
    ) -> Self::Output<Result<O, E>>
    where
        I: Send + 'src,
        E: Send + 'src,
        O: Send + 'src,
    {
        Box::pin(async move { f(self, input.await?).await })
    }
    #[inline(always)]
    fn and_then_option<I, O>(
        &'src self,
        input: Self::Output<Option<I>>,
        f: impl FnOnce(&'src Self, I) -> Self::Output<Option<O>> + Send + 'src,
    ) -> Self::Output<Option<O>>
    where
        I: Send + 'src,
        O: Send + 'src,
    {
        Box::pin(async move { f(self, input.await?).await })
    }
}