p2panda_stream/processors/
pipeline.rs

1// SPDX-License-Identifier: MIT OR Apache-2.0
2
3use std::marker::PhantomData;
4
5use crate::processors::Processor;
6use crate::processors::composed::ComposedProcessors;
7
8#[derive(Default)]
9pub struct PipelineBuilder<T> {
10    _marker: PhantomData<T>,
11}
12
13impl<T> PipelineBuilder<T> {
14    pub fn new() -> Self {
15        Self {
16            _marker: PhantomData,
17        }
18    }
19
20    pub fn layer<P>(self, processor: P) -> LayeredBuilder<P, T> {
21        LayeredBuilder {
22            processor,
23            _marker: PhantomData,
24        }
25    }
26}
27
28pub struct LayeredBuilder<P, T> {
29    processor: P,
30    _marker: PhantomData<T>,
31}
32
33impl<P, T> LayeredBuilder<P, T> {
34    pub fn layer<P2>(self, processor: P2) -> LayeredBuilder<ComposedProcessors<P, P2>, T>
35    where
36        P: Processor<T>,
37        P2: Processor<P::Output>,
38    {
39        LayeredBuilder {
40            processor: ComposedProcessors {
41                first: self.processor,
42                second: processor,
43            },
44            _marker: PhantomData,
45        }
46    }
47
48    pub fn build(self) -> Pipeline<P> {
49        Pipeline {
50            processor: self.processor,
51        }
52    }
53}
54
55pub struct Pipeline<P> {
56    processor: P,
57}
58
59impl<P, T> Processor<T> for Pipeline<P>
60where
61    P: Processor<T>,
62{
63    type Output = P::Output;
64
65    type Error = P::Error;
66
67    async fn process(&self, input: T) -> Result<(), Self::Error> {
68        self.processor.process(input).await
69    }
70
71    async fn next(&self) -> Result<Self::Output, Self::Error> {
72        self.processor.next().await
73    }
74}
p2panda_stream/processors/pipeline.rs

p2panda_stream/processors/
pipeline.rs
