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

Add WebSocketInterceptor

Browse Source 
See gh-162
This commit is contained in:
Rossen Stoyanchev
2021-10-20 21:21:31 +01:00
parent 811d32b647
commit 20bae75ed8
23 changed files with 361 additions and 70 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
graphql-spring-boot-starter/src/test/java/org/springframework/graphql/boot/GraphQlWebFluxAutoConfigurationTests.java
View File

@@ -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")));
}

2
graphql-spring-boot-starter/src/test/java/org/springframework/graphql/boot/GraphQlWebMvcAutoConfigurationTests.java
View File

@@ -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")));
}

2
samples/webflux-websocket/src/test/java/io/spring/sample/graphql/QueryTests.java
View File

@@ -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

2
samples/webflux-websocket/src/test/java/io/spring/sample/graphql/SubscriptionTests.java
View File

@@ -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

4
samples/webmvc-http-security/src/main/java/io/spring/sample/graphql/SampleApplication.java
View File

@@ -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));
};
}

8
spring-graphql-docs/src/docs/asciidoc/index.adoc
View File

@@ -115,12 +115,12 @@ transformation of the `graphql.ExecutionInput`:
class MyInterceptor implements WebInterceptor {
@Override
public Mono<WebOutput> intercept(WebInput webInput, WebGraphQlHandler next) {
public Mono<WebOutput> intercept(WebInput webInput, WebInterceptorChain chain) {
webInput.configureExecutionInput((executionInput, builder) -> {
Map<String, Object> 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<WebOutput> intercept(WebInput webInput, WebGraphQlHandler next) {
return next.handle(webInput)
public Mono<WebOutput> intercept(WebInput webInput, WebInterceptorChain chain) {
return chain.next(webInput)
.map(webOutput -> {
Object data = webOutput.getData();
Object updatedData = ... ;

2
spring-graphql-test/src/main/java/org/springframework/graphql/test/tester/WebGraphQlHandlerRequestStrategy.java
View File

@@ -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;
}

2
spring-graphql-test/src/test/java/org/springframework/graphql/test/tester/WebGraphQlTesterTests.java
View File

@@ -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

82
spring-graphql/src/main/java/org/springframework/graphql/web/DefaultWebGraphQlHandlerBuilder.java
View File

@@ -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<WebInterceptor> 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<WebOutput> handleRequest(WebInput input) {
return interceptorChain.next(input);
}
@Override
public Mono<Object> handleWebSocketInitialization(Map<String, Object> payload) {
return (webSocketInterceptor != null ?
webSocketInterceptor.handleConnectionInitialization(payload) : Mono.empty());
}
@Override
public Mono<Void> 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<WebInterceptor> 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<WebInterceptor> interceptors) {
List<WebSocketInterceptor> 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<WebOutput> handle(WebInput input) {
return this.delegate.handle(input).contextWrite((context) ->
public Mono<WebOutput> handleRequest(WebInput input) {
return this.delegate.handleRequest(input).contextWrite((context) ->
ReactorContextManager.extractThreadLocalValues(this.accessor, context));
}
@Override
public Mono<Object> handleWebSocketInitialization(Map<String, Object> payload) {
return this.delegate.handleWebSocketInitialization(payload).contextWrite((context) ->
ReactorContextManager.extractThreadLocalValues(this.accessor, context));
}
@Override
public Mono<Void> handleWebSocketCompletion() {
return this.delegate.handleWebSocketCompletion().contextWrite((context) ->
ReactorContextManager.extractThreadLocalValues(this.accessor, context));
}

33
spring-graphql/src/main/java/org/springframework/graphql/web/WebGraphQlHandler.java
View File

@@ -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<WebOutput> handle(WebInput input);
Mono<WebOutput> 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<Object> handleWebSocketInitialization(Map<String, Object> 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<Void> 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}.

6
spring-graphql/src/main/java/org/springframework/graphql/web/WebInterceptor.java
View File

@@ -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<WebOutput> intercept(WebInput webInput, WebGraphQlHandler next);
Mono<WebOutput> intercept(WebInput webInput, WebInterceptorChain next);
/**
* Return a composed {@link WebInterceptor} that invokes the current interceptor first

37
spring-graphql/src/main/java/org/springframework/graphql/web/WebInterceptorChain.java Normal file
View File

@@ -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<WebOutput> next(WebInput webInput);
}

57
spring-graphql/src/main/java/org/springframework/graphql/web/WebSocketInterceptor.java Normal file
View File

@@ -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<WebOutput> 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<Object> handleConnectionInitialization(Map<String, Object> 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<Void> handleConnectionCompletion() {
return Mono.empty();
}
}

2
spring-graphql/src/main/java/org/springframework/graphql/web/webflux/GraphQlHttpHandler.java
View File

@@ -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<String, Object> spec = output.toSpecification();

18
spring-graphql/src/main/java/org/springframework/graphql/web/webflux/GraphQlWebSocketHandler.java
View File

@@ -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<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;
return (payload != null ? payload : Collections.emptyMap());
}
@SuppressWarnings("unchecked")

2
spring-graphql/src/main/java/org/springframework/graphql/web/webmvc/GraphQlHttpHandler.java
View File

@@ -72,7 +72,7 @@ public class GraphQlHttpHandler {
if (logger.isDebugEnabled()) {
logger.debug("Executing: " + input);
}
Mono<ServerResponse> responseMono = this.graphQlHandler.handle(input).map((output) -> {
Mono<ServerResponse> responseMono = this.graphQlHandler.handleRequest(input).map((output) -> {
if (logger.isDebugEnabled()) {
logger.debug("Execution complete");
}

31
spring-graphql/src/main/java/org/springframework/graphql/web/webmvc/GraphQlWebSocketHandler.java
View File

@@ -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<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;
return (payload != null ? payload : Collections.emptyMap());
}
private SessionState getSessionInfo(WebSocketSession session) {

18
spring-graphql/src/test/java/org/springframework/graphql/data/querydsl/QuerydslDataFetcherTests.java
View File

@@ -66,7 +66,7 @@ class QuerydslDataFetcherTests {
BiConsumer<Consumer<TypeRuntimeWiring.Builder>, 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<Consumer<TypeRuntimeWiring.Builder>, 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<Predicate> predicateCaptor = ArgumentCaptor.forClass(Predicate.class);
@@ -202,7 +202,7 @@ class QuerydslDataFetcherTests {
BiConsumer<Consumer<TypeRuntimeWiring.Builder>, 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<Consumer<TypeRuntimeWiring.Builder>, 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(

4
spring-graphql/src/test/java/org/springframework/graphql/web/ConsumeOneAndNeverCompleteInterceptor.java
View File

@@ -25,8 +25,8 @@ import static org.assertj.core.api.Assertions.assertThat;
public class ConsumeOneAndNeverCompleteInterceptor implements WebInterceptor {
@Override
public Mono<WebOutput> intercept(WebInput webInput, WebGraphQlHandler next) {
return next.handle(webInput).map((output) -> output.transform((builder) -> {
public Mono<WebOutput> 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()));

12
spring-graphql/src/test/java/org/springframework/graphql/web/WebGraphQlHandlerTests.java
View File

@@ -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<String, Object> 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<String, Object> 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<String, Object> data = handler.handle(webInput).block().getData();
Map<String, Object> 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<GraphQLError> errors = webOutput.getErrors();
assertThat(errors.get(0).getMessage()).isEqualTo("Resolved error: Invalid greeting, name=007");

14
spring-graphql/src/test/java/org/springframework/graphql/web/WebInterceptorTests.java
View File

@@ -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<WebOutput> intercept(WebInput input, WebGraphQlHandler next) {
public Mono<WebOutput> 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;

43
spring-graphql/src/test/java/org/springframework/graphql/web/webflux/GraphQlWebSocketHandlerTests.java
View File

@@ -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<Object> handleConnectionInitialization(Map<String, Object> payload) {
Object value = payload.get("key");
return Mono.just(Collections.singletonMap("key", value + " acknowledged"));
}
});
StepVerifier.create(session.getOutput())
.consumeNextWith((message) -> {
Map<String, Object> content = decode(message);
assertThat(content).containsEntry("type", "connection_ack");
assertThat((Map<String, Object>) 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<Object> handleConnectionInitialization(Map<String, Object> 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)));

48
spring-graphql/src/test/java/org/springframework/graphql/web/webmvc/GraphQlWebSocketHandlerTests.java
View File

@@ -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<Object> handleConnectionInitialization(Map<String, Object> 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<String, Object> content = decode(message);
assertThat(content).containsEntry("type", "connection_ack");
assertThat((Map<String, Object>) 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<Object> handleConnectionInitialization(Map<String, Object> 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));