Make the subscriber in awaitOne less permissive #2586
Conversation
| } | ||
|
|
||
| override fun onNext(t: T) { | ||
| val sub = subscription.checkInitialized("onNext") | ||
| subscription.checkInitialized("onNext") |
There was a problem hiding this comment.
Double checkInitialized seems to be redundant
| } | ||
| if (seenValue) | ||
| // TODO: decide if we want to be lenient here: after all, nothing breaks if this isn't true | ||
| moreThanOneValueProvidedException(mode) |
There was a problem hiding this comment.
When writing a comment along with a statement, please wrap it in curly braces, otherwise it's a bit hard to read
There was a problem hiding this comment.
Throwing here breaks #2.13. This is expected, but still worth properly documenting to know why it is here even tho it violates the spec.
Also, removing this condition does not trigger any test failures, because double-resume also throws IllegalStateException. It's worth verifying the exception message at least once
There was a problem hiding this comment.
I am now verifying the exception messages. I don't like what I did in order to achieve it, but I like copy-pasting exception messages into tests even less, given how brittle that approach is. Maybe you have some better suggestions?
There was a problem hiding this comment.
The approach we're consistently using is the following:
- Do not expose exception messages to test in any way (e.g. via
internalfunction) - Instead of strictly comparing the error message as a whole, typically it's enough to do something like
e.message!!.contains("some_key_word_or_meaningful_part_of_the_message").
It simplifies the maintenance a bit, reduces the number of functions/string consts, but it's still brittle. Yet exception messages are rarely changed and the purpose of the test is rather fix the current behaviour
| @@ -169,6 +188,8 @@ private suspend fun <T> Publisher<T>.awaitOne( | |||
|
|
|||
| @Suppress("UNCHECKED_CAST") | |||
| override fun onComplete() { | |||
| subscription.checkInitialized("onComplete") // TODO: maybe don't enforce? | |||
There was a problem hiding this comment.
It seems like other implementations from reactive libraries do not enforce such rule, maybe we could stick with that as well
| block: Subscriber<in T>.(n: Long) -> Unit, | ||
| ) = | ||
| assertFailsWith<IllegalStateException> { publisher(block).operation() }.let { | ||
| message == null || it.message == message |
There was a problem hiding this comment.
This line doesn't actually test exception messages, the resulting boolean is ignored
There was a problem hiding this comment.
Yeah, sorry. Fixed in the commit that changes the whole error-passing mechanism here.
| } | ||
| if (seenValue) | ||
| // TODO: decide if we want to be lenient here: after all, nothing breaks if this isn't true | ||
| moreThanOneValueProvidedException(mode) |
There was a problem hiding this comment.
The approach we're consistently using is the following:
- Do not expose exception messages to test in any way (e.g. via
internalfunction) - Instead of strictly comparing the error message as a whole, typically it's enough to do something like
e.message!!.contains("some_key_word_or_meaningful_part_of_the_message").
It simplifies the maintenance a bit, reduces the number of functions/string consts, but it's still brittle. Yet exception messages are rarely changed and the purpose of the test is rather fix the current behaviour
The implementation of Reactive Streams' Subscriber used for `await*` operations was assuming that the publisher is correct. Now, the implementation detects some instances of problematic behavior for publishers and reports them. Fixes #2079
This is a more robust approach, as exceptions are prone to being lost in the async environments, especially if some components are (known to be) misbehaving, so it's worth it to report the fundamental errors to the system.
|
@qwwdfsad could you please check the latest commit? I spotted several errors during the final review, fixed them, and added a test to verify a new statement in a comment. |
The implementation of Reactive Streams' Subscriber used for `await*` operations was assuming that the publisher is correct. Now, the implementation detects some instances of problematic behavior for publishers and reports them. Fixes Kotlin#2079
Reactive Streams' Subscriber's implementation for
await*operations was assuming that the publisher is correct. Now, the implementation detects some instances of problematic behavior for publishers and reports them.Fixes #2079