From 1d97586ffdd09ffa028d30ba20e6b4fa954919e0 Mon Sep 17 00:00:00 2001 From: rstoyanchev Date: Thu, 6 Jun 2024 10:48:08 +0100 Subject: [PATCH] Apply ReactiveAdapterRegistry to return values CoroutinesUtils#invokeSuspendingFunction in adapts return values to Mono and Flux in Kotlin. However, Flow methods don't have to be suspending functions. This change ensures that ReactiveAdapterRegistry is applied wherever return values are handled. Closes gh-988 --- .../data/federation/EntityHandlerMethod.java | 14 +- .../AnnotatedControllerConfigurer.java | 13 +- .../support/BatchLoaderHandlerMethod.java | 40 +----- .../support/DataFetcherHandlerMethod.java | 21 +-- .../ContextDataFetcherDecorator.java | 29 ++-- .../ReactiveAdapterRegistryHelper.java | 135 ++++++++++++++++++ 6 files changed, 170 insertions(+), 82 deletions(-) create mode 100644 spring-graphql/src/main/java/org/springframework/graphql/execution/ReactiveAdapterRegistryHelper.java diff --git a/spring-graphql/src/main/java/org/springframework/graphql/data/federation/EntityHandlerMethod.java b/spring-graphql/src/main/java/org/springframework/graphql/data/federation/EntityHandlerMethod.java index 13771e96..2556bba9 100644 --- a/spring-graphql/src/main/java/org/springframework/graphql/data/federation/EntityHandlerMethod.java +++ b/spring-graphql/src/main/java/org/springframework/graphql/data/federation/EntityHandlerMethod.java @@ -18,7 +18,6 @@ package org.springframework.graphql.data.federation; import java.util.List; import java.util.Map; -import java.util.concurrent.CompletableFuture; import java.util.concurrent.Executor; import graphql.schema.DataFetchingEnvironment; @@ -26,6 +25,7 @@ import reactor.core.publisher.Mono; import org.springframework.graphql.data.method.HandlerMethodArgumentResolverComposite; import org.springframework.graphql.data.method.annotation.support.DataFetcherHandlerMethodSupport; +import org.springframework.graphql.execution.ReactiveAdapterRegistryHelper; import org.springframework.lang.Nullable; /** @@ -80,18 +80,8 @@ final class EntityHandlerMethod extends DataFetcherHandlerMethodSupport { } private Mono doInvoke(DataFetchingEnvironment env, Object[] args) { - Object result = doInvoke(env.getGraphQlContext(), args); - - if (result instanceof Mono mono) { - return mono.cast(Object.class); - } - else if (result instanceof CompletableFuture future) { - return Mono.fromFuture(future); - } - else { - return Mono.justOrEmpty(result); - } + return ReactiveAdapterRegistryHelper.toMono(result); } } diff --git a/spring-graphql/src/main/java/org/springframework/graphql/data/method/annotation/support/AnnotatedControllerConfigurer.java b/spring-graphql/src/main/java/org/springframework/graphql/data/method/annotation/support/AnnotatedControllerConfigurer.java index 804c0a63..d5cbe0e8 100644 --- a/spring-graphql/src/main/java/org/springframework/graphql/data/method/annotation/support/AnnotatedControllerConfigurer.java +++ b/spring-graphql/src/main/java/org/springframework/graphql/data/method/annotation/support/AnnotatedControllerConfigurer.java @@ -78,6 +78,7 @@ import org.springframework.graphql.data.method.annotation.SchemaMapping; import org.springframework.graphql.data.pagination.CursorStrategy; import org.springframework.graphql.data.query.SortStrategy; import org.springframework.graphql.execution.BatchLoaderRegistry; +import org.springframework.graphql.execution.ReactiveAdapterRegistryHelper; import org.springframework.graphql.execution.RuntimeWiringConfigurer; import org.springframework.graphql.execution.SelfDescribingDataFetcher; import org.springframework.graphql.execution.SubscriptionPublisherException; @@ -528,18 +529,24 @@ public class AnnotatedControllerConfigurer } @SuppressWarnings({"unchecked", "ReactiveStreamsUnusedPublisher"}) + @Nullable private Object applyExceptionHandling( DataFetchingEnvironment env, DataFetcherHandlerMethod handlerMethod, Object result) { - if (this.subscription && result instanceof Publisher publisher) { - result = Flux.from(publisher).onErrorResume((ex) -> handleSubscriptionError(ex, env, handlerMethod)); + if (this.subscription) { + return ReactiveAdapterRegistryHelper.toSubscriptionFlux(result) + .onErrorResume((ex) -> handleSubscriptionError(ex, env, handlerMethod)); } - else if (result instanceof Mono) { + + result = ReactiveAdapterRegistryHelper.toMonoOrFluxIfReactive(result); + + if (result instanceof Mono) { result = ((Mono) result).onErrorResume((ex) -> (Mono) handleException(ex, env, handlerMethod)); } else if (result instanceof Flux) { result = ((Flux) result).onErrorResume((ex) -> (Mono) handleException(ex, env, handlerMethod)); } + return result; } diff --git a/spring-graphql/src/main/java/org/springframework/graphql/data/method/annotation/support/BatchLoaderHandlerMethod.java b/spring-graphql/src/main/java/org/springframework/graphql/data/method/annotation/support/BatchLoaderHandlerMethod.java index 7de3f377..7e5610f9 100644 --- a/spring-graphql/src/main/java/org/springframework/graphql/data/method/annotation/support/BatchLoaderHandlerMethod.java +++ b/spring-graphql/src/main/java/org/springframework/graphql/data/method/annotation/support/BatchLoaderHandlerMethod.java @@ -21,9 +21,7 @@ import java.util.Arrays; import java.util.Collection; import java.util.Map; import java.util.concurrent.Callable; -import java.util.concurrent.CompletableFuture; import java.util.concurrent.Executor; -import java.util.function.Function; import graphql.GraphQLContext; import org.dataloader.BatchLoaderEnvironment; @@ -37,6 +35,7 @@ import org.springframework.core.ParameterNameDiscoverer; import org.springframework.graphql.data.method.HandlerMethod; import org.springframework.graphql.data.method.InvocableHandlerMethodSupport; import org.springframework.graphql.data.method.annotation.ContextValue; +import org.springframework.graphql.execution.ReactiveAdapterRegistryHelper; import org.springframework.lang.Nullable; import org.springframework.util.Assert; import org.springframework.util.ClassUtils; @@ -97,11 +96,11 @@ public class BatchLoaderHandlerMethod extends InvocableHandlerMethodSupport { Object[] args = getMethodArgumentValues(keys, environment); if (doesNotHaveAsyncArgs(args)) { Object result = doInvoke(environment.getContext(), args); - return toMonoMap(result); + return ReactiveAdapterRegistryHelper.toMono(result); } return toArgsMono(args).flatMap((argValues) -> { Object result = doInvoke(environment.getContext(), argValues); - return toMonoMap(result); + return ReactiveAdapterRegistryHelper.toMono(result); }); } @@ -117,11 +116,11 @@ public class BatchLoaderHandlerMethod extends InvocableHandlerMethodSupport { Object[] args = getMethodArgumentValues(keys, environment); if (doesNotHaveAsyncArgs(args)) { Object result = doInvoke(environment.getContext(), args); - return toFlux(result); + return ReactiveAdapterRegistryHelper.toFluxFromCollection(result); } return toArgsMono(args).flatMapMany((resolvedArgs) -> { Object result = doInvoke(environment.getContext(), resolvedArgs); - return toFlux(result); + return ReactiveAdapterRegistryHelper.toFluxFromCollection(result); }); } @@ -185,33 +184,4 @@ public class BatchLoaderHandlerMethod extends InvocableHandlerMethodSupport { return Arrays.stream(args).noneMatch((arg) -> arg instanceof Mono); } - @SuppressWarnings("unchecked") - private static Mono> toMonoMap(@Nullable Object result) { - if (result instanceof Map) { - return Mono.just((Map) result); - } - else if (result instanceof Mono) { - return (Mono>) result; - } - else if (result instanceof CompletableFuture) { - return Mono.fromFuture((CompletableFuture>) result); - } - return Mono.error(new IllegalStateException("Unexpected return value: " + result)); - } - - @SuppressWarnings("unchecked") - private static Flux toFlux(@Nullable Object result) { - if (result instanceof Collection) { - return Flux.fromIterable((Collection) result); - } - else if (result instanceof Flux) { - return (Flux) result; - } - else if (result instanceof CompletableFuture) { - return Mono.fromFuture((CompletableFuture>) result) - .flatMapIterable(Function.identity()); - } - return Flux.error(new IllegalStateException("Unexpected return value: " + result)); - } - } diff --git a/spring-graphql/src/main/java/org/springframework/graphql/data/method/annotation/support/DataFetcherHandlerMethod.java b/spring-graphql/src/main/java/org/springframework/graphql/data/method/annotation/support/DataFetcherHandlerMethod.java index 1befabcc..ee830009 100644 --- a/spring-graphql/src/main/java/org/springframework/graphql/data/method/annotation/support/DataFetcherHandlerMethod.java +++ b/spring-graphql/src/main/java/org/springframework/graphql/data/method/annotation/support/DataFetcherHandlerMethod.java @@ -18,18 +18,16 @@ package org.springframework.graphql.data.method.annotation.support; import java.util.Arrays; import java.util.concurrent.Callable; -import java.util.concurrent.CompletableFuture; import java.util.concurrent.Executor; import java.util.function.BiConsumer; import graphql.schema.DataFetchingEnvironment; -import org.reactivestreams.Publisher; -import reactor.core.publisher.Flux; import reactor.core.publisher.Mono; import org.springframework.graphql.data.method.HandlerMethod; import org.springframework.graphql.data.method.HandlerMethodArgumentResolver; import org.springframework.graphql.data.method.HandlerMethodArgumentResolverComposite; +import org.springframework.graphql.execution.ReactiveAdapterRegistryHelper; import org.springframework.lang.Nullable; import org.springframework.util.Assert; @@ -112,6 +110,7 @@ public class DataFetcherHandlerMethod extends DataFetcherHandlerMethodSupport { * @param providedArgs additional arguments to be matched by their type * @since 1.2.0 */ + @SuppressWarnings("ReactiveStreamsUnusedPublisher") @Nullable public Object invoke(DataFetchingEnvironment environment, Object... providedArgs) { Object[] args; @@ -129,23 +128,11 @@ public class DataFetcherHandlerMethod extends DataFetcherHandlerMethodSupport { return this.subscription ? toArgsMono(args).flatMapMany((argValues) -> { Object result = validateAndInvoke(argValues, environment); - Assert.state(result instanceof Publisher, "Expected a Publisher from a Subscription response"); - return Flux.from((Publisher) result); + return ReactiveAdapterRegistryHelper.toSubscriptionFlux(result); }) : toArgsMono(args).flatMap((argValues) -> { Object result = validateAndInvoke(argValues, environment); - if (result instanceof Mono mono) { - return mono; - } - else if (result instanceof Flux flux) { - return Flux.from(flux).collectList(); - } - else if (result instanceof CompletableFuture future) { - return Mono.fromFuture(future); - } - else { - return Mono.justOrEmpty(result); - } + return ReactiveAdapterRegistryHelper.toMono(result); }); } diff --git a/spring-graphql/src/main/java/org/springframework/graphql/execution/ContextDataFetcherDecorator.java b/spring-graphql/src/main/java/org/springframework/graphql/execution/ContextDataFetcherDecorator.java index 8ad85750..2e5daa1b 100644 --- a/spring-graphql/src/main/java/org/springframework/graphql/execution/ContextDataFetcherDecorator.java +++ b/spring-graphql/src/main/java/org/springframework/graphql/execution/ContextDataFetcherDecorator.java @@ -73,6 +73,7 @@ final class ContextDataFetcherDecorator implements DataFetcher { } + @SuppressWarnings("ReactiveStreamsUnusedPublisher") @Override public Object get(DataFetchingEnvironment env) throws Exception { @@ -86,24 +87,22 @@ final class ContextDataFetcherDecorator implements DataFetcher { Object value = snapshot.wrap(() -> this.delegate.get(env)).call(); if (this.subscription) { - Assert.state(value instanceof Publisher, "Expected Publisher for a subscription"); - Flux flux = Flux.from((Publisher) value).onErrorResume((exception) -> { - // Already handled, e.g. controller methods? - if (exception instanceof SubscriptionPublisherException) { - return Mono.error(exception); - } - return this.subscriptionExceptionResolver.resolveException(exception) - .flatMap((errors) -> Mono.error(new SubscriptionPublisherException(errors, exception))); - }); - return flux.contextWrite(snapshot::updateContext); + return ReactiveAdapterRegistryHelper.toSubscriptionFlux(value) + .onErrorResume((exception) -> { + // Already handled, e.g. controller methods? + if (exception instanceof SubscriptionPublisherException) { + return Mono.error(exception); + } + return this.subscriptionExceptionResolver.resolveException(exception) + .flatMap((errors) -> Mono.error(new SubscriptionPublisherException(errors, exception))); + }) + .contextWrite(snapshot::updateContext); } - if (value instanceof Flux) { - value = ((Flux) value).collectList(); - } + value = ReactiveAdapterRegistryHelper.toMonoIfReactive(value); - if (value instanceof Mono valueMono) { - value = valueMono.contextWrite(snapshot::updateContext).toFuture(); + if (value instanceof Mono mono) { + value = mono.contextWrite(snapshot::updateContext).toFuture(); } return value; diff --git a/spring-graphql/src/main/java/org/springframework/graphql/execution/ReactiveAdapterRegistryHelper.java b/spring-graphql/src/main/java/org/springframework/graphql/execution/ReactiveAdapterRegistryHelper.java new file mode 100644 index 00000000..c3efb3c2 --- /dev/null +++ b/spring-graphql/src/main/java/org/springframework/graphql/execution/ReactiveAdapterRegistryHelper.java @@ -0,0 +1,135 @@ +/* + * Copyright 2002-2024 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 java.util.Collection; + +import io.micrometer.context.Nullable; +import org.reactivestreams.Publisher; +import reactor.core.publisher.Flux; +import reactor.core.publisher.Mono; + +import org.springframework.core.ReactiveAdapter; +import org.springframework.core.ReactiveAdapterRegistry; +import org.springframework.util.Assert; + +/** + * Helper to adapt a result Object to {@link Mono} or {@link Flux} through + * {@link ReactiveAdapterRegistry}. + * + * @author Rossen Stoyanchev + * @since 1.3.1 + */ +@SuppressWarnings({"ReactiveStreamsUnusedPublisher", "unchecked"}) +public abstract class ReactiveAdapterRegistryHelper { + + private static final ReactiveAdapterRegistry registry = ReactiveAdapterRegistry.getSharedInstance(); + + + /** + * Return a {@link Mono} for the given Object by delegating to + * {@link #toMonoIfReactive}, and then applying {@link Mono#justOrEmpty} + * if necessary. + * @param result the result Object to adapt + * @param the type of element in the Mono to cast to + * @return a {@code Mono} that represents the result + */ + public static Mono toMono(@Nullable Object result) { + result = toMonoIfReactive(result); + return (Mono) ((result instanceof Mono mono) ? mono : Mono.justOrEmpty(result)); + } + + /** + * Return a {@link Mono} for the given result Object if it can be adapted + * to a {@link Publisher} via {@link ReactiveAdapterRegistry}. Multivalued + * publishers are collected to a List. + * @param result the result Object to adapt + * @return the same instance or a {@code Mono} if the object is known to + * {@code ReactiveAdapterRegistry} + */ + @Nullable + public static Object toMonoIfReactive(@Nullable Object result) { + ReactiveAdapter adapter = ((result != null) ? registry.getAdapter(result.getClass()) : null); + if (adapter == null) { + return result; + } + Publisher publisher = adapter.toPublisher(result); + return (adapter.isMultiValue() ? Flux.from(publisher).collectList() : Mono.from(publisher)); + } + + /** + * Adapt the given result Object to {@link Mono} or {@link Flux} if it can + * be adapted to a single or multi-value {@link Publisher} respectively + * via {@link ReactiveAdapterRegistry}. + * @param result the result Object to adapt + * @return the same instance, a {@code Mono}, or a {@code Flux} + */ + @Nullable + public static Object toMonoOrFluxIfReactive(@Nullable Object result) { + ReactiveAdapter adapter = ((result != null) ? registry.getAdapter(result.getClass()) : null); + if (adapter == null) { + return result; + } + Publisher publisher = adapter.toPublisher(result); + return (adapter.isMultiValue() ? Flux.from(publisher) : Mono.from(publisher)); + } + + /** + * Return a {@link Flux} for the given result Object, adapting to a + * {@link Publisher} first if necessary via {@link ReactiveAdapterRegistry}. + * @param result the result Object to adapt + * @return a {@link Flux}, possibly empty if the result is {@code null} + */ + public static Flux toSubscriptionFlux(@Nullable Object result) { + if (result == null) { + return Flux.empty(); + } + if (result instanceof Publisher publisher) { + return Flux.from(publisher); + } + ReactiveAdapter adapter = registry.getAdapter(result.getClass()); + Assert.state(adapter != null, "Expected Publisher for a subscription"); + return Flux.from(adapter.toPublisher(result)); + } + + /** + * Return a {@link Flux} for the given result Object that represents a + * logical collection of values. The Object must be a {@link Collection} + * or a publisher of a {@code Collection}, which is flattened with + * {@link Flux#fromIterable(Iterable)}, or a multi-value publisher. + * @param result the result Object to adapt + * @param the type of element in the collection to cast to + * @return a {@code Flux} that represents the collection + */ + public static Flux toFluxFromCollection(@Nullable Object result) { + if (result instanceof Collection) { + return Flux.fromIterable((Collection) result); + } + ReactiveAdapter adapter = ((result != null) ? registry.getAdapter(result.getClass()) : null); + if (adapter == null) { + return Flux.error(new IllegalStateException("Unexpected return value: " + result)); + } + Publisher publisher = adapter.toPublisher(result); + if (adapter.isMultiValue()) { + return (Flux) Flux.from(publisher); + } + else { + return Mono.from(publisher).flatMapMany((c) -> Flux.fromIterable((Collection) c)); + } + } + +}