Struct hydro_lang::stream::Stream

pub struct Stream<T, L, B, Order = TotalOrder> { /* private fields */ }

An ordered sequence stream of elements of type T.

Type Parameters:

• T: the type of elements in the stream
• L: the location where the stream is being materialized
• B: 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: Location<'a>, B, Order> Stream<T, L, B, Order>

pub fn map<U, F: Fn(T) -> U + 'a>( self, f: impl IntoQuotedMut<'a, F, L>, ) -> Stream<U, L, B, Order>

Produces a stream based on invoking f on each element in order. 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: IntoIterator<Item = U>, F: Fn(T) -> I + 'a>( self, f: impl IntoQuotedMut<'a, F, L>, ) -> Stream<U, L, B, Order>

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: IntoIterator<Item = U>, F: Fn(T) -> I + 'a>( self, f: impl IntoQuotedMut<'a, F, L>, ) -> Stream<U, L, B, NoOrder>

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, Order>

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>

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: Fn(&T) -> bool + 'a>( self, f: impl IntoQuotedMut<'a, F, L>, ) -> Stream<T, L, B, Order>

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: Fn(T) -> Option<U> + 'a>( self, f: impl IntoQuotedMut<'a, F, L>, ) -> Stream<U, L, B, Order>

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<O>( self, other: impl Into<Optional<O, L, Bounded>>, ) -> Stream<(T, O), L, B, Order>

where O: 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 = unsafe {
    process
        .source_iter(q!(vec![1, 2, 3, 4]))
        .tick_batch(&tick)
};
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, Order>

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, Order>

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

pub fn cross_product<O>( self, other: Stream<O, L, B, Order>, ) -> Stream<(T, O), L, B, NoOrder>

where T: Clone, O: 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, Order>

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>, ) -> Stream<T, L, Bounded, Order>

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 = unsafe {
   process
   .source_iter(q!(vec![ 1, 2, 3, 4 ]))
   .tick_batch(&tick)
};
let batch = unsafe {
    process
        .source_iter(q!(vec![1, 2]))
        .tick_batch(&tick)
};
stream.filter_not_in(batch).all_ticks()

pub fn inspect<F: Fn(&T) + 'a>( self, f: impl IntoQuotedMut<'a, F, L>, ) -> Stream<T, L, B, Order>

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 unsafe fn assume_ordering<O>(self) -> Stream<T, L, B, O>

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.

Safety

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.

Example

TODO: more sensible code after Shadaj merges

let nums = process.source_iter(q!({
    let now = std::time::SystemTime::now();
    match now.elapsed().unwrap().as_secs() % 2 {
        0 => vec![5, 4, 3, 2, 1],
        _ => vec![1, 2, 3, 4, 5],
    }
    .into_iter()
}));
// despite being generated by `source_iter`, the order of `nums` across runs is non-deterministic
let stream = unsafe { nums.assume_ordering::<NoOrder>() };
stream

impl<'a, T, L: Location<'a>, B, Order> Stream<&T, L, B, Order>

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

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: Location<'a>, B, Order> Stream<T, L, B, Order>

where Order: MinOrder<NoOrder, Min = NoOrder>,

pub fn fold_commutative<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 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 = unsafe { numbers.tick_batch(&tick) };
batch
    .fold_commutative(q!(|| 0), q!(|acc, x| *acc += x))
    .all_ticks()
// 10

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

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 = unsafe { numbers.tick_batch(&tick) };
batch
    .reduce_commutative(q!(|curr, new| *curr += new))
    .all_ticks()
// 10

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 = unsafe { numbers.tick_batch(&tick) };
batch.max().all_ticks()
// 4

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

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 = unsafe { numbers.tick_batch(&tick) };
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 = unsafe { numbers.tick_batch(&tick) };
batch.min().all_ticks()
// 1

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 = unsafe { numbers.tick_batch(&tick) };
batch.count().all_ticks()
// 4

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

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

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 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 = unsafe { numbers.tick_batch(&tick) };
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 = unsafe { numbers.tick_batch(&tick) };
batch.last().all_ticks()
// 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 = unsafe { words.tick_batch(&tick) };
batch
    .fold(q!(|| String::new()), q!(|acc, x| acc.push_str(x)))
    .all_ticks()
// "HELLOWORLD"

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 = unsafe { words.tick_batch(&tick) };
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> Stream<T, L, Unbounded, O>

