Skip to main content

fluxus_api/operators/
map.rs

1use async_trait::async_trait;
2use fluxus_transformers::Operator;
3use fluxus_utils::models::{Record, StreamResult};
4use std::marker::PhantomData;
5
6pub struct MapOperator<T, R, F> {
7    f: F,
8    _phantom: PhantomData<(T, R)>,
9}
10
11impl<T, R, F> MapOperator<T, R, F>
12where
13    F: Fn(T) -> R,
14{
15    pub fn new(f: F) -> Self {
16        Self {
17            f,
18            _phantom: PhantomData,
19        }
20    }
21}
22
23#[async_trait]
24impl<T, R, F> Operator<T, R> for MapOperator<T, R, F>
25where
26    T: Clone + Send + Sync + 'static,
27    R: Clone + Send + Sync + 'static,
28    F: Fn(T) -> R + Send + Sync,
29{
30    async fn process(&mut self, record: Record<T>) -> StreamResult<Vec<Record<R>>> {
31        let result = (self.f)(record.data);
32        Ok(vec![Record::with_timestamp(result, record.timestamp)])
33    }
34}