diff --git a/spring-graphql-test/src/test/java/org/springframework/graphql/test/tester/TestWebSocketClient.java b/spring-graphql-test/src/test/java/org/springframework/graphql/test/tester/TestWebSocketClient.java deleted file mode 100644 index b7fa1c37..00000000 --- a/spring-graphql-test/src/test/java/org/springframework/graphql/test/tester/TestWebSocketClient.java +++ /dev/null @@ -1,76 +0,0 @@ -/* - * Copyright 2002-2022 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.test.tester; - -import java.net.URI; -import java.util.List; -import java.util.concurrent.CopyOnWriteArrayList; - -import reactor.core.publisher.Mono; - -import org.springframework.http.HttpHeaders; -import org.springframework.util.Assert; -import org.springframework.web.reactive.socket.WebSocketHandler; -import org.springframework.web.reactive.socket.client.WebSocketClient; - -/** - * Copy of same class in spring-graphql tests. - */ -final class TestWebSocketClient implements WebSocketClient { - - private final WebSocketHandler serverHandler; - - private final List connections = new CopyOnWriteArrayList<>(); - - - public TestWebSocketClient(WebSocketHandler serverHandler) { - this.serverHandler = serverHandler; - } - - - /** - * Return the connection at the specified index from a list of connections - * based on order of execution. - */ - public TestWebSocketConnection getConnection(int index) { - Assert.isTrue(index < this.connections.size(), - "No connection at index=" + index + ", total=" + this.connections.size()); - return connections.get(index); - } - - /** - * Return the number of connections which corresponds to the number of calls - * to one of the execute methods. - */ - public int getConnectionCount() { - return this.connections.size(); - } - - - @Override - public Mono execute(URI url, WebSocketHandler clientHandler) { - return execute(URI.create("/"), HttpHeaders.EMPTY, clientHandler); - } - - @Override - public Mono execute(URI url, HttpHeaders headers, WebSocketHandler clientHandler) { - TestWebSocketConnection connection = new TestWebSocketConnection(url, headers); - this.connections.add(connection); - return connection.connect(clientHandler, this.serverHandler); - } - -} diff --git a/spring-graphql-test/src/test/java/org/springframework/graphql/test/tester/TestWebSocketConnection.java b/spring-graphql-test/src/test/java/org/springframework/graphql/test/tester/TestWebSocketConnection.java deleted file mode 100644 index 289fd765..00000000 --- a/spring-graphql-test/src/test/java/org/springframework/graphql/test/tester/TestWebSocketConnection.java +++ /dev/null @@ -1,267 +0,0 @@ -/* - * Copyright 2002-2022 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.test.tester; - -import java.net.URI; -import java.util.ArrayList; -import java.util.List; -import java.util.Queue; -import java.util.concurrent.ConcurrentLinkedQueue; -import java.util.concurrent.atomic.AtomicLong; - -import org.reactivestreams.Publisher; -import reactor.core.Scannable; -import reactor.core.publisher.Flux; -import reactor.core.publisher.Mono; -import reactor.core.publisher.Sinks; - -import org.springframework.core.io.buffer.DataBuffer; -import org.springframework.core.io.buffer.DefaultDataBufferFactory; -import org.springframework.http.HttpHeaders; -import org.springframework.lang.Nullable; -import org.springframework.util.Assert; -import org.springframework.web.reactive.socket.CloseStatus; -import org.springframework.web.reactive.socket.HandshakeInfo; -import org.springframework.web.reactive.socket.WebSocketHandler; -import org.springframework.web.reactive.socket.WebSocketMessage; -import org.springframework.web.reactive.socket.adapter.AbstractWebSocketSession; - -/** - * Copy of same class in spring-graphql tests. - */ -final class TestWebSocketConnection { - - private static final AtomicLong connectionIndex = new AtomicLong(); - - - private final URI url; - - private final HttpHeaders headers; - - private final TestWebSocketSession clientSession; - - private final TestWebSocketSession serverSession; - - - public TestWebSocketConnection(URI url, HttpHeaders headers) { - - this.url = url; - this.headers = headers; - - long id = connectionIndex.incrementAndGet(); - - Sinks.Many clientSink = Sinks.many().unicast().onBackpressureBuffer(); - Sinks.Many serverSink = Sinks.many().unicast().onBackpressureBuffer(); - - Sinks.One clientStatusSink = Sinks.one(); - Sinks.One serverStatusSink = Sinks.one(); - - this.clientSession = new TestWebSocketSession("client-session-" + id, url, headers, - clientSink, serverSink.asFlux(), clientStatusSink, serverStatusSink.asMono()); - - this.serverSession = new TestWebSocketSession("server-session-" + id, url, headers, - serverSink, clientSink.asFlux(), serverStatusSink, clientStatusSink.asMono()); - } - - - public URI getUrl() { - return this.url; - } - - public HttpHeaders getHeaders() { - return this.headers; - } - - /** - * Return {@code true} if both client and server sessions are open. - */ - public boolean isOpen() { - return (this.clientSession.isOpen() && this.serverSession.isOpen()); - } - - /** - * Return messages sent from the client side. - */ - public List getClientMessages() { - return this.clientSession.getSentMessages(); - } - - /** - * Return messages sent from the server side. - */ - public List getServerMessages() { - return this.serverSession.getSentMessages(); - } - - - /** - * Starts client and server session handling, and if either side errors or - * completes, close its session with either {@link CloseStatus#NORMAL} or - * {@link CloseStatus#PROTOCOL_ERROR} respectively. - * @param clientHandler the client session handler - * @param serverHandler the server session handler - * @return {@code Mono} that completes when either client or server - * session handling completes or when either is closed - */ - Mono connect(WebSocketHandler clientHandler, WebSocketHandler serverHandler) { - - Mono serverMono = invokeHandler(serverHandler, this.serverSession); - Mono clientMono = invokeHandler(clientHandler, this.clientSession); - - Mono serverStatusMono = this.serverSession.closeStatus().then(); - Mono clientStatusMono = this.clientSession.closeStatus().then(); - - return Mono.zip(serverMono, clientMono, serverStatusMono, clientStatusMono).then(); - } - - /** - * Handle the session and complete it when handling completes. - */ - private Mono invokeHandler(WebSocketHandler serverHandler, TestWebSocketSession session) { - return serverHandler.handle(session) - .then(Mono.defer(() -> session.close(CloseStatus.NORMAL))) - .onErrorResume(ex -> session.close(CloseStatus.PROTOCOL_ERROR).then(Mono.error(ex))); - } - - - /** - * Close the connection from the client side. - */ - public Mono closeClientSession(CloseStatus status) { - return this.clientSession.close(status); - } - - /** - * Close the connection from the server side. - */ - public Mono closeServerSession(CloseStatus status) { - return this.serverSession.close(status); - } - - /** - * Return the {@code CloseStatus} that may have come from either side. - */ - public Mono closeStatus() { - return this.clientSession.closeStatus().or(this.serverSession.closeStatus()); - } - - - /** - * Test WebSocketSession that sends to a given {@link Sinks.Many sink} and - * receives from a given {@code Flux}. - */ - private static class TestWebSocketSession extends AbstractWebSocketSession { - - private final Sinks.Many sendSink; - - private final Flux receiveFlux; - - private final Sinks.One closeStatusSink; - - private final Queue sentMessages = new ConcurrentLinkedQueue<>(); - - - TestWebSocketSession(String sessionId, URI url, HttpHeaders headers, - Sinks.Many sendSink, Flux receiveFlux, - Sinks.One closeStatusSink, Mono remoteCloseStatusMono) { - - super(new Object(), sessionId, new HandshakeInfo(url, headers, Mono.empty(), null), - DefaultDataBufferFactory.sharedInstance); - - this.sendSink = sendSink; - this.receiveFlux = receiveFlux.cache(); - this.closeStatusSink = closeStatusSink; - - // Close this side when the remote closes - remoteCloseStatusMono.doOnSuccess(this::handleRemoteClosure).subscribe(); - } - - private void handleRemoteClosure(@Nullable CloseStatus status) { - - if (!isOpen()) { - // when we close, remote closes, and we detect that - return; - } - - if (logger.isDebugEnabled()) { - logger.debug("Closing " + this + " due to remote " + status); - } - - closeInternal(status); - } - - - public List getSentMessages() { - return new ArrayList<>(this.sentMessages); - } - - @Override - public Mono send(Publisher messages) { - return Flux.from(messages) - .doOnNext(this::saveMessage) - .doOnNext(message -> { - Sinks.EmitResult result = this.sendSink.tryEmitNext(message); - Assert.state(result.isSuccess(), this + " failed to send: " + message + ", with " + result); - }) - .then(); - } - - private void saveMessage(WebSocketMessage message) { - DataBuffer payload = message.getPayload().retainedSlice(0, message.getPayload().readableByteCount()); - this.sentMessages.add(new WebSocketMessage(message.getType(), payload)); - } - - @Override - public Flux receive() { - return this.receiveFlux; - } - - @Override - public boolean isOpen() { - return !Boolean.TRUE.equals(this.closeStatusSink.scan(Scannable.Attr.TERMINATED)); - } - - @Override - public Mono closeStatus() { - return this.closeStatusSink.asMono(); - } - - public Mono close(CloseStatus status) { - if (logger.isDebugEnabled()) { - logger.debug("Closing " + this + " with " + status); - } - closeInternal(status); - return Mono.empty(); - } - - private void closeInternal(@Nullable CloseStatus status) { - if (status != null) { - this.closeStatusSink.tryEmitValue(status); - } - else { - this.closeStatusSink.tryEmitEmpty(); - }; - } - - @Override - public String toString() { - return getId(); - } - - } - -} diff --git a/spring-graphql-test/src/test/java/org/springframework/graphql/test/tester/WebGraphQlTesterBuilderTests.java b/spring-graphql-test/src/test/java/org/springframework/graphql/test/tester/WebGraphQlTesterBuilderTests.java index 64646309..f3d349c0 100644 --- a/spring-graphql-test/src/test/java/org/springframework/graphql/test/tester/WebGraphQlTesterBuilderTests.java +++ b/spring-graphql-test/src/test/java/org/springframework/graphql/test/tester/WebGraphQlTesterBuilderTests.java @@ -28,6 +28,8 @@ import org.junit.jupiter.params.provider.MethodSource; import reactor.core.publisher.Mono; import org.springframework.graphql.RequestOutput; +import org.springframework.graphql.web.TestWebSocketClient; +import org.springframework.graphql.web.TestWebSocketConnection; import org.springframework.graphql.support.DocumentSource; import org.springframework.graphql.web.WebGraphQlHandler; import org.springframework.graphql.web.WebInput; diff --git a/spring-graphql/build.gradle b/spring-graphql/build.gradle index af3ed273..eba8b596 100644 --- a/spring-graphql/build.gradle +++ b/spring-graphql/build.gradle @@ -57,6 +57,8 @@ dependencies { testRuntimeOnly 'org.apache.logging.log4j:log4j-core' testRuntimeOnly 'org.apache.logging.log4j:log4j-slf4j-impl' + + testFixturesApi 'org.springframework:spring-webflux' } test { diff --git a/spring-graphql/src/test/java/org/springframework/graphql/client/MockWebSocketGraphQlTransportTests.java b/spring-graphql/src/test/java/org/springframework/graphql/client/MockWebSocketGraphQlTransportTests.java index 7359a312..1995bcf7 100644 --- a/spring-graphql/src/test/java/org/springframework/graphql/client/MockWebSocketGraphQlTransportTests.java +++ b/spring-graphql/src/test/java/org/springframework/graphql/client/MockWebSocketGraphQlTransportTests.java @@ -34,6 +34,8 @@ import reactor.core.publisher.Mono; import reactor.test.StepVerifier; import org.springframework.graphql.GraphQlRequest; +import org.springframework.graphql.web.TestWebSocketClient; +import org.springframework.graphql.web.TestWebSocketConnection; import org.springframework.graphql.support.MapExecutionResult; import org.springframework.graphql.web.webflux.GraphQlWebSocketMessage; import org.springframework.http.HttpHeaders; diff --git a/spring-graphql/src/test/java/org/springframework/graphql/client/WebGraphQlClientBuilderTests.java b/spring-graphql/src/test/java/org/springframework/graphql/client/WebGraphQlClientBuilderTests.java index b9663e61..ad89a785 100644 --- a/spring-graphql/src/test/java/org/springframework/graphql/client/WebGraphQlClientBuilderTests.java +++ b/spring-graphql/src/test/java/org/springframework/graphql/client/WebGraphQlClientBuilderTests.java @@ -28,6 +28,8 @@ import org.junit.jupiter.params.provider.MethodSource; import reactor.core.publisher.Mono; import org.springframework.graphql.RequestOutput; +import org.springframework.graphql.web.TestWebSocketClient; +import org.springframework.graphql.web.TestWebSocketConnection; import org.springframework.graphql.support.DocumentSource; import org.springframework.graphql.web.WebGraphQlHandler; import org.springframework.graphql.web.WebInput; diff --git a/spring-graphql/src/test/java/org/springframework/graphql/client/TestWebSocketClient.java b/spring-graphql/src/testFixtures/java/org/springframework/graphql/web/TestWebSocketClient.java similarity index 95% rename from spring-graphql/src/test/java/org/springframework/graphql/client/TestWebSocketClient.java rename to spring-graphql/src/testFixtures/java/org/springframework/graphql/web/TestWebSocketClient.java index 115b02c2..9b4adb67 100644 --- a/spring-graphql/src/test/java/org/springframework/graphql/client/TestWebSocketClient.java +++ b/spring-graphql/src/testFixtures/java/org/springframework/graphql/web/TestWebSocketClient.java @@ -14,7 +14,7 @@ * limitations under the License. */ -package org.springframework.graphql.client; +package org.springframework.graphql.web; import java.net.URI; import java.util.List; @@ -37,7 +37,7 @@ import org.springframework.web.reactive.socket.client.WebSocketClient; * * @author Rossen Stoyanchev */ -final class TestWebSocketClient implements WebSocketClient { +public final class TestWebSocketClient implements WebSocketClient { private final WebSocketHandler serverHandler; diff --git a/spring-graphql/src/test/java/org/springframework/graphql/client/TestWebSocketConnection.java b/spring-graphql/src/testFixtures/java/org/springframework/graphql/web/TestWebSocketConnection.java similarity index 92% rename from spring-graphql/src/test/java/org/springframework/graphql/client/TestWebSocketConnection.java rename to spring-graphql/src/testFixtures/java/org/springframework/graphql/web/TestWebSocketConnection.java index a77c7343..4ca54749 100644 --- a/spring-graphql/src/test/java/org/springframework/graphql/client/TestWebSocketConnection.java +++ b/spring-graphql/src/testFixtures/java/org/springframework/graphql/web/TestWebSocketConnection.java @@ -14,7 +14,7 @@ * limitations under the License. */ -package org.springframework.graphql.client; +package org.springframework.graphql.web; import java.net.URI; import java.util.ArrayList; @@ -23,6 +23,8 @@ import java.util.Queue; import java.util.concurrent.ConcurrentLinkedQueue; import java.util.concurrent.atomic.AtomicLong; +import org.apache.commons.logging.Log; +import org.apache.commons.logging.LogFactory; import org.reactivestreams.Publisher; import reactor.core.Scannable; import reactor.core.publisher.Flux; @@ -49,7 +51,9 @@ import org.springframework.web.reactive.socket.adapter.AbstractWebSocketSession; * * @author Rossen Stoyanchev */ -final class TestWebSocketConnection { +public final class TestWebSocketConnection { + + private static Log logger = LogFactory.getLog(TestWebSocketConnection.class); private static final AtomicLong connectionIndex = new AtomicLong(); @@ -125,8 +129,8 @@ final class TestWebSocketConnection { */ Mono connect(WebSocketHandler clientHandler, WebSocketHandler serverHandler) { - Mono serverMono = invokeHandler(serverHandler, this.serverSession); - Mono clientMono = invokeHandler(clientHandler, this.clientSession); + Mono serverMono = invokeHandler(serverHandler, this.serverSession, false); + Mono clientMono = invokeHandler(clientHandler, this.clientSession, true); Mono serverStatusMono = this.serverSession.closeStatus().then(); Mono clientStatusMono = this.clientSession.closeStatus().then(); @@ -137,10 +141,13 @@ final class TestWebSocketConnection { /** * Handle the session and complete it when handling completes. */ - private Mono invokeHandler(WebSocketHandler serverHandler, TestWebSocketSession session) { - return serverHandler.handle(session) + private Mono invokeHandler(WebSocketHandler handler, TestWebSocketSession session, boolean isClient) { + return handler.handle(session) .then(Mono.defer(() -> session.close(CloseStatus.NORMAL))) - .onErrorResume(ex -> session.close(CloseStatus.PROTOCOL_ERROR).then(Mono.error(ex))); + .onErrorResume(ex -> { + logger.error("Unhandled " + (isClient ? "client" : "server") + " error: " + ex.getMessage()); + return session.close(CloseStatus.PROTOCOL_ERROR).then(Mono.error(ex)); + }); }