pub fn union<O2>( self, other: Stream<T, L, Unbounded, O2>, ) -> Stream<T, L, Unbounded, NoOrder>

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)).union(numbers)
// 2, 3, 4, 5, and 1, 2, 3, 4 interleaved in unknown order

impl<'a, T, L: Location<'a>, Order> Stream<T, L, Bounded, Order>

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

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 = unsafe { numbers.tick_batch(&tick) };
batch.sort().all_ticks()
// 1, 2, 3, 4

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

where Order: 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 = unsafe { numbers.tick_batch(&tick) };
batch.clone().map(q!(|x| x + 1)).chain(batch).all_ticks()
// 2, 3, 4, 5, 1, 2, 3, 4

impl<'a, K, V1, L: Location<'a>, B, Order> Stream<(K, V1), L, B, Order>

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

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>( self, n: Stream<K, L, Bounded, O2>, ) -> Stream<(K, V1), L, B, Order>

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 = unsafe {
   process
   .source_iter(q!(vec![ (1, 'a'), (2, 'b'), (3, 'c'), (4, 'd') ]))
   .tick_batch(&tick)
};
let batch = unsafe {
    process
        .source_iter(q!(vec![1, 2]))
        .tick_batch(&tick)
};
stream.anti_join(batch).all_ticks()

impl<'a, K: Eq + Hash, V, L: Location<'a>> Stream<(K, V), Tick<L>, Bounded>

pub fn fold_keyed<A, I: Fn() -> A + 'a, F: Fn(&mut A, V) + 'a>( self, init: impl IntoQuotedMut<'a, I, Tick<L>>, comb: impl IntoQuotedMut<'a, F, Tick<L>>, ) -> Stream<(K, A), Tick<L>, Bounded>

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 = unsafe { numbers.tick_batch(&tick) };
batch
    .fold_keyed(q!(|| 0), q!(|acc, x| *acc += x))
    .all_ticks()
// (1, 5), (2, 7)

pub fn reduce_keyed<F: Fn(&mut V, V) + 'a>( self, comb: impl IntoQuotedMut<'a, F, Tick<L>>, ) -> Stream<(K, V), Tick<L>, Bounded>

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 = unsafe { numbers.tick_batch(&tick) };
batch.reduce_keyed(q!(|acc, x| *acc += x)).all_ticks()
// (1, 5), (2, 7)

impl<'a, K: Eq + Hash, V, L: Location<'a>, Order> Stream<(K, V), Tick<L>, Bounded, Order>

pub fn fold_keyed_commutative<A, I: Fn() -> A + 'a, F: Fn(&mut A, V) + 'a>( self, init: impl IntoQuotedMut<'a, I, Tick<L>>, comb: impl IntoQuotedMut<'a, F, Tick<L>>, ) -> Stream<(K, A), Tick<L>, Bounded, Order>

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 = unsafe { numbers.tick_batch(&tick) };
batch
    .fold_keyed_commutative(q!(|| 0), q!(|acc, x| *acc += x))
    .all_ticks()
// (1, 5), (2, 7)

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

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 = unsafe { numbers.tick_batch(&tick) };
batch.keys().all_ticks()
// 1, 2

pub fn reduce_keyed_commutative<F: Fn(&mut V, V) + 'a>( self, comb: impl IntoQuotedMut<'a, F, Tick<L>>, ) -> Stream<(K, V), Tick<L>, Bounded, Order>

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 = unsafe { numbers.tick_batch(&tick) };
batch
    .reduce_keyed_commutative(q!(|acc, x| *acc += x))
    .all_ticks()
// (1, 5), (2, 7)

impl<'a, T, L: Location<'a> + NoTick, B, Order> Stream<T, Atomic<L>, B, Order>

pub unsafe fn tick_batch(self) -> Stream<T, Tick<L>, Bounded, Order>

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.

Safety

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

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

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

impl<'a, T, L: Location<'a> + NoTick + NoAtomic, B, Order> Stream<T, L, B, Order>

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

pub unsafe fn tick_batch( self, tick: &Tick<L>, ) -> Stream<T, Tick<L>, Bounded, Order>

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.

Safety

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

pub unsafe fn sample_every( self, interval: impl QuotedWithContext<'a, Duration, L> + Copy + 'a, ) -> Stream<T, L, Unbounded, Order>

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.

Safety

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

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

where Order: MinOrder<NoOrder, Min = NoOrder>,

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

Safety

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, T, L: Location<'a> + NoTick, B, Order> Stream<T, L, B, Order>

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

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

impl<'a, T, L: Location<'a>, Order> Stream<T, Tick<L>, Bounded, Order>

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

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

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

