dfir_lang/graph/ops/
cross_join.rs

Help
1use quote::quote_spanned;
2use syn::parse_quote;
3
4use super::{OperatorCategory, OperatorConstraints, WriteContextArgs, RANGE_1};
5
6/// > 2 input streams of type S and T, 1 output stream of type (S, T)
7///
8/// Forms the cross-join (Cartesian product) of the items in the input streams, returning all
9/// tupled pairs.
10///
11/// ```dfir
12/// source_iter(vec!["happy", "sad"]) -> [0]my_join;
13/// source_iter(vec!["dog", "cat"]) -> [1]my_join;
14/// my_join = cross_join() -> assert_eq([("happy", "dog"), ("sad", "dog"), ("happy", "cat"), ("sad", "cat")]);
15/// ```
16///
17/// `cross_join` can be provided with one or two generic lifetime persistence arguments
18/// in the same way as [`join`](#join), see [`join`'s documentation](#join) for more info.
19///
20/// ```rustbook
21/// let (input_send, input_recv) = dfir_rs::util::unbounded_channel::<&str>();
22/// let mut flow = dfir_rs::dfir_syntax! {
23///     my_join = cross_join::<'tick>();
24///     source_iter(["hello", "bye"]) -> [0]my_join;
25///     source_stream(input_recv) -> [1]my_join;
26///     my_join -> for_each(|(s, t)| println!("({}, {})", s, t));
27/// };
28/// input_send.send("oakland").unwrap();
29/// flow.run_tick();
30/// input_send.send("san francisco").unwrap();
31/// flow.run_tick();
32/// ```
33/// Prints only `"(hello, oakland)"` and `"(bye, oakland)"`. The `source_iter` is only included in
34/// the first tick, then forgotten, so when `"san francisco"` arrives on input `[1]` in the second tick,
35/// there is nothing for it to match with from input `[0]`, and therefore it does appear in the output.
36pub const CROSS_JOIN: OperatorConstraints = OperatorConstraints {
37    name: "cross_join",
38    categories: &[OperatorCategory::MultiIn],
39    hard_range_inn: &(2..=2),
40    soft_range_inn: &(2..=2),
41    hard_range_out: RANGE_1,
42    soft_range_out: RANGE_1,
43    num_args: 0,
44    persistence_args: &(0..=2),
45    type_args: &(0..=1),
46    is_external_input: false,
47    has_singleton_output: false,
48    flo_type: None,
49    ports_inn: Some(|| super::PortListSpec::Fixed(parse_quote! { 0, 1 })),
50    ports_out: None,
51    input_delaytype_fn: |_| None,
52    write_fn: |wc @ &WriteContextArgs {
53                   op_span,
54                   ident,
55                   inputs,
56                   ..
57               },
58               diagnostics| {
59        let mut output = (super::join::JOIN.write_fn)(wc, diagnostics)?;
60
61        let lhs = &inputs[0];
62        let rhs = &inputs[1];
63        let write_iterator = output.write_iterator;
64        output.write_iterator = quote_spanned!(op_span=>
65            let #lhs = #lhs.map(|a| ((), a));
66            let #rhs = #rhs.map(|b| ((), b));
67            #write_iterator
68            let #ident = #ident.map(|((), (a, b))| (a, b));
69        );
70
71        Ok(output)
72    },
73};
dfir_lang/graph/ops/cross_join.rs

dfir_lang/graph/ops/
cross_join.rs
