zrx_stream/stream/operator/
map.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//! Map operator.
27
28use std::marker::PhantomData;
29
30use zrx_scheduler::action::descriptor::Property;
31use zrx_scheduler::action::output::IntoOutputs;
32use zrx_scheduler::action::Descriptor;
33use zrx_scheduler::effect::{Item, Task};
34use zrx_scheduler::{Id, Value};
35
36use crate::stream::function::MapFn;
37use crate::stream::Stream;
38
39use super::{Operator, OperatorExt};
40
41// ----------------------------------------------------------------------------
42// Structs
43// ----------------------------------------------------------------------------
44
45/// Map operator.
46struct Map<F, U> {
47    /// Operator function.
48    function: F,
49    /// Capture types.
50    marker: PhantomData<U>,
51    /// Concurrency.
52    concurrency: Option<usize>,
53}
54
55// ----------------------------------------------------------------------------
56// Implementations
57// ----------------------------------------------------------------------------
58
59impl<I, T> Stream<I, T>
60where
61    I: Id,
62    T: Value + Clone,
63{
64    pub fn map<F, U>(&self, f: F) -> Stream<I, U>
65    where
66        F: MapFn<I, T, U> + Clone,
67        U: Value,
68    {
69        self.with_operator(Map {
70            function: f,
71            marker: PhantomData,
72            concurrency: None,
73        })
74    }
75
76    // @todo temporary solution to implement task concurrency, until we've
77    // implemented task groups in zrx, so we can properly manage concurrency
78    pub fn map_concurrency<F, U>(
79        &self, f: F, concurrency: usize,
80    ) -> Stream<I, U>
81    where
82        F: MapFn<I, T, U> + Clone,
83        U: Value,
84    {
85        self.with_operator(Map {
86            function: f,
87            marker: PhantomData,
88            concurrency: Some(concurrency),
89        })
90    }
91}
92
93// ----------------------------------------------------------------------------
94// Trait implementations
95// ----------------------------------------------------------------------------
96
97impl<I, T, F, U> Operator<I, T> for Map<F, U>
98where
99    I: Id,
100    T: Value + Clone,
101    F: MapFn<I, T, U> + Clone,
102    U: Value,
103{
104    type Item<'a> = Item<&'a I, &'a T>;
105
106    /// Handles the given item.
107    ///
108    /// This operator returns a task that produces an output item by applying
109    /// the operator function to the input item. The input item is moved into
110    /// the task, and the output item is sent back to the main thread when
111    /// the worker thread finishes.
112    #[cfg_attr(
113        feature = "tracing",
114        tracing::instrument(level = "debug", skip_all, fields(id = %item.id))
115    )]
116    fn handle(&mut self, item: Self::Item<'_>) -> impl IntoOutputs<I> {
117        let item = item.into_owned();
118        Task::new({
119            let function = self.function.clone();
120            move || {
121                function.execute(&item.id, item.data).map(|report| {
122                    report.map(|data| Item::new(item.id, Some(data)))
123                })
124            }
125        })
126    }
127
128    /// Returns the descriptor.
129    #[inline]
130    fn descriptor(&self) -> Descriptor {
131        let mut builder = Descriptor::builder()
132            .property(Property::Pure)
133            .property(Property::Stable)
134            .property(Property::Flush);
135
136        // Limit concurrency, if set
137        if let Some(concurrency) = self.concurrency {
138            builder = builder.property(Property::Concurrency(concurrency));
139        }
140        builder.build()
141    }
142}
zrx_stream/stream/operator/map.rs

zrx_stream/stream/operator/
map.rs
