graphql-java · gkesler · Dec 29, 2018 · Dec 29, 2018 · Dec 29, 2018 · Dec 30, 2018
diff --git a/src/main/java/graphql/execution/Async.java b/src/main/java/graphql/execution/Async.java
@@ -4,15 +4,18 @@
 import graphql.Internal;

 import java.util.ArrayList;
+import java.util.Collections;
 import java.util.Iterator;
 import java.util.List;
+import java.util.Optional;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.CompletionException;
 import java.util.concurrent.CompletionStage;
 import java.util.function.BiFunction;
 import java.util.function.Function;
 import java.util.function.Supplier;
 import java.util.stream.Collectors;
+import java.util.stream.StreamSupport;

 @Internal
 @SuppressWarnings("FutureReturnValueIgnored")
@@ -24,41 +27,27 @@ public interface CFFactory<T, U> {
    }

    public static <U> CompletableFuture<List<U>> each(List<CompletableFuture<U>> futures) {
-        CompletableFuture<List<U>> overallResult = new CompletableFuture<>();
-
-        CompletableFuture
-                .allOf(futures.toArray(new CompletableFuture[0]))
-                .whenComplete((noUsed, exception) -> {
-                    if (exception != null) {
-                        overallResult.completeExceptionally(exception);
-                        return;
-                    }
-                    List<U> results = new ArrayList<>();
-                    for (CompletableFuture<U> future : futures) {
-                        results.add(future.join());
-                    }
-                    overallResult.complete(results);
-                });
-        return overallResult;
+        Assert.assertNotNull(futures);
+
+        return CompletableFuture.allOf(futures.toArray(new CompletableFuture<?>[futures.size()]))
+                .thenApply(noUsed -> futures
+                    .stream()
+                    .map(CompletableFuture::join)
+                    .collect(Collectors.toList()));
    }

    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<>();
