Futures-rs: Publish/Broadcast stream transformer

[LalitMaganti]

One thing which I noticed was missing from the Streams implementation was some sort of "broadcast" transformer which allowed a stream to be subscribed to multiple times (i.e. polled/scheduled by multiple tasks in the language of futures-rs).

I was wondering if there was any interest in having this as I've been prototyping this and would love to upstream it!

[alexcrichton]

Thanks for the report! Could you detail a bit more what you're thinking this sort of combinator would be used for? We're all for adding more combinators of course :)

[LalitMaganti]

So the main use of the combinator would be to allow multiple downstream streams to receive the same notifications without running the stream multiple times.

For example, suppose I had a Stream which was emitting response objects which have a unique monotonically increasing id. At various points, I may wish to listen for a sub-stream of these response objects (maybe filtered on an id range) before I stop listening. However, I may still have other combinators on other tasks which are still listening to this stream and so would need the stream to continue emitting. This would be an ideal usecase for publish.

For a bit of context, I come from an Rx approach to streams and was actually prototyping bringing Rx to Rust before I saw you guys release this library. Publish is one of operators which is considered a cornerstone of Rx - you can see the description for it at http://reactivex.io/documentation/operators/publish.html. I don't particularily think the concept of "connection" is required in futures-rs. A combinator which does both publish and refcount ( http://reactivex.io/documentation/operators/refcount.html) is what I'm thinking about.

Thanks :)

[alexcrichton]

I was reading up a bit on Rx and from what I understand (which certainly isn't all of it) it looks like Stream is 90% of what Rx is? In that sense it seems like we may just be missing a few combinators, and you're thinking that the publish and refcount combinators are the ones we're missing?

[LalitMaganti]

There's a lot of combinators which I think could be added. The most useful ones in Rx are subscribeOn and observeOn but those don't fit tool well with the model of futures-rs with tasks. Publish is another one of those which is extremely useful for sure.

[alexcrichton]

Sounds good to me, and I'd be more than willing to add some more combinators. If you'd like to work on it, just let me know if you need any help!

[josephDunne]

I have added something like this as a History stream adaptor see #111

[alexcrichton]

I believe there's some relevant work in this area since this was opened (such as https://github.com/carllerche/futures-broadcast) and in general the work is concluding that it's a meaty enough implementation that it may not wish to belong in this repo, so I'm going to close this.