From e26eb7a719015f3d8fa6804d2159ec0845093226 Mon Sep 17 00:00:00 2001 From: Kyle Tomsic Date: Sun, 20 Oct 2019 10:05:55 -0400 Subject: [PATCH] Add `Stream::sum()` and `Stream::product()` implementations These are the stream equivalents to `std::iter::Iterator::sum()` and `std::iter::Iterator::product()`. Note that this changeset tweaks the `Stream::Sum` and `Stream::Product` traits a little: rather than returning a generic future `F`, they return a pinned, boxed, `Future` trait object now. This is in line with other traits that return a future, e.g. `FromStream`. --- src/option/mod.rs | 5 +++ src/option/product.rs | 66 +++++++++++++++++++++++++++++ src/option/sum.rs | 63 ++++++++++++++++++++++++++++ src/result/mod.rs | 5 +++ src/result/product.rs | 64 ++++++++++++++++++++++++++++ src/result/sum.rs | 64 ++++++++++++++++++++++++++++ src/stream/product.rs | 64 ++++++++++++++++++++++++++-- src/stream/stream/mod.rs | 90 ++++++++++++++++++++++++++++++++++++++++ src/stream/sum.rs | 62 +++++++++++++++++++++++++-- 9 files changed, 476 insertions(+), 7 deletions(-) create mode 100644 src/option/product.rs create mode 100644 src/option/sum.rs create mode 100644 src/result/product.rs create mode 100644 src/result/sum.rs diff --git a/src/option/mod.rs b/src/option/mod.rs index afb29ad..76f096b 100644 --- a/src/option/mod.rs +++ b/src/option/mod.rs @@ -7,3 +7,8 @@ mod from_stream; #[doc(inline)] pub use std::option::Option; + +cfg_unstable! { + mod product; + mod sum; +} diff --git a/src/option/product.rs b/src/option/product.rs new file mode 100644 index 0000000..8b66bc6 --- /dev/null +++ b/src/option/product.rs @@ -0,0 +1,66 @@ +use std::pin::Pin; + +use crate::prelude::*; +use crate::stream::{Stream, Product}; + +impl Product> for Option +where + T: Product, +{ + #[doc = r#" + Takes each element in the `Stream`: if it is a `None`, no further + elements are taken, and the `None` is returned. Should no `None` occur, + the product of all elements is returned. + + # Examples + + This multiplies every integer in a vector, rejecting the product if a negative element is + encountered: + + ``` + # fn main() { async_std::task::block_on(async { + # + use std::collections::VecDeque; + use async_std::prelude::*; + + let v: VecDeque<_> = vec![1, 2, 4].into_iter().collect(); + let prod: Option = v.map(|x| + if x < 0 { + None + } else { + Some(x) + }).product().await; + assert_eq!(prod, Some(8)); + # + # }) } + ``` + "#] + fn product<'a, S>(stream: S) -> Pin> + 'a>> + where S: Stream> + 'a + { + Box::pin(async move { + pin_utils::pin_mut!(stream); + + // Using `scan` here because it is able to stop the stream early + // if a failure occurs + let mut found_none = false; + let out = >::product(stream + .scan((), |_, elem| { + match elem { + Some(elem) => Some(elem), + None => { + found_none = true; + // Stop processing the stream on error + None + } + } + })).await; + + if found_none { + None + } else { + Some(out) + } + }) + } +} diff --git a/src/option/sum.rs b/src/option/sum.rs new file mode 100644 index 0000000..25dc920 --- /dev/null +++ b/src/option/sum.rs @@ -0,0 +1,63 @@ +use std::pin::Pin; + +use crate::prelude::*; +use crate::stream::{Stream, Sum}; + +impl Sum> for Option +where + T: Sum, +{ + #[doc = r#" + Takes each element in the `Iterator`: if it is a `None`, no further + elements are taken, and the `None` is returned. Should no `None` occur, + the sum of all elements is returned. + + # Examples + + This sums up the position of the character 'a' in a vector of strings, + if a word did not have the character 'a' the operation returns `None`: + + ``` + # fn main() { async_std::task::block_on(async { + # + use std::collections::VecDeque; + use async_std::prelude::*; + + let words: VecDeque<_> = vec!["have", "a", "great", "day"] + .into_iter() + .collect(); + let total: Option = words.map(|w| w.find('a')).sum().await; + assert_eq!(total, Some(5)); + # + # }) } + ``` + "#] + fn sum<'a, S>(stream: S) -> Pin> + 'a>> + where S: Stream> + 'a + { + Box::pin(async move { + pin_utils::pin_mut!(stream); + + // Using `scan` here because it is able to stop the stream early + // if a failure occurs + let mut found_none = false; + let out = >::sum(stream + .scan((), |_, elem| { + match elem { + Some(elem) => Some(elem), + None => { + found_none = true; + // Stop processing the stream on error + None + } + } + })).await; + + if found_none { + None + } else { + Some(out) + } + }) + } +} diff --git a/src/result/mod.rs b/src/result/mod.rs index 908f9c4..cae0ebd 100644 --- a/src/result/mod.rs +++ b/src/result/mod.rs @@ -7,3 +7,8 @@ mod from_stream; #[doc(inline)] pub use std::result::Result; + +cfg_unstable! { + mod product; + mod sum; +} diff --git a/src/result/product.rs b/src/result/product.rs new file mode 100644 index 0000000..17afa94 --- /dev/null +++ b/src/result/product.rs @@ -0,0 +1,64 @@ +use std::pin::Pin; + +use crate::prelude::*; +use crate::stream::{Stream, Product}; + +impl Product> for Result +where + T: Product, +{ + #[doc = r#" + Takes each element in the `Stream`: if it is an `Err`, no further + elements are taken, and the `Err` is returned. Should no `Err` occur, + the product of all elements is returned. + + # Examples + + This multiplies every integer in a vector, rejecting the product if a negative element is + encountered: + + ``` + # fn main() { async_std::task::block_on(async { + # + use std::collections::VecDeque; + use async_std::prelude::*; + + let v: VecDeque<_> = vec![1, 2, 4].into_iter().collect(); + let res: Result = v.map(|x| + if x < 0 { + Err("Negative element found") + } else { + Ok(x) + }).product().await; + assert_eq!(res, Ok(8)); + # + # }) } + ``` + "#] + fn product<'a, S>(stream: S) -> Pin> + 'a>> + where S: Stream> + 'a + { + Box::pin(async move { + pin_utils::pin_mut!(stream); + + // Using `scan` here because it is able to stop the stream early + // if a failure occurs + let mut found_error = None; + let out = >::product(stream + .scan((), |_, elem| { + match elem { + Ok(elem) => Some(elem), + Err(err) => { + found_error = Some(err); + // Stop processing the stream on error + None + } + } + })).await; + match found_error { + Some(err) => Err(err), + None => Ok(out) + } + }) + } +} diff --git a/src/result/sum.rs b/src/result/sum.rs new file mode 100644 index 0000000..caca4f6 --- /dev/null +++ b/src/result/sum.rs @@ -0,0 +1,64 @@ +use std::pin::Pin; + +use crate::prelude::*; +use crate::stream::{Stream, Sum}; + +impl Sum> for Result +where + T: Sum, +{ + #[doc = r#" + Takes each element in the `Stream`: if it is an `Err`, no further + elements are taken, and the `Err` is returned. Should no `Err` occur, + the sum of all elements is returned. + + # Examples + + This sums up every integer in a vector, rejecting the sum if a negative + element is encountered: + + ``` + # fn main() { async_std::task::block_on(async { + # + use std::collections::VecDeque; + use async_std::prelude::*; + + let v: VecDeque<_> = vec![1, 2].into_iter().collect(); + let res: Result = v.map(|x| + if x < 0 { + Err("Negative element found") + } else { + Ok(x) + }).sum().await; + assert_eq!(res, Ok(3)); + # + # }) } + ``` + "#] + fn sum<'a, S>(stream: S) -> Pin> + 'a>> + where S: Stream> + 'a + { + Box::pin(async move { + pin_utils::pin_mut!(stream); + + // Using `scan` here because it is able to stop the stream early + // if a failure occurs + let mut found_error = None; + let out = >::sum(stream + .scan((), |_, elem| { + match elem { + Ok(elem) => Some(elem), + Err(err) => { + found_error = Some(err); + // Stop processing the stream on error + None + } + } + })).await; + match found_error { + Some(err) => Err(err), + None => Ok(out) + } + }) + } +} diff --git a/src/stream/product.rs b/src/stream/product.rs index 5799990..71b14c7 100644 --- a/src/stream/product.rs +++ b/src/stream/product.rs @@ -1,7 +1,9 @@ +use std::pin::Pin; + use crate::future::Future; use crate::stream::Stream; -/// Trait to represent types that can be created by productming up a stream. +/// Trait to represent types that can be created by multiplying the elements of 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 @@ -16,8 +18,62 @@ use crate::stream::Stream; pub trait Product: Sized { /// Method which takes a stream and generates `Self` from the elements by /// multiplying the items. - fn product(stream: S) -> F + fn product<'a, S>(stream: S) -> Pin + 'a>> where - S: Stream, - F: Future; + S: Stream + 'a; +} + +use core::ops::Mul; +use core::num::Wrapping; +use crate::stream::stream::StreamExt; + +macro_rules! integer_product { + (@impls $one: expr, $($a:ty)*) => ($( + impl Product for $a { + fn product<'a, S>(stream: S) -> Pin+ 'a>> + where + S: Stream + 'a, + { + Box::pin(async move { stream.fold($one, Mul::mul).await } ) + } + } + impl<'a> Product<&'a $a> for $a { + fn product<'b, S>(stream: S) -> Pin + 'b>> + where + S: Stream + 'b, + { + Box::pin(async move { stream.fold($one, Mul::mul).await } ) + } + } + )*); + ($($a:ty)*) => ( + integer_product!(@impls 1, $($a)*); + integer_product!(@impls Wrapping(1), $(Wrapping<$a>)*); + ); } + +macro_rules! float_product { + ($($a:ty)*) => ($( + impl Product for $a { + fn product<'a, S>(stream: S) -> Pin+ 'a>> + where S: Stream + 'a, + { + Box::pin(async move { stream.fold(1.0, |a, b| a * b).await } ) + } + } + impl<'a> Product<&'a $a> for $a { + fn product<'b, S>(stream: S) -> Pin+ 'b>> + where S: Stream + 'b, + { + Box::pin(async move { stream.fold(1.0, |a, b| a * b).await } ) + } + } + )*); + ($($a:ty)*) => ( + float_product!($($a)*); + float_product!($(Wrapping<$a>)*); + ); +} + +integer_product!{ i8 i16 i32 i64 i128 isize u8 u16 u32 u64 u128 usize } +float_product!{ f32 f64 } diff --git a/src/stream/stream/mod.rs b/src/stream/stream/mod.rs index 501ece1..f2d3c6e 100644 --- a/src/stream/stream/mod.rs +++ b/src/stream/stream/mod.rs @@ -96,6 +96,7 @@ cfg_unstable! { use crate::future::Future; use crate::stream::FromStream; + use crate::stream::{Product, Sum}; pub use merge::Merge; @@ -1536,6 +1537,95 @@ extension_trait! { { LtFuture::new(self, other) } + + #[doc = r#" + Sums the elements of an iterator. + + Takes each element, adds them together, and returns the result. + + An empty iterator returns the zero value of the type. + + # Panics + + When calling `sum()` and a primitive integer type is being returned, this + method will panic if the computation overflows and debug assertions are + enabled. + + # Examples + + Basic usage: + + ``` + # fn main() { async_std::task::block_on(async { + # + use std::collections::VecDeque; + use async_std::prelude::*; + + let s: VecDeque<_> = vec![0u8, 1, 2, 3, 4].into_iter().collect(); + let sum: u8 = s.sum().await; + + assert_eq!(sum, 10); + # + # }) } + ``` + "#] + #[cfg(feature = "unstable")] + #[cfg_attr(feature = "docs", doc(cfg(unstable)))] + fn sum<'a, S>( + self, + ) -> impl Future + 'a [Pin + 'a>>] + where + Self: Sized + Stream + 'a, + S: Sum, + { + Sum::sum(self) + } + + #[doc = r#" + Iterates over the entire iterator, multiplying all the elements + + An empty iterator returns the one value of the type. + + # Panics + + When calling `product()` and a primitive integer type is being returned, + method will panic if the computation overflows and debug assertions are + enabled. + + # Examples + + This example calculates the factorial of n (i.e. the product of the numbers from 1 to + n, inclusive): + + ``` + # fn main() { async_std::task::block_on(async { + # + async fn factorial(n: u32) -> u32 { + use std::collections::VecDeque; + use async_std::prelude::*; + + let s: VecDeque<_> = (1..=n).collect(); + s.product().await + } + + assert_eq!(factorial(0).await, 1); + assert_eq!(factorial(1).await, 1); + assert_eq!(factorial(5).await, 120); + # + # }) } + ``` + "#] + #[cfg(feature = "unstable")] + #[cfg_attr(feature = "docs", doc(cfg(unstable)))] + fn product<'a, P>( + self, + ) -> impl Future + 'a [Pin + 'a>>] + where + Self: Sized + Stream + 'a, + P: Product, + { + Product::product(self) + } } impl Stream for Box { diff --git a/src/stream/sum.rs b/src/stream/sum.rs index a87ade1..dadbc34 100644 --- a/src/stream/sum.rs +++ b/src/stream/sum.rs @@ -1,3 +1,5 @@ +use std::pin::Pin; + use crate::future::Future; use crate::stream::Stream; @@ -16,8 +18,62 @@ use crate::stream::Stream; pub trait Sum: Sized { /// Method which takes a stream and generates `Self` from the elements by /// "summing up" the items. - fn sum(stream: S) -> F + fn sum<'a, S>(stream: S) -> Pin + 'a>> where - S: Stream, - F: Future; + S: Stream + 'a; +} + +use core::ops::Add; +use core::num::Wrapping; +use crate::stream::stream::StreamExt; + +macro_rules! integer_sum { + (@impls $zero: expr, $($a:ty)*) => ($( + impl Sum for $a { + fn sum<'a, S>(stream: S) -> Pin+ 'a>> + where + S: Stream + 'a, + { + Box::pin(async move { stream.fold($zero, Add::add).await } ) + } + } + impl<'a> Sum<&'a $a> for $a { + fn sum<'b, S>(stream: S) -> Pin + 'b>> + where + S: Stream + 'b, + { + Box::pin(async move { stream.fold($zero, Add::add).await } ) + } + } + )*); + ($($a:ty)*) => ( + integer_sum!(@impls 0, $($a)*); + integer_sum!(@impls Wrapping(0), $(Wrapping<$a>)*); + ); } + +macro_rules! float_sum { + ($($a:ty)*) => ($( + impl Sum for $a { + fn sum<'a, S>(stream: S) -> Pin + 'a>> + where S: Stream + 'a, + { + Box::pin(async move { stream.fold(0.0, |a, b| a + b).await } ) + } + } + impl<'a> Sum<&'a $a> for $a { + fn sum<'b, S>(stream: S) -> Pin + 'b>> + where S: Stream + 'b, + { + Box::pin(async move { stream.fold(0.0, |a, b| a + b).await } ) + } + } + )*); + ($($a:ty)*) => ( + float_sum!(@impls 0.0, $($a)*); + float_sum!(@impls Wrapping(0.0), $(Wrapping<$a>)*); + ); +} + +integer_sum!{ i8 i16 i32 i64 i128 isize u8 u16 u32 u64 u128 usize } +float_sum!{ f32 f64 }