Merge #149 #150

149: update deps r=stjepang a=yoshuawuyts

Updates all deps. Thanks!

150: split stream into multiple files r=stjepang a=yoshuawuyts

This splits `stream/mod.rs`'s combinators into multiple files, making it easier to contribute combinators. Additionally we've renamed `MinBy` to `MinByFuture` to make it the same as the other private futures. Ref #146 #129. Thanks!

Co-authored-by: Yoshua Wuyts <[email protected]>
Co-authored-by: Stjepan Glavina <[email protected]>

Author: 3 people

Cargo.toml

@@ -30,18 +30,18 @@ crossbeam-channel = "0.3.9"
 futures-core-preview = "0.3.0-alpha.18"
 futures-io-preview = "0.3.0-alpha.18"
 futures-timer = "0.4.0"
-lazy_static = "1.3.0"
+lazy_static = "1.4.0"
 log = { version = "0.4.8", features = ["kv_unstable"] }
 memchr = "2.2.1"
 mio = "0.6.19"
 mio-uds = "0.6.7"
-num_cpus = "1.10.0"
+num_cpus = "1.10.1"
 pin-utils = "0.1.0-alpha.4"
 slab = "0.4.2"
 
 [dev-dependencies]
-femme = "1.1.0"
-surf = "1.0.1"
+femme = "1.2.0"
+surf = "1.0.2"
 tempdir = "0.3.7"
 
 [dev-dependencies.futures-preview]

src/stream/mod.rs

@@ -27,7 +27,6 @@ pub use repeat::{repeat, Repeat};
 pub use stream::{Stream, Take};
 
 mod empty;
-mod min_by;
 mod once;
 mod repeat;
 mod stream;

src/stream/stream/all.rs

@@ -0,0 +1,52 @@
+use crate::future::Future;
+use crate::task::{Context, Poll};
+
+use std::marker::PhantomData;
+use std::pin::Pin;
+
+#[derive(Debug)]
+pub struct AllFuture<'a, S, F, T>
+where
+    F: FnMut(T) -> bool,
+{
+    pub(crate) stream: &'a mut S,
+    pub(crate) f: F,
+    pub(crate) result: bool,
+    pub(crate) __item: PhantomData<T>,
+}
+
+impl<'a, S, F, T> AllFuture<'a, S, F, T>
+where
+    F: FnMut(T) -> bool,
+{
+    pin_utils::unsafe_pinned!(stream: &'a mut S);
+    pin_utils::unsafe_unpinned!(result: bool);
+    pin_utils::unsafe_unpinned!(f: F);
+}
+
+impl<S, F> Future for AllFuture<'_, S, F, S::Item>
+where
+    S: futures_core::stream::Stream + Unpin + Sized,
+    F: FnMut(S::Item) -> bool,
+{
+    type Output = bool;
+
+    fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
+        use futures_core::stream::Stream;
+        let next = futures_core::ready!(self.as_mut().stream().poll_next(cx));
+        match next {
+            Some(v) => {
+                let result = (self.as_mut().f())(v);
+                *self.as_mut().result() = result;
+                if result {
+                    // don't forget to wake this task again to pull the next item from stream
+                    cx.waker().wake_by_ref();
+                    Poll::Pending
+                } else {
+                    Poll::Ready(false)
+                }
+            }
+            None => Poll::Ready(self.result),
+        }
+    }
+}

src/stream/stream/any.rs

@@ -0,0 +1,52 @@
+use crate::future::Future;
+use crate::task::{Context, Poll};
+
+use std::marker::PhantomData;
+use std::pin::Pin;
+
+#[derive(Debug)]
+pub struct AnyFuture<'a, S, F, T>
+where
+    F: FnMut(T) -> bool,
+{
+    pub(crate) stream: &'a mut S,
+    pub(crate) f: F,
+    pub(crate) result: bool,
+    pub(crate) __item: PhantomData<T>,
+}
+
+impl<'a, S, F, T> AnyFuture<'a, S, F, T>
+where
+    F: FnMut(T) -> bool,
+{
+    pin_utils::unsafe_pinned!(stream: &'a mut S);
+    pin_utils::unsafe_unpinned!(result: bool);
+    pin_utils::unsafe_unpinned!(f: F);
+}
+
+impl<S, F> Future for AnyFuture<'_, S, F, S::Item>
+where
+    S: futures_core::stream::Stream + Unpin + Sized,
+    F: FnMut(S::Item) -> bool,
+{
+    type Output = bool;
+
+    fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
+        use futures_core::stream::Stream;
+        let next = futures_core::ready!(self.as_mut().stream().poll_next(cx));
+        match next {
+            Some(v) => {
+                let result = (self.as_mut().f())(v);
+                *self.as_mut().result() = result;
+                if result {
+                    Poll::Ready(true)
+                } else {
+                    // don't forget to wake this task again to pull the next item from stream
+                    cx.waker().wake_by_ref();
+                    Poll::Pending
+                }
+            }
+            None => Poll::Ready(self.result),
+        }
+    }
+}

