join_streams! macro

[rubdos]

I've seen a lot of rumbling about the join/select macro's around, thought I'd make an issue about them.

The thing is: in #[async] code, joining/selecting futures/streams is kind of impossible, and I wonder how this was going to get solved. Something like

#[async]
for item in join_streams!(a, b, c) {
    match { /* somehow */ }
}

would be cool.

[rubdos]

I just came up with this ugly thing. The nice thing about this is that usage of bindings in the blocks don't move out of the context where this macro is called. This way, multiple streams can actually use the same bindings, which is kinda neat.

I do wonder whether it's possible in my macro to get rid of the intermediate types though. Probably the compiler gets rid of most overhead; but still...

Also, I'm using the ugly

let s = the_stream;
loop {
    let (event, rest) = await!(s.into_future()).map_err(|e| e.0)?;
    s = rest;
    ...;
}

instead of #[async] for, because that doesn't compile. I have no clue why not; could someone look into that? I think the usage of #[async] in a macro just doesn't work, even while the callee is properly tagged. Might be worthy of a separate issue.

[alexcrichton]

Thanks for the report! This may be a case for stream combinators though rather than macros? I think things like zip/merge/select may suit your needs here?

[rubdos]

Thanks for the report! This may be a case for stream combinators though rather than macros? I think things like zip/merge/select may suit your needs here?

I'm rather asking for an #[async]/await! like way to handle multiple streams. I've seen some discussion on IRC about it, but never saw an issue in this repository.

I previously used a very very ugly combinator based approach, but that had its limitations. For example, two streams having mutable access to the same object can only be modelled by selecting two streams, then processing them. This gets ugly very fast when those streams have different types of output. Consider

enum EventType {
    EventA(TypeA),
    EventB(TypeB),
}

...

let common_state = State { .. };

let stream_a = stream_a.map(|a| EventType::EventA(a));
let stream_b = stream_b.map(|b| EventType::EventB(b));
#[async]
for item in stream_a.select(stream_b) {
    match item {
        // Here we can use the common state
    }
}

With the macro I wrote/want/propose/... this becomes

join_streams! {
    stream_a(a) => { /* use state */ }
    stream_b(b) => { /* use state */ }
};

---

Note that this is just my ugly macro (which actually does the former select under the hood): I'm just asking whether there's been discussion about an await/async-style way of doing join and select.

Feel free to ping me on IRC btw.

[alexcrichton]

Ah ok that makes sense! I'd be a little hesitant to add such a macro to this crate, but it seems like a macro that would be possible to build on async/await?

[rubdos]

Yes, I linked one in my second comment.

While I give this macro as an example, I think I really just wanted a discussion on how to make an async/await-friendly way of selecting/joining streams.

[rubdos]

Seems like this would be pretty easily build on the select! macro in the upcoming futures 0.3. Related recent reddit post: https://www.reddit.com/r/rust/comments/a442w8/asyncawait_higher_level_stream_combinators_i_am/