Skip to main content

TryStreamTranspose

try_stream_transpose

Trait TryStreamTranspose 

Source 
pub trait TryStreamTranspose: TryStream {
    // Provided method
    fn transpose_with<Fut, F, T, E>(
        self,
        buf_size: usize,
        f: F,
    ) -> impl Future<Output = Result<T, E>>
       where F: FnMut(Receiver<Self::Ok>) -> Fut,
             Fut: Future<Output = Result<T, E>>,
             Self: Send + Sized + Stream<Item = Result<Self::Ok, Self::Error>> + 'static,
             Self::Ok: Send,
             Self::Error: Send,
             E: Send + From<Self::Error> + From<Error> + 'static { ... }
}
Expand description

Convenience for TryStream, when you want to work on a stream of Oks as a stream of the values inside the Oks.

Provided Methods§

Source

fn transpose_with<Fut, F, T, E>( self, buf_size: usize, f: F, ) -> impl Future<Output = Result<T, E>>
where F: FnMut(Receiver<Self::Ok>) -> Fut, Fut: Future<Output = Result<T, E>>, Self: Send + Sized + Stream<Item = Result<Self::Ok, Self::Error>> + 'static, Self::Ok: Send, Self::Error: Send, E: Send + From<Self::Error> + From<Error> + 'static,

Transposes a TryStream, in the sense that it passes a stream of its unwrapped items to the closure f.

If, while executing the closure, one of the items turns out an Err value, the in-closure stream stops yielding and this function returns said error. If neither the stream nor this function, while executing, produce an error, the closure’s result is returned.

Error types returned by this function must implement conversions from Error, in addition to the conversion from the items’s error type. Additionally, the buf_size parameter controls the size of the internal buffer.

use futures_util::{FutureExt, StreamExt, TryStreamExt, stream};
use try_stream_transpose::TryStreamTranspose;

let nums: [Result<i32, Error>; 4] = [Ok(4), Ok(5), Ok(6), Ok(7)];
core::assert_matches!(
    stream::iter(nums)
        .map_ok(|x| x + 1)
        .transpose_with(1024, |s| {
            s.fold(1, async |a, b| a * b).map(|x| Ok(x / 40))
        })
        .await,
    Ok::<i32, Error>(42)
);
let argh = [Ok(2), Err("3"), Ok(7)];
core::assert_matches!(
    stream::iter(argh)
        .map_ok(|x| x + 1)
        .transpose_with(1024, |s| {
            s.fold(1, async |a, b| a * b).map(|x| Ok(x / 8))
        })
        .await,
    Err(Error::Message("3"))
);

// The error type must support conversion from `Error`. Conversion from
// `&'static str` is implemented to support our code above.
#[derive(Debug, thiserror::Error)]
#[error(transparent)]
enum Error {
    #[error("{0}")]
    Message(&'static str),
    Internal(#[from] try_stream_transpose::Error),
}

impl From<&'static str> for Error {
    fn from(msg: &'static str) -> Self {
        Self::Message(msg)
    }
}

Dyn Compatibility§

This trait is dyn compatible.

In older versions of Rust, dyn compatibility was called "object safety".

Implementors§