Struct Stream

pub struct Stream<S, D>
where
    S: Scope,
{ /* private fields */ }

Abstraction of a stream of D: Data records timestamped with S::Timestamp.

Internally Stream maintains a list of data recipients who should be presented with data produced by the source of the stream.

Implementations

impl<S, D> Stream<S, D>

where S: Scope,

pub fn connect_to<P>(&self, target: Target, pusher: P, identifier: usize)

where P: Push<Message<Message<<S as ScopeParent>::Timestamp, D>>> + 'static,

Connects the stream to a destination.

The destination is described both by a Target, for progress tracking information, and a P: Push where the records should actually be sent. The identifier is unique to the edge and is used only for logging purposes.

pub fn new( source: Source, output: TeeHelper<<S as ScopeParent>::Timestamp, D>, scope: S, ) -> Stream<S, D>

Allocates a Stream from a supplied Source name and rendezvous point.

pub fn name(&self) -> &Source

The name of the stream’s source operator.

pub fn scope(&self) -> S

The scope immediately containing the stream.

Trait Implementations

impl<G, D> Accumulate<G, D> for Stream<G, D>

where G: Scope, D: Data,

fn accumulate<A>( &self, default: A, logic: impl Fn(&mut A, RefOrMut<'_, Vec<D>>) + 'static, ) -> Stream<G, A>

where A: Data,

Accumulates records within a timestamp.

fn count(&self) -> Stream<G, usize>

Counts the number of records observed at each time.

impl<S, K, V> Aggregate<S, K, V> for Stream<S, (K, V)>

where S: Scope, K: ExchangeData + Hash + Eq, V: ExchangeData,

fn aggregate<R, D, F, E, H>(&self, fold: F, emit: E, hash: H) -> Stream<S, R>

where R: Data, D: Default + 'static, F: Fn(&K, V, &mut D) + 'static, E: Fn(K, D) -> R + 'static, H: Fn(&K) -> u64 + 'static, <S as ScopeParent>::Timestamp: Eq,

Aggregates data of the form (key, val), using user-supplied logic.

impl<S, D> Branch<S, D> for Stream<S, D>

where S: Scope, D: Data,

fn branch( &self, condition: impl Fn(&<S as ScopeParent>::Timestamp, &D) -> bool + 'static, ) -> (Stream<S, D>, Stream<S, D>)

Takes one input stream and splits it into two output streams. For each record, the supplied closure is called with a reference to the data and its time. If it returns true, the record will be sent to the second returned stream, otherwise it will be sent to the first.

impl<S, D> BranchWhen<S, D> for Stream<S, D>

where S: Scope, D: Data,

fn branch_when( &self, condition: impl Fn(&<S as ScopeParent>::Timestamp) -> bool + 'static, ) -> (Stream<S, D>, Stream<S, D>)

Takes one input stream and splits it into two output streams. For each time, the supplied closure is called. If it returns true, the records for that will be sent to the second returned stream, otherwise they will be sent to the first.

impl<G, D> Broadcast<D> for Stream<G, D>

where G: Scope, D: ExchangeData,

fn broadcast(&self) -> Stream<G, D>

Broadcast records to all workers.

impl<S, D> Capture<<S as ScopeParent>::Timestamp, D> for Stream<S, D>

where S: Scope, D: Data,

fn capture_into<P>(&self, event_pusher: P)

where P: EventPusher<<S as ScopeParent>::Timestamp, D> + 'static,

Captures a stream of timestamped data for later replay.

fn capture(&self) -> Receiver<Event<T, D>>

Captures a stream using Rust’s MPSC channels.

impl<S, D> Clone for Stream<S, D>

where S: Clone + Scope, D: Clone, <S as ScopeParent>::Timestamp: Clone,

fn clone(&self) -> Stream<S, D>

Returns a duplicate of the value.

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

Performs copy-assignment from source.

impl<G, D> Concat<G, D> for Stream<G, D>

where G: Scope, D: Data,

fn concat(&self, other: &Stream<G, D>) -> Stream<G, D>

Merge the contents of two streams.

impl<G, D> Concatenate<G, D> for Stream<G, D>

where G: Scope, D: Data,

fn concatenate<I>(&self, sources: I) -> Stream<G, D>

where I: IntoIterator<Item = Stream<G, D>>,

Merge the contents of multiple streams.

impl<G, D> ConnectLoop<G, D> for Stream<G, D>

where G: Scope, D: Data,

fn connect_loop(&self, helper: Handle<G, D>)

Connect a Stream to be the input of a loop variable.

impl<G, D> Delay<G, D> for Stream<G, D>

where G: Scope, D: Data,

fn delay<L>(&self, func: L) -> Stream<G, D>

where L: FnMut(&D, &<G as ScopeParent>::Timestamp) -> <G as ScopeParent>::Timestamp + 'static,

Advances the timestamp of records using a supplied function.

fn delay_total<L>(&self, func: L) -> Stream<G, D>

where L: FnMut(&D, &<G as ScopeParent>::Timestamp) -> <G as ScopeParent>::Timestamp + 'static, <G as ScopeParent>::Timestamp: TotalOrder,

Advances the timestamp of records using a supplied function.

fn delay_batch<L>(&self, func: L) -> Stream<G, D>

where L: FnMut(&<G as ScopeParent>::Timestamp) -> <G as ScopeParent>::Timestamp + 'static,

Advances the timestamp of batches of records using a supplied function.

impl<G, T, D> Enter<G, T, D> for Stream<G, D>

where G: Scope, T: Timestamp + Refines<<G as ScopeParent>::Timestamp>, D: Data,

fn enter<'a>(&self, scope: &Child<'a, G, T>) -> Stream<Child<'a, G, T>, D>

Moves the Stream argument into a child of its current Scope.

impl<G, D> Exchange<<G as ScopeParent>::Timestamp, D> for Stream<G, D>

where G: Scope, D: ExchangeData,

fn exchange(&self, route: impl Fn(&D) -> u64 + 'static) -> Stream<G, D>

Exchange records between workers.

impl<G, D> Filter<D> for Stream<G, D>

where G: Scope, D: Data,

fn filter<P>(&self, predicate: P) -> Stream<G, D>

where P: FnMut(&D) -> bool + 'static,

Returns a new instance of self containing only records satisfying predicate.

impl<G, D> Inspect<G, D> for Stream<G, D>

where G: Scope, D: Data,

fn inspect_batch( &self, func: impl FnMut(&<G as ScopeParent>::Timestamp, &[D]) + 'static, ) -> Stream<G, D>

Runs a supplied closure on each observed data batch (time and data slice).

fn inspect(&self, func: impl FnMut(&D) + 'static) -> Stream<G, D>

Runs a supplied closure on each observed data element.

fn inspect_time( &self, func: impl FnMut(&<G as ScopeParent>::Timestamp, &D) + 'static, ) -> Stream<G, D>

Runs a supplied closure on each observed data element and associated time.

impl<'a, G, D, T> Leave<G, D> for Stream<Child<'a, G, T>, D>

where G: Scope, D: Data, T: Timestamp + Refines<<G as ScopeParent>::Timestamp>,

fn leave(&self) -> Stream<G, D>

Moves a Stream to the parent of its current Scope.

impl<S, D> Map<S, D> for Stream<S, D>

where S: Scope, D: Data,

fn map<D2, L>(&self, logic: L) -> Stream<S, D2>

where D2: Data, L: FnMut(D) -> D2 + 'static,

Consumes each element of the stream and yields a new element.

fn map_in_place<L>(&self, logic: L) -> Stream<S, D>

where L: FnMut(&mut D) + 'static,

Updates each element of the stream and yields the element, re-using memory where possible.

fn flat_map<I, L>(&self, logic: L) -> Stream<S, <I as IntoIterator>::Item>

where I: IntoIterator, L: FnMut(D) -> I + 'static, <I as IntoIterator>::Item: Data,

Consumes each element of the stream and yields some number of new elements.

impl<S, D> OkErr<S, D> for Stream<S, D>

where S: Scope, D: Data,

fn ok_err<D1, D2, L>(&self, logic: L) -> (Stream<S, D1>, Stream<S, D2>)

where D1: Data, D2: Data, L: FnMut(D) -> Result<D1, D2> + 'static,

Takes one input stream and splits it into two output streams. For each record, the supplied closure is called with the data. If it returns Ok(x), then x will be sent to the first returned stream; otherwise, if it returns Err(e), then e will be sent to the second.

impl<G, D1> Operator<G, D1> for Stream<G, D1>

where G: Scope, D1: Data,

fn unary_frontier<D2, B, L, P>( &self, pact: P, name: &str, constructor: B, ) -> Stream<G, D2>

where D2: Data, B: FnOnce(Capability<<G as ScopeParent>::Timestamp>, OperatorInfo) -> L, L: FnMut(&mut FrontieredInputHandle<'_, <G as ScopeParent>::Timestamp, D1, <P as ParallelizationContract<<G as ScopeParent>::Timestamp, D1>>::Puller>, &mut OutputHandle<'_, <G as ScopeParent>::Timestamp, D2, Tee<<G as ScopeParent>::Timestamp, D2>>) + 'static, P: ParallelizationContract<<G as ScopeParent>::Timestamp, D1>,

Creates a new dataflow operator that partitions its input stream by a parallelization strategy pact, and repeatedly invokes logic, the function returned by the function passed as constructor. logic can read from the input stream, write to the output stream, and inspect the frontier at the input.

fn unary_notify<D2, L, P>( &self, pact: P, name: &str, init: impl IntoIterator<Item = <G as ScopeParent>::Timestamp>, logic: L, ) -> Stream<G, D2>

where D2: Data, L: FnMut(&mut InputHandle<<G as ScopeParent>::Timestamp, D1, <P as ParallelizationContract<<G as ScopeParent>::Timestamp, D1>>::Puller>, &mut OutputHandle<'_, <G as ScopeParent>::Timestamp, D2, Tee<<G as ScopeParent>::Timestamp, D2>>, &mut Notificator<'_, <G as ScopeParent>::Timestamp>) + 'static, P: ParallelizationContract<<G as ScopeParent>::Timestamp, D1>,

Creates a new dataflow operator that partitions its input stream by a parallelization strategy pact, and repeatedly invokes logic, the function returned by the function passed as constructor. logic can read from the input stream, write to the output stream, and inspect the frontier at the input.

fn unary<D2, B, L, P>( &self, pact: P, name: &str, constructor: B, ) -> Stream<G, D2>

where D2: Data, B: FnOnce(Capability<<G as ScopeParent>::Timestamp>, OperatorInfo) -> L, L: FnMut(&mut InputHandle<<G as ScopeParent>::Timestamp, D1, <P as ParallelizationContract<<G as ScopeParent>::Timestamp, D1>>::Puller>, &mut OutputHandle<'_, <G as ScopeParent>::Timestamp, D2, Tee<<G as ScopeParent>::Timestamp, D2>>) + 'static, P: ParallelizationContract<<G as ScopeParent>::Timestamp, D1>,

Creates a new dataflow operator that partitions its input stream by a parallelization strategy pact, and repeatedly invokes logic, the function returned by the function passed as constructor. logic can read from the input stream, and write to the output stream.

fn binary_frontier<D2, D3, B, L, P1, P2>( &self, other: &Stream<G, D2>, pact1: P1, pact2: P2, name: &str, constructor: B, ) -> Stream<G, D3>

where D2: Data, D3: Data, B: FnOnce(Capability<<G as ScopeParent>::Timestamp>, OperatorInfo) -> L, L: FnMut(&mut FrontieredInputHandle<'_, <G as ScopeParent>::Timestamp, D1, <P1 as ParallelizationContract<<G as ScopeParent>::Timestamp, D1>>::Puller>, &mut FrontieredInputHandle<'_, <G as ScopeParent>::Timestamp, D2, <P2 as ParallelizationContract<<G as ScopeParent>::Timestamp, D2>>::Puller>, &mut OutputHandle<'_, <G as ScopeParent>::Timestamp, D3, Tee<<G as ScopeParent>::Timestamp, D3>>) + 'static, P1: ParallelizationContract<<G as ScopeParent>::Timestamp, D1>, P2: ParallelizationContract<<G as ScopeParent>::Timestamp, D2>,

Creates a new dataflow operator that partitions its input streams by a parallelization strategy pact, and repeatedly invokes logic, the function returned by the function passed as constructor. logic can read from the input streams, write to the output stream, and inspect the frontier at the inputs.

fn binary_notify<D2, D3, L, P1, P2>( &self, other: &Stream<G, D2>, pact1: P1, pact2: P2, name: &str, init: impl IntoIterator<Item = <G as ScopeParent>::Timestamp>, logic: L, ) -> Stream<G, D3>

where D2: Data, D3: Data, L: FnMut(&mut InputHandle<<G as ScopeParent>::Timestamp, D1, <P1 as ParallelizationContract<<G as ScopeParent>::Timestamp, D1>>::Puller>, &mut InputHandle<<G as ScopeParent>::Timestamp, D2, <P2 as ParallelizationContract<<G as ScopeParent>::Timestamp, D2>>::Puller>, &mut OutputHandle<'_, <G as ScopeParent>::Timestamp, D3, Tee<<G as ScopeParent>::Timestamp, D3>>, &mut Notificator<'_, <G as ScopeParent>::Timestamp>) + 'static, P1: ParallelizationContract<<G as ScopeParent>::Timestamp, D1>, P2: ParallelizationContract<<G as ScopeParent>::Timestamp, D2>,

Creates a new dataflow operator that partitions its input streams by a parallelization strategy pact, and repeatedly invokes logic, the function returned by the function passed as constructor. logic can read from the input streams, write to the output stream, and inspect the frontier at the inputs.

fn binary<D2, D3, B, L, P1, P2>( &self, other: &Stream<G, D2>, pact1: P1, pact2: P2, name: &str, constructor: B, ) -> Stream<G, D3>

where D2: Data, D3: Data, B: FnOnce(Capability<<G as ScopeParent>::Timestamp>, OperatorInfo) -> L, L: FnMut(&mut InputHandle<<G as ScopeParent>::Timestamp, D1, <P1 as ParallelizationContract<<G as ScopeParent>::Timestamp, D1>>::Puller>, &mut InputHandle<<G as ScopeParent>::Timestamp, D2, <P2 as ParallelizationContract<<G as ScopeParent>::Timestamp, D2>>::Puller>, &mut OutputHandle<'_, <G as ScopeParent>::Timestamp, D3, Tee<<G as ScopeParent>::Timestamp, D3>>) + 'static, P1: ParallelizationContract<<G as ScopeParent>::Timestamp, D1>, P2: ParallelizationContract<<G as ScopeParent>::Timestamp, D2>,

Creates a new dataflow operator that partitions its input streams by a parallelization strategy pact, and repeatedly invokes logic, the function returned by the function passed as constructor. logic can read from the input streams, write to the output stream, and inspect the frontier at the inputs.

fn sink<L, P>(&self, pact: P, name: &str, logic: L)

where L: FnMut(&mut FrontieredInputHandle<'_, <G as ScopeParent>::Timestamp, D1, <P as ParallelizationContract<<G as ScopeParent>::Timestamp, D1>>::Puller>) + 'static, P: ParallelizationContract<<G as ScopeParent>::Timestamp, D1>,

Creates a new dataflow operator that partitions its input stream by a parallelization strategy pact, and repeatedly invokes the function logic which can read from the input stream and inspect the frontier at the input.

impl<G, D, D2, F> Partition<G, D, D2, F> for Stream<G, D>

where G: Scope, D: Data, D2: Data, F: Fn(D) -> (u64, D2) + 'static,

fn partition(&self, parts: u64, route: F) -> Vec<Stream<G, D2>>

Produces parts output streams, containing records produced and assigned by route.

impl<G, D> Probe<G, D> for Stream<G, D>

where G: Scope, D: Data,

fn probe(&self) -> Handle<<G as ScopeParent>::Timestamp>

Constructs a progress probe which indicates which timestamps have elapsed at the operator.

fn probe_with( &self, handle: &mut Handle<<G as ScopeParent>::Timestamp>, ) -> Stream<G, D>

Inserts a progress probe in a stream.

impl<S, D> Reclock<S, D> for Stream<S, D>

where S: Scope, D: Data,

fn reclock(&self, clock: &Stream<S, ()>) -> Stream<S, D>

Delays records until an input is observed on the clock input.

impl<S, T, E> ResultStream<S, T, E> for Stream<S, Result<T, E>>

where S: Scope, T: Data, E: Data,

fn ok(&self) -> Stream<S, T>

Returns a new instance of self containing only ok records.

fn err(&self) -> Stream<S, E>

Returns a new instance of self containing only err records.

fn map_ok<T2, L>(&self, logic: L) -> Stream<S, Result<T2, E>>

where T2: Data, L: FnMut(T) -> T2 + 'static,

Returns a new instance of self applying logic on all Ok records.

fn map_err<E2, L>(&self, logic: L) -> Stream<S, Result<T, E2>>

where E2: Data, L: FnMut(E) -> E2 + 'static,

Returns a new instance of self applying logic on all Err records.

fn and_then<T2, L>(&self, logic: L) -> Stream<S, Result<T2, E>>

where T2: Data, L: FnMut(T) -> Result<T2, E> + 'static,

Returns a new instance of self applying logic on all Ok records, passes through Err records.

fn unwrap_or_else<L>(&self, logic: L) -> Stream<S, T>

where L: FnMut(E) -> T + 'static,

Returns a new instance of self applying logic on all Ok records.

impl<S, K, V> StateMachine<S, K, V> for Stream<S, (K, V)>

where S: Scope, K: ExchangeData + Hash + Eq, V: ExchangeData,

fn state_machine<R, D, I, F, H>(&self, fold: F, hash: H) -> Stream<S, R>

where R: Data, D: Default + 'static, I: IntoIterator<Item = R>, F: Fn(&K, V, &mut D) -> (bool, I) + 'static, H: Fn(&K) -> u64 + 'static, <S as ScopeParent>::Timestamp: Hash + Eq,

Tracks a state for each presented key, using user-supplied state transition logic.

impl<G: Scope> Sum<G> for Stream<G, usize>

fn sum(&self) -> Stream<G, usize>