Skip to content

Commit

Permalink
Merge #207
Browse files Browse the repository at this point in the history
207: Added the ability to collect a stream of results r=yoshuawuyts a=sunjay

As requested here: https://twitter.com/yoshuawuyts/status/1174026374316773377

The standard library has a very useful implementation of `FromIterator` that takes an iterator of `Result<T, E>` values and is able to produce a value of type `Result<Vec<T>, E>`. I asked for this in `async-std` and @yoshuawuyts recommended that I contribute the impl. It turns out that the implementation in the standard library is even more general than I initially thought. It allows any collection that implements `FromIterator` to be collected from an iterator of `Result<T, E>` values. That means that you can collect into `Result<Vec<T>, E>`, `Result<HashSet<T>, E>`, etc.

I wanted to add a similarly generic impl for this crate so we can also support collecting into any collection that implements `FromStream`. 

The implementation for this is based heavily on [what exists in `std`](https://github.com/rust-lang/rust/blob/9150f844e2624eb013ec78ca08c1d416e6644026/src/libcore/result.rs#L1379-L1429). I made a new `result` module since that's where this impl is in `std`. I still wanted to maintain the conventions of this repo, so I copied the `vec` module that @yoshuawuyts created in #125. Much like in that PR, the new `result` module is private.

There is a doctest in the documentation for `collect` that both teaches that this feature exists and tests that it works in some simple cases.

## Documentation Screenshot

![image](https://user-images.githubusercontent.com/530939/65075935-de89ae00-d965-11e9-9cd6-8b19b694ed3e.png)


Co-authored-by: Sunjay Varma <[email protected]>
  • Loading branch information
bors[bot] and sunjay authored Sep 17, 2019
2 parents 4f9e7d3 + c87dab2 commit c8475ca
Show file tree
Hide file tree
Showing 4 changed files with 73 additions and 0 deletions.
1 change: 1 addition & 0 deletions src/lib.rs
Original file line number Diff line number Diff line change
Expand Up @@ -48,6 +48,7 @@ pub mod io;
pub mod net;
pub mod os;
pub mod prelude;
mod result;
pub mod stream;
pub mod sync;
pub mod task;
Expand Down
47 changes: 47 additions & 0 deletions src/result/from_stream.rs
Original file line number Diff line number Diff line change
@@ -0,0 +1,47 @@
use crate::stream::{FromStream, IntoStream, Stream};

use std::pin::Pin;

impl<T: Send, E: Send, V> FromStream<Result<T, E>> for Result<V, E>
where
V: FromStream<T>,
{
/// 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, a container with the values of each `Result` is returned.
#[inline]
fn from_stream<'a, S: IntoStream<Item = Result<T, E>>>(
stream: S,
) -> Pin<Box<dyn core::future::Future<Output = Self> + Send + 'a>>
where
<S as IntoStream>::IntoStream: Send + 'a,
{
let stream = stream.into_stream();

Pin::from(Box::new(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: V = stream
.scan((), |_, elem| {
match elem {
Ok(elem) => Some(elem),
Err(err) => {
found_error = Some(err);
// Stop processing the stream on error
None
}
}
})
.collect()
.await;

match found_error {
Some(err) => Err(err),
None => Ok(out),
}
}))
}
}
9 changes: 9 additions & 0 deletions src/result/mod.rs
Original file line number Diff line number Diff line change
@@ -0,0 +1,9 @@
//! The Rust core error handling type
//!
//! This module provides the `Result<T, E>` type for returning and
//! propagating errors.
mod from_stream;

#[doc(inline)]
pub use std::result::Result;
16 changes: 16 additions & 0 deletions src/stream/stream/mod.rs
Original file line number Diff line number Diff line change
Expand Up @@ -726,6 +726,22 @@ pub trait Stream {
/// let buf: Vec<u8> = s.collect().await;
///
/// assert_eq!(buf, vec![9; 3]);
///
/// // You can also collect streams of Result values
/// // into any collection that implements FromStream
/// let s = stream::repeat(Ok(9)).take(3);
/// // We are using Vec here, but other collections
/// // are supported as well
/// let buf: Result<Vec<u8>, ()> = s.collect().await;
///
/// assert_eq!(buf, Ok(vec![9; 3]));
///
/// // The stream will stop on the first Err and
/// // return that instead
/// let s = stream::repeat(Err(5)).take(3);
/// let buf: Result<Vec<u8>, u8> = s.collect().await;
///
/// assert_eq!(buf, Err(5));
/// #
/// # }) }
/// ```
Expand Down

0 comments on commit c8475ca

Please sign in to comment.