Struct IndexedStream

pub struct IndexedStream<'a, T> { /* private fields */ }

A Stream optimized for parallel processing of ordered data.

The optimization is achieved by associating each item in the stream with its order. This can be leveraged by consumers of IndexedStreams to reason about the original ordering of the items as they become available. This can obviate the need to drive a stream to completion before operating on the results.

From an IntoIterator

A From implementation is provided for any IntoIterator, allowing for easy conversion from an arbitrary collection into an IndexedStream.

Example

let stream = IndexedStream::from(vec![1, 2, 3]);

As a Stream

A Stream implementation is also provided for IndexedStream, such that can be consumed as a normal stream.

Example

let mut stream = IndexedStream::from(vec![1, 2, 3]);
while let Some(Ok((idx, item))) = stream.next().await {
    println!("{}: {}", idx, item);
}

Directive, Functor, and Foldable

IndexedStream implements Directive, Functor, and Foldable, allowing it to be used directly to form a Directive chain. Note that because the stream is indexed, the Foldable implementation satisfies associativity of combination while folding in parallel.

Example

struct Multiply;
impl Monoid for Multiply {
    type Elem = i32;

    fn combine(&self, a: i32, b: i32) -> Result<i32> {
        Ok(a * b)
    }

    fn empty(&self) -> i32 {
        1
    }
}

struct MultiplyBy(i32);
impl Operation for MultiplyBy {
    type Input = i32;
    type Output = i32;

    fn execute(&self, input: i32) -> Result<i32> {
        Ok(self.0 * input)
    }
}

let computation = IndexedStream::from([1, 2, 3, 4, 5])
    .map(&MultiplyBy(2))
    .fold(&Multiply);
let result = computation.run(&runtime).await?;
assert_eq!(result, 3840);

Implementations

impl<'a, T> IndexedStream<'a, T>

pub fn new(inner: impl Stream<Item = Result<(usize, T)>> + Send + 'a) -> Self

Create a new IndexedStream.

pub async fn into_values_sorted(self) -> Result<impl IntoIterator<Item = T>>

Convert this IndexedStream into an IntoIterator of values, sorted by their index.

Note that this will drive the stream to completion.

Example

let stream = IndexedStream::from(vec![1, 2, 3]);
let values = stream.into_values_sorted().await?
    .into_iter()
    .collect::<Vec<_>>();
assert_eq!(vec![1, 2, 3], values);

Trait Implementations

impl<'a, T: Send> Directive for IndexedStream<'a, T>

type Input = IndexedStream<'a, T>

type Output = IndexedStream<'a, T>

fn run<'life0, 'async_trait>( self, _: &'life0 Runtime, ) -> Pin<Box<dyn Future<Output = Result<Self::Output>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait,

Run this Directive on the given Runtime.

fn map<'a, Op, F>(self, op: &'a Op) -> Map<'a, Op, Self>

where Op: Operation, F: Functor<'a, Op::Output, A = Op::Input>, Self: Sized + Directive<Output = F>,

Map this Directive’s output over the given Operation.

fn fold<'a, M, F>(self, m: &'a M) -> Fold<'a, M, Self>

where M: Monoid, F: Foldable<'a, M::Elem, A = M::Elem>, Self: Sized + Directive<Output = F>,

Fold this Directive over the given Monoid until a single value is produced.

impl<'a, A: Send + Sync + 'a, B: Send + 'a> Foldable<'a, B> for IndexedStream<'a, A>

fn f_fold<'life0, 'async_trait, M>( self, m: &'a M, runtime: &'life0 Runtime, ) -> Pin<Box<dyn Future<Output = Result<A>> + Send + 'async_trait>>

where M: 'static + 'async_trait + Monoid<Elem = A>, Self: 'async_trait, 'a: 'async_trait, 'life0: 'async_trait,

impl<'a, T: 'a, I: IntoIterator<Item = T>> From<I> for IndexedStream<'a, T>

where <I as IntoIterator>::IntoIter: Send + Sync + 'a,

Create an IndexedStream from an IntoIterator.

fn from(iter: I) -> Self

Converts to this type from the input type.

impl<'a, A: Send + Sync + 'a, B: Send + 'a> Functor<'a, B> for IndexedStream<'a, A>

fn f_map<'life0, 'async_trait, Op>( self, op: &'a Op, runtime: &'life0 Runtime, ) -> Pin<Box<dyn Future<Output = Result<Self::Target>> + Send + 'async_trait>>

where Op: Operation<Input = A, Output = B> + 'async_trait + Operation, Self: 'async_trait, 'a: 'async_trait, 'life0: 'async_trait,

impl<'a, T: 'a, U: 'a> HKT<U> for IndexedStream<'a, T>

type A = T

type Target = IndexedStream<'a, U>

impl<'a, T> Stream for IndexedStream<'a, T>

Stream implementation for IndexedStream.

This is a passthrough implementation, which simply delegates to the inner stream.

type Item = Result<(usize, T), Error>

Values yielded by the stream.

fn poll_next( self: Pin<&mut Self>, cx: &mut Context<'_>, ) -> Poll<Option<Self::Item>>

Attempt to pull out the next value of this stream, registering the current task for wakeup if the value is not yet available, and returning None if the stream is exhausted.

fn size_hint(&self) -> (usize, Option<usize>)

Returns the bounds on the remaining length of the stream.