[−][src]Trait futures::stream::StreamExt

pub trait StreamExt: Stream {
    default fn next(&mut self) -> Next<Self>
    where
        Self: Unpin,
    { ... }

    default fn into_future(self) -> StreamFuture<Self>
    where
        Self: Unpin,
    { ... }

    default fn map<T, F>(self, f: F) -> Map<Self, F>
    where
        F: FnMut(Self::Item) -> T,
    { ... }

    default fn filter<Fut, F>(self, f: F) -> Filter<Self, Fut, F>
    where
        F: FnMut(&Self::Item) -> Fut,
        Fut: Future<Output = bool>,
    { ... }

    default fn filter_map<Fut, T, F>(self, f: F) -> FilterMap<Self, Fut, F>
    where
        F: FnMut(Self::Item) -> Fut,
        Fut: Future<Output = Option<T>>,
    { ... }

    default fn then<Fut, F>(self, f: F) -> Then<Self, Fut, F>
    where
        F: FnMut(Self::Item) -> Fut,
        Fut: Future,
    { ... }

    default fn collect<C>(self) -> Collect<Self, C>
    where
        C: Default + Extend<Self::Item>,
    { ... }

    default fn concat(self) -> Concat<Self>
    where
        Self::Item: Extend<<Self::Item as IntoIterator>::Item>,
        Self::Item: IntoIterator,
        Self::Item: Default,
    { ... }

    default fn fold<T, Fut, F>(self, init: T, f: F) -> Fold<Self, Fut, T, F>
    where
        F: FnMut(T, Self::Item) -> Fut,
        Fut: Future<Output = T>,
    { ... }

    default fn flatten(self) -> Flatten<Self>
    where
        Self::Item: Stream,
    { ... }

    default fn skip_while<Fut, F>(self, f: F) -> SkipWhile<Self, Fut, F>
    where
        F: FnMut(&Self::Item) -> Fut,
        Fut: Future<Output = bool>,
    { ... }

    default fn take_while<Fut, F>(self, f: F) -> TakeWhile<Self, Fut, F>
    where
        F: FnMut(&Self::Item) -> Fut,
        Fut: Future<Output = bool>,
    { ... }

    default fn for_each<Fut, F>(self, f: F) -> ForEach<Self, Fut, F>
    where
        F: FnMut(Self::Item) -> Fut,
        Fut: Future<Output = ()>,
    { ... }

    default fn for_each_concurrent<Fut, F, impl Into>>(
        self, 
        limit: impl Into>, 
        f: F
    ) -> ForEachConcurrent<Self, Fut, F>
    where
        F: FnMut(Self::Item) -> Fut,
        Fut: Future<Output = ()>,
        impl Into>: Into<Option<usize>>,
    { ... }

    default fn take(self, n: u64) -> Take<Self> { ... }

    default fn skip(self, n: u64) -> Skip<Self> { ... }

    default fn fuse(self) -> Fuse<Self> { ... }

    default fn by_ref(&mut self) -> &mut Self { ... }

    default fn catch_unwind(self) -> CatchUnwind<Self>
    where
        Self: UnwindSafe,
    { ... }

    default fn boxed(self) -> Pin<Box<Self>> { ... }

    default fn buffered(self, n: usize) -> Buffered<Self>
    where
        Self::Item: Future,
    { ... }

    default fn buffer_unordered(self, n: usize) -> BufferUnordered<Self>
    where
        Self::Item: Future,
    { ... }

    default fn zip<St>(self, other: St) -> Zip<Self, St>
    where
        St: Stream,
    { ... }

    default fn chain<St>(self, other: St) -> Chain<Self, St>
    where
        St: Stream<Item = Self::Item>,
    { ... }

    default fn peekable(self) -> Peekable<Self> { ... }

    default fn chunks(self, capacity: usize) -> Chunks<Self> { ... }

    default fn select<St>(self, other: St) -> Select<Self, St>
    where
        St: Stream<Item = Self::Item>,
    { ... }

    default fn forward<S>(self, sink: S) -> Forward<Self, S>
    where
        S: Sink<Self::Ok>,
        Self: TryStream<Error = <S as Sink<Self::Ok>>::SinkError>,
    { ... }

