X Tutup
Skip to content

Added support for Flow publishers coming back from the subscription field data fetcher #3910

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
bbakerman merged 1 commit into from
Jul 1, 2025
Merged

Added support for Flow publishers coming back from the subscription field data fetcher #3910

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
39 changes: 29 additions & 10 deletions src/main/java/graphql/execution/SubscriptionExecutionStrategy.java
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,5 +1,6 @@
package graphql.execution;

import graphql.Assert;
import graphql.ExecutionResult;
import graphql.ExecutionResultImpl;
import graphql.GraphQLContext;
Expand All @@ -14,13 +15,14 @@
import graphql.language.Field;
import graphql.schema.GraphQLFieldDefinition;
import graphql.schema.GraphQLObjectType;
import org.reactivestreams.FlowAdapters;
import org.reactivestreams.Publisher;

import java.util.concurrent.CompletableFuture;
import java.util.concurrent.CompletionStage;
import java.util.concurrent.Flow;
import java.util.function.Function;

import static graphql.Assert.assertTrue;
import static graphql.execution.instrumentation.SimpleInstrumentationContext.nonNullCtx;
import static java.util.Collections.singletonMap;

Expand Down Expand Up @@ -70,14 +72,12 @@ public CompletableFuture<ExecutionResult> execute(ExecutionContext executionCont
CompletableFuture<ExecutionResult> overallResult = sourceEventStream.thenApply((publisher) ->
{
if (publisher == null) {
ExecutionResultImpl executionResult = new ExecutionResultImpl(null, executionContext.getErrors());
return executionResult;
return new ExecutionResultImpl(null, executionContext.getErrors());
}
Function<Object, CompletionStage<ExecutionResult>> mapperFunction = eventPayload -> executeSubscriptionEvent(executionContext, parameters, eventPayload);
boolean keepOrdered = keepOrdered(executionContext.getGraphQLContext());
SubscriptionPublisher mapSourceToResponse = new SubscriptionPublisher(publisher, mapperFunction, keepOrdered);
ExecutionResultImpl executionResult = new ExecutionResultImpl(mapSourceToResponse, executionContext.getErrors());
return executionResult;
return new ExecutionResultImpl(mapSourceToResponse, executionContext.getErrors());
Copy link
Member Author

@bbakerman bbakerman Apr 10, 2025

Choose a reason for hiding this comment

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

just redundant variables cleanup

});

// dispatched the subscription query
Expand Down Expand Up @@ -111,14 +111,33 @@ private CompletableFuture<Publisher<Object>> createSourceEventStream(ExecutionCo
CompletableFuture<FetchedValue> fieldFetched = Async.toCompletableFuture(fetchField(executionContext, newParameters));
return fieldFetched.thenApply(fetchedValue -> {
Object publisher = fetchedValue.getFetchedValue();
if (publisher != null) {
assertTrue(publisher instanceof Publisher, () -> "Your data fetcher must return a Publisher of events when using graphql subscriptions");
}
//noinspection unchecked,DataFlowIssue
return (Publisher<Object>) publisher;
return mkReactivePublisher(publisher);
});
}

/**
* The user code can return either a reactive stream {@link Publisher} or a JDK {@link Flow.Publisher}
* and we adapt it to a reactive streams one since we use reactive streams in our implementation.
*
* @param publisherObj - the object returned from the data fetcher as the source of events
*
* @return a reactive streams {@link Publisher} always
*/
@SuppressWarnings("unchecked")
private static Publisher<Object> mkReactivePublisher(Object publisherObj) {
if (publisherObj != null) {
if (publisherObj instanceof Publisher) {
return (Publisher<Object>) publisherObj;
} else if (publisherObj instanceof Flow.Publisher) {
Flow.Publisher<Object> flowPublisher = (Flow.Publisher<Object>) publisherObj;
return FlowAdapters.toPublisher(flowPublisher);
} else {
return Assert.assertShouldNeverHappen("Your data fetcher must return a Publisher of events when using graphql subscriptions");
}
}
return null; // null is valid - we return null data in this case
}

/*
ExecuteSubscriptionEvent(subscription, schema, variableValues, initialValue):

Expand Down
39 changes: 35 additions & 4 deletions src/test/groovy/graphql/execution/SubscriptionExecutionStrategyTest.groovy
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,5 +1,6 @@
package graphql.execution

import graphql.AssertException
import graphql.ErrorType
import graphql.ExecutionInput
import graphql.ExecutionResult
Expand All @@ -12,6 +13,7 @@ import graphql.execution.instrumentation.InstrumentationState
import graphql.execution.instrumentation.LegacyTestingInstrumentation
import graphql.execution.instrumentation.parameters.InstrumentationExecutionParameters
import graphql.execution.pubsub.CapturingSubscriber
import graphql.execution.pubsub.FlowMessagePublisher
import graphql.execution.pubsub.Message
import graphql.execution.pubsub.ReactiveStreamsMessagePublisher
import graphql.execution.pubsub.ReactiveStreamsObjectPublisher
Expand Down Expand Up @@ -138,7 +140,7 @@ class SubscriptionExecutionStrategyTest extends Specification {
def "subscription query sends out a stream of events using the '#why' implementation"() {

given:
Publisher<Object> publisher = eventStreamPublisher
Object publisher = eventStreamPublisher

DataFetcher newMessageDF = new DataFetcher() {
@Override
Expand Down Expand Up @@ -181,14 +183,15 @@ class SubscriptionExecutionStrategyTest extends Specification {
why | eventStreamPublisher
'reactive streams stream' | new ReactiveStreamsMessagePublisher(10)
'rxjava stream' | new RxJavaMessagePublisher(10)
'flow stream' | new FlowMessagePublisher(10)

}

@Unroll
def "subscription alias is correctly used in response messages using '#why' implementation"() {

given:
Publisher<Object> publisher = eventStreamPublisher
Object publisher = eventStreamPublisher

DataFetcher newMessageDF = new DataFetcher() {
@Override
Expand Down Expand Up @@ -227,6 +230,7 @@ class SubscriptionExecutionStrategyTest extends Specification {
why | eventStreamPublisher
'reactive streams stream' | new ReactiveStreamsMessagePublisher(1)
'rxjava stream' | new RxJavaMessagePublisher(1)
'flow stream' | new FlowMessagePublisher(1)
}


Expand All @@ -238,7 +242,7 @@ class SubscriptionExecutionStrategyTest extends Specification {
// capability and it costs us little to support it, lets have a test for it.
//
given:
Publisher<Object> publisher = eventStreamPublisher
Object publisher = eventStreamPublisher

DataFetcher newMessageDF = new DataFetcher() {
@Override
Expand Down Expand Up @@ -279,7 +283,7 @@ class SubscriptionExecutionStrategyTest extends Specification {
why | eventStreamPublisher
'reactive streams stream' | new ReactiveStreamsMessagePublisher(10)
'rxjava stream' | new RxJavaMessagePublisher(10)

'flow stream' | new FlowMessagePublisher(10)
}


Expand Down Expand Up @@ -312,6 +316,33 @@ class SubscriptionExecutionStrategyTest extends Specification {
executionResult.errors.size() == 1
}

def "if you dont return a Publisher we will assert"() {

DataFetcher newMessageDF = new DataFetcher() {
@Override
Object get(DataFetchingEnvironment environment) {
return "Not a Publisher"
}
}

GraphQL graphQL = buildSubscriptionQL(newMessageDF)

def executionInput = ExecutionInput.newExecutionInput().query("""
subscription NewMessages {
newMessage(roomId: 123) {
sender
text
}
}
""").build()

when:
graphQL.execute(executionInput)

then:
thrown(AssertException)
}

def "subscription query will surface event stream exceptions"() {

DataFetcher newMessageDF = new DataFetcher() {
Expand Down
44 changes: 44 additions & 0 deletions src/test/groovy/graphql/execution/pubsub/CommonMessagePublisher.java
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,44 @@
package graphql.execution.pubsub;

import org.reactivestreams.example.unicast.AsyncIterablePublisher;

import java.util.Iterator;
import java.util.concurrent.ForkJoinPool;
import java.util.function.Function;

class CommonMessagePublisher {

protected final AsyncIterablePublisher<Message> iterablePublisher;

protected CommonMessagePublisher(final int count) {
Iterable<Message> iterable = mkIterable(count, at -> {
Message message = new Message("sender" + at, "text" + at);
return examineMessage(message, at);
});
iterablePublisher = new AsyncIterablePublisher<>(iterable, ForkJoinPool.commonPool());
}

@SuppressWarnings("unused")
protected Message examineMessage(Message message, Integer at) {
return message;
}

private static Iterable<Message> mkIterable(int count, Function<Integer, Message> msgMaker) {
return () -> new Iterator<>() {
private int at = 0;

@Override
public boolean hasNext() {
return at < count;
}

@Override
public Message next() {
Message message = msgMaker.apply(at);
at++;
return message;
}
};
}

}
22 changes: 22 additions & 0 deletions src/test/groovy/graphql/execution/pubsub/FlowMessagePublisher.java
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,22 @@
package graphql.execution.pubsub;

import org.reactivestreams.FlowAdapters;

import java.util.concurrent.Flow;

/**
* This example publisher will create count "messages" and then terminate. It
* uses the reactive streams TCK as its implementation but presents itself
* as a {@link Flow.Publisher}
*/
public class FlowMessagePublisher extends CommonMessagePublisher implements Flow.Publisher<Message> {

public FlowMessagePublisher(int count) {
super(count);
}

@Override
public void subscribe(Flow.Subscriber<? super Message> s) {
iterablePublisher.subscribe(FlowAdapters.toSubscriber(s));
}
}
40 changes: 3 additions & 37 deletions src/test/groovy/graphql/execution/pubsub/ReactiveStreamsMessagePublisher.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -2,54 +2,20 @@

import org.reactivestreams.Publisher;
import org.reactivestreams.Subscriber;
import org.reactivestreams.example.unicast.AsyncIterablePublisher;

import java.util.Iterator;
import java.util.concurrent.ForkJoinPool;
import java.util.function.Function;

/**
* This example publisher will create count "messages" and then terminate. It
* uses the reactive streams TCK as its implementation
*/
public class ReactiveStreamsMessagePublisher implements Publisher<Message> {

private final AsyncIterablePublisher<Message> iterablePublisher;
public class ReactiveStreamsMessagePublisher extends CommonMessagePublisher implements Publisher<Message> {

public ReactiveStreamsMessagePublisher(final int count) {
Iterable<Message> iterable = mkIterable(count, at -> {
Message message = new Message("sender" + at, "text" + at);
return examineMessage(message, at);
});
iterablePublisher = new AsyncIterablePublisher<>(iterable, ForkJoinPool.commonPool());
super(count);
}

@Override
public void subscribe(Subscriber<? super Message> s) {
iterablePublisher.subscribe(s);
}

@SuppressWarnings("unused")
protected Message examineMessage(Message message, Integer at) {
return message;
}

private static Iterable<Message> mkIterable(int count, Function<Integer, Message> msgMaker) {
return () -> new Iterator<Message>() {
private int at = 0;

@Override
public boolean hasNext() {
return at < count;
}

@Override
public Message next() {
Message message = msgMaker.apply(at);
at++;
return message;
}
};
}

}
Copy link
Member Author

@bbakerman bbakerman Apr 10, 2025

Choose a reason for hiding this comment

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

Moved this code into CommonMessagePublisher for re-use reasons


Loading
X Tutup