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
}
}