fix: preserve ValueStream replay for listenOnlyOnce transformers#815
Open
rbellens wants to merge 2 commits into
Open
fix: preserve ValueStream replay for listenOnlyOnce transformers#815rbellens wants to merge 2 commits into
rbellens wants to merge 2 commits into
Conversation
DoStreamTransformer uses forwardStream with listenOnlyOnce, which previously used a plain broadcast StreamController. When the source is a ValueStream (e.g. BehaviorSubject), late listeners such as sequential Stream.first calls would hang because the latest value was not replayed. Forward ValueStream sources through a BehaviorSubject output so side effects still share a single upstream subscription while new listeners receive the latest value. Fixes ReactiveX#657 Relates to ReactiveX#605, ReactiveX#587
…ubscribe Replace the listenOnlyOnce single-subscription path with sideEffectsOnce broadcast forwarding so replay sources behave correctly (ReactiveX#605, ReactiveX#657) while doOn* callbacks still fire once per upstream event.
There was a problem hiding this comment.
Pull request overview
This PR addresses a long-standing edge case in RxDart’s doOn* operators when used on broadcast replay sources (e.g. BehaviorSubject), where late/serial listeners (like repeated Stream.first) could hang due to upstream subscription/cancellation handling.
Changes:
- Refactors
forwardStream()to use a namedsideEffectsOnceoption and introduces a dedicated broadcast forwarding path to ensuredoOn*side effects run only once per upstream event. - Fixes re-listen behavior by resetting the internal
cancelledflag on_forward’sonListen. - Adds regression tests for issues #657 and #605 around sequential
firstand cancel/relisten behavior withBehaviorSubject.
Reviewed changes
Copilot reviewed 3 out of 3 changed files in this pull request and generated 5 comments.
| File | Description |
|---|---|
| packages/rxdart/lib/src/utils/forwarding_stream.dart | Adds a new broadcast forwarding strategy for doOn* with sideEffectsOnce, and fixes _forward re-listen after cancel. |
| packages/rxdart/lib/src/transformers/do.dart | Updates DoStreamTransformer to use forwardStream(..., sideEffectsOnce: true). |
| packages/rxdart/test/transformers/do_test.dart | Adds regression tests covering sequential first and cancel/relisten behavior on BehaviorSubject. |
Comment on lines
+3
to
+4
| import 'package:rxdart/src/streams/value_stream.dart'; | ||
| import 'package:rxdart/src/subjects/behavior_subject.dart'; |
Comment on lines
+73
to
+95
| (v) { | ||
| if (controllers.first == controller) { | ||
| sink.onData(v); | ||
| } else { | ||
| // This listener has its own upstream subscription (source semantics), | ||
| // but side effects already ran on the primary listener's path. | ||
| sink.sink.add(v); | ||
| } | ||
| }, | ||
| onError: (Object e, StackTrace s) { | ||
| if (controllers.first == controller) { | ||
| sink.onError(e, s); | ||
| } else { | ||
| sink.sink.addError(e, s); | ||
| } | ||
| }, | ||
| onDone: () { | ||
| if (controllers.first == controller) { | ||
| sink.onDone(); | ||
| } else { | ||
| sink.sink.close(); | ||
| } | ||
| }, |
Comment on lines
+501
to
+502
| await subscription.cancel(); | ||
| }); |
Comment on lines
+517
to
+518
| await subscription.cancel(); | ||
| }); |
Comment on lines
+525
to
+527
| await stream.listen(null).cancel(); | ||
| expect(await stream.first, 'b'); | ||
| }); |
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Summary
DoStreamTransformer(and otherlistenOnlyOncetransformers) hanging on late listeners when the source is a replayValueStream(e.g. seededBehaviorSubject)cancelled = falsein_forward.onListenso upstream is re-subscribed after cancelBehaviorSubjectoutput controller when the source hasisReplayValueStreamStream.multi#605 and DoStreamTransformer not behaving as expected #657Test plan
dart test test/transformers/do_test.dartdart test test/transformers/do_test.dart --name "issue/605|issue/657"Fixes DoStreamTransformer not behaving as expected #657
Relates to Refactor forwardStream: uses
Stream.multi#605, Deadlocks in tests after upgrading rxdart #587