Join stream

[yoshuawuyts]

Depends on #1 to land first. Adds a new stream_join macro that can join multiple streams into a single stream:

let a = stream::once(1u8);
let b = stream::once(2u8);
let c = stream::once(3u8);

let mut s = join!(a, b, c);

assert_eq!(s.next().await, Some(1u8));
assert_eq!(s.next().await, Some(2u8));
assert_eq!(s.next().await, Some(3u8));
assert_eq!(s.next().await, None);

Thanks!

[matklad]

Does this need to be a macro? It seems that the most common use-case is joining two streams, and a.join(b) seems better than join!(a, b). a.join(b).join(c) is not too bad either, to my eye :) It also seems pretty similar to how Iterator::chain works.

[yoshuawuyts]

@matklad oh yeah, it doesn't need to be a macro for sure. I figured it'd be slightly more symmetrical with the other join macros:

async_std::future::join;
async_std::future::select;
async_std::future::try_join;
async_std::future::try_select;
async_std::stream::join;

I agree we may want to consider adding a join method to stream too, but for futures it's more tricky (because of the Result bound, which afaik requires TryFutureExt). So this represents a nice subset of things that's useful enough, and internally consistent inside async_std.

---

Perhaps a question then becomes: what about try_join and select for stream? I don't think we need a try variant because we don't need to short-circuit streams the way we do for join. And instead of select we can use match to pick a specific item from the stream. I think join is the only one that really makes sense to implement for Stream here, haha.

[matklad]

Ah, I see! (note: the following is not an argument for a particular position, just musings aloud). It seems like futures::join and stream::join do slightly different things though and probably could be named differently. Specifically, futures::join has to be a macro, because it joins futures of different types, and returns an (item1, item2, ...) which is awkward to express in types. stream::join, OTOH, requires that items of the input streams are the same, so it can be expressed without a macro. A more direct equivalent of futures::join seems to be a stream::zip, which yields tuples of values. This also seems they underlying reason why future::try_join makes sense, and stream::try_join doesn't: they are just different operations. A possible different (but not necessary better) name would be merge: a.merge(b).merge(c).

[yoshuawuyts]

@matklad These are all great points! Actually I've considered the name merge, not in the least because it has precedence in rxjs::merge.

But I still think join is a better name for this because where future::join converts multiple futures into a single future of their outputs, stream::join converts multiple streams into a single stream of their outputs. There's a nice symmetry in the use there, and I think between zip, chain and join we already have enough terminology that adding more might be confusing.

You make a good point though about this not having to be a macro in the same way that the futures variants have tho. I guess I'm still in favor of what I'm proposing here, but it's not really a strong opinion -- more of an intuitive preference that I'm curious how people will like it.

[matklad]

Some other terminology observations:

• futures-preview calls this operation join merge
• join can also be confusing, because it is also used in "join a thread/task". Ie, not joining two parallel strands of execution together, but waiting for a single child strand of execution. Something like await_all!(fut1, fut2) could be less amgigious

[vertexclique]

I've written about naming in here. Might be useful:
async-rs/async-std#14 (comment)

I think std::iter is not compliant with reactive streams specification.
Also, I think merge and friends should go to other crate defining these.

[yoshuawuyts]

futures-preview calls this operation merge

I've looked before, and can't find any reference to a merge operation in futures-preview (ref). Do you maybe have a link of where it's being used?

---

join can also be confusing, because it is also used in "join a thread/task".

Ah yeah, that's a fair point too. Though there are a lot of references to join in std already that it's reasonably context-specific already. Though perhaps this is more different because "joining a task" could refer to task.join() or join!(task1, task2). But that's futures, not streams though.

---

Also, I think merge and friends should go to other crate defining these.

@vertexclique Could you say more about why you think this is the case? From my perspective this seems like one of core concurrency primitives for streams, which I think makes sense to at least trial as part of what we're covering.

[matklad]

Do you maybe have a link of where it's being used?

D'oh, I was looking at futures 0.1: https://docs.rs/futures/0.1.29/futures/stream/trait.Stream.html#method.merge

[yoshuawuyts]

I've been mulling this over for the past day, and still think join is the right name to try out. I'm going to go ahead and merge this so we can try out how it feels. Thanks all the input!