Add async_stream to create streams via generators

[taiki-e]

This adds a proc-macro crate for experimenting with the async stream API that we may introduce as a feature of language in the future.

The implementation is basically similar to futures-await's #[async] for and #[async_stream]. (re-added some of the files deleted by #1050 and updated it.)
EDIT: After introducing .await syntax, it was not possible to keep the same implementation in stream scope (#[async_stream]) and future scope (async fn), so it was necessary to adjust it.

cc alexcrichton/futures-await#111

Rendered

[taiki-e]

This is the same as async fn.

[Nemo157]

Damn, that's my bad, I should have removed the explicit !Unpin implementation as part of rust-lang/rust#55704 (or transformed it into an explicit conditional implementation).

[Nemo157]

You shouldn't need this, T: ?Unpin so Self: ?Unpin via the structural rules (although because of PhantomData<U> you might be getting the Unpinness of U as well), if the generator transform ever allows non-self-referential static-marked generators to implement Unpin automatically then this could potentially generate impl Stream + Unpin.

[Nemo157]

One potential future addition to this would be #[async_sink], it seems much more niche than #[async_stream] though.

[Ekleog]

FWIW, here is something that would likely benefit from #[async_sink], which I have rage-quit trying to write as a manual Sink wrapper for now: https://github.com/Ekleog/yuubind/blob/master/smtp-message/src/data/sink.rs#L18 (it's almost with_flat_map except it's also adding some data after the end of the incoming stream)

[Nemo157]

(somewhat OT: @Ekleog it took a few upgrades to my async_sink_block implementation to workaround type inference issues, and I'm not sure how to deal with errors yet, but I got that example compiling with it: https://gist.github.com/Nemo157/0344ba1aa1d1d89ce2f5ae26ea7b1d46, so maybe it is worth considering adding something like it here in the future)

[yoshuawuyts]

I would love to see this included; this would allow us to make our tcp-echo and tcp-proxy examples in Runtime so much nicer!

[taiki-e]

One of the concerns with current implementations is that there is an Item type at the "return" position. In the futures-await's implementation, this is specified in the form of an attribute argument #[async_stream(item = i32)].

[yoshuawuyts]

@taiki-e that approach seems reasonable to me; are there any reasons why we can't / shouldn't follow future-await's approach?

[taiki-e]

@yoshuawuyts There is no particular reason. Also, it is easy to change the implementation to that.

(Because the stream has no final return value, the return position is always empty. The current implementation is just trying to reuse this blank.)

[kgv]

In futures-await we can use ? operator inside stream. If I understand correctly, with new syntax it's impossible. Is it true?

/// Fetch all provided urls one at a time
#[async_stream(item = String)]
fn fetch_all(client: hyper::Client, urls: Vec<&'static str>) -> io::Result<()> {
    for url in urls {
        let s = await!(fetch(client, url))?;
        stream_yield!(s);
    }
    Ok(())
}

[Nemo157]

@kgv yes, because of streams no longer being required to have an error and item type that no longer works.

You can probably do something mostly equivalent with a try block though:

/// Fetch all provided urls one at a time
#[async_stream(item = io::Result<String>)]
fn fetch_all(client: hyper::Client, urls: Vec<&'static str>) {
    let res = try {
        for url in urls {
            yield Ok(fetch(client, url).await?);
        }
    };
    if let Err(err) = res {
         yield Err(err);
    }
}

and it may be possible to add more sugar for this in the future (an async_try_stream maybe?).

[taiki-e]

@Nemo157

and it may be possible to add more sugar for this in the future (an async_try_stream maybe?).

I will try it after this PR is merged. And I think it will be something like the following:

#[async_try_stream(ok = String, error = io::Error)]
fn fetch_all(client: hyper::Client, urls: Vec<&'static str>) {
    for url in urls {
         yield fetch(client, url).await?;
    }
}

// desugaring into

fn fetch_all(client: hyper::Client, urls: Vec<&'static str>) -> impl Stream<Item = Result<String, io::Error>> {
    from_generator(static move || { // impl Generator<Yield = String, Return = Result<(), io::Error>>
        for url in urls {
             yield Poll::Ready(fetch(client, url).await?);
        }
        return Ok(()); // It is the same behavior as try_blocks that this is added automatically.
        loop { yield Poll::Pending }
    })
}

[taiki-e]

(Also, if we add async_try_stream, it may be preferable to write Item at the position of the attribute argument, not at the position of return, so I will revert it)

[JuanPotato]

I really appreciate this feature and would be happy to help with anything. I haven't been working with futures 0.3 for too long but am very willing to help.

[taiki-e]

@cramertj Any thoughts about this pr?

[taiki-e]

Published as futures-async-stream -- Thanks everybody for your patience. It may not be what was expected, but I believe it would be better to polish it as a separate project and then decide if it should eventually merge.