flowly-service 0.4.13

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

use std::pin::pin;

use futures::{Stream, StreamExt};

use crate::Service;

#[derive(Clone)]
pub struct TryFlattenMap<F> {
    pub(crate) f: F,
}

impl<I, G, Ft, U, E, F> Service<Result<I, E>> for TryFlattenMap<F>
where
    G: Send,
    E: Send,
    I: Send,
    Ft: Future<Output = Result<U, E>> + Send,
    F: FnMut(I) -> Ft + Send,
    U: Stream<Item = Result<G, E>> + Send,
{
    type Out = Result<G, E>;

    fn handle(
        mut self,
        input: impl Stream<Item = Result<I, E>> + Send,
    ) -> impl Stream<Item = Result<G, E>> + Send {
        async_stream::stream! {
            let mut stream = pin!(input);

            while let Some(res) = stream.next().await {
                match res {
                    Ok(item) => {
                        match (self.f)(item).await {
                            Ok(x) => {
                                let mut x = pin!(x);
                                while let Some(res) = x.next().await {
                                    yield res;
                                }
                            }
                            Err(err) => yield Err(err)
                        }
                    }
                    Err(err) => yield Err(err)
                }
            }
        }
    }
}