zrx_stream/stream/operator/
join_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//! Join 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::combinator::tuple::cons::IntoStreamTupleCons;
36use crate::stream::combinator::tuple::join::IntoJoinMap;
37use crate::stream::combinator::tuple::StreamTupleJoin;
38use crate::stream::function::{MapFn, Splat};
39use crate::stream::value::tuple::{All, Any, First, Presence};
40use crate::stream::value::Tuple;
41use crate::stream::Stream;
42
43use super::Operator;
44
45// ----------------------------------------------------------------------------
46// Structs
47// ----------------------------------------------------------------------------
48
49/// Join map operator.
50struct JoinMap<F, U, P> {
51    /// Operator function.
52    function: F,
53    /// Type marker.
54    marker: PhantomData<(U, P)>,
55}
56
57// ----------------------------------------------------------------------------
58// Implementations
59// ----------------------------------------------------------------------------
60
61impl<I, T> Stream<I, T>
62where
63    I: Id,
64    T: Value,
65{
66    pub fn join_map<S, O, F, U>(&self, streams: S, f: F) -> Stream<I, U>
67    where
68        S: IntoStreamTupleCons<I, T, Output = O>,
69        O: IntoJoinMap<I, All>,
70        F: MapFn<I, Splat<O::Item>, U> + Clone,
71        U: Value,
72    {
73        streams
74            .into_stream_tuple_cons(self.clone())
75            .into_join_map(f)
76    }
77
78    pub fn left_join_map<S, O, F, U>(&self, streams: S, f: F) -> Stream<I, U>
79    where
80        S: IntoStreamTupleCons<I, T, Output = O>,
81        O: IntoJoinMap<I, First>,
82        F: MapFn<I, Splat<O::Item>, U> + Clone,
83        U: Value,
84    {
85        streams
86            .into_stream_tuple_cons(self.clone())
87            .into_join_map(f)
88    }
89
90    pub fn full_join_map<S, O, F, U>(&self, streams: S, f: F) -> Stream<I, U>
91    where
92        S: IntoStreamTupleCons<I, T, Output = O>,
93        O: IntoJoinMap<I, Any>,
94        F: MapFn<I, Splat<O::Item>, U> + Clone,
95        U: Value,
96    {
97        streams
98            .into_stream_tuple_cons(self.clone())
99            .into_join_map(f)
100    }
101}
102
103// ----------------------------------------------------------------------------
104// Trait implementations
105// ----------------------------------------------------------------------------
106
107impl<I, T, F, U, P> Operator<I, T> for JoinMap<F, U, P>
108where
109    I: Id,
110    T: Tuple<P>,
111    F: MapFn<I, Splat<T>, U> + Clone,
112    U: Value,
113{
114    type Item<'a> = Item<&'a I, T::Arguments<'a>>;
115
116    /// Handles the given item.
117    ///
118    /// This operator returns a task that produces an output item by applying
119    /// the operator function to the input item. The input item is moved into
120    /// the task, and the output item is sent back to the main thread when
121    /// the worker thread finishes.
122    #[cfg_attr(
123        feature = "tracing",
124        tracing::instrument(level = "debug" skip_all, fields(id = %item.id))
125    )]
126    fn handle(&mut self, item: Self::Item<'_>) -> impl IntoOutputs<I> {
127        let item = item.into_owned().map(Splat::from);
128        Task::new({
129            let function = self.function.clone();
130            move || {
131                function.execute(&item.id, item.data).map(|report| {
132                    report.map(|data| Item::new(item.id, Some(data)))
133                })
134            }
135        })
136    }
137
138    /// Returns the descriptor.
139    #[inline]
140    fn descriptor(&self) -> Descriptor {
141        Descriptor::builder()
142            .property(Property::Pure)
143            .property(Property::Stable)
144            .property(Property::Flush)
145            .build()
146    }
147}
148
149// ----------------------------------------------------------------------------
150// Blanket implementations
151// ----------------------------------------------------------------------------
152
153impl<S, I, P> IntoJoinMap<I, P> for S
154where
155    S: StreamTupleJoin<I, P>,
156    I: Id,
157    P: Presence,
158{
159    fn into_join_map<F, U>(self, f: F) -> Stream<I, U>
160    where
161        F: MapFn<I, Splat<Self::Item>, U> + Clone,
162        U: Value,
163    {
164        self.workflow().add_operator(
165            self.ids(),
166            JoinMap {
167                function: f,
168                marker: PhantomData,
169            },
170        )
171    }
172}
zrx_stream/stream/operator/join_map.rs

zrx_stream/stream/operator/
join_map.rs
