zrx_stream/stream/
operator.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//! Stream operators.
27
28#![allow(clippy::must_use_candidate)]
29#![allow(clippy::return_self_not_must_use)]
30
31use zrx_scheduler::action::input::TryFromInputItem;
32use zrx_scheduler::action::output::IntoOutputs;
33use zrx_scheduler::action::Descriptor;
34use zrx_scheduler::effect::Signal;
35use zrx_scheduler::{Id, Value};
36
37use super::Stream;
38
39mod audit;
40mod chunks;
41mod coalesce;
42mod count;
43mod debounce;
44mod delta_count;
45mod delta_filter;
46mod delta_filter_map;
47mod delta_map;
48mod delta_reduce;
49mod difference;
50mod fill;
51mod filter;
52mod filter_map;
53mod group;
54mod inspect;
55mod intersection;
56mod join;
57mod join_filter;
58mod join_filter_map;
59mod join_map;
60mod lift;
61mod map;
62mod product;
63mod reduce;
64mod sample;
65mod select;
66mod sort;
67mod throttle;
68mod transpose;
69mod union;
70
71// ----------------------------------------------------------------------------
72// Traits
73// ----------------------------------------------------------------------------
74
75/// Operator.
76#[allow(unused_variables)]
77pub trait Operator<I, T> {
78    /// Item type handled by operator.
79    type Item<'a>: TryFromInputItem<'a, I>;
80
81    /// Handles the given item.
82    ///
83    /// This method is called by the scheduler to handle an item from a stream,
84    /// and receives a mutable reference, since operators are allowed to change
85    /// their internal state at any given time. Anything which can be converted
86    /// into [`Outputs`][] can be returned, e.g., items, signals and tasks.
87    ///
88    /// We deliberately decided to use an RPIT (return-position impl trait), as
89    /// it's the most convenient to work with, instead of requiring yet another
90    /// associated type to be defined.
91    ///
92    /// [`Outputs`]: zrx_scheduler::action::Outputs
93    #[inline]
94    fn handle(&mut self, item: Self::Item<'_>) -> impl IntoOutputs<I> {}
95
96    /// Notifies the operator of a signal.
97    ///
98    /// This method allows the operator to react to system messages or custom
99    /// events, which can be used to change its internal state or to trigger
100    /// specific actions. The default implementation just swallows the signal,
101    /// which is suitable for most cases.
102    ///
103    /// As with [`Operator::handle`], this method is allowed to return anything
104    /// that can be converted into [`Outputs`][], expecting it to emit one or
105    /// more output items, further signals, tasks or nothing at all.
106    ///
107    /// __Warning__: only implement this method if the operator is expected to
108    /// react to signals. Otherwise, the default implementation is sufficient.
109    ///
110    /// [`Outputs`]: zrx_scheduler::action::Outputs
111    #[inline]
112    fn notify(&mut self, signal: Signal<I>) -> impl IntoOutputs<I> {}
113
114    /// Returns the descriptor of the operator.
115    fn descriptor(&self) -> Descriptor;
116}
117
118// ----------------------------------------------------------------------------
119
120/// Operator extension trait.
121pub trait OperatorExt<I, T, U>
122where
123    I: Id,
124    T: Value,
125{
126    /// Applies the given operator and returns a stream.
127    fn with_operator<O>(&self, operator: O) -> Stream<I, U>
128    where
129        O: Operator<I, T> + 'static,
130        U: Value;
131}
132
133// ----------------------------------------------------------------------------
134// Implementations
135// ----------------------------------------------------------------------------
136
137impl<I, T, U> OperatorExt<I, T, U> for Stream<I, T>
138where
139    I: Id,
140    T: Value,
141{
142    /// Applies the given operator and returns a stream.
143    #[inline]
144    fn with_operator<O>(&self, operator: O) -> Stream<I, U>
145    where
146        O: Operator<I, T> + 'static,
147        U: Value,
148    {
149        self.workflow.add_operator([self.id], operator)
150    }
151}
zrx_stream/stream/operator.rs

zrx_stream/stream/
operator.rs
