diff --git a/graphql-spring-boot-starter/src/test/java/org/springframework/graphql/boot/GraphQlWebFluxAutoConfigurationTests.java b/graphql-spring-boot-starter/src/test/java/org/springframework/graphql/boot/GraphQlWebFluxAutoConfigurationTests.java index 7cf6db36..253de7c9 100644 --- a/graphql-spring-boot-starter/src/test/java/org/springframework/graphql/boot/GraphQlWebFluxAutoConfigurationTests.java +++ b/graphql-spring-boot-starter/src/test/java/org/springframework/graphql/boot/GraphQlWebFluxAutoConfigurationTests.java @@ -199,7 +199,7 @@ class GraphQlWebFluxAutoConfigurationTests { @Bean WebInterceptor customWebInterceptor() { - return (input, next) -> next.handle(input).map((output) -> + return (webInput, interceptorChain) -> interceptorChain.next(webInput).map((output) -> output.transform((builder) -> builder.responseHeader("X-Custom-Header", "42"))); } diff --git a/graphql-spring-boot-starter/src/test/java/org/springframework/graphql/boot/GraphQlWebMvcAutoConfigurationTests.java b/graphql-spring-boot-starter/src/test/java/org/springframework/graphql/boot/GraphQlWebMvcAutoConfigurationTests.java index cc798d39..4ab472d2 100644 --- a/graphql-spring-boot-starter/src/test/java/org/springframework/graphql/boot/GraphQlWebMvcAutoConfigurationTests.java +++ b/graphql-spring-boot-starter/src/test/java/org/springframework/graphql/boot/GraphQlWebMvcAutoConfigurationTests.java @@ -187,7 +187,7 @@ class GraphQlWebMvcAutoConfigurationTests { @Bean WebInterceptor customWebInterceptor() { - return (input, next) -> next.handle(input).map((output) -> + return (webInput, interceptorChain) -> interceptorChain.next(webInput).map((output) -> output.transform((builder) -> builder.responseHeader("X-Custom-Header", "42"))); } diff --git a/samples/webflux-websocket/src/test/java/io/spring/sample/graphql/QueryTests.java b/samples/webflux-websocket/src/test/java/io/spring/sample/graphql/QueryTests.java index c702c547..eb285d6a 100644 --- a/samples/webflux-websocket/src/test/java/io/spring/sample/graphql/QueryTests.java +++ b/samples/webflux-websocket/src/test/java/io/spring/sample/graphql/QueryTests.java @@ -36,7 +36,7 @@ public class QueryTests { @BeforeEach public void setUp(@Autowired WebGraphQlHandler handler) { this.graphQlTester = WebGraphQlTester.create(webInput -> - handler.handle(webInput).contextWrite(context -> context.put("name", "James"))); + handler.handleRequest(webInput).contextWrite(context -> context.put("name", "James"))); } @Test diff --git a/samples/webflux-websocket/src/test/java/io/spring/sample/graphql/SubscriptionTests.java b/samples/webflux-websocket/src/test/java/io/spring/sample/graphql/SubscriptionTests.java index 38ad6ebc..e5019a11 100644 --- a/samples/webflux-websocket/src/test/java/io/spring/sample/graphql/SubscriptionTests.java +++ b/samples/webflux-websocket/src/test/java/io/spring/sample/graphql/SubscriptionTests.java @@ -38,7 +38,7 @@ public class SubscriptionTests { @BeforeEach public void setUp(@Autowired WebGraphQlHandler handler) { this.graphQlTester = WebGraphQlTester.create(webInput -> - handler.handle(webInput).contextWrite(context -> context.put("name", "James"))); + handler.handleRequest(webInput).contextWrite(context -> context.put("name", "James"))); } @Test diff --git a/samples/webmvc-http-security/src/main/java/io/spring/sample/graphql/SampleApplication.java b/samples/webmvc-http-security/src/main/java/io/spring/sample/graphql/SampleApplication.java index a2fae4bb..b84d4e74 100644 --- a/samples/webmvc-http-security/src/main/java/io/spring/sample/graphql/SampleApplication.java +++ b/samples/webmvc-http-security/src/main/java/io/spring/sample/graphql/SampleApplication.java @@ -34,9 +34,9 @@ public class SampleApplication { @Bean public WebInterceptor interceptor() { - return (input, next) -> { + return (webInput, interceptorChain) -> { // Switch threads to prove ThreadLocal context propagation works - return Mono.delay(Duration.ofMillis(10)).flatMap(aLong -> next.handle(input)); + return Mono.delay(Duration.ofMillis(10)).flatMap(aLong -> interceptorChain.next(webInput)); }; } diff --git a/spring-graphql-docs/src/docs/asciidoc/index.adoc b/spring-graphql-docs/src/docs/asciidoc/index.adoc index 387c16f7..46d6a547 100644 --- a/spring-graphql-docs/src/docs/asciidoc/index.adoc +++ b/spring-graphql-docs/src/docs/asciidoc/index.adoc @@ -115,12 +115,12 @@ transformation of the `graphql.ExecutionInput`: class MyInterceptor implements WebInterceptor { @Override - public Mono intercept(WebInput webInput, WebGraphQlHandler next) { + public Mono intercept(WebInput webInput, WebInterceptorChain chain) { webInput.configureExecutionInput((executionInput, builder) -> { Map map = ... ; return builder.extensions(map).build(); }); - return next.handle(webInput); + return chain.next(webInput); } } ---- @@ -133,8 +133,8 @@ the `graphql.ExecutionResult`: class MyInterceptor implements WebInterceptor { @Override - public Mono intercept(WebInput webInput, WebGraphQlHandler next) { - return next.handle(webInput) + public Mono intercept(WebInput webInput, WebInterceptorChain chain) { + return chain.next(webInput) .map(webOutput -> { Object data = webOutput.getData(); Object updatedData = ... ; diff --git a/spring-graphql-test/src/main/java/org/springframework/graphql/test/tester/WebGraphQlHandlerRequestStrategy.java b/spring-graphql-test/src/main/java/org/springframework/graphql/test/tester/WebGraphQlHandlerRequestStrategy.java index ad94b044..4e704014 100644 --- a/spring-graphql-test/src/main/java/org/springframework/graphql/test/tester/WebGraphQlHandlerRequestStrategy.java +++ b/spring-graphql-test/src/main/java/org/springframework/graphql/test/tester/WebGraphQlHandlerRequestStrategy.java @@ -62,7 +62,7 @@ class WebGraphQlHandlerRequestStrategy extends DirectRequestStrategySupport impl } private WebOutput executeInternal(WebInput webInput) { - WebOutput webOutput = this.graphQlHandler.handle(webInput).block(getResponseTimeout()); + WebOutput webOutput = this.graphQlHandler.handleRequest(webInput).block(getResponseTimeout()); Assert.notNull(webOutput, "Expected WebOutput"); return webOutput; } diff --git a/spring-graphql-test/src/test/java/org/springframework/graphql/test/tester/WebGraphQlTesterTests.java b/spring-graphql-test/src/test/java/org/springframework/graphql/test/tester/WebGraphQlTesterTests.java index 59096515..619843f7 100644 --- a/spring-graphql-test/src/test/java/org/springframework/graphql/test/tester/WebGraphQlTesterTests.java +++ b/spring-graphql-test/src/test/java/org/springframework/graphql/test/tester/WebGraphQlTesterTests.java @@ -272,7 +272,7 @@ public class WebGraphQlTesterTests { } ExecutionResult result = builder.build(); WebOutput output = new WebOutput(mock(WebInput.class), result); - given(this.handler.handle(this.bodyCaptor.capture())).willReturn(Mono.just(output)); + given(this.handler.handleRequest(this.bodyCaptor.capture())).willReturn(Mono.just(output)); } @Override diff --git a/spring-graphql/src/main/java/org/springframework/graphql/web/DefaultWebGraphQlHandlerBuilder.java b/spring-graphql/src/main/java/org/springframework/graphql/web/DefaultWebGraphQlHandlerBuilder.java index b6f323e8..48f1da19 100644 --- a/spring-graphql/src/main/java/org/springframework/graphql/web/DefaultWebGraphQlHandlerBuilder.java +++ b/spring-graphql/src/main/java/org/springframework/graphql/web/DefaultWebGraphQlHandlerBuilder.java @@ -20,6 +20,8 @@ import java.util.ArrayList; import java.util.Arrays; import java.util.Collections; import java.util.List; +import java.util.Map; +import java.util.stream.Collectors; import reactor.core.publisher.Mono; @@ -80,21 +82,69 @@ class DefaultWebGraphQlHandlerBuilder implements WebGraphQlHandler.Builder { @Override public WebGraphQlHandler build() { + List interceptorsToUse = (this.interceptors != null) ? this.interceptors : Collections.emptyList(); - WebGraphQlHandler targetHandler = (webInput) -> - this.service.execute(webInput).map((result) -> new WebOutput(webInput, result)); + WebInterceptorChain interceptorChain = initWebInterceptorChain(interceptorsToUse); + WebSocketInterceptor webSocketInterceptor = initWebSocketInterceptor(interceptorsToUse); - WebGraphQlHandler interceptionChain = interceptorsToUse.stream() - .reduce(WebInterceptor::andThen) - .map((interceptor) -> (WebGraphQlHandler) (input) -> interceptor.intercept(input, targetHandler)) - .orElse(targetHandler); + WebGraphQlHandler graphQlHandler = new WebGraphQlHandler() { - return (CollectionUtils.isEmpty(this.accessors) ? interceptionChain - : new ThreadLocalExtractingHandler(interceptionChain, ThreadLocalAccessor.composite(this.accessors))); + @Override + public Mono handleRequest(WebInput input) { + return interceptorChain.next(input); + } + + @Override + public Mono handleWebSocketInitialization(Map payload) { + return (webSocketInterceptor != null ? + webSocketInterceptor.handleConnectionInitialization(payload) : Mono.empty()); + } + + @Override + public Mono handleWebSocketCompletion() { + return (webSocketInterceptor != null ? + webSocketInterceptor.handleConnectionCompletion() : Mono.empty()); + } + }; + + if (!CollectionUtils.isEmpty(this.accessors)) { + graphQlHandler = new ThreadLocalExtractingHandler( + graphQlHandler, ThreadLocalAccessor.composite(this.accessors)); + } + + return graphQlHandler; } + private WebInterceptorChain initWebInterceptorChain(List interceptors) { + + WebInterceptorChain targetHandler = + webInput -> service.execute(webInput).map((result) -> new WebOutput(webInput, result)); + + return interceptors.stream() + .reduce(WebInterceptor::andThen) + .map((interceptor) -> (WebInterceptorChain) (input) -> interceptor.intercept(input, targetHandler)) + .orElse(targetHandler); + } + + @Nullable + private WebSocketInterceptor initWebSocketInterceptor(List interceptors) { + + List filtered = interceptors.stream() + .filter(current -> current instanceof WebSocketInterceptor) + .map(current -> (WebSocketInterceptor) current) + .collect(Collectors.toList()); + + if (filtered.size() > 1) { + throw new IllegalArgumentException( + "There can be at most 1 WebSocketInterceptor. Found " + filtered.size() + "."); + } + + return (!filtered.isEmpty() ? filtered.get(0) : null); + } + + /** * {@link WebGraphQlHandler} that extracts ThreadLocal values and saves them in the * Reactor context for subsequent use for DataFetcher's. @@ -111,8 +161,20 @@ class DefaultWebGraphQlHandlerBuilder implements WebGraphQlHandler.Builder { } @Override - public Mono handle(WebInput input) { - return this.delegate.handle(input).contextWrite((context) -> + public Mono handleRequest(WebInput input) { + return this.delegate.handleRequest(input).contextWrite((context) -> + ReactorContextManager.extractThreadLocalValues(this.accessor, context)); + } + + @Override + public Mono handleWebSocketInitialization(Map payload) { + return this.delegate.handleWebSocketInitialization(payload).contextWrite((context) -> + ReactorContextManager.extractThreadLocalValues(this.accessor, context)); + } + + @Override + public Mono handleWebSocketCompletion() { + return this.delegate.handleWebSocketCompletion().contextWrite((context) -> ReactorContextManager.extractThreadLocalValues(this.accessor, context)); } diff --git a/spring-graphql/src/main/java/org/springframework/graphql/web/WebGraphQlHandler.java b/spring-graphql/src/main/java/org/springframework/graphql/web/WebGraphQlHandler.java index 9a43e13f..fe29b457 100644 --- a/spring-graphql/src/main/java/org/springframework/graphql/web/WebGraphQlHandler.java +++ b/spring-graphql/src/main/java/org/springframework/graphql/web/WebGraphQlHandler.java @@ -17,6 +17,7 @@ package org.springframework.graphql.web; import java.util.List; +import java.util.Map; import reactor.core.publisher.Mono; @@ -24,8 +25,8 @@ import org.springframework.graphql.GraphQlService; import org.springframework.graphql.execution.ThreadLocalAccessor; /** - * Common contract to handle a GraphQL request over HTTP or WebSocket for use - * with both Spring MVC and Spring WebFlux. + * Contract for common handling of a GraphQL request received over HTTP or + * WebSocket, and executed on Spring MVC or Spring WebFlux. * * @author Rossen Stoyanchev * @since 1.0.0 @@ -33,11 +34,32 @@ import org.springframework.graphql.execution.ThreadLocalAccessor; public interface WebGraphQlHandler { /** - * Perform request execution for the given input and return the result. + * Execute the given request and return the resulting output. * @param input the GraphQL request input container - * @return the execution result + * @return the result from execution */ - Mono handle(WebInput input); + Mono handleRequest(WebInput input); + + /** + * Handle the payload from the connection initialization message that a + * GraphQL over WebSocket client must send after the WebSocket session is + * established and before sending any requests. + * @param payload the payload from the {@code ConnectionInit} message + * @return an optional payload for the {@code ConnectionAck} message + */ + default Mono handleWebSocketInitialization(Map payload) { + return Mono.empty(); + } + + /** + * Handle the completion message that a GraphQL over WebSocket clients sends + * before closing the WebSocket connection. + * @return signals the end of completion handling + */ + default Mono handleWebSocketCompletion() { + return Mono.empty(); + } + /** * Provides access to a builder to create a {@link WebGraphQlHandler} instance. @@ -49,6 +71,7 @@ public interface WebGraphQlHandler { return new DefaultWebGraphQlHandlerBuilder(graphQlService); } + /** * Builder for a {@link WebGraphQlHandler} that executes a * {@link WebInterceptor} chain followed by a {@link GraphQlService}. diff --git a/spring-graphql/src/main/java/org/springframework/graphql/web/WebInterceptor.java b/spring-graphql/src/main/java/org/springframework/graphql/web/WebInterceptor.java index 6a91321d..31f52b54 100644 --- a/spring-graphql/src/main/java/org/springframework/graphql/web/WebInterceptor.java +++ b/spring-graphql/src/main/java/org/springframework/graphql/web/WebInterceptor.java @@ -42,13 +42,13 @@ public interface WebInterceptor { /** * Intercept a request and delegate for further handling and request execution via - * {@link WebGraphQlHandler#handle(WebInput)}. + * {@link WebGraphQlHandler#handleRequest(WebInput)}. * @param webInput container with HTTP request information and options to customize * the {@link ExecutionInput}. - * @param next the handler to delegate to for request execution + * @param next the rest of the chain to delegate to for request execution * @return a {@link Mono} with the result */ - Mono intercept(WebInput webInput, WebGraphQlHandler next); + Mono intercept(WebInput webInput, WebInterceptorChain next); /** * Return a composed {@link WebInterceptor} that invokes the current interceptor first diff --git a/spring-graphql/src/main/java/org/springframework/graphql/web/WebInterceptorChain.java b/spring-graphql/src/main/java/org/springframework/graphql/web/WebInterceptorChain.java new file mode 100644 index 00000000..ccc6cbd4 --- /dev/null +++ b/spring-graphql/src/main/java/org/springframework/graphql/web/WebInterceptorChain.java @@ -0,0 +1,37 @@ +/* + * Copyright 2002-2021 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.web; + +import reactor.core.publisher.Mono; + +/** + * Contract that allows a {@link WebInterceptor} to delegate to the remainder + * of the chain. + * + * @author Rossen Stoyanchev + * @since 1.0.0 + */ +public interface WebInterceptorChain { + + /** + * Delegate to the next rest of the chain which can consist of more + * {@code WebInterceptor} instances, and a {@link WebGraphQlHandler}. + * @param webInput the input for the request + * @return the output with the result from request execution + */ + Mono next(WebInput webInput); + +} diff --git a/spring-graphql/src/main/java/org/springframework/graphql/web/WebSocketInterceptor.java b/spring-graphql/src/main/java/org/springframework/graphql/web/WebSocketInterceptor.java new file mode 100644 index 00000000..2cb6e55f --- /dev/null +++ b/spring-graphql/src/main/java/org/springframework/graphql/web/WebSocketInterceptor.java @@ -0,0 +1,57 @@ +/* + * Copyright 2002-2021 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.web; + +import java.util.Map; + +import reactor.core.publisher.Mono; + +/** + * An extension of {@link WebInterceptor} with additional methods to handle the + * start and end of a WebSocket connection. Only a single interceptor of type + * {@link WebSocketInterceptor} may be declared. + * + * @author Rossen Stoyanchev + * @since 1.0.0 + */ +public interface WebSocketInterceptor extends WebInterceptor { + + @Override + default Mono intercept(WebInput webInput, WebInterceptorChain next) { + return next.next(webInput); + } + + /** + * Handle the payload from the connection initialization message that a + * GraphQL over WebSocket client must send after the WebSocket session is + * established and before sending any requests. + * @param payload the payload from the {@code ConnectionInit} message + * @return an optional payload for the {@code ConnectionAck} message + */ + default Mono handleConnectionInitialization(Map payload) { + return Mono.empty(); + } + + /** + * Handle the completion message that a GraphQL over WebSocket clients sends + * before closing the WebSocket connection. + * @return signals the end of completion handling + */ + default Mono handleConnectionCompletion() { + return Mono.empty(); + } + +} diff --git a/spring-graphql/src/main/java/org/springframework/graphql/web/webflux/GraphQlHttpHandler.java b/spring-graphql/src/main/java/org/springframework/graphql/web/webflux/GraphQlHttpHandler.java index 095fd591..6ab94779 100644 --- a/spring-graphql/src/main/java/org/springframework/graphql/web/webflux/GraphQlHttpHandler.java +++ b/spring-graphql/src/main/java/org/springframework/graphql/web/webflux/GraphQlHttpHandler.java @@ -66,7 +66,7 @@ public class GraphQlHttpHandler { if (logger.isDebugEnabled()) { logger.debug("Executing: " + input); } - return this.graphQlHandler.handle(input); + return this.graphQlHandler.handleRequest(input); }) .flatMap((output) -> { Map spec = output.toSpecification(); diff --git a/spring-graphql/src/main/java/org/springframework/graphql/web/webflux/GraphQlWebSocketHandler.java b/spring-graphql/src/main/java/org/springframework/graphql/web/webflux/GraphQlWebSocketHandler.java index cd0af25f..c6457343 100644 --- a/spring-graphql/src/main/java/org/springframework/graphql/web/webflux/GraphQlWebSocketHandler.java +++ b/spring-graphql/src/main/java/org/springframework/graphql/web/webflux/GraphQlWebSocketHandler.java @@ -18,6 +18,7 @@ package org.springframework.graphql.web.webflux; import java.time.Duration; import java.util.Arrays; +import java.util.Collections; import java.util.HashMap; import java.util.List; import java.util.Map; @@ -86,12 +87,13 @@ public class GraphQlWebSocketHandler implements WebSocketHandler { /** * Create a new instance. - * @param graphQlHandler common handler for GraphQL over HTTP requests + * @param graphQlHandler common handler for GraphQL over WebSocket requests * @param configurer codec configurer for JSON encoding and decoding * @param connectionInitTimeout the time within which the {@code CONNECTION_INIT} type * message must be received. */ - public GraphQlWebSocketHandler(WebGraphQlHandler graphQlHandler, ServerCodecConfigurer configurer, + public GraphQlWebSocketHandler( + WebGraphQlHandler graphQlHandler, ServerCodecConfigurer configurer, Duration connectionInitTimeout) { Assert.notNull(graphQlHandler, "WebGraphQlHandler is required"); @@ -163,7 +165,7 @@ public class GraphQlWebSocketHandler implements WebSocketHandler { if (logger.isDebugEnabled()) { logger.debug("Executing: " + input); } - return this.graphQlHandler.handle(input) + return this.graphQlHandler.handleRequest(input) .flatMapMany((output) -> handleWebOutput(session, id, subscriptions, output)) .doOnTerminate(() -> subscriptions.remove(id)); case COMPLETE: @@ -173,12 +175,15 @@ public class GraphQlWebSocketHandler implements WebSocketHandler { subscription.cancel(); } } - return Flux.empty(); + return this.graphQlHandler.handleWebSocketCompletion().thenMany(Flux.empty()); case CONNECTION_INIT: if (!connectionInitProcessed.compareAndSet(false, true)) { return GraphQlStatus.close(session, GraphQlStatus.TOO_MANY_INIT_REQUESTS_STATUS); } - return Flux.just(encode(session, null, MessageType.CONNECTION_ACK, null)); + return this.graphQlHandler.handleWebSocketInitialization(getPayload(map)) + .defaultIfEmpty(Collections.emptyMap()) + .flatMapMany(ackPayload -> Flux.just(encode(session, null, MessageType.CONNECTION_ACK, ackPayload))) + .onErrorResume(ex -> GraphQlStatus.close(session, GraphQlStatus.UNAUTHORIZED_STATUS)); default: return GraphQlStatus.close(session, GraphQlStatus.INVALID_MESSAGE_STATUS); } @@ -194,8 +199,7 @@ public class GraphQlWebSocketHandler implements WebSocketHandler { @SuppressWarnings("unchecked") private static Map getPayload(Map message) { Map payload = (Map) message.get("payload"); - Assert.notNull(payload, "No \"payload\" in message: " + message); - return payload; + return (payload != null ? payload : Collections.emptyMap()); } @SuppressWarnings("unchecked") diff --git a/spring-graphql/src/main/java/org/springframework/graphql/web/webmvc/GraphQlHttpHandler.java b/spring-graphql/src/main/java/org/springframework/graphql/web/webmvc/GraphQlHttpHandler.java index 75b3fc2a..768d4caa 100644 --- a/spring-graphql/src/main/java/org/springframework/graphql/web/webmvc/GraphQlHttpHandler.java +++ b/spring-graphql/src/main/java/org/springframework/graphql/web/webmvc/GraphQlHttpHandler.java @@ -72,7 +72,7 @@ public class GraphQlHttpHandler { if (logger.isDebugEnabled()) { logger.debug("Executing: " + input); } - Mono responseMono = this.graphQlHandler.handle(input).map((output) -> { + Mono responseMono = this.graphQlHandler.handleRequest(input).map((output) -> { if (logger.isDebugEnabled()) { logger.debug("Execution complete"); } diff --git a/spring-graphql/src/main/java/org/springframework/graphql/web/webmvc/GraphQlWebSocketHandler.java b/spring-graphql/src/main/java/org/springframework/graphql/web/webmvc/GraphQlWebSocketHandler.java index 6aad8607..c5c4d19d 100644 --- a/spring-graphql/src/main/java/org/springframework/graphql/web/webmvc/GraphQlWebSocketHandler.java +++ b/spring-graphql/src/main/java/org/springframework/graphql/web/webmvc/GraphQlWebSocketHandler.java @@ -24,6 +24,7 @@ import java.io.OutputStream; import java.net.URI; import java.time.Duration; import java.util.Arrays; +import java.util.Collections; import java.util.HashMap; import java.util.List; import java.util.Map; @@ -86,12 +87,13 @@ public class GraphQlWebSocketHandler extends TextWebSocketHandler implements Sub /** * Create a new instance. - * @param graphQlHandler common handler for GraphQL over HTTP requests + * @param graphQlHandler common handler for GraphQL over WebSocket requests * @param converter for JSON encoding and decoding * @param connectionInitTimeout the time within which the {@code CONNECTION_INIT} type * message must be received. */ - public GraphQlWebSocketHandler(WebGraphQlHandler graphQlHandler, HttpMessageConverter converter, + public GraphQlWebSocketHandler( + WebGraphQlHandler graphQlHandler, HttpMessageConverter converter, Duration connectionInitTimeout) { Assert.notNull(graphQlHandler, "WebGraphQlHandler is required"); @@ -158,7 +160,7 @@ public class GraphQlWebSocketHandler extends TextWebSocketHandler implements Sub if (logger.isDebugEnabled()) { logger.debug("Executing: " + input); } - this.graphQlHandler.handle(input) + this.graphQlHandler.handleRequest(input) .flatMapMany((output) -> handleWebOutput(session, input.getId(), output)) .publishOn(sessionState.getScheduler()) // Serial blocking send via single thread .subscribe(new SendMessageSubscriber(id, session, sessionState)); @@ -170,14 +172,30 @@ public class GraphQlWebSocketHandler extends TextWebSocketHandler implements Sub subscription.cancel(); } } + this.graphQlHandler.handleWebSocketCompletion().block(Duration.ofSeconds(10)); return; case CONNECTION_INIT: if (sessionState.setConnectionInitProcessed()) { GraphQlStatus.closeSession(session, GraphQlStatus.TOO_MANY_INIT_REQUESTS_STATUS); return; } - TextMessage outputMessage = encode(null, MessageType.CONNECTION_ACK, null); - session.sendMessage(outputMessage); + this.graphQlHandler.handleWebSocketInitialization(getPayload(map)) + .defaultIfEmpty(Collections.emptyMap()) + .publishOn(sessionState.getScheduler()) // Serial blocking send via single thread + .doOnNext(ackPayload -> { + TextMessage outputMessage = encode(null, MessageType.CONNECTION_ACK, ackPayload); + try { + session.sendMessage(outputMessage); + } + catch (IOException ex) { + throw new IllegalStateException(ex); + } + }) + .onErrorResume(ex -> { + GraphQlStatus.closeSession(session, GraphQlStatus.UNAUTHORIZED_STATUS); + return Mono.empty(); + }) + .block(Duration.ofSeconds(10)); return; default: GraphQlStatus.closeSession(session, GraphQlStatus.INVALID_MESSAGE_STATUS); @@ -193,8 +211,7 @@ public class GraphQlWebSocketHandler extends TextWebSocketHandler implements Sub @SuppressWarnings("unchecked") private static Map getPayload(Map message) { Map payload = (Map) message.get("payload"); - Assert.notNull(payload, "No \"payload\" in message: " + message); - return payload; + return (payload != null ? payload : Collections.emptyMap()); } private SessionState getSessionInfo(WebSocketSession session) { diff --git a/spring-graphql/src/test/java/org/springframework/graphql/data/querydsl/QuerydslDataFetcherTests.java b/spring-graphql/src/test/java/org/springframework/graphql/data/querydsl/QuerydslDataFetcherTests.java index b20fd82f..49ce15b4 100644 --- a/spring-graphql/src/test/java/org/springframework/graphql/data/querydsl/QuerydslDataFetcherTests.java +++ b/spring-graphql/src/test/java/org/springframework/graphql/data/querydsl/QuerydslDataFetcherTests.java @@ -66,7 +66,7 @@ class QuerydslDataFetcherTests { BiConsumer, QuerydslPredicateExecutor> tester = (wiringConfigurer, executor) -> { WebGraphQlHandler handler = initWebGraphQlHandler(wiringConfigurer, executor, null); - WebOutput output = handler.handle(input("{ bookById(id: 1) {name}}")).block(); + WebOutput output = handler.handleRequest(input("{ bookById(id: 1) {name}}")).block(); // TODO: getData interferes with method overrides assertThat((Object) output.getData()).isEqualTo( @@ -93,7 +93,7 @@ class QuerydslDataFetcherTests { BiConsumer, QuerydslPredicateExecutor> tester = (wiringConfigurer, executor) -> { WebGraphQlHandler handler = initWebGraphQlHandler(wiringConfigurer, mockRepository, null); - WebOutput output = handler.handle(input("{ books {name}}")).block(); + WebOutput output = handler.handleRequest(input("{ books {name}}")).block(); assertThat((Object) output.getData()).isEqualTo( Collections.singletonMap("books", Arrays.asList( @@ -118,7 +118,7 @@ class QuerydslDataFetcherTests { // 1) Automatic registration only WebGraphQlHandler handler = initWebGraphQlHandler(null, mockRepository, null); - WebOutput output = handler.handle(input("{ bookById(id: 1) {name}}")).block(); + WebOutput output = handler.handleRequest(input("{ bookById(id: 1) {name}}")).block(); assertThat((Object) output.getData()).isEqualTo( Collections.singletonMap("bookById", Collections.singletonMap("name", "Hitchhiker's Guide to the Galaxy"))); @@ -128,7 +128,7 @@ class QuerydslDataFetcherTests { builder -> builder.dataFetcher("bookById", env -> new Book(53L, "Breaking Bad", "Heisenberg")), mockRepository, null); - output = handler.handle(input("{ bookById(id: 1) {name}}")).block(); + output = handler.handleRequest(input("{ bookById(id: 1) {name}}")).block(); assertThat((Object) output.getData()).isEqualTo( Collections.singletonMap("bookById", Collections.singletonMap("name", "Breaking Bad"))); @@ -146,7 +146,7 @@ class QuerydslDataFetcherTests { .projectAs(BookProjection.class) .single())); - WebOutput output = handler.handle(input("{ bookById(id: 1) {name}}")).block(); + WebOutput output = handler.handleRequest(input("{ bookById(id: 1) {name}}")).block(); assertThat((Object) output.getData()).isEqualTo( Collections.singletonMap("bookById", @@ -165,7 +165,7 @@ class QuerydslDataFetcherTests { .projectAs(BookDto.class) .single())); - WebOutput output = handler.handle(input("{ bookById(id: 1) {name}}")).block(); + WebOutput output = handler.handleRequest(input("{ bookById(id: 1) {name}}")).block(); assertThat((Object) output.getData()).isEqualTo( Collections.singletonMap("bookById", @@ -183,7 +183,7 @@ class QuerydslDataFetcherTests { bindings.bind(book.name).firstOptional((path, value) -> value.map(path::startsWith))) .many())); - handler.handle(input("{ books(name: \"H\", author: \"Doug\") {name}}")).block(); + handler.handleRequest(input("{ books(name: \"H\", author: \"Doug\") {name}}")).block(); ArgumentCaptor predicateCaptor = ArgumentCaptor.forClass(Predicate.class); @@ -202,7 +202,7 @@ class QuerydslDataFetcherTests { BiConsumer, ReactiveQuerydslPredicateExecutor> tester = (wiringConfigurer, executor) -> { WebGraphQlHandler handler = initWebGraphQlHandler(wiringConfigurer, null, executor); - WebOutput output = handler.handle(input("{ bookById(id: 1) {name}}")).block(); + WebOutput output = handler.handleRequest(input("{ bookById(id: 1) {name}}")).block(); // TODO: getData interferes with method overrides assertThat((Object) output.getData()).isEqualTo( @@ -229,7 +229,7 @@ class QuerydslDataFetcherTests { BiConsumer, ReactiveQuerydslPredicateExecutor> tester = (wiringConfigurer, executor) -> { WebGraphQlHandler handler = initWebGraphQlHandler(wiringConfigurer, null, mockRepository); - WebOutput output = handler.handle(input("{ books {name}}")).block(); + WebOutput output = handler.handleRequest(input("{ books {name}}")).block(); assertThat((Object) output.getData()).isEqualTo( Collections.singletonMap("books", Arrays.asList( diff --git a/spring-graphql/src/test/java/org/springframework/graphql/web/ConsumeOneAndNeverCompleteInterceptor.java b/spring-graphql/src/test/java/org/springframework/graphql/web/ConsumeOneAndNeverCompleteInterceptor.java index 6e6e326a..3607bbd3 100644 --- a/spring-graphql/src/test/java/org/springframework/graphql/web/ConsumeOneAndNeverCompleteInterceptor.java +++ b/spring-graphql/src/test/java/org/springframework/graphql/web/ConsumeOneAndNeverCompleteInterceptor.java @@ -25,8 +25,8 @@ import static org.assertj.core.api.Assertions.assertThat; public class ConsumeOneAndNeverCompleteInterceptor implements WebInterceptor { @Override - public Mono intercept(WebInput webInput, WebGraphQlHandler next) { - return next.handle(webInput).map((output) -> output.transform((builder) -> { + public Mono intercept(WebInput webInput, WebInterceptorChain next) { + return next.next(webInput).map((output) -> output.transform((builder) -> { Publisher publisher = output.getData(); assertThat(publisher).isNotNull(); builder.data(Flux.from(publisher).take(1).concatWith(Flux.never())); diff --git a/spring-graphql/src/test/java/org/springframework/graphql/web/WebGraphQlHandlerTests.java b/spring-graphql/src/test/java/org/springframework/graphql/web/WebGraphQlHandlerTests.java index 7918a166..0bc5f5c5 100644 --- a/spring-graphql/src/test/java/org/springframework/graphql/web/WebGraphQlHandlerTests.java +++ b/spring-graphql/src/test/java/org/springframework/graphql/web/WebGraphQlHandlerTests.java @@ -62,7 +62,7 @@ public class WebGraphQlHandlerTests { GraphQlService service = new ExecutionGraphQlService(new TestGraphQlSource(graphQl)); WebGraphQlHandler handler = WebGraphQlHandler.builder(service).build(); - WebOutput webOutput = handler.handle(webInput).contextWrite((context) -> context.put("name", "007")).block(); + WebOutput webOutput = handler.handleRequest(webInput).contextWrite((context) -> context.put("name", "007")).block(); Map data = webOutput.getData(); assertThat(data).hasSize(1).containsEntry("greeting", "Hello 007"); @@ -82,7 +82,7 @@ public class WebGraphQlHandlerTests { GraphQlService service = new ExecutionGraphQlService(new TestGraphQlSource(graphQl)); WebGraphQlHandler handler = WebGraphQlHandler.builder(service).build(); - WebOutput webOutput = handler.handle(webInput).contextWrite((context) -> context.put("name", "007")).block(); + WebOutput webOutput = handler.handleRequest(webInput).contextWrite((context) -> context.put("name", "007")).block(); Map data = webOutput.getData(); assertThat(data).hasSize(1).containsEntry("greeting", null); @@ -105,11 +105,11 @@ public class WebGraphQlHandlerTests { GraphQlService service = new ExecutionGraphQlService(new TestGraphQlSource(graphQl)); WebGraphQlHandler handler = WebGraphQlHandler.builder(service) - .interceptor((input, next) -> Mono.delay(Duration.ofMillis(10)).flatMap((aLong) -> next.handle(input))) + .interceptor((input, next) -> Mono.delay(Duration.ofMillis(10)).flatMap((aLong) -> next.next(input))) .threadLocalAccessor(threadLocalAccessor) .build(); - Map data = handler.handle(webInput).block().getData(); + Map data = handler.handleRequest(webInput).block().getData(); assertThat(data).hasSize(1).containsEntry("greeting", "Hello 007"); } @@ -136,11 +136,11 @@ public class WebGraphQlHandlerTests { GraphQlService service = new ExecutionGraphQlService(new TestGraphQlSource(graphQl)); WebGraphQlHandler handler = WebGraphQlHandler.builder(service) - .interceptor((input, next) -> Mono.delay(Duration.ofMillis(10)).flatMap((aLong) -> next.handle(input))) + .interceptor((input, next) -> Mono.delay(Duration.ofMillis(10)).flatMap((aLong) -> next.next(input))) .threadLocalAccessor(threadLocalAccessor) .build(); - WebOutput webOutput = handler.handle(webInput).block(); + WebOutput webOutput = handler.handleRequest(webInput).block(); List errors = webOutput.getErrors(); assertThat(errors.get(0).getMessage()).isEqualTo("Resolved error: Invalid greeting, name=007"); diff --git a/spring-graphql/src/test/java/org/springframework/graphql/web/WebInterceptorTests.java b/spring-graphql/src/test/java/org/springframework/graphql/web/WebInterceptorTests.java index a1cff666..9894adb6 100644 --- a/spring-graphql/src/test/java/org/springframework/graphql/web/WebInterceptorTests.java +++ b/spring-graphql/src/test/java/org/springframework/graphql/web/WebInterceptorTests.java @@ -51,7 +51,7 @@ public class WebInterceptorTests { new OrderInterceptor(3, output))) .build(); - handler.handle(webInput).block(); + handler.handleRequest(webInput).block(); assertThat(output.toString()).isEqualTo(":pre1:pre2:pre3:post3:post2:post1"); } @@ -61,10 +61,10 @@ public class WebInterceptorTests { output.transform((builder) -> builder.responseHeader("testHeader", "testValue")); WebGraphQlHandler handler = WebGraphQlHandler.builder((input) -> emptyExecutionResult()) - .interceptor((input, next) -> next.handle(input).map(headerFunction)) + .interceptor((input, next) -> next.next(input).map(headerFunction)) .build(); - HttpHeaders headers = handler.handle(webInput).block().getResponseHeaders(); + HttpHeaders headers = handler.handleRequest(webInput).block().getResponseHeaders(); assertThat(headers.get("testHeader")).containsExactly("testValue"); } @@ -80,11 +80,11 @@ public class WebInterceptorTests { }) .interceptor((webInput, next) -> { webInput.configureExecutionInput((input, builder) -> builder.operationName("testOp").build()); - return next.handle(webInput); + return next.next(webInput); }) .build(); - handler.handle(webInput).block(); + handler.handleRequest(webInput).block(); assertThat(actualName.get()).isEqualTo("testOp"); } @@ -105,9 +105,9 @@ public class WebInterceptorTests { } @Override - public Mono intercept(WebInput input, WebGraphQlHandler next) { + public Mono intercept(WebInput input, WebInterceptorChain next) { this.output.append(":pre").append(this.order); - return next.handle(input) + return next.next(input) .map((output) -> { this.output.append(":post").append(this.order); return output; diff --git a/spring-graphql/src/test/java/org/springframework/graphql/web/webflux/GraphQlWebSocketHandlerTests.java b/spring-graphql/src/test/java/org/springframework/graphql/web/webflux/GraphQlWebSocketHandlerTests.java index 4c377d69..1d0fbf70 100644 --- a/spring-graphql/src/test/java/org/springframework/graphql/web/webflux/GraphQlWebSocketHandlerTests.java +++ b/spring-graphql/src/test/java/org/springframework/graphql/web/webflux/GraphQlWebSocketHandlerTests.java @@ -27,6 +27,7 @@ import java.util.function.BiConsumer; import org.assertj.core.api.InstanceOfAssertFactories; import org.junit.jupiter.api.Test; import reactor.core.publisher.Flux; +import reactor.core.publisher.Mono; import reactor.core.publisher.Sinks; import reactor.test.StepVerifier; @@ -36,6 +37,7 @@ import org.springframework.core.io.buffer.DefaultDataBufferFactory; import org.springframework.graphql.web.BookTestUtils; import org.springframework.graphql.web.ConsumeOneAndNeverCompleteInterceptor; import org.springframework.graphql.web.WebInterceptor; +import org.springframework.graphql.web.WebSocketInterceptor; import org.springframework.http.codec.ServerCodecConfigurer; import org.springframework.http.codec.json.Jackson2JsonDecoder; import org.springframework.web.reactive.socket.CloseStatus; @@ -123,6 +125,47 @@ public class GraphQlWebSocketHandlerTests { .verifyComplete(); } + @Test + @SuppressWarnings("unchecked") + void connectionInitHandling() { + TestWebSocketSession session = handle( + Flux.just(toWebSocketMessage("{\"type\":\"connection_init\",\"payload\":{\"key\":\"A\"}}")), + new WebSocketInterceptor() { + + @Override + public Mono handleConnectionInitialization(Map payload) { + Object value = payload.get("key"); + return Mono.just(Collections.singletonMap("key", value + " acknowledged")); + } + }); + + StepVerifier.create(session.getOutput()) + .consumeNextWith((message) -> { + Map content = decode(message); + assertThat(content).containsEntry("type", "connection_ack"); + assertThat((Map) content.get("payload")).containsEntry("key", "A acknowledged"); + }) + .verifyComplete(); + } + + @Test + void connectionInitRejected() { + TestWebSocketSession session = handle( + Flux.just(toWebSocketMessage("{\"type\":\"connection_init\"}")), + new WebSocketInterceptor() { + + @Override + public Mono handleConnectionInitialization(Map payload) { + return Mono.error(new IllegalStateException()); + } + }); + + StepVerifier.create(session.getOutput()).verifyComplete(); + StepVerifier.create(session.closeStatus()) + .expectNext(new CloseStatus(4401, "Unauthorized")) + .verifyComplete(); + } + @Test void unauthorizedWithoutConnectionInit() { TestWebSocketSession session = handle(Flux.just(toWebSocketMessage(BookTestUtils.BOOK_SUBSCRIPTION))); diff --git a/spring-graphql/src/test/java/org/springframework/graphql/web/webmvc/GraphQlWebSocketHandlerTests.java b/spring-graphql/src/test/java/org/springframework/graphql/web/webmvc/GraphQlWebSocketHandlerTests.java index 1ad0c24c..a07292de 100644 --- a/spring-graphql/src/test/java/org/springframework/graphql/web/webmvc/GraphQlWebSocketHandlerTests.java +++ b/spring-graphql/src/test/java/org/springframework/graphql/web/webmvc/GraphQlWebSocketHandlerTests.java @@ -21,6 +21,7 @@ import java.io.IOException; import java.io.InputStream; 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; @@ -28,11 +29,14 @@ import java.util.function.Consumer; import org.assertj.core.api.InstanceOfAssertFactories; import org.junit.jupiter.api.Test; +import reactor.core.publisher.Flux; +import reactor.core.publisher.Mono; import reactor.test.StepVerifier; import org.springframework.graphql.web.BookTestUtils; import org.springframework.graphql.web.ConsumeOneAndNeverCompleteInterceptor; import org.springframework.graphql.web.WebInterceptor; +import org.springframework.graphql.web.WebSocketInterceptor; import org.springframework.http.HttpHeaders; import org.springframework.http.HttpInputMessage; import org.springframework.http.converter.HttpMessageConverter; @@ -123,6 +127,50 @@ public class GraphQlWebSocketHandlerTests { assertThat(this.session.getCloseStatus()).isEqualTo(new CloseStatus(4400, "Invalid message")); } + @Test + @SuppressWarnings("unchecked") + void connectionInitHandling() throws Exception { + + WebSocketInterceptor interceptor = new WebSocketInterceptor() { + + @Override + public Mono handleConnectionInitialization(Map payload) { + Object value = payload.get("key"); + return Mono.just(Collections.singletonMap("key", value + " acknowledged")); + } + }; + + handle(initWebSocketHandler(interceptor), + new TextMessage("{\"type\":\"connection_init\",\"payload\":{\"key\":\"A\"}}")); + + StepVerifier.create(session.getOutput()) + .consumeNextWith((message) -> { + Map content = decode(message); + assertThat(content).containsEntry("type", "connection_ack"); + assertThat((Map) content.get("payload")).containsEntry("key", "A acknowledged"); + }) + .then(this.session::close) // Complete output Flux + .verifyComplete(); + } + + @Test + void connectionInitRejected() throws Exception { + + WebSocketInterceptor interceptor = new WebSocketInterceptor() { + + @Override + public Mono handleConnectionInitialization(Map payload) { + return Mono.error(new IllegalStateException()); + } + }; + + handle(initWebSocketHandler(interceptor), new TextMessage("{\"type\":\"connection_init\"}")); + + StepVerifier.create(session.closeStatus()) + .expectNext(new CloseStatus(4401, "Unauthorized")) + .verifyComplete(); + } + @Test void unauthorizedWithoutConnectionInit() throws Exception { handle(this.handler, new TextMessage(BookTestUtils.BOOK_SUBSCRIPTION));