    default fn split<Item>(self) -> (SplitSink<Self, Item>, SplitStream<Self>)
    where
        Self: Sink<Item>,
    { ... }

    default fn inspect<F>(self, f: F) -> Inspect<Self, F>
    where
        F: FnMut(&Self::Item),
    { ... }

    default fn left_stream<B>(self) -> Either<Self, B>
    where
        B: Stream<Item = Self::Item>,
    { ... }

    default fn right_stream<B>(self) -> Either<B, Self>
    where
        B: Stream<Item = Self::Item>,
    { ... }

    default fn poll_next_unpin(
        &mut self, 
        cx: &mut Context
    ) -> Poll<Option<Self::Item>>
    where
        Self: Unpin,
    { ... }

    default fn select_next_some(&mut self) -> SelectNextSome<Self>
    where
        Self: Unpin + FusedStream,
    { ... }
}

An extension trait for Streams that provides a variety of convenient combinator functions.

Provided methods

ⓘImportant traits for Next<'_, St>
default fn next(&mut self) -> Next<Self> where
    Self: Unpin, 

Creates a future that resolves to the next item in the stream.

Note that because next doesn't take ownership over the stream, the Stream type must be Unpin. If you want to use next with a !Unpin stream, you'll first have to pin the stream. This can be done by boxing the stream using Box::pin or pinning it to the stack using the pin_mut! macro from the pin_utils crate.

Examples

use futures::executor::block_on;
use futures::stream::{self, StreamExt};

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

assert_eq!(block_on(stream.next()), Some(1));
assert_eq!(block_on(stream.next()), Some(2));
assert_eq!(block_on(stream.next()), Some(3));
assert_eq!(block_on(stream.next()), None);

ⓘImportant traits for StreamFuture<St>
default fn into_future(self) -> StreamFuture<Self> where
    Self: Unpin, 

Converts this stream into a future of (next_item, tail_of_stream). If the stream terminates, then the next item is None.

The returned future can be used to compose streams and futures together by placing everything into the "world of futures".

Note that because into_future moves the stream, the Stream type must be Unpin. If you want to use into_future with a !Unpin stream, you'll first have to pin the stream. This can be done by boxing the stream using Box::pin or pinning it to the stack using the pin_mut! macro from the pin_utils crate.

Examples

use futures::executor::block_on;
use futures::stream::{self, StreamExt};

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

let (item, stream) = block_on(stream.into_future());
assert_eq!(Some(1), item);

let (item, stream) = block_on(stream.into_future());
assert_eq!(Some(2), item);

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

Maps this stream's items to a different type, returning a new stream of the resulting type.

The provided closure is executed over all elements of this stream as they are made available. It is executed inline with calls to poll_next.

Note that this function consumes the stream passed into it and returns a wrapped version of it, similar to the existing map methods in the standard library.

Examples

use futures::executor::block_on;
use futures::stream::{self, StreamExt};

let stream = stream::iter(1..=3);
let stream = stream.map(|x| x + 3);

assert_eq!(vec![4, 5, 6], block_on(stream.collect::<Vec<_>>()));

default fn filter<Fut, F>(self, f: F) -> Filter<Self, Fut, F> where
    F: FnMut(&Self::Item) -> Fut,
    Fut: Future<Output = bool>, 

Filters the values produced by this stream according to the provided asynchronous predicate.

As values of this stream are made available, the provided predicate f will be run against them. If the predicate returns a Future which resolves to true, then the stream will yield the value, but if the predicate returns a Future which resolves to false, then the value will be discarded and the next value will be produced.

Note that this function consumes the stream passed into it and returns a wrapped version of it, similar to the existing filter methods in the standard library.

Examples

use futures::executor::block_on;
use futures::future;
use futures::stream::{self, StreamExt};

let stream = stream::iter(1..=10);
let evens = stream.filter(|x| future::ready(x % 2 == 0));

assert_eq!(vec![2, 4, 6, 8, 10], block_on(evens.collect::<Vec<_>>()));

default fn filter_map<Fut, T, F>(self, f: F) -> FilterMap<Self, Fut, F> where
    F: FnMut(Self::Item) -> Fut,
    Fut: Future<Output = Option<T>>, 

