Consistently handle exceptions in reactive streams #2646
Merged
Conversation
This file contains 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
|
I will be adding some tests to this PR but would appreciate some feedback on the implementation in the meantime. |
dkhalanskyjb
force-pushed
the
fix-publishing-ignoring-cancellations
branch
from
f7c53ae
to
ed1cf6d
Compare
qwwdfsad
reviewed
Apr 13, 2021
dkhalanskyjb
force-pushed
the
fix-publishing-ignoring-cancellations
branch
from
ce6cc5f
to
bee8e90
Compare
* Fixed `PublisherCoroutine` and `rxObservable` ignoring cancellations. * Fatal exceptions are not treated in a special manner by us anymore. Instead, we follow the requirement in the reactive streams specification that, in case some method of `Subscriber` throws, that subscriber MUST be considered cancelled, and the exception MUST be reported in some place other than `onError`. * Fixed `trySend` sometimes throwing in `PublisherCoroutine` and `rxObservable`. * When an exception happens inside a cancellation handler, we now consistently throw the original exception passed to the handler, with the new exception added as suppressed.
dkhalanskyjb
force-pushed
the
fix-publishing-ignoring-cancellations
branch
from
bee8e90
to
8f9c70d
Compare
* Fixed `doLockedNext` not releasing the lock in `PublisherCoroutine` if `null` is emitted * Fixed `flux`, `publish`, `rxObservable` and `rxFlowable` incorrectly reporting `isClosedForSend == true` just after closing the channel
qwwdfsad
approved these changes
Apr 19, 2021
| subscriber.onComplete() | ||
| } else { | ||
| @Suppress("INVISIBLE_MEMBER") | ||
| val unwrappedCause = unwrap(cause) |
pablobaxter
pushed a commit
to pablobaxter/kotlinx.coroutines
that referenced
this pull request
Sep 14, 2022
* Fixed `PublisherCoroutine`, `rxObservable`, and `Flow.toPublisher` ignoring cancellations. * Fatal exceptions are not treated in a special manner by us anymore. Instead, we follow the requirement in the reactive streams specification that, in case some method of `Subscriber` throws, that subscriber MUST be considered canceled, and the exception MUST be reported in someplace other than `onError`. * Fixed `trySend` sometimes throwing in `PublisherCoroutine` and `rxObservable`. * When an exception happens inside a cancellation handler, we now consistently throw the original exception passed to the handler, with the new exception added as suppressed. * Fixed `PublisherCoroutine` and `rxObservable` claiming that the channel is not closed for send for some time after `close()` has finished. * Fixed publishers sometimes signalling `onComplete()` after cancellation even though their streams are not finite. Fixes Kotlin#2173
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.
PublisherCoroutine,rxObservable, andFlow.toPublisherignoring cancellations.anymore. Instead, we follow the requirement in the reactive
streams specification that, in case some method of
Subscriberthrows, that subscriber MUST be considered canceled, and the
exception MUST be reported in someplace other than
onError.trySendsometimes throwing inPublisherCoroutineandrxObservable.consistently throw the original exception passed to the handler,
with the new exception added as suppressed.
PublisherCoroutineandrxObservableclaiming that thechannel is not closed for send for some time after
close()hasfinished.
onComplete()aftercancellation even though their streams are not finite.
Fixes #2173