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

Add support for subscriptions over WebSocket in WebFlux

Browse Source
This commit is contained in:
Rossen Stoyanchev
2020-11-29 20:04:15 +00:00
parent 1dc38c8323
commit b10d5f7df7
8 changed files with 294 additions and 41 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
spring-graphql-web/build.gradle
View File

@@ -42,6 +42,7 @@ dependencies {
testImplementation 'com.fasterxml.jackson.core:jackson-databind'
testImplementation 'org.springframework:spring-webflux'
testImplementation 'org.springframework:spring-webmvc'
testImplementation 'io.projectreactor:reactor-test'
testImplementation 'javax.servlet:javax.servlet-api'
testImplementation 'org.springframework.boot:spring-boot-actuator-autoconfigure'
testImplementation 'org.springframework.boot:spring-boot-starter-test'

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

@@ -15,9 +15,7 @@
*/
package org.springframework.boot.graphql;
import org.springframework.boot.actuate.autoconfigure.metrics.AutoTimeProperties;
import org.springframework.boot.context.properties.ConfigurationProperties;
import org.springframework.boot.context.properties.NestedConfigurationProperty;
@ConfigurationProperties(prefix = "spring.graphql")
public class GraphQLProperties {
@@ -28,10 +26,15 @@ public class GraphQLProperties {
private String schemaLocation = "classpath:schema.graphqls";
/**
* Path of the GraphQL HTTP endpoint.
* Path of the GraphQL HTTP query endpoint.
*/
private String path = "/graphql";
/**
* Path of the GraphQL WebSocket subscription endpoint.
*/
private String webSocketPath = path + "/websocket";
public String getPath() {
return path;
}
@@ -40,6 +43,14 @@ public class GraphQLProperties {
this.path = path;
}
public String getWebSocketPath() {
return webSocketPath;
}
public void setWebSocketPath(String webSocketPath) {
this.webSocketPath = webSocketPath;
}
public String getSchemaLocation() {
return schemaLocation;
}

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

@@ -16,6 +16,7 @@
package org.springframework.boot.graphql;
import java.util.Collections;
import java.util.Map;
import graphql.GraphQL;
@@ -26,13 +27,21 @@ import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean
import org.springframework.boot.autoconfigure.condition.ConditionalOnWebApplication;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.core.ResolvableType;
import org.springframework.core.codec.Decoder;
import org.springframework.core.codec.Encoder;
import org.springframework.core.io.Resource;
import org.springframework.core.io.ResourceLoader;
import org.springframework.graphql.WebFluxGraphQLHandler;
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.web.reactive.HandlerMapping;
import org.springframework.web.reactive.function.server.RouterFunction;
import org.springframework.web.reactive.function.server.RouterFunctions;
import org.springframework.web.reactive.function.server.ServerResponse;
import org.springframework.web.reactive.handler.SimpleUrlHandlerMapping;
import static org.springframework.web.reactive.function.server.RequestPredicates.accept;
import static org.springframework.web.reactive.function.server.RequestPredicates.contentType;
@@ -46,21 +55,47 @@ public class WebFluxGraphQLAutoConfiguration {
@Bean
@ConditionalOnMissingBean
public WebFluxGraphQLHandler graphQLHandler(GraphQL.Builder graphQLBuilder) {
return new WebFluxGraphQLHandler(graphQLBuilder.build(), Collections.emptyList());
public WebFluxGraphQLHandler graphQLHandler(
GraphQL.Builder graphQLBuilder, ServerCodecConfigurer configurer) {
ResolvableType mapType = ResolvableType.forClass(Map.class);
Decoder<?> jsonDecoder = configurer.getReaders().stream()
.filter(reader -> reader.canRead(mapType, MediaType.APPLICATION_JSON))
.map(reader -> ((DecoderHttpMessageReader<?>) reader).getDecoder())
.findFirst()
.orElseThrow(() -> new IllegalArgumentException("No JSON Decoder"));
Encoder<?> jsonEncoder = configurer.getWriters().stream()
.filter(writer -> writer.canWrite(mapType, MediaType.APPLICATION_JSON))
.map(writer -> ((EncoderHttpMessageWriter<?>) writer).getEncoder())
.findFirst()
.orElseThrow(() -> new IllegalArgumentException("No JSON Encoder"));
return new WebFluxGraphQLHandler(
graphQLBuilder.build(), Collections.emptyList(), jsonDecoder, jsonEncoder);
}
@Bean
public RouterFunction<ServerResponse> graphQLQueryEndpoint(
ResourceLoader resourceLoader, WebFluxGraphQLHandler handler, GraphQLProperties properties) {
public RouterFunction<ServerResponse> graphQLEndpoint(
WebFluxGraphQLHandler handler, GraphQLProperties properties, ResourceLoader resourceLoader) {
String path = properties.getPath();
Resource resource = resourceLoader.getResource("classpath:graphiql/index.html");
return RouterFunctions.route()
.GET(path, req -> ServerResponse.ok().bodyValue(resource))
.POST(path, accept(MediaType.APPLICATION_JSON).and(contentType(MediaType.APPLICATION_JSON)), handler)
.POST(path, accept(MediaType.APPLICATION_JSON).and(contentType(MediaType.APPLICATION_JSON)), handler::handleQuery)
.build();
}
@Bean
public HandlerMapping graphQLWebSocketEndpoint(WebFluxGraphQLHandler handler, GraphQLProperties properties) {
String path = properties.getWebSocketPath();
SimpleUrlHandlerMapping mapping = new SimpleUrlHandlerMapping();
mapping.setUrlMap(Collections.singletonMap(path, handler.getSubscriptionWebSocketHandler()));
mapping.setOrder(-1); // Ahead of annotated controllers
return mapping;
}
}

91
spring-graphql-web/src/main/java/org/springframework/graphql/WebFluxGraphQLHandler.java
View File

@@ -15,23 +15,41 @@
*/
package org.springframework.graphql;
import java.util.Collections;
import java.util.List;
import java.util.Map;
import graphql.ExecutionResult;
import graphql.GraphQL;
import org.reactivestreams.Publisher;
import reactor.core.publisher.Mono;
import org.springframework.web.reactive.function.server.HandlerFunction;
import org.springframework.core.ResolvableType;
import org.springframework.core.codec.Decoder;
import org.springframework.core.codec.Encoder;
import org.springframework.core.io.buffer.DataBuffer;
import org.springframework.core.io.buffer.DataBufferUtils;
import org.springframework.util.CollectionUtils;
import org.springframework.util.MimeTypeUtils;
import org.springframework.web.reactive.function.server.ServerRequest;
import org.springframework.web.reactive.function.server.ServerResponse;
import org.springframework.web.reactive.socket.HandshakeInfo;
import org.springframework.web.reactive.socket.WebSocketHandler;
import org.springframework.web.reactive.socket.WebSocketMessage;
import org.springframework.web.reactive.socket.WebSocketSession;
/**
* GraphQL handler to expose as a WebFlux.fn endpoint via
* {@link org.springframework.web.reactive.function.server.RouterFunctions}.
*/
public class WebFluxGraphQLHandler implements HandlerFunction<ServerResponse> {
public class WebFluxGraphQLHandler {
private final WebInterceptorExecutionChain executionChain;
private final Decoder<?> jsonDecoder;
private final Encoder<?> jsonEncoder;
/**
* Create a handler that executes queries through the given {@link GraphQL}
@@ -39,13 +57,22 @@ public class WebFluxGraphQLHandler implements HandlerFunction<ServerResponse> {
* result from the execution of the query.
* @param graphQL the GraphQL instance to use for query execution
* @param interceptors 0 or more interceptors to customize input and output
* @param jsonDecoder to decode JSON for subscriptions over WebSocket
* @param jsonEncoder to encode JSON for subscriptions over WebSocket
*/
public WebFluxGraphQLHandler(GraphQL graphQL, List<WebInterceptor> interceptors) {
public WebFluxGraphQLHandler(GraphQL graphQL, List<WebInterceptor> interceptors,
Decoder<?> jsonDecoder, Encoder<?> jsonEncoder) {
this.executionChain = new WebInterceptorExecutionChain(graphQL, interceptors);
this.jsonDecoder = jsonDecoder;
this.jsonEncoder = jsonEncoder;
}
public Mono<ServerResponse> handle(ServerRequest request) {
/**
* Handle GraphQL query requests over HTTP.
*/
public Mono<ServerResponse> handleQuery(ServerRequest request) {
return request.bodyToMono(WebInput.MAP_PARAMETERIZED_TYPE_REF)
.flatMap(body -> {
WebInput webInput = new WebInput(request.uri(), request.headers().asHttpHeaders(), body);
@@ -60,4 +87,60 @@ public class WebFluxGraphQLHandler implements HandlerFunction<ServerResponse> {
});
}
/**
* Return a handler that supports subscriptions over WebSocket.
*/
public WebSocketHandler getSubscriptionWebSocketHandler() {
return new SubscriptionWebSocketHandler();
}
/**
* Handler for subscriptions over WebSocket.
*/
private class SubscriptionWebSocketHandler implements WebSocketHandler {
@Override
@SuppressWarnings("unchecked")
public Mono<Void> handle(WebSocketSession session) {
return session.send(session.receive()
.concatMap(message -> {
Map<String, Object> map = decode(message);
HandshakeInfo handshakeInfo = session.getHandshakeInfo();
WebInput webInput = new WebInput(handshakeInfo.getUri(), handshakeInfo.getHeaders(), map);
return executionChain.execute(webInput);
})
.concatMap(output -> {
if (!CollectionUtils.isEmpty(output.getErrors())) {
throw new IllegalStateException(
"Execution failed: " + output.getErrors());
}
if (!(output.getData() instanceof Publisher)) {
throw new IllegalStateException(
"Expected Publisher<ExecutionResult>: " + output.toSpecification());
}
return (Publisher<ExecutionResult>) output.getData();
})
.map(result -> encode(session, result.getData()))
);
}
@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());
}
@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());
return new WebSocketMessage(WebSocketMessage.Type.TEXT, buffer);
}
}
}

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

@@ -22,6 +22,7 @@ import java.util.Map;
import graphql.ExecutionInput;
import org.springframework.core.ParameterizedTypeReference;
import org.springframework.core.ResolvableType;
import org.springframework.http.HttpHeaders;
import org.springframework.lang.Nullable;
import org.springframework.util.CollectionUtils;
@@ -40,6 +41,8 @@ public class WebInput {
static final ParameterizedTypeReference<Map<String, Object>> MAP_PARAMETERIZED_TYPE_REF =
new ParameterizedTypeReference<Map<String, Object>>() {};
static final ResolvableType MAP_RESOLVABLE_TYPE = ResolvableType.forType(MAP_PARAMETERIZED_TYPE_REF);
private final UriComponents uri;

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

@@ -1,24 +1,45 @@
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;
import org.springframework.boot.autoconfigure.jackson.JacksonAutoConfiguration;
import org.springframework.boot.autoconfigure.web.reactive.HttpHandlerAutoConfiguration;
import org.springframework.boot.autoconfigure.web.reactive.WebFluxAutoConfiguration;
import org.springframework.boot.test.context.runner.ContextConsumer;
import org.springframework.boot.test.context.runner.ReactiveWebApplicationContextRunner;
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.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 {
@@ -27,10 +48,12 @@ class WebFluxApplicationContextTests {
CodecsAutoConfiguration.class, JacksonAutoConfiguration.class,
GraphQLAutoConfiguration.class, WebFluxGraphQLAutoConfiguration.class);
private static final String BASE_URL = "https://spring.example.org/graphql";
@Test
void endpointHandlesGraphQLQueries() {
testWith(client -> {
void query() {
testWithWebClient(client -> {
String query = "{" +
" bookById(id: \\\"book-1\\\"){ " +
" id" +
@@ -49,34 +72,75 @@ class WebFluxApplicationContextTests {
}
@Test
void missingQuery() {
testWith(client -> client.post().uri("").bodyValue("{}").exchange().expectStatus().isBadRequest());
void queryMissing() {
testWithWebClient(client -> client.post().uri("").bodyValue("{}").exchange().expectStatus().isBadRequest());
}
@Test
void invalidJson() {
testWith(client -> client.post().uri("").bodyValue(":)").exchange().expectStatus().isBadRequest());
void queryIsInvalidJson() {
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" +
" }" +
"}" +
"\"}";
private void testWith(Consumer<WebTestClient> consumer) {
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(WebFluxGraphQLHandler.class)
.getSubscriptionWebSocketHandler().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)
.configureClient()
.defaultHeaders(headers -> {
headers.setContentType(MediaType.APPLICATION_JSON);
headers.setAccept(Collections.singletonList(MediaType.APPLICATION_JSON));
})
.baseUrl(BASE_URL)
.build();
consumer.accept(client);
});
}
private void testWithApplicationContext(ContextConsumer<ApplicationContext> consumer) {
new ReactiveWebApplicationContextRunner()
.withConfiguration(AUTO_CONFIGURATIONS)
.withUserConfiguration(DataFetchersConfiguration.class)
.withPropertyValues(
"spring.main.web-application-type=reactive",
"spring.graphql.schema-location:classpath:books/schema.graphqls")
.run((context) -> {
WebTestClient client = WebTestClient.bindToApplicationContext(context)
.configureClient()
.defaultHeaders(headers -> {
headers.setContentType(MediaType.APPLICATION_JSON);
headers.setAccept(Collections.singletonList(MediaType.APPLICATION_JSON));
})
.baseUrl("https://spring.example.org/graphql")
.build();
consumer.accept(client);
});
.run(consumer);
}
@@ -85,9 +149,58 @@ class WebFluxApplicationContextTests {
@Bean
public RuntimeWiringCustomizer bookDataFetcher() {
return (runtimeWiring) -> runtimeWiring.type(newTypeWiring("Query")
.dataFetcher("bookById", GraphQLDataFetchers.getBookByIdDataFetcher()));
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();
}
}
}

19
spring-graphql-web/src/test/java/org/springframework/graphql/GraphQLDataFetchers.java
View File

@@ -4,6 +4,7 @@ import java.util.Arrays;
import java.util.List;
import graphql.schema.DataFetcher;
import reactor.core.publisher.Flux;
public class GraphQLDataFetchers {
@@ -15,13 +16,15 @@ public class GraphQLDataFetchers {
public static DataFetcher getBookByIdDataFetcher() {
return dataFetchingEnvironment -> {
String bookId = dataFetchingEnvironment.getArgument("id");
return books
.stream()
.filter(book -> book.getId().equals(bookId))
.findFirst()
.orElse(null);
};
return environment -> books.stream()
.filter(book -> book.getId().equals(environment.getArgument("id")))
.findFirst()
.orElse(null);
}
public static DataFetcher getBooksOnSale() {
return environment -> Flux.fromIterable(books)
.filter(book -> book.getPageCount() >= (int) environment.getArgument("minPages"));
}
}

6
spring-graphql-web/src/test/resources/books/schema.graphqls
View File

@@ -7,4 +7,8 @@ type Book {
name: String
pageCount: Int
author: String
}
}
type Subscription {
bookSearch(minPages:Int) : Book!
}