Filters the values produced by this stream while simultaneously mapping them to a different type according to the provided asynchronous closure.

As values of this stream are made available, the provided function will be run on them. If the future returned by the predicate f resolves to Some(item) then the stream will yield the value item, but if it resolves to None then the next value will be produced.

Note that this function consumes the stream passed into it and returns a wrapped version of it, similar to the existing filter_map methods in the standard library.

Examples

use futures::executor::block_on;
use futures::future;
use futures::stream::{self, StreamExt};

let stream = stream::iter(1..=10);
let evens = stream.filter_map(|x| {
    let ret = if x % 2 == 0 { Some(x + 1) } else { None };
    future::ready(ret)
});

assert_eq!(vec![3, 5, 7, 9, 11], block_on(evens.collect::<Vec<_>>()));

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

Computes from this stream's items new items of a different type using an asynchronous closure.

The provided closure f will be called with an Item once a value is ready, it returns a future which will then be run to completion to produce the next value on this stream.

Note that this function consumes the stream passed into it and returns a wrapped version of it.

Examples

use futures::executor::block_on;
use futures::future;
use futures::stream::{self, StreamExt};

let stream = stream::iter(1..=3);
let stream = stream.then(|x| future::ready(x + 3));

assert_eq!(vec![4, 5, 6], block_on(stream.collect::<Vec<_>>()));

ⓘImportant traits for Collect<St, C>
default fn collect<C>(self) -> Collect<Self, C> where
    C: Default + Extend<Self::Item>, 

Collect all of the values of this stream into a vector, returning a future representing the result of that computation.

The returned future will be resolved when the stream terminates.

Examples

use futures::channel::mpsc;
use futures::executor::block_on;
use futures::stream::StreamExt;
use std::thread;

let (tx, rx) = mpsc::unbounded();

thread::spawn(move || {
    for i in (1..=5) {
        tx.unbounded_send(i).unwrap();
    }
});

let output = block_on(rx.collect::<Vec<i32>>());
assert_eq!(output, vec![1, 2, 3, 4, 5]);

ⓘImportant traits for Concat<St>
default fn concat(self) -> Concat<Self> where
    Self::Item: Extend<<Self::Item as IntoIterator>::Item>,
    Self::Item: IntoIterator,
    Self::Item: Default, 

Concatenate all items of a stream into a single extendable destination, returning a future representing the end result.

This combinator will extend the first item with the contents of all the subsequent results of the stream. If the stream is empty, the default value will be returned.

Works with all collections that implement the Extend trait.

Examples

use futures::channel::mpsc;
use futures::executor::block_on;
use futures::stream::StreamExt;
use std::thread;

let (tx, rx) = mpsc::unbounded();

thread::spawn(move || {
    for i in (0..3).rev() {
        let n = i * 3;
        tx.unbounded_send(vec![n + 1, n + 2, n + 3]).unwrap();
    }
});

let result = block_on(rx.concat());

assert_eq!(result, vec![7, 8, 9, 4, 5, 6, 1, 2, 3]);

ⓘImportant traits for Fold<St, Fut, T, F>
default fn fold<T, Fut, F>(self, init: T, f: F) -> Fold<Self, Fut, T, F> where
    F: FnMut(T, Self::Item) -> Fut,
    Fut: Future<Output = T>, 

Execute an accumulating asynchronous computation over a stream, collecting all the values into one final result.

This combinator will accumulate all values returned by this stream according to the closure provided. The initial state is also provided to this method and then is returned again by each execution of the closure. Once the entire stream has been exhausted the returned future will resolve to this value.

Examples

use futures::executor::block_on;
use futures::future;
use futures::stream::{self, StreamExt};

let number_stream = stream::iter(0..6);
let sum = number_stream.fold(0, |acc, x| future::ready(acc + x));
assert_eq!(block_on(sum), 15);

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

Flattens a stream of streams into just one continuous stream.

Examples

use futures::channel::mpsc;
use futures::executor::block_on;
use futures::stream::StreamExt;
use std::thread;

let (tx1, rx1) = mpsc::unbounded();
let (tx2, rx2) = mpsc::unbounded();
let (tx3, rx3) = mpsc::unbounded();

