dataloader for subscriptions

andimarek · andimarek · commit d9c3abd1f9f5 · 2025-06-12T13:02:06.000+10:00
diff --git a/src/main/java/graphql/execution/SubscriptionExecutionStrategy.java b/src/main/java/graphql/execution/SubscriptionExecutionStrategy.java
@@ -107,7 +107,7 @@ private boolean keepOrdered(GraphQLContext graphQLContext) {
      */
 
     private CompletableFuture<Publisher<Object>> createSourceEventStream(ExecutionContext executionContext, ExecutionStrategyParameters parameters) {
-        ExecutionStrategyParameters newParameters = firstFieldOfSubscriptionSelection(executionContext, parameters);
+        ExecutionStrategyParameters newParameters = firstFieldOfSubscriptionSelection(executionContext, parameters, false);
 
         CompletableFuture<FetchedValue> fieldFetched = Async.toCompletableFuture(fetchField(executionContext, newParameters));
         return fieldFetched.thenApply(fetchedValue -> {
@@ -141,20 +141,20 @@ private CompletableFuture<ExecutionResult> executeSubscriptionEvent(ExecutionCon
                 .root(eventPayload)
                 .resetErrors()
         );
-        ExecutionStrategyParameters newParameters = firstFieldOfSubscriptionSelection(newExecutionContext, parameters);
+        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()
         ));
 
-        FetchedValue fetchedValue = unboxPossibleDataFetcherResult(newExecutionContext, parameters, eventPayload);
+        FetchedValue fetchedValue = unboxPossibleDataFetcherResult(newExecutionContext, newParameters, 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(val -> new ExecutionResultImpl(val, newParameters.getDeferredCallContext().getErrors()))
                 .thenApply(executionResult -> wrapWithRootFieldName(newParameters, executionResult));
 
         // dispatch instrumentation so they can know about each subscription event
@@ -182,24 +182,30 @@ private String getRootFieldName(ExecutionStrategyParameters parameters) {
         return rootField.getResultKey();
     }
 
-    private ExecutionStrategyParameters firstFieldOfSubscriptionSelection(ExecutionContext executionContext, ExecutionStrategyParameters parameters) {
+    private ExecutionStrategyParameters firstFieldOfSubscriptionSelection(ExecutionContext executionContext,
+                                                                          ExecutionStrategyParameters parameters,
+                                                                          boolean newCallContext) {
         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))
-        );
+        return parameters.transform(builder -> {
+            builder
+                    .field(firstField)
+                    .path(fieldPath)
+                    .nonNullFieldValidator(nonNullableFieldValidator);
+            if (newCallContext) {
+                builder.deferredCallContext(new DeferredCallContext(1, 1));
+            }
+        });
 
     }
 
-    private ExecutionStepInfo createSubscribedFieldStepInfo(ExecutionContext executionContext, ExecutionStrategyParameters parameters) {
+    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);
diff --git a/src/main/java/graphql/execution/incremental/DeferredCallContext.java b/src/main/java/graphql/execution/incremental/DeferredCallContext.java
@@ -4,8 +4,9 @@
 import graphql.Internal;
 import graphql.VisibleForTesting;
 
+import java.util.ArrayList;
+import java.util.Collections;
 import java.util.List;
-import java.util.concurrent.CopyOnWriteArrayList;
 
 /**
  * Contains data relevant to the execution of a {@link DeferredFragmentCall}.
@@ -24,7 +25,7 @@ public class DeferredCallContext {
     private final int startLevel;
     private final int fields;
 
-    private final List<GraphQLError> errors = new CopyOnWriteArrayList<>();
+    private final List<GraphQLError> errors = Collections.synchronizedList(new ArrayList<>());
 
     public DeferredCallContext(int startLevel, int fields) {
         this.startLevel = startLevel;