-                // Async.each makes sure that it is not a CompletionException inside a CompletionException
-                cf.completeExceptionally(new CompletionException(e));
-            }
-            futures.add(cf);
-        }
-        return each(futures);
-
+        Assert.assertNotNull(list);
+        Assert.assertNotNull(cfFactory);
+
+        int index[] = {0};
+        return each(
+            StreamSupport
+                .stream(list.spliterator(), false)
+                .map(o -> tryCatch(() -> 
+                    Assert.assertNotNull(cfFactory.apply(o, index[0]++), "cfFactory must return a non null value")))
+                .collect(Collectors.toList())
+        );
    }

    public static <T, U> CompletableFuture<List<U>> eachSequentially(Iterable<T> list, CFFactory<T, U> cfFactory) {

diff --git a/src/main/java/graphql/execution/FieldCollector.java b/src/main/java/graphql/execution/FieldCollector.java
@@ -120,7 +120,7 @@ private String getFieldEntryKey(Field field) {
    }


-    private boolean doesFragmentConditionMatch(FieldCollectorParameters parameters, InlineFragment inlineFragment) {
+    protected boolean doesFragmentConditionMatch(FieldCollectorParameters parameters, InlineFragment inlineFragment) {
        if (inlineFragment.getTypeCondition() == null) {
            return true;
        }
@@ -129,7 +129,7 @@ private boolean doesFragmentConditionMatch(FieldCollectorParameters parameters,
        return checkTypeCondition(parameters, conditionType);
    }

-    private boolean doesFragmentConditionMatch(FieldCollectorParameters parameters, FragmentDefinition fragmentDefinition) {
+    protected boolean doesFragmentConditionMatch(FieldCollectorParameters parameters, FragmentDefinition fragmentDefinition) {
        GraphQLType conditionType;
        conditionType = getTypeFromAST(parameters.getGraphQLSchema(), fragmentDefinition.getTypeCondition());
        return checkTypeCondition(parameters, conditionType);

diff --git a/src/main/java/graphql/execution3/DAGExecutionStrategy.java b/src/main/java/graphql/execution3/DAGExecutionStrategy.java
@@ -0,0 +1,234 @@
+/*
+ * To change this license header, choose License Headers in Project Properties.
+ * To change this template file, choose Tools | Templates
+ * and open the template in the editor.
+ */
+package graphql.execution3;
+
+import static graphql.Assert.assertNotNull;
+import graphql.ExecutionResult;
+import graphql.ExecutionResultImpl;
+import graphql.execution.Async;
+import graphql.execution.ExecutionContext;
+import graphql.execution.ExecutionPath;
+import graphql.execution.ExecutionStepInfo;
+import static graphql.execution.ExecutionStepInfo.newExecutionStepInfo;
+import graphql.execution.ExecutionStepInfoFactory;
+import graphql.execution.FetchedValue;
+import graphql.execution.MergedField;
+import graphql.execution.nextgen.ValueFetcher;
+import graphql.language.Field;
+import graphql.language.Node;
+import graphql.schema.GraphQLOutputType;
+import graphql.schema.GraphQLType;
+import graphql.schema.GraphQLTypeUtil;
+import graphql.util.DependenciesIterator;
+import graphql.util.Edge;
+import graphql.util.TriFunction;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.Iterator;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Objects;
+import java.util.concurrent.CompletableFuture;
+import java.util.stream.Collectors;
+import java.util.stream.Stream;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ *
+ * @author gkesler
+ */
+public class DAGExecutionStrategy implements ExecutionStrategy {
+    public DAGExecutionStrategy (ExecutionContext executionContext) {
+        this.executionContext = Objects.requireNonNull(executionContext);
+        this.executionInfoFactory = new ExecutionStepInfoFactory();
+        this.valueFetcher = new ValueFetcher();
+    }
+
+    /**
+     * Executes a graphql request according to the schedule
+     * provided by executionPlan
+     * 
+     * @param executionPlan a {@code graphql.util.DependencyGraph} specialization that provides
+     * order of field resolution requests
+     * 
+     * @return a CompletableFuture holding the result of execution.
+     */
+    @Override
+    public CompletableFuture<ExecutionResult> execute(ExecutionPlan executionPlan) {
+        assertNotNull(executionPlan);
+
+        ExecutionPlanContextImpl executionPlanContext = new ExecutionPlanContextImpl();
+        return CompletableFuture
+            .completedFuture(executionPlan.orderDependencies(executionPlanContext))
+            .thenCompose(this::resolveClosure)
+            .thenApply(noResult -> executionPlanContext.getResult());
+    }
+
+    private CompletableFuture<DependenciesIterator<NodeVertex<Node, GraphQLType>>> resolveClosure (DependenciesIterator<NodeVertex<Node, GraphQLType>> closure) {
+        if (closure.hasNext()) {
+            Collection<NodeVertex<Node, GraphQLType>> nodes = closure.next();
+            List<CompletableFuture<Void>> tasks = new ArrayList<>(nodes.size());
+
+            Iterator<NodeVertex<Node, GraphQLType>> toBeResolved = nodes.iterator();
+            while (toBeResolved.hasNext()) {
+                NodeVertex<Node, GraphQLType> field = toBeResolved.next();
+                toBeResolved.remove();
+
+                tasks.add(
+                    CompletableFuture
+                        .completedFuture(field)
+                        .thenCompose(this::fetchNode)
+                        .thenAccept(closure::close)
+                );
+            }
+
+            // execute fetch&resove asynchronously
+            return Async
+                .each(tasks)
+                .thenApply(noResult -> closure)
+                .thenCompose(this::resolveClosure);
+        } else {
+            return CompletableFuture
+                .completedFuture(closure);
+        }               
+    }
+
+    private void provideSource (NodeVertex<Node, GraphQLType> source, NodeVertex<Node, GraphQLType> sink) {
+        LOGGER.debug("provideSource: source={}, sink={}", source, sink);
+        source.accept(sink, new NodeVertexVisitor<NodeVertex<? extends Node, ? extends GraphQLType>>() {
+            @Override
+            public NodeVertex<? extends Node, ? extends GraphQLType> visit(OperationVertex source, NodeVertex<? extends Node, ? extends GraphQLType> sink) {
+                source
+                    .executionStepInfo(
+                        newExecutionStepInfo()
+                            .type((GraphQLOutputType)source.getType())
+                            .path(ExecutionPath.rootPath())
+                            .build()
+                    )
+                    // prepare placeholder for this operation result
+                    // since operation does not have its external reslution,
+                    // it's result is stored in the source object 
+                    .result(source.getSource());
+
+                return visitNode(source, sink);
+            }
+
+            @Override
+            public NodeVertex<? extends Node, ? extends GraphQLType> visitNode(NodeVertex<? extends Node, ? extends GraphQLType> source, NodeVertex<? extends Node, ? extends GraphQLType> sink) {
+                return sink
+                    .parentExecutionStepInfo(source.getExecutionStepInfo())
+                    .source(Results.flatten((List<Object>)source.getResult()));
+            }
+        });
+    }
+
+    private void joinResults (FieldVertex source, NodeVertex<Node, GraphQLType> sink) {
+        LOGGER.debug("afterResolve: source={}, sink={}", source, sink);
+        Results.joinResultsOf(source);
+    }
+
+    private boolean tryResolve (NodeVertex<Node, GraphQLType> node) {
+        LOGGER.debug("tryResolve: node={}", node);
+        return node.accept(true, new NodeVertexVisitor<Boolean>() {
+            @Override
+            public Boolean visit(FieldVertex node, Boolean data) {
+                List<Object> sources = (List<Object>)node.getSource();
+                // if sources is empty, no need to fetch data.
+                // even if it is fetched, it won't be joined anyways
+                return sources == null || sources.isEmpty();
+            }
+
+            @Override
+            public Boolean visit(DocumentVertex node, Boolean data) {
+                node.result(result);
+                return true;
+            }            
+        });
+    }
+
+    private CompletableFuture<NodeVertex<Node, GraphQLType>> fetchNode (NodeVertex<Node, GraphQLType> node) {  
+        LOGGER.debug("fetchNode: node={}", node);
+
+        FieldVertex fieldNode = node.as(FieldVertex.class);
+        MergedField sameFields = MergedField
+                .newMergedField(fieldNode.getNode())
+                .build();
+        ExecutionStepInfo sourceExecutionStepInfo = node.getParentExecutionStepInfo();
+        GraphQLOutputType parentType = (GraphQLOutputType)GraphQLTypeUtil.unwrapAll(sourceExecutionStepInfo.getType());
+        ExecutionStepInfo parentExecutionStepInfo = sourceExecutionStepInfo
+            .transform(builder -> builder
+                    .parentInfo(sourceExecutionStepInfo.getParent())
+                    .type(parentType));
+        ExecutionStepInfo executionStepInfo = executionInfoFactory
+            .newExecutionStepInfoForSubField(executionContext, sameFields, parentExecutionStepInfo);
+
+        TriFunction<FieldVertex, MergedField, ExecutionStepInfo, CompletableFuture<List<FetchedValue>>> valuesFetcher = 
+            fieldNode.isRoot() ? this::fetchRootValues : this::fetchBatchedValues;
+
+
+        return valuesFetcher.apply(fieldNode, sameFields, executionStepInfo)
+                .thenApply(fetchedValues -> 
+                    fetchedValues
+                        .stream()
+                        .map(FetchedValue::getFetchedValue)
+                        .map(fv -> Results.checkAndFixNILs(fv, fieldNode))
+                        .collect(Collectors.toList())
+                )
+                .thenApply(fetchedValues -> 
+                    (NodeVertex<Node, GraphQLType>)node
+                        .executionStepInfo(executionStepInfo)
+                        .result(fetchedValues)
+                );
+    }
+
+    private CompletableFuture<List<FetchedValue>> fetchRootValues (FieldVertex fieldNode, MergedField sameFields, ExecutionStepInfo executionStepInfo) {
+        return valueFetcher
+            .fetchValue(executionContext, executionContext.getRoot(), null/*toDoLocalContext*/, sameFields, executionStepInfo)
+            .thenApply(Collections::singletonList);
+    }
+
+    private CompletableFuture<List<FetchedValue>> fetchBatchedValues (FieldVertex fieldNode, MergedField sameFields, ExecutionStepInfo executionStepInfo) {
+        List<Object> sources = (List<Object>)fieldNode.getSource();
+        return valueFetcher
+            .fetchBatchedValues(executionContext, sources, sameFields, 
+                Stream
+                    .generate(() -> executionStepInfo)
+                    .limit(sources.size())
+                    .collect(Collectors.toList())
+            );
+    }
+
+    private class ExecutionPlanContextImpl implements ExecutionPlanContext {
+        @Override
+        public void prepareResolve(Edge<? extends NodeVertex<? extends Node, ? extends GraphQLType>, ?> edge) {
+            provideSource((NodeVertex<Node, GraphQLType>)edge.getSource(), (NodeVertex<Node, GraphQLType>)edge.getSink());
+        }
+
+        @Override
+        public void whenResolved(Edge<? extends NodeVertex<? extends Node, ? extends GraphQLType>, ?> edge) {
+            joinResults((FieldVertex)edge.getSource(), (NodeVertex<Node, GraphQLType>)edge.getSink());
+        }
+
+        @Override
+        public boolean resolve(NodeVertex<? extends Node, ? extends GraphQLType> node) {
+            return tryResolve((NodeVertex<Node, GraphQLType>)node);
+        }
+
+        public ExecutionResult getResult() {
+            return new ExecutionResultImpl(result.get(0), executionContext.getErrors());
+        }
+    };
+
+    private final ExecutionContext executionContext;
+    private final ExecutionStepInfoFactory executionInfoFactory;
+    private final ValueFetcher valueFetcher;
+    private final List<Object> result = Arrays.asList(new LinkedHashMap<String, Object>());
+
+    private static final Logger LOGGER = LoggerFactory.getLogger(DAGExecutionStrategy.class);
+}
diff --git a/src/main/java/graphql/execution3/DocumentVertex.java b/src/main/java/graphql/execution3/DocumentVertex.java
@@ -0,0 +1,45 @@
+/*
+ * To change this license header, choose License Headers in Project Properties.
+ * To change this template file, choose Tools | Templates
+ * and open the template in the editor.
+ */
+package graphql.execution3;
+
+import graphql.execution.ExecutionStepInfo;
+import graphql.language.Document;
+import graphql.schema.GraphQLType;
+import java.util.Objects;
+
+/**
+ * ExecutionPlan Vertex created around Document node
+ */
+public class DocumentVertex extends NodeVertex<Document, GraphQLType> {
+    public DocumentVertex(Document node) {
+        super(Objects.requireNonNull(node), null);
+    }
+
+    @Override
+    public DocumentVertex parentExecutionStepInfo(ExecutionStepInfo parentExecutionStepInfo) {
+        return (DocumentVertex)super.parentExecutionStepInfo(parentExecutionStepInfo);
+    }
+
+    @Override
+    public DocumentVertex executionStepInfo(ExecutionStepInfo value) {
+        return (DocumentVertex)super.executionStepInfo(value);
+    }
+
+    @Override
+    public DocumentVertex source(Object source) {
+        return (DocumentVertex)super.source(source);
+    }
+
+    @Override
+    public DocumentVertex result(Object result) {
+        return (DocumentVertex)super.result(result);
+    }
+
+    @Override
+    <U> U accept(U data, NodeVertexVisitor<? super U> visitor) {
+        return (U)visitor.visit(this, data);
+    }    
+}