Problem using Stream in Tokio Task

help
1

With following code, I get error that "implementation of std::ops::FnOnce is not general enough" and "no two async blocks, even if identical, have the same type. consider pinning your async block and casting it to a trait object". The error is on spawn line, and detail line number points to the whole closure of inner_stream call in then. I tried to pin the closure or the block, nothing seems to work. Maybe I did not do it right. How can I fix this?

PS: This code works in Playground, but not locally with Rust Stable. I have no idea why.

use futures::StreamExt;
use futures::TryStreamExt;
use futures::stream;
use futures::stream::BoxStream;
use tokio::task::JoinSet;

#[derive(Debug)]
struct Error;

async fn inner_stream(_i: usize) -> Result<BoxStream<'static, Result<usize, Error>>, Error> {
    Err(Error)
}

async fn outer_stream(_x: String) -> Result<(), Error> {
    let iter = [].into_iter().filter(|x: &Vec<usize>| !x.is_empty());

    stream::iter(iter)
        .then(async |x| {
            // some processing
            Ok(x)
        })
        .map_ok(async |x| {
            stream::iter(x.into_iter())
//                .then(|y| Box::pin(async move { inner_stream(y).await }))
                .then(async |y| inner_stream(y).await ))
                .try_flatten()
                .try_collect::<Vec<_>>()
                .await
        })
        .try_for_each_concurrent(8, async |x| {
            let _ = x.await;
            Ok(())
        })
        .await?;
    Ok(())
}

#[tokio::main]
async fn main() -> Result<(), Error> {
    let xs: Vec<String> = vec![];
    let mut js = JoinSet::new();
    for x in xs {
        js.spawn(async move { outer_stream(x).await });
    }
    js.join_all().await;
    Ok(())
}
2

In general when asking these sorts of questions, I suggest posting code that reproduces the problem in the playground, or at least the full error from running cargo run in a terminal.

Here's what I ended up with.

error[E0308]: mismatched types
  --> src/lib.rs:39:43
   |
33 |           .then(async |y| {
   |  _________________________-
34 | |           inner_stream(y, todo!(), todo!()).await
35 | |         })
   | |_________- the found `async` closure body
...
39 |       .try_for_each_concurrent(8, async |x| {
   |  ___________________________________________^
40 | |       x.await// ; Ok(())
41 | |       // some processing
42 | |     })
   | |_____^ expected `Result<(), _>`, found `TryCollect<TryFlatten<_>, _>`
   |
   = note: expected `async` closure body `{async closure body@src/lib.rs:39:43: 42:6}` (`Result<(), _>`)
              found `async` closure body `{async closure body@src/lib.rs:39:43: 42:6}` (`TryCollect<TryFlatten<_>, _>`)
   = note: no two async blocks, even if identical, have the same type
   = help: consider pinning your async block and casting it to a trait object

And I think the error is actively misleading, and that this is the pertinent part of the error:

   | |_____^ expected `Result<(), _>`, found `TryCollect<TryFlatten<_>, _>`

Because try_for_each_concurrent wants a Result<(), _>.

However, it's possible your error is something else if you've omitted too much ("some processing").

If that is your error, try this:

    .try_for_each_concurrent(8, async |x| {
-     x.await
+     x.await;
      // some processing
+     Ok(())
    })

(I don't know if it's what you want semantically, but if it gets rid of the error, it's a starting point.)

3

Filed a diagnostic issue.

4

Sorry, the second error in try_for_each_concurrent is my bad. The first one is the one I got, "no two async blocks are the same".

I could not produce with a minimal code in playground, it generally passed. I updated the original code which only gives the error I got, with Rust Stable.

5

Now I get the sample code working, by revert the BoxStream to its original impl TryStream. But this does not help in actual code. Giving I still cannot access Github, will update later.

6

I got it working. After revert BoxStream, I give + Send + 'static to the impl TryStream, just a guess, and rustc prompts one of the function parameters does not have proper lifetime. Set it to 'static makes it work.