In serial mode, resolve the fields in given order

Author: Karthick Sankarachary

src/main/java/graphql/async/ExecutionFuture.java

@@ -8,12 +8,10 @@
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.ForkJoinPool;
-import java.util.function.Function;
 
 import graphql.ExceptionWhileDataFetching;
 import graphql.ExecutionResult;
 import graphql.ExecutionResultImpl;
-import graphql.GraphQLError;
 
 public class ExecutionFuture {
 
@@ -31,17 +29,17 @@ public static CompletableFuture<ExecutionResult> completable(ExecutionResult exe
     Object executionData = executionResult != null ? executionResult.getData() : null;
     if (executionData instanceof CompletableFuture) {
       return ((CompletableFuture) executionData)
-        .exceptionally((throwable) -> {
+        .exceptionally(throwable -> {
           ((ExecutionResultImpl) executionResult).addErrors(Arrays.asList(
             new ExceptionWhileDataFetching((Throwable) throwable)
           ));
           return executionResult;
         })
         .thenApplyAsync(
-            (completedData) -> {
-              ((ExecutionResultImpl) executionResult).setData(completedData);
-              return executionResult;
-            }, executorService);
+          completedData -> {
+            ((ExecutionResultImpl) executionResult).setData(completedData);
+            return executionResult;
+          }, executorService);
     } else if (executionData instanceof Map) {
       final Map<String, Object> results = (Map<String, Object>) executionData;
       List<CompletableFuture<Object>> completables = new ArrayList<>();
@@ -50,19 +48,19 @@ public static CompletableFuture<ExecutionResult> completable(ExecutionResult exe
         if (fieldValue instanceof CompletableFuture) {
           completables.add(
             ((CompletableFuture) fieldValue)
-              .thenAccept((fieldValueDone) -> results.put(fieldName, fieldValueDone)));
+              .thenAccept(fieldValueDone -> results.put(fieldName, fieldValueDone)));
         }
       }
       if (!completables.isEmpty()) {
         return CompletableFuture.allOf(array(completables))
-          .exceptionally((throwable) -> {
+          .exceptionally(throwable -> {
             ((ExecutionResultImpl) executionResult).addErrors(Arrays.asList(
               new ExceptionWhileDataFetching(throwable)
             ));
             return null;
           })
           .thenApply(
-              (completablesDone) -> executionResult);
+            completablesDone -> executionResult);
       }
     }
     return CompletableFuture.completedFuture(executionResult);

src/main/java/graphql/async/GraphQL.java

@@ -8,7 +8,6 @@
 import graphql.ExecutionResult;
 import graphql.execution.AsyncExecutionStrategy;
 import graphql.execution.ExecutionStrategy;
-import graphql.execution.SimpleExecutionStrategy;
 import graphql.schema.GraphQLSchema;
 
 import static graphql.Assert.assertNotNull;
@@ -44,7 +43,7 @@ public GraphQL(GraphQLSchema graphQLSchema) {
    */
   public GraphQL(GraphQLSchema graphQLSchema, ExecutionStrategy queryStrategy) {
     //noinspection deprecation
-    super(graphQLSchema, queryStrategy, AsyncExecutionStrategy.parallel());
+    super(graphQLSchema, queryStrategy, AsyncExecutionStrategy.serial());
   }
 
   /**
@@ -59,8 +58,12 @@ public GraphQL(GraphQLSchema graphQLSchema, ExecutionStrategy queryStrategy) {
   public GraphQL(GraphQLSchema graphQLSchema, ExecutionStrategy queryStrategy,
                  ExecutionStrategy mutationStrategy) {
     super(graphQLSchema, queryStrategy, mutationStrategy);
-    assert queryStrategy instanceof AsyncExecutionStrategy;
-    assert mutationStrategy instanceof AsyncExecutionStrategy;
+    assert queryStrategy instanceof AsyncExecutionStrategy :
+      "Async graphql requires an async query strategy";
+    assert mutationStrategy instanceof AsyncExecutionStrategy  :
+      "Async graphql requires an async mutation strategy";
+    assert ((AsyncExecutionStrategy) mutationStrategy).isSerial() :
+      "Async graphql requires a serial mutation strategy";
   }
 
   /**

src/main/java/graphql/execution/AsyncExecutionStrategy.java

@@ -1,7 +1,6 @@
 package graphql.execution;
 
 import java.util.ArrayList;
-import java.util.Arrays;
 import java.util.LinkedHashMap;
 import java.util.List;
 import java.util.Map;
@@ -11,7 +10,6 @@
 import java.util.concurrent.ForkJoinPool;
 import java.util.function.Function;
 import java.util.stream.Collectors;
-import java.util.stream.Stream;
 
 import graphql.ExceptionWhileDataFetching;
 import graphql.ExecutionResult;
@@ -21,8 +19,8 @@
 import graphql.schema.GraphQLObjectType;
 import graphql.schema.GraphQLType;
 
-import static graphql.async.ExecutionFuture.completable;
 import static graphql.async.ExecutionFuture.array;
+import static graphql.async.ExecutionFuture.completable;
 
 /**
  * <p>AsyncExecutionStrategy implements the {@link ExecutionStrategy} in a non-blocking manner.</p>
@@ -42,8 +40,8 @@
  */
 public class AsyncExecutionStrategy extends ExecutionStrategy {
 
-  protected ExecutorService executorService;
   protected boolean serial;
+  protected ExecutorService executorService;
 
   public static AsyncExecutionStrategy serial() {
     return new AsyncExecutionStrategy(true);
@@ -70,6 +68,10 @@ protected AsyncExecutionStrategy(boolean serial, ExecutorService executorService
     this.executorService = executorService;
   }
 
+  public boolean isSerial() {
+    return serial;
+  }
+
   /**
    * Resolve the given fields in parallel and return an execution result without blocking.
    *
@@ -87,17 +89,32 @@ public ExecutionResult execute(final ExecutionContext executionContext,
     Set<String> fieldNames = fields.keySet();
 
     // Create tasks to resolve each of the fields
-    (serial ? fieldNames.stream() : fieldNames.parallelStream()).forEach((fieldName) -> {
-      fieldFutures.put(fieldName, CompletableFuture.supplyAsync(
-        () -> resolveField(executionContext, parentType, source, fields.get(fieldName)),
-        executorService));
-    });
+    CompletableFuture<ExecutionResult> previousField = CompletableFuture.completedFuture(null);
+    for (String fieldName : fieldNames) {
+      CompletableFuture<ExecutionResult> fieldFuture;
+      if (serial) {
+        // Block the current field until the previous field is done
+        fieldFuture = previousField.thenCompose(result -> CompletableFuture.supplyAsync(
+          () -> resolveField(executionContext, parentType, source, fields.get(fieldName)),
+          executorService
+        ));
+        previousField = fieldFuture;
+      } else {
+        // Resolve every field in parallel, independent of each other
+        fieldFuture = CompletableFuture.supplyAsync(
+          () -> resolveField(executionContext, parentType, source, fields.get(fieldName)),
+          executorService
+        );
+      }
+      fieldFutures.put(fieldName, fieldFuture);
+    }
+
 
     // Prepare a completable for the map of field results
     CompletableFuture<Map<String, Object>> resultsFuture = CompletableFuture
       // First, wait for all the tasks above to complete
       .allOf(array(fieldFutures.values()))
-      .exceptionally((throwable) -> {
+      .exceptionally(throwable -> {
         executionContext.addError(new ExceptionWhileDataFetching(throwable));
         return null;
       })
@@ -107,25 +124,25 @@ public ExecutionResult execute(final ExecutionContext executionContext,
         return CompletableFuture
           .allOf(array(
             fieldNames.stream()
-              .map((fieldName) -> {
+              .map(fieldName -> {
                 final ExecutionResult fieldResult = fieldFutures.get(fieldName).join();
                 fieldResults.put(fieldName, fieldResult);
                 return completable(fieldResult, executorService);
               })
               .collect(Collectors.<CompletableFuture<ExecutionResult>>toList())
           ))
-          .exceptionally((throwable) -> {
+          .exceptionally(throwable -> {
             executionContext.addError(new ExceptionWhileDataFetching(throwable));
             return null;
           })
-          .thenApplyAsync((resultsDone) -> fieldResults, executorService);
+          .thenApplyAsync(resultsDone -> fieldResults, executorService);
       }, executorService)
       // Last, collect the results of the field into a map
-      .exceptionally((throwable) -> {
+      .exceptionally(throwable -> {
         executionContext.addError(new ExceptionWhileDataFetching(throwable));
         return new LinkedHashMap<>();
       })
-      .thenApplyAsync((fieldResults) -> {
+      .thenApplyAsync(fieldResults -> {
         Map<String, Object> results = new LinkedHashMap<>();
         for (String fieldName : fieldResults.keySet()) {
           ExecutionResult fieldResult = fieldResults.get(fieldName);
@@ -155,10 +172,10 @@ protected ExecutionResult completeValue(final ExecutionContext executionContext,
     return ((CompletableFuture<ExecutionResult>) result)
       .thenComposeAsync(
         (Function<Object, CompletableFuture<ExecutionResult>>)
-          (completedResult) -> completable(
+          completedResult -> completable(
             executionStrategy.completeValue(executionContext, fieldType, fields, completedResult),
             executorService), executorService)
-      .exceptionally((throwable) -> {
+      .exceptionally(throwable -> {
         executionContext.addError(new ExceptionWhileDataFetching(throwable));
         return null;
       })
@@ -180,7 +197,7 @@ protected ExecutionResult completeValueForList(ExecutionContext executionContext
       super.completeValueForList(executionContext, fieldType, fields, result);
     List<Object> completableResults = (List<Object>) executionResult.getData();
     List<Object> completedResults = new ArrayList<>();
-    completableResults.forEach((completedResult) -> {
+    completableResults.forEach(completedResult -> {
       completedResults.add((completedResult instanceof CompletableFuture) ?
                            ((CompletableFuture) completedResult).join() : completedResult);
     });

src/test/groovy/graphql/execution/AsyncExecutionStrategyTest.groovy

@@ -64,7 +64,6 @@ class AsyncExecutionStrategyTest extends Specification {
                 new ThreadPoolExecutor.CallerRunsPolicy());
 
         def graphQL = GraphQL.newAsyncGraphQL(NewsSchema.newsSchema)
-                .queryExecutionStrategy(AsyncExecutionStrategy.serial(threadPoolExecutor))
                 .build()
 
         def received = new AtomicReference<Map>()