Future and Stream nesting with type problems

Question

I'd like to use a future which returns a Vec<String>, iterate over this in a future-stream and give the values to another future and the result of this future should be handled. The complete thing should be a future, too.

What's the way to go? I've tried different approaches and with all I've got type problems, which I don't understand.

Why there are these nested future result type signatures? Shouldn't this become the final result? Why doesn't the compiler know the types?

error[E0631]: type mismatch in closure arguments
  --> src/lib.rs:45:18
   |
45 |                 .then(|x: Result<(), ()>| ok(()))
   |                  ^^^^ -------------------------- found signature of `fn(std::result::Result<(), ()>) -> _`
   |                  |
   |                  expected signature of `fn(std::result::Result<std::vec::Vec<tokio::prelude::future::Then<tokio::prelude::future::Then<impl tokio::prelude::Future, tokio::prelude::future::FutureResult<(), ()>, [closure@src/lib.rs:35:31: 41:26]>, tokio::prelude::future::FutureResult<(), _>, [closure@src/lib.rs:42:31: 42:57]>>, _>) -> _`

I've setup a Playground for this

extern crate tokio;

use tokio::prelude::future::ok;
use tokio::prelude::*;

#[allow(dead_code)]
pub fn test_future<F>(f: F) -> Result<F::Item, F::Error>
where
    F: IntoFuture,
    F::Future: Send + 'static,
    F::Item: Send + 'static,
    F::Error: Send + 'static,
{
    let mut runtime = tokio::runtime::Runtime::new().expect("Unable to create a runtime");
    runtime.block_on(f.into_future())
}

#[allow(dead_code)]
fn fut(el: &String) -> impl Future<Item = String, Error = std::io::Error> {
    ok((el.to_string() + "-ok").to_string())
}

#[test]
fn reporting_future_result_test() {
    let v = vec![
        vec!["a".to_string(), "b".to_string()],
        vec!["a".to_string(), "b".to_string()],
    ];

    let f = stream::iter_ok(v.iter().cloned())
        .map(|el: Vec<String>| {
            stream::iter_ok(el.iter().cloned())
                .map(|ell: String| {
                    fut(&ell)
                        .then(|x: Result<String, std::io::Error>| {
                            match x {
                                Ok(s) => println!("{}", s),
                                Err(e) => println!("{:?}", e),
                            };
                            ok(())
                        })
                        .then(|x: Result<(), ()>| ok(()))
                })
                .collect()
                .then(|x: Result<(), ()>| ok(()))
        })
        .collect()
        .then(|x: Result<Vec<_>, std::io::Error>| ok(()));

    let r = test_future(f);

    match r {
        Ok(x) => println!("{:?}", x),
        Err(_) => println!("error"),
    }
}

[Why is it discouraged to accept a reference to a String (&String), Vec (&Vec) or Box (&Box) as a function argument?](https://stackoverflow.com/q/40006219/155423) — Shepmaster, Nov 02 '18 at 00:46
*expected signature of `fn(Result, ...>`* — why does your closure have a `Result<(), ...>`? What do you think the result of calling `collect` will be? — Shepmaster, Nov 02 '18 at 00:48
*a future which returns a `Vec`* You don't have that. You have a *stream* that returns `Vec`. — Shepmaster, Nov 02 '18 at 00:50
@Shepmaster Yes, in the sample I've a stream. Does this matter so much? I don't know how to get a future from a Vec in the same short way. About the result I'm not quite sure: I think it will be a Vec out of (). At the end it doesn't matter for me. I'd like to throw away all results. I've thought I use the collect to get a single value in the "result tree" to come closer to the end result Future. — Markus, Nov 02 '18 at 10:32
Ah. I've forgot to say that the collect is more or less an embarrassment, because of the Type errors. In my understanding a final then() at the outer end could be enough. But then there is no then() found for the traits... or something like that. I don't understand these error messages of the futures. — Markus, Nov 02 '18 at 10:47
Maybe I see this too much like JS Promises to understand what's going on. But I lack some documentation/information to get lucidity. — Markus, Nov 02 '18 at 11:12

score 0 · Accepted Answer · answered Nov 02 '18 at 13:14

0

extern crate tokio; // 0.1.11

use tokio::prelude::*;

// a future which returns a Vec<String>
fn makes_strings() -> impl Future<Item = Vec<String>, Error = ()> {
    future::ok(vec![])
}

fn make_new_string(el: String) -> impl Future<Item = String, Error = ()> {
    future::ok(el + "-ok")
}

fn iterate_over() -> impl Future<Item = Vec<String>, Error = ()> {
    makes_strings().and_then(|v| {
        // iterate over this
        let strings = v.into_iter();
        // give the values to another future
        let futures = strings.map(make_new_string);
        // The complete thing should be a future
        future::join_all(futures)
    })
}

answered Nov 02 '18 at 13:14

Shepmaster

388,571
95
1,107
1,366

Ok. I'll try this way. – Markus Nov 02 '18 at 13:35
Do you know why the inner futures only get polled if I use the block_on()? On run() or spawn() I have to either wait the inner ones or use another spawn(). That's the cause why I got into trouble with it. I thought I have something wrong with building my future chain... but maybe this probably works like this. At place of future::ok in your code, I've another_future().then(|_| {println!("works");ok(())}) The "works" I only see when using block_on(iterate_over), wait() or tokio::spawn. I return all the inner futures back to the caller, shouldn't there be a "relation/knowledge"?! – Markus Nov 03 '18 at 10:35
Here is one which starts with vec->stream: https://play.rust-lang.org/?version=stable&mode=debug&edition=2015&gist=2fc0a1532bc7b80520be2a31d593b461 – Markus Nov 03 '18 at 13:33

Future and Stream nesting with type problems

1 Answers1