zrx_stream/stream/
operator.rs

1// Copyright (c) 2025-2026 Zensical and contributors
2
3// SPDX-License-Identifier: MIT
4// Third-party contributions licensed under 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//! Stream operators.
27
28#![allow(clippy::new_without_default)]
29
30use std::marker::PhantomData;
31
32use zrx_scheduler::action::Action;
33use zrx_scheduler::schedule::Subscriber;
34use zrx_scheduler::Id;
35
36use super::Stream;
37
38mod filter;
39mod filter_map;
40mod join;
41mod map;
42mod product;
43mod select;
44
45pub use filter::Filter;
46pub use filter_map::FilterMap;
47pub use join::Join;
48pub use map::Map;
49pub use product::Product;
50pub use select::Select;
51
52// ----------------------------------------------------------------------------
53// Traits
54// ----------------------------------------------------------------------------
55
56/// Operator.
57pub trait Operator<'a, I, A>
58where
59    A: Action<I>,
60{
61    /// Subscribe the given subscriber.
62    fn subscribe<S>(&self, subscriber: S) -> Stream<I, A::Output<'a>>
63    where
64        S: Into<Subscriber<'a, I, A>>;
65}
66
67// ----------------------------------------------------------------------------
68// Trait implementations
69// ----------------------------------------------------------------------------
70
71impl<'a, I, A, T> Operator<'a, I, A> for Stream<I, T>
72where
73    I: Id,
74    A: Action<I, Inputs = (T,)> + 'static,
75{
76    /// Subscribe the given subscriber.
77    #[inline]
78    fn subscribe<S>(&self, subscriber: S) -> Stream<I, A::Output<'a>>
79    where
80        S: Into<Subscriber<'a, I, A>>,
81    {
82        // We can safely use expect here, as the stream interface prevents us
83        // from creating subscribers with invalid nodes. Otherwise, it's a bug.
84        let id = self.workflow.with(|builder| {
85            builder.add([self.id], subscriber).expect("invariant")
86        });
87        Stream {
88            id,
89            workflow: self.workflow.clone(),
90            marker: PhantomData,
91        }
92    }
93}
94
95// ----------------------------------------------------------------------------
96// Macros
97// ----------------------------------------------------------------------------
98
99/// Implements operator trait for a tuple.
100macro_rules! impl_operator_for_tuple {
101    ($T1:ident $(, $T:ident)+ $(,)?) => {
102        impl<'a, I, A, $T1, $($T,)+> Operator<'a, I, A>
103            for (Stream<I, $T1>, $(Stream<I, $T>,)+)
104        where
105            I: Id,
106            A: Action<I, Inputs = ($T1, $($T,)+)> + 'static,
107        {
108            #[inline]
109            fn subscribe<S>(&self, subscriber: S) -> Stream<I, A::Output<'a>>
110            where
111                S: Into<Subscriber<'a, I, A>>,
112            {
113                #[allow(non_snake_case)]
114                let ($T1, $($T,)*) = self;
115                // Albeit this can practically never happen, we can technically
116                // have different workflows here if we somehow alter the stream
117                // interface. Thus, this assertion is just a cautionary measure
118                // to prevent our future selves from breaking it by accident.
119                $(assert_eq!($T1.workflow, $T.workflow);)+
120                let id = $T1.workflow.with(|builder| {
121                    builder
122                        .add([$T1.id, $($T.id,)*], subscriber)
123                        .expect("invariant")
124                });
125                Stream {
126                    id,
127                    workflow: $T1.workflow.clone(),
128                    marker: PhantomData,
129                }
130            }
131        }
132    };
133}
134
135// ----------------------------------------------------------------------------
136
137impl_operator_for_tuple!(T1, T2);
138impl_operator_for_tuple!(T1, T2, T3);
139impl_operator_for_tuple!(T1, T2, T3, T4);
140impl_operator_for_tuple!(T1, T2, T3, T4, T5);
141impl_operator_for_tuple!(T1, T2, T3, T4, T5, T6);
142impl_operator_for_tuple!(T1, T2, T3, T4, T5, T6, T7);
143impl_operator_for_tuple!(T1, T2, T3, T4, T5, T6, T7, T8);
zrx_stream/stream/operator.rs

zrx_stream/stream/
operator.rs
