tokio_stream/
once.rs

1use crate::Stream;
2
3use core::pin::Pin;
4use core::task::{Context, Poll};
5
6/// Stream for the [`once`](fn@once) function.
7#[derive(Debug)]
8#[must_use = "streams do nothing unless polled"]
9pub struct Once<T> {
10    value: Option<T>,
11}
12
13impl<I> Unpin for Once<I> {}
14
15/// Creates a stream that emits an element exactly once.
16///
17/// The returned stream is immediately ready and emits the provided value once.
18///
19/// # Examples
20///
21/// ```
22/// use tokio_stream::{self as stream, StreamExt};
23///
24/// # #[tokio::main(flavor = "current_thread")]
25/// # async fn main() {
26/// // one is the loneliest number
27/// let mut one = stream::once(1);
28///
29/// assert_eq!(Some(1), one.next().await);
30///
31/// // just one, that's all we get
32/// assert_eq!(None, one.next().await);
33/// # }
34/// ```
35pub fn once<T>(value: T) -> Once<T> {
36    Once { value: Some(value) }
37}
38
39impl<T> Stream for Once<T> {
40    type Item = T;
41
42    fn poll_next(mut self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll<Option<T>> {
43        #[cfg(feature = "rt")]
44        {
45            use tokio::task::coop;
46
47            let coop = std::task::ready!(coop::poll_proceed(_cx));
48
49            coop.made_progress();
50        }
51
52        Poll::Ready(self.value.take())
53    }
54
55    fn size_hint(&self) -> (usize, Option<usize>) {
56        if self.value.is_some() {
57            (1, Some(1))
58        } else {
59            (0, Some(0))
60        }
61    }
62}
tokio_stream/once.rs

tokio_stream/
once.rs
