zrx_stream/stream/operator/
coalesce.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//! Coalesce 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::{IntoStreamSet, StreamSet};
37use crate::stream::Stream;
38
39use super::Operator;
40
41// ----------------------------------------------------------------------------
42// Structs
43// ----------------------------------------------------------------------------
44
45/// Coalesce operator.
46struct Coalesce<T> {
47    /// Capture types.
48    marker: PhantomData<T>,
49}
50
51// ----------------------------------------------------------------------------
52// Implementations
53// ----------------------------------------------------------------------------
54
55impl<I, T> Stream<I, T>
56where
57    I: Id,
58    T: Value + Clone,
59{
60    pub fn coalesce<S>(&self, streams: S) -> Stream<I, T>
61    where
62        S: IntoStreamSet<I, T>,
63    {
64        let set = self.into_stream_set().union(streams);
65        self.workflow.add_operator(
66            set.into_iter().map(|stream| stream.id),
67            Coalesce::<T> { marker: PhantomData },
68        )
69    }
70}
71
72// ----------------------------------------------------------------------------
73
74impl<I, T> StreamSet<I, T>
75where
76    I: Id,
77    T: Value + Clone,
78{
79    pub fn into_coalesce(self) -> Option<Stream<I, T>> {
80        self.get(0)
81            .map(|head| head.workflow.clone())
82            .map(|workflow| {
83                workflow.add_operator(
84                    self.into_iter().map(|stream| stream.id),
85                    Coalesce::<T> { marker: PhantomData },
86                )
87            })
88    }
89}
90
91// ----------------------------------------------------------------------------
92// Trait implementations
93// ----------------------------------------------------------------------------
94
95impl<I, T> Operator<I, T> for Coalesce<T>
96where
97    I: Id,
98    T: Value + Clone,
99{
100    type Item<'a> = Item<&'a I, Vec<Option<&'a T>>>;
101
102    /// Handles the given item.
103    ///
104    /// Coalescing is implemented by checking the streams one after another for
105    /// the presence of an item, and returning the first occurrence.
106    #[cfg_attr(
107        feature = "tracing",
108        tracing::instrument(level = "debug", skip_all, fields(id = %item.id))
109    )]
110    fn handle(&mut self, item: Self::Item<'_>) -> impl IntoOutputs<I> {
111        item.map(|data| data.into_iter().find_map(|opt| opt))
112            .into_owned()
113    }
114
115    /// Returns the descriptor.
116    #[inline]
117    fn descriptor(&self) -> Descriptor {
118        Descriptor::builder()
119            .property(Property::Pure)
120            .property(Property::Stable)
121            .property(Property::Flush)
122            .build()
123    }
124}
zrx_stream/stream/operator/coalesce.rs

zrx_stream/stream/operator/
coalesce.rs