thread::spawn(move || {
    tx1.unbounded_send(1).unwrap();
    tx1.unbounded_send(2).unwrap();
});
thread::spawn(move || {
    tx2.unbounded_send(3).unwrap();
    tx2.unbounded_send(4).unwrap();
});
thread::spawn(move || {
    tx3.unbounded_send(rx1).unwrap();
    tx3.unbounded_send(rx2).unwrap();
});

let output = block_on(rx3.flatten().collect::<Vec<i32>>());
assert_eq!(output, vec![1, 2, 3, 4]);

default fn skip_while<Fut, F>(self, f: F) -> SkipWhile<Self, Fut, F> where
    F: FnMut(&Self::Item) -> Fut,
    Fut: Future<Output = bool>, 

Skip elements on this stream while the provided asynchronous predicate resolves to true.

This function, like Iterator::skip_while, will skip elements on the stream until the predicate f resolves to false. Once one element returns false all future elements will be returned from the underlying stream.

Examples

use futures::executor::block_on;
use futures::future;
use futures::stream::{self, StreamExt};

let stream = stream::iter(1..=10);

let stream = stream.skip_while(|x| future::ready(*x <= 5));

assert_eq!(vec![6, 7, 8, 9, 10], block_on(stream.collect::<Vec<_>>()));

default fn take_while<Fut, F>(self, f: F) -> TakeWhile<Self, Fut, F> where
    F: FnMut(&Self::Item) -> Fut,
    Fut: Future<Output = bool>, 

Take elements from this stream while the provided asynchronous predicate resolves to true.

This function, like Iterator::take_while, will take elements from the stream until the predicate f resolves to false. Once one element returns false it will always return that the stream is done.

Examples

use futures::executor::block_on;
use futures::future;
use futures::stream::{self, StreamExt};

let stream = stream::iter(1..=10);

let stream = stream.take_while(|x| future::ready(*x <= 5));

assert_eq!(vec![1, 2, 3, 4, 5], block_on(stream.collect::<Vec<_>>()));

ⓘImportant traits for ForEach<St, Fut, F>
default fn for_each<Fut, F>(self, f: F) -> ForEach<Self, Fut, F> where
    F: FnMut(Self::Item) -> Fut,
    Fut: Future<Output = ()>, 

Runs this stream to completion, executing the provided asynchronous closure for each element on the stream.

The closure provided will be called for each item this stream produces, yielding a future. That future will then be executed to completion before moving on to the next item.

The returned value is a Future where the Output type is (); it is executed entirely for its side effects.

To process each item in the stream and produce another stream instead of a single future, use then instead.

Examples

use futures::executor::block_on;
use futures::future;
use futures::stream::{self, StreamExt};

let mut x = 0;

{
    let fut = stream::repeat(1).take(3).for_each(|item| {
        x += item;
        future::ready(())
    });
    block_on(fut);
}

assert_eq!(x, 3);

ⓘImportant traits for ForEachConcurrent<St, Fut, F>
default fn for_each_concurrent<Fut, F, impl Into>>(
    self,
    limit: impl Into>,
    f: F
) -> ForEachConcurrent<Self, Fut, F> where
    F: FnMut(Self::Item) -> Fut,
    Fut: Future<Output = ()>,
    impl Into>: Into<Option<usize>>, 

Runs this stream to completion, executing the provided asynchronous closure for each element on the stream concurrently as elements become available.

This is similar to StreamExt::for_each, but the futures produced by the closure are run concurrently (but not in parallel-- this combinator does not introduce any threads).

The closure provided will be called for each item this stream produces, yielding a future. That future will then be executed to completion concurrently with the other futures produced by the closure.

The first argument is an optional limit on the number of concurrent futures. If this limit is not None, no more than limit futures will be run concurrently. The limit argument is of type Into<Option<usize>>, and so can be provided as either None, Some(10), or just 10. Note: a limit of zero is interpreted as no limit at all, and will have the same result as passing in None.

This method is only available when the std feature of this library is activated, and it is activated by default.

Examples

#![feature(async_await, await_macro, futures_api)]
use futures::channel::oneshot;
use futures::stream::{self, StreamExt};

