diff --git a/pom.xml b/pom.xml
index 8f9b227e..186bde02 100644
--- a/pom.xml
+++ b/pom.xml
@@ -2,7 +2,7 @@
4.0.0
com.intuit.graphql
graphql-orchestrator-java
- 5.0.19-SNAPSHOT
+ 5.0.20-DEFER-SNAPSHOT
jar
graphql-orchestrator-java
GraphQL Orchestrator combines multiple graphql services into a single unified schema
@@ -39,6 +39,7 @@
3.0.1
2.26.0
17.4
+ 3.4.24
@@ -159,6 +160,17 @@
jackson-databind
2.13.4.1
+
+ io.projectreactor
+ reactor-core
+ ${reactorVersion}
+
+
+ io.projectreactor
+ reactor-test
+ ${reactorVersion}
+ test
+
diff --git a/src/main/java/com/intuit/graphql/orchestrator/GraphQLOrchestrator.java b/src/main/java/com/intuit/graphql/orchestrator/GraphQLOrchestrator.java
index 4812344b..b0b71ab8 100644
--- a/src/main/java/com/intuit/graphql/orchestrator/GraphQLOrchestrator.java
+++ b/src/main/java/com/intuit/graphql/orchestrator/GraphQLOrchestrator.java
@@ -1,8 +1,12 @@
package com.intuit.graphql.orchestrator;
+import com.intuit.graphql.orchestrator.deferDirective.DeferDirectiveInstrumentation;
+import com.intuit.graphql.orchestrator.deferDirective.DeferOptions;
import com.intuit.graphql.orchestrator.schema.RuntimeGraph;
+import com.intuit.graphql.orchestrator.utils.MultiEIGenerator;
import graphql.ExecutionInput;
import graphql.ExecutionResult;
+import graphql.ExecutionResultImpl;
import graphql.GraphQL;
import graphql.GraphQLContext;
import graphql.execution.AsyncExecutionStrategy;
@@ -11,11 +15,14 @@
import graphql.execution.instrumentation.ChainedInstrumentation;
import graphql.execution.instrumentation.Instrumentation;
import graphql.execution.instrumentation.dataloader.DataLoaderDispatcherInstrumentation;
+import graphql.execution.reactive.SubscriptionPublisher;
import graphql.schema.GraphQLSchema;
import lombok.extern.slf4j.Slf4j;
import org.dataloader.BatchLoader;
import org.dataloader.DataLoader;
import org.dataloader.DataLoaderRegistry;
+import reactor.core.publisher.Flux;
+import reactor.core.scheduler.Schedulers;
import java.util.Arrays;
import java.util.LinkedList;
@@ -23,16 +30,20 @@
import java.util.Map;
import java.util.Objects;
import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.atomic.AtomicInteger;
import java.util.function.Function;
import java.util.function.UnaryOperator;
import java.util.stream.Collectors;
+import static com.intuit.graphql.orchestrator.utils.DirectivesUtil.USE_DEFER;
import static java.util.Objects.requireNonNull;
@Slf4j
public class GraphQLOrchestrator {
public static final String DATA_LOADER_REGISTRY_CONTEXT_KEY = DataLoaderRegistry.class.getName() + ".context.key";
+ private static final DeferOptions DEFAULT_DEFER_OPTIONS = DeferOptions.builder().nestedDefersAllowed(false).build();
+ private static final boolean DISABLED_DEFER = false;
private final RuntimeGraph runtimeGraph;
private final List instrumentations;
@@ -41,8 +52,8 @@ public class GraphQLOrchestrator {
private final ExecutionStrategy mutationExecutionStrategy;
private GraphQLOrchestrator(final RuntimeGraph runtimeGraph, final List instrumentations,
- final ExecutionIdProvider executionIdProvider, final ExecutionStrategy queryExecutionStrategy,
- final ExecutionStrategy mutationExecutionStrategy) {
+ final ExecutionIdProvider executionIdProvider, final ExecutionStrategy queryExecutionStrategy,
+ final ExecutionStrategy mutationExecutionStrategy) {
this.runtimeGraph = runtimeGraph;
this.instrumentations = instrumentations;
this.executionIdProvider = executionIdProvider;
@@ -63,16 +74,80 @@ private DataLoaderRegistry buildNewDataLoaderRegistry() {
// to create a new DataLoader per request. Else it will use the cache which is shared
// across request.
final Map temporaryMap = this.runtimeGraph.getBatchLoaderMap().values().stream().distinct()
- .collect(Collectors.toMap(Function.identity(), DataLoader::new));
+ .collect(Collectors.toMap(Function.identity(), DataLoader::new));
this.runtimeGraph.getBatchLoaderMap()
- .forEach((key, batchLoader) ->
- dataLoaderRegistry.register(key, temporaryMap.getOrDefault(batchLoader, new DataLoader(batchLoader))));
+ .forEach((key, batchLoader) ->
+ dataLoaderRegistry.register(key, temporaryMap.getOrDefault(batchLoader, new DataLoader(batchLoader))));
return dataLoaderRegistry;
}
public CompletableFuture execute(ExecutionInput executionInput) {
+ return execute(executionInput, DEFAULT_DEFER_OPTIONS, DISABLED_DEFER);
+ }
+
+ public CompletableFuture execute(ExecutionInput executionInput, DeferOptions deferOptions, boolean hasDefer) {
+ if(hasDefer) {
+ return executeWithDefer(executionInput, deferOptions);
+ }
+ final GraphQL graphQL = constructGraphQL();
+
+ final ExecutionInput newExecutionInput = executionInput
+ .transform(builder -> builder.dataLoaderRegistry(buildNewDataLoaderRegistry()));
+
+ if (newExecutionInput.getContext() instanceof GraphQLContext) {
+ ((GraphQLContext) newExecutionInput.getContext())
+ .put(DATA_LOADER_REGISTRY_CONTEXT_KEY, newExecutionInput.getDataLoaderRegistry());
+ ((GraphQLContext) newExecutionInput.getContext())
+ .put(USE_DEFER , false);
+ }
+ return graphQL.executeAsync(newExecutionInput);
+ }
+ private CompletableFuture executeWithDefer(ExecutionInput executionInput, DeferOptions options) {
+ AtomicInteger responses = new AtomicInteger(0);
+ MultiEIGenerator eiGenerator = new MultiEIGenerator(executionInput, options, this.getSchema());
+
+ Flux