Struct DataStream 

pub struct DataStream { /* private fields */ }

A stream of events with chainable operators

Implementations

impl DataStream

pub fn from_events(events: Vec<StreamEvent>) -> Self

Create a new data stream from events

pub fn new() -> Self

Create an empty data stream

pub fn push(&mut self, event: StreamEvent)

Add an event to the stream

pub fn len(&self) -> usize

Get the number of events in the stream

pub fn is_empty(&self) -> bool

Check if stream is empty

pub fn filter<F>(self, predicate: F) -> Self

where F: Fn(&StreamEvent) -> bool,

Filter events based on a predicate

Example

stream.filter(|e| e.get_numeric("amount").unwrap_or(0.0) > 100.0)

pub fn map<F>(self, mapper: F) -> Self

where F: Fn(StreamEvent) -> StreamEvent,

Transform each event using a mapping function

Example

stream.map(|mut e| {
    e.add_tag("processed", "true");
    e
})

pub fn flat_map<F>(self, mapper: F) -> Self

where F: Fn(StreamEvent) -> Vec<StreamEvent>,

Transform each event into multiple events (flatMap)

Example

stream.flat_map(|e| {
    // Split event into multiple events
    vec![e.clone(), e]
})

pub fn key_by<F, K>(self, key_selector: F) -> KeyedStream<K>

where F: Fn(&StreamEvent) -> K, K: Hash + Eq + Clone,

Key events by a specific field or function

Example

stream.key_by(|e| e.get_string("user_id").unwrap_or("default").to_string())

pub fn window(self, config: WindowConfig) -> WindowedStream

Apply a window to the stream

Example

stream.window(WindowConfig::sliding(Duration::from_secs(60)))

pub fn reduce<F>(self, reducer: F) -> Option<StreamEvent>

where F: Fn(StreamEvent, StreamEvent) -> StreamEvent,

Reduce events to a single result

Example

stream.reduce(|acc, e| {
    // Accumulate values
    acc
})

pub fn count(self) -> usize

Count the number of events

pub fn collect(self) -> Vec<StreamEvent>

Collect events into a vector

pub fn take(self, n: usize) -> Self

Take only the first n events

pub fn skip(self, n: usize) -> Self

Skip the first n events

pub fn for_each<F>(self, action: F) -> Self

where F: Fn(&StreamEvent),

Process each event with a side effect (doesn’t modify the stream)

Example

stream.for_each(|e| {
    println!("Processing: {:?}", e);
})

pub fn union(self, other: DataStream) -> Self

Union with another stream

pub fn find<F>(self, predicate: F) -> Option<StreamEvent>

where F: Fn(&StreamEvent) -> bool,

Find events matching a pattern

pub fn any<F>(&self, predicate: F) -> bool

where F: Fn(&StreamEvent) -> bool,

Check if any event matches the predicate

pub fn all<F>(&self, predicate: F) -> bool

where F: Fn(&StreamEvent) -> bool,

Check if all events match the predicate

pub fn sort_by<F, K>(self, key_fn: F) -> Self

where F: Fn(&StreamEvent) -> K, K: Ord,

Sort events by a key function

pub fn group_by<F, K>(self, key_selector: F) -> GroupedStream<K>

where F: Fn(&StreamEvent) -> K, K: Hash + Eq + Clone,

Group events by a key and apply aggregation

pub fn aggregate<A>(self, aggregator: A) -> AggregateResult

where A: Aggregation,

Apply an aggregation function

Trait Implementations

impl Clone for DataStream

fn clone(&self) -> DataStream

Returns a duplicate of the value.

fn clone_from(&mut self, source: &Self)

Performs copy-assignment from source.

impl Default for DataStream

fn default() -> Self

Returns the “default value” for a type.