let (tx1, rx1) = oneshot::channel();
let (tx2, rx2) = oneshot::channel();
let (tx3, rx3) = oneshot::channel();

let fut = stream::iter(vec![rx1, rx2, rx3]).for_each_concurrent(
    /* limit */ 2,
    async move |rx| {
        await!(rx).unwrap();
    }
);
tx1.send(()).unwrap();
tx2.send(()).unwrap();
tx3.send(()).unwrap();
await!(fut);

default fn take(self, n: u64) -> Take<Self>

Creates a new stream of at most n items of the underlying stream.

Once n items have been yielded from this stream then it will always return that the stream is done.

Examples

use futures::executor::block_on;
use futures::stream::{self, StreamExt};

let stream = stream::iter(1..=10).take(3);

assert_eq!(vec![1, 2, 3], block_on(stream.collect::<Vec<_>>()));

default fn skip(self, n: u64) -> Skip<Self>

Creates a new stream which skips n items of the underlying stream.

Once n items have been skipped from this stream then it will always return the remaining items on this stream.

Examples

use futures::executor::block_on;
use futures::stream::{self, StreamExt};

let stream = stream::iter(1..=10).skip(5);

assert_eq!(vec![6, 7, 8, 9, 10], block_on(stream.collect::<Vec<_>>()));

default fn fuse(self) -> Fuse<Self>

Fuse a stream such that poll_next will never again be called once it has finished. This method can be used t turn any Stream into a FusedStream.

Normally, once a stream has returned None from poll_next any further calls could exhibit bad behavior such as block forever, panic, never return, etc. If it is known that poll_next may be called after stream has already finished, then this method can be used to ensure that it has defined semantics.

The poll_next method of a fused stream is guaranteed to return None after the underlying stream has finished.

Examples

#![feature(futures_api)]
use futures::executor::block_on_stream;
use futures::stream::{self, StreamExt};
use futures::task::Poll;

let mut x = 0;
let stream = stream::poll_fn(|_| {
    x += 1;
    match x {
        0..=2 => Poll::Ready(Some(x)),
        3 => Poll::Ready(None),
        _ => panic!("should not happen")
    }
}).fuse();

let mut iter = block_on_stream(stream);
assert_eq!(Some(1), iter.next());
assert_eq!(Some(2), iter.next());
assert_eq!(None, iter.next());
assert_eq!(None, iter.next());
// ...

ⓘImportant traits for &'_ mut F
default fn by_ref(&mut self) -> &mut Self

Borrows a stream, rather than consuming it.

This is useful to allow applying stream adaptors while still retaining ownership of the original stream.

Examples

use futures::executor::block_on;
use futures::future;
use futures::stream::{self, StreamExt};

let mut stream = stream::iter(1..5);

let sum = block_on(stream.by_ref()
                         .take(2)
                         .fold(0, |a, b| future::ready(a + b)));
assert_eq!(sum, 3);

// You can use the stream again
let sum = block_on(stream.take(2).fold(0, |a, b| future::ready(a + b)));
assert_eq!(sum, 7);

default fn catch_unwind(self) -> CatchUnwind<Self> where
    Self: UnwindSafe, 

Catches unwinding panics while polling the stream.

Caught panic (if any) will be the last element of the resulting stream.

In general, panics within a stream can propagate all the way out to the task level. This combinator makes it possible to halt unwinding within the stream itself. It's most commonly used within task executors. This method should not be used for error handling.

Note that this method requires the UnwindSafe bound from the standard library. This isn't always applied automatically, and the standard library provides an AssertUnwindSafe wrapper type to apply it after-the fact. To assist using this method, the Stream trait is also implemented for AssertUnwindSafe<St> where St implements Stream.

This method is only available when the std feature of this library is activated, and it is activated by default.

Examples

use futures::executor::block_on;
use futures::stream::{self, StreamExt};

let stream = stream::iter(vec![Some(10), None, Some(11)]);
// Panic on second element
let stream_panicking = stream.map(|o| o.unwrap());
// Collect all the results
let stream = stream_panicking.catch_unwind();

let results: Vec<Result<i32, _>> = block_on(stream.collect());
match results[0] {
    Ok(10) => {}
    _ => panic!("unexpected result!"),
}
assert!(results[1].is_err());
assert_eq!(results.len(), 2);

