Trait par_stream::TryIndexStreamExt

pub trait TryIndexStreamExt where
    Self: Stream<Item = Result<(usize, Self::Ok), Self::Error>>, {
    type Ok;
    type Error;
    fn try_reorder_enumerated(
        self
    ) -> TryReorderEnumerated<Self, Self::Ok, Self::Error>;
}

The trait extends TryStream types with ordering manipulation combinators.

Associated Types

type Ok

type Error

Required methods

fn try_reorder_enumerated(
    self
) -> TryReorderEnumerated<Self, Self::Ok, Self::Error>

Reorders the input items Ok((index, item)) according to the index number and returns item.

It can be combined with try_enumerate and unordered parallel tasks.

If an Err item is received, it stops receiving future items and flushes buffered values, and sends the Err in the end.

use futures::prelude::*;
use par_stream::prelude::*;

let result: Result<Vec<_>, _> = stream::iter(0..100)
    .map(|val| if val < 50 { Ok(val) } else { Err(val) })
    // add index number
    .try_enumerate()
    // double the values in parallel
    .try_par_then_unordered(None, move |(index, value)| {
        // the closure is sent to parallel worker
        async move { Ok((index, value * 2)) }
    })
    // add values by one in parallel
    .try_par_then_unordered(None, move |(index, value)| {
        // the closure is sent to parallel worker
        async move {
            let new_val = value + 1;
            if new_val < 50 {
                Ok((index, new_val))
            } else {
                Err(value)
            }
        }
    })
    // reorder the values according to index number
    .try_reorder_enumerated()
    .try_collect()
    .await;

Implementors

impl<S, T, E> TryIndexStreamExt for S where
    S: Stream<Item = Result<(usize, T), E>>, 

type Ok = T

type Error = E