Skip to content

Preserve DataFetcherResult extensions on subscription events #4476

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
kalayciburak wants to merge 1 commit into
base: master
Choose a base branch
from
Open

Preserve DataFetcherResult extensions on subscription events #4476

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
14 changes: 12 additions & 2 deletions src/main/java/graphql/execution/SubscriptionExecutionStrategy.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -22,6 +22,7 @@
import org.reactivestreams.FlowAdapters;
import org.reactivestreams.Publisher;

import java.util.Map;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.CompletionStage;
import java.util.concurrent.Flow;
Expand Down Expand Up @@ -184,12 +185,13 @@ private CompletableFuture<ExecutionResult> executeSubscriptionEvent(ExecutionCon
newParameters.getAlternativeCallContext(),
"alternativeCallContext must not be null");
executionContext.getDataLoaderDispatcherStrategy().newSubscriptionExecution(alternativeCallContext);
Map<Object, Object> eventExtensions = extensionsFromEventPayload(eventPayload);
Object fetchedValue = unboxPossibleDataFetcherResult(newExecutionContext, newParameters, eventPayload);
FieldValueInfo fieldValueInfo = completeField(newExecutionContext, newParameters, fetchedValue);
executionContext.getDataLoaderDispatcherStrategy().subscriptionEventCompletionDone(alternativeCallContext);
CompletableFuture<ExecutionResult> overallResult = fieldValueInfo
.getFieldValueFuture()
.thenApply(val -> new ExecutionResultImpl(val, alternativeCallContext.getErrors()))
.thenApply(val -> new ExecutionResultImpl(val, alternativeCallContext.getErrors(), eventExtensions))
.thenApply(executionResult -> wrapWithRootFieldName(newParameters, executionResult))
.whenComplete((executionResult, throwable) -> {
executionContext.getDataLoaderDispatcherStrategy().subscriptionEventExecutionDone(alternativeCallContext);
Expand All @@ -211,10 +213,18 @@ private ExecutionResult wrapWithRootFieldName(ExecutionStrategyParameters parame
String rootFieldName = getRootFieldName(parameters);
return new ExecutionResultImpl(
singletonMap(rootFieldName, executionResult.getData()),
executionResult.getErrors()
executionResult.getErrors(),
executionResult.getExtensions()
);
}

private static @Nullable Map<Object, Object> extensionsFromEventPayload(Object eventPayload) {
if (eventPayload instanceof DataFetcherResult) {
return ((DataFetcherResult<?>) eventPayload).getExtensions();
}
return null;
}

private String getRootFieldName(ExecutionStrategyParameters parameters) {
Field rootField = assertNotNull(parameters.getField(), "field must not be null").getSingleField();
return rootField.getResultKey();
Expand Down
45 changes: 45 additions & 0 deletions src/test/groovy/graphql/execution/SubscriptionExecutionStrategyTest.groovy
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -604,6 +604,51 @@ class SubscriptionExecutionStrategyTest extends Specification {
}
}

def "subscription event DataFetcherResult extensions are included on the emitted ExecutionResult"() {

DataFetcher newMessageDF = new DataFetcher() {
@Override
Object get(DataFetchingEnvironment environment) {
def objectMaker = { int index ->
def message = new Message("sender" + index, "text" + index)
return DataFetcherResult.newResult()
.data(message)
.extensions(["eventKey": "event-" + index])
.build()
}
return new ReactiveStreamsObjectPublisher(3, objectMaker)
}
}

GraphQL graphQL = buildSubscriptionQL(newMessageDF)

def executionInput = ExecutionInput.newExecutionInput().query("""
subscription NewMessages {
newMessage(roomId: 123) {
sender
text
}
}
""").build()

when:
def executionResult = graphQL.execute(executionInput)
Publisher<ExecutionResult> msgStream = executionResult.getData()
def capturingSubscriber = new CapturingSubscriber<ExecutionResult>()
msgStream.subscribe(capturingSubscriber)

then:
Awaitility.await().untilTrue(capturingSubscriber.isDone())

def messages = capturingSubscriber.events
messages.size() == 3
for (int i = 0; i < messages.size(); i++) {
def message = messages[i]
assert message.data == ["newMessage": [sender: "sender" + i, text: "text" + i]]
assert message.extensions == [eventKey: "event-" + i]
}
}

def "instrumentation gets called on subscriptions"() {

//
Expand Down
Loading