Async Execution Strategy #242
There are no files selected for viewing
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,55 @@ | ||
| package graphql; | ||
|
|
||
| import graphql.execution.ExecutionStrategy; | ||
| import graphql.schema.GraphQLSchema; | ||
|
|
||
| import java.util.Collections; | ||
| import java.util.Map; | ||
| import java.util.concurrent.CompletionStage; | ||
|
|
||
| import static java.util.concurrent.CompletableFuture.completedFuture; | ||
|
|
||
| public class GraphQLAsync extends GraphQL { | ||
|
|
||
| public GraphQLAsync(GraphQLSchema graphQLSchema) { | ||
| super(graphQLSchema); | ||
| } | ||
|
|
||
| public GraphQLAsync(GraphQLSchema graphQLSchema, ExecutionStrategy queryStrategy) { | ||
| super(graphQLSchema, queryStrategy); | ||
| } | ||
|
|
||
| public GraphQLAsync(GraphQLSchema graphQLSchema, ExecutionStrategy queryStrategy, ExecutionStrategy mutationStrategy) { | ||
| super(graphQLSchema, queryStrategy, mutationStrategy); | ||
| } | ||
|
|
||
| public CompletionStage<ExecutionResult> executeAsync(String requestString) { | ||
| return executeAsync(requestString, null); | ||
|
|
||
| } | ||
|
|
||
| public CompletionStage<ExecutionResult> executeAsync(String requestString, Object context) { | ||
| return executeAsync(requestString, context, Collections.emptyMap()); | ||
|
|
||
| } | ||
|
|
||
| public CompletionStage<ExecutionResult> executeAsync(String requestString, String operationName, Object context) { | ||
| return executeAsync(requestString, operationName, context, Collections.emptyMap()); | ||
|
|
||
| } | ||
|
|
||
| public CompletionStage<ExecutionResult> executeAsync(String requestString, Object context, Map<String, Object> arguments) { | ||
| return executeAsync(requestString, null, context, arguments); | ||
|
|
||
| } | ||
|
|
||
| @SuppressWarnings("unchecked") | ||
| public CompletionStage<ExecutionResult> executeAsync(String requestString, String operationName, Object context, Map<String, Object> arguments) { | ||
| ExecutionResult result = execute(requestString, operationName, context, arguments); | ||
| Object data1 = result.getData(); | ||
| if (data1 instanceof CompletionStage) { | ||
| return ((CompletionStage<Map<String, Object>>) data1).thenApply(data -> new ExecutionResultImpl(data, result.getErrors())); | ||
| } | ||
| return completedFuture(result); | ||
| } | ||
| } |
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,131 @@ | ||
| package graphql.execution.async; | ||
|
|
||
| import graphql.ExceptionWhileDataFetching; | ||
| import graphql.ExecutionResult; | ||
| import graphql.GraphQLError; | ||
| import graphql.execution.ExecutionContext; | ||
| import graphql.execution.ExecutionStrategy; | ||
| import graphql.language.Field; | ||
| import graphql.schema.*; | ||
| import org.slf4j.Logger; | ||
| import org.slf4j.LoggerFactory; | ||
|
|
||
| import java.util.ArrayList; | ||
| import java.util.List; | ||
| import java.util.Map; | ||
| import java.util.concurrent.CompletableFuture; | ||
| import java.util.concurrent.CompletionStage; | ||
| import java.util.function.Supplier; | ||
|
|
||
| import static com.spotify.futures.CompletableFutures.successfulAsList; | ||
|
Member
There was a problem hiding this comment. do you really want Spotify as a dependency. I thought graphql-java was trying to be as lightweight as possible
Contributor
Author
There was a problem hiding this comment. That's just in there while I work this out. That artifact is tiny (50 LOC?) and can be copy/pasted into this project license-permitting.
Contributor
Author
There was a problem hiding this comment. Thanks for taking a look! |
||
| import static java.util.concurrent.CompletableFuture.completedFuture; | ||
| import static java.util.stream.Collectors.toList; | ||
| import static java.util.stream.Collectors.toMap; | ||
|
|
||
|
|
||
| public final class AsyncExecutionStrategy extends ExecutionStrategy { | ||
|
|
||
| public static AsyncExecutionStrategy serial() { | ||
| return new AsyncExecutionStrategy(true); | ||
| } | ||
|
|
||
| public static AsyncExecutionStrategy parallel() { | ||
| return new AsyncExecutionStrategy(false); | ||
| } | ||
|
|
||
| private static final Logger log = LoggerFactory.getLogger(AsyncExecutionStrategy.class); | ||
|
|
||
| private final boolean serial; | ||
|
|
||
| private AsyncExecutionStrategy(boolean serial) { | ||
| this.serial = serial; | ||
| } | ||
|
|
||
| @Override | ||
| public ExecutionResult execute(ExecutionContext executionContext, GraphQLObjectType parentType, Object source, Map<String, List<Field>> fields) { | ||
| Map<String, Supplier<CompletionStage<ExecutionResult>>> fieldResolvers = fields.entrySet() | ||
| .stream() | ||
| .collect(toMap(Map.Entry::getKey, entry -> () -> resolveFieldAsync(executionContext, parentType, source, entry.getValue()))); | ||
|
|
||
| AsyncFieldsCoordinator coordinator = new AsyncFieldsCoordinator(fieldResolvers); | ||
|
|
||
| CompletionStage<Map<String, Object>> data = serial ? coordinator.executeSerially() : coordinator.executeParallelly(); | ||
| return new ExecutionResultImpl(data, executionContext.getErrors()); | ||
| } | ||
|
|
||
| private CompletionStage<ExecutionResult> resolveFieldAsync(ExecutionContext executionContext, GraphQLObjectType parentType, Object source, List<Field> fieldList) { | ||
| GraphQLFieldDefinition fieldDef = getFieldDef(executionContext.getGraphQLSchema(), parentType, fieldList.get(0)); | ||
| GraphQLOutputType fieldType = fieldDef.getType(); | ||
|
|
||
| DataFetchingEnvironment env = new DataFetchingEnvironment( | ||
| source, | ||
| valuesResolver.getArgumentValues(fieldDef.getArguments(), fieldList.get(0).getArguments(), executionContext.getVariables()), | ||
| executionContext.getRoot(), | ||
| fieldList, | ||
| fieldDef.getType(), | ||
| parentType, | ||
| executionContext.getGraphQLSchema() | ||
| ); | ||
|
|
||
| try { | ||
| Object obj1 = fieldDef.getDataFetcher().get(env); | ||
| if (obj1 instanceof CompletionStage) { | ||
| return ((CompletionStage<?>) obj1) | ||
| .exceptionally(e -> { | ||
| logExceptionWhileFetching(e, fieldList.get(0)); | ||
| executionContext.addError(new ExceptionWhileDataFetching(e)); | ||
| return null; | ||
| }) | ||
| .thenCompose(obj2 -> completeValueAsync(executionContext, fieldType, fieldList, obj2)); | ||
| } else { | ||
| return completeValueAsync(executionContext, fieldType, fieldList, obj1); | ||
| } | ||
| } catch (Exception e) { | ||
| logExceptionWhileFetching(e, fieldList.get(0)); | ||
| executionContext.addError(new ExceptionWhileDataFetching(e)); | ||
| return completedFuture(new ExecutionResultImpl(null, null)); | ||
| } | ||
| } | ||
|
|
||
| @SuppressWarnings("unchecked") | ||
| private CompletionStage<ExecutionResult> completeValueAsync(ExecutionContext executionContext, GraphQLType fieldType, List<Field> fieldList, Object result) { | ||
| if (fieldType instanceof GraphQLList) { | ||
| List<Object> collect = ((List<Object>) result); | ||
|
|
||
| List<CompletionStage<ExecutionResult>> collect1 = collect.stream() | ||
| .map(object -> { | ||
| CompletionStage<Object> stage = object instanceof CompletionStage ? (CompletionStage<Object>) object : completedFuture(object); | ||
| return stage | ||
| .thenCompose(result5 -> completeValueAsync(executionContext, ((GraphQLList) fieldType).getWrappedType(), fieldList, result5)); | ||
| }) | ||
| .collect(toList()); | ||
|
|
||
| return successfulAsList(collect1, t -> null).thenApply(results -> { | ||
| List<Object> list = new ArrayList<>(); | ||
| List<GraphQLError> errors = new ArrayList<>(); | ||
| for (ExecutionResult executionResult : results) { | ||
| list.add(executionResult.getData()); | ||
| errors.addAll(executionResult.getErrors()); | ||
| } | ||
| return new ExecutionResultImpl(list, errors); | ||
| }); | ||
|
|
||
| } else { | ||
| ExecutionResult completed = completeValue(executionContext, fieldType, fieldList, result); | ||
| // Happens when the data fetcher returns null for nullable field | ||
| if (completed == null) { | ||
| return completedFuture(new ExecutionResultImpl(null, null)); | ||
| } | ||
| if (!(completed.getData() instanceof CompletionStage)) { | ||
| return completedFuture(completed); | ||
| } | ||
| return ((CompletionStage<?>) completed.getData()) | ||
| .thenApply(data -> new ExecutionResultImpl(data, completed.getErrors())); | ||
| } | ||
| } | ||
|
|
||
| private void logExceptionWhileFetching(Throwable e, Field field) { | ||
| log.debug("Exception while fetching data for field {}", field.getName(), e); | ||
| } | ||
|
|
||
| } | ||
Oops, something went wrong.
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
The GraphQL spec says that mutations WILL ALWAYS be serial. So why allow people to wire in anything BUT a serial implementation?
I guess it makes it more configureable than today but you have no way of knowing that they are following spec or not.
Maybe its a case of "dont shoot yourself" by wiring in a asynch mutation execution strategy
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
My serial implementation is async. The current serial implementation is not async. Both are serial though.