Struct mz_timely_util::builder_async::OperatorBuilder

pub struct OperatorBuilder<G: Scope> { /* private fields */ }

Builds async operators with generic shape.

Implementations

impl<G: Scope> OperatorBuilder<G>

pub fn new(name: String, scope: G) -> Self

Allocates a new generic async operator builder from its containing scope.

pub fn new_input_for<D: Container, P>( &mut self, stream: &StreamCore<G, D>, pact: P, output: &dyn OutputIndex ) -> AsyncInputHandle<G::Timestamp, D, ConnectedToOne>

where P: ParallelizationContract<G::Timestamp, D>,

Adds a new input that is connected to the specified output, returning the async input handle to use.

pub fn new_input_for_many<const N: usize, D: Container, P>( &mut self, stream: &StreamCore<G, D>, pact: P, outputs: [&dyn OutputIndex; N] ) -> AsyncInputHandle<G::Timestamp, D, ConnectedToMany<N>>

where P: ParallelizationContract<G::Timestamp, D>,

Adds a new input that is connected to the specified outputs, returning the async input handle to use.

pub fn new_disconnected_input<D: Container, P>( &mut self, stream: &StreamCore<G, D>, pact: P ) -> AsyncInputHandle<G::Timestamp, D, Disconnected>

where P: ParallelizationContract<G::Timestamp, D>,

Adds a new input that is not connected to any output, returning the async input handle to use.

pub fn new_input_connection<D: Container, P, C>( &mut self, stream: &StreamCore<G, D>, pact: P, connection: C ) -> AsyncInputHandle<G::Timestamp, D, C>

where P: ParallelizationContract<G::Timestamp, D>, C: InputConnection<G::Timestamp> + 'static,

Adds a new input with connection information, returning the async input handle to use.

pub fn new_output<D: Container>( &mut self ) -> (AsyncOutputHandle<G::Timestamp, D, Tee<G::Timestamp, D>>, StreamCore<G, D>)

Adds a new output, returning the output handle and stream.

pub fn build<B, L>(self, constructor: B) -> Button

where B: FnOnce(Vec<Capability<G::Timestamp>>) -> L, L: Future + 'static,

Creates an operator implementation from supplied logic constructor. It returns a shutdown button that when pressed it will cause the logic future to be dropped and input handles to be drained. The button can be converted into a token by using Button::press_on_drop

pub fn build_fallible<E: 'static, F>( self, constructor: F ) -> (Button, StreamCore<G, Vec<Rc<E>>>)

where F: for<'a> FnOnce(&'a mut [CapabilitySet<G::Timestamp>]) -> Pin<Box<dyn Future<Output = Result<(), E>> + 'a>> + 'static,

Creates a fallible operator implementation from supplied logic constructor. If the Future resolves to an error it will be emitted in the returned error stream and then the operator will wait indefinitely until the shutdown button is pressed.

Capability handling

Unlike OperatorBuilder::build, this method does not give owned capabilities to the constructor. All initial capabilities are wrapped in a CapabilitySet and a mutable reference to them is given instead. This is done to avoid storing owned capabilities in the state of the logic future which would make using the ? operator unsafe, since the frontiers would incorrectly advance, potentially causing incorrect actions downstream.

builder.build_fallible(|caps| Box::pin(async move {
    // Assert that we have the number of capabilities we expect
    // `cap` will be a `&mut Option<Capability<T>>`:
    let [cap_set]: &mut [_; 1] = caps.try_into().unwrap();

    // Using cap to send data:
    output.give(&cap_set[0], 42);

    // Using cap_set to downgrade it:
    cap_set.downgrade([]);

    // Explicitly dropping the capability:
    // Simply running `drop(cap_set)` will only drop the reference and not the capability set itself!
    *cap_set = CapabilitySet::new();

    // !! BIG WARNING !!:
    // It is tempting to `take` the capability out of the set for convenience. This will
    // move the capability into the future state, tying its lifetime to it, which will get
    // dropped when an error is hit, causing incorrect progress statements.
    let cap = cap_set.delayed(&Timestamp::minimum());
    *cap_set = CapabilitySet::new(); // DO NOT DO THIS
}));

pub fn operator_info(&self) -> OperatorInfo

Creates operator info for the operator.

pub fn activator(&self) -> &Activator

Returns the activator for the operator.