Refactor of setting up our transformers #412
Conversation
Codecov Report
@@ Coverage Diff @@
## master #412 +/- ##
==========================================
- Coverage 93.55% 92.68% -0.87%
==========================================
Files 60 61 +1
Lines 2047 2037 -10
==========================================
- Hits 1915 1888 -27
- Misses 132 149 +17 |
There was a problem hiding this comment.
I took a few hours to review this one, and overall I think the approach makes sense.
Should we release this as a -dev release to gather early feedback and see if the change hits folks in a negative way overall? I still think it will be worth fixing, but hard to get a sense of the impact this change has since it fixes more subtle bugs.
| } | ||
|
|
||
| @override | ||
| FutureOr onCancel(EventSink<S> sink) => _mapperSubscription?.cancel(); |
There was a problem hiding this comment.
Should we close the output sink at this point as well?
There was a problem hiding this comment.
Not at cancel time, onCancel will be called if all subscribers have cancelled for example, but the Stream could still be active and be subscribed to again at a later stage.
| // this method is ran before any other events might be added. | ||
| // Once the first event(s) is/are successfully added, this method | ||
| // will not trigger again. | ||
| void _safeAddFirstEvent() { |
There was a problem hiding this comment.
Is this a good idea? I've actually run into exactly the bug you're talking about, but this wold effectively turn any sync Stream into an async Stream. I think the bug is sort of "intended" as there are special considerations to take when using Sync StreamControllers.
| // forwarded events. | ||
| if (stream is Subject<T>) { | ||
| controller = stream.createForwardingController( | ||
| onListen: onListen, onCancel: onCancel, sync: true); |
There was a problem hiding this comment.
Should these sync property always be true here? Very hard to know without a stream.isSync or something :/
There was a problem hiding this comment.
Heh no idea :) I've checked stream_transform from the google devs, they transform using sync true always.
There was a problem hiding this comment.
I think, since we use transform sinks here, if the root Stream is async, then the transformed Streams will be too. It's like awaiting a Future and mapping the result in sync,
You still need to await the initial time.
|
|
||
| // Create a new Controller, which will serve as a trampoline for | ||
| // forwarded events. | ||
| if (stream is Subject<T>) { |
There was a problem hiding this comment.
Do we need this special case? It looks like we listen to the source Stream in onListen and add all events from the source Stream to this new sink. Therefore, it seems like the events should be properly replayed in that manner?
There was a problem hiding this comment.
I looked at this a bit more, and it might point to a general problem with our Subject implementations that we're fixing here: Our subjects currently replay events after the close method is called. Do you think that's actually the correct behavior? It kinda seems like they should just emitDone like a normal StreamController at that point.
There was a problem hiding this comment.
Yeah we do need this behavior, if for example the source is a ReplaySubject, then using an intermediate broadcast one will not replay events. It's different from our previous impl. where binding would spawn a new Controller always.
If we omit the Subject duplication, our tests will fail accordingly.
I'll add a better explanation later, in short, the very root Stream will not always be resubscribed to when new subscripers register.
Ok, this turned out to be bigger than expected (as per usual haha).
Some issues made me take a closer look at how we are setting up our
Streamtransformations,TL;DR : we were doing it wrong, slightly!
The way we did things before, is by always creating a new
StreamControllerin our transformers, and then immediately subscribing to them. Afterwards, in theonListenhandler, we would then actually subscribe to the underlyingStream.This was causing some side effects, I'll elaborate more on those later on.
Ideally, we want to have the same effects that default transformers have, like for example the built-in
map.At first, I took a look at the stream_transform library and copied how they do it, which was a better way but still not giving the exact same behaviour as the default transformers.
Finally, I went with using
Stream.eventTransformed, which as a result also cleans up a lot of things in a better way.This PR should fix that up.
As an effect of doing things differently, our do transformer has different behaviour in 2 cases, but the new behaviour is actually in line with how it is supposed to work for Dart streams:
onListenwill now only fire on the first subscriber, if there are more, then all current subscribers need to cancel first. If then another new subscriber registers,onListenwill be called a second time.onCancelwill now only fire when all active subscribers have cancelled.This is a big change, and will sure spark a conversation, please do test and ask away :)