Skip to content

Commit

Permalink
Merge pull request #667 from olegnn/option_take_while
Browse files Browse the repository at this point in the history 
Use `take_while` instead of `scan` in `impl` of `Product`, `Sum` and `FromStream` for `Option` and `Result`
  • Loading branch information
@yoshuawuyts
yoshuawuyts authored Jan 27, 2020
2 parents beb8d24 + ed7ddac commit 6c1b5eb
Show file tree
Hide file tree
Showing 7 changed files with 113 additions and 82 deletions.
21 changes: 10 additions & 11 deletions src/option/from_stream.rs
View file
Original file line number Diff line number Diff line change
Expand Up @@ -2,6 +2,7 @@ use std::pin::Pin;

use crate::prelude::*;
use crate::stream::{FromStream, IntoStream};
use std::convert::identity;

impl<T, V> FromStream<Option<T>> for Option<V>
where
Expand All @@ -17,24 +18,22 @@ where
let stream = stream.into_stream();

Box::pin(async move {
// Using `scan` here because it is able to stop the stream early
// Using `take_while` here because it is able to stop the stream early
// if a failure occurs
let mut found_error = false;
let mut found_none = false;
let out: V = stream
.scan((), |_, elem| {
match elem {
Some(elem) => Some(elem),
None => {
found_error = true;
// Stop processing the stream on error
None
}
.take_while(|elem| {
elem.is_some() || {
found_none = true;
// Stop processing the stream on `None`
false
}
})
.filter_map(identity)
.collect()
.await;

if found_error { None } else { Some(out) }
if found_none { None } else { Some(out) }
})
}
}
33 changes: 16 additions & 17 deletions src/option/product.rs
View file
Original file line number Diff line number Diff line change
@@ -1,7 +1,8 @@
use std::pin::Pin;

use crate::prelude::*;
use crate::stream::{Stream, Product};
use crate::stream::{Product, Stream};
use std::convert::identity;

impl<T, U> Product<Option<U>> for Option<T>
where
Expand Down Expand Up @@ -36,29 +37,27 @@ where
```
"#]
fn product<'a, S>(stream: S) -> Pin<Box<dyn Future<Output = Option<T>> + 'a>>
where S: Stream<Item = Option<U>> + 'a
where
S: Stream<Item = Option<U>> + 'a,
{
Box::pin(async move {
// Using `scan` here because it is able to stop the stream early
// Using `take_while` here because it is able to stop the stream early
// if a failure occurs
let mut found_none = false;
let out = <T as Product<U>>::product(stream
.scan((), |_, elem| {
match elem {
Some(elem) => Some(elem),
None => {
let out = <T as Product<U>>::product(
stream
.take_while(|elem| {
elem.is_some() || {
found_none = true;
// Stop processing the stream on error
None
// Stop processing the stream on `None`
false
}
}
})).await;
})
.filter_map(identity),
)
.await;

if found_none {
None
} else {
Some(out)
}
if found_none { None } else { Some(out) }
})
}
}
31 changes: 15 additions & 16 deletions src/option/sum.rs
View file
Original file line number Diff line number Diff line change
Expand Up @@ -2,6 +2,7 @@ use std::pin::Pin;

use crate::prelude::*;
use crate::stream::{Stream, Sum};
use std::convert::identity;

impl<T, U> Sum<Option<U>> for Option<T>
where
Expand Down Expand Up @@ -31,29 +32,27 @@ where
```
"#]
fn sum<'a, S>(stream: S) -> Pin<Box<dyn Future<Output = Option<T>> + 'a>>
where S: Stream<Item = Option<U>> + 'a
where
S: Stream<Item = Option<U>> + 'a,
{
Box::pin(async move {
// Using `scan` here because it is able to stop the stream early
// Using `take_while` here because it is able to stop the stream early
// if a failure occurs
let mut found_none = false;
let out = <T as Sum<U>>::sum(stream
.scan((), |_, elem| {
match elem {
Some(elem) => Some(elem),
None => {
let out = <T as Sum<U>>::sum(
stream
.take_while(|elem| {
elem.is_some() || {
found_none = true;
// Stop processing the stream on error
None
// Stop processing the stream on `None`
false
}
}
})).await;
})
.filter_map(identity),
)
.await;

if found_none {
None
} else {
Some(out)
}
if found_none { None } else { Some(out) }
})
}
}
32 changes: 20 additions & 12 deletions src/result/from_stream.rs
View file
Original file line number Diff line number Diff line change
Expand Up @@ -34,26 +34,34 @@ where
let stream = stream.into_stream();

Box::pin(async move {
// Using `scan` here because it is able to stop the stream early
// Using `take_while` here because it is able to stop the stream early
// if a failure occurs
let mut is_error = false;
let mut found_error = None;
let out: V = stream
.scan((), |_, elem| {
match elem {
Ok(elem) => Some(elem),
Err(err) => {
found_error = Some(err);
// Stop processing the stream on error
None
}
.take_while(|elem| {
// Stop processing the stream on `Err`
!is_error
&& (elem.is_ok() || {
is_error = true;
// Capture first `Err`
true
})
})
.filter_map(|elem| match elem {
Ok(value) => Some(value),
Err(err) => {
found_error = Some(err);
None
}
})
.collect()
.await;

match found_error {
Some(err) => Err(err),
None => Ok(out),
if is_error {
Err(found_error.unwrap())
} else {
Ok(out)
}
})
}
Expand Down
39 changes: 26 additions & 13 deletions src/result/product.rs
View file
Original file line number Diff line number Diff line change
@@ -1,7 +1,7 @@
use std::pin::Pin;

use crate::prelude::*;
use crate::stream::{Stream, Product};
use crate::stream::{Product, Stream};

impl<T, U, E> Product<Result<U, E>> for Result<T, E>
where
Expand Down Expand Up @@ -36,26 +36,39 @@ where
```
"#]
fn product<'a, S>(stream: S) -> Pin<Box<dyn Future<Output = Result<T, E>> + 'a>>
where S: Stream<Item = Result<U, E>> + 'a
where
S: Stream<Item = Result<U, E>> + 'a,
{
Box::pin(async move {
// Using `scan` here because it is able to stop the stream early
// Using `take_while` here because it is able to stop the stream early
// if a failure occurs
let mut is_error = false;
let mut found_error = None;
let out = <T as Product<U>>::product(stream
.scan((), |_, elem| {
match elem {
Ok(elem) => Some(elem),
let out = <T as Product<U>>::product(
stream
.take_while(|elem| {
// Stop processing the stream on `Err`
!is_error
&& (elem.is_ok() || {
is_error = true;
// Capture first `Err`
true
})
})
.filter_map(|elem| match elem {
Ok(value) => Some(value),
Err(err) => {
found_error = Some(err);
// Stop processing the stream on error
None
}
}
})).await;
match found_error {
Some(err) => Err(err),
None => Ok(out)
}),
)
.await;

if is_error {
Err(found_error.unwrap())
} else {
Ok(out)
}
})
}
Expand Down
37 changes: 25 additions & 12 deletions src/result/sum.rs
View file
Original file line number Diff line number Diff line change
Expand Up @@ -36,26 +36,39 @@ where
```
"#]
fn sum<'a, S>(stream: S) -> Pin<Box<dyn Future<Output = Result<T, E>> + 'a>>
where S: Stream<Item = Result<U, E>> + 'a
where
S: Stream<Item = Result<U, E>> + 'a,
{
Box::pin(async move {
// Using `scan` here because it is able to stop the stream early
// Using `take_while` here because it is able to stop the stream early
// if a failure occurs
let mut is_error = false;
let mut found_error = None;
let out = <T as Sum<U>>::sum(stream
.scan((), |_, elem| {
match elem {
Ok(elem) => Some(elem),
let out = <T as Sum<U>>::sum(
stream
.take_while(|elem| {
// Stop processing the stream on `Err`
!is_error
&& (elem.is_ok() || {
is_error = true;
// Capture first `Err`
true
})
})
.filter_map(|elem| match elem {
Ok(value) => Some(value),
Err(err) => {
found_error = Some(err);
// Stop processing the stream on error
None
}
}
})).await;
match found_error {
Some(err) => Err(err),
None => Ok(out)
}),
)
.await;

if is_error {
Err(found_error.unwrap())
} else {
Ok(out)
}
})
}
Expand Down
2 changes: 1 addition & 1 deletion src/unit/from_stream.rs
View file
Original file line number Diff line number Diff line change
Expand Up @@ -8,6 +8,6 @@ impl FromStream<()> for () {
fn from_stream<'a, S: IntoStream<Item = ()> + 'a>(
stream: S,
) -> Pin<Box<dyn Future<Output = Self> + 'a>> {
Box::pin(stream.into_stream().for_each(|_| ()))
Box::pin(stream.into_stream().for_each(drop))
}
}

0 comments on commit 6c1b5eb

Please sign in to comment.