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

zrx_stream/stream/
operator.rs
