zrx_stream/stream/operator/
inspect.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//! Inspect operator.
27
28use zrx_scheduler::action::descriptor::Property;
29use zrx_scheduler::action::output::IntoOutputs;
30use zrx_scheduler::action::Descriptor;
31use zrx_scheduler::effect::{Item, Task};
32use zrx_scheduler::{Id, Value};
33
34use crate::stream::function::InspectFn;
35use crate::stream::Stream;
36
37use super::{Operator, OperatorExt};
38
39// ----------------------------------------------------------------------------
40// Structs
41// ----------------------------------------------------------------------------
42
43/// Inspect operator.
44struct Inspect<F> {
45    /// Operator function.
46    function: F,
47}
48
49// ----------------------------------------------------------------------------
50// Implementations
51// ----------------------------------------------------------------------------
52
53impl<I, T> Stream<I, T>
54where
55    I: Id,
56    T: Value + Clone,
57{
58    pub fn inspect<F>(&self, f: F) -> Stream<I, T>
59    where
60        F: InspectFn<I, T> + Clone,
61    {
62        self.with_operator(Inspect { function: f })
63    }
64}
65
66// ----------------------------------------------------------------------------
67// Trait implementations
68// ----------------------------------------------------------------------------
69
70impl<I, T, F> Operator<I, T> for Inspect<F>
71where
72    I: Id,
73    T: Value + Clone,
74    F: InspectFn<I, T> + Clone,
75{
76    type Item<'a> = Item<&'a I, &'a T>;
77
78    /// Handles the given item.
79    ///
80    /// This operator allows to inspect each item passing through the stream
81    /// without modifying it. The provided function is executed for the given
82    /// item as part of a task, allowing for side effects such as logging or
83    /// debugging to be performed.
84    #[cfg_attr(
85        feature = "tracing",
86        tracing::instrument(level = "debug", skip_all, fields(id = %item.id))
87    )]
88    fn handle(&mut self, item: Self::Item<'_>) -> impl IntoOutputs<I> {
89        let item = item.into_owned();
90        Task::new({
91            let function = self.function.clone();
92            move || {
93                function
94                    .execute(&item.id, &item.data)
95                    .map(|report| report.map(|()| item.map(Some)))
96            }
97        })
98    }
99
100    /// Returns the descriptor.
101    #[inline]
102    fn descriptor(&self) -> Descriptor {
103        Descriptor::builder()
104            .property(Property::Pure)
105            .property(Property::Stable)
106            .property(Property::Flush)
107            .build()
108    }
109}
zrx_stream/stream/operator/inspect.rs

zrx_stream/stream/operator/
inspect.rs
