diff --git a/spring-graphql/src/main/java/org/springframework/graphql/execution/SingleErrorExceptionResolver.java b/spring-graphql/src/main/java/org/springframework/graphql/execution/SingleErrorExceptionResolver.java deleted file mode 100644 index 0683ef31..00000000 --- a/spring-graphql/src/main/java/org/springframework/graphql/execution/SingleErrorExceptionResolver.java +++ /dev/null @@ -1,42 +0,0 @@ -/* - * 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.execution; - -import java.util.Collections; -import java.util.List; - -import graphql.GraphQLError; -import graphql.schema.DataFetchingEnvironment; -import reactor.core.publisher.Mono; - -/** - * Simple adapter for {@link DataFetcherExceptionResolver} implementations that - * resolve exceptions to a single error only. - */ -public abstract class SingleErrorExceptionResolver implements DataFetcherExceptionResolver { - - - @Override - public final Mono> resolveException(Throwable exception, DataFetchingEnvironment environment) { - return doResolve(exception, environment).map(Collections::singletonList); - } - - /** - * Implement this method to resolve the exception to an error. - */ - protected abstract Mono doResolve(Throwable exception, DataFetchingEnvironment environment); - -} diff --git a/spring-graphql/src/test/java/org/springframework/graphql/web/Book.java b/spring-graphql/src/test/java/org/springframework/graphql/web/Book.java index 9a053fef..f0cda89d 100644 --- a/spring-graphql/src/test/java/org/springframework/graphql/web/Book.java +++ b/spring-graphql/src/test/java/org/springframework/graphql/web/Book.java @@ -2,50 +2,41 @@ package org.springframework.graphql.web; public class Book { - String id; + Long id; String name; - int pageCount; - String author; + public Book() { } - public Book(String id, String name, int pageCount, String author) { + public Book(Long id, String name, String author) { this.id = id; this.name = name; - this.pageCount = pageCount; this.author = author; } - public String getId() { - return id; + + public Long getId() { + return this.id; } - public void setId(String id) { + public void setId(Long id) { this.id = id; } public String getName() { - return name; + return this.name; } public void setName(String name) { this.name = name; } - public int getPageCount() { - return pageCount; - } - - public void setPageCount(int pageCount) { - this.pageCount = pageCount; - } - public String getAuthor() { - return author; + return this.author; } public void setAuthor(String author) { diff --git a/spring-graphql/src/test/java/org/springframework/graphql/web/BookTestUtils.java b/spring-graphql/src/test/java/org/springframework/graphql/web/BookTestUtils.java new file mode 100644 index 00000000..0d8d585a --- /dev/null +++ b/spring-graphql/src/test/java/org/springframework/graphql/web/BookTestUtils.java @@ -0,0 +1,91 @@ +/* + * 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.Arrays; +import java.util.HashMap; +import java.util.Map; + +import graphql.schema.idl.RuntimeWiring; +import reactor.core.publisher.Flux; + +import org.springframework.core.io.ClassPathResource; +import org.springframework.graphql.execution.ExecutionGraphQlService; +import org.springframework.graphql.execution.GraphQlSource; + +import static graphql.schema.idl.TypeRuntimeWiring.newTypeWiring; + +public abstract class BookTestUtils { + + public static final String SUBSCRIPTION_ID = "1"; + + public static final String BOOK_QUERY = "{" + + "\"id\":\"" + BookTestUtils.SUBSCRIPTION_ID + "\"," + + "\"type\":\"subscribe\"," + + "\"payload\":{\"query\": \"" + + " query TestQuery {" + + " bookById(id: \\\"1\\\"){ " + + " id" + + " name" + + " author" + + " }}\"}" + + "}"; + + public static final String BOOK_SUBSCRIPTION = "{" + + "\"id\":\"" + SUBSCRIPTION_ID + "\"," + + "\"type\":\"subscribe\"," + + "\"payload\":{\"query\": \"" + + " subscription TestSubscription {" + + " bookSearch(author: \\\"George\\\") {" + + " id" + + " name" + + " author" + + " }}\"}" + + "}"; + + private static final Map booksMap = new HashMap<>(4); + static { + booksMap.put(1L, new Book(1L, "Nineteen Eighty-Four", "George Orwell")); + booksMap.put(2L, new Book(2L, "The Great Gatsby", "F. Scott Fitzgerald")); + booksMap.put(3L, new Book(3L, "Catch-22", "Joseph Heller")); + booksMap.put(4L, new Book(4L, "To The Lighthouse", "Virginia Woolf")); + booksMap.put(5L, new Book(5L, "Animal Farm", "George Orwell")); + } + + + public static WebGraphQlHandler initWebGraphQlHandler(WebInterceptor... interceptors) { + return WebGraphQlHandler.builder(new ExecutionGraphQlService(graphQlSource())) + .interceptors(Arrays.asList(interceptors)) + .build(); + } + + private static GraphQlSource graphQlSource() { + RuntimeWiring.Builder builder = RuntimeWiring.newRuntimeWiring(); + builder.type(newTypeWiring("Query").dataFetcher("bookById", env -> { + Long id = Long.parseLong(env.getArgument("id")); + return booksMap.get(id); + })); + builder.type(newTypeWiring("Subscription").dataFetcher("bookSearch", env -> { + String author = env.getArgument("author"); + return Flux.fromIterable(booksMap.values()).filter(book -> book.getAuthor().contains(author)); + })); + return GraphQlSource.builder() + .schemaResource(new ClassPathResource("books/schema.graphqls")) + .runtimeWiring(builder.build()) + .build(); + } + +} diff --git a/spring-graphql/src/test/java/org/springframework/graphql/web/GraphQlDataFetchers.java b/spring-graphql/src/test/java/org/springframework/graphql/web/GraphQlDataFetchers.java deleted file mode 100644 index 47e1576e..00000000 --- a/spring-graphql/src/test/java/org/springframework/graphql/web/GraphQlDataFetchers.java +++ /dev/null @@ -1,30 +0,0 @@ -package org.springframework.graphql.web; - -import java.util.Arrays; -import java.util.List; - -import graphql.schema.DataFetcher; -import reactor.core.publisher.Flux; - -public class GraphQlDataFetchers { - - private static List books = Arrays.asList( - new Book("book-1", "GraphQL for beginners", 100, "John GraphQL"), - new Book("book-2", "Harry Potter and the Philosopher's Stone", 223, "Joanne Rowling"), - new Book("book-3", "Moby Dick", 635, "Moby Dick"), - new Book("book-3", "Moby Dick", 635, "Moby Dick")); - - - public static DataFetcher getBookByIdDataFetcher() { - return env -> books.stream() - .filter(book -> book.getId().equals(env.getArgument("id"))) - .findFirst() - .orElse(null); - } - - public static DataFetcher getBooksOnSale() { - return env -> Flux.fromIterable(books) - .filter(book -> book.getPageCount() >= (int) env.getArgument("minPages")); - } - -} 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 796f65c4..3229b82d 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 @@ -15,7 +15,6 @@ */ package org.springframework.graphql.web.webflux; -import java.net.URI; import java.nio.charset.StandardCharsets; import java.time.Duration; import java.util.ArrayList; @@ -24,34 +23,22 @@ import java.util.List; import java.util.Map; import java.util.function.BiConsumer; -import graphql.schema.idl.RuntimeWiring; import org.junit.jupiter.api.Test; -import org.reactivestreams.Publisher; import reactor.core.publisher.Flux; -import reactor.core.publisher.Mono; import reactor.core.publisher.Sinks; import reactor.test.StepVerifier; -import org.springframework.core.io.ClassPathResource; import org.springframework.core.io.buffer.DataBuffer; import org.springframework.core.io.buffer.DataBufferUtils; import org.springframework.core.io.buffer.DefaultDataBufferFactory; -import org.springframework.graphql.execution.ExecutionGraphQlService; -import org.springframework.graphql.execution.GraphQlSource; +import org.springframework.graphql.web.BookTestUtils; import org.springframework.graphql.web.ConsumeOneAndNeverCompleteInterceptor; -import org.springframework.graphql.web.GraphQlDataFetchers; -import org.springframework.graphql.web.WebGraphQlHandler; import org.springframework.graphql.web.WebInterceptor; -import org.springframework.http.HttpHeaders; import org.springframework.http.codec.ServerCodecConfigurer; import org.springframework.http.codec.json.Jackson2JsonDecoder; -import org.springframework.lang.Nullable; import org.springframework.web.reactive.socket.CloseStatus; -import org.springframework.web.reactive.socket.HandshakeInfo; import org.springframework.web.reactive.socket.WebSocketMessage; -import org.springframework.web.reactive.socket.adapter.AbstractWebSocketSession; -import static graphql.schema.idl.TypeRuntimeWiring.newTypeWiring; import static org.assertj.core.api.Assertions.as; import static org.assertj.core.api.Assertions.assertThat; import static org.assertj.core.api.InstanceOfAssertFactories.map; @@ -63,72 +50,38 @@ public class GraphQlWebSocketHandlerTests { private static final Jackson2JsonDecoder decoder = new Jackson2JsonDecoder(); - private static final String SUBSCRIPTION_ID = "123"; - - private static final String BOOK_SEARCH_QUERY = "{" + - "\"id\":\"" + SUBSCRIPTION_ID + "\"," + - "\"type\":\"subscribe\"," + - "\"payload\":{\"query\": \"" + - " subscription TestSubscription {" + - " bookSearch(minPages: 200) {" + - " id" + - " name" + - " pageCount" + - " author" + - " }}\"}" + - "}"; - @Test - void query() throws Exception { - String bookQuery = "{" + - "\"id\":\"" + SUBSCRIPTION_ID + "\"," + - "\"type\":\"subscribe\"," + - "\"payload\":{\"query\": \"" + - " query TestQuery {" + - " bookById(id: \\\"book-1\\\"){ " + - " id" + - " name" + - " pageCount" + - " author" + - " }}\"}" + - "}"; - - Flux input = Flux.just( + void query() { + TestWebSocketSession session = handle(Flux.just( toWebSocketMessage("{\"type\":\"connection_init\"}"), - toWebSocketMessage(bookQuery)); - - TestWebSocketSession session = new TestWebSocketSession(input); - initWebSocketHandler().handle(session).block(); + toWebSocketMessage(BookTestUtils.BOOK_QUERY))); StepVerifier.create(session.getOutput()) .consumeNextWith(message -> assertMessageType(message, "connection_ack")) .consumeNextWith(message -> assertThat(decode(message)) .hasSize(3) - .containsEntry("id", SUBSCRIPTION_ID) + .containsEntry("id", BookTestUtils.SUBSCRIPTION_ID) .containsEntry("type", "next") .extractingByKey("payload", as(map(String.class, Object.class))) .extractingByKey("data", as(map(String.class, Object.class))) .extractingByKey("bookById", as(map(String.class, Object.class))) - .containsEntry("name", "GraphQL for beginners")) + .containsEntry("name", "Nineteen Eighty-Four")) .consumeNextWith(message -> assertMessageType(message, "complete")) .verifyComplete(); } @Test - void subscription() throws Exception { - Flux input = Flux.just( + void subscription() { + TestWebSocketSession session = handle(Flux.just( toWebSocketMessage("{\"type\":\"connection_init\"}"), - toWebSocketMessage(BOOK_SEARCH_QUERY)); - - TestWebSocketSession session = new TestWebSocketSession(input); - initWebSocketHandler().handle(session).block(); + toWebSocketMessage(BookTestUtils.BOOK_SUBSCRIPTION))); BiConsumer bookPayloadAssertion = (message, bookId) -> assertThat(decode(message)) .hasSize(3) - .containsEntry("id", SUBSCRIPTION_ID) + .containsEntry("id", BookTestUtils.SUBSCRIPTION_ID) .containsEntry("type", "next") .extractingByKey("payload", as(map(String.class, Object.class))) .extractingByKey("data", as(map(String.class, Object.class))) @@ -137,21 +90,17 @@ public class GraphQlWebSocketHandlerTests { StepVerifier.create(session.getOutput()) .consumeNextWith(message -> assertMessageType(message, "connection_ack")) - .consumeNextWith(message -> bookPayloadAssertion.accept(message, "book-2")) - .consumeNextWith(message -> bookPayloadAssertion.accept(message, "book-3")) - .consumeNextWith(message -> bookPayloadAssertion.accept(message, "book-3")) + .consumeNextWith(message -> bookPayloadAssertion.accept(message, "1")) + .consumeNextWith(message -> bookPayloadAssertion.accept(message, "5")) .consumeNextWith(message -> assertMessageType(message, "complete")) .verifyComplete(); } @Test - void unauthorizedWithoutMessageType() throws Exception { - Flux input = Flux.just( + void unauthorizedWithoutMessageType() { + TestWebSocketSession session = handle(Flux.just( toWebSocketMessage("{\"type\":\"connection_init\"}"), - toWebSocketMessage("{\"id\":\"" + SUBSCRIPTION_ID + "\"}")); // No message type - - TestWebSocketSession session = new TestWebSocketSession(input); - initWebSocketHandler().handle(session).block(); + toWebSocketMessage("{\"id\":\"" + BookTestUtils.SUBSCRIPTION_ID + "\"}"))); StepVerifier.create(session.getOutput()) .consumeNextWith(message -> assertMessageType(message, "connection_ack")) @@ -163,13 +112,12 @@ public class GraphQlWebSocketHandlerTests { } @Test - void invalidMessageWithoutId() throws Exception { + void invalidMessageWithoutId() { Flux input = Flux.just( toWebSocketMessage("{\"type\":\"connection_init\"}"), toWebSocketMessage("{\"type\":\"subscribe\"}")); // No message id - TestWebSocketSession session = new TestWebSocketSession(input); - initWebSocketHandler().handle(session).block(); + TestWebSocketSession session = handle(input); StepVerifier.create(session.getOutput()) .consumeNextWith(message -> assertMessageType(message, "connection_ack")) @@ -181,9 +129,8 @@ public class GraphQlWebSocketHandlerTests { } @Test - void unauthorizedWithoutConnectionInit() throws Exception { - TestWebSocketSession session = new TestWebSocketSession(Flux.just(toWebSocketMessage(BOOK_SEARCH_QUERY))); - initWebSocketHandler().handle(session).block(); + void unauthorizedWithoutConnectionInit() { + TestWebSocketSession session = handle(Flux.just(toWebSocketMessage(BookTestUtils.BOOK_SUBSCRIPTION))); StepVerifier.create(session.getOutput()).verifyComplete(); StepVerifier.create(session.closeStatus()) @@ -192,13 +139,10 @@ public class GraphQlWebSocketHandlerTests { } @Test - void tooManyConnectionInitRequests() throws Exception { - Flux input = Flux.just( + void tooManyConnectionInitRequests() { + TestWebSocketSession session = handle(Flux.just( toWebSocketMessage("{\"type\":\"connection_init\"}"), - toWebSocketMessage("{\"type\":\"connection_init\"}")); - - TestWebSocketSession session = new TestWebSocketSession(input); - initWebSocketHandler().handle(session).block(); + toWebSocketMessage("{\"type\":\"connection_init\"}"))); StepVerifier.create(session.getOutput()) .consumeNextWith(message -> assertMessageType(message, "connection_ack")) @@ -210,9 +154,12 @@ public class GraphQlWebSocketHandlerTests { } @Test - void connectionInitTimeout() throws Exception { + void connectionInitTimeout() { + GraphQlWebSocketHandler handler = new GraphQlWebSocketHandler( + BookTestUtils.initWebGraphQlHandler(), ServerCodecConfigurer.create(), Duration.ofMillis(50)); + TestWebSocketSession session = new TestWebSocketSession(Flux.empty()); - initWebSocketHandler(Collections.emptyList(), Duration.ofMillis(50)).handle(session).block(); + handler.handle(session).block(); StepVerifier.create(session.closeStatus()) .expectNext(new CloseStatus(4408, "Connection initialisation timeout")) @@ -220,23 +167,18 @@ public class GraphQlWebSocketHandlerTests { } @Test - void subscriptionExists() throws Exception { - Flux input = Flux.just( + void subscriptionExists() { + TestWebSocketSession session = handle(Flux.just( toWebSocketMessage("{\"type\":\"connection_init\"}"), - toWebSocketMessage(BOOK_SEARCH_QUERY), - toWebSocketMessage(BOOK_SEARCH_QUERY)); - - List interceptors = Collections.singletonList(new ConsumeOneAndNeverCompleteInterceptor()); - - TestWebSocketSession session = new TestWebSocketSession(input); - initWebSocketHandler(interceptors, null).handle(session).block(); + toWebSocketMessage(BookTestUtils.BOOK_SUBSCRIPTION), + toWebSocketMessage(BookTestUtils.BOOK_SUBSCRIPTION)), new ConsumeOneAndNeverCompleteInterceptor()); // Collect messages until session closed List> messages = new ArrayList<>(); session.getOutput().subscribe(message -> messages.add(decode(message))); StepVerifier.create(session.closeStatus()) - .expectNext(new CloseStatus(4409, "Subscriber for " + SUBSCRIPTION_ID + " already exists")) + .expectNext(new CloseStatus(4409, "Subscriber for " + BookTestUtils.SUBSCRIPTION_ID + " already exists")) .verifyComplete(); assertThat(messages.size()).isEqualTo(2); @@ -245,55 +187,36 @@ public class GraphQlWebSocketHandlerTests { } @Test - void clientCompletion() throws Exception { + void clientCompletion() { Sinks.Many input = Sinks.many().unicast().onBackpressureBuffer(); input.tryEmitNext(toWebSocketMessage("{\"type\":\"connection_init\"}")); - input.tryEmitNext(toWebSocketMessage(BOOK_SEARCH_QUERY)); + input.tryEmitNext(toWebSocketMessage(BookTestUtils.BOOK_SUBSCRIPTION)); - List interceptors = Collections.singletonList(new ConsumeOneAndNeverCompleteInterceptor()); - TestWebSocketSession session = new TestWebSocketSession(input.asFlux()); - initWebSocketHandler(interceptors, null).handle(session).block(); + TestWebSocketSession session = + handle(input.asFlux(), new ConsumeOneAndNeverCompleteInterceptor()); - String completeMessage = "{\"id\":\"" + SUBSCRIPTION_ID + "\",\"type\":\"complete\"}"; + String completeMessage = "{\"id\":\"" + BookTestUtils.SUBSCRIPTION_ID + "\",\"type\":\"complete\"}"; StepVerifier.create(session.getOutput()) .consumeNextWith(message -> assertMessageType(message, "connection_ack")) .consumeNextWith(message -> assertMessageType(message, "next")) .then(() -> input.tryEmitNext(toWebSocketMessage(completeMessage))) .as("Second subscription with same id is possible only if the first was properly removed") - .then(() -> input.tryEmitNext(toWebSocketMessage(BOOK_SEARCH_QUERY))) + .then(() -> input.tryEmitNext(toWebSocketMessage(BookTestUtils.BOOK_SUBSCRIPTION))) .consumeNextWith(message -> assertMessageType(message, "next")) .then(() -> input.tryEmitNext(toWebSocketMessage(completeMessage))) .verifyTimeout(Duration.ofMillis(500)); } - private GraphQlWebSocketHandler initWebSocketHandler() throws Exception { - return initWebSocketHandler(Collections.emptyList(), Duration.ofSeconds(60)); - } - - private GraphQlWebSocketHandler initWebSocketHandler( - @Nullable List interceptors, @Nullable Duration initTimeoutDuration) { - - WebGraphQlHandler graphQlHandler = - WebGraphQlHandler.builder(new ExecutionGraphQlService(graphQlSource())) - .interceptors(interceptors != null ? interceptors : Collections.emptyList()) - .build(); - - return new GraphQlWebSocketHandler(graphQlHandler, + private TestWebSocketSession handle(Flux input, WebInterceptor... interceptors) { + GraphQlWebSocketHandler handler = new GraphQlWebSocketHandler( + BookTestUtils.initWebGraphQlHandler(interceptors), ServerCodecConfigurer.create(), - (initTimeoutDuration != null ? initTimeoutDuration : Duration.ofSeconds(60))); - } + Duration.ofSeconds(60)); - private static GraphQlSource graphQlSource() { - RuntimeWiring.Builder builder = RuntimeWiring.newRuntimeWiring(); - builder.type(newTypeWiring("Query").dataFetcher("bookById", GraphQlDataFetchers.getBookByIdDataFetcher())); - builder.type(newTypeWiring("Subscription").dataFetcher("bookSearch", GraphQlDataFetchers.getBooksOnSale())); - RuntimeWiring runtimeWiring = builder.build(); - - return GraphQlSource.builder() - .schemaResource(new ClassPathResource("books/schema.graphqls")) - .runtimeWiring(runtimeWiring) - .build(); + TestWebSocketSession session = new TestWebSocketSession(input); + handler.handle(session).block(); + return session; } private static WebSocketMessage toWebSocketMessage(String data) { @@ -312,61 +235,7 @@ public class GraphQlWebSocketHandlerTests { Map map = decode(message); assertThat(map).containsEntry("type", messageType); if (!messageType.equals("connection_ack")) { - assertThat(map).containsEntry("id", SUBSCRIPTION_ID); - } - } - - - private static class TestWebSocketSession extends AbstractWebSocketSession { - - private final Flux input; - - private Flux output = Flux.empty(); - - private final Sinks.One closeStatusSink = Sinks.one(); - - - public TestWebSocketSession(Flux input) { - this("1", URI.create("https://example.org/graphql"), input); - } - - public TestWebSocketSession(String id, URI uri, Flux input) { - super(new Object(), id, - new HandshakeInfo(uri, new HttpHeaders(), Mono.empty(), null), - DefaultDataBufferFactory.sharedInstance); - this.input = input; - } - - - @Override - public Flux receive() { - return this.input; - } - - @Override - public Mono send(Publisher messages) { - this.output = Flux.from(messages); - return Mono.empty(); - } - - public Flux getOutput() { - return this.output; - } - - @Override - public boolean isOpen() { - throw new java.lang.UnsupportedOperationException(); - } - - @Override - public Mono close(CloseStatus status) { - this.closeStatusSink.tryEmitValue(status); - return Mono.empty(); - } - - @Override - public Mono closeStatus() { - return this.closeStatusSink.asMono(); + assertThat(map).containsEntry("id", BookTestUtils.SUBSCRIPTION_ID); } } diff --git a/spring-graphql/src/test/java/org/springframework/graphql/web/webflux/TestWebSocketSession.java b/spring-graphql/src/test/java/org/springframework/graphql/web/webflux/TestWebSocketSession.java new file mode 100644 index 00000000..33f027bb --- /dev/null +++ b/spring-graphql/src/test/java/org/springframework/graphql/web/webflux/TestWebSocketSession.java @@ -0,0 +1,87 @@ +/* + * 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.webflux; + +import java.net.URI; + +import org.reactivestreams.Publisher; +import reactor.core.publisher.Flux; +import reactor.core.publisher.Mono; +import reactor.core.publisher.Sinks; + +import org.springframework.core.io.buffer.DefaultDataBufferFactory; +import org.springframework.http.HttpHeaders; +import org.springframework.web.reactive.socket.CloseStatus; +import org.springframework.web.reactive.socket.HandshakeInfo; +import org.springframework.web.reactive.socket.WebSocketMessage; +import org.springframework.web.reactive.socket.adapter.AbstractWebSocketSession; + +/** + * {@link org.springframework.web.reactive.socket.WebSocketSession} that is given + * an input Flux of messages and exposes a Flux of published output messages. + */ +class TestWebSocketSession extends AbstractWebSocketSession { + + private final Flux input; + + private Flux output = Flux.empty(); + + private final Sinks.One closeStatusSink = Sinks.one(); + + + public TestWebSocketSession(Flux input) { + this("1", URI.create("https://example.org/graphql"), input); + } + + public TestWebSocketSession(String id, URI uri, Flux input) { + super(new Object(), id, + new HandshakeInfo(uri, new HttpHeaders(), Mono.empty(), null), + DefaultDataBufferFactory.sharedInstance); + this.input = input; + } + + + @Override + public Flux receive() { + return this.input; + } + + @Override + public Mono send(Publisher messages) { + this.output = Flux.from(messages); + return Mono.empty(); + } + + public Flux getOutput() { + return this.output; + } + + @Override + public boolean isOpen() { + throw new UnsupportedOperationException(); + } + + @Override + public Mono close(CloseStatus status) { + this.closeStatusSink.tryEmitValue(status); + return Mono.empty(); + } + + @Override + public Mono closeStatus() { + return this.closeStatusSink.asMono(); + } +} 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 5f99c148..65fb25f7 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 @@ -20,33 +20,25 @@ 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; import java.util.function.Consumer; -import graphql.schema.idl.RuntimeWiring; import org.junit.jupiter.api.Test; import reactor.test.StepVerifier; -import org.springframework.core.io.ClassPathResource; -import org.springframework.graphql.execution.ExecutionGraphQlService; -import org.springframework.graphql.execution.GraphQlSource; +import org.springframework.graphql.web.BookTestUtils; import org.springframework.graphql.web.ConsumeOneAndNeverCompleteInterceptor; -import org.springframework.graphql.web.GraphQlDataFetchers; -import org.springframework.graphql.web.WebGraphQlHandler; import org.springframework.graphql.web.WebInterceptor; import org.springframework.http.HttpHeaders; import org.springframework.http.HttpInputMessage; import org.springframework.http.converter.HttpMessageConverter; import org.springframework.http.converter.json.MappingJackson2HttpMessageConverter; -import org.springframework.lang.Nullable; import org.springframework.web.socket.CloseStatus; import org.springframework.web.socket.TextMessage; import org.springframework.web.socket.WebSocketMessage; -import static graphql.schema.idl.TypeRuntimeWiring.newTypeWiring; import static org.assertj.core.api.Assertions.as; import static org.assertj.core.api.Assertions.assertThat; import static org.assertj.core.api.InstanceOfAssertFactories.map; @@ -56,166 +48,134 @@ import static org.assertj.core.api.InstanceOfAssertFactories.map; */ public class GraphQlWebSocketHandlerTests { - private static final String SUBSCRIPTION_ID = "123"; - - private static final String BOOK_SEARCH_QUERY = "{" + - "\"id\":\"" + SUBSCRIPTION_ID + "\"," + - "\"type\":\"subscribe\"," + - "\"payload\":{\"query\": \"" + - " subscription TestSubscription {" + - " bookSearch(minPages: 200) {" + - " id" + - " name" + - " pageCount" + - " author" + - " }}\"}" + - "}"; - private static final HttpMessageConverter converter = new MappingJackson2HttpMessageConverter(); private final TestWebSocketSession session = new TestWebSocketSession(); - private final GraphQlWebSocketHandler handler = - initWebSocketHandler(Collections.emptyList(), Duration.ofSeconds(60)); + private final GraphQlWebSocketHandler handler = initWebSocketHandler(); @Test void query() throws Exception { - String bookQuery = "{" + - "\"id\":\"" + SUBSCRIPTION_ID + "\"," + - "\"type\":\"subscribe\"," + - "\"payload\":{\"query\": \"" + - " query TestQuery {" + - " bookById(id: \\\"book-1\\\"){ " + - " id" + - " name" + - " pageCount" + - " author" + - " }}\"}" + - "}"; + handle(this.handler, + new TextMessage("{\"type\":\"connection_init\"}"), + new TextMessage(BookTestUtils.BOOK_QUERY)); - this.handler.afterConnectionEstablished(session); - this.handler.handleTextMessage(session, new TextMessage("{\"type\":\"connection_init\"}")); - this.handler.handleTextMessage(session, new TextMessage(bookQuery)); - - StepVerifier.create(session.getOutput()) + StepVerifier.create(this.session.getOutput()) .consumeNextWith(message -> assertMessageType(message, "connection_ack")) .consumeNextWith(message -> assertThat(decode(message)) .hasSize(3) - .containsEntry("id", SUBSCRIPTION_ID) + .containsEntry("id", BookTestUtils.SUBSCRIPTION_ID) .containsEntry("type", "next") .extractingByKey("payload", as(map(String.class, Object.class))) .extractingByKey("data", as(map(String.class, Object.class))) .extractingByKey("bookById", as(map(String.class, Object.class))) - .containsEntry("name", "GraphQL for beginners")) + .containsEntry("name", "Nineteen Eighty-Four")) .consumeNextWith(message -> assertMessageType(message, "complete")) - .then(session::close) // Complete output Flux + .then(this.session::close) // Complete output Flux .verifyComplete(); } @Test void subscription() throws Exception { - this.handler.afterConnectionEstablished(session); - this.handler.handleTextMessage(session, new TextMessage("{\"type\":\"connection_init\"}")); - this.handler.handleTextMessage(session, new TextMessage(BOOK_SEARCH_QUERY)); + handle(this.handler, + new TextMessage("{\"type\":\"connection_init\"}"), + new TextMessage(BookTestUtils.BOOK_SUBSCRIPTION)); BiConsumer, String> bookPayloadAssertion = (message, bookId) -> assertThat(decode(message)) .hasSize(3) - .containsEntry("id", SUBSCRIPTION_ID) + .containsEntry("id", BookTestUtils.SUBSCRIPTION_ID) .containsEntry("type", "next") .extractingByKey("payload", as(map(String.class, Object.class))) .extractingByKey("data", as(map(String.class, Object.class))) .extractingByKey("bookSearch", as(map(String.class, Object.class))) .containsEntry("id", bookId); - StepVerifier.create(session.getOutput()) + StepVerifier.create(this.session.getOutput()) .consumeNextWith(message -> assertMessageType(message, "connection_ack")) - .consumeNextWith(message -> bookPayloadAssertion.accept(message, "book-2")) - .consumeNextWith(message -> bookPayloadAssertion.accept(message, "book-3")) - .consumeNextWith(message -> bookPayloadAssertion.accept(message, "book-3")) + .consumeNextWith(message -> bookPayloadAssertion.accept(message, "1")) + .consumeNextWith(message -> bookPayloadAssertion.accept(message, "5")) .consumeNextWith(message -> assertMessageType(message, "complete")) - .then(session::close) // Complete output Flux + .then(this.session::close) // Complete output Flux .verifyComplete(); } @Test void unauthorizedWithoutMessageType() throws Exception { - this.handler.afterConnectionEstablished(session); - this.handler.handleTextMessage(session, new TextMessage("{\"type\":\"connection_init\"}")); - this.handler.handleTextMessage(session, new TextMessage("{\"id\":\"" + SUBSCRIPTION_ID + "\"}")); // No message type + handle(this.handler, + new TextMessage("{\"type\":\"connection_init\"}"), + new TextMessage("{\"id\":\"" + BookTestUtils.SUBSCRIPTION_ID + "\"}")); // No message type - StepVerifier.create(session.getOutput()) + StepVerifier.create(this.session.getOutput()) .consumeNextWith(message -> assertMessageType(message, "connection_ack")) .verifyComplete(); - assertThat(session.getCloseStatus()).isEqualTo(new CloseStatus(4400, "Invalid message")); + assertThat(this.session.getCloseStatus()).isEqualTo(new CloseStatus(4400, "Invalid message")); } @Test void invalidMessageWithoutId() throws Exception { - this.handler.afterConnectionEstablished(session); - this.handler.handleTextMessage(session, new TextMessage("{\"type\":\"connection_init\"}")); - this.handler.handleTextMessage(session, new TextMessage("{\"type\":\"subscribe\"}")); // No message id + handle(this.handler, + new TextMessage("{\"type\":\"connection_init\"}"), + new TextMessage("{\"type\":\"subscribe\"}")); // No message id - StepVerifier.create(session.getOutput()) + StepVerifier.create(this.session.getOutput()) .consumeNextWith(message -> assertMessageType(message, "connection_ack")) .verifyComplete(); - assertThat(session.getCloseStatus()).isEqualTo(new CloseStatus(4400, "Invalid message")); + assertThat(this.session.getCloseStatus()).isEqualTo(new CloseStatus(4400, "Invalid message")); } @Test void unauthorizedWithoutConnectionInit() throws Exception { - this.handler.afterConnectionEstablished(session); - this.handler.handleTextMessage(session, new TextMessage(BOOK_SEARCH_QUERY)); + handle(this.handler, new TextMessage(BookTestUtils.BOOK_SUBSCRIPTION)); - StepVerifier.create(session.getOutput()).verifyComplete(); - assertThat(session.getCloseStatus()).isEqualTo(new CloseStatus(4401, "Unauthorized")); + StepVerifier.create(this.session.getOutput()).verifyComplete(); + assertThat(this.session.getCloseStatus()).isEqualTo(new CloseStatus(4401, "Unauthorized")); } @Test void tooManyConnectionInitRequests() throws Exception { - this.handler.afterConnectionEstablished(session); - this.handler.handleTextMessage(session, new TextMessage("{\"type\":\"connection_init\"}")); - this.handler.handleTextMessage(session, new TextMessage("{\"type\":\"connection_init\"}")); + handle(this.handler, + new TextMessage("{\"type\":\"connection_init\"}"), + new TextMessage("{\"type\":\"connection_init\"}")); - StepVerifier.create(session.getOutput()) + StepVerifier.create(this.session.getOutput()) .consumeNextWith(message -> assertMessageType(message, "connection_ack")) .verifyComplete(); - assertThat(session.getCloseStatus()) + assertThat(this.session.getCloseStatus()) .isEqualTo(new CloseStatus(4429, "Too many initialisation requests")); } @Test void connectionInitTimeout() { - GraphQlWebSocketHandler handler = initWebSocketHandler(Collections.emptyList(), Duration.ofMillis(50)); - handler.afterConnectionEstablished(session); + GraphQlWebSocketHandler handler = new GraphQlWebSocketHandler( + BookTestUtils.initWebGraphQlHandler(), converter, Duration.ofMillis(50)); - StepVerifier.create(session.closeStatus()) + handler.afterConnectionEstablished(this.session); + + StepVerifier.create(this.session.closeStatus()) .expectNext(new CloseStatus(4408, "Connection initialisation timeout")) .verifyComplete(); } @Test void subscriptionExists() throws Exception { - GraphQlWebSocketHandler handler = initWebSocketHandler( - Collections.singletonList(new ConsumeOneAndNeverCompleteInterceptor()), null); - - handler.afterConnectionEstablished(session); - handler.handleTextMessage(session, new TextMessage("{\"type\":\"connection_init\"}")); - handler.handleTextMessage(session, new TextMessage(BOOK_SEARCH_QUERY)); - handler.handleTextMessage(session, new TextMessage(BOOK_SEARCH_QUERY)); + handle(initWebSocketHandler(new ConsumeOneAndNeverCompleteInterceptor()), + new TextMessage("{\"type\":\"connection_init\"}"), + new TextMessage(BookTestUtils.BOOK_SUBSCRIPTION), + new TextMessage(BookTestUtils.BOOK_SUBSCRIPTION)); // Collect messages until session closed List> messages = new ArrayList<>(); session.getOutput().subscribe(message -> messages.add(decode(message))); - StepVerifier.create(session.closeStatus()) - .expectNext(new CloseStatus(4409, "Subscriber for " + SUBSCRIPTION_ID + " already exists")) + StepVerifier.create(this.session.closeStatus()) + .expectNext(new CloseStatus(4409, "Subscriber for " + BookTestUtils.SUBSCRIPTION_ID + " already exists")) .verifyComplete(); assertThat(messages.size()).isEqualTo(2); @@ -225,69 +185,56 @@ public class GraphQlWebSocketHandlerTests { @Test void clientCompletion() throws Exception { - GraphQlWebSocketHandler handler = initWebSocketHandler( - Collections.singletonList(new ConsumeOneAndNeverCompleteInterceptor()), null); + GraphQlWebSocketHandler handler = + initWebSocketHandler(new ConsumeOneAndNeverCompleteInterceptor()); - handler.afterConnectionEstablished(session); - handler.handleTextMessage(session, new TextMessage("{\"type\":\"connection_init\"}")); - handler.handleTextMessage(session, new TextMessage(BOOK_SEARCH_QUERY)); + handle(handler, + new TextMessage("{\"type\":\"connection_init\"}"), + new TextMessage(BookTestUtils.BOOK_SUBSCRIPTION)); - String completeMessage = "{\"id\":\"" + SUBSCRIPTION_ID + "\",\"type\":\"complete\"}"; + String completeMessage = "{\"id\":\"" + BookTestUtils.SUBSCRIPTION_ID + "\",\"type\":\"complete\"}"; Consumer messageSender = body -> { try { - handler.handleTextMessage(session, new TextMessage(body)); + handler.handleTextMessage(this.session, new TextMessage(body)); } catch (Exception ex) { throw new IllegalStateException(ex); } }; - StepVerifier.create(session.getOutput()) + StepVerifier.create(this.session.getOutput()) .consumeNextWith(message -> assertMessageType(message, "connection_ack")) .consumeNextWith(message -> assertMessageType(message, "next")) .then(() -> messageSender.accept(completeMessage)) .as("Second subscription with same id is possible only if the first was properly removed") - .then(() -> messageSender.accept(BOOK_SEARCH_QUERY)) + .then(() -> messageSender.accept(BookTestUtils.BOOK_SUBSCRIPTION)) .consumeNextWith(message -> assertMessageType(message, "next")) .then(() -> messageSender.accept(completeMessage)) .verifyTimeout(Duration.ofMillis(500)); } + private void handle(GraphQlWebSocketHandler handler, TextMessage... textMessages) throws Exception { + handler.afterConnectionEstablished(this.session); + for (TextMessage message : textMessages) { + handler.handleTextMessage(this.session, message); + } + } - private GraphQlWebSocketHandler initWebSocketHandler( - @Nullable List interceptors, @Nullable Duration initTimeoutDuration) { - + private GraphQlWebSocketHandler initWebSocketHandler(WebInterceptor... interceptors) { try { - WebGraphQlHandler graphQlHandler = - WebGraphQlHandler.builder(new ExecutionGraphQlService(graphQlSource())) - .interceptors(interceptors != null ? interceptors : Collections.emptyList()) - .build(); - - return new GraphQlWebSocketHandler(graphQlHandler, converter, - (initTimeoutDuration != null ? initTimeoutDuration : Duration.ofSeconds(60))); + return new GraphQlWebSocketHandler( + BookTestUtils.initWebGraphQlHandler(interceptors), converter, Duration.ofSeconds(60)); } catch (Exception ex) { throw new IllegalStateException(ex); } } - private static GraphQlSource graphQlSource() { - RuntimeWiring.Builder builder = RuntimeWiring.newRuntimeWiring(); - builder.type(newTypeWiring("Query").dataFetcher("bookById", GraphQlDataFetchers.getBookByIdDataFetcher())); - builder.type(newTypeWiring("Subscription").dataFetcher("bookSearch", GraphQlDataFetchers.getBooksOnSale())); - RuntimeWiring runtimeWiring = builder.build(); - - return GraphQlSource.builder() - .schemaResource(new ClassPathResource("books/schema.graphqls")) - .runtimeWiring(runtimeWiring) - .build(); - } - private void assertMessageType(WebSocketMessage message, String messageType) { Map map = decode(message, Map.class); assertThat(map).containsEntry("type", messageType); if (!messageType.equals("connection_ack")) { - assertThat(map).containsEntry("id", SUBSCRIPTION_ID); + assertThat(map).containsEntry("id", BookTestUtils.SUBSCRIPTION_ID); } } diff --git a/spring-graphql/src/test/java/org/springframework/graphql/web/webmvc/TestWebSocketSession.java b/spring-graphql/src/test/java/org/springframework/graphql/web/webmvc/TestWebSocketSession.java index 1283d38d..a7b9f307 100644 --- a/spring-graphql/src/test/java/org/springframework/graphql/web/webmvc/TestWebSocketSession.java +++ b/spring-graphql/src/test/java/org/springframework/graphql/web/webmvc/TestWebSocketSession.java @@ -35,7 +35,8 @@ import org.springframework.web.socket.WebSocketMessage; import org.springframework.web.socket.WebSocketSession; /** - * + * WebSocketSession that saves sent messages and exposes them as a Flux which + * makes assertions comparable to the same for WebFlux. */ public class TestWebSocketSession implements WebSocketSession { diff --git a/spring-graphql/src/test/resources/books/schema.graphqls b/spring-graphql/src/test/resources/books/schema.graphqls index 7ec591e5..cb787d52 100644 --- a/spring-graphql/src/test/resources/books/schema.graphqls +++ b/spring-graphql/src/test/resources/books/schema.graphqls @@ -5,10 +5,9 @@ type Query { type Book { id: ID name: String - pageCount: Int author: String } type Subscription { - bookSearch(minPages:Int) : Book! + bookSearch(author: String) : Book! }