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

use crate::link::{Error, ErrorFuc, Linkable, Pipeline};
use crate::Start;
use std::future::Future;

pub struct AsyncConnect<P, F> {
    pub(crate) prev: P,
    pub(crate) next: F,
}

impl<NXT, Fut, F, L> Linkable for AsyncConnect<L, F>
where
    F: 'static + Fn(L::OUT) -> Fut + Send + Sync,
    Fut: 'static + Future<Output = NXT> + Send + Sync,
    NXT: Send + Sync,
    L: Linkable + Send + Sync,
{
    type OUT = NXT;
}

#[async_trait::async_trait]
impl<NXT, Fut, F, P> Pipeline for AsyncConnect<P, F>
where
    P: Pipeline + Sync + Send,
    F: Fn(P::OUT) -> Fut + 'static + Send + Sync,
    Fut: Future<Output = NXT> + Send + Sync + 'static,
    NXT: Send + Sync,
{
    type IN = P::IN;
    async fn process(self: &Self, input: Self::IN) -> Result<NXT, Error> {
        let out = self.prev.process(input).await?;
        Ok((self.next)(out).await)
    }
}

#[async_trait::async_trait]
impl<NXT, Fut, F, IN> Pipeline for AsyncConnect<Start<IN>, F>
where
    F: Fn(IN) -> Fut + 'static + Send + Sync,
    Fut: Future<Output = NXT> + Send + Sync + 'static,
    NXT: Send + Sync,
    IN: Send + Sync,
{
    type IN = IN;
    async fn process(self: &Self, input: IN) -> Result<NXT, Error> {
        Ok((self.next)(input).await)
    }
}

impl<NXT, Fut, F, L> Linkable for AsyncConnect<L, ErrorFuc<F>>
where
    F: Fn(L::OUT) -> Fut + Send + Sync,
    Fut: Future<Output = Result<NXT, Error>> + Send + Sync + 'static,
    NXT: Send + Sync,
    L: Linkable + Send + Sync,
{
    type OUT = NXT;
}

#[async_trait::async_trait]
impl<NXT, Fut, F, P> Pipeline for AsyncConnect<P, ErrorFuc<F>>
where
    F: Fn(P::OUT) -> Fut + Send + Sync,
    Fut: Future<Output = Result<NXT, Error>> + Send + Sync + 'static,
    NXT: Send + Sync,
    P: Pipeline + Send + Sync,
{
    type IN = P::IN;
    async fn process(self: &Self, input: Self::IN) -> Result<NXT, Error> {
        let out = self.prev.process(input).await?;
        (self.next.f)(out).await
    }
}

#[async_trait::async_trait]
impl<NXT, Fut, F, IN> Pipeline for AsyncConnect<Start<IN>, ErrorFuc<F>>
where
    IN: Sync + Send,
    F: Fn(IN) -> Fut + Sync + Send,
    Fut: Future<Output = Result<NXT, Error>> + Send + Sync + 'static,
    NXT: Send + Sync,
{
    type IN = IN;
    async fn process(self: &Self, input: Self::IN) -> Result<NXT, Error> {
        (self.next.f)(input).await
    }
}