Let the ? operator work natively in try_stream!.

[goffrie]

Insteads of desugaring ? in the macro, we can have the async block
itself return Result<(), E>, and adjust the supporting code so that
? just works. The benefit is that this allows ? operators that are
hidden behind macros.

[taiki-e]

LGTM, thanks!

[albel727]

Is this supposed to break code that compiled before? Because it does for me. Before this PR one could use "return;" in the stream body to stop yielding. Now this fails with

expected type `()`
              found enum `std::result::Result<(), _>`

[albel727]

Ah yes, now that I read the PR with some coffee, of course it's supposed to break code. Well, simply returning Ok(()) is not a big deal, and the ability to return an Err is more flexible too. But I would really want a v0.3.1 release of async_stream, which lifts the Unpin on Stream::Item requirement, but with this PR included it would probably have to wait for a v0.4.

[goffrie]

Ah, that's a good point. I didn't realize it, but yes this is a breaking change.

I'll leave it to @taiki-e (or other maintainer 🙂) to decide whether to cut a 0.4 or revert this PR.

[taiki-e]

@albel727 good catch!

I tend to accept this at 0.4, but for now, I'd like to revert this. (and I'd like to add more tests before re-land in 0.4.)

[SergioBenitez]

What is the difference between try_stream! and stream!, besides the expected type of the expr in yield expr? If you take a try_stream! and replace every yield expr with yield Ok(expr), can you simply swap the try_stream! invocation for a stream! invocation?

If the answer is yes, then this leads me to ask: are both stream and try_stream necessary, and should ? just work in stream!? If the answer is no, the docs should be updated to reflect this. Consider that these two functions appears to work identically:

fn f<S>(stream: S) -> impl Stream<Item = io::Result<u8>>
    where S: Stream<Item = io::Result<&'static str>>
{
    try_stream! {
        for await s in stream {
            let num = s?.parse();
            let num = num.map_err(|e| io::Error::new(io::ErrorKind::Other, e))?;
            yield num;
        }
    }
}

fn g<S>(stream: S) -> impl Stream<Item = io::Result<u8>>
    where S: Stream<Item = io::Result<&'static str>>
{
    stream! {
        for await s in stream {
            let num = s?.parse();
            let num = num.map_err(|e| io::Error::new(io::ErrorKind::Other, e))?;
            yield Ok(num);
        }
    }
}

[Kestrer]

? cannot be used inside stream! because the async block must evaluate to (). We could modify stream! to support evaluating to Result<()> in which case the difference with try_stream! would be that it requires an Ok(()) at the end (ok-wrapping the yields can be done by AsyncStream). However, that wouldn't work well with the Try trait as we'd have overlapping impls of when the async block evaluates to () or T: Try.

If Try was implemented for () I think the two macros could be unified however. I haven't seen any plans for that though.

[SergioBenitez]

? cannot be used inside stream! because the async block must evaluate to ().

I don't understand. My example code above uses ? in stream! and works exactly as the try_stream! version does. Can you post an example where they would work differently?

[Kestrer]

Ah, right - it works because ? in stream! expands to a match where the Err branch yields and returns. So yes, I think you're right - other than native ? expansion and Ok-wrapping stream! and try_stream! are identical.

The native ? expansion does have an impact on the public API by the way. Consider this (contrived) example, which doesn't compile:

macro_rules! questionmark {
    ($e:expr) => { $e? };
}
let _ = stream! {
    questionmark!(Err(5));
    yield Ok::<i32, i32>(6);
};

Although this case is so rare there's no point in worrying about it much.