src/stream/min_by.rs renamed to src/stream/stream/min_by.rs

@@ -1,31 +1,31 @@
 use std::cmp::Ordering;
 use std::pin::Pin;
 
-use super::stream::Stream;
 use crate::future::Future;
+use crate::stream::Stream;
 use crate::task::{Context, Poll};
 
 /// A future that yields the minimum item in a stream by a given comparison function.
 #[derive(Clone, Debug)]
-pub struct MinBy<S: Stream, F> {
+pub struct MinByFuture<S: Stream, F> {
     stream: S,
     compare: F,
     min: Option<S::Item>,
 }
 
-impl<S: Stream + Unpin, F> Unpin for MinBy<S, F> {}
+impl<S: Stream + Unpin, F> Unpin for MinByFuture<S, F> {}
 
-impl<S: Stream + Unpin, F> MinBy<S, F> {
+impl<S: Stream + Unpin, F> MinByFuture<S, F> {
     pub(super) fn new(stream: S, compare: F) -> Self {
-        MinBy {
+        MinByFuture {
             stream,
             compare,
             min: None,
         }
     }
 }
 
-impl<S, F> Future for MinBy<S, F>
+impl<S, F> Future for MinByFuture<S, F>
 where
     S: futures_core::stream::Stream + Unpin,
     S::Item: Copy,

src/stream/stream.rs renamed to src/stream/stream/mod.rs

@@ -21,16 +21,24 @@
 //! # }) }
 //! ```
 
-use std::cmp::Ordering;
-use std::pin::Pin;
+mod all;
+mod any;
+mod min_by;
+mod next;
+mod take;
 
-use cfg_if::cfg_if;
+pub use take::Take;
+
+use all::AllFuture;
+use any::AnyFuture;
+use min_by::MinByFuture;
+use next::NextFuture;
 
-use super::min_by::MinBy;
-use crate::future::Future;
-use crate::task::{Context, Poll};
+use std::cmp::Ordering;
 use std::marker::PhantomData;
 
+use cfg_if::cfg_if;
+
 cfg_if! {
     if #[cfg(feature = "docs")] {
         #[doc(hidden)]
@@ -145,12 +153,12 @@ pub trait Stream {
     /// #
     /// # }) }
     /// ```
-    fn min_by<F>(self, compare: F) -> MinBy<Self, F>
+    fn min_by<F>(self, compare: F) -> MinByFuture<Self, F>
     where
         Self: Sized + Unpin,
         F: FnMut(&Self::Item, &Self::Item) -> Ordering,
     {
-        MinBy::new(self, compare)
+        MinByFuture::new(self, compare)
     }
 
     /// Tests if every element of the stream matches a predicate.
@@ -278,142 +286,3 @@ impl<T: futures_core::stream::Stream + Unpin + ?Sized> Stream for T {
         NextFuture { stream: self }
     }
 }
-
-#[doc(hidden)]
-#[allow(missing_debug_implementations)]
-pub struct NextFuture<'a, T: Unpin + ?Sized> {
-    stream: &'a mut T,
-}
-
-impl<T: futures_core::stream::Stream + Unpin + ?Sized> Future for NextFuture<'_, T> {
-    type Output = Option<T::Item>;
-
-    fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
-        Pin::new(&mut *self.stream).poll_next(cx)
-    }
-}
-
-/// A stream that yields the first `n` items of another stream.
-#[derive(Clone, Debug)]
-pub struct Take<S> {
-    stream: S,
-    remaining: usize,
-}
-
-impl<S: Unpin> Unpin for Take<S> {}
-
-impl<S: futures_core::stream::Stream> Take<S> {
-    pin_utils::unsafe_pinned!(stream: S);
-    pin_utils::unsafe_unpinned!(remaining: usize);
-}
-
-impl<S: futures_core::stream::Stream> futures_core::stream::Stream for Take<S> {
-    type Item = S::Item;
-
-    fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<S::Item>> {
-        if self.remaining == 0 {
-            Poll::Ready(None)
-        } else {
-            let next = futures_core::ready!(self.as_mut().stream().poll_next(cx));
-            match next {
-                Some(_) => *self.as_mut().remaining() -= 1,
-                None => *self.as_mut().remaining() = 0,
-            }
-            Poll::Ready(next)
-        }
-    }
-}
-
-#[derive(Debug)]
-pub struct AllFuture<'a, S, F, T>
-where
-    F: FnMut(T) -> bool,
-{
-    stream: &'a mut S,
-    f: F,
-    result: bool,
-    __item: PhantomData<T>,
-}
-
-impl<'a, S, F, T> AllFuture<'a, S, F, T>
-where
-    F: FnMut(T) -> bool,
-{
-    pin_utils::unsafe_pinned!(stream: &'a mut S);
-    pin_utils::unsafe_unpinned!(result: bool);
-    pin_utils::unsafe_unpinned!(f: F);
-}
-
-impl<S, F> Future for AllFuture<'_, S, F, S::Item>
-where
-    S: futures_core::stream::Stream + Unpin + Sized,
-    F: FnMut(S::Item) -> bool,
-{
-    type Output = bool;
-
-    fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
-        use futures_core::stream::Stream;
-        let next = futures_core::ready!(self.as_mut().stream().poll_next(cx));
-        match next {
-            Some(v) => {
-                let result = (self.as_mut().f())(v);
-                *self.as_mut().result() = result;
-                if result {
-                    // don't forget to wake this task again to pull the next item from stream
-                    cx.waker().wake_by_ref();
-                    Poll::Pending
-                } else {
-                    Poll::Ready(false)
-                }
-            }
-            None => Poll::Ready(self.result),
-        }
-    }
-}
-
-#[derive(Debug)]
-pub struct AnyFuture<'a, S, F, T>
-where
-    F: FnMut(T) -> bool,
-{
-    stream: &'a mut S,
-    f: F,
-    result: bool,
-    __item: PhantomData<T>,
-}
-
-impl<'a, S, F, T> AnyFuture<'a, S, F, T>
-where
-    F: FnMut(T) -> bool,
-{
-    pin_utils::unsafe_pinned!(stream: &'a mut S);
-    pin_utils::unsafe_unpinned!(result: bool);
-    pin_utils::unsafe_unpinned!(f: F);
-}
-
-impl<S, F> Future for AnyFuture<'_, S, F, S::Item>
-where
-    S: futures_core::stream::Stream + Unpin + Sized,
-    F: FnMut(S::Item) -> bool,
-{
-    type Output = bool;
-
-    fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
-        use futures_core::stream::Stream;
-        let next = futures_core::ready!(self.as_mut().stream().poll_next(cx));
-        match next {
-            Some(v) => {
-                let result = (self.as_mut().f())(v);
-                *self.as_mut().result() = result;
-                if result {
-                    Poll::Ready(true)
-                } else {
-                    // don't forget to wake this task again to pull the next item from stream
-                    cx.waker().wake_by_ref();
-                    Poll::Pending
-                }
-            }
-            None => Poll::Ready(self.result),
-        }
-    }
-}

src/stream/stream/next.rs

@@ -0,0 +1,17 @@
+use crate::future::Future;
+use crate::task::{Context, Poll};
+use std::pin::Pin;
+
+#[doc(hidden)]
+#[allow(missing_debug_implementations)]
+pub struct NextFuture<'a, T: Unpin + ?Sized> {
+    pub(crate) stream: &'a mut T,
+}
+
+impl<T: futures_core::stream::Stream + Unpin + ?Sized> Future for NextFuture<'_, T> {
+    type Output = Option<T::Item>;
+
+    fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
+        Pin::new(&mut *self.stream).poll_next(cx)
+    }
+}