Repository navigation
Expand file tree
/
Copy pathSubscriptionExecutionStrategy.java
More file actions
252 lines (211 loc) · 13.7 KB
/
Copy pathSubscriptionExecutionStrategy.java
File metadata and controls
252 lines (211 loc) · 13.7 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
package graphql.execution;
import graphql.Assert;
import graphql.ExecutionResult;
import graphql.ExecutionResultImpl;
import graphql.GraphQLContext;
import graphql.PublicApi;
import org.jspecify.annotations.NullMarked;
import org.jspecify.annotations.Nullable;
import graphql.execution.incremental.AlternativeCallContext;
import graphql.execution.instrumentation.ExecutionStrategyInstrumentationContext;
import graphql.execution.instrumentation.Instrumentation;
import graphql.execution.instrumentation.InstrumentationContext;
import graphql.execution.instrumentation.parameters.InstrumentationExecutionParameters;
import graphql.execution.instrumentation.parameters.InstrumentationExecutionStrategyParameters;
import graphql.execution.instrumentation.parameters.InstrumentationFieldParameters;
import graphql.execution.instrumentation.parameters.InstrumentationReactiveResultsParameters;
import graphql.execution.reactive.SubscriptionPublisher;
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.assertNotNull;
import static graphql.execution.instrumentation.SimpleInstrumentationContext.nonNullCtx;
import static java.util.Collections.singletonMap;
/**
* An execution strategy that implements graphql subscriptions by using reactive-streams
* as the output result of the subscription query.
* <p>
* Afterwards each object delivered on that stream will be mapped via running the original selection set over that object and hence producing an ExecutionResult
* just like a normal graphql query.
* <p>
* See <a href="https://spec.graphql.org/draft/#sec-Subscription">https://spec.graphql.org/draft/#sec-Subscription</a>
* <p>
* See <a href="https://www.reactive-streams.org/">https://www.reactive-streams.org/</a>
*/
@PublicApi
@NullMarked
public class SubscriptionExecutionStrategy extends ExecutionStrategy {
/**
* If a boolean value is placed into the {@link GraphQLContext} with this key then the order
* of the subscription events can be controlled. By default, subscription events are published
* as the graphql subselection calls complete, and not in the order they originally arrived from the
* source publisher. But this can be changed to {@link Boolean#TRUE} to keep them in order.
*/
public static final String KEEP_SUBSCRIPTION_EVENTS_ORDERED = "KEEP_SUBSCRIPTION_EVENTS_ORDERED";
public SubscriptionExecutionStrategy() {
super();
}
public SubscriptionExecutionStrategy(DataFetcherExceptionHandler dataFetcherExceptionHandler) {
super(dataFetcherExceptionHandler);
}
@Override
public CompletableFuture<ExecutionResult> execute(ExecutionContext executionContext, ExecutionStrategyParameters parameters) throws NonNullableFieldWasNullException {
Instrumentation instrumentation = executionContext.getInstrumentation();
InstrumentationExecutionStrategyParameters instrumentationParameters = new InstrumentationExecutionStrategyParameters(executionContext, parameters);
ExecutionStrategyInstrumentationContext executionStrategyCtx = ExecutionStrategyInstrumentationContext.nonNullCtx(instrumentation.beginExecutionStrategy(
instrumentationParameters,
executionContext.getInstrumentationState()
));
CompletableFuture<Publisher<Object>> sourceEventStream = createSourceEventStream(executionContext, parameters);
//
// when the upstream source event stream completes, subscribe to it and wire in our adapter
CompletableFuture<ExecutionResult> overallResult = sourceEventStream.thenApply((publisher) ->
{
if (publisher == null) {
return new ExecutionResultImpl(null, executionContext.getErrors());
}
Function<Object, CompletionStage<ExecutionResult>> mapperFunction = eventPayload -> executeSubscriptionEvent(executionContext, parameters, eventPayload);
boolean keepOrdered = keepOrdered(executionContext.getGraphQLContext());
InstrumentationReactiveResultsParameters instrumentationReactiveResultsParameters = new InstrumentationReactiveResultsParameters(executionContext, InstrumentationReactiveResultsParameters.ResultType.SUBSCRIPTION);
InstrumentationContext<Void> reactiveCtx = nonNullCtx(executionContext.getInstrumentation().beginReactiveResults(instrumentationReactiveResultsParameters, executionContext.getInstrumentationState()));
reactiveCtx.onDispatched();
SubscriptionPublisher mapSourceToResponse = new SubscriptionPublisher(publisher, mapperFunction, keepOrdered,
throwable -> reactiveCtx.onCompleted(null, throwable));
return new ExecutionResultImpl(mapSourceToResponse, executionContext.getErrors());
});
// dispatched the subscription query
executionStrategyCtx.onDispatched();
overallResult.whenComplete(executionStrategyCtx::onCompleted);
return overallResult;
}
private boolean keepOrdered(GraphQLContext graphQLContext) {
return graphQLContext.getOrDefault(KEEP_SUBSCRIPTION_EVENTS_ORDERED, false);
}
/*
https://github.com/facebook/graphql/blob/master/spec/Section%206%20--%20Execution.md
CreateSourceEventStream(subscription, schema, variableValues, initialValue):
Let {subscriptionType} be the root Subscription type in {schema}.
Assert: {subscriptionType} is an Object type.
Let {selectionSet} be the top level Selection Set in {subscription}.
Let {rootField} be the first top level field in {selectionSet}.
Let {argumentValues} be the result of {CoerceArgumentValues(subscriptionType, rootField, variableValues)}.
Let {fieldStream} be the result of running {ResolveFieldEventStream(subscriptionType, initialValue, rootField, argumentValues)}.
Return {fieldStream}.
*/
private CompletableFuture<Publisher<Object>> createSourceEventStream(ExecutionContext executionContext, ExecutionStrategyParameters parameters) {
ExecutionStrategyParameters newParameters = firstFieldOfSubscriptionSelection(executionContext, parameters, false);
CompletableFuture<Object> fieldFetched = Async.toCompletableFuture(fetchField(executionContext, newParameters));
return fieldFetched.thenApply(fetchedValue -> {
Object publisher = FetchedValue.getFetchedValue(fetchedValue);
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 @Nullable Publisher<Object> mkReactivePublisher(@Nullable 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):
Let {subscriptionType} be the root Subscription type in {schema}.
Assert: {subscriptionType} is an Object type.
Let {selectionSet} be the top level Selection Set in {subscription}.
Let {data} be the result of running {ExecuteSelectionSet(selectionSet, subscriptionType, initialValue, variableValues)} normally (allowing parallelization).
Let {errors} be any field errors produced while executing the selection set.
Return an unordered map containing {data} and {errors}.
Note: The {ExecuteSubscriptionEvent()} algorithm is intentionally similar to {ExecuteQuery()} since this is how each event result is produced.
*/
private CompletableFuture<ExecutionResult> executeSubscriptionEvent(ExecutionContext executionContext, ExecutionStrategyParameters parameters, Object eventPayload) {
Instrumentation instrumentation = executionContext.getInstrumentation();
ExecutionContext newExecutionContext = executionContext.transform(builder -> builder
.root(eventPayload)
.resetErrors()
);
ExecutionStrategyParameters newParameters = firstFieldOfSubscriptionSelection(newExecutionContext, parameters, true);
ExecutionStepInfo subscribedFieldStepInfo = createSubscribedFieldStepInfo(executionContext, newParameters);
InstrumentationFieldParameters i13nFieldParameters = new InstrumentationFieldParameters(executionContext, () -> subscribedFieldStepInfo);
InstrumentationContext<ExecutionResult> subscribedFieldCtx = nonNullCtx(instrumentation.beginSubscribedFieldEvent(
i13nFieldParameters, executionContext.getInstrumentationState()
));
AlternativeCallContext alternativeCallContext = assertNotNull(
newParameters.getAlternativeCallContext(),
"alternativeCallContext must not be null");
executionContext.getDataLoaderDispatcherStrategy().newSubscriptionExecution(alternativeCallContext);
Object fetchedValue = unboxPossibleDataFetcherResult(newExecutionContext, newParameters, eventPayload);
FieldValueInfo fieldValueInfo = completeField(newExecutionContext, newParameters, fetchedValue);
executionContext.getDataLoaderDispatcherStrategy().subscriptionEventCompletionDone(alternativeCallContext);
CompletableFuture<ExecutionResult> overallResult = fieldValueInfo
.getFieldValueFuture()
.thenApply(val -> new ExecutionResultImpl(val, alternativeCallContext.getErrors()))
.thenApply(executionResult -> wrapWithRootFieldName(newParameters, executionResult))
.whenComplete((executionResult, throwable) -> {
executionContext.getDataLoaderDispatcherStrategy().subscriptionEventExecutionDone(alternativeCallContext);
});
// dispatch instrumentation so they can know about each subscription event
subscribedFieldCtx.onDispatched();
overallResult.whenComplete(subscribedFieldCtx::onCompleted);
// allow them to instrument each ER should they want to
InstrumentationExecutionParameters i13nExecutionParameters = new InstrumentationExecutionParameters(
executionContext.getExecutionInput(), executionContext.getGraphQLSchema());
overallResult = overallResult.thenCompose(executionResult -> instrumentation.instrumentExecutionResult(executionResult, i13nExecutionParameters, executionContext.getInstrumentationState()));
return overallResult;
}
private ExecutionResult wrapWithRootFieldName(ExecutionStrategyParameters parameters, ExecutionResult executionResult) {
String rootFieldName = getRootFieldName(parameters);
return new ExecutionResultImpl(
singletonMap(rootFieldName, executionResult.getData()),
executionResult.getErrors()
);
}
private String getRootFieldName(ExecutionStrategyParameters parameters) {
Field rootField = assertNotNull(parameters.getField(), "field must not be null").getSingleField();
return rootField.getResultKey();
}
private ExecutionStrategyParameters firstFieldOfSubscriptionSelection(ExecutionContext executionContext,
ExecutionStrategyParameters parameters,
boolean newCallContext) {
MergedSelectionSet fields = parameters.getFields();
MergedField firstField = assertNotNull(fields.getSubField(fields.getKeys().get(0)), "firstField must not be null");
ResultPath fieldPath = parameters.getPath().segment(mkNameForPath(firstField.getSingleField()));
NonNullableFieldValidator nonNullableFieldValidator = new NonNullableFieldValidator(executionContext);
return parameters.transform(builder -> {
builder
.field(firstField)
.path(fieldPath)
.nonNullFieldValidator(nonNullableFieldValidator);
if (newCallContext) {
builder.alternativeCallContext(new AlternativeCallContext(1, 1));
}
});
}
private ExecutionStepInfo createSubscribedFieldStepInfo(ExecutionContext
executionContext, ExecutionStrategyParameters parameters) {
Field field = assertNotNull(parameters.getField(), "field must not be null").getSingleField();
GraphQLObjectType parentType = parameters.getExecutionStepInfo().getUnwrappedNonNullTypeAs();
GraphQLFieldDefinition fieldDef = getFieldDef(executionContext.getGraphQLSchema(), parentType, field);
return createExecutionStepInfo(executionContext, parameters, fieldDef, parentType);
}
}