Struct timely::dataflow::operators::generic::FrontieredInputHandle

pub struct FrontieredInputHandle<'a, T: Timestamp, D: 'a, P: Pull<Bundle<T, D>> + 'a> {
    pub handle: &'a mut InputHandle<T, D, P>,
    pub frontier: &'a MutableAntichain<T>,
}

Handle to an operator’s input stream and frontier.

Fields

handle: &'a mut InputHandle<T, D, P>

The underlying input handle.

frontier: &'a MutableAntichain<T>

The frontier as reported by timely progress tracking.

Implementations

impl<'a, T: Timestamp, D: Data, P: Pull<Bundle<T, D>> + 'a> FrontieredInputHandle<'a, T, D, P>

pub fn new(
    handle: &'a mut InputHandle<T, D, P>,
    frontier: &'a MutableAntichain<T>
) -> Self

Allocate a new frontiered input handle.

pub fn next(&mut self) -> Option<(CapabilityRef<'_, T>, RefOrMut<'_, Vec<D>>)>

Reads the next input buffer (at some timestamp t) and a corresponding capability for t. The timestamp t of the input buffer can be retrieved by invoking .time() on the capability. Returns None when there’s no more data available.

pub fn for_each<F: FnMut(CapabilityRef<'_, T>, RefOrMut<'_, Vec<D>>)>(
    &mut self,
    logic: F
)

Repeatedly calls logic till exhaustion of the available input data. logic receives a capability and an input buffer.

Examples

use timely::dataflow::operators::ToStream;
use timely::dataflow::operators::generic::Operator;
use timely::dataflow::channels::pact::Pipeline;

timely::example(|scope| {
    (0..10).to_stream(scope)
           .unary(Pipeline, "example", |_cap,_info| |input, output| {
               input.for_each(|cap, data| {
                   output.session(&cap).give_vec(&mut data.replace(Vec::new()));
               });
           });
});

pub fn frontier(&self) -> &'a MutableAntichain<T>

Inspect the frontier associated with this input.