Trait StreamExt

pub trait StreamExt: Stream {
    // Provided methods
    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 drain(&mut self) -> Drain<'_, Self> { ... }
    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 drain(&mut self) -> Drain<'_, Self>

Yields all immediately available values from a stream.

This is intended to be used as a way of polling a stream without waiting, similar to the try_iter function on std::sync::mpsc::Receiver. For instance, running this stream on an async_channel::Receiver will return all messages that are currently in the channel, but will not wait for new messages.

This returns a Stream instead of an Iterator because it still needs access to the polling context in order to poll the underlying stream. Since this stream will never return Poll::Pending, wrapping it in block_on will allow it to be effectively used as an Iterator.

This stream is not necessarily fused. After it returns None, it can return Some(x) in the future if it is polled again.

Examples

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

// A stream that yields two values, returns `Pending`, and then yields one more value.
let pend_once = stream::once_future(async {
    future::yield_now().await;
    3
});
let s = stream::iter(vec![1, 2]).chain(pend_once);
pin!(s);

// This will return the first two values, and then `None` because the stream returns
// `Pending` after that.
let mut iter = stream::block_on(s.drain());
assert_eq!(iter.next(), Some(1));
assert_eq!(iter.next(), Some(2));
assert_eq!(iter.next(), None);

// This will return the last value, because the stream returns `Ready` when polled.
assert_eq!(iter.next(), Some(3));
assert_eq!(iter.next(), None);

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()];

Dyn Compatibility

This trait is not dyn compatible.

In older versions of Rust, dyn compatibility was called "object safety", so this trait is not object safe.

Implementors

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