arcon 0.2.1

A runtime for writing streaming applications
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38

use crate::{
    data::ArconType,
    dataflow::{
        builder::OperatorBuilder,
        stream::{OperatorExt, Stream},
    },
    index::EmptyState,
    stream::operator::function,
    util::ArconFnBounds,
};
use std::sync::Arc;

/// Extension trait for filter operations
pub trait FilterExt<T: ArconType> {
    /// Filter out records based on the given predicate
    ///
    /// # Example
    /// ```rust
    /// use arcon::prelude::*;
    /// let stream: Stream<u64> = (0..100)
    ///     .to_stream(|conf| {
    ///         conf.set_arcon_time(ArconTime::Process);
    ///     })
    ///     .filter(|x| x < &50);
    /// ```
    fn filter<F: Fn(&T) -> bool + ArconFnBounds>(self, f: F) -> Self;
}

impl<T: ArconType> FilterExt<T> for Stream<T> {
    #[must_use]
    fn filter<F: Fn(&T) -> bool + ArconFnBounds>(self, f: F) -> Self {
        self.operator(OperatorBuilder {
            operator: Arc::new(move || function::Filter::new(f.clone())),
            state: Arc::new(|_| EmptyState),
            conf: Default::default(),
        })
    }
}