1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
use quote::{ToTokens, quote_spanned};
use syn::parse_quote;
use super::{
DelayType, OpInstGenerics, OperatorCategory, OperatorConstraints, OperatorInstance,
OperatorWriteOutput, Persistence, RANGE_0, RANGE_1, WriteContextArgs,
};
use crate::graph::PortIndexValue;
/// > 2 input streams, 1 output stream, no arguments.
///
/// Operates like cross-join, but treats one of the inputs as a "singleton"-like stream, emitting
/// ignoring everything after the first element. This operator blocks on the singleton input, and
/// then joins it with all the elements in the other stream if an element is present. This operator
/// is useful when a singleton input must be used to transform elements of a stream, since unlike
/// cross-product it avoids cloning the stream of inputs. It is also useful for creating conditional
/// branches, since the operator short circuits if the singleton input produces no values.
///
/// There are two inputs to `cross_singleton`, they are `input` and `single`.
/// `input` is the input data flow, and `single` is the singleton input.
///
/// ```dfir
/// join = cross_singleton();
///
/// source_iter([1, 2, 3]) -> [input]join;
/// source_iter([0]) -> [single]join;
///
/// join -> assert_eq([(1, 0), (2, 0), (3, 0)]);
/// ```
pub const CROSS_SINGLETON: OperatorConstraints = OperatorConstraints {
name: "cross_singleton",
categories: &[OperatorCategory::MultiIn],
persistence_args: &(0..=1),
type_args: RANGE_0,
hard_range_inn: &(2..=2),
soft_range_inn: &(2..=2),
hard_range_out: RANGE_1,
soft_range_out: RANGE_1,
num_args: 0,
is_external_input: false,
flo_type: None,
ports_inn: Some(|| super::PortListSpec::Fixed(parse_quote! { input, single })),
ports_out: None,
input_delaytype_fn: |idx| match idx {
PortIndexValue::Path(path) if "single" == path.to_token_stream().to_string() => {
Some(DelayType::Stratum)
}
_else => None,
},
write_fn: |wc @ &WriteContextArgs {
root,
ident,
op_span,
inputs,
is_pull,
op_inst:
OperatorInstance {
generics: OpInstGenerics { persistence_args, .. },
..
},
..
},
_diagnostics| {
assert!(is_pull);
let persistence = match persistence_args[..] {
[] => Persistence::Tick,
[p] => p,
_ => panic!(),
};
let item_stream = &inputs[0];
let singleton_stream = &inputs[1];
let singleton_state_ident = wc.make_ident("singleton_state");
let write_prologue = quote_spanned! {op_span=>
let mut #singleton_state_ident: ::std::option::Option<_> = ::std::option::Option::None;
};
let write_tick_end = match persistence {
Persistence::Static => Default::default(),
_ => quote_spanned! {op_span=>
#singleton_state_ident = ::std::option::Option::None;
},
};
let write_iterator = quote_spanned! {op_span=>
let #ident = #root::dfir_pipes::pull::Pull::cross_singleton_state(#item_stream, #singleton_stream, &mut #singleton_state_ident);
};
Ok(OperatorWriteOutput {
write_prologue,
write_iterator,
write_tick_end,
..Default::default()
})
},
};