p2panda_stream/processors/
composed.rs

1// SPDX-License-Identifier: MIT OR Apache-2.0
2
3use crate::processors::Processor;
4
5pub struct ComposedProcessors<P1, P2> {
6    pub first: P1,
7    pub second: P2,
8}
9
10impl<P1, P2, T> Processor<T> for ComposedProcessors<P1, P2>
11where
12    P1: Processor<T>,
13    P2: Processor<P1::Output>,
14{
15    type Output = P2::Output;
16
17    type Error = ComposedError<P1::Error, P2::Error>;
18
19    async fn process(&self, input: T) -> Result<(), Self::Error> {
20        self.first
21            .process(input)
22            .await
23            .map_err(ComposedError::First)?;
24        Ok(())
25    }
26
27    async fn next(&self) -> Result<Self::Output, Self::Error> {
28        loop {
29            tokio::select! {
30                intermediate = self.first.next() => {
31                    match intermediate {
32                        Ok(intermediate) => {
33                            self.second
34                                .process(intermediate)
35                                .await
36                                .map_err(ComposedError::Second)?;
37
38                            // Yield to prevent runtime starvation.
39                            tokio::task::yield_now().await;
40                        },
41                        Err(err) => {
42                            return Err(ComposedError::First(err));
43                        },
44                    }
45                }
46
47                output = self.second.next() => {
48                    return output.map_err(ComposedError::Second);
49                }
50            }
51        }
52    }
53}
54
55#[derive(Debug, PartialEq, Eq)]
56pub enum ComposedError<P1, P2> {
57    First(P1),
58    Second(P2),
59}
p2panda_stream/processors/composed.rs

p2panda_stream/processors/
composed.rs
