1 | use super::assert_stream; |
2 | use core::pin::Pin; |
3 | use futures_core::future::Future; |
4 | use futures_core::ready; |
5 | use futures_core::stream::{FusedStream, Stream}; |
6 | use futures_core::task::{Context, Poll}; |
7 | use 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 | /// ``` |
20 | pub fn once<Fut: Future>(future: Fut) -> Once<Fut> { |
21 | assert_stream::<Fut::Output, _>(Once::new(future)) |
22 | } |
23 | |
24 | pin_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 | |
34 | impl<Fut> Once<Fut> { |
35 | pub(crate) fn new(future: Fut) -> Self { |
36 | Self { future: Some(future) } |
37 | } |
38 | } |
39 | |
40 | impl<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 | |
63 | impl<Fut: Future> FusedStream for Once<Fut> { |
64 | fn is_terminated(&self) -> bool { |
65 | self.future.is_none() |
66 | } |
67 | } |
68 | |