Struct timely::dataflow::stream::StreamCore

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

Abstraction of a stream of D: Container 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: Scope, D: Container> StreamCore<S, D>

pub fn connect_to<P: Push<BundleCore<S::Timestamp, D>> + 'static>( &self, target: Target, pusher: P, identifier: usize )

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::Timestamp, D>, scope: S) -> Self

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<S: Scope, C: Container> BranchWhen<<S as ScopeParent>::Timestamp> for StreamCore<S, C>

fn branch_when( &self, condition: impl Fn(&S::Timestamp) -> bool + 'static ) -> (Self, Self)

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<S: Scope, D: Container> Capture<<S as ScopeParent>::Timestamp, D> for StreamCore<S, D>

fn capture_into<P: EventPusherCore<S::Timestamp, D> + 'static>( &self, event_pusher: P )

Captures a stream of timestamped data for later replay.

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

Captures a stream using Rust’s MPSC channels.

impl<S: Clone + Scope, D: Clone> Clone for StreamCore<S, D>where S::Timestamp: Clone,

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

Returns a copy of the value.

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

Performs copy-assignment from source.

impl<G: Scope, D: Container> Concat<G, D> for StreamCore<G, D>

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

Merge the contents of two streams.

impl<G: Scope, D: Container> Concatenate<G, D> for StreamCore<G, D>

fn concatenate<I>(&self, sources: I) -> StreamCore<G, D>where I: IntoIterator<Item = StreamCore<G, D>>,

Merge the contents of multiple streams.

impl<G: Scope, D: Container> ConnectLoop<G, D> for StreamCore<G, D>

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

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

impl<S, D> Debug for StreamCore<S, D>where S: Scope,

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

Formats the value using the given formatter.

impl<G: Scope, T: Timestamp + Refines<G::Timestamp>, C: Data + Container> Enter<G, T, C> for StreamCore<G, C>

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

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

impl<G: Scope, C> Exchange<<C as Container>::Item> for StreamCore<G, C>where C: PushPartitioned + ExchangeData, C::Item: ExchangeData,

fn exchange( &self, route: impl FnMut(&C::Item) -> u64 + 'static ) -> StreamCore<G, C>

Exchange records between workers.

impl<G: Scope, C> Inspect<G, Rc<C>> for StreamCore<G, Rc<C>>where C: AsRef<[C::Item]> + Container,

fn inspect_core<F>(&self, func: F) -> Selfwhere F: FnMut(Result<(&G::Timestamp, &[C::Item]), &[G::Timestamp]>) + 'static,

Runs a supplied closure on each observed data batch, and each frontier advancement.

fn inspect(&self, func: impl FnMut(&C::Item) + 'static) -> Self

Runs a supplied closure on each observed data element.

fn inspect_time( &self, func: impl FnMut(&G::Timestamp, &C::Item) + 'static ) -> Self

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

fn inspect_batch( &self, func: impl FnMut(&G::Timestamp, &[C::Item]) + 'static ) -> Self

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

impl<G: Scope, D: Data + Columnation> Inspect<G, TimelyStack<D>> for StreamCore<G, TimelyStack<D>>

fn inspect_core<F>(&self, func: F) -> Selfwhere F: FnMut(Result<(&G::Timestamp, &[D]), &[G::Timestamp]>) + 'static,

Runs a supplied closure on each observed data batch, and each frontier advancement.

fn inspect(&self, func: impl FnMut(&C::Item) + 'static) -> Self

Runs a supplied closure on each observed data element.

fn inspect_time( &self, func: impl FnMut(&G::Timestamp, &C::Item) + 'static ) -> Self

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

fn inspect_batch( &self, func: impl FnMut(&G::Timestamp, &[C::Item]) + 'static ) -> Self

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

impl<G: Scope, D: Data> Inspect<G, Vec<D, Global>> for StreamCore<G, Vec<D>>

fn inspect_core<F>(&self, func: F) -> Selfwhere F: FnMut(Result<(&G::Timestamp, &[D]), &[G::Timestamp]>) + 'static,

Runs a supplied closure on each observed data batch, and each frontier advancement.

fn inspect(&self, func: impl FnMut(&C::Item) + 'static) -> Self

Runs a supplied closure on each observed data element.

fn inspect_time( &self, func: impl FnMut(&G::Timestamp, &C::Item) + 'static ) -> Self

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

fn inspect_batch( &self, func: impl FnMut(&G::Timestamp, &[C::Item]) + 'static ) -> Self

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

impl<G: Scope, C: Container> InspectCore<G, C> for StreamCore<G, C>

fn inspect_container<F>(&self, func: F) -> StreamCore<G, C>where F: FnMut(Result<(&G::Timestamp, &C), &[G::Timestamp]>) + 'static,

Runs a supplied closure on each observed container, and each frontier advancement.

impl<'a, G: Scope, D: Clone + Container, T: Timestamp + Refines<G::Timestamp>> Leave<G, D> for StreamCore<Child<'a, G, T>, D>

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

Moves a Stream to the parent of its current Scope.

impl<G: Scope, D1: Container> Operator<G, D1> for StreamCore<G, D1>

fn unary_frontier<D2, B, L, P>( &self, pact: P, name: &str, constructor: B ) -> StreamCore<G, D2>where D2: Container, B: FnOnce(Capability<G::Timestamp>, OperatorInfo) -> L, L: FnMut(&mut FrontieredInputHandleCore<'_, G::Timestamp, D1, P::Puller>, &mut OutputHandleCore<'_, G::Timestamp, D2, TeeCore<G::Timestamp, D2>>) + 'static, P: ParallelizationContractCore<G::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: Container, L: FnMut(&mut InputHandleCore<G::Timestamp, D1, P::Puller>, &mut OutputHandleCore<'_, G::Timestamp, D2, TeeCore<G::Timestamp, D2>>, &mut Notificator<'_, G::Timestamp>) + 'static, P: ParallelizationContractCore<G::Timestamp, D1>>( &self, pact: P, name: &str, init: impl IntoIterator<Item = G::Timestamp>, logic: L ) -> StreamCore<G, D2>

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 ) -> StreamCore<G, D2>where D2: Container, B: FnOnce(Capability<G::Timestamp>, OperatorInfo) -> L, L: FnMut(&mut InputHandleCore<G::Timestamp, D1, P::Puller>, &mut OutputHandleCore<'_, G::Timestamp, D2, TeeCore<G::Timestamp, D2>>) + 'static, P: ParallelizationContractCore<G::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: &StreamCore<G, D2>, pact1: P1, pact2: P2, name: &str, constructor: B ) -> StreamCore<G, D3>where D2: Container, D3: Container, B: FnOnce(Capability<G::Timestamp>, OperatorInfo) -> L, L: FnMut(&mut FrontieredInputHandleCore<'_, G::Timestamp, D1, P1::Puller>, &mut FrontieredInputHandleCore<'_, G::Timestamp, D2, P2::Puller>, &mut OutputHandleCore<'_, G::Timestamp, D3, TeeCore<G::Timestamp, D3>>) + 'static, P1: ParallelizationContractCore<G::Timestamp, D1>, P2: ParallelizationContractCore<G::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: Container, D3: Container, L: FnMut(&mut InputHandleCore<G::Timestamp, D1, P1::Puller>, &mut InputHandleCore<G::Timestamp, D2, P2::Puller>, &mut OutputHandleCore<'_, G::Timestamp, D3, TeeCore<G::Timestamp, D3>>, &mut Notificator<'_, G::Timestamp>) + 'static, P1: ParallelizationContractCore<G::Timestamp, D1>, P2: ParallelizationContractCore<G::Timestamp, D2>>( &self, other: &StreamCore<G, D2>, pact1: P1, pact2: P2, name: &str, init: impl IntoIterator<Item = G::Timestamp>, logic: L ) -> StreamCore<G, D3>

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: &StreamCore<G, D2>, pact1: P1, pact2: P2, name: &str, constructor: B ) -> StreamCore<G, D3>where D2: Container, D3: Container, B: FnOnce(Capability<G::Timestamp>, OperatorInfo) -> L, L: FnMut(&mut InputHandleCore<G::Timestamp, D1, P1::Puller>, &mut InputHandleCore<G::Timestamp, D2, P2::Puller>, &mut OutputHandleCore<'_, G::Timestamp, D3, TeeCore<G::Timestamp, D3>>) + 'static, P1: ParallelizationContractCore<G::Timestamp, D1>, P2: ParallelizationContractCore<G::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 FrontieredInputHandleCore<'_, G::Timestamp, D1, P::Puller>) + 'static, P: ParallelizationContractCore<G::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: Scope, D: Container> Probe<G, D> for StreamCore<G, D>

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

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

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

Inserts a progress probe in a stream.

impl<S: Scope, C: Container> Reclock<S> for StreamCore<S, C>

fn reclock<TC: Container<Item = ()>>( &self, clock: &StreamCore<S, TC> ) -> StreamCore<S, C>

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

impl<S: Scope, C: Container> SharedStream<S, C> for StreamCore<S, C>

fn shared(&self) -> StreamCore<S, Rc<C>>

Convert a stream into a stream of shared data