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 /// ```
once<Fut: Future>(future: Fut) -> Once<Fut>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> {
new(future: Fut) -> Self35 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
poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>>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
size_hint(&self) -> (usize, Option<usize>)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> {
is_terminated(&self) -> bool64 fn is_terminated(&self) -> bool {
65 self.future.is_none()
66 }
67 }
68