Spring/spring-graphql
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Merge branch '1.3.x'

Browse Source
This commit is contained in:
Brian Clozel
2025-03-14 09:39:20 +01:00
parent cf1c99f489 a6c69eb7ac
commit a5ec819748
14 changed files with 202 additions and 200 deletions
Download Patch File Download Diff File Expand all files Collapse all files

8
spring-graphql-docs/modules/ROOT/pages/request-execution.adoc
View File

@@ -417,14 +417,6 @@ include-code::HttpTimeoutConfiguration[]
For more transport-specific timeouts, there are dedicated properties on the handler implementations like
`GraphQlWebSocketHandler` and `GraphQlSseHandler`.
NOTE: While reactive data fetchers are cancelled automatically, this cannot be done for others
as there is no consistent way to cancel processing. In this case, controller methods can get
the cancellation signal from a `Mono` in the GraphQL context and manually cancel work.
Here is an example of using the cancellation signal to abort processing inside a controller method:
include-code::TimeoutController[tag=cancel,indent=0]
[[execution.reactivedatafetcher]]
== Reactive `DataFetcher`

56
spring-graphql-docs/src/main/java/org/springframework/graphql/docs/execution/timeout/TimeoutController.java
View File

@@ -1,56 +0,0 @@
/*
* Copyright 2020-2025 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* https://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.graphql.docs.execution.timeout;
import java.util.concurrent.Future;
import graphql.GraphQLContext;
import reactor.core.publisher.Mono;
import org.springframework.graphql.ExecutionGraphQlRequest;
import org.springframework.graphql.data.method.annotation.Argument;
import org.springframework.graphql.data.method.annotation.QueryMapping;
import org.springframework.stereotype.Controller;
@Controller
public class TimeoutController {
BookCache bookCache = new BookCache();
// tag::cancel[]
@QueryMapping
public Book bookById(@Argument Long id, GraphQLContext context) throws Exception {
Mono<Void> cancel = context.get(ExecutionGraphQlRequest.CANCEL_PUBLISHER_CONTEXT_KEY);
Future<Book> bookFuture = this.bookCache.fetchBook(id);
cancel.doOnCancel(() -> bookFuture.cancel(true)).subscribe();
return bookFuture.get();
}
// end::cancel[]
record Book(String title, String author) {
}
class BookCache {
public Future<Book> fetchBook(Long id) {
return null;
}
}
}

6
spring-graphql/src/main/java/org/springframework/graphql/ExecutionGraphQlRequest.java
View File

@@ -36,12 +36,6 @@ import org.springframework.lang.Nullable;
*/
public interface ExecutionGraphQlRequest extends GraphQlRequest {
/**
* Key of the GraphQL context entry that holds a {@code Mono<Void>} that completes
* when the inbound GraphQL request is cancelled at the transport level.
*/
String CANCEL_PUBLISHER_CONTEXT_KEY = ExecutionGraphQlRequest.class.getName() + ".cancelled";
/**
* Return the transport assigned id for the request that in turn sets
* {@link ExecutionInput.Builder#executionId(ExecutionId) executionId}.

4
spring-graphql/src/main/java/org/springframework/graphql/data/method/InvocableHandlerMethodSupport.java
View File

@@ -32,7 +32,7 @@ import reactor.core.publisher.Mono;
import org.springframework.core.CoroutinesUtils;
import org.springframework.core.KotlinDetector;
import org.springframework.data.util.KotlinReflectionUtils;
import org.springframework.graphql.execution.ContextSnapshotFactoryHelper;
import org.springframework.graphql.execution.ContextPropagationHelper;
import org.springframework.lang.Nullable;
import org.springframework.util.Assert;
@@ -153,7 +153,7 @@ public abstract class InvocableHandlerMethodSupport extends HandlerMethod {
CompletableFuture<Object> future = new CompletableFuture<>();
this.executor.execute(() -> {
try {
ContextSnapshot snapshot = ContextSnapshotFactoryHelper.captureFrom(graphQLContext);
ContextSnapshot snapshot = ContextPropagationHelper.captureFrom(graphQLContext);
Object value = snapshot.wrap((Callable<?>) result).call();
future.complete(value);
}

22
spring-graphql/src/main/java/org/springframework/graphql/execution/ContextDataFetcherDecorator.java
View File

@@ -39,7 +39,6 @@ import org.reactivestreams.Publisher;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import org.springframework.graphql.ExecutionGraphQlRequest;
import org.springframework.lang.Nullable;
import org.springframework.util.Assert;
@@ -80,24 +79,22 @@ final class ContextDataFetcherDecorator implements DataFetcher<Object> {
public Object get(DataFetchingEnvironment env) throws Exception {
GraphQLContext graphQlContext = env.getGraphQlContext();
ContextSnapshotFactory snapshotFactory = ContextSnapshotFactoryHelper.getInstance(graphQlContext);
ContextSnapshotFactory snapshotFactory = ContextPropagationHelper.getInstance(graphQlContext);
ContextSnapshot snapshot = (env.getLocalContext() instanceof GraphQLContext localContext) ?
snapshotFactory.captureFrom(graphQlContext, localContext) :
snapshotFactory.captureFrom(graphQlContext);
Mono<Void> cancelledRequest = graphQlContext.get(ExecutionGraphQlRequest.CANCEL_PUBLISHER_CONTEXT_KEY);
Object value = snapshot.wrap(() -> this.delegate.get(env)).call();
if (value instanceof DataFetcherResult<?> dataFetcherResult) {
Object adapted = updateValue(dataFetcherResult.getData(), snapshot, cancelledRequest);
Object adapted = updateValue(dataFetcherResult.getData(), snapshot, graphQlContext);
value = DataFetcherResult.newResult()
.data(adapted)
.errors(dataFetcherResult.getErrors())
.localContext(dataFetcherResult.getLocalContext()).build();
}
else {
value = updateValue(value, snapshot, cancelledRequest);
value = updateValue(value, snapshot, graphQlContext);
}
return value;
@@ -105,7 +102,7 @@ final class ContextDataFetcherDecorator implements DataFetcher<Object> {
@SuppressWarnings("ReactiveStreamsUnusedPublisher")
private @Nullable Object updateValue(
@Nullable Object value, ContextSnapshot snapshot, @Nullable Mono<Void> cancelledRequest) {
@Nullable Object value, ContextSnapshot snapshot, GraphQLContext graphQlContext) {
if (value == null) {
return null;
@@ -121,19 +118,14 @@ final class ContextDataFetcherDecorator implements DataFetcher<Object> {
return this.subscriptionExceptionResolver.resolveException(exception)
.flatMap((errors) -> Mono.error(new SubscriptionPublisherException(errors, exception)));
});
if (cancelledRequest != null) {
subscriptionResult = subscriptionResult.takeUntilOther(cancelledRequest);
}
return subscriptionResult.contextWrite(snapshot::updateContext);
return ContextPropagationHelper.bindCancelFrom(subscriptionResult, graphQlContext)
.contextWrite(snapshot::updateContext);
}
value = ReactiveAdapterRegistryHelper.toMonoIfReactive(value);
if (value instanceof Mono<?> mono) {
if (cancelledRequest != null) {
mono = mono.takeUntilOther(cancelledRequest);
}
value = mono.contextWrite(snapshot::updateContext).toFuture();
value = ContextPropagationHelper.bindCancelFrom(mono, graphQlContext).contextWrite(snapshot::updateContext).toFuture();
}
return value;

170
spring-graphql/src/main/java/org/springframework/graphql/execution/ContextPropagationHelper.java Normal file
View File

@@ -0,0 +1,170 @@
/*
* Copyright 2020-2025 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* https://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.graphql.execution;
import graphql.GraphQLContext;
import io.micrometer.context.ContextSnapshot;
import io.micrometer.context.ContextSnapshotFactory;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import reactor.core.publisher.Sinks;
import reactor.util.context.Context;
import reactor.util.context.ContextView;
import org.springframework.lang.Nullable;
/**
* Helper for propagating context values from and to Reactor and GraphQL contexts.
*
* @author Rossen Stoyanchev
* @author Brian Clozel
* @since 1.3.5
*/
public abstract class ContextPropagationHelper {
private static final ContextSnapshotFactory sharedInstance = ContextSnapshotFactory.builder().build();
private static final String CONTEXT_SNAPSHOT_FACTORY_KEY = ContextPropagationHelper.class.getName() + ".KEY";
private static final String CANCEL_PUBLISHER_KEY = ContextPropagationHelper.class.getName() + ".cancelled";
/**
* Select a {@code ContextSnapshotFactory} instance to use, either the one
* passed in if it is not {@code null}, or a shared, static instance.
* @param factory the candidate factory instance to use if not {@code null}
* @return the instance to use
*/
public static ContextSnapshotFactory selectInstance(@Nullable ContextSnapshotFactory factory) {
if (factory != null) {
return factory;
}
return sharedInstance;
}
/**
* Save the {@code ContextSnapshotFactory} in the given {@link Context}.
* @param factory the instance to save
* @param context the context to save the instance to
* @return a new context with the saved instance
*/
public static Context saveInstance(ContextSnapshotFactory factory, Context context) {
return context.put(CONTEXT_SNAPSHOT_FACTORY_KEY, factory);
}
/**
* Save the {@code ContextSnapshotFactory} in the given {@link Context}.
* @param factory the instance to save
* @param context the context to save the instance to
*/
public static void saveInstance(ContextSnapshotFactory factory, GraphQLContext context) {
context.put(CONTEXT_SNAPSHOT_FACTORY_KEY, factory);
}
/**
* Access the {@code ContextSnapshotFactory} from the given {@link ContextView}
* or return a shared, static instance.
* @param contextView the context where the instance is saved
* @return the instance to use
*/
public static ContextSnapshotFactory getInstance(ContextView contextView) {
ContextSnapshotFactory factory = contextView.getOrDefault(CONTEXT_SNAPSHOT_FACTORY_KEY, null);
return selectInstance(factory);
}
/**
* Access the {@code ContextSnapshotFactory} from the given {@link GraphQLContext}
* or return a shared, static instance.
* @param context the context where the instance is saved
* @return the instance to use
*/
public static ContextSnapshotFactory getInstance(GraphQLContext context) {
ContextSnapshotFactory factory = context.get(CONTEXT_SNAPSHOT_FACTORY_KEY);
return selectInstance(factory);
}
/**
* Shortcut to obtain the {@code ContextSnapshotFactory} instance, and to
* capture from the given {@link ContextView}.
* @param contextView the context to capture from
* @return a snapshot from the capture
*/
public static ContextSnapshot captureFrom(ContextView contextView) {
ContextSnapshotFactory factory = getInstance(contextView);
return selectInstance(factory).captureFrom(contextView);
}
/**
* Shortcut to obtain the {@code ContextSnapshotFactory} instance, and to
* capture from the given {@link GraphQLContext}.
* @param context the context to capture from
* @return a snapshot from the capture
*/
public static ContextSnapshot captureFrom(GraphQLContext context) {
ContextSnapshotFactory factory = getInstance(context);
return selectInstance(factory).captureFrom(context);
}
/**
* Create a publisher and store it into the given {@link GraphQLContext}.
* This publisher can then be used to propagate cancel signals to upstream publishers.
* @param context the current GraphQL context
* @since 1.3.5
*/
public static Sinks.Empty<Void> createCancelPublisher(GraphQLContext context) {
Sinks.Empty<Void> requestCancelled = Sinks.empty();
context.put(CANCEL_PUBLISHER_KEY, requestCancelled.asMono());
return requestCancelled;
}
/**
* Bind the source {@link Mono} to the publisher from the given {@link GraphQLContext}.
* The returned {@code Mono} will be cancelled when this publisher completes.
* Subscribers must use the returned {@code Mono} instance.
* @param source the source {@code Mono}
* @param context the current GraphQL context
* @param <T> the type of published elements
* @return the new {@code Mono} that will be cancelled when notified
* @since 1.3.5
*/
public static <T> Mono<T> bindCancelFrom(Mono<T> source, GraphQLContext context) {
Mono<Void> cancelSignal = context.get(CANCEL_PUBLISHER_KEY);
if (cancelSignal != null) {
return source.takeUntilOther(cancelSignal);
}
return source;
}
/**
* Bind the source {@link Flux} to the publisher from the given {@link GraphQLContext}.
* The returned {@code Flux} will be cancelled when this publisher completes.
* Subscribers must use the returned {@code Mono} instance.
* @param source the source {@code Mono}
* @param context the current GraphQL context
* @param <T> the type of published elements
* @return the new {@code Mono} that will be cancelled when notified
* @since 1.3.5
*/
public static <T> Flux<T> bindCancelFrom(Flux<T> source, GraphQLContext context) {
Mono<Void> cancelSignal = context.get(CANCEL_PUBLISHER_KEY);
if (cancelSignal != null) {
return source.takeUntilOther(cancelSignal);
}
return source;
}
}

91
spring-graphql/src/main/java/org/springframework/graphql/execution/ContextSnapshotFactoryHelper.java
View File

@@ -16,13 +16,7 @@
package org.springframework.graphql.execution;
import graphql.GraphQLContext;
import io.micrometer.context.ContextSnapshot;
import io.micrometer.context.ContextSnapshotFactory;
import reactor.util.context.Context;
import reactor.util.context.ContextView;
import org.springframework.lang.Nullable;
/**
* Helper to use a single {@link ContextSnapshotFactory} instance by saving and
@@ -30,88 +24,9 @@ import org.springframework.lang.Nullable;
*
* @author Rossen Stoyanchev
* @since 1.3.0
* @deprecated since 1.3.5 in favor of {@link ContextPropagationHelper}.
*/
public abstract class ContextSnapshotFactoryHelper {
private static final ContextSnapshotFactory sharedInstance = ContextSnapshotFactory.builder().build();
private static final String CONTEXT_SNAPSHOT_FACTORY_KEY = ContextSnapshotFactoryHelper.class.getName() + ".KEY";
/**
* Select a {@code ContextSnapshotFactory} instance to use, either the one
* passed in if it is not {@code null}, or a shared, static instance.
* @param factory the candidate factory instance to use if not {@code null}
* @return the instance to use
*/
public static ContextSnapshotFactory selectInstance(@Nullable ContextSnapshotFactory factory) {
if (factory != null) {
return factory;
}
return sharedInstance;
}
/**
* Save the {@code ContextSnapshotFactory} in the given {@link Context}.
* @param factory the instance to save
* @param context the context to save the instance to
* @return a new context with the saved instance
*/
public static Context saveInstance(ContextSnapshotFactory factory, Context context) {
return context.put(CONTEXT_SNAPSHOT_FACTORY_KEY, factory);
}
/**
* Save the {@code ContextSnapshotFactory} in the given {@link Context}.
* @param factory the instance to save
* @param context the context to save the instance to
*/
public static void saveInstance(ContextSnapshotFactory factory, GraphQLContext context) {
context.put(CONTEXT_SNAPSHOT_FACTORY_KEY, factory);
}
/**
* Access the {@code ContextSnapshotFactory} from the given {@link ContextView}
* or return a shared, static instance.
* @param contextView the context where the instance is saved
* @return the instance to use
*/
public static ContextSnapshotFactory getInstance(ContextView contextView) {
ContextSnapshotFactory factory = contextView.getOrDefault(CONTEXT_SNAPSHOT_FACTORY_KEY, null);
return selectInstance(factory);
}
/**
* Access the {@code ContextSnapshotFactory} from the given {@link GraphQLContext}
* or return a shared, static instance.
* @param context the context where the instance is saved
* @return the instance to use
*/
public static ContextSnapshotFactory getInstance(GraphQLContext context) {
ContextSnapshotFactory factory = context.get(CONTEXT_SNAPSHOT_FACTORY_KEY);
return selectInstance(factory);
}
/**
* Shortcut to obtain the {@code ContextSnapshotFactory} instance, and to
* capture from the given {@link ContextView}.
* @param contextView the context to capture from
* @return a snapshot from the capture
*/
public static ContextSnapshot captureFrom(ContextView contextView) {
ContextSnapshotFactory factory = getInstance(contextView);
return selectInstance(factory).captureFrom(contextView);
}
/**
* Shortcut to obtain the {@code ContextSnapshotFactory} instance, and to
* capture from the given {@link GraphQLContext}.
* @param context the context to capture from
* @return a snapshot from the capture
*/
public static ContextSnapshot captureFrom(GraphQLContext context) {
ContextSnapshotFactory factory = getInstance(context);
return selectInstance(factory).captureFrom(context);
}
@Deprecated(since = "1.3.5", forRemoval = true)
public abstract class ContextSnapshotFactoryHelper extends ContextPropagationHelper {
}

2
spring-graphql/src/main/java/org/springframework/graphql/execution/DataFetcherExceptionResolverAdapter.java
View File

@@ -96,7 +96,7 @@ public abstract class DataFetcherExceptionResolverAdapter implements DataFetcher
private List<GraphQLError> resolveInternal(Throwable exception, DataFetchingEnvironment env) {
try {
return (this.threadLocalContextAware) ?
ContextSnapshotFactoryHelper.captureFrom(env.getGraphQlContext())
ContextPropagationHelper.captureFrom(env.getGraphQlContext())
.wrap(() -> resolveToMultipleErrors(exception, env))
.call() :
resolveToMultipleErrors(exception, env);

4
spring-graphql/src/main/java/org/springframework/graphql/execution/DefaultBatchLoaderRegistry.java
View File

@@ -233,7 +233,7 @@ public class DefaultBatchLoaderRegistry implements BatchLoaderRegistry {
@Override
public CompletionStage<List<V>> load(List<K> keys, BatchLoaderEnvironment environment) {
GraphQLContext graphQLContext = environment.getContext();
ContextSnapshot snapshot = ContextSnapshotFactoryHelper.captureFrom(graphQLContext);
ContextSnapshot snapshot = ContextPropagationHelper.captureFrom(graphQLContext);
try {
return snapshot.wrap(() ->
this.loader.apply(keys, environment)
@@ -281,7 +281,7 @@ public class DefaultBatchLoaderRegistry implements BatchLoaderRegistry {
@Override
public CompletionStage<Map<K, V>> load(Set<K> keys, BatchLoaderEnvironment environment) {
GraphQLContext graphQLContext = environment.getContext();
ContextSnapshot snapshot = ContextSnapshotFactoryHelper.captureFrom(graphQLContext);
ContextSnapshot snapshot = ContextPropagationHelper.captureFrom(graphQLContext);
try {
return snapshot.wrap(() ->
this.loader.apply(keys, environment)

9
spring-graphql/src/main/java/org/springframework/graphql/execution/DefaultExecutionGraphQlService.java
View File

@@ -87,20 +87,19 @@ public class DefaultExecutionGraphQlService implements ExecutionGraphQlService {
ExecutionInput executionInput = request.toExecutionInput();
ContextSnapshotFactory factory = ContextSnapshotFactoryHelper.getInstance(contextView);
ContextSnapshotFactory factory = ContextPropagationHelper.getInstance(contextView);
GraphQLContext graphQLContext = executionInput.getGraphQLContext();
ContextSnapshotFactoryHelper.saveInstance(factory, graphQLContext);
ContextPropagationHelper.saveInstance(factory, graphQLContext);
factory.captureFrom(contextView).updateContext(graphQLContext);
Sinks.Empty<Void> requestCancelled = Sinks.empty();
graphQLContext.put(ExecutionGraphQlRequest.CANCEL_PUBLISHER_CONTEXT_KEY, requestCancelled.asMono());
ExecutionInput executionInputToUse = registerDataLoaders(executionInput);
Sinks.Empty<Void> cancelPublisher = ContextPropagationHelper.createCancelPublisher(graphQLContext);
return Mono.fromFuture(this.graphQlSource.graphQl().executeAsync(executionInputToUse))
.onErrorResume((ex) -> ex instanceof GraphQLError, (ex) ->
Mono.just(ExecutionResult.newExecutionResult().addError((GraphQLError) ex).build()))
.map((result) -> new DefaultExecutionGraphQlResponse(executionInputToUse, result))
.doOnCancel(requestCancelled::tryEmitEmpty);
.doOnCancel(cancelPublisher::tryEmitEmpty);
});
}

2
spring-graphql/src/main/java/org/springframework/graphql/execution/ExceptionResolversExceptionHandler.java
View File

@@ -69,7 +69,7 @@ class ExceptionResolversExceptionHandler implements DataFetcherExceptionHandler
Throwable exception = unwrapException(handlerParameters);
DataFetchingEnvironment env = handlerParameters.getDataFetchingEnvironment();
ContextSnapshot snapshot = ContextSnapshotFactoryHelper.captureFrom(env.getGraphQlContext());
ContextSnapshot snapshot = ContextPropagationHelper.captureFrom(env.getGraphQlContext());
try {
return Flux.fromIterable(this.resolvers)
.flatMap((resolver) -> resolver.resolveException(exception, env))

2
spring-graphql/src/main/java/org/springframework/graphql/execution/SubscriptionExceptionResolverAdapter.java
View File

@@ -83,7 +83,7 @@ public abstract class SubscriptionExceptionResolverAdapter implements Subscripti
public final Mono<List<GraphQLError>> resolveException(Throwable exception) {
if (this.threadLocalContextAware) {
return Mono.deferContextual((contextView) -> {
ContextSnapshot snapshot = ContextSnapshotFactoryHelper.captureFrom(contextView);
ContextSnapshot snapshot = ContextPropagationHelper.captureFrom(contextView);
try {
List<GraphQLError> errors = snapshot.wrap(() -> resolveToMultipleErrors(exception)).call();
return Mono.justOrEmpty(errors);

6
spring-graphql/src/main/java/org/springframework/graphql/server/DefaultWebGraphQlHandlerBuilder.java
View File

@@ -25,7 +25,7 @@ import io.micrometer.context.ContextSnapshotFactory;
import reactor.core.publisher.Mono;
import org.springframework.graphql.ExecutionGraphQlService;
import org.springframework.graphql.execution.ContextSnapshotFactoryHelper;
import org.springframework.graphql.execution.ContextPropagationHelper;
import org.springframework.graphql.server.WebGraphQlInterceptor.Chain;
import org.springframework.lang.Nullable;
import org.springframework.util.Assert;
@@ -81,7 +81,7 @@ class DefaultWebGraphQlHandlerBuilder implements WebGraphQlHandler.Builder {
@Override
public WebGraphQlHandler build() {
ContextSnapshotFactory snapshotFactory = ContextSnapshotFactoryHelper.selectInstance(this.snapshotFactory);
ContextSnapshotFactory snapshotFactory = ContextPropagationHelper.selectInstance(this.snapshotFactory);
Chain endOfChain = (request) -> this.service.execute(request).map(WebGraphQlResponse::new);
@@ -107,7 +107,7 @@ class DefaultWebGraphQlHandlerBuilder implements WebGraphQlHandler.Builder {
public Mono<WebGraphQlResponse> handleRequest(WebGraphQlRequest request) {
ContextSnapshot snapshot = snapshotFactory.captureAll();
return executionChain.next(request).contextWrite((context) -> {
context = ContextSnapshotFactoryHelper.saveInstance(snapshotFactory, context);
context = ContextPropagationHelper.saveInstance(snapshotFactory, context);
return snapshot.updateContext(context);
});
}

20
spring-graphql/src/test/java/org/springframework/graphql/execution/ContextDataFetcherDecoratorTests.java
View File

@@ -19,7 +19,6 @@ package org.springframework.graphql.execution;
import java.time.Duration;
import java.util.Collections;
import java.util.List;
import java.util.Map;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.function.BiConsumer;
@@ -47,7 +46,6 @@ import reactor.core.publisher.Mono;
import reactor.core.publisher.Sinks;
import reactor.test.StepVerifier;
import org.springframework.graphql.ExecutionGraphQlRequest;
import org.springframework.graphql.GraphQlSetup;
import org.springframework.graphql.ResponseHelper;
import org.springframework.graphql.TestThreadLocalAccessor;
@@ -300,9 +298,8 @@ public class ContextDataFetcherDecoratorTests {
)
.toGraphQl();
Sinks.Empty<Void> requestCancelled = Sinks.empty();
ExecutionInput input = ExecutionInput.newExecutionInput().query("{ greeting }")
.graphQLContext(Map.of(ExecutionGraphQlRequest.CANCEL_PUBLISHER_CONTEXT_KEY, requestCancelled.asMono())).build();
ExecutionInput input = ExecutionInput.newExecutionInput().query("{ greeting }").build();
Sinks.Empty<Void> requestCancelled = ContextPropagationHelper.createCancelPublisher(input.getGraphQLContext());
CompletableFuture<ExecutionResult> asyncResult = graphQl.executeAsync(input);
requestCancelled.tryEmitEmpty();
@@ -320,9 +317,8 @@ public class ContextDataFetcherDecoratorTests {
)
.toGraphQl();
Sinks.Empty<Void> requestCancelled = Sinks.empty();
ExecutionInput input = ExecutionInput.newExecutionInput().query("{ greeting }")
.graphQLContext(Map.of(ExecutionGraphQlRequest.CANCEL_PUBLISHER_CONTEXT_KEY, requestCancelled.asMono())).build();
ExecutionInput input = ExecutionInput.newExecutionInput().query("{ greeting }").build();
Sinks.Empty<Void> requestCancelled = ContextPropagationHelper.createCancelPublisher(input.getGraphQLContext());
CompletableFuture<ExecutionResult> asyncResult = graphQl.executeAsync(input);
requestCancelled.tryEmitEmpty();
@@ -339,14 +335,14 @@ public class ContextDataFetcherDecoratorTests {
.doOnCancel(() -> dataFetcherCancelled.set(true))
)
.toGraphQl();
Sinks.Empty<Void> requestCancelled = Sinks.empty();
ExecutionInput input = ExecutionInput.newExecutionInput().query("subscription { greetings }")
.graphQLContext(Map.of(ExecutionGraphQlRequest.CANCEL_PUBLISHER_CONTEXT_KEY, requestCancelled.asMono())).build();
ExecutionInput input = ExecutionInput.newExecutionInput().query("subscription { greetings }").build();
Sinks.Empty<Void> requestCancelled = ContextPropagationHelper.createCancelPublisher(input.getGraphQLContext());
ExecutionResult executionResult = graphQl.executeAsync(input).get();
ResponseHelper.forSubscription(executionResult).subscribe();
requestCancelled.tryEmitEmpty();
await().atMost(Duration.ofSeconds(2)).until(dataFetcherCancelled::get);
assertThat(dataFetcherCancelled).isTrue();
}