@@ -136,34 +136,31 @@ pub trait Stream {
136
136
/// use async_std::prelude::*;
137
137
/// use async_std::stream;
138
138
///
139
- /// let mut s = stream::repeat(9).take(3);
139
+ /// let mut s = stream::repeat::<u32>(42).take(3);
140
+ /// assert!(s.all(|x| x == 42).await);
140
141
///
141
- /// while let Some(v) = s.next().await {
142
- /// assert_eq!(v, 9);
143
- /// }
144
142
/// #
145
143
/// # }) }
146
144
/// ```
147
- /// ```
148
- /// let a = [1, 2, 3];
149
145
///
150
- /// assert!(a.iter().all(|&x| x > 0));
146
+ /// Empty stream:
151
147
///
152
- /// assert!(!a.iter().all(|&x| x > 2));
153
148
/// ```
149
+ /// # fn main() { async_std::task::block_on(async {
150
+ /// #
151
+ /// use async_std::prelude::*;
152
+ /// use async_std::stream;
154
153
///
155
- /// Stopping at the first `false`:
154
+ /// let mut s = stream::empty::<u32>();
155
+ /// assert!(s.all(|_| false).await);
156
156
///
157
+ /// #
158
+ /// # }) }
157
159
/// ```
158
- /// let a = [1, 2, 3];
159
- ///
160
- /// let mut iter = a.iter();
161
- ///
162
- /// assert!(!iter.all(|&x| x != 2));
160
+ /// Stopping at the first `false`:
163
161
///
164
- /// // we can still use `iter`, as there are more elements.
165
- /// assert_eq!(iter.next(), Some(&3));
166
- /// ```
162
+ /// TODO: add example here
163
+ /// TODO: add more examples
167
164
#[ inline]
168
165
fn all < F > ( & mut self , f : F ) -> AllFuture < ' _ , Self , F , Self :: Item >
169
166
where
@@ -235,9 +232,9 @@ impl<S: futures::Stream> futures::Stream for Take<S> {
235
232
}
236
233
}
237
234
235
+ #[ derive( Debug ) ]
238
236
pub struct AllFuture < ' a , S , F , T >
239
237
where
240
- S : ?Sized ,
241
238
F : FnMut ( T ) -> bool ,
242
239
{
243
240
stream : & ' a mut S ,
@@ -248,7 +245,6 @@ where
248
245
249
246
impl < ' a , S , F , T > AllFuture < ' a , S , F , T >
250
247
where
251
- S : ?Sized ,
252
248
F : FnMut ( T ) -> bool ,
253
249
{
254
250
pin_utils:: unsafe_pinned!( stream: & ' a mut S ) ;
@@ -258,8 +254,9 @@ where
258
254
259
255
impl < S , F > Future for AllFuture < ' _ , S , F , S :: Item >
260
256
where
261
- S : futures:: Stream + Unpin + ? Sized ,
257
+ S : futures:: Stream + Unpin + Sized ,
262
258
F : FnMut ( S :: Item ) -> bool ,
259
+ S :: Item : std:: fmt:: Debug ,
263
260
{
264
261
type Output = bool ;
265
262
@@ -268,9 +265,15 @@ where
268
265
let next = futures:: ready!( self . as_mut( ) . stream( ) . poll_next( cx) ) ;
269
266
match next {
270
267
Some ( v) => {
271
- // me: *screams*
272
- * self . as_mut ( ) . result ( ) = ( self . as_mut ( ) . f ( ) ) ( v) ;
273
- Poll :: Pending
268
+ let result = ( self . as_mut ( ) . f ( ) ) ( v) ;
269
+ * self . as_mut ( ) . result ( ) = result;
270
+ if result {
271
+ // don't forget to wake this task again to pull the next item from stream
272
+ cx. waker ( ) . wake_by_ref ( ) ;
273
+ Poll :: Pending
274
+ } else {
275
+ Poll :: Ready ( false )
276
+ }
274
277
}
275
278
None => Poll :: Ready ( self . result ) ,
276
279
}
0 commit comments