How to combine two or more features, and process them back in the order that they are resolved?

Hi, newbie here, I have been looking the documentation, I can't figure out how to make this work.

let say, I have two futures, f1 and f2, I want to combine them in a new future or stream and process them in the order in which they were resolved, without the need to wait for all to be complete.

for example, f2.join(f1), will not works because this wait till both futures are done.

I am looking for something more like this

combine([f1, f2]).map(|result|{
do_computation;
return future
}).map(|another_result|{
another computation;
return result
}).collect()

any help will be welcome.

If you want to process the results independently, why not call map on the individual futures and only join at the end?

@HadrienG thanks, your suggestion seems pretty logical, but now the program does not show anything

https://gist.github.com/criloz/7f1b5b7cdef086e05eba999a13f94dfc

for reference, this is the working code

https://gist.github.com/criloz/2aae69f5e84c737155ea3110219fa928

the working code is pretty ugly, but works,, any idea?

@vitalyd

does not work either.

https://gist.github.com/criloz/94489592261033c5269a1c10aa49ca8e

I'm not entirely sure what your end goal is, but you should use for_each to work through the futures:

::futures::stream::futures_unordered(vs).for_each(|res|{
                res.body().concat2().and_then(move |body|{
                    let value:Story = serde_json::from_slice(&body).unwrap();
                    println!("{:?}", value.title);
                    Ok(())
                })
            })

And you don't need the collect() there (again, not sure what your real usecase is though).

1 Like

@vitalyd, that worked just perfect, the real use case is about made multiples queries to different servers, and parse the response to json as they come. I also wanted to understand more about futures and how can I compose pipelines with them

Thanks

1 Like

Gotcha - yeah, good exercise.

In case you’re wondering, the issue with your initial version using futures_unordered is the map() - this function just translates/maps the result of each item in the stream but crucially doesn’t return something that’s a future itself. Or rather, you can return something that’s a future but the framework doesn’t "hook it up" - nothing drives it to completion. So the res.body().concat2()... future was created but immediately dropped, and no output was generated.