Repository navigation
Expand file tree
/
Copy pathSubscriptionExecutionStrategy.java
More file actions
208 lines (171 loc) · 11.1 KB
/
Copy pathSubscriptionExecutionStrategy.java
File metadata and controls
208 lines (171 loc) · 11.1 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
package graphql.execution;
import graphql.ExecutionResult;
import graphql.ExecutionResultImpl;
import graphql.GraphQLContext;
import graphql.PublicApi;
import graphql.execution.incremental.DeferredCallContext;
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.reactive.SubscriptionPublisher;
import graphql.language.Field;
import graphql.schema.GraphQLFieldDefinition;
import graphql.schema.GraphQLObjectType;
import org.reactivestreams.Publisher;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.CompletionStage;
import java.util.function.Function;
import static graphql.Assert.assertTrue;
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
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) {
ExecutionResultImpl executionResult = new ExecutionResultImpl(null, executionContext.getErrors());
return executionResult;
}
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;
});
// 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);
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;
});
}
/*
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);
ExecutionStepInfo subscribedFieldStepInfo = createSubscribedFieldStepInfo(executionContext, newParameters);
InstrumentationFieldParameters i13nFieldParameters = new InstrumentationFieldParameters(executionContext, () -> subscribedFieldStepInfo);
InstrumentationContext<ExecutionResult> subscribedFieldCtx = nonNullCtx(instrumentation.beginSubscribedFieldEvent(
i13nFieldParameters, executionContext.getInstrumentationState()
));
FetchedValue fetchedValue = unboxPossibleDataFetcherResult(newExecutionContext, parameters, eventPayload);
FieldValueInfo fieldValueInfo = completeField(newExecutionContext, newParameters, fetchedValue);
executionContext.getDataLoaderDispatcherStrategy().newSubscriptionExecution(fieldValueInfo, newParameters.getDeferredCallContext());
CompletableFuture<ExecutionResult> overallResult = fieldValueInfo
.getFieldValueFuture()
.thenApply(val -> new ExecutionResultImpl(val, newExecutionContext.getErrors()))
.thenApply(executionResult -> wrapWithRootFieldName(newParameters, executionResult));
// 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 = parameters.getField().getSingleField();
return rootField.getResultKey();
}
private ExecutionStrategyParameters firstFieldOfSubscriptionSelection(ExecutionContext executionContext, ExecutionStrategyParameters parameters) {
MergedSelectionSet fields = parameters.getFields();
MergedField firstField = fields.getSubField(fields.getKeys().get(0));
ResultPath fieldPath = parameters.getPath().segment(mkNameForPath(firstField.getSingleField()));
NonNullableFieldValidator nonNullableFieldValidator = new NonNullableFieldValidator(executionContext);
return parameters.transform(builder -> builder
.field(firstField)
.path(fieldPath)
.nonNullFieldValidator(nonNullableFieldValidator)
.deferredCallContext(new DeferredCallContext(1, 1))
);
}
private ExecutionStepInfo createSubscribedFieldStepInfo(ExecutionContext executionContext, ExecutionStrategyParameters parameters) {
Field field = parameters.getField().getSingleField();
GraphQLObjectType parentType = (GraphQLObjectType) parameters.getExecutionStepInfo().getUnwrappedNonNullType();
GraphQLFieldDefinition fieldDef = getFieldDef(executionContext.getGraphQLSchema(), parentType, field);
return createExecutionStepInfo(executionContext, parameters, fieldDef, parentType);
}
}