Added support for reactive Publishers to be returned from data fetchers #3731

bbakerman · 2024-10-26T08:14:24Z

This adds support for Reactive Publishers, both reactive streams Publishers and JDK Flow Publishers, to be returned from a DataFetcher as a value.

They will be turned into CompletableFuture by using a subscriber to ask for one value from them.

This is a lot like how reactor Mono has a .toFuture() method except now they can just return the Mono itself, since a Mono is a reactive streams Publisher.

The cost of this is 2 more instanceof checks like we already for for CompleteableFutures

bbakerman · 2024-10-26T08:15:46Z

src/main/java/graphql/execution/reactive/ReactiveSupport.java

+     * @param <T> for two
+     * @param <S> for subscription
+     */
+    private static abstract class PublisherToCompletableFuture<T, S> extends CompletableFuture<T> {


This class is heavily inspired from the Reactor reactor.core.publisher.MonoToCompletableFuture class. Its uses pretty much the same pattern to turn a Publisher into a CF

Note it "IS-A" CompletableFuture class

bbakerman · 2024-10-26T08:17:55Z

src/main/java/graphql/execution/ExecutionStrategy.java

@@ -197,8 +199,8 @@ public static String mkNameForPath(List<Field> currentField) {
     * @throws NonNullableFieldWasNullException in the {@link CompletableFuture} if a non-null field resolved to a null value
     */
    @SuppressWarnings("unchecked")
-    protected Object /* CompletableFuture<Map<String, Object>> | Map<String, Object> */
-    executeObject(ExecutionContext executionContext, ExecutionStrategyParameters parameters) throws NonNullableFieldWasNullException {
+    @DuckTyped(shape = "CompletableFuture<Map<String, Object>> | Map<String, Object>")


I think this annotation reads better in the IDEA rather than a comment. Maybe we can use it later in some way but honestly its just a more type safe way of saying the same as we try to say in the comment

Nice idea. Makes it more official than a comment

bbakerman · 2024-10-26T08:19:05Z

src/main/java/graphql/execution/reactive/ReactiveSupport.java

+public class ReactiveSupport {
+
+    @DuckTyped(shape = "CompletableFuture | Object")
+    public static Object fetchedObject(Object fetchedObject) {


Also ducked typed

…rs - review tweaks

bbakerman · 2024-10-26T08:25:41Z

src/test/groovy/graphql/execution/reactive/ReactiveSupportTest.groovy

+        val = ReactiveSupport.fetchedObject(cf)
+        then:
+        val === cf
+    }


wont touch CFs or plain old objects

bbakerman · 2024-10-26T08:27:02Z

src/test/groovy/graphql/execution/reactive/ReactiveSupportTest.groovy

+        !cf.isDone()
+
+        when:
+        def cfCancelled = cf.cancel(true)


cancelled before it produce a value

bbakerman · 2024-10-26T09:50:36Z

src/test/groovy/graphql/execution/pubsub/CountingFlux.groovy

+            return counter + 1;
+        })
+    }
+}


Just a test helper

bbakerman · 2024-10-26T09:51:10Z

src/test/groovy/graphql/execution/reactive/ReactiveSupportTest.groovy

+        reactiveObject         || _
+        Mono.just("X")         || _
+        toFlow(Mono.just("X")) || _
+    }


tests with both Reactor and Java Flows

bbakerman · 2024-10-26T09:52:52Z

src/test/groovy/graphql/execution/reactive/ReactiveSupportTest.groovy

+    }
+
+
+    def "can get a reactive Flux and only take one value and make a CF from it"() {


even a Flux of N values only has 1 value taken from it because that is what the graphql contact is - a Query field is 1 value not a stream of values - at least not yet (aka @stream future work perhaps)

bbakerman · 2024-10-26T09:54:03Z

src/test/groovy/graphql/execution/reactive/ReactiveSupportTest.groovy

+                cfField          : "cf",
+                materialisedField: "materialised"
+        ]
+    }


full integration tests showing it working.

We know CF error handling works already so there is no need for integration tests showing all that

bbakerman · 2024-10-26T09:54:37Z

src/main/java/graphql/execution/ExecutionStrategy.java

