Struct StreamSet 

pub struct StreamSet<I, T> { /* private fields */ }

Stream set.

Stream sets are homogeneous collections of streams, implementing common set operations like union, intersection, and difference. They are represented as a vector of streams, in order to preserve the ordering in which they were added, which is crucial for operators like Stream::coalesce.

Operators implemented with stream sets include:

• Stream::union
• Stream::intersection
• Stream::difference
• Stream::coalesce

Note that stream sets implement set operations themselves, including union, intersection, and difference, which means they can be combined with other stream sets to implement combinatorical structural operators. However, the union set operation is likely the most common use case.

Implementations

impl<I, T> StreamSet<I, T>

pub fn union<S>(self, streams: S) -> Self

where S: IntoStreamSet<I, T>,

Creates the union of two stream sets.

Examples

use zrx_stream::combinator::IntoStreamSet;
use zrx_stream::workspace::Workspace;

// Create workspace and workflow
let workspace = Workspace::<&str>::new();
let workflow = workspace.add_workflow();

// Create streams (homogeneous)
let a = workflow.add_source::<i32>();
let b = workflow.add_source::<i32>();
let c = workflow.add_source::<i32>();

// Create union of stream sets
let set = [&a, &b].into_stream_set().union([&b, &c]);
assert_eq!(set.len(), 3);

pub fn intersection<S>(self, streams: S) -> Self

where S: IntoStreamSet<I, T>,

Creates the intersection of two stream sets.

Examples

use zrx_stream::combinator::IntoStreamSet;
use zrx_stream::workspace::Workspace;

// Create workspace and workflow
let workspace = Workspace::<&str>::new();
let workflow = workspace.add_workflow();

// Create streams (homogeneous)
let a = workflow.add_source::<i32>();
let b = workflow.add_source::<i32>();
let c = workflow.add_source::<i32>();

// Create intersection of stream sets
let set = [&a, &b].into_stream_set().intersection([&b, &c]);
assert_eq!(set.len(), 1);

pub fn difference<S>(self, streams: S) -> Self

where S: IntoStreamSet<I, T>,

Creates the difference of two stream sets.

Examples

use zrx_stream::combinator::IntoStreamSet;
use zrx_stream::workspace::Workspace;

// Create workspace and workflow
let workspace = Workspace::<&str>::new();
let workflow = workspace.add_workflow();

// Create streams (homogeneous)
let a = workflow.add_source::<i32>();
let b = workflow.add_source::<i32>();
let c = workflow.add_source::<i32>();

// Create difference of stream sets
let set = [&a, &b].into_stream_set().difference([&b, &c]);
assert_eq!(set.len(), 1);

pub fn is_subset<S>(&self, streams: S) -> bool

where S: IntoStreamSet<I, T>,

Returns whether the stream set is a subset.

Examples

use zrx_stream::combinator::IntoStreamSet;
use zrx_stream::workspace::Workspace;

// Create workspace and workflow
let workspace = Workspace::<&str>::new();
let workflow = workspace.add_workflow();

// Create streams (homogeneous)
let a = workflow.add_source::<i32>();
let b = workflow.add_source::<i32>();

// Create stream set and check for subset
let set = a.into_stream_set();
assert!(set.is_subset([&a, &b]));

pub fn is_superset<S>(&self, streams: S) -> bool

where S: IntoStreamSet<I, T>,

Returns whether the stream set is a superset.

Examples

use zrx_stream::combinator::IntoStreamSet;
use zrx_stream::workspace::Workspace;

// Create workspace and workflow
let workspace = Workspace::<&str>::new();
let workflow = workspace.add_workflow();

// Create streams (homogeneous)
let a = workflow.add_source::<i32>();
let b = workflow.add_source::<i32>();

// Create stream set and check for superset
let set = [&a, &b].into_stream_set();
assert!(set.is_superset(&a));

pub fn get(&self, index: usize) -> Option<&Stream<I, T>>

Returns a reference to the stream at the given index.

Examples

use zrx_stream::combinator::IntoStreamSet;
use zrx_stream::workspace::Workspace;

// Create workspace and workflow
let workspace = Workspace::<&str>::new();
let workflow = workspace.add_workflow();

// Create streams (homogeneous)
let a = workflow.add_source::<i32>();
let b = workflow.add_source::<i32>();

// Create stream set and obtain stream reference
let set = [&a, &b].into_stream_set();
assert_eq!(set.get(0), Some(&a));

pub fn contains(&self, stream: &Stream<I, T>) -> bool

Returns whether the stream set contains the given stream.

Examples

use zrx_stream::combinator::IntoStreamSet;
use zrx_stream::workspace::Workspace;

// Create workspace and workflow
let workspace = Workspace::<&str>::new();
let workflow = workspace.add_workflow();

// Create streams (homogeneous)
let a = workflow.add_source::<i32>();
let b = workflow.add_source::<i32>();

// Create stream set and ensure presence of stream
let set = [&a, &b].into_stream_set();
assert!(set.contains(&a));

impl<I, T> StreamSet<I, T>

pub fn len(&self) -> usize

Returns the number of streams.

pub fn is_empty(&self) -> bool

Returns whether there are any streams.

impl<I, T> StreamSet<I, T>

where I: Id, T: Value + Clone,

pub fn into_coalesce(self) -> Option<Stream<I, T>>

impl<I, T> StreamSet<I, T>

where I: Id, T: Value + Clone + Eq,

pub fn into_difference(self) -> Option<Stream<I, T>>

impl<I, T> StreamSet<I, T>

where I: Id, T: Value + Clone + Eq,

pub fn into_intersection(self) -> Option<Stream<I, T>>

impl<I, T> StreamSet<I, T>

where I: Id, T: Value + Clone + Eq,

pub fn into_union(self) -> Option<Stream<I, T>>

Trait Implementations

impl<I: Clone, T: Clone> Clone for StreamSet<I, T>

fn clone(&self) -> StreamSet<I, T>

Returns a duplicate of the value.

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

Performs copy-assignment from source.

impl<I: Debug, T: Debug> Debug for StreamSet<I, T>

fn fmt(&self, f: &mut Formatter<'_>) -> Result

Formats the value using the given formatter.

impl<I, T> FromIterator<Stream<I, T>> for StreamSet<I, T>

fn from_iter<S>(iter: S) -> Self

where S: IntoIterator<Item = Stream<I, T>>,

Creates a stream set from an iterator.

Examples

use zrx_stream::combinator::{IntoStreamSet, StreamSet};
use zrx_stream::workspace::Workspace;

// Create workspace and workflow
let workspace = Workspace::<&str>::new();
let workflow = workspace.add_workflow();

// Create streams (homogeneous)
let a = workflow.add_source::<i32>();
let b = workflow.add_source::<i32>();

// Create stream set from iterator
let set = StreamSet::from_iter([a, b]);

impl<I, T> IntoIterator for StreamSet<I, T>

fn into_iter(self) -> Self::IntoIter

Creates an iterator over the stream set.

Examples

use zrx_stream::combinator::{IntoStreamSet, StreamSet};
use zrx_stream::workspace::Workspace;

// Create workspace and workflow
let workspace = Workspace::<&str>::new();
let workflow = workspace.add_workflow();

// Create streams (homogeneous)
let a = workflow.add_source::<i32>();
let b = workflow.add_source::<i32>();

// Create and iterate over stream set
let set = StreamSet::from_iter([a, b]);
for stream in set {
    println!("{stream:?}");
}

type Item = Stream<I, T>

The type of the elements being iterated over.

type IntoIter = IntoIter<<StreamSet<I, T> as IntoIterator>::Item>

Which kind of iterator are we turning this into?