zrx_stream/stream/operator/
join.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 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;
34use zrx_scheduler::{Id, Value};
35
36use crate::stream::combinator::tuple::cons::IntoStreamTupleCons;
37use crate::stream::combinator::tuple::join::IntoJoin;
38use crate::stream::combinator::tuple::StreamTupleJoin;
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 operator.
50struct Join<T, P> {
51    /// Capture types.
52    marker: PhantomData<(T, P)>,
53}
54
55// ----------------------------------------------------------------------------
56// Implementations
57// ----------------------------------------------------------------------------
58
59impl<I, T> Stream<I, T>
60where
61    I: Id,
62    T: Value,
63{
64    pub fn join<S, O>(&self, streams: S) -> Stream<I, O::Item>
65    where
66        S: IntoStreamTupleCons<I, T, Output = O>,
67        O: IntoJoin<I, All>,
68    {
69        streams // fmt
70            .into_stream_tuple_cons(self.clone())
71            .into_join()
72    }
73
74    pub fn left_join<S, O>(&self, streams: S) -> Stream<I, O::Item>
75    where
76        S: IntoStreamTupleCons<I, T, Output = O>,
77        O: IntoJoin<I, First>,
78    {
79        streams // fmt
80            .into_stream_tuple_cons(self.clone())
81            .into_join()
82    }
83
84    pub fn full_join<S, O>(&self, streams: S) -> Stream<I, O::Item>
85    where
86        S: IntoStreamTupleCons<I, T, Output = O>,
87        O: IntoJoin<I, Any>,
88    {
89        streams // fmt
90            .into_stream_tuple_cons(self.clone())
91            .into_join()
92    }
93}
94
95// ----------------------------------------------------------------------------
96// Trait implementations
97// ----------------------------------------------------------------------------
98
99impl<I, T, P> Operator<I, T> for Join<T, P>
100where
101    I: Id,
102    T: Tuple<P>,
103{
104    type Item<'a> = Item<&'a I, T::Arguments<'a>>;
105
106    /// Handles the given item.
107    ///
108    /// Joins are one of the most complex operations in stream processing, yet
109    /// the implementation here is surprisingly straightforward. The core idea
110    /// is to leverage the [`Tuple`] trait to represent the combined state of
111    /// all streams being joined.
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        item.into_owned().map(Some)
118    }
119
120    /// Returns the descriptor.
121    #[inline]
122    fn descriptor(&self) -> Descriptor {
123        Descriptor::builder()
124            .property(Property::Pure)
125            .property(Property::Stable)
126            .property(Property::Flush)
127            .build()
128    }
129}
130
131// ----------------------------------------------------------------------------
132// Blanket implementations
133// ----------------------------------------------------------------------------
134
135impl<S, I, P> IntoJoin<I, P> for S
136where
137    S: StreamTupleJoin<I, P>,
138    I: Id,
139    P: Presence,
140{
141    fn into_join(self) -> Stream<I, Self::Item> {
142        self.workflow().add_operator(
143            self.ids(),
144            Join::<Self::Item, P> { marker: PhantomData },
145        )
146    }
147}
zrx_stream/stream/operator/join.rs

zrx_stream/stream/operator/
join.rs
