add stream::{Sum,Product} (#343)

Signed-off-by: Yoshua Wuyts <yoshuawuyts@gmail.com>

Author: yoshuawuyts

src/stream/mod.rs

@@ -42,19 +42,23 @@ cfg_if! {
     if #[cfg(any(feature = "unstable", feature = "docs"))] {
         mod double_ended_stream;
         mod exact_size_stream;
-        mod fused_stream;
         mod extend;
         mod from_stream;
-        mod into_stream;
+        mod fused_stream;
         mod interval;
+        mod into_stream;
+        mod product;
+        mod sum;
 
         pub use double_ended_stream::DoubleEndedStream;
         pub use exact_size_stream::ExactSizeStream;
         pub use extend::Extend;
         pub use from_stream::FromStream;
         pub use fused_stream::FusedStream;
-        pub use into_stream::IntoStream;
         pub use interval::{interval, Interval};
+        pub use into_stream::IntoStream;
+        pub use product::Product;
+        pub use sum::Sum;
 
         pub use stream::Merge;
     }

src/stream/product.rs

@@ -0,0 +1,23 @@
+use crate::future::Future;
+use crate::stream::Stream;
+
+/// Trait to represent types that can be created by productming up a stream.
+///
+/// This trait is used to implement the [`product`] method on streams. Types which
+/// implement the trait can be generated by the [`product`] method. Like
+/// [`FromStream`] this trait should rarely be called directly and instead
+/// interacted with through [`Stream::product`].
+///
+/// [`product`]: trait.Product.html#tymethod.product
+/// [`FromStream`]: trait.FromStream.html
+/// [`Stream::product`]: trait.Stream.html#method.product
+#[cfg_attr(feature = "docs", doc(cfg(unstable)))]
+#[cfg(any(feature = "unstable", feature = "docs"))]
+pub trait Product<A = Self>: Sized {
+    /// Method which takes a stream and generates `Self` from the elements by
+    /// multiplying the items.
+    fn product<S, F>(stream: S) -> F
+    where
+        S: Stream<Item = A>,
+        F: Future<Output = Self>;
+}

src/stream/sum.rs

@@ -0,0 +1,23 @@
+use crate::future::Future;
+use crate::stream::Stream;
+
+/// Trait to represent types that can be created by summing up a stream.
+///
+/// This trait is used to implement the [`sum`] method on streams. Types which
+/// implement the trait can be generated by the [`sum`] method. Like
+/// [`FromStream`] this trait should rarely be called directly and instead
+/// interacted with through [`Stream::sum`].
+///
+/// [`sum`]: trait.Sum.html#tymethod.sum
+/// [`FromStream`]: trait.FromStream.html
+/// [`Stream::sum`]: trait.Stream.html#method.sum
+#[cfg_attr(feature = "docs", doc(cfg(unstable)))]
+#[cfg(any(feature = "unstable", feature = "docs"))]
+pub trait Sum<A = Self>: Sized {
+    /// Method which takes a stream and generates `Self` from the elements by
+    /// "summing up" the items.
+    fn sum<S, F>(stream: S) -> F
+    where
+        S: Stream<Item = A>,
+        F: Future<Output = Self>;
+}

src/task/blocking.rs

@@ -1,6 +1,5 @@
 //! A thread pool for running blocking functions asynchronously.
 
-use std::future::Future;
 use std::sync::atomic::{AtomicU64, Ordering};
 use std::thread;
 use std::time::Duration;