where T: Clone,

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

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

impl<'a, T, L: Location<'a> + NoTick, B, Order> Stream<T, L, B, Order>

pub fn send_bincode<L2: Location<'a>, CoreType>( self, other: &L2, ) -> Stream<<L::Root as CanSend<'a, L2>>::Out<CoreType>, L2, Unbounded, Order::Min>

where L::Root: CanSend<'a, L2, In<CoreType> = T>, CoreType: Serialize + DeserializeOwned, Order: MinOrder<<L::Root as CanSend<'a, L2>>::OutStrongestOrder<Order>>,

pub fn send_bincode_external<L2: 'a, CoreType>( self, other: &ExternalProcess<'_, L2>, ) -> ExternalBincodeStream<L::Out<CoreType>>

where L: CanSend<'a, ExternalProcess<'a, L2>, In<CoreType> = T, Out<CoreType> = CoreType>, CoreType: Serialize + DeserializeOwned,

pub fn send_bytes<L2: Location<'a>>( self, other: &L2, ) -> Stream<<L::Root as CanSend<'a, L2>>::Out<Bytes>, L2, Unbounded, Order::Min>

where L::Root: CanSend<'a, L2, In<Bytes> = T>, Order: MinOrder<<L::Root as CanSend<'a, L2>>::OutStrongestOrder<Order>>,

pub fn send_bytes_external<L2: 'a>( self, other: &ExternalProcess<'_, L2>, ) -> ExternalBytesPort

where L::Root: CanSend<'a, ExternalProcess<'a, L2>, In<Bytes> = T, Out<Bytes> = Bytes>,

pub fn send_bincode_anonymous<L2: Location<'a>, Tag, CoreType>( self, other: &L2, ) -> Stream<CoreType, L2, Unbounded, Order::Min>

where L::Root: CanSend<'a, L2, In<CoreType> = T, Out<CoreType> = (Tag, CoreType)>, CoreType: Serialize + DeserializeOwned, Order: MinOrder<<L::Root as CanSend<'a, L2>>::OutStrongestOrder<Order>>,

pub fn send_bytes_interleaved<L2: Location<'a>, Tag>( self, other: &L2, ) -> Stream<Bytes, L2, Unbounded, Order::Min>

where L::Root: CanSend<'a, L2, In<Bytes> = T, Out<Bytes> = (Tag, Bytes)>, Order: MinOrder<<L::Root as CanSend<'a, L2>>::OutStrongestOrder<Order>>,

pub fn broadcast_bincode<C2: 'a>( self, other: &Cluster<'a, C2>, ) -> Stream<<L::Root as CanSend<'a, Cluster<'a, C2>>>::Out<T>, Cluster<'a, C2>, Unbounded, Order::Min>

where L::Root: CanSend<'a, Cluster<'a, C2>, In<T> = (ClusterId<C2>, T)>, T: Clone + Serialize + DeserializeOwned, Order: MinOrder<<L::Root as CanSend<'a, Cluster<'a, C2>>>::OutStrongestOrder<Order>>,

pub fn broadcast_bincode_interleaved<C2: 'a, Tag>( self, other: &Cluster<'a, C2>, ) -> Stream<T, Cluster<'a, C2>, Unbounded, Order::Min>

where L::Root: CanSend<'a, Cluster<'a, C2>, In<T> = (ClusterId<C2>, T), Out<T> = (Tag, T)> + 'a, T: Clone + Serialize + DeserializeOwned, Order: MinOrder<<L::Root as CanSend<'a, Cluster<'a, C2>>>::OutStrongestOrder<Order>>,

pub fn broadcast_bytes<C2: 'a>( self, other: &Cluster<'a, C2>, ) -> Stream<<L::Root as CanSend<'a, Cluster<'a, C2>>>::Out<Bytes>, Cluster<'a, C2>, Unbounded, Order::Min>

where L::Root: CanSend<'a, Cluster<'a, C2>, In<Bytes> = (ClusterId<C2>, T)> + 'a, T: Clone, Order: MinOrder<<L::Root as CanSend<'a, Cluster<'a, C2>>>::OutStrongestOrder<Order>>,

pub fn broadcast_bytes_interleaved<C2: 'a, Tag>( self, other: &Cluster<'a, C2>, ) -> Stream<Bytes, Cluster<'a, C2>, Unbounded, Order::Min>

