/rust/registry/src/index.crates.io-1949cf8c6b5b557f/tokio-stream-0.1.19/src/iter.rs
Line | Count | Source |
1 | | use crate::Stream; |
2 | | |
3 | | use core::pin::Pin; |
4 | | use core::task::{Context, Poll}; |
5 | | |
6 | | /// Stream for the [`iter`](fn@iter) function. |
7 | | #[derive(Debug)] |
8 | | #[must_use = "streams do nothing unless polled"] |
9 | | pub struct Iter<I> { |
10 | | iter: I, |
11 | | #[cfg(not(feature = "rt"))] |
12 | | yield_amt: usize, |
13 | | } |
14 | | |
15 | | impl<I> Unpin for Iter<I> {} |
16 | | |
17 | | /// Converts an `Iterator` into a `Stream` which is always ready |
18 | | /// to yield the next value. |
19 | | /// |
20 | | /// Iterators in Rust don't express the ability to block, so this adapter |
21 | | /// simply always calls `iter.next()` and returns that. |
22 | | /// |
23 | | /// ``` |
24 | | /// # async fn dox() { |
25 | | /// use tokio_stream::{self as stream, StreamExt}; |
26 | | /// |
27 | | /// let mut stream = stream::iter(vec![17, 19]); |
28 | | /// |
29 | | /// assert_eq!(stream.next().await, Some(17)); |
30 | | /// assert_eq!(stream.next().await, Some(19)); |
31 | | /// assert_eq!(stream.next().await, None); |
32 | | /// # } |
33 | | /// ``` |
34 | 0 | pub fn iter<I>(i: I) -> Iter<I::IntoIter> |
35 | 0 | where |
36 | 0 | I: IntoIterator, |
37 | | { |
38 | 0 | Iter { |
39 | 0 | iter: i.into_iter(), |
40 | 0 | #[cfg(not(feature = "rt"))] |
41 | 0 | yield_amt: 0, |
42 | 0 | } |
43 | 0 | } |
44 | | |
45 | | impl<I> Stream for Iter<I> |
46 | | where |
47 | | I: Iterator, |
48 | | { |
49 | | type Item = I::Item; |
50 | | |
51 | 0 | fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<I::Item>> { |
52 | | #[cfg(feature = "rt")] |
53 | | { |
54 | | use tokio::task::coop; |
55 | | |
56 | | let coop = std::task::ready!(coop::poll_proceed(cx)); |
57 | | let item = self.iter.next(); |
58 | | |
59 | | coop.made_progress(); |
60 | | |
61 | | Poll::Ready(item) |
62 | | } |
63 | | |
64 | | #[cfg(not(feature = "rt"))] |
65 | | { |
66 | 0 | if self.yield_amt >= 32 { |
67 | 0 | self.yield_amt = 0; |
68 | | |
69 | 0 | cx.waker().wake_by_ref(); |
70 | | |
71 | 0 | Poll::Pending |
72 | | } else { |
73 | 0 | let item = self.iter.next(); |
74 | | |
75 | 0 | self.yield_amt += 1; |
76 | | |
77 | 0 | Poll::Ready(item) |
78 | | } |
79 | | } |
80 | 0 | } |
81 | | |
82 | 0 | fn size_hint(&self) -> (usize, Option<usize>) { |
83 | 0 | self.iter.size_hint() |
84 | 0 | } |
85 | | } |