ⓘImportant traits for Pin<P>
default fn boxed(self) -> Pin<Box<Self>>

Wrap the stream in a Box, pinning it.

default fn buffered(self, n: usize) -> Buffered<Self> where
    Self::Item: Future, 

An adaptor for creating a buffered list of pending futures.

If this stream's item can be converted into a future, then this adaptor will buffer up to at most n futures and then return the outputs in the same order as the underlying stream. No more than n futures will be buffered at any point in time, and less than n may also be buffered depending on the state of each future.

The returned stream will be a stream of each future's output.

This method is only available when the std feature of this library is activated, and it is activated by default.

default fn buffer_unordered(self, n: usize) -> BufferUnordered<Self> where
    Self::Item: Future, 

An adaptor for creating a buffered list of pending futures (unordered).

If this stream's item can be converted into a future, then this adaptor will buffer up to n futures and then return the outputs in the order in which they complete. No more than n futures will be buffered at any point in time, and less than n may also be buffered depending on the state of each future.

The returned stream will be a stream of each future's output.

This method is only available when the std feature of this library is activated, and it is activated by default.

Examples

#![feature(async_await, await_macro, futures_api)]
use futures::channel::oneshot;
use futures::stream::{self, StreamExt};

let (send_one, recv_one) = oneshot::channel();
let (send_two, recv_two) = oneshot::channel();

let stream_of_futures = stream::iter(vec![recv_one, recv_two]);
let mut buffered = stream_of_futures.buffer_unordered(10);

send_two.send(2i32);
assert_eq!(await!(buffered.next()), Some(Ok(2i32)));

send_one.send(1i32);
assert_eq!(await!(buffered.next()), Some(Ok(1i32)));

assert_eq!(await!(buffered.next()), None);

default fn zip<St>(self, other: St) -> Zip<Self, St> where
    St: Stream, 

An adapter for zipping two streams together.

The zipped stream waits for both streams to produce an item, and then returns that pair. If either stream ends then the zipped stream will also end.

Examples

use futures::executor::block_on;
use futures::stream::{self, StreamExt};

let stream1 = stream::iter(1..=3);
let stream2 = stream::iter(5..=10);

let vec = block_on(stream1.zip(stream2)
                          .collect::<Vec<_>>());
assert_eq!(vec![(1, 5), (2, 6), (3, 7)], vec);

default fn chain<St>(self, other: St) -> Chain<Self, St> where
    St: Stream<Item = Self::Item>, 

Adapter for chaining two stream.

The resulting stream emits elements from the first stream, and when first stream reaches the end, emits the elements from the second stream.

use futures::executor::block_on;
use futures::stream::{self, StreamExt};

let stream1 = stream::iter(vec![Ok(10), Err(false)]);
let stream2 = stream::iter(vec![Err(true), Ok(20)]);

let stream = stream1.chain(stream2);

let result: Vec<_> = block_on(stream.collect());
assert_eq!(result, vec![
    Ok(10),
    Err(false),
    Err(true),
    Ok(20),
]);

default fn peekable(self) -> Peekable<Self>

Creates a new stream which exposes a peek method.

Calling peek returns a reference to the next item in the stream.

default fn chunks(self, capacity: usize) -> Chunks<Self>

An adaptor for chunking up items of the stream inside a vector.

This combinator will attempt to pull items from this stream and buffer them into a local vector. At most capacity items will get buffered before they're yielded from the returned stream.

Note that the vectors returned from this iterator may not always have capacity elements. If the underlying stream ended and only a partial vector was created, it'll be returned. Additionally if an error happens from the underlying stream then the currently buffered items will be yielded.

This method is only available when the std feature of this library is activated, and it is activated by default.

Panics

This method will panic of capacity is zero.

default fn select<St>(self, other: St) -> Select<Self, St> where
    St: Stream<Item = Self::Item>, 

This combinator will attempt to pull items from both streams. Each stream will be polled in a round-robin fashion, and whenever a stream is ready to yield an item that item is yielded.

After one of the two input stream completes, the remaining one will be polled exclusively. The returned stream completes when both input streams have completed.

Note that this method consumes both streams and returns a wrapped version of them.

