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

RSocket support for SubscriptionErrorException

Browse Source 
See gh-339
This commit is contained in:
rstoyanchev
2022-03-28 07:37:01 +00:00
parent 3090328666
commit 5b93c1fce7
11 changed files with 172 additions and 41 deletions
Download Patch File Download Diff File Expand all files Collapse all files

16
spring-graphql/src/main/java/org/springframework/graphql/client/AbstractGraphQlClientBuilder.java
View File

@@ -118,6 +118,14 @@ public abstract class AbstractGraphQlClientBuilder<B extends AbstractGraphQlClie
this.jsonEncoder = encoder;
}
/**
* Access to the configured JSON encoder.
*/
protected Encoder<?> getJsonEncoder() {
Assert.notNull(this.jsonEncoder, "JSON Encoder not set");
return this.jsonEncoder;
}
/**
* Variant of {@link #setJsonCodecs} for setting each codec individually.
*/
@@ -125,6 +133,14 @@ public abstract class AbstractGraphQlClientBuilder<B extends AbstractGraphQlClie
this.jsonDecoder = decoder;
}
/**
* Access to the configured JSON encoder.
*/
protected Decoder<?> getJsonDecoder() {
Assert.notNull(this.jsonDecoder, "JSON Encoder not set");
return this.jsonDecoder;
}
/**
* Return the configured interceptors. For subclasses that look for a
* transport specific interceptor extensions.

5
spring-graphql/src/main/java/org/springframework/graphql/client/CodecDelegate.java
View File

@@ -25,7 +25,6 @@ import org.springframework.core.io.buffer.DataBuffer;
import org.springframework.core.io.buffer.DataBufferUtils;
import org.springframework.graphql.server.support.GraphQlWebSocketMessage;
import org.springframework.http.MediaType;
import org.springframework.http.codec.ClientCodecConfigurer;
import org.springframework.http.codec.CodecConfigurer;
import org.springframework.http.codec.DecoderHttpMessageReader;
import org.springframework.http.codec.EncoderHttpMessageWriter;
@@ -52,10 +51,6 @@ final class CodecDelegate {
private final Encoder<?> encoder;
CodecDelegate() {
this(ClientCodecConfigurer.create());
}
CodecDelegate(CodecConfigurer configurer) {
Assert.notNull(configurer, "CodecConfigurer is required");
this.codecConfigurer = configurer;

8
spring-graphql/src/main/java/org/springframework/graphql/client/DefaultRSocketGraphQlClient.java
View File

@@ -147,16 +147,16 @@ final class DefaultRSocketGraphQlClient extends AbstractDelegatingGraphQlClient
@Override
public RSocketGraphQlClient build() {
Assert.state(this.clientTransport != null, "Neither WebSocket nor TCP networking configured");
RSocketRequester requester = this.requesterBuilder.transport(this.clientTransport);
RSocketGraphQlTransport graphQlTransport = new RSocketGraphQlTransport(this.route, requester);
// Pass the codecs to the parent for response decoding
this.requesterBuilder.rsocketStrategies(builder -> {
builder.decoders(decoders -> setJsonDecoder(CodecDelegate.findJsonDecoder(decoders)));
builder.encoders(encoders -> setJsonEncoder(CodecDelegate.findJsonEncoder(encoders)));
});
Assert.state(this.clientTransport != null, "Neither WebSocket nor TCP networking configured");
RSocketRequester requester = this.requesterBuilder.transport(this.clientTransport);
RSocketGraphQlTransport graphQlTransport = new RSocketGraphQlTransport(this.route, requester, getJsonDecoder());
return new DefaultRSocketGraphQlClient(
super.buildGraphQlClient(graphQlTransport),
this.requesterBuilder, this.clientTransport, this.route, getBuilderInitializer());

33
spring-graphql/src/main/java/org/springframework/graphql/client/RSocketGraphQlTransport.java
View File

@@ -16,12 +16,21 @@
package org.springframework.graphql.client;
import java.nio.charset.StandardCharsets;
import java.util.Collections;
import java.util.List;
import java.util.Map;
import graphql.GraphQLError;
import io.rsocket.exceptions.RejectedException;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import org.springframework.core.ParameterizedTypeReference;
import org.springframework.core.ResolvableType;
import org.springframework.core.codec.Decoder;
import org.springframework.core.codec.DecodingException;
import org.springframework.core.io.buffer.DefaultDataBufferFactory;
import org.springframework.graphql.GraphQlRequest;
import org.springframework.graphql.GraphQlResponse;
import org.springframework.messaging.rsocket.RSocketRequester;
@@ -43,17 +52,23 @@ final class RSocketGraphQlTransport implements GraphQlTransport {
private static final ParameterizedTypeReference<Map<String, Object>> MAP_TYPE =
new ParameterizedTypeReference<Map<String, Object>>() {};
private static final ResolvableType LIST_TYPE = ResolvableType.forClass(List.class);
private final String route;
private final RSocketRequester rsocketRequester;
private final Decoder<?> jsonDecoder;
RSocketGraphQlTransport(String route, RSocketRequester requester) {
RSocketGraphQlTransport(String route, RSocketRequester requester, Decoder<?> jsonDecoder) {
Assert.notNull(route, "'route' is required");
Assert.notNull(requester, "RSocketRequester is required");
Assert.notNull(jsonDecoder, "JSON Decoder is required");
this.route = route;
this.rsocketRequester = requester;
this.jsonDecoder = jsonDecoder;
}
@@ -68,7 +83,23 @@ final class RSocketGraphQlTransport implements GraphQlTransport {
public Flux<GraphQlResponse> executeSubscription(GraphQlRequest request) {
return this.rsocketRequester.route(this.route).data(request.toMap())
.retrieveFlux(MAP_TYPE)
.doOnError(ex -> System.out.println(ex))
.onErrorResume(RejectedException.class, ex -> Flux.error(decodeErrors(request, ex)))
.map(ResponseMapGraphQlResponse::new);
}
@SuppressWarnings("unchecked")
private Exception decodeErrors(GraphQlRequest request, RejectedException ex) {
try {
byte[] errorData = ex.getMessage().getBytes(StandardCharsets.UTF_8);
List<GraphQLError> errors = (List<GraphQLError>) this.jsonDecoder.decode(
DefaultDataBufferFactory.sharedInstance.wrap(errorData), LIST_TYPE, null, null);
GraphQlResponse response = new ResponseMapGraphQlResponse(Collections.singletonMap("errors", errors));
return new SubscriptionErrorException(request, response.getErrors());
}
catch (DecodingException ex2) {
return ex;
}
}
}

11
spring-graphql/src/main/java/org/springframework/graphql/client/ResponseMapGraphQlResponse.java
View File

@@ -136,10 +136,13 @@ class ResponseMapGraphQlResponse extends AbstractGraphQlResponse {
@SuppressWarnings("unchecked")
private static String initPath(Map<String, Object> errorMap) {
return ((List<Object>) errorMap.getOrDefault("path", Collections.emptyList())).stream()
.reduce("",
(s, o) -> s + (o instanceof Integer ? "[" + o + "]" : (s.isEmpty() ? o : "." + o)),
(s, s2) -> null);
List<Object> path = (List<Object>) errorMap.get("path");
if (path == null) {
return "";
}
return path.stream().reduce("",
(s, o) -> s + (o instanceof Integer ? "[" + o + "]" : (s.isEmpty() ? o : "." + o)),
(s, s2) -> null);
}
@Override

9
spring-graphql/src/main/java/org/springframework/graphql/server/DefaultWebGraphQlHandlerBuilder.java
View File

@@ -25,6 +25,7 @@ import reactor.core.publisher.Mono;
import org.springframework.graphql.ExecutionGraphQlService;
import org.springframework.graphql.execution.ReactorContextManager;
import org.springframework.graphql.execution.ThreadLocalAccessor;
import org.springframework.graphql.server.WebGraphQlInterceptor.Chain;
import org.springframework.lang.Nullable;
import org.springframework.util.Assert;
import org.springframework.util.CollectionUtils;
@@ -88,18 +89,18 @@ class DefaultWebGraphQlHandlerBuilder implements WebGraphQlHandler.Builder {
@Override
public WebGraphQlHandler build() {
WebGraphQlInterceptor.Chain endOfChain = request -> this.service.execute(request).map(WebGraphQlResponse::new);
Chain endOfChain = request -> this.service.execute(request).map(WebGraphQlResponse::new);
WebGraphQlInterceptor.Chain chain = this.interceptors.stream()
Chain executionChain = this.interceptors.stream()
.reduce(WebGraphQlInterceptor::andThen)
.map(interceptor -> (WebGraphQlInterceptor.Chain) (request) -> interceptor.intercept(request, endOfChain))
.map(interceptor -> (Chain) (request) -> interceptor.intercept(request, endOfChain))
.orElse(endOfChain);
return new WebGraphQlHandler() {
@Override
public Mono<WebGraphQlResponse> handleRequest(WebGraphQlRequest request) {
return chain.next(request)
return executionChain.next(request)
.contextWrite(context -> {
if (!CollectionUtils.isEmpty(accessors)) {
ThreadLocalAccessor accessor = ThreadLocalAccessor.composite(accessors);

63
spring-graphql/src/main/java/org/springframework/graphql/server/GraphQlRSocketHandler.java
View File

@@ -17,19 +17,29 @@
package org.springframework.graphql.server;
import java.nio.charset.StandardCharsets;
import java.util.List;
import java.util.Map;
import graphql.ExecutionResult;
import graphql.GraphQLError;
import io.rsocket.exceptions.InvalidException;
import io.rsocket.exceptions.RejectedException;
import org.reactivestreams.Publisher;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import org.springframework.core.ResolvableType;
import org.springframework.core.codec.Encoder;
import org.springframework.core.io.buffer.DataBuffer;
import org.springframework.core.io.buffer.DefaultDataBufferFactory;
import org.springframework.graphql.ExecutionGraphQlResponse;
import org.springframework.graphql.ExecutionGraphQlService;
import org.springframework.graphql.server.RSocketGraphQlInterceptor.Chain;
import org.springframework.util.AlternativeJdkIdGenerator;
import org.springframework.util.Assert;
import org.springframework.util.IdGenerator;
import org.springframework.util.MimeTypeUtils;
/**
@@ -67,7 +77,12 @@ import org.springframework.util.IdGenerator;
*/
public class GraphQlRSocketHandler {
private final RSocketGraphQlInterceptor.Chain executionChain;
private static final ResolvableType LIST_TYPE = ResolvableType.forClass(List.class);
private final Chain executionChain;
private final Encoder<?> jsonEncoder;
private final IdGenerator idGenerator = new AlternativeJdkIdGenerator();
@@ -75,17 +90,33 @@ public class GraphQlRSocketHandler {
/**
* Create a new instance that handles requests through a chain of interceptors
* followed by the given {@link ExecutionGraphQlService}.
* @param graphQlService the service that will execute the request
* @param interceptors interceptors to form the processing chain
* @param jsonEncoder a JSON encoder for serializing a
* {@link graphql.GraphQLError} list for a failed subscription
*/
public GraphQlRSocketHandler(
ExecutionGraphQlService service, List<RSocketGraphQlInterceptor> interceptors) {
ExecutionGraphQlService graphQlService, List<RSocketGraphQlInterceptor> interceptors,
Encoder<?> jsonEncoder) {
RSocketGraphQlInterceptor.Chain endOfChain = request -> service.execute(request).map(RSocketGraphQlResponse::new);
Assert.notNull(graphQlService, "ExecutionGraphQlService is required");
Assert.notNull(jsonEncoder, "JSON Encoder is required");
this.executionChain = (interceptors.isEmpty() ? endOfChain :
this.executionChain = initExecutionChain(graphQlService, interceptors);
this.jsonEncoder = jsonEncoder;
}
private static Chain initExecutionChain(
ExecutionGraphQlService graphQlService, List<RSocketGraphQlInterceptor> interceptors) {
Chain endOfChain = request ->
graphQlService.execute(request).map(RSocketGraphQlResponse::new);
return interceptors.isEmpty() ? endOfChain :
interceptors.stream()
.reduce(RSocketGraphQlInterceptor::andThen)
.map(interceptor -> (RSocketGraphQlInterceptor.Chain) request -> interceptor.intercept(request, endOfChain))
.orElse(endOfChain));
.map(interceptor -> (Chain) request -> interceptor.intercept(request, endOfChain))
.orElse(endOfChain);
}
@@ -106,12 +137,13 @@ public class GraphQlRSocketHandler {
Publisher<ExecutionResult> publisher = response.getData();
return Flux.from(publisher).map(ExecutionResult::toSpecification);
}
String message = (!response.isValid() ?
response.toMap().get("errors").toString() :
"Response is not a stream, is the operation actually a subscription?");
return Flux.error(new RejectedException(message));
else if (response.isValid()) {
return Flux.error(new InvalidException(
"Expected a Publisher for a subscription operation. " +
"This is either a server error or the operation is not a subscription"));
}
String errorData = encodeErrors(response).toString(StandardCharsets.UTF_8);
return Flux.error(new RejectedException(errorData));
});
}
@@ -120,4 +152,11 @@ public class GraphQlRSocketHandler {
return this.executionChain.next(new RSocketGraphQlRequest(payload, requestId, null));
}
@SuppressWarnings("unchecked")
private DataBuffer encodeErrors(RSocketGraphQlResponse response) {
return ((Encoder<List<GraphQLError>>) this.jsonEncoder).encodeValue(
response.getExecutionResult().getErrors(),
DefaultDataBufferFactory.sharedInstance, LIST_TYPE, MimeTypeUtils.APPLICATION_JSON, null);
}
}