@@ -498,6 +500,8 @@ Async.CombinedBuilder<FieldValueInfo> getAsyncFieldValueInfo(
        executionContext.getDataLoaderDispatcherStrategy().fieldFetched(executionContext, parameters, dataFetcher, fetchedObject);
        fetchCtx.onDispatched();
        fetchCtx.onFetchedValue(fetchedObject);
+        // possible convert reactive objects into CompletableFutures
+        fetchedObject = ReactiveSupport.fetchedObject(fetchedObject);


This is the main chance. One liner. Turns a reactive publisher into a CF

…rs - never on subscriptions

…rs - never on subscriptions with tests working

bbakerman · 2024-10-27T06:21:06Z

src/main/java/graphql/execution/ExecutionContext.java

+     */
+    public boolean isSubscriptionOperation() {
+        return isOpType(OperationDefinition.Operation.SUBSCRIPTION);
+    }


I hav long wanted this. Checking operationDefinition.getOperation() is a PITA

bbakerman · 2024-10-27T06:21:53Z

src/main/java/graphql/execution/ExecutionStrategy.java

+        if (!executionContext.isSubscriptionOperation()) {
+            // possible convert reactive objects into CompletableFutures
+            fetchedObject = ReactiveSupport.fetchedObject(fetchedObject);
+        }


This is the main code - if its NOT a Subscription, we turns the Pubisher into a CF

bbakerman · 2024-10-27T22:07:08Z

src/main/java/graphql/DuckTyped.java

+@Target(value = {METHOD, PARAMETER})
+public @interface DuckTyped {
+    String shape();
+}


I think this is better than a comment

@DuckedTyped(shape="Object | CompletableFuture<Object>")

versus

/* Object | CompletableFuture<Object> */

bbakerman · 2024-10-27T22:08:57Z

src/main/java/graphql/execution/reactive/ReactiveSupport.java

+            return reactivePublisherToCF((Publisher<?>) fetchedObject);
+        }
+        return fetchedObject;
+    }


Leaves other things alone = only acts on Flow.Publisher and Pubisher

Added support for reactive Publishers to be returned from data fetchers

e514e7f

bbakerman added this to the 23.x breaking changes milestone Oct 26, 2024

bbakerman commented Oct 26, 2024

View reviewed changes

Added support for reactive Publishers to be returned from data fetche…

cf2038e

…rs - review tweaks

bbakerman commented Oct 26, 2024

View reviewed changes

src/test/groovy/graphql/execution/pubsub/CountingFlux.groovy

return counter + 1;

})

}

}

Copy link

Member Author

bbakerman Oct 26, 2024

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Just a test helper

bbakerman commented Oct 26, 2024

View reviewed changes

bbakerman added 2 commits October 27, 2024 12:31

Added support for reactive Publishers to be returned from data fetche…

484e534

…rs - never on subscriptions

Added support for reactive Publishers to be returned from data fetche…

a8cbd97

…rs - never on subscriptions with tests working

bbakerman commented Oct 27, 2024

View reviewed changes

bbakerman merged commit d854425 into master Nov 28, 2024
1 check passed

dondonz deleted the reactive-values-from-datafetchers branch November 30, 2024 07:09

bclozel mentioned this pull request Apr 3, 2025

Remove Publisher to CompletableFuture conversion spring-projects/spring-graphql#1170

Closed

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Added support for reactive Publishers to be returned from data fetchers #3731

Added support for reactive Publishers to be returned from data fetchers #3731

bbakerman commented Oct 26, 2024

bbakerman Oct 26, 2024

bbakerman Oct 26, 2024

bbakerman Oct 26, 2024

dondonz Nov 6, 2024

bbakerman Oct 26, 2024

bbakerman Oct 26, 2024

bbakerman Oct 26, 2024

bbakerman Oct 26, 2024

bbakerman Oct 26, 2024

bbakerman Oct 26, 2024

bbakerman Oct 26, 2024

bbakerman Oct 26, 2024

bbakerman Oct 26, 2024

bbakerman Oct 27, 2024

dondonz Nov 6, 2024

bbakerman Oct 27, 2024

bbakerman Oct 27, 2024

bbakerman Oct 27, 2024

		}


		def "can get a reactive Flux and only take one value and make a CF from it"() {

Added support for reactive Publishers to be returned from data fetchers #3731

Added support for reactive Publishers to be returned from data fetchers #3731

Conversation

bbakerman commented Oct 26, 2024

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment