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: Projection<'_, Fut> = self.project();
45 let v: ::Output = match this.future.as_mut().as_pin_mut() {
46 Some(fut: Pin<&mut 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}
68