where L::Root: CanSend<'a, Cluster<'a, C2>, In<Bytes> = (ClusterId<C2>, T), Out<Bytes> = (Tag, Bytes)> + 'a, T: Clone, Order: MinOrder<<L::Root as CanSend<'a, Cluster<'a, C2>>>::OutStrongestOrder<Order>>,

impl<'a, T, L: Location<'a> + NoTick, B> Stream<T, L, B, TotalOrder>

pub fn round_robin_bincode<C2: 'a>( self, other: &Cluster<'a, C2>, ) -> Stream<<L::Root as CanSend<'a, Cluster<'a, C2>>>::Out<T>, Cluster<'a, C2>, Unbounded, <TotalOrder as MinOrder<<L::Root as CanSend<'a, Cluster<'a, C2>>>::OutStrongestOrder<TotalOrder>>>::Min>

where L::Root: CanSend<'a, Cluster<'a, C2>, In<T> = (ClusterId<C2>, T)>, T: Clone + Serialize + DeserializeOwned, TotalOrder: MinOrder<<L::Root as CanSend<'a, Cluster<'a, C2>>>::OutStrongestOrder<TotalOrder>>,

pub fn round_robin_bincode_interleaved<C2: 'a, Tag>( self, other: &Cluster<'a, C2>, ) -> Stream<T, Cluster<'a, C2>, Unbounded, <TotalOrder as MinOrder<<L::Root as CanSend<'a, Cluster<'a, C2>>>::OutStrongestOrder<TotalOrder>>>::Min>

where L::Root: CanSend<'a, Cluster<'a, C2>, In<T> = (ClusterId<C2>, T), Out<T> = (Tag, T)> + 'a, T: Clone + Serialize + DeserializeOwned, TotalOrder: MinOrder<<L::Root as CanSend<'a, Cluster<'a, C2>>>::OutStrongestOrder<TotalOrder>>,

pub fn round_robin_bytes<C2: 'a>( self, other: &Cluster<'a, C2>, ) -> Stream<<L::Root as CanSend<'a, Cluster<'a, C2>>>::Out<Bytes>, Cluster<'a, C2>, Unbounded, <TotalOrder as MinOrder<<L::Root as CanSend<'a, Cluster<'a, C2>>>::OutStrongestOrder<TotalOrder>>>::Min>

where L::Root: CanSend<'a, Cluster<'a, C2>, In<Bytes> = (ClusterId<C2>, T)> + 'a, T: Clone, TotalOrder: MinOrder<<L::Root as CanSend<'a, Cluster<'a, C2>>>::OutStrongestOrder<TotalOrder>>,

pub fn round_robin_bytes_interleaved<C2: 'a, Tag>( self, other: &Cluster<'a, C2>, ) -> Stream<Bytes, Cluster<'a, C2>, Unbounded, <TotalOrder as MinOrder<<L::Root as CanSend<'a, Cluster<'a, C2>>>::OutStrongestOrder<TotalOrder>>>::Min>

where L::Root: CanSend<'a, Cluster<'a, C2>, In<Bytes> = (ClusterId<C2>, T), Out<Bytes> = (Tag, Bytes)> + 'a, T: Clone, TotalOrder: MinOrder<<L::Root as CanSend<'a, Cluster<'a, C2>>>::OutStrongestOrder<TotalOrder>>,

Trait Implementations

impl<'a, T: Clone, L: Location<'a>, B, Order> Clone for Stream<T, L, B, Order>

fn clone(&self) -> Self

Returns a copy of the value.

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

Performs copy-assignment from source.

impl<'a, T, L: Location<'a> + NoTick, B, Order> CycleCollection<'a, ForwardRefMarker> for Stream<T, L, B, Order>

type Location = L

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

impl<'a, T, L: Location<'a>, Order> CycleCollection<'a, TickCycleMarker> for Stream<T, Tick<L>, Bounded, Order>

type Location = Tick<L>

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

impl<'a, T, L: Location<'a> + NoTick, B, Order> CycleComplete<'a, ForwardRefMarker> for Stream<T, L, B, Order>

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

impl<'a, T, L: Location<'a>, Order> CycleComplete<'a, TickCycleMarker> for Stream<T, Tick<L>, Bounded, Order>

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

impl<'a, T, L: Location<'a>, Order> DeferTick for Stream<T, Tick<L>, Bounded, Order>

fn defer_tick(self) -> Self

impl<'a, T, L: Location<'a>, B> From<Stream<T, L, B>> for Stream<T, L, B, NoOrder>

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

Converts to this type from the input type.

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

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

Converts to this type from the input type.