make list completion not sequential, but fully async

andimarek · andimarek · commit 0ff4424c9a19 · 2017-08-21T07:37:21.000+02:00
diff --git a/src/main/java/graphql/execution/Async.java b/src/main/java/graphql/execution/Async.java
@@ -7,6 +7,7 @@
 import java.util.Iterator;
 import java.util.List;
 import java.util.concurrent.CompletableFuture;
+import java.util.function.BiFunction;
 
 @Internal
 public class Async {
@@ -34,6 +35,23 @@ public static <U> CompletableFuture<List<U>> each(List<CompletableFuture<U>> fut
         return overallResult;
     }
 
+    public static <T, U> CompletableFuture<List<U>> each(Iterable<T> list, BiFunction<T, Integer, CompletableFuture<U>> cfFactory) {
+        List<CompletableFuture<U>> futures = new ArrayList<>();
+        int index = 0;
+        for (T t : list) {
+            CompletableFuture<U> cf;
+            try {
+                cf = cfFactory.apply(t, index++);
+                Assert.assertNotNull(cf, "cfFactory must return a non null value");
+            } catch (Exception e) {
+                cf = new CompletableFuture<>();
+                cf.completeExceptionally(e);
+            }
+            futures.add(cf);
+        }
+        return each(futures);
+
+    }
 
     public static <T, U> CompletableFuture<List<U>> eachSequentially(Iterable<T> list, CFFactory<T, U> cfFactory) {
         CompletableFuture<List<U>> result = new CompletableFuture<>();
diff --git a/src/main/java/graphql/execution/ExecutionStrategy.java b/src/main/java/graphql/execution/ExecutionStrategy.java
@@ -510,7 +510,7 @@ protected CompletableFuture<ExecutionResult> completeValueForList(ExecutionConte
         ExecutionTypeInfo typeInfo = parameters.typeInfo();
         GraphQLList fieldType = typeInfo.castType(GraphQLList.class);
 
-        CompletableFuture<List<ExecutionResult>> resultsFuture = Async.eachSequentially(iterableValues, (item, index, prevResults) -> {
+        CompletableFuture<List<ExecutionResult>> resultsFuture = Async.each(iterableValues, (item, index) -> {
 
             ExecutionPath indexedPath = parameters.path().segment(index);
 
diff --git a/src/test/groovy/graphql/execution/AsyncTest.groovy b/src/test/groovy/graphql/execution/AsyncTest.groovy
@@ -3,6 +3,10 @@ package graphql.execution
 import spock.lang.Specification
 
 import java.util.concurrent.CompletableFuture
+import java.util.concurrent.CompletionException
+import java.util.function.BiFunction
+
+import static java.util.concurrent.CompletableFuture.completedFuture
 
 class AsyncTest extends Specification {
 
@@ -47,7 +51,7 @@ class AsyncTest extends Specification {
         given:
         def input = ['a', 'b', 'c']
         def cfFactory = Mock(Async.CFFactory)
-        cfFactory.apply('a', 0, _) >> CompletableFuture.completedFuture("x")
+        cfFactory.apply('a', 0, _) >> completedFuture("x")
         cfFactory.apply('b', 1, _) >> {
             def cf = new CompletableFuture<>()
             cf.completeExceptionally(new RuntimeException("some error"))
@@ -71,7 +75,7 @@ class AsyncTest extends Specification {
         given:
         def input = ['a', 'b', 'c']
         def cfFactory = Mock(Async.CFFactory)
-        cfFactory.apply('a', 0, _) >> CompletableFuture.completedFuture("x")
+        cfFactory.apply('a', 0, _) >> completedFuture("x")
         cfFactory.apply('b', 1, _) >> { throw new RuntimeException("some error") }
 
         when:
@@ -87,4 +91,55 @@ class AsyncTest extends Specification {
         exception.getMessage() == "some error"
     }
 
+    def "each works for mapping function"() {
+        given:
+        def input = ['a', 'b', 'c']
+        def cfFactory = Mock(BiFunction)
+        cfFactory.apply('a', 0) >> completedFuture('x')
+        cfFactory.apply('b', 1) >> completedFuture('y')
+        cfFactory.apply('c', 2) >> completedFuture('z')
+
+
+        when:
+        def result = Async.each(input, cfFactory)
+
+        then:
+        result.isDone()
+        result.get() == ['x', 'y', 'z']
+    }
+
+    def "each with mapping function propagates factory exception"() {
+        given:
+        def input = ['a', 'b', 'c']
+        def cfFactory = Mock(BiFunction)
+
+
+        when:
+        def result = Async.each(input, cfFactory)
+
+        then:
+        1 * cfFactory.apply('a', 0) >> completedFuture('x')
+        1 * cfFactory.apply('b', 1) >> { throw new RuntimeException('some error') }
+        1 * cfFactory.apply('c', 2) >> completedFuture('z')
+        result.isCompletedExceptionally()
+        Throwable exception
+        result.exceptionally({ e ->
+            exception = e
+        })
+        exception instanceof CompletionException
+        exception.getCause().getMessage() == "some error"
+    }
+
+    def "each works for list of futures "() {
+        given:
+        completedFuture('x')
+
+        when:
+        def result = Async.each([completedFuture('x'), completedFuture('y'), completedFuture('z')])
+
+        then:
+        result.isDone()
+        result.get() == ['x', 'y', 'z']
+    }
+
 }
