zrx_stream/stream/operator/
product.rs

1// Copyright (c) 2025-2026 Zensical and contributors
2
3// SPDX-License-Identifier: MIT
4// All contributions are certified under the DCO
5
6// Permission is hereby granted, free of charge, to any person obtaining a copy
7// of this software and associated documentation files (the "Software"), to
8// deal in the Software without restriction, including without limitation the
9// rights to use, copy, modify, merge, publish, distribute, sublicense, and/or
10// sell copies of the Software, and to permit persons to whom the Software is
11// furnished to do so, subject to the following conditions:
12
13// The above copyright notice and this permission notice shall be included in
14// all copies or substantial portions of the Software.
15
16// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
17// IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
18// FITNESS FOR A PARTICULAR PURPOSE AND NON-INFRINGEMENT. IN NO EVENT SHALL THE
19// AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
20// LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING
21// FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS
22// IN THE SOFTWARE.
23
24// ----------------------------------------------------------------------------
25
26//! Product operator.
27
28use std::marker::PhantomData;
29
30use zrx_scheduler::action::context::Binding;
31use zrx_scheduler::action::{Action, Context};
32use zrx_scheduler::step::{IntoSteps, Scope};
33use zrx_scheduler::{Id, Value};
34
35use crate::stream::combinator::convert::IntoStreamTupleCons;
36use crate::stream::Stream;
37
38use super::Operator;
39
40// ----------------------------------------------------------------------------
41// Structs
42// ----------------------------------------------------------------------------
43
44/// Product operator.
45pub struct Product<T, U> {
46    /// Capture types.
47    marker: PhantomData<(T, U)>,
48}
49
50// ----------------------------------------------------------------------------
51// Implementations
52// ----------------------------------------------------------------------------
53
54impl<I, T> Stream<I, T>
55where
56    I: Id,
57    T: Value,
58{
59    /// Maps the stream using the provided function.
60    #[inline]
61    pub fn product<U>(&self, stream: &Stream<I, U>) -> Stream<I, (T, U)>
62    where
63        U: Value,
64    {
65        stream
66            .into_stream_tuple_cons(self.clone())
67            .subscribe(Product { marker: PhantomData })
68    }
69}
70
71// ----------------------------------------------------------------------------
72// Trait implementations
73// ----------------------------------------------------------------------------
74
75impl<I, T, U> Action<I> for Product<T, U>
76where
77    I: Id,
78    T: Value,
79    U: Value,
80{
81    type Inputs = (T, U);
82    type Output<'a> = (T, U);
83
84    /// Executes the operator.
85    fn execute(&mut self, ctx: Context<I, Self>) -> impl IntoSteps<I, Self> {
86        let Binding { scopes, inputs, mut output, .. } = ctx.bind();
87        scopes.into_iter().flat_map(move |scope| {
88            let (left, right) = *inputs;
89            let mut scoped = vec![];
90
91            // If the key exists in the left scope,
92            if let Some(l_value) = left.get(scope.key()) {
93                for (r_scope, r_value) in right.iter() {
94                    let combined = scope.key().concat(r_scope);
95                    output.insert(
96                        combined.clone(),
97                        (l_value.clone(), r_value.clone()),
98                    );
99                    scoped.push(Scope::from(combined).done());
100                }
101            } else {
102                for r_scope in right.keys() {
103                    let combined = scope.key().concat(r_scope);
104                    output.remove(&combined);
105                    scoped.push(Scope::from(combined).done());
106                }
107            }
108
109            // If the key exists in the left scope,
110            if let Some(r_value) = right.get(scope.key()) {
111                // omit double-emit
112                for (l_scope, l_value) in
113                    left.iter().filter(|(k, _)| *k != scope.key())
114                {
115                    let combined = scope.key().concat(l_scope);
116                    output.insert(
117                        combined.clone(),
118                        (l_value.clone(), r_value.clone()),
119                    );
120                    scoped.push(Scope::from(combined).done());
121                }
122            } else {
123                for l_scope in left.keys() {
124                    let combined = l_scope.concat(scope.key());
125                    output.remove(&combined);
126                    scoped.push(Scope::from(combined).done());
127                }
128            }
129            scoped
130        })
131    }
132}
zrx_stream/stream/operator/product.rs

zrx_stream/stream/operator/
product.rs
