flowly-service 0.6.0

Flowly is a library of modular and reusable components for building robust pipelines processing audio, video and other.
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

use std::marker::PhantomData;

use futures::{FutureExt, Stream, StreamExt, TryStreamExt};

use crate::Service;

pub fn scope<I, M, E, S, F>(f: F, service: S) -> Scope<I, M, E, S, F> {
    Scope {
        service,
        f,
        _m: PhantomData,
    }
}

pub fn scope_each<I: Clone, M, S, F, E>(f: F, service: S) -> ScopeEach<I, M, S, F, E> {
    ScopeEach {
        service,
        f,
        _m: PhantomData,
    }
}

#[derive(Clone)]
pub struct Scope<I, M, E, S, F> {
    service: S,
    f: F,
    _m: PhantomData<(I, M, E)>,
}

impl<I, M, E1, O, E, S, F> Service<I> for Scope<I, M, E1, S, F>
where
    S: Service<M, Out = Result<O, E>> + Send,
    F: Send + Fn(&I) -> Result<M, E1> + Sync,
    M: Send + Sync,
    O: Send,
    I: Send + Sync,
    E: Send + Sync + From<E1>,
    E1: Send + Sync,
{
    type Out = Result<(I, Vec<O>), E>;

    fn handle(&self, msg: I, cx: &crate::Context) -> impl Stream<Item = Self::Out> + Send {
        async move {
            match (self.f)(&msg) {
                Ok(m) => self
                    .service
                    .handle(m, cx)
                    .try_collect()
                    .await
                    .map(move |dat| (msg, dat)),
                Err(err) => Err(E::from(err)),
            }
        }
        .into_stream()
    }
}

#[derive(Clone)]
pub struct ScopeEach<I, M, S, F, E> {
    service: S,
    f: F,
    _m: PhantomData<(I, M, E)>,
}

impl<I, M, O, E, E1, S, F> Service<I> for ScopeEach<I, M, S, F, E1>
where
    S: Service<M, Out = Result<O, E>> + Send,
    F: Send + Sync + Fn(&I) -> Result<M, E1>,
    M: Send + Sync,
    O: Send,
    I: Send + Clone + Sync + 'static,
    E: Send + Sync + From<E1>,
    E1: Send + Sync,
{
    type Out = Result<(I, O), E>;

    fn handle(&self, msg: I, cx: &crate::Context) -> impl Stream<Item = Self::Out> + Send {
        async move {
            match (self.f)(&msg) {
                Ok(m) => Ok(self
                    .service
                    .handle(m, cx)
                    .map(move |x| x.map(|x| (msg.clone(), x)))),
                Err(err) => Err(E::from(err)),
            }
        }
        .into_stream()
        .try_flatten()
    }
}