diff --git a/src/main/java/graphql/execution/Execution.java b/src/main/java/graphql/execution/Execution.java index 37d43a4713..38c82f5a53 100644 --- a/src/main/java/graphql/execution/Execution.java +++ b/src/main/java/graphql/execution/Execution.java @@ -21,6 +21,8 @@ import graphql.execution.instrumentation.dataloader.PerLevelDataLoaderDispatchStrategy; import graphql.execution.instrumentation.parameters.InstrumentationExecuteOperationParameters; import graphql.execution.instrumentation.parameters.InstrumentationExecutionParameters; +import graphql.execution.instrumentation.parameters.InstrumentationReactiveResultsParameters; +import graphql.execution.reactive.ReactiveSupport; import graphql.extensions.ExtensionsBuilder; import graphql.incremental.DelayedIncrementalPartialResult; import graphql.incremental.IncrementalExecutionResultImpl; @@ -245,9 +247,17 @@ private CompletableFuture incrementalSupport(ExecutionContext e return result.thenApply(er -> { IncrementalCallState incrementalCallState = executionContext.getIncrementalCallState(); if (incrementalCallState.getIncrementalCallsDetected()) { + InstrumentationReactiveResultsParameters parameters = new InstrumentationReactiveResultsParameters(executionContext, InstrumentationReactiveResultsParameters.ResultType.DEFER); + InstrumentationContext ctx = nonNullCtx(executionContext.getInstrumentation().beginReactiveResults(parameters, executionContext.getInstrumentationState())); + // we start the rest of the query now to maximize throughput. We have the initial important results, // and now we can start the rest of the calls as early as possible (even before someone subscribes) Publisher publisher = incrementalCallState.startDeferredCalls(); + ctx.onDispatched(); + + // + // 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)); return IncrementalExecutionResultImpl.fromExecutionResult(er) // "hasNext" can, in theory, be "false" when all the incremental items are delivered in the diff --git a/src/main/java/graphql/execution/SubscriptionExecutionStrategy.java b/src/main/java/graphql/execution/SubscriptionExecutionStrategy.java index 50cfcb4bca..f1e1a9a919 100644 --- a/src/main/java/graphql/execution/SubscriptionExecutionStrategy.java +++ b/src/main/java/graphql/execution/SubscriptionExecutionStrategy.java @@ -12,6 +12,7 @@ import graphql.execution.instrumentation.parameters.InstrumentationExecutionParameters; import graphql.execution.instrumentation.parameters.InstrumentationExecutionStrategyParameters; import graphql.execution.instrumentation.parameters.InstrumentationFieldParameters; +import graphql.execution.instrumentation.parameters.InstrumentationReactiveResultsParameters; import graphql.execution.reactive.SubscriptionPublisher; import graphql.language.Field; import graphql.schema.GraphQLFieldDefinition; @@ -77,7 +78,13 @@ public CompletableFuture execute(ExecutionContext executionCont } Function> mapperFunction = eventPayload -> executeSubscriptionEvent(executionContext, parameters, eventPayload); boolean keepOrdered = keepOrdered(executionContext.getGraphQLContext()); - SubscriptionPublisher mapSourceToResponse = new SubscriptionPublisher(publisher, mapperFunction, keepOrdered); + + InstrumentationReactiveResultsParameters instrumentationReactiveResultsParameters = new InstrumentationReactiveResultsParameters(executionContext, InstrumentationReactiveResultsParameters.ResultType.SUBSCRIPTION); + InstrumentationContext reactiveCtx = nonNullCtx(executionContext.getInstrumentation().beginReactiveResults(instrumentationReactiveResultsParameters, executionContext.getInstrumentationState())); + reactiveCtx.onDispatched(); + + SubscriptionPublisher mapSourceToResponse = new SubscriptionPublisher(publisher, mapperFunction, keepOrdered, + throwable -> reactiveCtx.onCompleted(null, throwable)); return new ExecutionResultImpl(mapSourceToResponse, executionContext.getErrors()); }); diff --git a/src/main/java/graphql/execution/instrumentation/ChainedInstrumentation.java b/src/main/java/graphql/execution/instrumentation/ChainedInstrumentation.java index e48e20fa61..2779a10a01 100644 --- a/src/main/java/graphql/execution/instrumentation/ChainedInstrumentation.java +++ b/src/main/java/graphql/execution/instrumentation/ChainedInstrumentation.java @@ -15,6 +15,7 @@ import graphql.execution.instrumentation.parameters.InstrumentationFieldCompleteParameters; import graphql.execution.instrumentation.parameters.InstrumentationFieldFetchParameters; import graphql.execution.instrumentation.parameters.InstrumentationFieldParameters; +import graphql.execution.instrumentation.parameters.InstrumentationReactiveResultsParameters; import graphql.execution.instrumentation.parameters.InstrumentationValidationParameters; import graphql.language.Document; import graphql.schema.DataFetcher; @@ -137,6 +138,10 @@ public InstrumentationContext beginExecuteOperation(Instrumenta return chainedCtx(state, (instrumentation, specificState) -> instrumentation.beginExecuteOperation(parameters, specificState)); } + @Override + public @Nullable InstrumentationContext beginReactiveResults(InstrumentationReactiveResultsParameters parameters, InstrumentationState state) { + return chainedCtx(state, (instrumentation, specificState) -> instrumentation.beginReactiveResults(parameters, specificState)); + } @Override public ExecutionStrategyInstrumentationContext beginExecutionStrategy(InstrumentationExecutionStrategyParameters parameters, InstrumentationState state) { diff --git a/src/main/java/graphql/execution/instrumentation/Instrumentation.java b/src/main/java/graphql/execution/instrumentation/Instrumentation.java index 1ca6c268ed..565c4333da 100644 --- a/src/main/java/graphql/execution/instrumentation/Instrumentation.java +++ b/src/main/java/graphql/execution/instrumentation/Instrumentation.java @@ -12,6 +12,7 @@ import graphql.execution.instrumentation.parameters.InstrumentationFieldCompleteParameters; import graphql.execution.instrumentation.parameters.InstrumentationFieldFetchParameters; import graphql.execution.instrumentation.parameters.InstrumentationFieldParameters; +import graphql.execution.instrumentation.parameters.InstrumentationReactiveResultsParameters; import graphql.execution.instrumentation.parameters.InstrumentationValidationParameters; import graphql.language.Document; import graphql.schema.DataFetcher; @@ -120,6 +121,21 @@ default InstrumentationContext beginExecuteOperation(Instrument return noOp(); } + /** + * This is called just before the execution of any reactive results, namely incremental deferred results or subscriptions. When the {@link org.reactivestreams.Publisher} + * finally ends (with either a {@link Throwable} or none) then the {@link InstrumentationContext} wil be called back to say the reactive results + * have finished. + * + * @param parameters the parameters to this step + * @param state the state created during the call to {@link #createStateAsync(InstrumentationCreateStateParameters)} + * + * @return a nullable {@link InstrumentationContext} object that will be called back when the step ends (assuming it's not null) + */ + @Nullable + default InstrumentationContext beginReactiveResults(InstrumentationReactiveResultsParameters parameters, InstrumentationState state) { + return noOp(); + } + /** * This is called each time an {@link graphql.execution.ExecutionStrategy} is invoked, which may be multiple times * per query as the engine recursively descends over the query. diff --git a/src/main/java/graphql/execution/instrumentation/NoContextChainedInstrumentation.java b/src/main/java/graphql/execution/instrumentation/NoContextChainedInstrumentation.java index 5ae45dc9df..8d56230825 100644 --- a/src/main/java/graphql/execution/instrumentation/NoContextChainedInstrumentation.java +++ b/src/main/java/graphql/execution/instrumentation/NoContextChainedInstrumentation.java @@ -8,6 +8,7 @@ import graphql.execution.instrumentation.parameters.InstrumentationFieldCompleteParameters; import graphql.execution.instrumentation.parameters.InstrumentationFieldFetchParameters; import graphql.execution.instrumentation.parameters.InstrumentationFieldParameters; +import graphql.execution.instrumentation.parameters.InstrumentationReactiveResultsParameters; import graphql.execution.instrumentation.parameters.InstrumentationValidationParameters; import graphql.language.Document; import graphql.validation.ValidationError; @@ -73,6 +74,11 @@ public InstrumentationContext beginExecuteOperation(Instrumenta return runAll(state, (instrumentation, specificState) -> instrumentation.beginExecuteOperation(parameters, specificState)); } + @Override + public @Nullable InstrumentationContext beginReactiveResults(InstrumentationReactiveResultsParameters parameters, InstrumentationState state) { + return runAll(state, (instrumentation, specificState) -> instrumentation.beginReactiveResults(parameters, specificState)); + } + @Override public ExecutionStrategyInstrumentationContext beginExecutionStrategy(InstrumentationExecutionStrategyParameters parameters, InstrumentationState state) { return runAll(state, (instrumentation, specificState) -> instrumentation.beginExecutionStrategy(parameters, specificState)); diff --git a/src/main/java/graphql/execution/instrumentation/parameters/InstrumentationReactiveResultsParameters.java b/src/main/java/graphql/execution/instrumentation/parameters/InstrumentationReactiveResultsParameters.java new file mode 100644 index 0000000000..a4e312efd6 --- /dev/null +++ b/src/main/java/graphql/execution/instrumentation/parameters/InstrumentationReactiveResultsParameters.java @@ -0,0 +1,39 @@ +package graphql.execution.instrumentation.parameters; + +import graphql.PublicApi; +import graphql.execution.ExecutionContext; +import graphql.execution.instrumentation.Instrumentation; +import org.jspecify.annotations.NullMarked; + +/** + * Parameters sent to {@link Instrumentation} methods + */ +@SuppressWarnings("TypeParameterUnusedInFormals") +@PublicApi +@NullMarked +public class InstrumentationReactiveResultsParameters { + + /** + * What type of reactive results was the {@link org.reactivestreams.Publisher} + */ + public enum ResultType { + DEFER, SUBSCRIPTION + } + + private final ExecutionContext executionContext; + private final ResultType resultType; + + public InstrumentationReactiveResultsParameters(ExecutionContext executionContext, ResultType resultType) { + this.executionContext = executionContext; + this.resultType = resultType; + } + + + public ExecutionContext getExecutionContext() { + return executionContext; + } + + public ResultType getResultType() { + return resultType; + } +} diff --git a/src/main/java/graphql/execution/reactive/ReactiveSupport.java b/src/main/java/graphql/execution/reactive/ReactiveSupport.java index 80deb38be0..c513cc75a5 100644 --- a/src/main/java/graphql/execution/reactive/ReactiveSupport.java +++ b/src/main/java/graphql/execution/reactive/ReactiveSupport.java @@ -4,11 +4,14 @@ import graphql.Internal; import org.reactivestreams.FlowAdapters; import org.reactivestreams.Publisher; +import org.reactivestreams.Subscriber; +import org.reactivestreams.Subscription; import java.util.Objects; import java.util.concurrent.CompletableFuture; import java.util.concurrent.Flow; import java.util.concurrent.atomic.AtomicReference; +import java.util.function.Consumer; /** * This provides support for a DataFetcher to be able to @@ -141,4 +144,55 @@ public void onComplete() { onCompleteImpl(); } } + + /** + * Our reactive {@link SingleSubscriberPublisher} supports only a single subscription + * so this can be used a delegate to perform a call back when the given Publisher + * actually finishes without adding an extra subscription to the delegate Publisher + * + * @param publisher the publisher to wrap + * @param atTheEndCallback the callback when the {@link Publisher} has finished + * @param for two + */ + public static Publisher whenPublisherFinishes(Publisher publisher, Consumer atTheEndCallback) { + return new AtTheEndPublisher<>(publisher, atTheEndCallback); + } + + static class AtTheEndPublisher implements Publisher { + + private final Publisher delegatePublisher; + private final Consumer atTheEndCallback; + + public AtTheEndPublisher(Publisher delegatePublisher, Consumer atTheEndCallback) { + this.delegatePublisher = delegatePublisher; + this.atTheEndCallback = atTheEndCallback; + } + + @Override + public void subscribe(Subscriber originalSubscriber) { + delegatePublisher.subscribe(new Subscriber<>() { + @Override + public void onSubscribe(Subscription s) { + originalSubscriber.onSubscribe(s); + } + + @Override + public void onNext(T t) { + originalSubscriber.onNext(t); + } + + @Override + public void onError(Throwable t) { + originalSubscriber.onError(t); + atTheEndCallback.accept(t); + } + + @Override + public void onComplete() { + originalSubscriber.onComplete(); + atTheEndCallback.accept(null); + } + }); + } + } } diff --git a/src/main/java/graphql/execution/reactive/SubscriptionPublisher.java b/src/main/java/graphql/execution/reactive/SubscriptionPublisher.java index 7b8b08d19f..4951b451df 100644 --- a/src/main/java/graphql/execution/reactive/SubscriptionPublisher.java +++ b/src/main/java/graphql/execution/reactive/SubscriptionPublisher.java @@ -7,6 +7,7 @@ import org.reactivestreams.Subscriber; import java.util.concurrent.CompletionStage; +import java.util.function.Consumer; import java.util.function.Function; @@ -25,6 +26,7 @@ public class SubscriptionPublisher implements Publisher { private final CompletionStageMappingPublisher mappingPublisher; + private final Publisher publisher; /** * Subscription consuming code is not expected to create instances of this class @@ -34,12 +36,13 @@ public class SubscriptionPublisher implements Publisher { * @param keepOrdered this indicates that the order of results should be kep in the same order as the source events arrive */ @Internal - public SubscriptionPublisher(Publisher upstreamPublisher, Function> mapper, boolean keepOrdered) { + public SubscriptionPublisher(Publisher upstreamPublisher, Function> mapper, boolean keepOrdered, Consumer whenDone) { if (keepOrdered) { mappingPublisher = new CompletionStageMappingOrderedPublisher<>(upstreamPublisher, mapper); } else { mappingPublisher = new CompletionStageMappingPublisher<>(upstreamPublisher, mapper); } + publisher = ReactiveSupport.whenPublisherFinishes(mappingPublisher, whenDone); } /** @@ -52,6 +55,6 @@ public Publisher getUpstreamPublisher() { @Override public void subscribe(Subscriber subscriber) { - mappingPublisher.subscribe(subscriber); + publisher.subscribe(subscriber); } } diff --git a/src/test/groovy/graphql/TestUtil.groovy b/src/test/groovy/graphql/TestUtil.groovy index 48a1ad11a7..fce61c315c 100644 --- a/src/test/groovy/graphql/TestUtil.groovy +++ b/src/test/groovy/graphql/TestUtil.groovy @@ -329,6 +329,81 @@ class TestUtil { } + /** + * Helper to say that a sub list is contained inside rhe master list in order for its entire length + * + * @param source the source list to check + * @param target the target list + * @return true if the target lists are inside the source list in order + */ + static boolean listContainsInOrder(List source, List target, List... targets) { + def index = indexOfSubListFrom(0, source, target) + if (index == -1) { + return false + } + for (List list : targets) { + index = indexOfSubListFrom(index, source, list) + if (index == -1) { + return false + } + } + return true + } + + /** + * Finds the index of the target list inside the source list starting from the specified index + * + * @param startIndex the starting index + * @param source the source list + * @param target the target list + * @return the index of the target list or -1 + */ + static int indexOfSubListFrom(int startIndex, List source, List target) { + def subListSize = target.size() + def masterListSize = source.size() + if (masterListSize < subListSize) { + return -1 + } + if (target.isEmpty() || source.isEmpty()) { + return -1 + } + for (int i = startIndex; i < masterListSize; i++) { + // starting at each index look for the sub list + if (i + subListSize > masterListSize) { + return -1 + } + + boolean matches = true + for (int j = 0; j < subListSize; j++) { + T sub = target.get(j) + T m = source.get(i + j) + if (!eq(sub, m)) { + matches = false + break + } + } + if (matches) { + return i + } + } + return -1 + } + + private static boolean eq(T t1, T t2) { + if (t1 == null && t2 == null) { + return true + } + if (t1 != null && t2 != null) { + return t1 == t2 + } + return false + } + + + static T last(List list) { + return list.get(list.size()-1) + } + static List> getIncrementalResults(IncrementalExecutionResult initialResult) { Publisher deferredResultStream = initialResult.incrementalItemPublisher diff --git a/src/test/groovy/graphql/TestUtilTest.groovy b/src/test/groovy/graphql/TestUtilTest.groovy new file mode 100644 index 0000000000..faf63d66aa --- /dev/null +++ b/src/test/groovy/graphql/TestUtilTest.groovy @@ -0,0 +1,104 @@ +package graphql + +import spock.lang.Specification + +class TestUtilTest extends Specification { + + def "list contains in order"() { + given: + def masterList = ["a", "b", "c", "d", "e", "f", "g"] + + when: + def actual = TestUtil.listContainsInOrder(masterList, subList) + + then: + actual == expected + + where: + subList | expected + ["a"] | true + ["f"] | true + ["c", "d"] | true + ["f", "g"] | true + ["a", "b", "c", "d", "e", "f", "g"] | true + ["f", "g", "X"] | false + ["b", "c", "e"] | false + [] | false + ["a", "b", "c", "d", "e", "f", "g", "X"] | false + ["A", "B", "C"] | false + } + + def "list contains in order edge cases"() { + when: + def actual = TestUtil.listContainsInOrder([], []) + then: + !actual + + when: + actual = TestUtil.listContainsInOrder(["a"], []) + then: + !actual + + when: + actual = TestUtil.listContainsInOrder([], ["a"]) + then: + !actual + + when: + actual = TestUtil.listContainsInOrder([null, "a", null], [null, "a"]) + then: + actual + } + + def "list contains in order many lists"() { + def master = ["a", "b", "c", "d", "e", "f", "g"] + when: + def actual = TestUtil.listContainsInOrder(master, + ["b", "c"], ["e", "f"]) + then: + actual + + when: + actual = TestUtil.listContainsInOrder(master, + ["b", "c"], ["g"]) + then: + actual + + when: + actual = TestUtil.listContainsInOrder(master, + ["b", "c"], ["f", "g", "X"]) + then: + !actual + + when: + actual = TestUtil.listContainsInOrder(master, + ["a"], ["b"], ["c"], ["d"], ["e"], ["f"], ["g"]) + then: + actual + + when: + actual = TestUtil.listContainsInOrder(master, + ["a", "b"], ["c", "d"], ["e"], ["f"], ["g"]) + then: + actual + + when: + actual = TestUtil.listContainsInOrder(master, + ["a"], ["b"], ["c"], ["d"], ["e"], ["f"], ["g"], ["X"]) + then: + !actual + + when: "empty" + actual = TestUtil.listContainsInOrder(master, + ["a"], [], ["c"]) + then: + !actual + + when: + actual = TestUtil.listContainsInOrder(master, + ["a"], ["b"], ["c"], ["X"], ["e"], ["f"], ["g"]) + then: + !actual + + } +} diff --git a/src/test/groovy/graphql/execution/SubscriptionExecutionStrategyTest.groovy b/src/test/groovy/graphql/execution/SubscriptionExecutionStrategyTest.groovy index 5168593d59..af3dcaf5ce 100644 --- a/src/test/groovy/graphql/execution/SubscriptionExecutionStrategyTest.groovy +++ b/src/test/groovy/graphql/execution/SubscriptionExecutionStrategyTest.groovy @@ -12,6 +12,7 @@ import graphql.TypeMismatchError import graphql.execution.instrumentation.InstrumentationState import graphql.execution.instrumentation.LegacyTestingInstrumentation import graphql.execution.instrumentation.dataloader.DataLoaderDispatchingContextKeys +import graphql.execution.instrumentation.ModernTestingInstrumentation import graphql.execution.instrumentation.parameters.InstrumentationExecutionParameters import graphql.execution.pubsub.CapturingSubscriber import graphql.execution.pubsub.FlowMessagePublisher @@ -881,4 +882,73 @@ class SubscriptionExecutionStrategyTest extends Specification { where: exhaustedStrategy << [false, true] } + + + def "can instrument subscription reactive ending"() { + + given: + Object publisher = new ReactiveStreamsMessagePublisher(2) + + DataFetcher newMessageDF = new DataFetcher() { + @Override + Object get(DataFetchingEnvironment environment) { + return publisher + } + } + + def wiringBuilder = buildBaseSubscriptionWiring( + PropertyDataFetcher.fetching("sender"), PropertyDataFetcher.fetching("text") + ) + RuntimeWiring runtimeWiring = wiringBuilder + .type(newTypeWiring("Subscription").dataFetcher("newMessage", newMessageDF).build()) + .build() + + def instrumentation = new ModernTestingInstrumentation() + + def graphQL = TestUtil.graphQL(idl, runtimeWiring) + .instrumentation(instrumentation) + .subscriptionExecutionStrategy(new SubscriptionExecutionStrategy()).build() + + def executionInput = ExecutionInput.newExecutionInput().query(""" + subscription NewMessages { + newMessage(roomId: 123) { + sender + text + } + } + """).build() + + def executionResult = graphQL.execute(executionInput) + + when: + Publisher msgStream = executionResult.getData() + def capturingSubscriber = new CapturingSubscriber() + msgStream.subscribe(capturingSubscriber) + + then: + msgStream instanceof SubscriptionPublisher + Awaitility.await().untilTrue(capturingSubscriber.isDone()) + + TestUtil.listContainsInOrder(instrumentation.executionList, [ + "start:execution", + "start:parse", + "end:parse", + "start:validation", + "end:validation", + "start:execute-operation", + "start:execution-strategy", + "start:fetch-newMessage", + "end:fetch-newMessage", + "start:reactive-results-subscription", + "end:execution-strategy", + "end:execute-operation", + "end:execution", + ], [ + // followed by + "end:reactive-results-subscription" + ]) + + // last of all it finishes + TestUtil.last(instrumentation.executionList) == "end:reactive-results-subscription" + } } diff --git a/src/test/groovy/graphql/execution/instrumentation/InstrumentationTest.groovy b/src/test/groovy/graphql/execution/instrumentation/InstrumentationTest.groovy index 08ea4850c0..43767d9348 100644 --- a/src/test/groovy/graphql/execution/instrumentation/InstrumentationTest.groovy +++ b/src/test/groovy/graphql/execution/instrumentation/InstrumentationTest.groovy @@ -26,7 +26,6 @@ import java.util.concurrent.atomic.AtomicBoolean class InstrumentationTest extends Specification { - def 'Instrumentation of simple serial execution'() { given: @@ -341,7 +340,7 @@ class InstrumentationTest extends Specification { def instrumentation = new ModernTestingInstrumentation() { @Override InstrumentationContext beginExecution(InstrumentationExecutionParameters parameters, InstrumentationState state) { - executionList.add("start:execution") + this.executionList.add("start:execution") return null } } @@ -561,6 +560,7 @@ class InstrumentationTest extends Specification { "end:field-hero", "end:execution-strategy", "end:execute-operation", + "start:reactive-results-defer", "end:execution", // // the deferred field resolving now happens after the operation has come back @@ -572,6 +572,67 @@ class InstrumentationTest extends Specification { "end:complete-name", "end:field-name", "end:deferred-field-name", + + "end:reactive-results-defer", ] } + + def "can instrument defer reactive ending"() { + + given: + + def query = """ + { + hero { + id + ... @defer(label: "name") { + name + } + } + } + """ + + + when: + + def instrumentation = new ModernTestingInstrumentation() + + def graphQL = GraphQL + .newGraphQL(StarWarsSchema.starWarsSchema) + .queryExecutionStrategy(new AsyncExecutionStrategy()) + .instrumentation(instrumentation) + .build() + + def ei = ExecutionInput.newExecutionInput(query).graphQLContext { it -> + GraphQL.unusualConfiguration(it).incrementalSupport().enableIncrementalSupport(true) + }.build() + + IncrementalExecutionResult incrementalER = graphQL.execute(ei) as IncrementalExecutionResult + // + // cause the defer Publish to be finished + def results = TestUtil.getIncrementalResults(incrementalER) + then: + + TestUtil.listContainsInOrder(instrumentation.executionList, [ + "start:execution", + "start:parse", + "end:parse", + "start:validation", + "end:validation", + "start:execute-operation", + ], [ + // then it ends initial operation + "end:execution-strategy", + "end:execute-operation", + "start:reactive-results-defer", + "end:execution", + ], [ + // followed by + "end:reactive-results-defer" + ]) + + // last of all it finishes + TestUtil.last(instrumentation.executionList) == "end:reactive-results-defer" + } + } diff --git a/src/test/groovy/graphql/execution/instrumentation/ModernTestingInstrumentation.groovy b/src/test/groovy/graphql/execution/instrumentation/ModernTestingInstrumentation.groovy index 6cbb23941c..1053ffd0d7 100644 --- a/src/test/groovy/graphql/execution/instrumentation/ModernTestingInstrumentation.groovy +++ b/src/test/groovy/graphql/execution/instrumentation/ModernTestingInstrumentation.groovy @@ -10,6 +10,7 @@ import graphql.execution.instrumentation.parameters.InstrumentationExecutionStra import graphql.execution.instrumentation.parameters.InstrumentationFieldCompleteParameters import graphql.execution.instrumentation.parameters.InstrumentationFieldFetchParameters import graphql.execution.instrumentation.parameters.InstrumentationFieldParameters +import graphql.execution.instrumentation.parameters.InstrumentationReactiveResultsParameters import graphql.execution.instrumentation.parameters.InstrumentationValidationParameters import graphql.language.Document import graphql.schema.DataFetcher @@ -61,6 +62,13 @@ class ModernTestingInstrumentation implements Instrumentation { return new TestingExecutionStrategyInstrumentationContext("execution-strategy", executionList, throwableList, useOnDispatch) } + @Override + InstrumentationContext beginReactiveResults(InstrumentationReactiveResultsParameters parameters, InstrumentationState state) { + assert state == instrumentationState + def resultType = parameters.resultType.toString().toLowerCase() + return new TestingInstrumentContext("reactive-results-$resultType", executionList, throwableList, useOnDispatch) + } + @Override ExecuteObjectInstrumentationContext beginExecuteObject(InstrumentationExecutionStrategyParameters parameters, InstrumentationState state) { assert state == instrumentationState diff --git a/src/test/groovy/graphql/execution/reactive/ReactiveSupportTest.groovy b/src/test/groovy/graphql/execution/reactive/ReactiveSupportTest.groovy index a5197af4ed..145c6d4efe 100644 --- a/src/test/groovy/graphql/execution/reactive/ReactiveSupportTest.groovy +++ b/src/test/groovy/graphql/execution/reactive/ReactiveSupportTest.groovy @@ -5,6 +5,7 @@ import graphql.TestUtil import graphql.execution.pubsub.CapturingSubscriber import graphql.execution.pubsub.CountingFlux import graphql.schema.DataFetcher +import org.awaitility.Awaitility import reactor.adapter.JdkFlowAdapter import reactor.core.publisher.Mono import spock.lang.Specification @@ -219,4 +220,61 @@ class ReactiveSupportTest extends Specification { materialisedField: "materialised" ] } + + def "can be called back when there is a Publisher ends successfully or otherwise"() { + when: + def called = false + def throwable = null + + SingleSubscriberPublisher publisher = new SingleSubscriberPublisher<>() + def publisherFinishes = ReactiveSupport.whenPublisherFinishes(publisher, { t -> + throwable = t + called = true + }) + + + def capturingSubscriber = new CapturingSubscriber() + publisherFinishes.subscribe(capturingSubscriber) + + publisher.offer("a") + publisher.offer("b") + publisher.offer("c") + publisher.noMoreData() + + then: + + Awaitility.await().untilTrue(capturingSubscriber.isDone()) + + capturingSubscriber.events["a", "b", "c"] + + called + throwable == null + + when: "it has an error thrown" + + called = false + throwable = null + + publisher = new SingleSubscriberPublisher<>() + publisherFinishes = ReactiveSupport.whenPublisherFinishes(publisher, { t -> + throwable = t + called = true + }) + + + capturingSubscriber = new CapturingSubscriber() + publisherFinishes.subscribe(capturingSubscriber) + + publisher.offer("a") + publisher.offerError(new RuntimeException("BANG")) + + then: + + Awaitility.await().untilTrue(capturingSubscriber.isDone()) + + capturingSubscriber.events == ["a"] + + called + throwable.message == "BANG" + } }