hydroflow_lang/graph/ops/
tee.rs

1use quote::{quote_spanned, ToTokens};
2
3use super::{
4    OperatorCategory, OperatorConstraints, OperatorWriteOutput,
5    WriteContextArgs, RANGE_0, RANGE_1, RANGE_ANY,
6};
7
8/// > 1 input stream, *n* output streams
9///
10/// Takes the input stream and delivers a copy of each item to each output.
11/// > Note: Downstream operators may need explicit type annotations.
12///
13/// ```hydroflow
14/// my_tee = source_iter(vec!["Hello", "World"]) -> tee();
15/// my_tee -> map(|x: &str| x.to_uppercase()) -> assert_eq(["HELLO", "WORLD"]);
16/// my_tee -> map(|x: &str| x.to_lowercase()) -> assert_eq(["hello", "world"]);
17/// my_tee -> assert_eq(["Hello", "World"]);
18/// ```
19pub const TEE: OperatorConstraints = OperatorConstraints {
20    name: "tee",
21    categories: &[OperatorCategory::MultiOut],
22    hard_range_inn: RANGE_1,
23    soft_range_inn: RANGE_1,
24    hard_range_out: RANGE_ANY,
25    soft_range_out: &(2..),
26    num_args: 0,
27    persistence_args: RANGE_0,
28    type_args: RANGE_0,
29    is_external_input: false,
30    has_singleton_output: false,
31    ports_inn: None,
32    ports_out: None,
33    input_delaytype_fn: |_| None,
34    write_fn: |&WriteContextArgs {
35                   root,
36                   op_span,
37                   ident,
38                   inputs,
39                   outputs,
40                   is_pull,
41                   ..
42               },
43               _| {
44        let write_iterator = if !is_pull {
45            let tees = outputs
46                .iter()
47                .rev()
48                .map(|i| i.to_token_stream())
49                .reduce(|b, a| quote_spanned! {op_span=> #root::pusherator::tee::Tee::new(#a, #b) })
50                .unwrap_or_else(
51                    || quote_spanned! {op_span=> #root::pusherator::for_each::ForEach::new(std::mem::drop) },
52                );
53            quote_spanned! {op_span=>
54                let #ident = #tees;
55            }
56        } else {
57            assert_eq!(1, inputs.len());
58            let input = &inputs[0];
59            quote_spanned! {op_span=>
60                let #ident = #input;
61            }
62        };
63        Ok(OperatorWriteOutput {
64            write_iterator,
65            ..Default::default()
66        })
67    },
68};
hydroflow_lang/graph/ops/tee.rs

hydroflow_lang/graph/ops/
tee.rs
