Skip to content

Commit 338ffdd

Browse files
bbakermanbbakerman
authored
Merge pull request #3910 from graphql-java/support-flow-in-subscriptions
Added support for Flow publishers coming back from the subscription field data fetcher
2 parents 64f3a9d + 93a66bb commit 338ffdd

5 files changed

Lines changed: 133 additions & 51 deletions

File tree

‎src/main/java/graphql/execution/SubscriptionExecutionStrategy.java‎

Lines changed: 29 additions & 10 deletions
Original file line numberDiff line numberDiff line change
@@ -1,5 +1,6 @@
11
package graphql.execution;
22

3+
import graphql.Assert;
34
import graphql.ExecutionResult;
45
import graphql.ExecutionResultImpl;
56
import graphql.GraphQLContext;
@@ -15,13 +16,14 @@
1516
import graphql.language.Field;
1617
import graphql.schema.GraphQLFieldDefinition;
1718
import graphql.schema.GraphQLObjectType;
19+
import org.reactivestreams.FlowAdapters;
1820
import org.reactivestreams.Publisher;
1921

2022
import java.util.concurrent.CompletableFuture;
2123
import java.util.concurrent.CompletionStage;
24+
import java.util.concurrent.Flow;
2225
import java.util.function.Function;
2326

24-
import static graphql.Assert.assertTrue;
2527
import static graphql.execution.instrumentation.SimpleInstrumentationContext.nonNullCtx;
2628
import static java.util.Collections.singletonMap;
2729

