1 //! An unbounded set of streams
2
3 use core::fmt::{self, Debug};
4 use core::iter::FromIterator;
5 use core::pin::Pin;
6
7 use futures_core::stream::{Stream, FusedStream};
8 use futures_core::task::{Context, Poll};
9
10 use crate::stream::{StreamExt, StreamFuture, FuturesUnordered};
11
12 /// An unbounded set of streams
13 ///
14 /// This "combinator" provides the ability to maintain a set of streams
15 /// and drive them all to completion.
16 ///
17 /// Streams are pushed into this set and their realized values are
18 /// yielded as they become ready. Streams will only be polled when they
19 /// generate notifications. This allows to coordinate a large number of streams.
20 ///
21 /// Note that you can create a ready-made `SelectAll` via the
22 /// `select_all` function in the `stream` module, or you can start with an
23 /// empty set with the `SelectAll::new` constructor.
24 #[must_use = "streams do nothing unless polled"]
25 pub struct SelectAll<St> {
26 inner: FuturesUnordered<StreamFuture<St>>,
27 }
28
29 impl<St: Debug> Debug for SelectAll<St> {
fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result30 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
31 write!(f, "SelectAll {{ ... }}")
32 }
33 }
34
35 impl<St: Stream + Unpin> SelectAll<St> {
36 /// Constructs a new, empty `SelectAll`
37 ///
38 /// The returned `SelectAll` does not contain any streams and, in this
39 /// state, `SelectAll::poll` will return `Poll::Ready(None)`.
new() -> SelectAll<St>40 pub fn new() -> SelectAll<St> {
41 SelectAll { inner: FuturesUnordered::new() }
42 }
43
44 /// Returns the number of streams contained in the set.
45 ///
46 /// This represents the total number of in-flight streams.
len(&self) -> usize47 pub fn len(&self) -> usize {
48 self.inner.len()
49 }
50
51 /// Returns `true` if the set contains no streams
is_empty(&self) -> bool52 pub fn is_empty(&self) -> bool {
53 self.inner.is_empty()
54 }
55
56 /// Push a stream into the set.
57 ///
58 /// This function submits the given stream to the set for managing. This
59 /// function will not call `poll` on the submitted stream. The caller must
60 /// ensure that `SelectAll::poll` is called in order to receive task
61 /// notifications.
push(&mut self, stream: St)62 pub fn push(&mut self, stream: St) {
63 self.inner.push(stream.into_future());
64 }
65 }
66
67 impl<St: Stream + Unpin> Default for SelectAll<St> {
default() -> SelectAll<St>68 fn default() -> SelectAll<St> {
69 SelectAll::new()
70 }
71 }
72
73 impl<St: Stream + Unpin> Stream for SelectAll<St> {
74 type Item = St::Item;
75
poll_next( mut self: Pin<&mut Self>, cx: &mut Context<'_>, ) -> Poll<Option<Self::Item>>76 fn poll_next(
77 mut self: Pin<&mut Self>,
78 cx: &mut Context<'_>,
79 ) -> Poll<Option<Self::Item>> {
80 loop {
81 match ready!(self.inner.poll_next_unpin(cx)) {
82 Some((Some(item), remaining)) => {
83 self.push(remaining);
84 return Poll::Ready(Some(item));
85 }
86 Some((None, _)) => {
87 // `FuturesUnordered` thinks it isn't terminated
88 // because it yielded a Some.
89 // We do not return, but poll `FuturesUnordered`
90 // in the next loop iteration.
91 }
92 None => return Poll::Ready(None),
93 }
94 }
95 }
96 }
97
98 impl<St: Stream + Unpin> FusedStream for SelectAll<St> {
is_terminated(&self) -> bool99 fn is_terminated(&self) -> bool {
100 self.inner.is_terminated()
101 }
102 }
103
104 /// Convert a list of streams into a `Stream` of results from the streams.
105 ///
106 /// This essentially takes a list of streams (e.g. a vector, an iterator, etc.)
107 /// and bundles them together into a single stream.
108 /// The stream will yield items as they become available on the underlying
109 /// streams internally, in the order they become available.
110 ///
111 /// Note that the returned set can also be used to dynamically push more
112 /// futures into the set as they become available.
113 ///
114 /// This function is only available when the `std` or `alloc` feature of this
115 /// library is activated, and it is activated by default.
select_all<I>(streams: I) -> SelectAll<I::Item> where I: IntoIterator, I::Item: Stream + Unpin116 pub fn select_all<I>(streams: I) -> SelectAll<I::Item>
117 where I: IntoIterator,
118 I::Item: Stream + Unpin
119 {
120 let mut set = SelectAll::new();
121
122 for stream in streams {
123 set.push(stream);
124 }
125
126 set
127 }
128
129 impl<St: Stream + Unpin> FromIterator<St> for SelectAll<St> {
from_iter<T: IntoIterator<Item = St>>(iter: T) -> Self130 fn from_iter<T: IntoIterator<Item = St>>(iter: T) -> Self {
131 select_all(iter)
132 }
133 }
134