zrx_stream/stream/operator/
map.rs

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

zrx_stream/stream/operator/
map.rs