ⓘImportant traits for Forward<St, Si>
default fn forward<S>(self, sink: S) -> Forward<Self, S> where
    S: Sink<Self::Ok>,
    Self: TryStream<Error = <S as Sink<Self::Ok>>::SinkError>, 

A future that completes after the given stream has been fully processed into the sink and the sink has been flushed and closed.

This future will drive the stream to keep producing items until it is exhausted, sending each item to the sink. It will complete once the stream is exhausted, the sink has received and flushed all items, and the sink is closed.

default fn split<Item>(self) -> (SplitSink<Self, Item>, SplitStream<Self>) where
    Self: Sink<Item>, 

Splits this Stream + Sink object into separate Stream and Sink objects.

This can be useful when you want to split ownership between tasks, or allow direct interaction between the two objects (e.g. via Sink::send_all).

This method is only available when the std feature of this library is activated, and it is activated by default.

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

Do something with each item of this stream, afterwards passing it on.

This is similar to the Iterator::inspect method in the standard library where it allows easily inspecting each value as it passes through the stream, for example to debug what's going on.

default fn left_stream<B>(self) -> Either<Self, B> where
    B: Stream<Item = Self::Item>, 

Wrap this stream in an Either stream, making it the left-hand variant of that Either.

This can be used in combination with the right_stream method to write if statements that evaluate to different streams in different branches.

default fn right_stream<B>(self) -> Either<B, Self> where
    B: Stream<Item = Self::Item>, 

Wrap this stream in an Either stream, making it the right-hand variant of that Either.

This can be used in combination with the left_stream method to write if statements that evaluate to different streams in different branches.

default fn poll_next_unpin(
    &mut self,
    cx: &mut Context
) -> Poll<Option<Self::Item>> where
    Self: Unpin, 

A convenience method for calling Stream::poll_next on Unpin stream types.

ⓘImportant traits for SelectNextSome<'a, St>
default fn select_next_some(&mut self) -> SelectNextSome<Self> where
    Self: Unpin + FusedStream, 

Returns a [Future] that resolves when the next item in this stream is ready.

This is similar to the next method, but it won't resolve to None if used on an empty Stream. Instead, the returned future type will return [true] from FusedFuture::is_terminated when the Stream is empty, allowing select_next_some to be easily used with the select! macro.

If the future is polled after this Stream is empty it will panic. Using the future with a FusedFuture-aware primitive like the select! macro will prevent this.

Examples

#![feature(async_await, await_macro, futures_api)]
use futures::{future, select};
use futures::stream::{StreamExt, FuturesUnordered};

let mut fut = future::ready(1);
let mut async_tasks = FuturesUnordered::new();
let mut total = 0;
loop {
    select! {
        num = fut => {
            // First, the `ready` future completes.
            total += num;
            // Then we spawn a new task onto `async_tasks`,
            async_tasks.push(async { 5 });
        },
        // On the next iteration of the loop, the task we spawned
        // completes.
        num = async_tasks.select_next_some() => {
            total += num;
        }
        // Finally, both the `ready` future and `async_tasks` have
        // finished, so we enter the `complete` branch.
        complete => break,
    }
}
assert_eq!(total, 6);
Loading content...

Implementors

impl<T> StreamExt for T where
    T: Stream + ?Sized, [src]

ⓘImportant traits for Next<'_, St>
default fn next(&mut self) -> Next<Self> where
    Self: Unpin, [src]

ⓘImportant traits for StreamFuture<St>
default fn into_future(self) -> StreamFuture<Self> where
    Self: Unpin, [src]

default fn map<T, F>(self, f: F) -> Map<Self, F> where
    F: FnMut(Self::Item) -> T, [src]

default fn filter<Fut, F>(self, f: F) -> Filter<Self, Fut, F> where
    F: FnMut(&Self::Item) -> Fut,
    Fut: Future<Output = bool>, [src]

default fn filter_map<Fut, T, F>(self, f: F) -> FilterMap<Self, Fut, F> where
    F: FnMut(Self::Item) -> Fut,
    Fut: Future<Output = Option<T>>, [src]

default fn then<Fut, F>(self, f: F) -> Then<Self, Fut, F> where
    F: FnMut(Self::Item) -> Fut,
    Fut: Future, [src]

