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.

Dinnage

you link two questions that fix your issue – Starstudded

@Starstudded in fact, they don't since I cannot use into_iter() without implementing it. – Dinnage

Have you discovered the source of the error (trait Try not implemented for String)? With a quick glance, it looks like you need to stop using ? inside the map function and see if collect can do this for you (the linked docs has an example which is similar to what you want). You can't just copy-paste other solutions; you'll need to understand the types you are working with and develop a solution that fits. – Sidonia

@swalladge that's exactly what I've tried, but I still end up with Vec<Result<>> I think is collect() does this for Iterator but can't do it for Stream and I can't do into_iter(). – Dinnage

I.e., if I remove ? I receive ``` = note: expected type std::vec::Vec<i64> found type `std::vec::Vec<std::result::Result<i64, errors::CustomError>>``` – Dinnage

@Starstudded please take another look, I've updated the question with a piece of code on a playground. – Dinnage

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).

Sidonia

Answer 2

map with a function that returns Result

Don't do this, that's not when you should use map. Instead, use and_then:

let vec = into_many(10)
    .and_then(|number| convert_to_string(number))
    .collect()
    .wait()
    .unwrap();

You should practice with simpler Rust concepts like Option, Result, and iterators before diving into futures. Many concepts transfer over.

See also:

• How do I unwrap an arbitrary number of nested Option types?
• What is the idiomatic way to handle/unwrap nested Result types?

Georgie

Answer 3

https://docs.rs/futures/latest/futures/stream/trait.TryStreamExt.html#method.try_collect

Use try_collect() over .collect::<Vec<Result<_, _>>().into_iter().collect::<Result<Vec<_>, _>() as try_collect will terminate early on error`.

Lovable