zrx_stream/stream/operator/
fill.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//! Fill operator.
27
28use zrx_scheduler::action::descriptor::Property;
29use zrx_scheduler::action::output::IntoOutputs;
30use zrx_scheduler::action::{Descriptor, Report};
31use zrx_scheduler::effect::{Item, Task};
32use zrx_scheduler::{Id, Value};
33
34use crate::stream::function::DefaultFn;
35use crate::stream::Stream;
36
37use super::{Operator, OperatorExt};
38
39// ----------------------------------------------------------------------------
40// Structs
41// ----------------------------------------------------------------------------
42
43/// Fill operator.
44struct Fill<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 fill(&self, default: T) -> Stream<I, T> {
59        self.fill_with(move || Some(default.clone()))
60    }
61
62    pub fn fill_with<F>(&self, f: F) -> Stream<I, T>
63    where
64        F: DefaultFn<I, T> + Clone,
65    {
66        self.with_operator(Fill { function: f })
67    }
68}
69
70// ----------------------------------------------------------------------------
71// Trait implementations
72// ----------------------------------------------------------------------------
73
74impl<I, T, F> Operator<I, T> for Fill<F>
75where
76    I: Id,
77    T: Value + Clone,
78    F: DefaultFn<I, T> + Clone,
79{
80    type Item<'a> = Item<&'a I, Option<&'a T>>;
81
82    /// Handles the given item.
83    ///
84    /// If the given item holds associated data, this operator just passes it
85    /// through. Otherwise, it invokes the operator function. If the function
86    /// returns a value, it is set as the associated data for the item. Note
87    /// that the default might be specific to the identifier.
88    #[cfg_attr(
89        feature = "tracing",
90        tracing::instrument(level = "debug", skip_all, fields(id = %item.id))
91    )]
92    fn handle(&mut self, item: Self::Item<'_>) -> impl IntoOutputs<I> {
93        let item = item.into_owned();
94        Task::new({
95            let function = self.function.clone();
96            move || {
97                if item.data.is_some() {
98                    Ok(Report::new(item))
99                } else {
100                    function.execute(&item.id).map(|report| {
101                        report.map(|data| Item::new(item.id, data))
102                    })
103                }
104            }
105        })
106    }
107
108    /// Returns the descriptor.
109    #[inline]
110    fn descriptor(&self) -> Descriptor {
111        Descriptor::builder()
112            .property(Property::Pure)
113            .property(Property::Stable)
114            .property(Property::Flush)
115            .build()
116    }
117}
zrx_stream/stream/operator/fill.rs

zrx_stream/stream/operator/
fill.rs
