Struct Stream 

pub struct Stream<Type, Loc, Bound: Boundedness, Order = TotalOrder, Retries = ExactlyOnce> { /* private fields */ }

An ordered sequence stream of elements of type T.

Type Parameters:

• Type: the type of elements in the stream
• Loc: the location where the stream is being materialized
• Bound: the boundedness of the stream, which is either Bounded or Unbounded
• Order: the ordering of the stream, which is either TotalOrder or NoOrder (default is TotalOrder)

Implementations

impl<'a, T, L, B: Boundedness, O, R> Stream<T, Cluster<'a, L>, B, O, R>

pub fn send_bincode<L2>( self, other: &Process<'a, L2>, ) -> KeyedStream<MemberId<L>, T, Process<'a, L2>, Unbounded, O, R>

where T: Serialize + DeserializeOwned,

pub fn broadcast_bincode<L2: 'a>( self, other: &Cluster<'a, L2>, nondet_membership: NonDet, ) -> KeyedStream<MemberId<L>, T, Cluster<'a, L2>, Unbounded, O, R>

where T: Clone + Serialize + DeserializeOwned,

impl<'a, T, L, L2, B: Boundedness, O, R> Stream<(MemberId<L2>, T), Process<'a, L>, B, O, R>

pub fn demux_bincode( self, other: &Cluster<'a, L2>, ) -> Stream<T, Cluster<'a, L2>, Unbounded, O, R>

where T: Serialize + DeserializeOwned,

impl<'a, T, L, B: Boundedness> Stream<T, Process<'a, L>, B, TotalOrder, ExactlyOnce>

pub fn round_robin_bincode<L2: 'a>( self, other: &Cluster<'a, L2>, nondet_membership: NonDet, ) -> Stream<T, Cluster<'a, L2>, Unbounded, TotalOrder, ExactlyOnce>

where T: Serialize + DeserializeOwned,

impl<'a, T, L, L2, B: Boundedness, O, R> Stream<(MemberId<L2>, T), Cluster<'a, L>, B, O, R>

