Adding Instrumentation for reactive results and when they finish #4084
Conversation
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
bbakerman
commented
Aug 12, 2025
| */ | ||
| @SuppressWarnings("TypeParameterUnusedInFormals") | ||
| @PublicApi | ||
| public class InstrumentationReactiveResultsParameters { |
There was a problem hiding this comment.
New parameters for the Instrumentation method
bbakerman
commented
Aug 12, 2025
| @Nullable | ||
| default InstrumentationContext<Void> beginReactiveResults(InstrumentationReactiveResultsParameters parameters, InstrumentationState state) { | ||
| return noOp(); | ||
| } |
There was a problem hiding this comment.
Notice how its Void - the end of a Publisher has no result object - it has many objects but there is no actual result at the end
bbakerman
commented
Aug 12, 2025
| */ | ||
| public static <T> Publisher<T> whenPublisherFinishes(Publisher<T> publisher, Consumer<? super Throwable> atTheEndCallback) { | ||
| return new AtTheEndPublisher<>(publisher, atTheEndCallback); | ||
| } |
There was a problem hiding this comment.
This helper allows us to wrap a Publisher so that a callback is made when its finished.
bbakerman
commented
Aug 12, 2025
| if (keepOrdered) { | ||
| mappingPublisher = new CompletionStageMappingOrderedPublisher<>(upstreamPublisher, mapper); | ||
| } else { | ||
| mappingPublisher = new CompletionStageMappingPublisher<>(upstreamPublisher, mapper); | ||
| } | ||
| publisher = ReactiveSupport.whenPublisherFinishes(mappingPublisher, whenDone); |
There was a problem hiding this comment.
we now delegate and invoke the whenDone callback
bbakerman
commented
Aug 12, 2025
|
|
||
| // | ||
| // wrap this Publisher into one that can call us back when the publishing is done either in error or successful | ||
| publisher = ReactiveSupport.whenPublisherFinishes(publisher, throwable -> ctx.onCompleted(null, throwable)); |
There was a problem hiding this comment.
Same here - we delegate to the original Publisher but the callback here is done when the Publisher is finished
Test Results 326 files - 649 326 suites - 649 3m 31s ⏱️ - 7m 5s Results for commit d064c7f. ± Comparison against base commit 3ba9750. This pull request removes 558 and adds 203 tests. Note that renamed tests count towards both.This pull request skips 1 test.♻️ This comment has been updated with latest results. |
…ainedInstrumentation fixups
bbakerman
commented
Aug 13, 2025
| reactiveCtx.onDispatched(); | ||
|
|
||
| SubscriptionPublisher mapSourceToResponse = new SubscriptionPublisher(publisher, mapperFunction, keepOrdered, | ||
| throwable -> reactiveCtx.onCompleted(null, throwable)); |
There was a problem hiding this comment.
we pass in the call back here. Why?
We want the SubscriptionPublisher to be the implementation we pass back - we have tests for this.
So I changed SubscriptionPublisher internally to know when its finished
bbakerman
commented
Aug 13, 2025
|
|
||
| called | ||
| throwable.message == "BANG" | ||
| } |
There was a problem hiding this comment.
Testing that the new helper works
bbakerman
commented
Aug 13, 2025
| return t1 == t2 | ||
| } | ||
| return false | ||
| } |
There was a problem hiding this comment.
Just to make it easier to check large list of values for ordered things. Helpful with Instrumentation "step" testing
bbakerman
commented
Aug 13, 2025
| ["a"], ["b"], ["c"], ["X"], ["e"], ["f"], ["g"]) | ||
| then: | ||
| !actual | ||
|
|
There was a problem hiding this comment.
Test thats the TestUtil helpers work
bbakerman
commented
Aug 13, 2025
| ]) | ||
|
|
||
| // last of all it finishes | ||
| instrumentation.executionList.last == "end:reactive-results-defer" |
There was a problem hiding this comment.
Test that @defer publisher can tell us when they are finished
…tation callback - fixed tests
…lback-after-reactive-publishers-finish
…tation callback -null marked
bbakerman
commented
Aug 13, 2025
| ]) | ||
|
|
||
| // last of all it finishes | ||
| TestUtil.last(instrumentation.executionList) == "end:reactive-results-defer" |
There was a problem hiding this comment.
Rather than test ALL the instrumentation steps again - we look at sub sections and make sure they are in order
This balances completeness with code readability / maintain ability
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.
This adds Insrtrumentation support for reactive results - namely defer Publishers and Subscription Publisher.
When the Publisher is finished (throwable or not error) then the Instrumentation context is considered finished and called.
See #4083 for related code