tinyflow_framework/runtime/chain_segment/
middle.rs

1use std::any::Any;
2
3use tinyflow_api::functions::{FilterFunction, FlatMapFunction, MapFunction};
4
5use super::middle_tail_step;
6
7middle_tail_step! {
8    struct MapChainSegment<T, O, F>;
9    marker: (T, O);
10    where {
11        T: Send + 'static,
12        O: Clone + Send + 'static,
13        F: MapFunction<T, O> + Clone + Send + 'static,
14    }
15    process { |this, data, runtime| {
16        let out = this.function.map(data, runtime).await?;
17        Ok(vec![Box::new(out) as Box<dyn Any + Send>])
18    }}
19    factory map_chain_factory
20}
21
22middle_tail_step! {
23    struct FilterChainSegment<T, F>;
24    marker: T;
25    where {
26        T: Clone + Send + 'static,
27        F: FilterFunction<T> + Clone + Send + 'static,
28    }
29    process { |this, data, runtime| {
30        if this.function.filter(&data, runtime).await? {
31            Ok(vec![Box::new(data) as Box<dyn Any + Send>])
32        } else {
33            Ok(vec![])
34        }
35    }}
36    factory filter_chain_factory
37}
38
39middle_tail_step! {
40    struct FlatMapChainSegment<T, O, F>;
41    marker: (T, O);
42    where {
43        T: Send + 'static,
44        O: Clone + Send + 'static,
45        F: FlatMapFunction<T, O> + Clone + Send + 'static,
46    }
47    process { |this, data, runtime| {
48        let outs = this.function.flat_map(data, runtime).await?;
49        Ok(outs
50            .into_iter()
51            .map(|d| Box::new(d) as Box<dyn Any + Send>)
52            .collect::<Vec<_>>())
53    }}
54    factory flat_map_chain_factory
55}
tinyflow_framework/runtime/chain_segment/middle.rs

tinyflow_framework/runtime/chain_segment/
middle.rs
