futures_util/stream/
once.rs

Help
1use super::assert_stream;
2use core::pin::Pin;
3use futures_core::future::Future;
4use futures_core::ready;
5use futures_core::stream::{FusedStream, Stream};
6use futures_core::task::{Context, Poll};
7use pin_project_lite::pin_project;
8
9/// Creates a stream of a single element.
10///
11/// ```
12/// # futures::executor::block_on(async {
13/// use futures::stream::{self, StreamExt};
14///
15/// let stream = stream::once(async { 17 });
16/// let collected = stream.collect::<Vec<i32>>().await;
17/// assert_eq!(collected, vec![17]);
18/// # });
19/// ```
20pub fn once<Fut: Future>(future: Fut) -> Once<Fut> {
21    assert_stream::<Fut::Output, _>(Once::new(future))
22}
23
24pin_project! {
25    /// A stream which emits single element and then EOF.
26    #[derive(Debug)]
27    #[must_use = "streams do nothing unless polled"]
28    pub struct Once<Fut> {
29        #[pin]
30        future: Option<Fut>
31    }
32}
33
34impl<Fut> Once<Fut> {
35    pub(crate) fn new(future: Fut) -> Self {
36        Self { future: Some(future) }
37    }
38}
39
40impl<Fut: Future> Stream for Once<Fut> {
41    type Item = Fut::Output;
42
43    fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
44        let mut this = self.project();
45        let v = match this.future.as_mut().as_pin_mut() {
46            Some(fut) => ready!(fut.poll(cx)),
47            None => return Poll::Ready(None),
48        };
49
50        this.future.set(None);
51        Poll::Ready(Some(v))
52    }
53
54    fn size_hint(&self) -> (usize, Option<usize>) {
55        if self.future.is_some() {
56            (1, Some(1))
57        } else {
58            (0, Some(0))
59        }
60    }
61}
62
63impl<Fut: Future> FusedStream for Once<Fut> {
64    fn is_terminated(&self) -> bool {
65        self.future.is_none()
66    }
67}
futures_util/stream/once.rs

futures_util/stream/
once.rs
