Rxdart: More methods

[qwilbird]

Hi,

awesome work in progress with this library!

Just wondering when you intend to add more methods, like: "distinctBy", "groupBy", "bufferByTime", "delay", "swithMap", "retryWhen" as they would really help!

Thank you and good job again!

[brianegan]

Thanks much! Glad ya like it :)

• switchMap is called flatMapLatest (the old name). We're planning on changing this to switchMap soon!
• distinctBy -- Do distinct or distinctUnique work for this case? Both methods accept an optional function to determine equality.
• groupBy -- We removed this a while back since Dart was planning on including it in the base Stream class. However, that change was reverted and I'm not sure if it's coming back. If it doesn't come back as part of Dart2, we'll bring it back in!

Need to implement:

• [x] delay
• [x] retryWhen
• [x] bufferByTime

[qwilbird]

Awesome! Any chance they will be added in a future release? :)

What do you recommend for unittesting? With RXJS, we use a lot marble testing

Thank you again!

[brianegan]

Sure, just depends on how much free time we've got, or if we run into a situation where we personally need them and have to add them, haha. Always happy to have contributions as well!

For unit testing, I'd recommend the default test package. It has "Stream Matchers".

Docs here: https://pub.dartlang.org/packages/test

Example can be seen here: https://github.com/ReactiveX/rxdart/blob/master/test/streams/error_test.dart#L11

[frankpepermans]

delay was added

[brianegan]

bufferTime added.

[brianegan]

flatMapLatest has been renamed to switchMap as part of 0.16.5.

[frankpepermans]

Loads of new ways to buffer or window streams were added as well

[escamoteur]

please add replay()

[frankpepermans]

Replay is a bit tricky, since you need to be aware when a new listen happens at the end of the downstream.
That's only possible if the Stream which is being transformed, is a broadcast Stream.

i.e.

    .replay()
    .asBroadcastStream(); // will not work with multiple listeners, as the transform happens only once

    .asBroadcastStream()
    .replay(); // works as expected

The safe approach is to always throw when the Stream provided to replay is not a broadcast Stream

[brianegan]

@frankpepermans Yah, replay is actually a pretty big topic in itself. I was originally thinking we could essentially convert the Stream into a ReplaySubject in the replay transformer, but Rx has some interesting conventions around replay and how it should work (it returns a ConnectableObservable, for example).

It's another case where renaming the library and making functionality specific to Dart might make more sense than trying to follow the Observable contract perfectly.

[frankpepermans]

@brianegan yup, I assumed it would be as easy as just using a ReplaySubject in the transformer, but sadly it's not:

var s = stream // not a broadcast Stream
    .replay()
    .asBroadcastStream();

The replay tranformer is called just once here, regardless of the amount of subscribers, which is contrary to what we need: to know when someone subscribes, and pass him/her the existing queue immediately.

If you make it a broadcast Stream before replay kicks in, then we get what we want, the transformer is actually called on each new subscriber.

[brianegan]

Hrm, that's interesting, I'll have to take a look. I assumed we would only want the transformer called once. The transformer would be responsible for listening to the original stream (just once) and then piping all events to the ReplaySubject. We would return the subject.stream in the bind function of the transformer.

Once it's a ReplaySubject stream, it should be broadcast by default and everyone who subscribes should be all good?

[long1eu]

What about this implementation of retryWhen https://gist.github.com/long1eu/9c9e257f111e0c8d3ba44cbf840a8767

[brianegan]

Thanks @long1eu! I think we'll need to make this a Stream class rather than a Transformer, since retry / retryWhen should re-subscribe to the original Stream rather than switching to a new Stream when the original Stream fails. Since this won't work in Dart as Streams are single-subscription, the StreamFactory needs to be the "Source of Truth" -- it should create the original Stream that could fail, and should re-create that same Stream when it does. In your implementation, it starts with 1 stream then switches to another created by the Factory upon failure, which means you'll often be repeating the same code before the retryWhen and inside the factory.

[long1eu]

I've tried to build it based on your suggestions on top of the retry count implementation.

https://gist.github.com/long1eu/db508dcf2c1a28ec226a939ca5fe28e0

[qwilbird]

@long1eu That looks nice - looking forward being able to use retryWhen!
@brianegan, what do you think of this implementation?

[brianegan]

Oh dang, sorry I missed this one. Overall -- Looking much better! Thanks for taking the time to write that up :)

I think it would make sense to extract the ErrorAndStackTrace class out into it's own file in the lib/src/utils and share it between retry and retryWhen.

It'd also be great to have Docs + Tests for this code so we can let folks know how they can use this constructor and so we verify it works correctly now and into the future. You can check out the retry tests for an example of how to write these types of tests, or if ya need help please let me know.

Once all that is done, it'd be great to submit this work as a PR! Thanks again for all the work :)

[brianegan]

Hey all -- just published version 0.18.1 that includes retryWhen -- Thanks so much to @long1eu for contributing 🔥 🎉 ❤️

I'm going to go ahead and close this issue out now since we've covered the original 3 operators. The replay work has been split out into it's own issue for tracking since it's a larger chunk of work.

[qwilbird]

awesome work guys! Thank you so much!