atomr-streams 0.6.0

Typed reactive streams DSL for atomr — Source / Flow / Sink, junctions, framing, kill switches, hubs, stream refs.
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

//! Flow — a linear transformation from `In` to `Out`.
//!
//! A `Flow<A, B>` is a boxed closure that turns a `Stream<A>` into a
//! `Stream<B>`. Composition is by function chaining, which mirrors the
//! semantics of `Dsl/FlowOperations.cs` for the linear subset of
//! operators we provide.

use std::future::Future;
use std::time::Duration;

use futures::stream::{BoxStream, StreamExt};

pub struct Flow<In, Out> {
    pub(crate) transform: Box<dyn FnOnce(BoxStream<'static, In>) -> BoxStream<'static, Out> + Send + 'static>,
}

impl<T: Send + 'static> Flow<T, T> {
    pub fn identity() -> Self {
        Flow { transform: Box::new(|s| s) }
    }
}

impl<In: Send + 'static, Out: Send + 'static> Flow<In, Out> {
    /// Pure synchronous mapping./ `Select`.
    pub fn from_fn<F>(f: F) -> Self
    where
        F: FnMut(In) -> Out + Send + 'static,
    {
        Flow { transform: Box::new(move |s: BoxStream<'static, In>| s.map(f).boxed()) }
    }

    /// Asynchronous mapping with ordered bounded parallelism.
    pub fn map_async<F, Fut>(parallelism: usize, f: F) -> Self
    where
        F: FnMut(In) -> Fut + Send + 'static,
        Fut: Future<Output = Out> + Send + 'static,
    {
        let p = parallelism.max(1);
        Flow { transform: Box::new(move |s: BoxStream<'static, In>| s.map(f).buffered(p).boxed()) }
    }

    /// Chain another flow after this one.
    pub fn via<Out2: Send + 'static>(self, next: Flow<Out, Out2>) -> Flow<In, Out2> {
        Flow {
            transform: Box::new(move |s: BoxStream<'static, In>| {
                let mid = (self.transform)(s);
                (next.transform)(mid)
            }),
        }
    }

    /// Compose with a post-processing closure./ `Select`.
    pub fn then<Out2, F>(self, g: F) -> Flow<In, Out2>
    where
        Out2: Send + 'static,
        F: FnMut(Out) -> Out2 + Send + 'static,
    {
        Flow {
            transform: Box::new(move |s: BoxStream<'static, In>| {
                let out = (self.transform)(s);
                out.map(g).boxed()
            }),
        }
    }
}

impl<In: Send + 'static> Flow<In, In> {
    pub fn filter<F>(mut f: F) -> Self
    where
        F: FnMut(&In) -> bool + Send + 'static,
    {
        Flow {
            transform: Box::new(move |s: BoxStream<'static, In>| {
                s.filter(move |v| futures::future::ready(f(v))).boxed()
            }),
        }
    }

    pub fn take(n: usize) -> Self {
        Flow { transform: Box::new(move |s: BoxStream<'static, In>| s.take(n).boxed()) }
    }

    pub fn skip(n: usize) -> Self {
        Flow { transform: Box::new(move |s: BoxStream<'static, In>| s.skip(n).boxed()) }
    }

    pub fn throttle(interval: Duration) -> Self {
        Flow {
            transform: Box::new(move |s: BoxStream<'static, In>| {
                s.then(move |v| async move {
                    tokio::time::sleep(interval).await;
                    v
                })
                .boxed()
            }),
        }
    }
}

impl<In: Send + 'static, Out: Send + 'static> Flow<In, Out> {
    /// / `flatMapConcat`.
    pub fn flat_map_concat<F, S, U>(mut f: F) -> Flow<In, U>
    where
        F: FnMut(In) -> S + Send + 'static,
        S: IntoIterator<Item = U> + Send + 'static,
        S::IntoIter: Send + 'static,
        U: Send + 'static,
        // Keep Out type linked for inference; unused here.
        In: 'static,
    {
        Flow {
            transform: Box::new(move |s: BoxStream<'static, In>| {
                s.flat_map(move |x| futures::stream::iter(f(x))).boxed()
            }),
        }
    }
}