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

zrx_stream/stream/operator/
join_map.rs