@@ -71,14 +73,12 @@ public CompletableFuture<ExecutionResult> execute(ExecutionContext executionCont
7173
CompletableFuture<ExecutionResult> overallResult = sourceEventStream.thenApply((publisher) ->
7274
{
7375
if (publisher == null) {
74-
ExecutionResultImpl executionResult = new ExecutionResultImpl(null, executionContext.getErrors());
75-
return executionResult;
76+
return new ExecutionResultImpl(null, executionContext.getErrors());
7677
}
7778
Function<Object, CompletionStage<ExecutionResult>> mapperFunction = eventPayload -> executeSubscriptionEvent(executionContext, parameters, eventPayload);
7879
boolean keepOrdered = keepOrdered(executionContext.getGraphQLContext());
7980
SubscriptionPublisher mapSourceToResponse = new SubscriptionPublisher(publisher, mapperFunction, keepOrdered);
80-
ExecutionResultImpl executionResult = new ExecutionResultImpl(mapSourceToResponse, executionContext.getErrors());
81-
return executionResult;
81+
return new ExecutionResultImpl(mapSourceToResponse, executionContext.getErrors());
8282
});
8383

8484
// dispatched the subscription query
@@ -112,14 +112,33 @@ private CompletableFuture<Publisher<Object>> createSourceEventStream(ExecutionCo
112112
CompletableFuture<FetchedValue> fieldFetched = Async.toCompletableFuture(fetchField(executionContext, newParameters));
113113
return fieldFetched.thenApply(fetchedValue -> {
114114
Object publisher = fetchedValue.getFetchedValue();
115-
if (publisher != null) {
116-
assertTrue(publisher instanceof Publisher, () -> "Your data fetcher must return a Publisher of events when using graphql subscriptions");
117-
}
118-
//noinspection unchecked,DataFlowIssue
119-
return (Publisher<Object>) publisher;
115+
return mkReactivePublisher(publisher);
120116
});
121117
}
122118

119+
/**
120+
* The user code can return either a reactive stream {@link Publisher} or a JDK {@link Flow.Publisher}
121+
* and we adapt it to a reactive streams one since we use reactive streams in our implementation.
122+
*
123+
* @param publisherObj - the object returned from the data fetcher as the source of events
124+
*
125+
* @return a reactive streams {@link Publisher} always
126+
*/
127+
@SuppressWarnings("unchecked")
128+
private static Publisher<Object> mkReactivePublisher(Object publisherObj) {
129+
if (publisherObj != null) {
130+
if (publisherObj instanceof Publisher) {
131+
return (Publisher<Object>) publisherObj;
132+
} else if (publisherObj instanceof Flow.Publisher) {
133+
Flow.Publisher<Object> flowPublisher = (Flow.Publisher<Object>) publisherObj;
134+
return FlowAdapters.toPublisher(flowPublisher);
135+
} else {
136+
return Assert.assertShouldNeverHappen("Your data fetcher must return a Publisher of events when using graphql subscriptions");
137+
}
138+
}
139+
return null; // null is valid - we return null data in this case
140+
}
141+
123142
/*
124143
ExecuteSubscriptionEvent(subscription, schema, variableValues, initialValue):
125144

‎src/test/groovy/graphql/execution/SubscriptionExecutionStrategyTest.groovy‎

Lines changed: 35 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -1,5 +1,6 @@
11
package graphql.execution
22

3+
import graphql.AssertException
34
import graphql.ErrorType
45
import graphql.ExecutionInput
56
import graphql.ExecutionResult
@@ -12,6 +13,7 @@ import graphql.execution.instrumentation.InstrumentationState
1213
import graphql.execution.instrumentation.LegacyTestingInstrumentation
1314
import graphql.execution.instrumentation.parameters.InstrumentationExecutionParameters
1415
import graphql.execution.pubsub.CapturingSubscriber
16+
import graphql.execution.pubsub.FlowMessagePublisher
1517
import graphql.execution.pubsub.Message
1618
import graphql.execution.pubsub.ReactiveStreamsMessagePublisher
1719
import graphql.execution.pubsub.ReactiveStreamsObjectPublisher
@@ -142,7 +144,7 @@ class SubscriptionExecutionStrategyTest extends Specification {
142144
def "subscription query sends out a stream of events using the '#why' implementation"() {
143145
144146
given:
145-
Publisher<Object> publisher = eventStreamPublisher
147+
Object publisher = eventStreamPublisher
146148
147149
DataFetcher newMessageDF = new DataFetcher() {
148150
@Override
@@ -185,14 +187,15 @@ class SubscriptionExecutionStrategyTest extends Specification {
185187
why | eventStreamPublisher
186188
'reactive streams stream' | new ReactiveStreamsMessagePublisher(10)
187189
'rxjava stream' | new RxJavaMessagePublisher(10)
190+
'flow stream' | new FlowMessagePublisher(10)
188191
189192
}
190193
191194
@Unroll
192195
def "subscription alias is correctly used in response messages using '#why' implementation"() {
193196
194197
given:
195-
Publisher<Object> publisher = eventStreamPublisher
198+
Object publisher = eventStreamPublisher
196199
197200
DataFetcher newMessageDF = new DataFetcher() {
198201
@Override
@@ -231,6 +234,7 @@ class SubscriptionExecutionStrategyTest extends Specification {
231234
why | eventStreamPublisher
232235
'reactive streams stream' | new ReactiveStreamsMessagePublisher(1)
233236
'rxjava stream' | new RxJavaMessagePublisher(1)
237+
'flow stream' | new FlowMessagePublisher(1)
234238
}
235239
236240
@@ -242,7 +246,7 @@ class SubscriptionExecutionStrategyTest extends Specification {
242246
// capability and it costs us little to support it, lets have a test for it.
243247
//
244248
given:
245-
Publisher<Object> publisher = eventStreamPublisher
249+
Object publisher = eventStreamPublisher
246250
247251
DataFetcher newMessageDF = new DataFetcher() {
248252
@Override
@@ -283,7 +287,7 @@ class SubscriptionExecutionStrategyTest extends Specification {
283287
why | eventStreamPublisher
284288
'reactive streams stream' | new ReactiveStreamsMessagePublisher(10)
285289
'rxjava stream' | new RxJavaMessagePublisher(10)
286-
290+
'flow stream' | new FlowMessagePublisher(10)
287291
}
288292
289293
@@ -316,6 +320,33 @@ class SubscriptionExecutionStrategyTest extends Specification {
316320
executionResult.errors.size() == 1
317321
}
318322
323+
def "if you dont return a Publisher we will assert"() {
324+
325+
DataFetcher newMessageDF = new DataFetcher() {
326+
@Override
327+
Object get(DataFetchingEnvironment environment) {
328+
return "Not a Publisher"
329+
}
330+
}
331+
332+
GraphQL graphQL = buildSubscriptionQL(newMessageDF)
333+
334+
def executionInput = ExecutionInput.newExecutionInput().query("""
335+
subscription NewMessages {
336+
newMessage(roomId: 123) {
337+
sender
338+
text
339+
}
340+
}
341+
""").build()
342+
343+
when:
344+
graphQL.execute(executionInput)
345+
346+
then:
347+
thrown(AssertException)
348+
}
349+
319350
def "subscription query will surface event stream exceptions"() {
320351
321352
DataFetcher newMessageDF = new DataFetcher() {

‎src/test/groovy/graphql/execution/pubsub/CommonMessagePublisher.java‎

Lines changed: 44 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,44 @@
1+
package graphql.execution.pubsub;
2+
3+
import org.reactivestreams.example.unicast.AsyncIterablePublisher;
4+
5+
import java.util.Iterator;
6+
import java.util.concurrent.ForkJoinPool;
7+
import java.util.function.Function;
8+
9+
class CommonMessagePublisher {
10+
11+
protected final AsyncIterablePublisher<Message> iterablePublisher;
12+
13+
protected CommonMessagePublisher(final int count) {
14+
Iterable<Message> iterable = mkIterable(count, at -> {
15+
Message message = new Message("sender" + at, "text" + at);
16+
return examineMessage(message, at);
17+
});
18+
iterablePublisher = new AsyncIterablePublisher<>(iterable, ForkJoinPool.commonPool());
19+
}
20+
21+
@SuppressWarnings("unused")
22+
protected Message examineMessage(Message message, Integer at) {
23+
return message;
24+
}
25+
26+
private static Iterable<Message> mkIterable(int count, Function<Integer, Message> msgMaker) {
27+
return () -> new Iterator<>() {
28+
private int at = 0;
29+
30+
@Override
31+
public boolean hasNext() {
32+
return at < count;
33+
}
34+
35+
@Override
36+
public Message next() {
37+
Message message = msgMaker.apply(at);
38+
at++;
39+
return message;
40+
}
41+
};
42+
}
43+
44+
}

‎src/test/groovy/graphql/execution/pubsub/FlowMessagePublisher.java‎

Lines changed: 22 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,22 @@
1+
package graphql.execution.pubsub;
2+
3+
import org.reactivestreams.FlowAdapters;
4+
5+
import java.util.concurrent.Flow;
6+
7+
/**
8+
* This example publisher will create count "messages" and then terminate. It
9+
* uses the reactive streams TCK as its implementation but presents itself
10+
* as a {@link Flow.Publisher}
11+
*/
12+
public class FlowMessagePublisher extends CommonMessagePublisher implements Flow.Publisher<Message> {
13+
14+
public FlowMessagePublisher(int count) {
15+
super(count);
16+
}
17+
18+
@Override
19+
public void subscribe(Flow.Subscriber<? super Message> s) {
20+
iterablePublisher.subscribe(FlowAdapters.toSubscriber(s));
21+
}
22+
}

‎src/test/groovy/graphql/execution/pubsub/ReactiveStreamsMessagePublisher.java‎

Lines changed: 3 additions & 37 deletions
Original file line numberDiff line numberDiff line change
@@ -2,54 +2,20 @@
22

33
import org.reactivestreams.Publisher;
44
import org.reactivestreams.Subscriber;
5-
import org.reactivestreams.example.unicast.AsyncIterablePublisher;
6-
7-
import java.util.Iterator;
8-
import java.util.concurrent.ForkJoinPool;
9-
import java.util.function.Function;
105

116
/**
127
* This example publisher will create count "messages" and then terminate. It
138
* uses the reactive streams TCK as its implementation
149
*/
15-
public class ReactiveStreamsMessagePublisher implements Publisher<Message> {
16-
17-
private final AsyncIterablePublisher<Message> iterablePublisher;
10+
public class ReactiveStreamsMessagePublisher extends CommonMessagePublisher implements Publisher<Message> {
1811

1912
public ReactiveStreamsMessagePublisher(final int count) {
20-
Iterable<Message> iterable = mkIterable(count, at -> {
21-
Message message = new Message("sender" + at, "text" + at);
22-
return examineMessage(message, at);
23-
});
24-
iterablePublisher = new AsyncIterablePublisher<>(iterable, ForkJoinPool.commonPool());
13+
super(count);
2514
}
2615

2716
@Override
2817
public void subscribe(Subscriber<? super Message> s) {
2918
iterablePublisher.subscribe(s);
3019
}
31-
32-
@SuppressWarnings("unused")
33-
protected Message examineMessage(Message message, Integer at) {
34-
return message;
35-
}
36-
37-
private static Iterable<Message> mkIterable(int count, Function<Integer, Message> msgMaker) {
38-
return () -> new Iterator<Message>() {
39-
private int at = 0;
40-
41-
@Override
42-
public boolean hasNext() {
43-
return at < count;
44-
}
45-
46-
@Override
47-
public Message next() {
48-
Message message = msgMaker.apply(at);
49-
at++;
50-
return message;
51-
}
52-
};
53-
}
54-
5520
}
21+

0 commit comments

Comments
 (0)