From b10d5f7df7f27038f2d02c9c0e9d5b70cc604e0f Mon Sep 17 00:00:00 2001 From: Rossen Stoyanchev Date: Sun, 29 Nov 2020 20:04:15 +0000 Subject: [PATCH] Add support for subscriptions over WebSocket in WebFlux --- spring-graphql-web/build.gradle | 1 + .../boot/graphql/GraphQLProperties.java | 17 +- .../WebFluxGraphQLAutoConfiguration.java | 45 +++++- .../graphql/WebFluxGraphQLHandler.java | 91 ++++++++++- .../org/springframework/graphql/WebInput.java | 3 + .../WebFluxApplicationContextTests.java | 153 +++++++++++++++--- .../graphql/GraphQLDataFetchers.java | 19 ++- .../src/test/resources/books/schema.graphqls | 6 +- 8 files changed, 294 insertions(+), 41 deletions(-) diff --git a/spring-graphql-web/build.gradle b/spring-graphql-web/build.gradle index b77c3b42..1795db8e 100644 --- a/spring-graphql-web/build.gradle +++ b/spring-graphql-web/build.gradle @@ -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' diff --git a/spring-graphql-web/src/main/java/org/springframework/boot/graphql/GraphQLProperties.java b/spring-graphql-web/src/main/java/org/springframework/boot/graphql/GraphQLProperties.java index 5d1ddb06..5031c619 100644 --- a/spring-graphql-web/src/main/java/org/springframework/boot/graphql/GraphQLProperties.java +++ b/spring-graphql-web/src/main/java/org/springframework/boot/graphql/GraphQLProperties.java @@ -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; } diff --git a/spring-graphql-web/src/main/java/org/springframework/boot/graphql/WebFluxGraphQLAutoConfiguration.java b/spring-graphql-web/src/main/java/org/springframework/boot/graphql/WebFluxGraphQLAutoConfiguration.java index 3b251975..9e157e99 100644 --- a/spring-graphql-web/src/main/java/org/springframework/boot/graphql/WebFluxGraphQLAutoConfiguration.java +++ b/spring-graphql-web/src/main/java/org/springframework/boot/graphql/WebFluxGraphQLAutoConfiguration.java @@ -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 graphQLQueryEndpoint( - ResourceLoader resourceLoader, WebFluxGraphQLHandler handler, GraphQLProperties properties) { + public RouterFunction 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; + } + } diff --git a/spring-graphql-web/src/main/java/org/springframework/graphql/WebFluxGraphQLHandler.java b/spring-graphql-web/src/main/java/org/springframework/graphql/WebFluxGraphQLHandler.java index 4a4cd274..06bcbcae 100644 --- a/spring-graphql-web/src/main/java/org/springframework/graphql/WebFluxGraphQLHandler.java +++ b/spring-graphql-web/src/main/java/org/springframework/graphql/WebFluxGraphQLHandler.java @@ -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 { +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 { * 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 interceptors) { + public WebFluxGraphQLHandler(GraphQL graphQL, List interceptors, + Decoder jsonDecoder, Encoder jsonEncoder) { + this.executionChain = new WebInterceptorExecutionChain(graphQL, interceptors); + this.jsonDecoder = jsonDecoder; + this.jsonEncoder = jsonEncoder; } - public Mono handle(ServerRequest request) { + /** + * Handle GraphQL query requests over HTTP. + */ + public Mono 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 { }); } + /** + * 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 handle(WebSocketSession session) { + return session.send(session.receive() + .concatMap(message -> { + Map 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: " + output.toSpecification()); + } + return (Publisher) output.getData(); + }) + .map(result -> encode(session, result.getData())) + ); + } + + @SuppressWarnings({"unchecked", "ConstantConditions"}) + private Map decode(WebSocketMessage message) { + DataBuffer buffer = message.getPayload(); + return (Map) jsonDecoder.decode( + DataBufferUtils.retain(buffer), WebInput.MAP_RESOLVABLE_TYPE, null, Collections.emptyMap()); + } + + @SuppressWarnings("unchecked") + private WebSocketMessage encode(WebSocketSession session, Object data) { + DataBuffer buffer = ((Encoder) jsonEncoder).encodeValue((T) data, + session.bufferFactory(), + ResolvableType.forInstance(data), + MimeTypeUtils.APPLICATION_JSON, + Collections.emptyMap()); + return new WebSocketMessage(WebSocketMessage.Type.TEXT, buffer); + } + } + } diff --git a/spring-graphql-web/src/main/java/org/springframework/graphql/WebInput.java b/spring-graphql-web/src/main/java/org/springframework/graphql/WebInput.java index 2809030c..043dbc08 100644 --- a/spring-graphql-web/src/main/java/org/springframework/graphql/WebInput.java +++ b/spring-graphql-web/src/main/java/org/springframework/graphql/WebInput.java @@ -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_PARAMETERIZED_TYPE_REF = new ParameterizedTypeReference>() {}; + static final ResolvableType MAP_RESOLVABLE_TYPE = ResolvableType.forType(MAP_PARAMETERIZED_TYPE_REF); + private final UriComponents uri; diff --git a/spring-graphql-web/src/test/java/org/springframework/boot/graphql/WebFluxApplicationContextTests.java b/spring-graphql-web/src/test/java/org/springframework/boot/graphql/WebFluxApplicationContextTests.java index 58581788..58c54b60 100644 --- a/spring-graphql-web/src/test/java/org/springframework/boot/graphql/WebFluxApplicationContextTests.java +++ b/spring-graphql-web/src/test/java/org/springframework/boot/graphql/WebFluxApplicationContextTests.java @@ -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 consumer) { + DataBuffer buffer = DefaultDataBufferFactory.sharedInstance.wrap(query.getBytes(StandardCharsets.UTF_8)); + Flux 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 extractBook(WebSocketMessage message) { + Map map = (Map) new Jackson2JsonDecoder().decode( + DataBufferUtils.retain(message.getPayload()), + ResolvableType.forClass(Map.class), null, Collections.emptyMap()); + return (Map) map.get("bookSearch"); + } + + private void testWithWebClient(Consumer 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 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 { + + private final Flux input; + + private Flux output; + + public TestWebSocketSession(String id, URI uri, Flux input) { + super(new Object(), id, + new HandshakeInfo(uri, new HttpHeaders(), Mono.empty(), null), + DefaultDataBufferFactory.sharedInstance); + this.input = input; } + @Override + public Flux receive() { + return this.input; + } + + @Override + public Mono send(Publisher messages) { + this.output = Flux.from(messages); + return Mono.empty(); + } + + public Flux getOutput() { + return this.output; + } + + @Override + public boolean isOpen() { + throw new java.lang.UnsupportedOperationException(); + } + + @Override + public Mono close(CloseStatus status) { + throw new java.lang.UnsupportedOperationException(); + } + + @Override + public Mono closeStatus() { + throw new java.lang.UnsupportedOperationException(); + } } + } diff --git a/spring-graphql-web/src/test/java/org/springframework/graphql/GraphQLDataFetchers.java b/spring-graphql-web/src/test/java/org/springframework/graphql/GraphQLDataFetchers.java index 475c763f..4e143634 100644 --- a/spring-graphql-web/src/test/java/org/springframework/graphql/GraphQLDataFetchers.java +++ b/spring-graphql-web/src/test/java/org/springframework/graphql/GraphQLDataFetchers.java @@ -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")); + } + } diff --git a/spring-graphql-web/src/test/resources/books/schema.graphqls b/spring-graphql-web/src/test/resources/books/schema.graphqls index 975b7ca6..7ec591e5 100644 --- a/spring-graphql-web/src/test/resources/books/schema.graphqls +++ b/spring-graphql-web/src/test/resources/books/schema.graphqls @@ -7,4 +7,8 @@ type Book { name: String pageCount: Int author: String -} \ No newline at end of file +} + +type Subscription { + bookSearch(minPages:Int) : Book! +}