Struct actyxos_data_flow::flow::Input

pub struct Input<T: ExchangeData>(_, _);

An input to a Flow

An Input is the entry point by which data records enter a Flow. Both of these are created together by the Flow::new method.

Implementations

impl<T: ExchangeData> Input<T>

pub fn advance_to(&mut self, time: usize)

Advance the input timestamp to the given value

This is usually done after ingesting a batch of data, followed by a .flush() to send the ingested collection elements through the flow and have them generate their deltas.

pub fn flush(&mut self)

Flush this input’s inserted elements into the collection

The worker executing this flow can then be stepped until the resulting deltas reach their designated outputs.

pub fn insert(&mut self, value: T)

Insert a new element into the collection

pub fn remove(&mut self, value: T)

Remove an element from the collection

pub fn look_back(&self) -> Option<Duration>

Query this input’s desired look back interval

In general, correct function of a flow requires that all data are inserted so that all desired outputs are computed. When restarting this process, ingesting all data from the beginning can be quite time-consuming. It is not uncommon that the logic expressed by the flow does not care about elements of arbitrary age to correctly compute deltas for new (current) elements.

For example, in a factory each production order is only relevant for a few days or weeks, matching the timespan needed to fulfil that order. Or an operations dashboard may focus on the behavior of the system over the past 24 hours and thus inputs from one week ago have no influence anymore on what shall be on the screens.

In these cases, limited look back is a suitable performance optimization to speed up system restarts. When full precision is paramount, this shall be set to None to always recompute the full state of the flow after a restart.

This parameter can be set by using Flow::new_limited.

Trait Implementations

impl<T: ExchangeData + DeserializeOwned> Inputs for Input<Event<T>>

type Elem = Event<Payload>

fn advance_clock(&mut self, time: usize)

fn feed(&mut self, input: &Event<Payload>) -> Result<()>

fn look_back(&self) -> Option<Duration>