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, Scoped};
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    #[must_use]
62    pub fn product<U>(&self, stream: &Stream<I, U>) -> Stream<I, (T, U)>
63    where
64        U: Value,
65    {
66        stream
67            .into_stream_tuple_cons(self.clone())
68            .subscribe(Product { marker: PhantomData })
69    }
70}
71
72// ----------------------------------------------------------------------------
73// Trait implementations
74// ----------------------------------------------------------------------------
75
76impl<I, T, U> Action<I> for Product<T, U>
77where
78    I: Id,
79    T: Value,
80    U: Value,
81{
82    type Inputs = (T, U);
83    type Output<'a> = (T, U);
84
85    /// Executes the operator.
86    fn execute(&mut self, ctx: Context<I, Self>) -> impl IntoSteps<I, Self> {
87        let Binding { scopes, inputs, mut output, .. } = ctx.bind();
88        scopes.into_iter().flat_map(move |scope| {
89            let (left, right) = *inputs;
90            let mut scoped = vec![];
91
92            // If the key exists in the left scope,
93            if let Some(l_value) = left.get(&scope) {
94                for (r_scope, r_value) in right.iter() {
95                    let combined = scope.concat(r_scope);
96                    output.insert(
97                        combined.clone(),
98                        (l_value.clone(), r_value.clone()),
99                    );
100                    scoped.push(Scoped::from(combined).done());
101                }
102            } else {
103                for r_scope in right.keys() {
104                    let combined = scope.concat(r_scope);
105                    output.remove(&combined);
106                    scoped.push(Scoped::from(combined).done());
107                }
108            }
109
110            // If the key exists in the left scope,
111            if let Some(r_value) = right.get(&scope) {
112                // omit double-emit
113                for (l_scope, l_value) in
114                    left.iter().filter(|(k, _)| *k != &*scope)
115                {
116                    let combined = scope.concat(l_scope);
117                    output.insert(
118                        combined.clone(),
119                        (l_value.clone(), r_value.clone()),
120                    );
121                    scoped.push(Scoped::from(combined).done());
122                }
123            } else {
124                for l_scope in left.keys() {
125                    let combined = l_scope.concat(&scope);
126                    output.remove(&combined);
127                    scoped.push(Scoped::from(combined).done());
128                }
129            }
130            scoped
131        })
132    }
133}
zrx_stream/stream/operator/product.rs

zrx_stream/stream/operator/
product.rs
