Spring/spring-graphql
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Update to latest GraphQL over WebSocket protocol

Browse Source
This commit is contained in:
Rossen Stoyanchev
2021-01-03 21:24:55 +00:00
parent bb9d1dc300
commit 2e32a06a55
16 changed files with 858 additions and 164 deletions
Download Patch File Download Diff File Expand all files Collapse all files

20
samples/webflux-websocket/build.gradle Normal file
View File

@@ -0,0 +1,20 @@
plugins {
id 'org.springframework.boot' version '2.4.0-SNAPSHOT'
id 'io.spring.dependency-management' version '1.0.10.RELEASE'
id 'java'
}
group = 'com.example'
version = '0.0.1-SNAPSHOT'
description = "GraphQL over WebSocket sample"
sourceCompatibility = '1.8'
dependencies {
implementation project(':spring-graphql-web')
implementation 'org.springframework.boot:spring-boot-starter-webflux'
implementation 'org.springframework.boot:spring-boot-starter-actuator'
developmentOnly 'org.springframework.boot:spring-boot-devtools'
testImplementation 'org.springframework.boot:spring-boot-starter-test'
}
test {
useJUnitPlatform()
}

28
samples/webflux-websocket/src/main/java/io/spring/sample/graphql/SampleApplication.java Normal file
View File

@@ -0,0 +1,28 @@
/*
* Copyright 2002-2020 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* https://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package io.spring.sample.graphql;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
@SpringBootApplication
public class SampleApplication {
public static void main(String[] args) {
SpringApplication.run(SampleApplication.class, args);
}
}

23
samples/webflux-websocket/src/main/java/io/spring/sample/graphql/SampleWiring.java Normal file
View File

@@ -0,0 +1,23 @@
package io.spring.sample.graphql;
import java.time.Duration;
import graphql.schema.idl.RuntimeWiring;
import reactor.core.publisher.Flux;
import org.springframework.boot.graphql.RuntimeWiringCustomizer;
import org.springframework.stereotype.Component;
@Component
public class SampleWiring implements RuntimeWiringCustomizer {
@Override
public void customize(RuntimeWiring.Builder builder) {
builder.type("Query", wiringBuilder -> wiringBuilder.dataFetcher("hello",
env -> "Hello world!"));
builder.type("Subscription", wiringBuilder -> wiringBuilder.dataFetcher("greetings",
env -> Flux.just("Hi", "Bonjour", "Hola", "Cio", "Zdravo")
.delayElements(Duration.ofMillis(500))));
}
}

5
samples/webflux-websocket/src/main/resources/application.properties Normal file
View File

@@ -0,0 +1,5 @@
management.endpoints.web.exposure.include=health,metrics,info
logging.level.org.springframework.web=debug
logging.level.org.springframework.http=debug
logging.level.org.springframework.graphql=debug
logging.level.reactor.netty=debug

11
samples/webflux-websocket/src/main/resources/schema.graphqls Normal file
View File

@@ -0,0 +1,11 @@
schema {
query: Query
subscription : Subscription
}
type Query {
hello: String
}
type Subscription {
greetings: String
}

55
samples/webflux-websocket/src/main/resources/static/index.html Normal file
View File

@@ -0,0 +1,55 @@
<!DOCTYPE html>
<html>
<head>
<meta charset="utf-8" />
<title>GraphQL over WebSocket</title>
<script type="text/javascript" src="https://unpkg.com/graphql-ws/umd/graphql-ws.js"></script>
</head>
<body>
<script type="text/javascript">
const client = graphqlWs.createClient({
url: 'ws://localhost:8080/graphql/websocket',
});
// query
(async () => {
const result = await new Promise((resolve, reject) => {
let result;
client.subscribe(
{
query: '{ hello }',
},
{
next: (data) => (result = data),
error: reject,
complete: () => resolve(result),
},
);
});
console.log("Query result: " + result);
})();
// subscription
(async () => {
const onNext = () => {
console.log("Subscription data")
};
await new Promise((resolve, reject) => {
client.subscribe(
{
query: 'subscription { greetings }',
},
{
next: onNext,
error: reject,
complete: resolve,
},
);
});
})();
</script>
</body>
</html>

1
settings.gradle
View File

@@ -16,3 +16,4 @@ pluginManagement {
rootProject.name = 'spring-graphql'
include 'spring-graphql-web'
include 'samples:webmvc-http-endpoint'
include 'samples:webflux-websocket'

16
spring-graphql-web/src/main/java/org/springframework/boot/graphql/GraphQLProperties.java
View File

@@ -15,6 +15,8 @@
*/
package org.springframework.boot.graphql;
import java.time.Duration;
import org.springframework.boot.context.properties.ConfigurationProperties;
@ConfigurationProperties(prefix = "spring.graphql")
@@ -35,6 +37,13 @@ public class GraphQLProperties {
*/
private String webSocketPath = path + "/websocket";
/**
* For the GraphQL over WebSocket endpoint, this is time within which the
* initial {@code CONNECTION_INIT} type message must be received.
*/
private Duration connectionInitTimeoutDuration = Duration.ofSeconds(60);
public String getPath() {
return path;
}
@@ -59,4 +68,11 @@ public class GraphQLProperties {
this.schemaLocation = schemaLocation;
}
public Duration getConnectionInitTimeoutDuration() {
return this.connectionInitTimeoutDuration;
}
public void setConnectionInitTimeoutDuration(Duration connectionInitTimeoutDuration) {
this.connectionInitTimeoutDuration = connectionInitTimeoutDuration;
}
}

