Struct actyxos_data_flow::flow::Flow

pub struct Flow<'a, T: ExchangeData, St: NeedsState>(_, _);

Differential dataflow Collection wrapper

This wrapper type fixes the timestamp type to usize, the multiplicity to isize and the scope to the level below a Worker. These choices present restrictions that we find commonly useful for a certain class of problems when ingesting ActyxOS events and turning them into database rows.

Flows are constructed from a scope like this:

use actyxos_data_flow::flow::{Scope, Flow, Input, Stateless};

fn mk_logic<'a>(scope: &mut Scope<'a>) -> (Input<String>, Flow<'a, usize, Stateless>) {
    let (input, flow) = Flow::<String, _>::new(scope);
    let out = flow.map(|s| s.len());
    (input, out)
}

When the flow’s calculations depend only on a limited amount of historical data after a restart, you may use the look_back feature of then input collection:

use std::time::Duration;
let (input, flow) = Flow::<String, _>::new_limited(scope, Duration::from_secs(3600));

Implementations

impl<'a, T: ExchangeData> Flow<'a, T, Stateless>

pub fn new(
    scope: &mut Child<'a, Worker<Thread>, usize>
) -> (Input<T>, Flow<'a, T, Stateless>)

Create a new flow within the given scope

pub fn new_limited(
    scope: &mut Child<'a, Worker<Thread>, usize>,
    look_back: Duration
) -> (Input<T>, Flow<'a, T, Stateless>)

Create a new flow with limited look_back period within the given scope

see also Input::look_back

impl<'a, T: ExchangeData, St: NeedsState> Flow<'a, T, St>

pub fn filter(&self, f: impl Fn(&T) -> bool + 'static) -> Self

Filter this collection with the given predicate

pub fn filter_mut(
    &self,
    f: impl FnMut(&T) -> bool + 'static
) -> Flow<'a, T, Stateful>

Filter this collection with the given stateful predicate

pub fn map<U: ExchangeData>(
    &self,
    f: impl Fn(T) -> U + 'static
) -> Flow<'a, U, St>

Transform this collection’s elements 1:1

pub fn map_mut<U: ExchangeData>(
    &self,
    f: impl FnMut(T) -> U + 'static
) -> Flow<'a, U, Stateful>

Transform this collection’s elements 1:1 with a stateful function

pub fn map_in_place(&self, f: impl Fn(&mut T) + 'static) -> Self

Transform this collection’s elements 1:1 while keeping the same type

pub fn map_in_place_mut(
    &self,
    f: impl FnMut(&mut T) + 'static
) -> Flow<'a, T, Stateful>

Transform this collection’s elements 1:1 with a stateful function while keeping the same type

pub fn flat_map<U, I>(&self, f: impl Fn(T) -> I + 'static) -> Flow<'a, U, St> where
    U: ExchangeData,
    I: IntoIterator<Item = U>, 

Transform this collection’s elements 1:many

pub fn flat_map_mut<U, I>(
    &self,
    f: impl FnMut(T) -> I + 'static
) -> Flow<'a, U, Stateful> where
    U: ExchangeData,
    I: IntoIterator<Item = U>, 

Transform this collection’s elements 1:many with a stateful function

pub fn monotonic_max_by<K: ExchangeData>(
    &self,
    f: impl Fn(&T) -> K + 'static
) -> Flow<'a, T, Stateful>

Retain only the maximum element for each key computed by the given function

This function is an optimization over using .group().max() in that it does not retain the elements previously added to the collection. Therefore it cannot deal with the situation that the currently known maximum for a group is removed.

pub fn monotonic_representative_by<K: ExchangeData>(
    &self,
    f: impl Fn(&T) -> K + 'static
) -> Flow<'a, T, Stateful>

Retain only one representative for each key computed by the given function

This function is an optimization over using .group().min() in that it does not retain the elements previously added to the collection. Therefore it cannot deal with the situation that the chosen representative is removed.

The chosen representative is the first element to be seen for each key.

pub fn negate(&self) -> Self

Turn additions into removals and vice versa

pub fn group_by<K: ExchangeData + Hashable>(
    &self,
    f: impl FnMut(&T) -> K + 'static
) -> Grouped<'a, K, T>

Arrange this collection according to the computed keys

This function is used to access the join, reduce, etc. methods of the Grouped type, it has no inherent value by itself.

pub fn inspect(&self, f: impl Fn(&(T, usize, isize)) + 'static) -> Self

Inspect elements as they flow through the underlying timely dataflow stream

pub fn inspect_mut(
    &self,
    f: impl FnMut(&(T, usize, isize)) + 'static
) -> Flow<'a, T, Stateful>

Inspect elements as they flow through the underlying timely dataflow stream using a stateful function

pub fn probe(&self) -> Probe

Attach a probe to this collection to check the propagation of input timestamps

pub fn output(&self) -> Output<T>

Turn this flow into an output to be consumed by a machine

see also Machine

impl<'a, T: ExchangeData> Flow<'a, T, Stateless>

pub fn concat<St: NeedsState>(&self, other: &Flow<'a, T, St>) -> Flow<'a, T, St>

Compute the union with the other flow

pub fn concat_many<St: NeedsState>(
    &self,
    others: impl IntoIterator<Item = Flow<'a, T, St>>
) -> Flow<'a, T, St>

Compute the union with many other flows

impl<'a, T: ExchangeData> Flow<'a, T, Stateful>

pub fn concat<St: NeedsState>(
    &self,
    other: &Flow<'a, T, St>
) -> Flow<'a, T, Stateful>

Compute the union with the other flow

pub fn concat_many<St: NeedsState>(
    &self,
    others: impl IntoIterator<Item = Flow<'a, T, St>>
) -> Flow<'a, T, Stateful>

Compute the union with many other flows

impl<'a, T: ExchangeData + Hashable, St: NeedsState> Flow<'a, T, St>

pub fn distinct(&self) -> Flow<'a, T, Stateful>

Reduce the multiplicity of each element in this flow to 1

pub fn threshold(
    &self,
    f: impl FnMut(&T, isize) -> isize + 'static
) -> Flow<'a, T, Stateful>

Transform the multiplicity of each element in this flow with the given function

pub fn count(&self) -> Flow<'a, (T, isize), Stateful>

Count the number of elements in this collection

impl<'a, K: ExchangeData + Hashable, V: ExchangeData, St: NeedsState> Flow<'a, (K, V), St>

pub fn group(&self) -> Grouped<'a, K, V>

Group this flow of K-V pairs by the first element (the key) of the pair