zrx_stream/stream/operator/
difference.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//! Difference 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/// Difference operator.
46struct Difference<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 + Eq,
59{
60    pub fn difference<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            Difference::<T> { marker: PhantomData },
68        )
69    }
70}
71
72// ----------------------------------------------------------------------------
73
74impl<I, T> StreamSet<I, T>
75where
76    I: Id,
77    T: Value + Clone + Eq,
78{
79    pub fn into_difference(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                    Difference::<T> { marker: PhantomData },
86                )
87            })
88    }
89}
90
91// ----------------------------------------------------------------------------
92// Trait implementations
93// ----------------------------------------------------------------------------
94
95impl<I, T> Operator<I, T> for Difference<T>
96where
97    I: Id,
98    T: Value + Clone + Eq,
99{
100    type Item<'a> = Item<&'a I, Vec<Option<&'a T>>>;
101
102    /// Handles the given item.
103    ///
104    /// Differences of streams are computed by checking that each subsequent
105    /// stream does not have an exact copy of the item of the first stream.
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        let item = item.map(|data| {
112            let mut iter = data.into_iter();
113            iter.next()?.and_then(|head| {
114                iter.all(|opt| opt != Some(head)).then_some(head)
115            })
116        });
117
118        // Return item
119        item.into_owned()
120    }
121
122    /// Returns the descriptor.
123    #[inline]
124    fn descriptor(&self) -> Descriptor {
125        Descriptor::builder()
126            .property(Property::Pure)
127            .property(Property::Stable)
128            .property(Property::Flush)
129            .build()
130    }
131}
zrx_stream/stream/operator/difference.rs

zrx_stream/stream/operator/
difference.rs