pub fn demux_bincode( self, other: &Cluster<'a, L2>, ) -> KeyedStream<MemberId<L>, T, Cluster<'a, L2>, Unbounded, O, R>

where T: Serialize + DeserializeOwned,

impl<'a, T, L, B: Boundedness, O, R> Stream<T, Process<'a, L>, B, O, R>

pub fn send_bincode<L2>( self, other: &Process<'a, L2>, ) -> Stream<T, Process<'a, L2>, Unbounded, O, R>

where T: Serialize + DeserializeOwned,

pub fn broadcast_bincode<L2: 'a>( self, other: &Cluster<'a, L2>, nondet_membership: NonDet, ) -> Stream<T, Cluster<'a, L2>, Unbounded, O, R>

where T: Clone + Serialize + DeserializeOwned,

pub fn send_bincode_external<L2>( self, other: &External<'_, L2>, ) -> ExternalBincodeStream<T>

where T: Serialize + DeserializeOwned,

impl<'a, T, L, B: Boundedness, O, R> Stream<T, L, B, O, R>

where L: Location<'a>,

pub fn map<U, F>(self, f: impl IntoQuotedMut<'a, F, L>) -> Stream<U, L, B, O, R>

where F: Fn(T) -> U + 'a,

Produces a stream based on invoking f on each element. If you do not want to modify the stream and instead only want to view each item use Stream::inspect instead.

Example

let words = process.source_iter(q!(vec!["hello", "world"]));
words.map(q!(|x| x.to_uppercase()))

pub fn flat_map_ordered<U, I, F>( self, f: impl IntoQuotedMut<'a, F, L>, ) -> Stream<U, L, B, O, R>

where I: IntoIterator<Item = U>, F: Fn(T) -> I + 'a,

For each item i in the input stream, transform i using f and then treat the result as an Iterator to produce items one by one. The implementation for Iterator for the output type U must produce items in a deterministic order.

For example, U could be a Vec, but not a HashSet. If the order of the items in U is not deterministic, use Stream::flat_map_unordered instead.

Example

process
    .source_iter(q!(vec![vec![1, 2], vec![3, 4]]))
    .flat_map_ordered(q!(|x| x))
// 1, 2, 3, 4

pub fn flat_map_unordered<U, I, F>( self, f: impl IntoQuotedMut<'a, F, L>, ) -> Stream<U, L, B, NoOrder, R>

where I: IntoIterator<Item = U>, F: Fn(T) -> I + 'a,

Like Stream::flat_map_ordered, but allows the implementation of Iterator for the output type U to produce items in any order.

Example

process
    .source_iter(q!(vec![
        std::collections::HashSet::<i32>::from_iter(vec![1, 2]),
        std::collections::HashSet::from_iter(vec![3, 4]),
    ]))
    .flat_map_unordered(q!(|x| x))
// 1, 2, 3, 4, but in no particular order

pub fn flatten_ordered<U>(self) -> Stream<U, L, B, O, R>

where T: IntoIterator<Item = U>,

For each item i in the input stream, treat i as an Iterator and produce its items one by one. The implementation for Iterator for the element type T must produce items in a deterministic order.

For example, T could be a Vec, but not a HashSet. If the order of the items in T is not deterministic, use Stream::flatten_unordered instead.

process
    .source_iter(q!(vec![vec![1, 2], vec![3, 4]]))
    .flatten_ordered()
// 1, 2, 3, 4

pub fn flatten_unordered<U>(self) -> Stream<U, L, B, NoOrder, R>

where T: IntoIterator<Item = U>,

Like Stream::flatten_ordered, but allows the implementation of Iterator for the element type T to produce items in any order.

Example

process
    .source_iter(q!(vec![
        std::collections::HashSet::<i32>::from_iter(vec![1, 2]),
        std::collections::HashSet::from_iter(vec![3, 4]),
    ]))
    .flatten_unordered()
// 1, 2, 3, 4, but in no particular order

pub fn filter<F>(self, f: impl IntoQuotedMut<'a, F, L>) -> Stream<T, L, B, O, R>

where F: Fn(&T) -> bool + 'a,

Creates a stream containing only the elements of the input stream that satisfy a predicate f, preserving the order of the elements.

The closure f receives a reference &T rather than an owned value T because filtering does not modify or take ownership of the values. If you need to modify the values while filtering use Stream::filter_map instead.

Example

process
    .source_iter(q!(vec![1, 2, 3, 4]))
    .filter(q!(|&x| x > 2))
// 3, 4

pub fn filter_map<U, F>( self, f: impl IntoQuotedMut<'a, F, L>, ) -> Stream<U, L, B, O, R>

where F: Fn(T) -> Option<U> + 'a,

An operator that both filters and maps. It yields only the items for which the supplied closure f returns Some(value).

Example

process
    .source_iter(q!(vec!["1", "hello", "world", "2"]))
    .filter_map(q!(|s| s.parse::<usize>().ok()))
// 1, 2

pub fn cross_singleton<O2>( self, other: impl Into<Optional<O2, L, Bounded>>, ) -> Stream<(T, O2), L, B, O, R>

where O2: Clone,

Generates a stream that maps each input element i to a tuple (i, x), where x is the final value of other, a bounded Singleton.

Example

let tick = process.tick();
let batch = process
  .source_iter(q!(vec![1, 2, 3, 4]))
  .batch(&tick, nondet!(/** test */));
let count = batch.clone().count(); // `count()` returns a singleton
batch.cross_singleton(count).all_ticks()
// (1, 4), (2, 4), (3, 4), (4, 4)

pub fn continue_if<U>( self, signal: Optional<U, L, Bounded>, ) -> Stream<T, L, B, O, R>

Allow this stream through if the argument (a Bounded Optional) is non-empty, otherwise the output is empty.

pub fn continue_unless<U>( self, other: Optional<U, L, Bounded>, ) -> Stream<T, L, B, O, R>

Allow this stream through if the argument (a Bounded Optional) is empty, otherwise the output is empty.

pub fn cross_product<T2, O2>( self, other: Stream<T2, L, B, O2, R>, ) -> Stream<(T, T2), L, B, NoOrder, R>

where T: Clone, T2: Clone,

Forms the cross-product (Cartesian product, cross-join) of the items in the 2 input streams, returning all tupled pairs in a non-deterministic order.

Example

let tick = process.tick();
let stream1 = process.source_iter(q!(vec!['a', 'b', 'c']));
let stream2 = process.source_iter(q!(vec![1, 2, 3]));
stream1.cross_product(stream2)

pub fn unique(self) -> Stream<T, L, B, O, ExactlyOnce>

where T: Eq + Hash,

Takes one stream as input and filters out any duplicate occurrences. The output contains all unique values from the input.

Example

let tick = process.tick();
    process.source_iter(q!(vec![1, 2, 3, 2, 1, 4])).unique()

pub fn filter_not_in<O2>( self, other: Stream<T, L, Bounded, O2, R>, ) -> Stream<T, L, Bounded, O, R>

where T: Eq + Hash,

Outputs everything in this stream that is not contained in the other stream.

The other stream must be Bounded, since this function will wait until all its elements are available before producing any output.

Example

let tick = process.tick();
let stream = process
  .source_iter(q!(vec![ 1, 2, 3, 4 ]))
  .batch(&tick, nondet!(/** test */));
let batch = process
  .source_iter(q!(vec![1, 2]))
  .batch(&tick, nondet!(/** test */));
stream.filter_not_in(batch).all_ticks()

pub fn inspect<F>( self, f: impl IntoQuotedMut<'a, F, L>, ) -> Stream<T, L, B, O, R>

where F: Fn(&T) + 'a,

An operator which allows you to “inspect” each element of a stream without modifying it. The closure f is called on a reference to each item. This is mainly useful for debugging, and should not be used to generate side-effects.

Example

let nums = process.source_iter(q!(vec![1, 2]));
// prints "1 * 10 = 10" and "2 * 10 = 20"
nums.inspect(q!(|x| println!("{} * 10 = {}", x, x * 10)))

pub fn ir_node_named(self, name: &str) -> Stream<T, L, B, O, R>

An operator which allows you to “name” a HydroNode. This is only used for testing, to correlate certain HydroNodes with IDs.

pub fn assume_ordering<O2>(self, _nondet: NonDet) -> Stream<T, L, B, O2, R>

Explicitly “casts” the stream to a type with a different ordering guarantee. Useful in unsafe code where the ordering cannot be proven by the type-system.

Non-Determinism

This function is used as an escape hatch, and any mistakes in the provided ordering guarantee will propagate into the guarantees for the rest of the program.

pub fn weakest_ordering(self) -> Stream<T, L, B, NoOrder, R>

Weakens the ordering guarantee provided by the stream to NoOrder, which is always safe because that is the weakest possible guarantee.

pub fn assume_retries<R2>(self, _nondet: NonDet) -> Stream<T, L, B, O, R2>

Explicitly “casts” the stream to a type with a different retries guarantee. Useful in unsafe code where the lack of retries cannot be proven by the type-system.

Non-Determinism

This function is used as an escape hatch, and any mistakes in the provided retries guarantee will propagate into the guarantees for the rest of the program.

pub fn weakest_retries(self) -> Stream<T, L, B, O, AtLeastOnce>

Weakens the retries guarantee provided by the stream to AtLeastOnce, which is always safe because that is the weakest possible guarantee.

pub fn weaken_retries<R2>( self, ) -> Stream<T, L, B, O, <R as MinRetries<R2>>::Min>

where R: MinRetries<R2>,

Weakens the retries guarantee provided by the stream to be the weaker of the current guarantee and R2. This is safe because the output guarantee will always be weaker than the input.

impl<'a, T, L, B: Boundedness, O> Stream<T, L, B, O, ExactlyOnce>

where L: Location<'a>,

pub fn weaker_retries<R2>(self) -> Stream<T, L, B, O, R2>

Given a stream with ExactlyOnce retry guarantees, weakens it to an arbitrary guarantee R2, which is safe because all guarantees are equal to or weaker than ExactlyOnce

impl<'a, T, L, B: Boundedness, O, R> Stream<&T, L, B, O, R>

where L: Location<'a>,

pub fn cloned(self) -> Stream<T, L, B, O, R>

where T: Clone,

Clone each element of the stream; akin to map(q!(|d| d.clone())).

Example

process.source_iter(q!(&[1, 2, 3])).cloned()
// 1, 2, 3

impl<'a, T, L, B: Boundedness, O, R> Stream<T, L, B, O, R>

where L: Location<'a>,

pub fn fold_commutative_idempotent<A, I, F>( self, init: impl IntoQuotedMut<'a, I, L>, comb: impl IntoQuotedMut<'a, F, L>, ) -> Singleton<A, L, B>

where I: Fn() -> A + 'a, F: Fn(&mut A, T),

Combines elements of the stream into a Singleton, by starting with an initial value, generated by the init closure, and then applying the comb closure to each element in the stream. Unlike iterators, comb takes the accumulator by &mut reference, so that it can be modified in place.

The comb closure must be commutative AND idempotent, as the order of input items is not guaranteed and there may be duplicates.

Example

let tick = process.tick();
let bools = process.source_iter(q!(vec![false, true, false]));
let batch = bools.batch(&tick, nondet!(/** test */));
batch
    .fold_commutative_idempotent(q!(|| false), q!(|acc, x| *acc |= x))
    .all_ticks()
// true

pub fn reduce_commutative_idempotent<F>( self, comb: impl IntoQuotedMut<'a, F, L>, ) -> Optional<T, L, B>

where F: Fn(&mut T, T) + 'a,

Combines elements of the stream into an Optional, by starting with the first element in the stream, and then applying the comb closure to each element in the stream. The Optional will be empty until the first element in the input arrives. Unlike iterators, comb takes the accumulator by &mut reference, so that it can be modified in place.

The comb closure must be commutative AND idempotent, as the order of input items is not guaranteed and there may be duplicates.

Example

let tick = process.tick();
let bools = process.source_iter(q!(vec![false, true, false]));
let batch = bools.batch(&tick, nondet!(/** test */));
batch
    .reduce_commutative_idempotent(q!(|acc, x| *acc |= x))
    .all_ticks()
// true

pub fn max(self) -> Optional<T, L, B>

where T: Ord,

Computes the maximum element in the stream as an Optional, which will be empty until the first element in the input arrives.

Example

let tick = process.tick();
let numbers = process.source_iter(q!(vec![1, 2, 3, 4]));
let batch = numbers.batch(&tick, nondet!(/** test */));
batch.max().all_ticks()
// 4

pub fn max_by_key<K, F>( self, key: impl IntoQuotedMut<'a, F, L> + Copy, ) -> Optional<T, L, B>

where K: Ord, F: Fn(&T) -> K + 'a,

Computes the maximum element in the stream as an Optional, where the maximum is determined according to the key function. The Optional will be empty until the first element in the input arrives.

Example

let tick = process.tick();
let numbers = process.source_iter(q!(vec![1, 2, 3, 4]));
let batch = numbers.batch(&tick, nondet!(/** test */));
batch.max_by_key(q!(|x| -x)).all_ticks()
// 1

pub fn min(self) -> Optional<T, L, B>

where T: Ord,

Computes the minimum element in the stream as an Optional, which will be empty until the first element in the input arrives.

Example

let tick = process.tick();
let numbers = process.source_iter(q!(vec![1, 2, 3, 4]));
let batch = numbers.batch(&tick, nondet!(/** test */));
batch.min().all_ticks()
// 1

impl<'a, T, L, B: Boundedness, O> Stream<T, L, B, O, ExactlyOnce>

where L: Location<'a>,

pub fn fold_commutative<A, I, F>( self, init: impl IntoQuotedMut<'a, I, L>, comb: impl IntoQuotedMut<'a, F, L>, ) -> Singleton<A, L, B>

where I: Fn() -> A + 'a, F: Fn(&mut A, T),

Combines elements of the stream into a Singleton, by starting with an initial value, generated by the init closure, and then applying the comb closure to each element in the stream. Unlike iterators, comb takes the accumulator by &mut reference, so that it can be modified in place.

The comb closure must be commutative, as the order of input items is not guaranteed.

Example

let tick = process.tick();
let numbers = process.source_iter(q!(vec![1, 2, 3, 4]));
let batch = numbers.batch(&tick, nondet!(/** test */));
batch
    .fold_commutative(q!(|| 0), q!(|acc, x| *acc += x))
    .all_ticks()
// 10

pub fn reduce_commutative<F>( self, comb: impl IntoQuotedMut<'a, F, L>, ) -> Optional<T, L, B>

where F: Fn(&mut T, T) + 'a,

Combines elements of the stream into a Optional, by starting with the first element in the stream, and then applying the comb closure to each element in the stream. The Optional will be empty until the first element in the input arrives. Unlike iterators, comb takes the accumulator by &mut reference, so that it can be modified in place.

The comb closure must be commutative, as the order of input items is not guaranteed.

Example

let tick = process.tick();
let numbers = process.source_iter(q!(vec![1, 2, 3, 4]));
let batch = numbers.batch(&tick, nondet!(/** test */));
batch
    .reduce_commutative(q!(|curr, new| *curr += new))
    .all_ticks()
// 10

pub fn count(self) -> Singleton<usize, L, B>

Computes the number of elements in the stream as a Singleton.

Example

let tick = process.tick();
let numbers = process.source_iter(q!(vec![1, 2, 3, 4]));
let batch = numbers.batch(&tick, nondet!(/** test */));
batch.count().all_ticks()
// 4

impl<'a, T, L, B: Boundedness, R> Stream<T, L, B, TotalOrder, R>

where L: Location<'a>,

pub fn fold_idempotent<A, I, F>( self, init: impl IntoQuotedMut<'a, I, L>, comb: impl IntoQuotedMut<'a, F, L>, ) -> Singleton<A, L, B>

where I: Fn() -> A + 'a, F: Fn(&mut A, T),

Combines elements of the stream into a Singleton, by starting with an initial value, generated by the init closure, and then applying the comb closure to each element in the stream. Unlike iterators, comb takes the accumulator by &mut reference, so that it can be modified in place.

The comb closure must be idempotent, as there may be non-deterministic duplicates.

Example

let tick = process.tick();
let bools = process.source_iter(q!(vec![false, true, false]));
let batch = bools.batch(&tick, nondet!(/** test */));
batch
    .fold_idempotent(q!(|| false), q!(|acc, x| *acc |= x))
    .all_ticks()
// true

pub fn reduce_idempotent<F>( self, comb: impl IntoQuotedMut<'a, F, L>, ) -> Optional<T, L, B>

where F: Fn(&mut T, T) + 'a,

Combines elements of the stream into an Optional, by starting with the first element in the stream, and then applying the comb closure to each element in the stream. The Optional will be empty until the first element in the input arrives. Unlike iterators, comb takes the accumulator by &mut reference, so that it can be modified in place.

The comb closure must be idempotent, as there may be non-deterministic duplicates.

Example

let tick = process.tick();
let bools = process.source_iter(q!(vec![false, true, false]));
let batch = bools.batch(&tick, nondet!(/** test */));
batch.reduce_idempotent(q!(|acc, x| *acc |= x)).all_ticks()
// true

pub fn first(self) -> Optional<T, L, B>

Computes the first element in the stream as an Optional, which will be empty until the first element in the input arrives.

This requires the stream to have a TotalOrder guarantee, otherwise re-ordering of elements may cause the first element to change.

Example

let tick = process.tick();
let numbers = process.source_iter(q!(vec![1, 2, 3, 4]));
let batch = numbers.batch(&tick, nondet!(/** test */));
batch.first().all_ticks()
// 1

pub fn last(self) -> Optional<T, L, B>

Computes the last element in the stream as an Optional, which will be empty until an element in the input arrives.

This requires the stream to have a TotalOrder guarantee, otherwise re-ordering of elements may cause the last element to change.

Example

let tick = process.tick();
let numbers = process.source_iter(q!(vec![1, 2, 3, 4]));
let batch = numbers.batch(&tick, nondet!(/** test */));
batch.last().all_ticks()
// 4

impl<'a, T, L, B: Boundedness> Stream<T, L, B, TotalOrder, ExactlyOnce>

where L: Location<'a>,

pub fn enumerate(self) -> Stream<(usize, T), L, B, TotalOrder, ExactlyOnce>

Returns a stream with the current count tupled with each element in the input stream.

Example

let tick = process.tick();
let numbers = process.source_iter(q!(vec![1, 2, 3, 4]));
numbers.enumerate()
// (0, 1), (1, 2), (2, 3), (3, 4)

pub fn fold<A, I: Fn() -> A + 'a, F: Fn(&mut A, T)>( self, init: impl IntoQuotedMut<'a, I, L>, comb: impl IntoQuotedMut<'a, F, L>, ) -> Singleton<A, L, B>

Combines elements of the stream into a Singleton, by starting with an intitial value, generated by the init closure, and then applying the comb closure to each element in the stream. Unlike iterators, comb takes the accumulator by &mut reference, so that it can be modified in place.

The input stream must have a TotalOrder guarantee, which means that the comb closure is allowed to depend on the order of elements in the stream.

Example

let tick = process.tick();
let words = process.source_iter(q!(vec!["HELLO", "WORLD"]));
let batch = words.batch(&tick, nondet!(/** test */));
batch
    .fold(q!(|| String::new()), q!(|acc, x| acc.push_str(x)))
    .all_ticks()
// "HELLOWORLD"

pub fn collect_vec(self) -> Singleton<Vec<T>, L, B>

pub fn scan<A, U, I, F>( self, init: impl IntoQuotedMut<'a, I, L>, f: impl IntoQuotedMut<'a, F, L>, ) -> Stream<U, L, B, TotalOrder, ExactlyOnce>

where I: Fn() -> A + 'a, F: Fn(&mut A, T) -> Option<U> + 'a,

Applies a function to each element of the stream, maintaining an internal state (accumulator) and emitting each intermediate result.

Unlike fold which only returns the final accumulated value, scan produces a new stream containing all intermediate accumulated values. The scan operation can also terminate early by returning None.

The function takes a mutable reference to the accumulator and the current element, and returns an Option<U>. If the function returns Some(value), value is emitted to the output stream. If the function returns None, the stream is terminated and no more elements are processed.

Examples

Basic usage - running sum:

process.source_iter(q!(vec![1, 2, 3, 4])).scan(
    q!(|| 0),
    q!(|acc, x| {
        *acc += x;
        Some(*acc)
    }),
)
// Output: 1, 3, 6, 10

Early termination example:

process.source_iter(q!(vec![1, 2, 3, 4])).scan(
    q!(|| 1),
    q!(|state, x| {
        *state = *state * x;
        if *state > 6 {
            None // Terminate the stream
        } else {
            Some(-*state)
        }
    }),
)
// Output: -1, -2, -6

pub fn reduce<F: Fn(&mut T, T) + 'a>( self, comb: impl IntoQuotedMut<'a, F, L>, ) -> Optional<T, L, B>

Combines elements of the stream into an Optional, by starting with the first element in the stream, and then applying the comb closure to each element in the stream. The Optional will be empty until the first element in the input arrives.

The input stream must have a TotalOrder guarantee, which means that the comb closure is allowed to depend on the order of elements in the stream.

Example

let tick = process.tick();
let words = process.source_iter(q!(vec!["HELLO", "WORLD"]));
let batch = words.batch(&tick, nondet!(/** test */));
batch
    .map(q!(|x| x.to_string()))
    .reduce(q!(|curr, new| curr.push_str(&new)))
    .all_ticks()
// "HELLOWORLD"

impl<'a, T, L: Location<'a> + NoTick + NoAtomic, O, R> Stream<T, L, Unbounded, O, R>

pub fn interleave<O2, R2: MinRetries<R>>( self, other: Stream<T, L, Unbounded, O2, R2>, ) -> Stream<T, L, Unbounded, NoOrder, R::Min>

where R: MinRetries<R2, Min = R2::Min>,

Produces a new stream that interleaves the elements of the two input streams. The result has NoOrder because the order of interleaving is not guaranteed.

Currently, both input streams must be Unbounded. When the streams are Bounded, you can use Stream::chain instead.

Example

let numbers = process.source_iter(q!(vec![1, 2, 3, 4]));
numbers.clone().map(q!(|x| x + 1)).interleave(numbers)
// 2, 3, 4, 5, and 1, 2, 3, 4 interleaved in unknown order

impl<'a, T, L, O, R> Stream<T, L, Bounded, O, R>

where L: Location<'a>,

pub fn sort(self) -> Stream<T, L, Bounded, TotalOrder, R>

where T: Ord,

Produces a new stream that emits the input elements in sorted order.

The input stream can have any ordering guarantee, but the output stream will have a TotalOrder guarantee. This operator will block until all elements in the input stream are available, so it requires the input stream to be Bounded.

Example

let tick = process.tick();
let numbers = process.source_iter(q!(vec![4, 2, 3, 1]));
let batch = numbers.batch(&tick, nondet!(/** test */));
batch.sort().all_ticks()
// 1, 2, 3, 4

pub fn chain<O2>( self, other: Stream<T, L, Bounded, O2, R>, ) -> Stream<T, L, Bounded, O::Min, R>

where O: MinOrder<O2>,

Produces a new stream that first emits the elements of the self stream, and then emits the elements of the other stream. The output stream has a TotalOrder guarantee if and only if both input streams have a TotalOrder guarantee.

Currently, both input streams must be Bounded. This operator will block on the first stream until all its elements are available. In a future version, we will relax the requirement on the other stream.

Example

let tick = process.tick();
let numbers = process.source_iter(q!(vec![1, 2, 3, 4]));
let batch = numbers.batch(&tick, nondet!(/** test */));
batch.clone().map(q!(|x| x + 1)).chain(batch).all_ticks()
// 2, 3, 4, 5, 1, 2, 3, 4

pub fn cross_product_nested_loop<T2, O2>( self, other: Stream<T2, L, Bounded, O2, R>, ) -> Stream<(T, T2), L, Bounded, O::Min, R>

where T: Clone, T2: Clone, O: MinOrder<O2>,

Forms the cross-product (Cartesian product, cross-join) of the items in the 2 input streams. Unlike Stream::cross_product, the output order is totally ordered when the inputs are because this is compiled into a nested loop.

impl<'a, K, V1, L, B: Boundedness, O, R> Stream<(K, V1), L, B, O, R>

where L: Location<'a>,

pub fn join<V2, O2>( self, n: Stream<(K, V2), L, B, O2, R>, ) -> Stream<(K, (V1, V2)), L, B, NoOrder, R>

where K: Eq + Hash,

Given two streams of pairs (K, V1) and (K, V2), produces a new stream of nested pairs (K, (V1, V2)) by equi-joining the two streams on the key attribute K.

Example

let tick = process.tick();
let stream1 = process.source_iter(q!(vec![(1, 'a'), (2, 'b')]));
let stream2 = process.source_iter(q!(vec![(1, 'x'), (2, 'y')]));
stream1.join(stream2)
// (1, ('a', 'x')), (2, ('b', 'y'))

pub fn anti_join<O2, R2>( self, n: Stream<K, L, Bounded, O2, R2>, ) -> Stream<(K, V1), L, B, O, R>

where K: Eq + Hash,

Given a stream of pairs (K, V1) and a bounded stream of keys K, computes the anti-join of the items in the input – i.e. returns unique items in the first input that do not have a matching key in the second input.

Example

let tick = process.tick();
let stream = process
  .source_iter(q!(vec![ (1, 'a'), (2, 'b'), (3, 'c'), (4, 'd') ]))
  .batch(&tick, nondet!(/** test */));
let batch = process
  .source_iter(q!(vec![1, 2]))
  .batch(&tick, nondet!(/** test */));
stream.anti_join(batch).all_ticks()

impl<'a, K, V, L: Location<'a>, B: Boundedness, O, R> Stream<(K, V), L, B, O, R>

pub fn into_keyed(self) -> KeyedStream<K, V, L, B, O, R>

impl<'a, K, V, L, B: Boundedness> Stream<(K, V), L, B, TotalOrder, ExactlyOnce>

where K: Eq + Hash, L: Location<'a>,

pub fn scan_keyed<A, U, I, F>( self, init: impl IntoQuotedMut<'a, I, L> + Copy, f: impl IntoQuotedMut<'a, F, L> + Copy, ) -> Stream<(K, U), L, B, TotalOrder, ExactlyOnce>

where K: Clone, I: Fn() -> A + 'a, F: Fn(&mut A, V) -> Option<U> + 'a,

A special case of Stream::scan, in the spirit of SQL’s GROUP BY and aggregation constructs. The input tuples are partitioned into groups by the first element (“keys”), and for each group the values in the second element are transformed via the f combinator.

Unlike Stream::fold_keyed which only returns the final accumulated value, scan produces a new stream containing all intermediate accumulated values paired with the key. The scan operation can also terminate early by returning None.

The function takes a mutable reference to the accumulator and the current element, and returns an Option<U>. If the function returns Some(value), value is emitted to the output stream. If the function returns None, the stream is terminated and no more elements are processed.

Example

process
    .source_iter(q!(vec![(0, 1), (0, 2), (1, 3), (1, 4)]))
    .scan_keyed(
        q!(|| 0),
        q!(|acc, x| {
            *acc += x;
            Some(*acc)
        }),
    )
// Output: (0, 1), (0, 3), (1, 3), (1, 7)

pub fn fold_keyed_early_stop<A, I, F>( self, init: impl IntoQuotedMut<'a, I, L> + Copy, f: impl IntoQuotedMut<'a, F, L> + Copy, ) -> Stream<(K, A), L, B, TotalOrder, ExactlyOnce>

where K: Clone, I: Fn() -> A + 'a, F: Fn(&mut A, V) -> bool + 'a,

Like Stream::fold_keyed, in the spirit of SQL’s GROUP BY and aggregation constructs. But the aggregation function returns a boolean, which when true indicates that the aggregated result is complete and can be released to downstream computation. Unlike Stream::fold_keyed, this means that even if the input stream is Unbounded, the outputs of the fold can be processed like normal stream elements.

Example

process
    .source_iter(q!(vec![(0, 2), (0, 3), (1, 3), (1, 6)]))
    .fold_keyed_early_stop(
        q!(|| 0),
        q!(|acc, x| {
            *acc += x;
            x % 2 == 0
        }),
    )
// Output: (0, 2), (1, 9)

impl<'a, K, V, L> Stream<(K, V), Tick<L>, Bounded, TotalOrder, ExactlyOnce>

where K: Eq + Hash, L: Location<'a>,

pub fn fold_keyed<A, I, F>( self, init: impl IntoQuotedMut<'a, I, Tick<L>>, comb: impl IntoQuotedMut<'a, F, Tick<L>>, ) -> Stream<(K, A), Tick<L>, Bounded, NoOrder, ExactlyOnce>

where I: Fn() -> A + 'a, F: Fn(&mut A, V) + 'a,

👎Deprecated: use .into_keyed().fold(…) instead

A special case of Stream::fold, in the spirit of SQL’s GROUP BY and aggregation constructs. The input tuples are partitioned into groups by the first element (“keys”), and for each group the values in the second element are accumulated via the comb closure.

The input stream must have a TotalOrder guarantee, which means that the comb closure is allowed to depend on the order of elements in the stream.

If the input and output value types are the same and do not require initialization then use Stream::reduce_keyed.

Example

let tick = process.tick();
let numbers = process.source_iter(q!(vec![(1, 2), (2, 3), (1, 3), (2, 4)]));
let batch = numbers.batch(&tick, nondet!(/** test */));
batch
    .fold_keyed(q!(|| 0), q!(|acc, x| *acc += x))
    .all_ticks()
// (1, 5), (2, 7)

pub fn reduce_keyed<F>( self, comb: impl IntoQuotedMut<'a, F, Tick<L>>, ) -> Stream<(K, V), Tick<L>, Bounded, NoOrder, ExactlyOnce>

where F: Fn(&mut V, V) + 'a,

👎Deprecated: use .into_keyed().reduce(…) instead

A special case of Stream::reduce, in the spirit of SQL’s GROUP BY and aggregation constructs. The input tuples are partitioned into groups by the first element (“keys”), and for each group the values in the second element are accumulated via the comb closure.

The input stream must have a TotalOrder guarantee, which means that the comb closure is allowed to depend on the order of elements in the stream.

If you need the accumulated value to have a different type than the input, use Stream::fold_keyed.

Example

let tick = process.tick();
let numbers = process.source_iter(q!(vec![(1, 2), (2, 3), (1, 3), (2, 4)]));
let batch = numbers.batch(&tick, nondet!(/** test */));
batch.reduce_keyed(q!(|acc, x| *acc += x)).all_ticks()
// (1, 5), (2, 7)

impl<'a, K, V, L, O, R> Stream<(K, V), Tick<L>, Bounded, O, R>

where K: Eq + Hash, L: Location<'a>,

pub fn fold_keyed_commutative_idempotent<A, I, F>( self, init: impl IntoQuotedMut<'a, I, Tick<L>>, comb: impl IntoQuotedMut<'a, F, Tick<L>>, ) -> Stream<(K, A), Tick<L>, Bounded, NoOrder, ExactlyOnce>

where I: Fn() -> A + 'a, F: Fn(&mut A, V) + 'a,

👎Deprecated: use .into_keyed().fold_commutative_idempotent(…) instead

A special case of Stream::fold_commutative_idempotent, in the spirit of SQL’s GROUP BY and aggregation constructs. The input tuples are partitioned into groups by the first element (“keys”), and for each group the values in the second element are accumulated via the comb closure.

The comb closure must be commutative, as the order of input items is not guaranteed, and idempotent, as there may be non-deterministic duplicates.

If the input and output value types are the same and do not require initialization then use Stream::reduce_keyed_commutative_idempotent.

Example

let tick = process.tick();
let numbers = process.source_iter(q!(vec![(1, false), (2, true), (1, false), (2, false)]));
let batch = numbers.batch(&tick, nondet!(/** test */));
batch
    .fold_keyed_commutative_idempotent(q!(|| false), q!(|acc, x| *acc |= x))
    .all_ticks()
// (1, false), (2, true)

pub fn keys(self) -> Stream<K, Tick<L>, Bounded, NoOrder, ExactlyOnce>

Given a stream of pairs (K, V), produces a new stream of unique keys K.

Example

let tick = process.tick();
let numbers = process.source_iter(q!(vec![(1, 2), (2, 3), (1, 3), (2, 4)]));
let batch = numbers.batch(&tick, nondet!(/** test */));
batch.keys().all_ticks()
// 1, 2

pub fn reduce_keyed_commutative_idempotent<F>( self, comb: impl IntoQuotedMut<'a, F, Tick<L>>, ) -> Stream<(K, V), Tick<L>, Bounded, NoOrder, ExactlyOnce>

where F: Fn(&mut V, V) + 'a,

👎Deprecated: use .into_keyed().reduce_commutative_idempotent(…) instead

A special case of Stream::reduce_commutative_idempotent, in the spirit of SQL’s GROUP BY and aggregation constructs. The input tuples are partitioned into groups by the first element (“keys”), and for each group the values in the second element are accumulated via the comb closure.

The comb closure must be commutative, as the order of input items is not guaranteed, and idempotent, as there may be non-deterministic duplicates.

If you need the accumulated value to have a different type than the input, use Stream::fold_keyed_commutative_idempotent.

Example

let tick = process.tick();
let numbers = process.source_iter(q!(vec![(1, false), (2, true), (1, false), (2, false)]));
let batch = numbers.batch(&tick, nondet!(/** test */));
batch
    .reduce_keyed_commutative_idempotent(q!(|acc, x| *acc |= x))
    .all_ticks()
// (1, false), (2, true)

impl<'a, K, V, L, O> Stream<(K, V), Tick<L>, Bounded, O, ExactlyOnce>

where K: Eq + Hash, L: Location<'a>,

pub fn fold_keyed_commutative<A, I, F>( self, init: impl IntoQuotedMut<'a, I, Tick<L>>, comb: impl IntoQuotedMut<'a, F, Tick<L>>, ) -> Stream<(K, A), Tick<L>, Bounded, NoOrder, ExactlyOnce>

where I: Fn() -> A + 'a, F: Fn(&mut A, V) + 'a,

👎Deprecated: use .into_keyed().fold_commutative(…) instead

A special case of Stream::fold_commutative, in the spirit of SQL’s GROUP BY and aggregation constructs. The input tuples are partitioned into groups by the first element (“keys”), and for each group the values in the second element are accumulated via the comb closure.

The comb closure must be commutative, as the order of input items is not guaranteed.

If the input and output value types are the same and do not require initialization then use Stream::reduce_keyed_commutative.

Example

let tick = process.tick();
let numbers = process.source_iter(q!(vec![(1, 2), (2, 3), (1, 3), (2, 4)]));
let batch = numbers.batch(&tick, nondet!(/** test */));
batch
    .fold_keyed_commutative(q!(|| 0), q!(|acc, x| *acc += x))
    .all_ticks()
// (1, 5), (2, 7)

pub fn reduce_keyed_commutative<F>( self, comb: impl IntoQuotedMut<'a, F, Tick<L>>, ) -> Stream<(K, V), Tick<L>, Bounded, NoOrder, ExactlyOnce>

where F: Fn(&mut V, V) + 'a,

👎Deprecated: use .into_keyed().reduce_commutative(…) instead

A special case of Stream::reduce_commutative, in the spirit of SQL’s GROUP BY and aggregation constructs. The input tuples are partitioned into groups by the first element (“keys”), and for each group the values in the second element are accumulated via the comb closure.

The comb closure must be commutative, as the order of input items is not guaranteed.

If you need the accumulated value to have a different type than the input, use Stream::fold_keyed_commutative.

Example

let tick = process.tick();
let numbers = process.source_iter(q!(vec![(1, 2), (2, 3), (1, 3), (2, 4)]));
let batch = numbers.batch(&tick, nondet!(/** test */));
batch
    .reduce_keyed_commutative(q!(|acc, x| *acc += x))
    .all_ticks()
// (1, 5), (2, 7)

impl<'a, K, V, L, R> Stream<(K, V), Tick<L>, Bounded, TotalOrder, R>

where K: Eq + Hash, L: Location<'a>,

pub fn fold_keyed_idempotent<A, I, F>( self, init: impl IntoQuotedMut<'a, I, Tick<L>>, comb: impl IntoQuotedMut<'a, F, Tick<L>>, ) -> Stream<(K, A), Tick<L>, Bounded, NoOrder, ExactlyOnce>

where I: Fn() -> A + 'a, F: Fn(&mut A, V) + 'a,

👎Deprecated: use .into_keyed().fold_idempotent(…) instead

A special case of Stream::fold_idempotent, in the spirit of SQL’s GROUP BY and aggregation constructs. The input tuples are partitioned into groups by the first element (“keys”), and for each group the values in the second element are accumulated via the comb closure.

The comb closure must be idempotent as there may be non-deterministic duplicates.

If the input and output value types are the same and do not require initialization then use Stream::reduce_keyed_idempotent.

Example

let tick = process.tick();
let numbers = process.source_iter(q!(vec![(1, false), (2, true), (1, false), (2, false)]));
let batch = numbers.batch(&tick, nondet!(/** test */));
batch
    .fold_keyed_idempotent(q!(|| false), q!(|acc, x| *acc |= x))
    .all_ticks()
// (1, false), (2, true)

pub fn reduce_keyed_idempotent<F>( self, comb: impl IntoQuotedMut<'a, F, Tick<L>>, ) -> Stream<(K, V), Tick<L>, Bounded, NoOrder, ExactlyOnce>

where F: Fn(&mut V, V) + 'a,

👎Deprecated: use .into_keyed().reduce_idempotent(…) instead

A special case of Stream::reduce_idempotent, in the spirit of SQL’s GROUP BY and aggregation constructs. The input tuples are partitioned into groups by the first element (“keys”), and for each group the values in the second element are accumulated via the comb closure.

The comb closure must be idempotent, as there may be non-deterministic duplicates.

If you need the accumulated value to have a different type than the input, use Stream::fold_keyed_idempotent.

Example

let tick = process.tick();
let numbers = process.source_iter(q!(vec![(1, false), (2, true), (1, false), (2, false)]));
let batch = numbers.batch(&tick, nondet!(/** test */));
batch
    .reduce_keyed_idempotent(q!(|acc, x| *acc |= x))
    .all_ticks()
// (1, false), (2, true)

impl<'a, T, L, B: Boundedness, O, R> Stream<T, Atomic<L>, B, O, R>

where L: Location<'a> + NoTick,

pub fn batch(self, _nondet: NonDet) -> Stream<T, Tick<L>, Bounded, O, R>

Returns a stream corresponding to the latest batch of elements being atomically processed. These batches are guaranteed to be contiguous across ticks and preserve the order of the input.

Non-Determinism

The batch boundaries are non-deterministic and may change across executions.

pub fn end_atomic(self) -> Stream<T, L, B, O, R>

pub fn atomic_source(&self) -> Tick<L>

impl<'a, T, L, B: Boundedness, O, R> Stream<T, L, B, O, R>

where L: Location<'a> + NoTick + NoAtomic,

pub fn atomic(self, tick: &Tick<L>) -> Stream<T, Atomic<L>, B, O, R>

pub fn resolve_futures<T2>(self) -> Stream<T2, L, B, NoOrder, R>

where T: Future<Output = T2>,

Consumes a stream of Future<T>, produces a new stream of the resulting T outputs. Future outputs are produced as available, regardless of input arrival order.

Example

process.source_iter(q!([2, 3, 1, 9, 6, 5, 4, 7, 8]))
    .map(q!(|x| async move {
        tokio::time::sleep(tokio::time::Duration::from_millis(10)).await;
        x
    }))
    .resolve_futures()
// 1, 2, 3, 4, 5, 6, 7, 8, 9 (in any order)

pub fn batch( self, tick: &Tick<L>, nondet: NonDet, ) -> Stream<T, Tick<L>, Bounded, O, R>

Given a tick, returns a stream corresponding to a batch of elements segmented by that tick. These batches are guaranteed to be contiguous across ticks and preserve the order of the input.

Non-Determinism

The batch boundaries are non-deterministic and may change across executions.

pub fn sample_every( self, interval: impl QuotedWithContext<'a, Duration, L> + Copy + 'a, nondet: NonDet, ) -> Stream<T, L, Unbounded, O, AtLeastOnce>

Given a time interval, returns a stream corresponding to samples taken from the stream roughly at that interval. The output will have elements in the same order as the input, but with arbitrary elements skipped between samples. There is also no guarantee on the exact timing of the samples.

Non-Determinism

The output stream is non-deterministic in which elements are sampled, since this is controlled by a clock.

pub fn timeout( self, duration: impl QuotedWithContext<'a, Duration, Tick<L>> + Copy + 'a, nondet: NonDet, ) -> Optional<(), L, Unbounded>

Given a timeout duration, returns an Optional which will have a value if the stream has not emitted a value since that duration.

Non-Determinism

Timeout relies on non-deterministic sampling of the stream, so depending on when samples take place, timeouts may be non-deterministically generated or missed, and the notification of the timeout may be delayed as well. There is also no guarantee on how long the Optional will have a value after the timeout is detected based on when the next sample is taken.

impl<'a, F, T, L, B: Boundedness, O, R> Stream<F, L, B, O, R>

where L: Location<'a> + NoTick + NoAtomic, F: Future<Output = T>,

pub fn resolve_futures_ordered(self) -> Stream<T, L, B, O, R>

Consumes a stream of Future<T>, produces a new stream of the resulting T outputs. Future outputs are produced in the same order as the input stream.

Example

process.source_iter(q!([2, 3, 1, 9, 6, 5, 4, 7, 8]))
    .map(q!(|x| async move {
        tokio::time::sleep(tokio::time::Duration::from_millis(10)).await;
        x
    }))
    .resolve_futures_ordered()
// 2, 3, 1, 9, 6, 5, 4, 7, 8

impl<'a, T, L, B: Boundedness, O, R> Stream<T, L, B, O, R>

where L: Location<'a> + NoTick,

pub fn for_each<F: Fn(T) + 'a>(self, f: impl IntoQuotedMut<'a, F, L>)

pub fn dest_sink<S>(self, sink: impl QuotedWithContext<'a, S, L>)

where S: 'a + Sink<T> + Unpin,

impl<'a, T, L, O, R> Stream<T, Tick<L>, Bounded, O, R>

where L: Location<'a>,

pub fn all_ticks(self) -> Stream<T, L, Unbounded, O, R>

pub fn all_ticks_atomic(self) -> Stream<T, Atomic<L>, Unbounded, O, R>

pub fn persist(self) -> Stream<T, Tick<L>, Bounded, O, R>

where T: Clone,

pub fn defer_tick(self) -> Stream<T, Tick<L>, Bounded, O, R>

pub fn delta(self) -> Stream<T, Tick<L>, Bounded, O, R>

Trait Implementations

impl<'a, T, L, B: Boundedness, O, R> Clone for Stream<T, L, B, O, R>

where T: Clone, L: Location<'a>,

fn clone(&self) -> Self

Returns a duplicate of the value.

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

Performs copy-assignment from source.

impl<'a, T, L, B: Boundedness, O, R> CycleCollection<'a, ForwardRefMarker> for Stream<T, L, B, O, R>

where L: Location<'a> + NoTick,

type Location = L

fn create_source(ident: Ident, location: L) -> Self

impl<'a, T, L, O, R> CycleCollection<'a, TickCycleMarker> for Stream<T, Tick<L>, Bounded, O, R>

where L: Location<'a>,

type Location = Tick<L>

fn create_source(ident: Ident, location: Tick<L>) -> Self

impl<'a, T, L, B: Boundedness, O, R> CycleComplete<'a, ForwardRefMarker> for Stream<T, L, B, O, R>

where L: Location<'a> + NoTick,

fn complete(self, ident: Ident, expected_location: LocationId)

impl<'a, T, L, O, R> CycleComplete<'a, TickCycleMarker> for Stream<T, Tick<L>, Bounded, O, R>

where L: Location<'a>,

fn complete(self, ident: Ident, expected_location: LocationId)

impl<'a, T, L, O, R> DeferTick for Stream<T, Tick<L>, Bounded, O, R>

where L: Location<'a>,

fn defer_tick(self) -> Self

impl<'a, T, L, B: Boundedness, O> From<Stream<T, L, B, O>> for Stream<T, L, B, O, AtLeastOnce>

where L: Location<'a>,

fn from( stream: Stream<T, L, B, O, ExactlyOnce>, ) -> Stream<T, L, B, O, AtLeastOnce>

Converts to this type from the input type.

impl<'a, T, L, B: Boundedness, R> From<Stream<T, L, B, TotalOrder, R>> for Stream<T, L, B, NoOrder, R>

where L: Location<'a>,

fn from(stream: Stream<T, L, B, TotalOrder, R>) -> Stream<T, L, B, NoOrder, R>

Converts to this type from the input type.

impl<'a, T, L, O, R> From<Stream<T, L, Bounded, O, R>> for Stream<T, L, Unbounded, O, R>

where L: Location<'a>,

fn from(stream: Stream<T, L, Bounded, O, R>) -> Stream<T, L, Unbounded, O, R>

Converts to this type from the input type.