diff --git a/spring-graphql/src/main/java/org/springframework/graphql/client/AbstractGraphQlClientBuilder.java b/spring-graphql/src/main/java/org/springframework/graphql/client/AbstractGraphQlClientBuilder.java index fca0f447..d95e8535 100644 --- a/spring-graphql/src/main/java/org/springframework/graphql/client/AbstractGraphQlClientBuilder.java +++ b/spring-graphql/src/main/java/org/springframework/graphql/client/AbstractGraphQlClientBuilder.java @@ -118,6 +118,14 @@ public abstract class AbstractGraphQlClientBuilder 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 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. diff --git a/spring-graphql/src/main/java/org/springframework/graphql/client/CodecDelegate.java b/spring-graphql/src/main/java/org/springframework/graphql/client/CodecDelegate.java index 74e66edb..6f433c78 100644 --- a/spring-graphql/src/main/java/org/springframework/graphql/client/CodecDelegate.java +++ b/spring-graphql/src/main/java/org/springframework/graphql/client/CodecDelegate.java @@ -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; diff --git a/spring-graphql/src/main/java/org/springframework/graphql/client/DefaultRSocketGraphQlClient.java b/spring-graphql/src/main/java/org/springframework/graphql/client/DefaultRSocketGraphQlClient.java index caad0cd4..e043954f 100644 --- a/spring-graphql/src/main/java/org/springframework/graphql/client/DefaultRSocketGraphQlClient.java +++ b/spring-graphql/src/main/java/org/springframework/graphql/client/DefaultRSocketGraphQlClient.java @@ -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()); diff --git a/spring-graphql/src/main/java/org/springframework/graphql/client/RSocketGraphQlTransport.java b/spring-graphql/src/main/java/org/springframework/graphql/client/RSocketGraphQlTransport.java index 669eec78..479f1b63 100644 --- a/spring-graphql/src/main/java/org/springframework/graphql/client/RSocketGraphQlTransport.java +++ b/spring-graphql/src/main/java/org/springframework/graphql/client/RSocketGraphQlTransport.java @@ -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_TYPE = new ParameterizedTypeReference>() {}; + 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 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 errors = (List) 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; + } + } + } diff --git a/spring-graphql/src/main/java/org/springframework/graphql/client/ResponseMapGraphQlResponse.java b/spring-graphql/src/main/java/org/springframework/graphql/client/ResponseMapGraphQlResponse.java index 5a1a1fbb..8df91d5b 100644 --- a/spring-graphql/src/main/java/org/springframework/graphql/client/ResponseMapGraphQlResponse.java +++ b/spring-graphql/src/main/java/org/springframework/graphql/client/ResponseMapGraphQlResponse.java @@ -136,10 +136,13 @@ class ResponseMapGraphQlResponse extends AbstractGraphQlResponse { @SuppressWarnings("unchecked") private static String initPath(Map errorMap) { - return ((List) errorMap.getOrDefault("path", Collections.emptyList())).stream() - .reduce("", - (s, o) -> s + (o instanceof Integer ? "[" + o + "]" : (s.isEmpty() ? o : "." + o)), - (s, s2) -> null); + List path = (List) 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 diff --git a/spring-graphql/src/main/java/org/springframework/graphql/server/DefaultWebGraphQlHandlerBuilder.java b/spring-graphql/src/main/java/org/springframework/graphql/server/DefaultWebGraphQlHandlerBuilder.java index c7434e77..9c7b6dd5 100644 --- a/spring-graphql/src/main/java/org/springframework/graphql/server/DefaultWebGraphQlHandlerBuilder.java +++ b/spring-graphql/src/main/java/org/springframework/graphql/server/DefaultWebGraphQlHandlerBuilder.java @@ -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 handleRequest(WebGraphQlRequest request) { - return chain.next(request) + return executionChain.next(request) .contextWrite(context -> { if (!CollectionUtils.isEmpty(accessors)) { ThreadLocalAccessor accessor = ThreadLocalAccessor.composite(accessors); diff --git a/spring-graphql/src/main/java/org/springframework/graphql/server/GraphQlRSocketHandler.java b/spring-graphql/src/main/java/org/springframework/graphql/server/GraphQlRSocketHandler.java index d647df5d..de691b4b 100644 --- a/spring-graphql/src/main/java/org/springframework/graphql/server/GraphQlRSocketHandler.java +++ b/spring-graphql/src/main/java/org/springframework/graphql/server/GraphQlRSocketHandler.java @@ -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 interceptors) { + ExecutionGraphQlService graphQlService, List 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 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 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>) this.jsonEncoder).encodeValue( + response.getExecutionResult().getErrors(), + DefaultDataBufferFactory.sharedInstance, LIST_TYPE, MimeTypeUtils.APPLICATION_JSON, null); + } + } diff --git a/spring-graphql/src/main/java/org/springframework/graphql/server/webflux/CodecDelegate.java b/spring-graphql/src/main/java/org/springframework/graphql/server/webflux/CodecDelegate.java index 8d01264f..60676fcf 100644 --- a/spring-graphql/src/main/java/org/springframework/graphql/server/webflux/CodecDelegate.java +++ b/spring-graphql/src/main/java/org/springframework/graphql/server/webflux/CodecDelegate.java @@ -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 diff --git a/spring-graphql/src/test/java/org/springframework/graphql/client/MockGraphQlWebSocketServer.java b/spring-graphql/src/test/java/org/springframework/graphql/client/MockGraphQlWebSocketServer.java index 6fce6d5d..d5facd87 100644 --- a/spring-graphql/src/test/java/org/springframework/graphql/client/MockGraphQlWebSocketServer.java +++ b/spring-graphql/src/test/java/org/springframework/graphql/client/MockGraphQlWebSocketServer.java @@ -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, Exchange> expectedExchanges = new LinkedHashMap<>(); - private final CodecDelegate codecDelegate = new CodecDelegate(); + private final CodecDelegate codecDelegate = new CodecDelegate(ClientCodecConfigurer.create()); /** diff --git a/spring-graphql/src/test/java/org/springframework/graphql/client/RSocketGraphQlClientBuilderTests.java b/spring-graphql/src/test/java/org/springframework/graphql/client/RSocketGraphQlClientTests.java similarity index 76% rename from spring-graphql/src/test/java/org/springframework/graphql/client/RSocketGraphQlClientBuilderTests.java rename to spring-graphql/src/test/java/org/springframework/graphql/client/RSocketGraphQlClientTests.java index b58b447e..bdb6917c 100644 --- a/spring-graphql/src/test/java/org/springframework/graphql/client/RSocketGraphQlClientBuilderTests.java +++ b/spring-graphql/src/test/java/org/springframework/graphql/client/RSocketGraphQlClientTests.java @@ -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 responseFlux = this.builderSetup.initBuilder().build() + .document(document).executeSubscription(); + + StepVerifier.create(responseFlux) + .expectErrorSatisfies(ex -> { + assertThat(ex).isInstanceOf(SubscriptionErrorException.class); + List 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 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(); + } + } } diff --git a/spring-graphql/src/test/java/org/springframework/graphql/client/MockWebSocketGraphQlTransportTests.java b/spring-graphql/src/test/java/org/springframework/graphql/client/WebSocketGraphQlTransportTests.java similarity index 98% rename from spring-graphql/src/test/java/org/springframework/graphql/client/MockWebSocketGraphQlTransportTests.java rename to spring-graphql/src/test/java/org/springframework/graphql/client/WebSocketGraphQlTransportTests.java index 00ac5db3..27e976fb 100644 --- a/spring-graphql/src/test/java/org/springframework/graphql/client/MockWebSocketGraphQlTransportTests.java +++ b/spring-graphql/src/test/java/org/springframework/graphql/client/WebSocketGraphQlTransportTests.java @@ -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