3
spring-graphql/src/main/java/org/springframework/graphql/server/webflux/CodecDelegate.java
View File

@@ -36,8 +36,7 @@ import org.springframework.web.reactive.socket.WebSocketMessage;
import org.springframework.web.reactive.socket.WebSocketSession;
/**
* Delegate that can be embedded in a class to help with encoding and decoding
* GraphQL over WebSocket messages.
* Helper class for encoding and decoding GraphQL messages.
*
* @author Rossen Stoyanchev
* @since 1.0.0

3
spring-graphql/src/test/java/org/springframework/graphql/client/MockGraphQlWebSocketServer.java
View File

@@ -31,6 +31,7 @@ import org.springframework.graphql.GraphQlRequest;
import org.springframework.graphql.GraphQlResponse;
import org.springframework.graphql.support.DefaultGraphQlRequest;
import org.springframework.graphql.server.support.GraphQlWebSocketMessage;
import org.springframework.http.codec.ClientCodecConfigurer;
import org.springframework.lang.Nullable;
import org.springframework.web.reactive.socket.WebSocketHandler;
import org.springframework.web.reactive.socket.WebSocketSession;
@@ -52,7 +53,7 @@ public final class MockGraphQlWebSocketServer implements WebSocketHandler {
private final Map<Map<String, Object>, Exchange> expectedExchanges = new LinkedHashMap<>();
private final CodecDelegate codecDelegate = new CodecDelegate();
private final CodecDelegate codecDelegate = new CodecDelegate(ClientCodecConfigurer.create());
/**

54
spring-graphql/src/test/java/org/springframework/graphql/client/RSocketGraphQlClientBuilderTests.java → spring-graphql/src/test/java/org/springframework/graphql/client/RSocketGraphQlClientTests.java
View File

@@ -19,26 +19,34 @@ package org.springframework.graphql.client;
import java.time.Duration;
import java.util.Collections;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import graphql.ExecutionInput;
import graphql.ExecutionResult;
import graphql.ExecutionResultImpl;
import graphql.GraphQLError;
import graphql.GraphqlErrorBuilder;
import io.rsocket.Closeable;
import io.rsocket.SocketAcceptor;
import io.rsocket.core.RSocketServer;
import io.rsocket.transport.local.LocalClientTransport;
import io.rsocket.transport.local.LocalServerTransport;
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.Test;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import reactor.test.StepVerifier;
import org.springframework.graphql.ExecutionGraphQlResponse;
import org.springframework.graphql.ExecutionGraphQlService;
import org.springframework.graphql.GraphQlRequest;
import org.springframework.graphql.ResponseError;
import org.springframework.graphql.support.DefaultExecutionGraphQlResponse;
import org.springframework.graphql.server.GraphQlRSocketHandler;
import org.springframework.http.codec.json.Jackson2JsonDecoder;
import org.springframework.http.codec.json.Jackson2JsonEncoder;
import org.springframework.lang.Nullable;
import org.springframework.messaging.handler.annotation.MessageMapping;
import org.springframework.messaging.rsocket.RSocketStrategies;
import org.springframework.messaging.rsocket.annotation.support.RSocketMessageHandler;
@@ -47,11 +55,15 @@ import org.springframework.util.Assert;
import static org.assertj.core.api.Assertions.assertThat;
/**
* {@code RSocketGraphQlClient} tests using {@link LocalClientTransport} and
* {@link LocalServerTransport} for RSocket exchanges in memory, with stubbed
* GraphQL responses.
*
* @author Rossen Stoyanchev
*/
public class RSocketGraphQlClientBuilderTests {
public class RSocketGraphQlClientTests {
private static final String DOCUMENT = "{ Query }";
@@ -61,6 +73,11 @@ public class RSocketGraphQlClientBuilderTests {
private final BuilderSetup builderSetup = new BuilderSetup();
@AfterEach
void tearDown() {
this.builderSetup.shutDown();
}
@Test
void mutate() {
@@ -72,7 +89,7 @@ public class RSocketGraphQlClientBuilderTests {
GraphQlRequest request = this.builderSetup.getGraphQlRequest();
assertThat(request).isNotNull();
// Mutate
// Mutate: still works (carries over original default)
client = client.mutate().build();
client.document(DOCUMENT).execute().block(TIMEOUT);
@@ -80,6 +97,26 @@ public class RSocketGraphQlClientBuilderTests {
assertThat(request).isNotNull();
}
@Test
void subscriptionError() {
String document = "subscription { greetings }";
GraphQLError error = GraphqlErrorBuilder.newError().message("boo").build();
ExecutionResult result = ExecutionResultImpl.newExecutionResult().addError(error).build();
this.builderSetup.setMockResponse(document, result);
Flux<ClientGraphQlResponse> responseFlux = this.builderSetup.initBuilder().build()
.document(document).executeSubscription();
StepVerifier.create(responseFlux)
.expectErrorSatisfies(ex -> {
assertThat(ex).isInstanceOf(SubscriptionErrorException.class);
List<ResponseError> errors = ((SubscriptionErrorException) ex).getErrors();
assertThat(errors).hasSize(1);
assertThat(errors.get(0).getMessage()).isEqualTo("boo");
})
.verify(TIMEOUT);
}
private static class BuilderSetup {
@@ -88,6 +125,9 @@ public class RSocketGraphQlClientBuilderTests {
private final Map<String, ExecutionGraphQlResponse> responses = new HashMap<>();
@Nullable
private Closeable server;
public BuilderSetup() {
ExecutionGraphQlResponse defaultResponse = new DefaultExecutionGraphQlResponse(
@@ -108,9 +148,9 @@ public class RSocketGraphQlClientBuilderTests {
};
GraphQlRSocketController controller = new GraphQlRSocketController(
new GraphQlRSocketHandler(graphQlService, Collections.emptyList()));
new GraphQlRSocketHandler(graphQlService, Collections.emptyList(), new Jackson2JsonEncoder()));
RSocketServer.create()
this.server = RSocketServer.create()
.acceptor(createSocketAcceptor(controller))
.bind(LocalServerTransport.create("local"))
.block();
@@ -142,6 +182,12 @@ public class RSocketGraphQlClientBuilderTests {
public GraphQlRequest getGraphQlRequest() {
return this.graphQlRequest;
}
public void shutDown() {
if (this.server != null) {
this.server.dispose();
}
}
}

8
spring-graphql/src/test/java/org/springframework/graphql/client/MockWebSocketGraphQlTransportTests.java → spring-graphql/src/test/java/org/springframework/graphql/client/WebSocketGraphQlTransportTests.java
View File

@@ -57,11 +57,11 @@ import static org.mockito.Mockito.when;
*
* @author Rossen Stoyanchev
*/
public class MockWebSocketGraphQlTransportTests {
public class WebSocketGraphQlTransportTests {
private final static Duration TIMEOUT = Duration.ofSeconds(5);
private static final CodecDelegate CODEC_DELEGATE = new CodecDelegate();
private static final CodecDelegate CODEC_DELEGATE = new CodecDelegate(ClientCodecConfigurer.create());
private final MockGraphQlWebSocketServer mockServer = new MockGraphQlWebSocketServer();
@@ -351,7 +351,7 @@ public class MockWebSocketGraphQlTransportTests {
private final GraphQlResponse response;
private final CodecDelegate codecDelegate = new CodecDelegate();
private final CodecDelegate codecDelegate = new CodecDelegate(ClientCodecConfigurer.create());
private PingResponseHandler(GraphQlResponse response) {
this.response = response;
@@ -387,7 +387,7 @@ public class MockWebSocketGraphQlTransportTests {
*/
private static class UnexpectedResponseHandler implements WebSocketHandler {
private final CodecDelegate codecDelegate = new CodecDelegate();
private final CodecDelegate codecDelegate = new CodecDelegate(ClientCodecConfigurer.create());
@Override