futures_util/stream/
once.rs

1use futures_core::{Poll, Async, Stream, IntoFuture, Future};
2use futures_core::task;
3
4/// A stream which emits single element and then EOF.
5///
6/// This stream will never block and is always ready.
7#[derive(Debug)]
8#[must_use = "streams do nothing unless polled"]
9pub struct Once<F>(Option<F>);
10
11/// Creates a stream of single element
12///
13/// ```rust
14/// # extern crate futures;
15/// # extern crate futures_executor;
16/// use futures::prelude::*;
17/// use futures::stream;
18/// use futures_executor::block_on;
19///
20/// # fn main() {
21/// let mut stream = stream::once::<Result<(), _>>(Err(17));
22/// let collected: Result<Vec<_>, _> = block_on(stream.collect());
23/// assert_eq!(collected, Err(17));
24///
25/// let mut stream = stream::once::<Result<_, ()>>(Ok(92));
26/// let collected: Result<Vec<_>, _> = block_on(stream.collect());
27/// assert_eq!(collected, Ok(vec![92]));
28/// # }
29/// ```
30pub fn once<F: IntoFuture>(item: F) -> Once<F::Future> {
31    Once(Some(item.into_future()))
32}
33
34impl<F: Future> Stream for Once<F> {
35    type Item = F::Item;
36    type Error = F::Error;
37
38    fn poll_next(&mut self, cx: &mut task::Context) -> Poll<Option<F::Item>, F::Error> {
39        if let Some(mut f) = self.0.take() {
40            match f.poll(cx)? {
41                Async::Ready(x) => Ok(Async::Ready(Some(x))),
42                Async::Pending => {
43                    self.0 = Some(f);
44                    Ok(Async::Pending)
45                }
46            }
47        } else {
48            Ok(Async::Ready(None))
49        }
50    }
51}
futures_util/stream/once.rs

futures_util/stream/
once.rs
