fluxus_api/operators/
flat_map.rs

1use async_trait::async_trait;
2use fluxus_transformers::Operator;
3use fluxus_utils::models::{Record, StreamResult};
4use std::marker::PhantomData;
5
6pub struct FlatMapOperator<T, R, F, I>
7where
8    I: IntoIterator<Item = R>,
9    F: Fn(T) -> I,
10{
11    f: F,
12    _phantom: PhantomData<(T, R)>,
13}
14
15impl<T, R, F, I> FlatMapOperator<T, R, F, I>
16where
17    I: IntoIterator<Item = R>,
18    F: Fn(T) -> I,
19{
20    pub fn new(f: F) -> Self {
21        Self {
22            f,
23            _phantom: PhantomData,
24        }
25    }
26}
27
28#[async_trait]
29impl<T, R, F, I> Operator<T, R> for FlatMapOperator<T, R, F, I>
30where
31    T: Clone + Send + Sync + 'static,
32    R: Clone + Send + Sync + 'static,
33    F: Fn(T) -> I + Send + Sync,
34    I: IntoIterator<Item = R>,
35{
36    async fn process(&mut self, record: Record<T>) -> StreamResult<Vec<Record<R>>> {
37        let Record { data, timestamp } = record;
38        let result = (self.f)(data);
39        Ok(result
40            .into_iter()
41            .map(|r| Record::with_timestamp(r, timestamp))
42            .collect())
43    }
44}
fluxus_api/operators/flat_map.rs

fluxus_api/operators/
flat_map.rs