ⓘImportant traits for Collect<St, C>
default fn collect<C>(self) -> Collect<Self, C> where
    C: Default + Extend<Self::Item>, [src]

ⓘImportant traits for Concat<St>
default fn concat(self) -> Concat<Self> where
    Self::Item: Extend<<Self::Item as IntoIterator>::Item>,
    Self::Item: IntoIterator,
    Self::Item: Default, [src]

ⓘImportant traits for Fold<St, Fut, T, F>
default fn fold<T, Fut, F>(self, init: T, f: F) -> Fold<Self, Fut, T, F> where
    F: FnMut(T, Self::Item) -> Fut,
    Fut: Future<Output = T>, [src]

default fn flatten(self) -> Flatten<Self> where
    Self::Item: Stream, [src]

default fn skip_while<Fut, F>(self, f: F) -> SkipWhile<Self, Fut, F> where
    F: FnMut(&Self::Item) -> Fut,
    Fut: Future<Output = bool>, [src]

default fn take_while<Fut, F>(self, f: F) -> TakeWhile<Self, Fut, F> where
    F: FnMut(&Self::Item) -> Fut,
    Fut: Future<Output = bool>, [src]

ⓘImportant traits for ForEach<St, Fut, F>
default fn for_each<Fut, F>(self, f: F) -> ForEach<Self, Fut, F> where
    F: FnMut(Self::Item) -> Fut,
    Fut: Future<Output = ()>, [src]

ⓘImportant traits for ForEachConcurrent<St, Fut, F>
default fn for_each_concurrent<Fut, F, impl Into>>(
    self,
    limit: impl Into>,
    f: F
) -> ForEachConcurrent<Self, Fut, F> where
    F: FnMut(Self::Item) -> Fut,
    Fut: Future<Output = ()>,
    impl Into>: Into<Option<usize>>, [src]

default fn take(self, n: u64) -> Take<Self>[src]

default fn skip(self, n: u64) -> Skip<Self>[src]

default fn fuse(self) -> Fuse<Self>[src]

ⓘImportant traits for &'_ mut F
default fn by_ref(&mut self) -> &mut Self[src]

default fn catch_unwind(self) -> CatchUnwind<Self> where
    Self: UnwindSafe, [src]

ⓘImportant traits for Pin<P>
default fn boxed(self) -> Pin<Box<Self>>[src]

default fn buffered(self, n: usize) -> Buffered<Self> where
    Self::Item: Future, [src]

default fn buffer_unordered(self, n: usize) -> BufferUnordered<Self> where
    Self::Item: Future, [src]

default fn zip<St>(self, other: St) -> Zip<Self, St> where
    St: Stream, [src]

default fn chain<St>(self, other: St) -> Chain<Self, St> where
    St: Stream<Item = Self::Item>, [src]

default fn peekable(self) -> Peekable<Self>[src]

default fn chunks(self, capacity: usize) -> Chunks<Self>[src]

default fn select<St>(self, other: St) -> Select<Self, St> where
    St: Stream<Item = Self::Item>, [src]

ⓘImportant traits for Forward<St, Si>
default fn forward<S>(self, sink: S) -> Forward<Self, S> where
    S: Sink<Self::Ok>,
    Self: TryStream<Error = <S as Sink<Self::Ok>>::SinkError>, [src]

default fn split<Item>(self) -> (SplitSink<Self, Item>, SplitStream<Self>) where
    Self: Sink<Item>, [src]

default fn inspect<F>(self, f: F) -> Inspect<Self, F> where
    F: FnMut(&Self::Item), [src]

default fn left_stream<B>(self) -> Either<Self, B> where
    B: Stream<Item = Self::Item>, [src]

default fn right_stream<B>(self) -> Either<B, Self> where
    B: Stream<Item = Self::Item>, [src]

default fn poll_next_unpin(
    &mut self,
    cx: &mut Context
) -> Poll<Option<Self::Item>> where
    Self: Unpin, [src]

ⓘImportant traits for SelectNextSome<'a, St>
default fn select_next_some(&mut self) -> SelectNextSome<Self> where
    Self: Unpin + FusedStream, [src]

Loading content...