How can I use Stream::map with a function that returns Result?

Question

I've got the following piece of code (see playground):

use futures::{stream, Future, Stream}; // 0.1.25
use std::num::ParseIntError;

fn into_many(i: i32) -> impl Stream<Item = i32, Error = ParseIntError> {
    stream::iter_ok(0..i)
}

fn convert_to_string(number: i32) -> Result<String, ParseIntError> {
    Ok(number.to_string())
}

fn main() {
    println!("start:");
    let vec = into_many(10)
        .map(|number| convert_to_string(number))
        .collect()
        .wait()
        .unwrap();

    println!("vec={:#?}", vec);

    println!("finish:");
}

It outputs the following (i.e., Vec<Result<i32, ParseIntError>>):

start:
vec=[
    Ok(
        "0"
    ),
    Ok(
        "1"
    ),
    Ok(
        "2"
    ), ...

Is there any way to make it output a Vec<i32> and if any error happens than immediately stop execution and return from the function (e.g., like this example)?

Note: I do want to use use futures::Stream; // 0.1.25 even if it doesn't make sense for this particular example.

Answer 1

The following code (playground link) as a modification of your current code in your question gets the result you want:

use futures::{stream, Future, Stream}; // 0.1.25
use std::num::ParseIntError;

fn into_many(i: i32) -> impl Stream<Item = i32, Error = ParseIntError> {
    stream::iter_ok(0..i)
}

fn convert_to_string(number: i32) -> Result<String, ParseIntError> {
    Ok(number.to_string())
}

fn main() {
    println!("start:");
    let vec: Result<Vec<String>, ParseIntError> = into_many(10)
        .map(|number| convert_to_string(number))
        .collect()
        .wait()
        .unwrap()
        .into_iter()
        .collect();

    println!("vec={:#?}", vec);

    println!("finish:");
}

Since your current code returned a Vec, we can turn that into an iterator and collect that into the type you want. Type annotations are needed so that collect knows what type to collect the iterator into.

Note that the collect method on the Iterator trait isn't to be confused with the collect method on a Stream.

Finally, while this works, it may not be exactly what you want, since it still waits for all results from the stream to be collected into a vector, before using collect to transform the vector. I don't have experience with futures so not sure how possible this is (it probably is but may require a less neat functional programming style solution).