Trait futures_lite::stream::StreamExt

pub trait StreamExt: Stream {
    fn poll_next(&mut self, cx: &mut Context<'_>) -> Poll<Option<Self::Item>>
    where
        Self: Unpin,
    { ... }
    fn next(&mut self) -> NextFuture<'_, Self>
    where
        Self: Unpin,
    { ... }
    fn try_next<T, E>(&mut self) -> TryNextFuture<'_, Self>
    where
        Self: Stream<Item = Result<T, E>> + Unpin,
    { ... }
    fn count(self) -> CountFuture<Self>
    where
        Self: Sized,
    { ... }
    fn map<T, F>(self, f: F) -> Map<Self, F>
    where
        Self: Sized,
        F: FnMut(Self::Item) -> T,
    { ... }
    fn flat_map<U, F>(self, f: F) -> FlatMap<Self, U, F>
    where
        Self: Sized,
        U: Stream,
        F: FnMut(Self::Item) -> U,
    { ... }
    fn flatten(self) -> Flatten<Self>
    where
        Self: Sized,
        Self::Item: Stream,
    { ... }
    fn then<F, Fut>(self, f: F) -> Then<Self, F, Fut>
    where
        Self: Sized,
        F: FnMut(Self::Item) -> Fut,
        Fut: Future,
    { ... }
    fn filter<P>(self, predicate: P) -> Filter<Self, P>
    where
        Self: Sized,
        P: FnMut(&Self::Item) -> bool,
    { ... }
    fn filter_map<T, F>(self, f: F) -> FilterMap<Self, F>
    where
        Self: Sized,
        F: FnMut(Self::Item) -> Option<T>,
    { ... }
    fn take(self, n: usize) -> Take<Self>
    where
        Self: Sized,
    { ... }
    fn take_while<P>(self, predicate: P) -> TakeWhile<Self, P>
    where
        Self: Sized,
        P: FnMut(&Self::Item) -> bool,
    { ... }
    fn skip(self, n: usize) -> Skip<Self>
    where
        Self: Sized,
    { ... }
    fn skip_while<P>(self, predicate: P) -> SkipWhile<Self, P>
    where
        Self: Sized,
        P: FnMut(&Self::Item) -> bool,
    { ... }
    fn step_by(self, step: usize) -> StepBy<Self>
    where
        Self: Sized,
    { ... }
    fn chain<U>(self, other: U) -> Chain<Self, U>
    where
        Self: Sized,
        U: Stream<Item = Self::Item> + Sized,
    { ... }
    fn cloned<'a, T>(self) -> Cloned<Self>
    where
        Self: Stream<Item = &'a T> + Sized,
        T: Clone + 'a,
    { ... }
    fn copied<'a, T>(self) -> Copied<Self>
    where
        Self: Stream<Item = &'a T> + Sized,
        T: Copy + 'a,
    { ... }
    fn collect<C>(self) -> CollectFuture<Self, C>
    where
        Self: Sized,
        C: Default + Extend<Self::Item>,
    { ... }
    fn try_collect<T, E, C>(self) -> TryCollectFuture<Self, C>
    where
        Self: Stream<Item = Result<T, E>> + Sized,
        C: Default + Extend<T>,
    { ... }
    fn partition<B, P>(self, predicate: P) -> PartitionFuture<Self, P, B>
    where
        Self: Sized,
        B: Default + Extend<Self::Item>,
        P: FnMut(&Self::Item) -> bool,
    { ... }
    fn fold<T, F>(self, init: T, f: F) -> FoldFuture<Self, F, T>
    where
        Self: Sized,
        F: FnMut(T, Self::Item) -> T,
    { ... }
    fn try_fold<T, E, F, B>(
        &mut self, 
        init: B, 
        f: F
    ) -> TryFoldFuture<'_, Self, F, B>
    where
        Self: Stream<Item = Result<T, E>> + Unpin + Sized,
        F: FnMut(B, T) -> Result<B, E>,
    { ... }
    fn scan<St, B, F>(self, initial_state: St, f: F) -> Scan<Self, St, F>
    where
        Self: Sized,
        F: FnMut(&mut St, Self::Item) -> Option<B>,
    { ... }
    fn fuse(self) -> Fuse<Self>
    where
        Self: Sized,
    { ... }
    fn cycle(self) -> Cycle<Self>
    where
        Self: Clone + Sized,
    { ... }
    fn enumerate(self) -> Enumerate<Self>
    where
        Self: Sized,
    { ... }
    fn inspect<F>(self, f: F) -> Inspect<Self, F>
    where
        Self: Sized,
        F: FnMut(&Self::Item),
    { ... }
    fn nth(&mut self, n: usize) -> NthFuture<'_, Self>
    where
        Self: Unpin,
    { ... }
    fn last(self) -> LastFuture<Self>
    where
        Self: Sized,
    { ... }
    fn find<P>(&mut self, predicate: P) -> FindFuture<'_, Self, P>
    where
        Self: Unpin,
        P: FnMut(&Self::Item) -> bool,
    { ... }
    fn find_map<F, B>(&mut self, f: F) -> FindMapFuture<'_, Self, F>
    where
        Self: Unpin,
        F: FnMut(Self::Item) -> Option<B>,
    { ... }
    fn position<P>(&mut self, predicate: P) -> PositionFuture<'_, Self, P>
    where
        Self: Unpin,
        P: FnMut(Self::Item) -> bool,
    { ... }
    fn all<P>(&mut self, predicate: P) -> AllFuture<'_, Self, P>
    where
        Self: Unpin,
        P: FnMut(Self::Item) -> bool,
    { ... }
    fn any<P>(&mut self, predicate: P) -> AnyFuture<'_, Self, P>
    where
        Self: Unpin,
        P: FnMut(Self::Item) -> bool,
    { ... }
    fn for_each<F>(self, f: F) -> ForEachFuture<Self, F>
    where
        Self: Sized,
        F: FnMut(Self::Item),
    { ... }
    fn try_for_each<F, E>(&mut self, f: F) -> TryForEachFuture<'_, Self, F>
    where
        Self: Unpin,
        F: FnMut(Self::Item) -> Result<(), E>,
    { ... }
    fn zip<U>(self, other: U) -> Zip<Self, U>
    where
        Self: Sized,
        U: Stream,
    { ... }
    fn unzip<A, B, FromA, FromB>(self) -> UnzipFuture<Self, FromA, FromB>
    where
        FromA: Default + Extend<A>,
        FromB: Default + Extend<B>,
        Self: Stream<Item = (A, B)> + Sized,
    { ... }
    fn or<S>(self, other: S) -> Or<Self, S>
    where
        Self: Sized,
        S: Stream<Item = Self::Item>,
    { ... }
    fn race<S>(self, other: S) -> Race<Self, S>
    where
        Self: Sized,
        S: Stream<Item = Self::Item>,
    { ... }
    fn boxed<'a>(self) -> Pin<Box<dyn Stream<Item = Self::Item> + Send + 'a>>
    where
        Self: Send + Sized + 'a,
    { ... }
    fn boxed_local<'a>(self) -> Pin<Box<dyn Stream<Item = Self::Item> + 'a>>
    where
        Self: Sized + 'a,
    { ... }
}

Extension trait for Stream.

Provided methods

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

A convenience for calling Stream::poll_next() on !Unpin types.

fn next(&mut self) -> NextFuture<'_, Self> where
    Self: Unpin, 

Retrieves the next item in the stream.

Returns None when iteration is finished. Stream implementations may choose to or not to resume iteration after that.

Examples

use futures_lite::stream::{self, StreamExt};

let mut s = stream::iter(1..=3);

assert_eq!(s.next().await, Some(1));
assert_eq!(s.next().await, Some(2));
assert_eq!(s.next().await, Some(3));
assert_eq!(s.next().await, None);

fn try_next<T, E>(&mut self) -> TryNextFuture<'_, Self> where
    Self: Stream<Item = Result<T, E>> + Unpin, 

Retrieves the next item in the stream.

This is similar to the next() method, but returns Result<Option<T>, E> rather than Option<Result<T, E>>.

Note that s.try_next().await is equivalent to s.next().await.transpose().

Examples

use futures_lite::stream::{self, StreamExt};

let mut s = stream::iter(vec![Ok(1), Ok(2), Err("error")]);

assert_eq!(s.try_next().await, Ok(Some(1)));
assert_eq!(s.try_next().await, Ok(Some(2)));
assert_eq!(s.try_next().await, Err("error"));
assert_eq!(s.try_next().await, Ok(None));

fn count(self) -> CountFuture<Self> where
    Self: Sized, 

Counts the number of items in the stream.

Examples

use futures_lite::stream::{self, StreamExt};

let s1 = stream::iter(vec![0]);
let s2 = stream::iter(vec![1, 2, 3]);

assert_eq!(s1.count().await, 1);
assert_eq!(s2.count().await, 3);

fn map<T, F>(self, f: F) -> Map<Self, F> where
    Self: Sized,
    F: FnMut(Self::Item) -> T, 

Maps items of the stream to new values using a closure.

Examples

use futures_lite::stream::{self, StreamExt};

let s = stream::iter(vec![1, 2, 3]);
let mut s = s.map(|x| 2 * x);

assert_eq!(s.next().await, Some(2));
assert_eq!(s.next().await, Some(4));
assert_eq!(s.next().await, Some(6));
assert_eq!(s.next().await, None);

fn flat_map<U, F>(self, f: F) -> FlatMap<Self, U, F> where
    Self: Sized,
    U: Stream,
    F: FnMut(Self::Item) -> U, 

Maps items to streams and then concatenates them.

Examples

use futures_lite::stream::{self, StreamExt};

let words = stream::iter(vec!["one", "two"]);

let s: String = words
    .flat_map(|s| stream::iter(s.chars()))
    .collect()
    .await;

assert_eq!(s, "onetwo");

fn flatten(self) -> Flatten<Self> where
    Self: Sized,
    Self::Item: Stream, 

Concatenates inner streams.

Examples

use futures_lite::stream::{self, StreamExt};

let s1 = stream::iter(vec![1, 2, 3]);
let s2 = stream::iter(vec![4, 5]);

let s = stream::iter(vec![s1, s2]);
let v: Vec<_> = s.flatten().collect().await;
assert_eq!(v, [1, 2, 3, 4, 5]);

fn then<F, Fut>(self, f: F) -> Then<Self, F, Fut> where
    Self: Sized,
    F: FnMut(Self::Item) -> Fut,
    Fut: Future, 

Maps items of the stream to new values using an async closure.

Examples

use futures_lite::pin;
use futures_lite::stream::{self, StreamExt};

let s = stream::iter(vec![1, 2, 3]);
let mut s = s.then(|x| async move { 2 * x });

pin!(s);
assert_eq!(s.next().await, Some(2));
assert_eq!(s.next().await, Some(4));
assert_eq!(s.next().await, Some(6));
assert_eq!(s.next().await, None);

fn filter<P>(self, predicate: P) -> Filter<Self, P> where
    Self: Sized,
    P: FnMut(&Self::Item) -> bool, 

Keeps items of the stream for which predicate returns true.

Examples

use futures_lite::stream::{self, StreamExt};

let s = stream::iter(vec![1, 2, 3, 4]);
let mut s = s.filter(|i| i % 2 == 0);

assert_eq!(s.next().await, Some(2));
assert_eq!(s.next().await, Some(4));
assert_eq!(s.next().await, None);

fn filter_map<T, F>(self, f: F) -> FilterMap<Self, F> where
    Self: Sized,
    F: FnMut(Self::Item) -> Option<T>, 

Filters and maps items of the stream using a closure.

Examples

use futures_lite::stream::{self, StreamExt};

let s = stream::iter(vec!["1", "lol", "3", "NaN", "5"]);
let mut s = s.filter_map(|a| a.parse::<u32>().ok());

assert_eq!(s.next().await, Some(1));
assert_eq!(s.next().await, Some(3));
assert_eq!(s.next().await, Some(5));
assert_eq!(s.next().await, None);

fn take(self, n: usize) -> Take<Self> where
    Self: Sized, 

Takes only the first n items of the stream.

Examples

use futures_lite::stream::{self, StreamExt};

let mut s = stream::repeat(7).take(2);

assert_eq!(s.next().await, Some(7));
assert_eq!(s.next().await, Some(7));
assert_eq!(s.next().await, None);

fn take_while<P>(self, predicate: P) -> TakeWhile<Self, P> where
    Self: Sized,
    P: FnMut(&Self::Item) -> bool, 

Takes items while predicate returns true.

Examples

use futures_lite::stream::{self, StreamExt};

let s = stream::iter(vec![1, 2, 3, 4]);
let mut s = s.take_while(|x| *x < 3);

assert_eq!(s.next().await, Some(1));
assert_eq!(s.next().await, Some(2));
assert_eq!(s.next().await, None);

fn skip(self, n: usize) -> Skip<Self> where
    Self: Sized, 

Skips the first n items of the stream.

Examples

use futures_lite::stream::{self, StreamExt};

let s = stream::iter(vec![1, 2, 3]);
let mut s = s.skip(2);

assert_eq!(s.next().await, Some(3));
assert_eq!(s.next().await, None);

fn skip_while<P>(self, predicate: P) -> SkipWhile<Self, P> where
    Self: Sized,
    P: FnMut(&Self::Item) -> bool, 

Skips items while predicate returns true.

Examples

use futures_lite::stream::{self, StreamExt};

let s = stream::iter(vec![-1i32, 0, 1]);
let mut s = s.skip_while(|x| x.is_negative());

assert_eq!(s.next().await, Some(0));
assert_eq!(s.next().await, Some(1));
assert_eq!(s.next().await, None);

fn step_by(self, step: usize) -> StepBy<Self> where
    Self: Sized, 

Yields every stepth item.

Panics

This method will panic if the step is 0.

Examples

use futures_lite::stream::{self, StreamExt};

let s = stream::iter(vec![0, 1, 2, 3, 4]);
let mut s = s.step_by(2);

assert_eq!(s.next().await, Some(0));
assert_eq!(s.next().await, Some(2));
assert_eq!(s.next().await, Some(4));
assert_eq!(s.next().await, None);

fn chain<U>(self, other: U) -> Chain<Self, U> where
    Self: Sized,
    U: Stream<Item = Self::Item> + Sized, 

Appends another stream to the end of this one.

Examples

use futures_lite::stream::{self, StreamExt};

let s1 = stream::iter(vec![1, 2]);
let s2 = stream::iter(vec![7, 8]);
let mut s = s1.chain(s2);

assert_eq!(s.next().await, Some(1));
assert_eq!(s.next().await, Some(2));
assert_eq!(s.next().await, Some(7));
assert_eq!(s.next().await, Some(8));
assert_eq!(s.next().await, None);

fn cloned<'a, T>(self) -> Cloned<Self> where
    Self: Stream<Item = &'a T> + Sized,
    T: Clone + 'a, 

Clones all items.

Examples

use futures_lite::stream::{self, StreamExt};

let s = stream::iter(vec![&1, &2]);
let mut s = s.cloned();

assert_eq!(s.next().await, Some(1));
assert_eq!(s.next().await, Some(2));
assert_eq!(s.next().await, None);

fn copied<'a, T>(self) -> Copied<Self> where
    Self: Stream<Item = &'a T> + Sized,
    T: Copy + 'a, 

Copies all items.

Examples

use futures_lite::stream::{self, StreamExt};

let s = stream::iter(vec![&1, &2]);
let mut s = s.copied();

assert_eq!(s.next().await, Some(1));
assert_eq!(s.next().await, Some(2));
assert_eq!(s.next().await, None);

fn collect<C>(self) -> CollectFuture<Self, C> where
    Self: Sized,
    C: Default + Extend<Self::Item>, 

Collects all items in the stream into a collection.

Examples

use futures_lite::stream::{self, StreamExt};

let mut s = stream::iter(1..=3);

let items: Vec<_> = s.collect().await;
assert_eq!(items, [1, 2, 3]);

fn try_collect<T, E, C>(self) -> TryCollectFuture<Self, C> where
    Self: Stream<Item = Result<T, E>> + Sized,
    C: Default + Extend<T>, 

Collects all items in the fallible stream into a collection.

use futures_lite::stream::{self, StreamExt};

let s = stream::iter(vec![Ok(1), Err(2), Ok(3)]);
let res: Result<Vec<i32>, i32> = s.try_collect().await;
assert_eq!(res, Err(2));

let s = stream::iter(vec![Ok(1), Ok(2), Ok(3)]);
let res: Result<Vec<i32>, i32> = s.try_collect().await;
assert_eq!(res, Ok(vec![1, 2, 3]));

fn partition<B, P>(self, predicate: P) -> PartitionFuture<Self, P, B> where
    Self: Sized,
    B: Default + Extend<Self::Item>,
    P: FnMut(&Self::Item) -> bool, 

Partitions items into those for which predicate is true and those for which it is false, and then collects them into two collections.

Examples

use futures_lite::stream::{self, StreamExt};

let s = stream::iter(vec![1, 2, 3]);
let (even, odd): (Vec<_>, Vec<_>) = s.partition(|&n| n % 2 == 0).await;

assert_eq!(even, &[2]);
assert_eq!(odd, &[1, 3]);

fn fold<T, F>(self, init: T, f: F) -> FoldFuture<Self, F, T> where
    Self: Sized,
    F: FnMut(T, Self::Item) -> T, 

Accumulates a computation over the stream.

The computation begins with the accumulator value set to init, and then applies f to the accumulator and each item in the stream. The final accumulator value is returned.

Examples

use futures_lite::stream::{self, StreamExt};

let s = stream::iter(vec![1, 2, 3]);
let sum = s.fold(0, |acc, x| acc + x).await;

assert_eq!(sum, 6);

fn try_fold<T, E, F, B>(
    &mut self,
    init: B,
    f: F
) -> TryFoldFuture<'_, Self, F, B> where
    Self: Stream<Item = Result<T, E>> + Unpin + Sized,
    F: FnMut(B, T) -> Result<B, E>, 

Accumulates a fallible computation over the stream.

The computation begins with the accumulator value set to init, and then applies f to the accumulator and each item in the stream. The final accumulator value is returned, or an error if f failed the computation.

Examples

use futures_lite::stream::{self, StreamExt};

let mut s = stream::iter(vec![Ok(1), Ok(2), Ok(3)]);

let sum = s.try_fold(0, |acc, v| {
    if (acc + v) % 2 == 1 {
        Ok(acc + v)
    } else {
        Err("fail")
    }
})
.await;

assert_eq!(sum, Err("fail"));

fn scan<St, B, F>(self, initial_state: St, f: F) -> Scan<Self, St, F> where
    Self: Sized,
    F: FnMut(&mut St, Self::Item) -> Option<B>, 

Maps items of the stream to new values using a state value and a closure.

Scanning begins with the inital state set to initial_state, and then applies f to the state and each item in the stream. The stream stops when f returns None.

Examples

use futures_lite::stream::{self, StreamExt};

let s = stream::iter(vec![1, 2, 3]);
let mut s = s.scan(1, |state, x| {
    *state = *state * x;
    Some(-*state)
});

assert_eq!(s.next().await, Some(-1));
assert_eq!(s.next().await, Some(-2));
assert_eq!(s.next().await, Some(-6));
assert_eq!(s.next().await, None);

fn fuse(self) -> Fuse<Self> where
    Self: Sized, 

Fuses the stream so that it stops yielding items after the first None.

Examples

use futures_lite::stream::{self, StreamExt};

let mut s = stream::once(1).fuse();

assert_eq!(s.next().await, Some(1));
assert_eq!(s.next().await, None);
assert_eq!(s.next().await, None);

fn cycle(self) -> Cycle<Self> where
    Self: Clone + Sized, 

Repeats the stream from beginning to end, forever.

Examples

use futures_lite::stream::{self, StreamExt};

let mut s = stream::iter(vec![1, 2]).cycle();

assert_eq!(s.next().await, Some(1));
assert_eq!(s.next().await, Some(2));
assert_eq!(s.next().await, Some(1));
assert_eq!(s.next().await, Some(2));

fn enumerate(self) -> Enumerate<Self> where
    Self: Sized, 

Enumerates items, mapping them to (index, item).

Examples

use futures_lite::stream::{self, StreamExt};

let s = stream::iter(vec!['a', 'b', 'c']);
let mut s = s.enumerate();

assert_eq!(s.next().await, Some((0, 'a')));
assert_eq!(s.next().await, Some((1, 'b')));
assert_eq!(s.next().await, Some((2, 'c')));
assert_eq!(s.next().await, None);

fn inspect<F>(self, f: F) -> Inspect<Self, F> where
    Self: Sized,
    F: FnMut(&Self::Item), 

Calls a closure on each item and passes it on.

Examples

use futures_lite::stream::{self, StreamExt};

let s = stream::iter(vec![1, 2, 3, 4, 5]);

let sum = s
   .inspect(|x| println!("about to filter {}", x))
   .filter(|x| x % 2 == 0)
   .inspect(|x| println!("made it through filter: {}", x))
   .fold(0, |sum, i| sum + i)
   .await;

fn nth(&mut self, n: usize) -> NthFuture<'_, Self> where
    Self: Unpin, 

Gets the nth item of the stream.

In the end, n+1 items of the stream will be consumed.

Examples

use futures_lite::stream::{self, StreamExt};

let mut s = stream::iter(vec![0, 1, 2, 3, 4, 5, 6, 7]);

assert_eq!(s.nth(2).await, Some(2));
assert_eq!(s.nth(2).await, Some(5));
assert_eq!(s.nth(2).await, None);

fn last(self) -> LastFuture<Self> where
    Self: Sized, 

Returns the last item in the stream.

Examples

use futures_lite::stream::{self, StreamExt};

let s = stream::iter(vec![1, 2, 3, 4]);
assert_eq!(s.last().await, Some(4));

let s = stream::empty::<i32>();
assert_eq!(s.last().await, None);

fn find<P>(&mut self, predicate: P) -> FindFuture<'_, Self, P> where
    Self: Unpin,
    P: FnMut(&Self::Item) -> bool, 

Finds the first item of the stream for which predicate returns true.

Examples

use futures_lite::stream::{self, StreamExt};

let mut s = stream::iter(vec![11, 12, 13, 14]);

assert_eq!(s.find(|x| *x % 2 == 0).await, Some(12));
assert_eq!(s.next().await, Some(13));

fn find_map<F, B>(&mut self, f: F) -> FindMapFuture<'_, Self, F> where
    Self: Unpin,
    F: FnMut(Self::Item) -> Option<B>, 

Applies a closure to items in the stream and returns the first Some result.

Examples

use futures_lite::stream::{self, StreamExt};

let mut s = stream::iter(vec!["lol", "NaN", "2", "5"]);
let number = s.find_map(|s| s.parse().ok()).await;

assert_eq!(number, Some(2));

fn position<P>(&mut self, predicate: P) -> PositionFuture<'_, Self, P> where
    Self: Unpin,
    P: FnMut(Self::Item) -> bool, 

Finds the index of the first item of the stream for which predicate returns true.

Examples

use futures_lite::stream::{self, StreamExt};

let mut s = stream::iter(vec![0, 1, 2, 3, 4, 5]);

assert_eq!(s.position(|x| x == 2).await, Some(2));
assert_eq!(s.position(|x| x == 3).await, Some(0));
assert_eq!(s.position(|x| x == 9).await, None);

fn all<P>(&mut self, predicate: P) -> AllFuture<'_, Self, P> where
    Self: Unpin,
    P: FnMut(Self::Item) -> bool, 

Tests if predicate returns true for all items in the stream.

The result is true for an empty stream.

Examples

use futures_lite::stream::{self, StreamExt};

let mut s = stream::iter(vec![1, 2, 3]);
assert!(!s.all(|x| x % 2 == 0).await);

let mut s = stream::iter(vec![2, 4, 6, 8]);
assert!(s.all(|x| x % 2 == 0).await);

let mut s = stream::empty::<i32>();
assert!(s.all(|x| x % 2 == 0).await);

fn any<P>(&mut self, predicate: P) -> AnyFuture<'_, Self, P> where
    Self: Unpin,
    P: FnMut(Self::Item) -> bool, 

Tests if predicate returns true for any item in the stream.

The result is false for an empty stream.

Examples

use futures_lite::stream::{self, StreamExt};

let mut s = stream::iter(vec![1, 3, 5, 7]);
assert!(!s.any(|x| x % 2 == 0).await);

let mut s = stream::iter(vec![1, 2, 3]);
assert!(s.any(|x| x % 2 == 0).await);

let mut s = stream::empty::<i32>();
assert!(!s.any(|x| x % 2 == 0).await);

fn for_each<F>(self, f: F) -> ForEachFuture<Self, F> where
    Self: Sized,
    F: FnMut(Self::Item), 

Calls a closure on each item of the stream.

Examples

use futures_lite::stream::{self, StreamExt};

let mut s = stream::iter(vec![1, 2, 3]);
s.for_each(|s| println!("{}", s)).await;

fn try_for_each<F, E>(&mut self, f: F) -> TryForEachFuture<'_, Self, F> where
    Self: Unpin,
    F: FnMut(Self::Item) -> Result<(), E>, 

Calls a fallible closure on each item of the stream, stopping on first error.

Examples

use futures_lite::stream::{self, StreamExt};

let mut s = stream::iter(vec![0, 1, 2, 3]);

let mut v = vec![];
let res = s
    .try_for_each(|n| {
        if n < 2 {
            v.push(n);
            Ok(())
        } else {
            Err("too big")
        }
    })
    .await;

assert_eq!(v, &[0, 1]);
assert_eq!(res, Err("too big"));

fn zip<U>(self, other: U) -> Zip<Self, U> where
    Self: Sized,
    U: Stream, 

Zips up two streams into a single stream of pairs.

The stream of pairs stops when either of the original two streams is exhausted.

Examples

use futures_lite::stream::{self, StreamExt};

let l = stream::iter(vec![1, 2, 3]);
let r = stream::iter(vec![4, 5, 6, 7]);
let mut s = l.zip(r);

assert_eq!(s.next().await, Some((1, 4)));
assert_eq!(s.next().await, Some((2, 5)));
assert_eq!(s.next().await, Some((3, 6)));
assert_eq!(s.next().await, None);

fn unzip<A, B, FromA, FromB>(self) -> UnzipFuture<Self, FromA, FromB> where
    FromA: Default + Extend<A>,
    FromB: Default + Extend<B>,
    Self: Stream<Item = (A, B)> + Sized, 

Collects a stream of pairs into a pair of collections.

Examples

use futures_lite::stream::{self, StreamExt};

let s = stream::iter(vec![(1, 2), (3, 4)]);
let (left, right): (Vec<_>, Vec<_>) = s.unzip().await;

assert_eq!(left, [1, 3]);
assert_eq!(right, [2, 4]);

fn or<S>(self, other: S) -> Or<Self, S> where
    Self: Sized,
    S: Stream<Item = Self::Item>, 

Merges with other stream, preferring items from self whenever both streams are ready.

Examples

use futures_lite::stream::{self, StreamExt};
use futures_lite::stream::{once, pending};

assert_eq!(once(1).or(pending()).next().await, Some(1));
assert_eq!(pending().or(once(2)).next().await, Some(2));

// The first future wins.
assert_eq!(once(1).or(once(2)).next().await, Some(1));

fn race<S>(self, other: S) -> Race<Self, S> where
    Self: Sized,
    S: Stream<Item = Self::Item>, 

Merges with other stream, with no preference for either stream when both are ready.

Examples

use futures_lite::stream::{self, StreamExt};
use futures_lite::stream::{once, pending};

assert_eq!(once(1).race(pending()).next().await, Some(1));
assert_eq!(pending().race(once(2)).next().await, Some(2));

// One of the two stream is randomly chosen as the winner.
let res = once(1).race(once(2)).next().await;

fn boxed<'a>(self) -> Pin<Box<dyn Stream<Item = Self::Item> + Send + 'a>> where
    Self: Send + Sized + 'a, 

Boxes the stream and changes its type to dyn Stream + Send + 'a.

Examples

use futures_lite::stream::{self, StreamExt};

let a = stream::once(1);
let b = stream::empty();

// Streams of different types can be stored in
// the same collection when they are boxed:
let streams = vec![a.boxed(), b.boxed()];

fn boxed_local<'a>(self) -> Pin<Box<dyn Stream<Item = Self::Item> + 'a>> where
    Self: Sized + 'a, 

Boxes the stream and changes its type to dyn Stream + 'a.

Examples

use futures_lite::stream::{self, StreamExt};

let a = stream::once(1);
let b = stream::empty();

// Streams of different types can be stored in
// the same collection when they are boxed:
let streams = vec![a.boxed_local(), b.boxed_local()];

Loading content...

Implementors

impl<S: Stream + ?Sized> StreamExt for S

Loading content...