7
spring-graphql-web/src/main/java/org/springframework/boot/graphql/WebFluxGraphQLAutoConfiguration.java
View File

@@ -57,9 +57,12 @@ public class WebFluxGraphQLAutoConfiguration {
@Bean
@ConditionalOnMissingBean
public WebFluxGraphQLWebSocketHandler graphQLWebSocketHandler(
GraphQL.Builder graphQLBuilder, ServerCodecConfigurer configurer) {
GraphQL.Builder graphQLBuilder, GraphQLProperties properties, ServerCodecConfigurer configurer) {
return new WebFluxGraphQLWebSocketHandler(graphQLBuilder.build(), Collections.emptyList(), configurer);
return new WebFluxGraphQLWebSocketHandler(
graphQLBuilder.build(), Collections.emptyList(),
configurer, properties.getConnectionInitTimeoutDuration()
);
}
@Bean

269
spring-graphql-web/src/main/java/org/springframework/graphql/WebFluxGraphQLWebSocketHandler.java
View File

@@ -15,18 +15,25 @@
*/
package org.springframework.graphql;
import java.time.Duration;
import java.util.Collections;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicBoolean;
import graphql.ErrorType;
import graphql.ExecutionResult;
import graphql.GraphQL;
import graphql.GraphqlErrorBuilder;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.reactivestreams.Publisher;
import org.reactivestreams.Subscription;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import org.springframework.core.ResolvableType;
import org.springframework.core.codec.Decoder;
import org.springframework.core.codec.Encoder;
import org.springframework.core.io.buffer.DataBuffer;
@@ -35,8 +42,11 @@ import org.springframework.http.MediaType;
import org.springframework.http.codec.DecoderHttpMessageReader;
import org.springframework.http.codec.EncoderHttpMessageWriter;
import org.springframework.http.codec.ServerCodecConfigurer;
import org.springframework.lang.Nullable;
import org.springframework.util.Assert;
import org.springframework.util.CollectionUtils;
import org.springframework.util.MimeTypeUtils;
import org.springframework.web.reactive.socket.CloseStatus;
import org.springframework.web.reactive.socket.HandshakeInfo;
import org.springframework.web.reactive.socket.WebSocketHandler;
import org.springframework.web.reactive.socket.WebSocketMessage;
@@ -50,33 +60,45 @@ public class WebFluxGraphQLWebSocketHandler implements WebSocketHandler {
private static final Log logger = LogFactory.getLog(WebFluxGraphQLWebSocketHandler.class);
private static final ResolvableType MAP_TYPE = ResolvableType.forClass(Map.class);
private static final List<String> SUB_PROTOCOL_LIST = Collections.singletonList("graphql-transport-ws");
private final WebInterceptorExecutionChain executionChain;
private final Decoder<?> jsonDecoder;
private final Duration initTimeoutDuration;
private final Encoder<?> jsonEncoder;
private final Decoder<?> decoder;
private final Encoder<?> encoder;
private final Map<String, Subscription> subscriptions = new ConcurrentHashMap<>();
@Override
public List<String> getSubProtocols() {
return SUB_PROTOCOL_LIST;
}
/**
* Create a new instance.
* @param graphQL the GraphQL instance to use for query execution
* @param interceptors 0 or more interceptors to customize input and output
* @param configurer codec configurer for JSON encoding and decoding
* @param initTimeoutDuration the time within which the
* {@code CONNECTION_INIT} type message must be received.
*/
public WebFluxGraphQLWebSocketHandler(GraphQL graphQL, List<WebInterceptor> interceptors,
ServerCodecConfigurer configurer) {
ServerCodecConfigurer configurer, Duration initTimeoutDuration) {
this.executionChain = new WebInterceptorExecutionChain(graphQL, interceptors);
this.jsonDecoder = initDecoder(configurer);
this.jsonEncoder = initEncoder(configurer);
this.initTimeoutDuration = initTimeoutDuration;
this.decoder = initDecoder(configurer);
this.encoder = initEncoder(configurer);
}
private static Decoder<?> initDecoder(ServerCodecConfigurer configurer) {
return configurer.getReaders().stream()
.filter(reader -> reader.canRead(MAP_TYPE, MediaType.APPLICATION_JSON))
.filter(reader -> reader.canRead(WebInput.MAP_RESOLVABLE_TYPE, MediaType.APPLICATION_JSON))
.map(reader -> ((DecoderHttpMessageReader<?>) reader).getDecoder())
.findFirst()
.orElseThrow(() -> new IllegalArgumentException("No JSON Decoder"));
@@ -84,7 +106,7 @@ public class WebFluxGraphQLWebSocketHandler implements WebSocketHandler {
private static Encoder<?> initEncoder(ServerCodecConfigurer configurer) {
return configurer.getWriters().stream()
.filter(writer -> writer.canWrite(MAP_TYPE, MediaType.APPLICATION_JSON))
.filter(writer -> writer.canWrite(WebInput.MAP_RESOLVABLE_TYPE, MediaType.APPLICATION_JSON))
.map(writer -> ((EncoderHttpMessageWriter<?>) writer).getEncoder())
.findFirst()
.orElseThrow(() -> new IllegalArgumentException("No JSON Encoder"));
@@ -92,54 +114,211 @@ public class WebFluxGraphQLWebSocketHandler implements WebSocketHandler {
@Override
@SuppressWarnings("unchecked")
public Mono<Void> handle(WebSocketSession session) {
return session.send(session.receive()
.concatMap(message -> {
AtomicBoolean initialized = new AtomicBoolean();
Mono.delay(this.initTimeoutDuration)
.then(Mono.defer(() ->
initialized.compareAndSet(false, true) ?
GraphQLStatus.initTimeout(session) :
Mono.empty()))
.subscribe();
Flux<WebSocketMessage> responseFlux = session.receive()
.flatMap(message -> {
Map<String, Object> map = decode(message);
HandshakeInfo handshakeInfo = session.getHandshakeInfo();
WebInput webInput = new WebInput(handshakeInfo.getUri(), handshakeInfo.getHeaders(), map);
if (logger.isDebugEnabled()) {
logger.debug("Executing: " + webInput);
String id = (String) map.get("id");
MessageType messageType = MessageType.resolve((String) map.get("type"));
if (id == null || messageType == null) {
return GraphQLStatus.invalidMessage(session);
}
return executionChain.execute(webInput);
})
.concatMap(output -> {
if (!CollectionUtils.isEmpty(output.getErrors())) {
throw new IllegalStateException(
"Execution failed: " + output.getErrors());
switch (messageType) {
case SUBSCRIBE:
if (!initialized.get()) {
return GraphQLStatus.unauthorized(session);
}
HandshakeInfo handshakeInfo = session.getHandshakeInfo();
WebSocketInput input = new WebSocketInput(handshakeInfo, id, getPayload(map));
if (logger.isDebugEnabled()) {
logger.debug("Executing: " + input);
}
return executionChain.execute(input)
.flatMapMany(output -> handleWebOutput(session, input.id(), output));
case COMPLETE:
Subscription subscription = this.subscriptions.remove(id);
if (subscription != null) {
subscription.cancel();
}
return Flux.empty();
case CONNECTION_INIT:
if (!initialized.compareAndSet(false, true)) {
return GraphQLStatus.tooManyInitRequests(session);
}
return Flux.just(encode(session, id, MessageType.CONNECTION_ACK, null));
default:
return GraphQLStatus.invalidMessage(session);
}
if (!(output.getData() instanceof Publisher)) {
throw new IllegalStateException(
"Expected Publisher<ExecutionResult>: " + output.toSpecification());
}
if (logger.isDebugEnabled()) {
logger.debug("Execution complete, subscribing for events.");
}
return (Publisher<ExecutionResult>) output.getData();
})
.map(result -> {
Object data = result.getData();
return encode(session, data);
})
);
});
return session.send(responseFlux);
}
@SuppressWarnings({"unchecked", "ConstantConditions"})
private Map<String, Object> decode(WebSocketMessage message) {
DataBuffer buffer = message.getPayload();
return (Map<String, Object>) jsonDecoder.decode(
DataBufferUtils.retain(buffer), WebInput.MAP_RESOLVABLE_TYPE, null, Collections.emptyMap());
DataBuffer buffer = DataBufferUtils.retain(message.getPayload());
return (Map<String, Object>) decoder.decode(buffer, WebInput.MAP_RESOLVABLE_TYPE, null, null);
}
@SuppressWarnings("unchecked")
private <T> WebSocketMessage encode(WebSocketSession session, Object data) {
DataBuffer buffer = ((Encoder<T>) jsonEncoder).encodeValue((T) data,
session.bufferFactory(),
ResolvableType.forInstance(data),
MimeTypeUtils.APPLICATION_JSON,
Collections.emptyMap());
private static Map<String, Object> getPayload(Map<String, Object> message) {
Map<String, Object> payload = (Map<String, Object>) message.get("payload");
Assert.notNull(payload, "No \"payload\" in message: " + message);
return payload;
}
@SuppressWarnings("unchecked")
private Flux<WebSocketMessage> handleWebOutput(WebSocketSession session, String id, WebOutput output) {
if (logger.isDebugEnabled()) {
logger.debug("Execution result ready" +
(!CollectionUtils.isEmpty(output.getErrors()) ?
" with errors: " + output.getErrors() : "") + ".");
}
Flux<ExecutionResult> outputFlux;
if (output.getData() instanceof Publisher) {
// Subscription
outputFlux = Flux.from((Publisher<ExecutionResult>) output.getData())
.doOnSubscribe(subscription -> {
Subscription previous = this.subscriptions.putIfAbsent(id, subscription);
if (previous != null) {
throw new SubscriptionExistsException();
}
});
}
else {
// Query
outputFlux = (CollectionUtils.isEmpty(output.getErrors()) ?
Flux.just(output) :
Flux.error(new IllegalStateException("Execution failed: " + output.getErrors())));
}
return outputFlux
.map(result -> {
Map<String, Object> dataMap = result.toSpecification();
return encode(session, id, MessageType.NEXT, dataMap);
})
.concatWith(Mono.defer(() -> Mono.just(encode(session, id, MessageType.COMPLETE, null))))
.onErrorResume(ex -> {
if (ex instanceof SubscriptionExistsException) {
return GraphQLStatus.subscriptionExists(session, id);
}
ErrorType errorType = ErrorType.DataFetchingException;
String message = ex.getMessage();
Map<String, Object> errorMap = GraphqlErrorBuilder.newError()
.errorType(errorType)
.message(message)
.build()
.toSpecification();
return Mono.just(encode(session, id, MessageType.ERROR, errorMap));
});
}
@SuppressWarnings("unchecked")
private <T> WebSocketMessage encode(
WebSocketSession session, String id, MessageType messageType, @Nullable Object payload) {
Map<String, Object> payloadMap = new HashMap<>(3);
payloadMap.put("id", id);
payloadMap.put("type", messageType.getType());
if (payload != null) {
payloadMap.put("payload", payload);
}
DataBuffer buffer = ((Encoder<T>) encoder).encodeValue(
(T) payloadMap, session.bufferFactory(), WebInput.MAP_RESOLVABLE_TYPE,
MimeTypeUtils.APPLICATION_JSON, null);
return new WebSocketMessage(WebSocketMessage.Type.TEXT, buffer);
}
private enum MessageType {
CONNECTION_INIT("connection_init"),
CONNECTION_ACK("connection_ack"),
SUBSCRIBE("subscribe"),
NEXT("next"),
ERROR("error"),
COMPLETE("complete");
private static final Map<String, MessageType> messageTypes = new HashMap<>(6);
static {
for (MessageType messageType : MessageType.values()) {
messageTypes.put(messageType.getType(), messageType);
}
}
private final String type;
MessageType(String type) {
this.type = type;
}
public String getType() {
return this.type;
}
@Nullable
public static MessageType resolve(@Nullable String type) {
return (type != null ? messageTypes.get(type) : null);
}
}
private static class GraphQLStatus {
private static final CloseStatus INVALID_MESSAGE_STATUS =
new CloseStatus(4400, "Invalid message");
private static final CloseStatus UNAUTHORIZED_STATUS =
new CloseStatus(4401, "Unauthorized");
private static final CloseStatus INIT_TIMEOUT_STATUS =
new CloseStatus(4408, "Connection initialisation timeout");
private static final CloseStatus TOO_MANY_INIT_REQUESTS_STATUS =
new CloseStatus(4429, "Too many initialisation requests");
public static <V> Flux<V> invalidMessage(WebSocketSession session) {
return closeSession(session, INVALID_MESSAGE_STATUS);
}
public static <V> Flux<V> unauthorized(WebSocketSession session) {
return closeSession(session, UNAUTHORIZED_STATUS);
}
public static Mono<Void> initTimeout(WebSocketSession session) {
return session.close(INIT_TIMEOUT_STATUS);
}
public static <V> Flux<V> subscriptionExists(WebSocketSession session, String id) {
return closeSession(session, new CloseStatus(4409, "Subscriber for " + id + " already exists"));
}
public static <V> Flux<V> tooManyInitRequests(WebSocketSession session) {
return closeSession(session, TOO_MANY_INIT_REQUESTS_STATUS);
}
private static <V> Flux<V> closeSession(WebSocketSession session, CloseStatus status) {
return session.close(status).thenMany(Mono.empty());
}
}
private static class SubscriptionExistsException extends RuntimeException {
}
}

4
spring-graphql-web/src/main/java/org/springframework/graphql/WebInput.java
View File

@@ -25,6 +25,7 @@ import org.springframework.core.ParameterizedTypeReference;
import org.springframework.core.ResolvableType;
import org.springframework.http.HttpHeaders;
import org.springframework.lang.Nullable;
import org.springframework.util.Assert;
import org.springframework.util.CollectionUtils;
import org.springframework.util.StringUtils;
import org.springframework.web.server.ServerWebInputException;
@@ -58,6 +59,9 @@ public class WebInput {
@SuppressWarnings("unchecked")
public WebInput(URI uri, HttpHeaders headers, Map<String, Object> body) {
Assert.notNull(uri, "URI is required'");
Assert.notNull(body, "HttpHeaders is required'");
Assert.notNull(body, "'body' is required'");
this.uri = UriComponentsBuilder.fromUri(uri).build(true);
this.headers = headers;
this.query = getAndValidateQuery(body);

10
spring-graphql-web/src/main/java/org/springframework/graphql/WebInterceptorExecutionChain.java
View File

@@ -48,10 +48,10 @@ class WebInterceptorExecutionChain {
}
public Mono<WebOutput> execute(WebInput webInput) {
return createInputChain(webInput).flatMap(executionInput -> {
public Mono<WebOutput> execute(WebInput input) {
return createInputChain(input).flatMap(executionInput -> {
CompletableFuture<ExecutionResult> future = this.graphQL.executeAsync(executionInput);
return createOutputChain(Mono.fromFuture(future));
return createOutputChain(input, Mono.fromFuture(future));
});
}
@@ -63,8 +63,8 @@ class WebInterceptorExecutionChain {
return preHandleMono;
}
private Mono<WebOutput> createOutputChain(Mono<ExecutionResult> resultMono) {
Mono<WebOutput> outputMono = resultMono.map((ExecutionResult executionResult) -> new WebOutput(executionResult, null));
private Mono<WebOutput> createOutputChain(WebInput input, Mono<ExecutionResult> resultMono) {
Mono<WebOutput> outputMono = resultMono.map((ExecutionResult result) -> new WebOutput(input, result, null));
for (int i = this.interceptors.size() - 1 ; i >= 0; i--) {
WebInterceptor interceptor = this.interceptors.get(i);
outputMono = outputMono.flatMap(interceptor::postHandle);

23
spring-graphql-web/src/main/java/org/springframework/graphql/WebOutput.java
View File

@@ -26,6 +26,7 @@ import graphql.GraphQLError;
import org.springframework.http.HttpHeaders;
import org.springframework.lang.Nullable;
import org.springframework.util.Assert;
/**
@@ -34,6 +35,8 @@ import org.springframework.lang.Nullable;
*/
public class WebOutput implements ExecutionResult {
private final WebInput input;
private final ExecutionResult executionResult;
@Nullable
@@ -43,12 +46,22 @@ public class WebOutput implements ExecutionResult {
/**
* Create an instance that wraps the given {@link ExecutionResult}.
*/
public WebOutput(ExecutionResult executionResult, @Nullable HttpHeaders headers) {
public WebOutput(WebInput input, ExecutionResult executionResult, @Nullable HttpHeaders headers) {
Assert.notNull(input, "WebInput is required.");
Assert.notNull(executionResult, "ExecutionResult is required.");
this.input = input;
this.executionResult = executionResult;
this.headers = headers;
}
/**
* Return the associated {@link WebInput} used for the execution.
*/
public WebInput getWebInput() {
return this.input;
}
@Nullable
@Override
public <T> T getData() {
@@ -95,6 +108,8 @@ public class WebOutput implements ExecutionResult {
public static class Builder {
private final WebInput input;
@Nullable
private Object data;
@@ -107,7 +122,8 @@ public class WebOutput implements ExecutionResult {
private HttpHeaders headers;
public Builder(WebOutput output) {
private Builder(WebOutput output) {
this.input = output.getWebInput();
this.data = output.getData();
this.errors = output.getErrors();
this.extensions = output.getExtensions();
@@ -158,7 +174,8 @@ public class WebOutput implements ExecutionResult {
}
public WebOutput build() {
return new WebOutput(new ExecutionResultImpl(this.data, this.errors, this.extensions), this.headers);
ExecutionResult result = new ExecutionResultImpl(this.data, this.errors, this.extensions);
return new WebOutput(this.input, result, this.headers);
}
}

60
spring-graphql-web/src/main/java/org/springframework/graphql/WebSocketInput.java Normal file
View File

@@ -0,0 +1,60 @@
/*
* Copyright 2002-2020 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* https://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.graphql;
import java.util.Map;
import org.springframework.web.reactive.socket.HandshakeInfo;
/**
* Extension of {@link WebInput} that contains a GraphQL subscription received
* over a WebSocket connection.
*/
public class WebSocketInput extends WebInput {
private final HandshakeInfo handshakeInfo;
private final String id;
public WebSocketInput(HandshakeInfo handshakeInfo, String id, Map<String, Object> payload) {
super(handshakeInfo.getUri(), handshakeInfo.getHeaders(), payload);
this.handshakeInfo = handshakeInfo;
this.id = id;
}
/**
* Return information about the WebSocket handshake.
*/
public HandshakeInfo handshakeInfo() {
return this.handshakeInfo;
}
/**
* Return the id that will correlate server responses to client requests
* within a multiplexed WebSocket connection.
*/
public String id() {
return this.id;
}
@Override
public String toString() {
return "id='" + id() + "', " + super.toString();
}
}

127
spring-graphql-web/src/test/java/org/springframework/boot/graphql/WebFluxApplicationContextTests.java
View File

@@ -1,17 +1,24 @@
/*
* Copyright 2002-2020 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* https://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.boot.graphql;
import java.net.URI;
import java.nio.charset.StandardCharsets;
import java.util.Collections;
import java.util.Map;
import java.util.function.Consumer;
import org.junit.jupiter.api.Test;
import org.reactivestreams.Publisher;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import reactor.test.StepVerifier;
import org.springframework.boot.autoconfigure.AutoConfigurations;
import org.springframework.boot.autoconfigure.http.codec.CodecsAutoConfiguration;
@@ -23,24 +30,11 @@ import org.springframework.boot.test.context.runner.ReactiveWebApplicationContex
import org.springframework.context.ApplicationContext;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.core.ResolvableType;
import org.springframework.core.io.buffer.DataBuffer;
import org.springframework.core.io.buffer.DataBufferUtils;
import org.springframework.core.io.buffer.DefaultDataBufferFactory;
import org.springframework.graphql.GraphQLDataFetchers;
import org.springframework.graphql.WebFluxGraphQLHandler;
import org.springframework.graphql.WebFluxGraphQLWebSocketHandler;
import org.springframework.http.HttpHeaders;
import org.springframework.http.MediaType;
import org.springframework.http.codec.json.Jackson2JsonDecoder;
import org.springframework.test.web.reactive.server.WebTestClient;
import org.springframework.web.reactive.socket.CloseStatus;
import org.springframework.web.reactive.socket.HandshakeInfo;
import org.springframework.web.reactive.socket.WebSocketMessage;
import org.springframework.web.reactive.socket.adapter.AbstractWebSocketSession;
import static graphql.schema.idl.TypeRuntimeWiring.newTypeWiring;
import static org.assertj.core.api.Assertions.assertThat;
class WebFluxApplicationContextTests {
@@ -82,43 +76,6 @@ class WebFluxApplicationContextTests {
testWithWebClient(client -> client.post().uri("").bodyValue(":)").exchange().expectStatus().isBadRequest());
}
@Test
void subscription() {
testWithApplicationContext(context -> {
String query =
"{ \"query\": \"" +
" subscription TestSubscription {" +
" bookSearch(minPages: 200) {" +
" id" +
" name" +
" pageCount" +
" author" +
" }" +
"}" +
"\"}";
DataBuffer buffer = DefaultDataBufferFactory.sharedInstance.wrap(query.getBytes(StandardCharsets.UTF_8));
Flux<WebSocketMessage> input = Flux.just(new WebSocketMessage(WebSocketMessage.Type.TEXT, buffer));
TestWebSocketSession session = new TestWebSocketSession("1", URI.create(BASE_URL), input);
context.getBean(WebFluxGraphQLWebSocketHandler.class).handle(session).block();
StepVerifier.create(session.getOutput())
.consumeNextWith(message -> assertThat(extractBook(message)).containsEntry("id", "book-2"))
.consumeNextWith(message -> assertThat(extractBook(message)).containsEntry("id", "book-3"))
.consumeNextWith(message -> assertThat(extractBook(message)).containsEntry("id", "book-3"))
.verifyComplete();
});
}
@SuppressWarnings({"unchecked", "ConstantConditions"})
private Map<String, Object> extractBook(WebSocketMessage message) {
Map<String, Object> map = (Map<String, Object>) new Jackson2JsonDecoder().decode(
DataBufferUtils.retain(message.getPayload()),
ResolvableType.forClass(Map.class), null, Collections.emptyMap());
return (Map<String, Object>) map.get("bookSearch");
}
private void testWithWebClient(Consumer<WebTestClient> consumer) {
testWithApplicationContext(context -> {
WebTestClient client = WebTestClient.bindToApplicationContext(context)
@@ -149,57 +106,9 @@ class WebFluxApplicationContextTests {
@Bean
public RuntimeWiringCustomizer bookDataFetcher() {
return (runtimeWiring) -> {
runtimeWiring.type(newTypeWiring("Query")
.dataFetcher("bookById", GraphQLDataFetchers.getBookByIdDataFetcher()));
runtimeWiring.type(newTypeWiring("Subscription")
.dataFetcher("bookSearch", GraphQLDataFetchers.getBooksOnSale()));
};
}
}
private static class TestWebSocketSession extends AbstractWebSocketSession<Object> {
private final Flux<WebSocketMessage> input;
private Flux<WebSocketMessage> output;
public TestWebSocketSession(String id, URI uri, Flux<WebSocketMessage> input) {
super(new Object(), id,
new HandshakeInfo(uri, new HttpHeaders(), Mono.empty(), null),
DefaultDataBufferFactory.sharedInstance);
this.input = input;
}
@Override
public Flux<WebSocketMessage> receive() {
return this.input;
}
@Override
public Mono<Void> send(Publisher<WebSocketMessage> messages) {
this.output = Flux.from(messages);
return Mono.empty();
}
public Flux<WebSocketMessage> getOutput() {
return this.output;
}
@Override
public boolean isOpen() {
throw new java.lang.UnsupportedOperationException();
}
@Override
public Mono<Void> close(CloseStatus status) {
throw new java.lang.UnsupportedOperationException();
}
@Override
public Mono<CloseStatus> closeStatus() {
throw new java.lang.UnsupportedOperationException();
return (runtimeWiring) ->
runtimeWiring.type(newTypeWiring("Query")
.dataFetcher("bookById", GraphQLDataFetchers.getBookByIdDataFetcher()));
}
}

363
spring-graphql-web/src/test/java/org/springframework/graphql/WebFluxGraphQLWebSocketHandlerTests.java Normal file
View File

@@ -0,0 +1,363 @@
/*
* Copyright 2002-2020 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* https://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.graphql;
import java.io.File;
import java.net.URI;
import java.nio.charset.StandardCharsets;
import java.time.Duration;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.Map;
import java.util.function.BiConsumer;
import graphql.GraphQL;
import graphql.schema.GraphQLSchema;
import graphql.schema.idl.RuntimeWiring;
import graphql.schema.idl.SchemaGenerator;
import graphql.schema.idl.SchemaParser;
import graphql.schema.idl.TypeDefinitionRegistry;
import org.junit.jupiter.api.Test;
import org.reactivestreams.Publisher;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import reactor.core.publisher.Sinks;
import reactor.test.StepVerifier;
import org.springframework.core.io.buffer.DataBuffer;
import org.springframework.core.io.buffer.DataBufferUtils;
import org.springframework.core.io.buffer.DefaultDataBufferFactory;
import org.springframework.http.HttpHeaders;
import org.springframework.http.codec.ServerCodecConfigurer;
import org.springframework.http.codec.json.Jackson2JsonDecoder;
import org.springframework.lang.Nullable;
import org.springframework.util.ResourceUtils;
import org.springframework.web.reactive.socket.CloseStatus;
import org.springframework.web.reactive.socket.HandshakeInfo;
import org.springframework.web.reactive.socket.WebSocketMessage;
import org.springframework.web.reactive.socket.adapter.AbstractWebSocketSession;
import static graphql.schema.idl.TypeRuntimeWiring.newTypeWiring;
import static org.assertj.core.api.Assertions.as;
import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.InstanceOfAssertFactories.map;
/**
* Unit tests for {@link WebFluxGraphQLWebSocketHandler}.
*/
public class WebFluxGraphQLWebSocketHandlerTests {
private static final Jackson2JsonDecoder decoder = new Jackson2JsonDecoder();
private static final String SUBSCRIPTION_ID = "123";
private static final String BOOK_SEARCH_QUERY = "{" +
"\"id\":\"" + SUBSCRIPTION_ID + "\"," +
"\"type\":\"subscribe\"," +
"\"payload\":{\"query\": \"" +
" subscription TestSubscription {" +
" bookSearch(minPages: 200) {" +
" id" +
" name" +
" pageCount" +
" author" +
" }}\"}" +
"}";
@Test
void query() throws Exception {
String bookQuery = "{" +
"\"id\":\"" + SUBSCRIPTION_ID + "\"," +
"\"type\":\"subscribe\"," +
"\"payload\":{\"query\": \"" +
" query TestQuery {" +
" bookById(id: \\\"book-1\\\"){ " +
" id" +
" name" +
" pageCount" +
" author" +
" }}\"}" +
"}";
Flux<WebSocketMessage> input = Flux.just(
toWebSocketMessage("{\"id\":\"" + SUBSCRIPTION_ID + "\",\"type\":\"connection_init\"}"),
toWebSocketMessage(bookQuery));
TestWebSocketSession session = new TestWebSocketSession(input);
initWebSocketHandler().handle(session).block();
StepVerifier.create(session.getOutput())
.consumeNextWith(message -> assertMessageType(message, "connection_ack"))
.consumeNextWith(message ->
assertThat(decode(message))
.hasSize(3)
.containsEntry("id", SUBSCRIPTION_ID)
.containsEntry("type", "next")
.extractingByKey("payload", as(map(String.class, Object.class)))
.extractingByKey("data", as(map(String.class, Object.class)))
.extractingByKey("bookById", as(map(String.class, Object.class)))
.containsEntry("name", "GraphQL for beginners"))
.consumeNextWith(message -> assertMessageType(message, "complete"))
.verifyComplete();
}
@Test
void subscription() throws Exception {
Flux<WebSocketMessage> input = Flux.just(
toWebSocketMessage("{\"id\":\"" + SUBSCRIPTION_ID + "\",\"type\":\"connection_init\"}"),
toWebSocketMessage(BOOK_SEARCH_QUERY));
TestWebSocketSession session = new TestWebSocketSession(input);
initWebSocketHandler().handle(session).block();
BiConsumer<WebSocketMessage, String> bookPayloadAssertion = (message, bookId) ->
assertThat(decode(message))
.hasSize(3)
.containsEntry("id", SUBSCRIPTION_ID)
.containsEntry("type", "next")
.extractingByKey("payload", as(map(String.class, Object.class)))
.extractingByKey("data", as(map(String.class, Object.class)))
.extractingByKey("bookSearch", as(map(String.class, Object.class)))
.containsEntry("id", bookId);
StepVerifier.create(session.getOutput())
.consumeNextWith(message -> assertMessageType(message, "connection_ack"))
.consumeNextWith(message -> bookPayloadAssertion.accept(message, "book-2"))
.consumeNextWith(message -> bookPayloadAssertion.accept(message, "book-3"))
.consumeNextWith(message -> bookPayloadAssertion.accept(message, "book-3"))
.consumeNextWith(message -> assertMessageType(message, "complete"))
.verifyComplete();
}
@Test
void unauthorizedWithoutMessageType() throws Exception {
Flux<WebSocketMessage> input = Flux.just(
toWebSocketMessage("{\"id\":\"" + SUBSCRIPTION_ID + "\",\"type\":\"connection_init\"}"),
toWebSocketMessage("{\"id\":\"" + SUBSCRIPTION_ID + "\"}")); // No message type
TestWebSocketSession session = new TestWebSocketSession(input);
initWebSocketHandler().handle(session).block();
StepVerifier.create(session.getOutput())
.consumeNextWith(message -> assertMessageType(message, "connection_ack"))
.verifyComplete();
StepVerifier.create(session.closeStatus())
.expectNext(new CloseStatus(4400, "Invalid message"))
.verifyComplete();
}
@Test
void unauthorizedWithoutConnectionInit() throws Exception {
TestWebSocketSession session = new TestWebSocketSession(Flux.just(toWebSocketMessage(BOOK_SEARCH_QUERY)));
initWebSocketHandler().handle(session).block();
StepVerifier.create(session.getOutput()).verifyComplete();
StepVerifier.create(session.closeStatus())
.expectNext(new CloseStatus(4401, "Unauthorized"))
.verifyComplete();
}
@Test
void tooManyConnectionInitRequests() throws Exception {
Flux<WebSocketMessage> input = Flux.just(
toWebSocketMessage("{\"id\":\"" + SUBSCRIPTION_ID + "\",\"type\":\"connection_init\"}"),
toWebSocketMessage("{\"id\":\"" + SUBSCRIPTION_ID + "\",\"type\":\"connection_init\"}"));
TestWebSocketSession session = new TestWebSocketSession(input);
initWebSocketHandler().handle(session).block();
StepVerifier.create(session.getOutput())
.consumeNextWith(message -> assertMessageType(message, "connection_ack"))
.verifyComplete();
StepVerifier.create(session.closeStatus())
.expectNext(new CloseStatus(4429, "Too many initialisation requests"))
.verifyComplete();
}
@Test
void connectionInitTimeout() throws Exception {
TestWebSocketSession session = new TestWebSocketSession(Flux.empty());
initWebSocketHandler(Collections.emptyList(), Duration.ofMillis(50)).handle(session).block();
StepVerifier.create(session.closeStatus())
.expectNext(new CloseStatus(4408, "Connection initialisation timeout"))
.verifyComplete();
}
@Test
void subscriptionExists() throws Exception {
Flux<WebSocketMessage> input = Flux.just(toWebSocketMessage(
"{\"id\":\"" + SUBSCRIPTION_ID + "\",\"type\":\"connection_init\"}"),
toWebSocketMessage(BOOK_SEARCH_QUERY),
toWebSocketMessage(BOOK_SEARCH_QUERY));
List<WebInterceptor> interceptors = Collections.singletonList(new TakeOneAndNeverCompleteInterceptor());
TestWebSocketSession session = new TestWebSocketSession(input);
initWebSocketHandler(interceptors, null).handle(session).block();
// Collect messages until session closed
List<Map<String, Object>> messages = new ArrayList<>();
session.getOutput().subscribe(message -> messages.add(decode(message)));
StepVerifier.create(session.closeStatus())
.expectNext(new CloseStatus(4409, "Subscriber for " + SUBSCRIPTION_ID + " already exists"))
.verifyComplete();
assertThat(messages.size()).isEqualTo(2);
assertThat(messages.get(0).get("type")).isEqualTo("connection_ack");
assertThat(messages.get(1).get("type")).isEqualTo("next");
}
@Test
void clientCompletion() throws Exception {
Sinks.Many<WebSocketMessage> input = Sinks.many().unicast().onBackpressureBuffer();
input.tryEmitNext(toWebSocketMessage("{\"id\":\"" + SUBSCRIPTION_ID + "\",\"type\":\"connection_init\"}"));
input.tryEmitNext(toWebSocketMessage(BOOK_SEARCH_QUERY));
List<WebInterceptor> interceptors = Collections.singletonList(new TakeOneAndNeverCompleteInterceptor());
TestWebSocketSession session = new TestWebSocketSession(input.asFlux());
initWebSocketHandler(interceptors, null).handle(session).block();
String completeMessage = "{\"id\":\"" + SUBSCRIPTION_ID + "\",\"type\":\"complete\"}";
StepVerifier.create(session.getOutput())
.consumeNextWith(message -> assertMessageType(message, "connection_ack"))
.consumeNextWith(message -> assertMessageType(message, "next"))
.then(() -> input.tryEmitNext(toWebSocketMessage(completeMessage)))
.as("Second subscription with same id is possible only if the first was properly removed")
.then(() -> input.tryEmitNext(toWebSocketMessage(BOOK_SEARCH_QUERY)))
.consumeNextWith(message -> assertMessageType(message, "next"))
.then(() -> input.tryEmitNext(toWebSocketMessage(completeMessage)))
.verifyTimeout(Duration.ofMillis(500));
}
private WebFluxGraphQLWebSocketHandler initWebSocketHandler() throws Exception {
return initWebSocketHandler(Collections.emptyList(), Duration.ofSeconds(69));
}
private WebFluxGraphQLWebSocketHandler initWebSocketHandler(
@Nullable List<WebInterceptor> interceptors, @Nullable Duration initTimeoutDuration) throws Exception {
GraphQL graphQL = initGraphQL();
return new WebFluxGraphQLWebSocketHandler(graphQL,
(interceptors != null ? interceptors : Collections.emptyList()),
ServerCodecConfigurer.create(),
(initTimeoutDuration != null ? initTimeoutDuration : Duration.ofSeconds(60)));
}
private static GraphQL initGraphQL() throws Exception {
File schemaFile = ResourceUtils.getFile("classpath:books/schema.graphqls");
TypeDefinitionRegistry typeDefinitionRegistry = new SchemaParser().parse(schemaFile);
RuntimeWiring.Builder builder = RuntimeWiring.newRuntimeWiring();
builder.type(newTypeWiring("Query").dataFetcher("bookById", GraphQLDataFetchers.getBookByIdDataFetcher()));
builder.type(newTypeWiring("Subscription").dataFetcher("bookSearch", GraphQLDataFetchers.getBooksOnSale()));
RuntimeWiring runtimeWiring = builder.build();
GraphQLSchema schema = new SchemaGenerator().makeExecutableSchema(typeDefinitionRegistry, runtimeWiring);
return GraphQL.newGraphQL(schema).build();
}
private static WebSocketMessage toWebSocketMessage(String data) {
DataBuffer buffer = DefaultDataBufferFactory.sharedInstance.wrap(data.getBytes(StandardCharsets.UTF_8));
return new WebSocketMessage(WebSocketMessage.Type.TEXT, buffer);
}
@SuppressWarnings({"unchecked", "ConstantConditions"})
private Map<String, Object> decode(WebSocketMessage message) {
return (Map<String, Object>) decoder.decode(
DataBufferUtils.retain(message.getPayload()),
WebInput.MAP_RESOLVABLE_TYPE, null, Collections.emptyMap());
}
private void assertMessageType(WebSocketMessage message, String messageType) {
assertThat(decode(message))
.containsEntry("id", SUBSCRIPTION_ID)
.containsEntry("type", messageType);
}
private static class TestWebSocketSession extends AbstractWebSocketSession<Object> {
private final Flux<WebSocketMessage> input;
private Flux<WebSocketMessage> output = Flux.empty();
private final Sinks.One<CloseStatus> closeStatusSink = Sinks.one();
public TestWebSocketSession(Flux<WebSocketMessage> input) {
this("1", URI.create("https://example.org/graphql"), input);
}
public TestWebSocketSession(String id, URI uri, Flux<WebSocketMessage> input) {
super(new Object(), id,
new HandshakeInfo(uri, new HttpHeaders(), Mono.empty(), null),
DefaultDataBufferFactory.sharedInstance);
this.input = input;
}
@Override
public Flux<WebSocketMessage> receive() {
return this.input;
}
@Override
public Mono<Void> send(Publisher<WebSocketMessage> messages) {
this.output = Flux.from(messages);
return Mono.empty();
}
public Flux<WebSocketMessage> getOutput() {
return this.output;
}
@Override
public boolean isOpen() {
throw new java.lang.UnsupportedOperationException();
}
@Override
public Mono<Void> close(CloseStatus status) {
this.closeStatusSink.tryEmitValue(status);
return Mono.empty();
}
@Override
public Mono<CloseStatus> closeStatus() {
return this.closeStatusSink.asMono();
}
}
private static class TakeOneAndNeverCompleteInterceptor implements WebInterceptor {
@Override
public Mono<WebOutput> postHandle(WebOutput output) {
return Mono.just(output.transform(builder -> {
Publisher<?> publisher = output.getData();
assertThat(publisher).isNotNull();
builder.data(Flux.from(publisher).take(1).concatWith(Flux.never()));
}));
}
}
}