diff --git a/spring-graphql/src/main/java/org/springframework/graphql/client/DefaultWebSocketGraphQlClientBuilder.java b/spring-graphql/src/main/java/org/springframework/graphql/client/DefaultWebSocketGraphQlClientBuilder.java index 2d28960c..5bcb3a72 100644 --- a/spring-graphql/src/main/java/org/springframework/graphql/client/DefaultWebSocketGraphQlClientBuilder.java +++ b/spring-graphql/src/main/java/org/springframework/graphql/client/DefaultWebSocketGraphQlClientBuilder.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2023 the original author or authors. + * Copyright 2002-2024 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. @@ -17,6 +17,7 @@ package org.springframework.graphql.client; import java.net.URI; +import java.time.Duration; import java.util.Arrays; import java.util.List; import java.util.function.Consumer; @@ -26,6 +27,7 @@ import reactor.core.publisher.Mono; import org.springframework.http.HttpHeaders; import org.springframework.http.codec.ClientCodecConfigurer; import org.springframework.http.codec.CodecConfigurer; +import org.springframework.lang.Nullable; import org.springframework.util.Assert; import org.springframework.web.reactive.socket.client.WebSocketClient; import org.springframework.web.util.DefaultUriBuilderFactory; @@ -49,7 +51,8 @@ final class DefaultWebSocketGraphQlClientBuilder private final CodecConfigurer codecConfigurer; - private long keepalive; + @Nullable + private Duration keepAlive; /** * Constructor to start via {@link WebSocketGraphQlClient#builder(String, WebSocketClient)}. @@ -58,28 +61,13 @@ final class DefaultWebSocketGraphQlClientBuilder this(toURI(url), client); } - /** - * Constructor to start via {@link WebSocketGraphQlClient#builder(String, WebSocketClient, long)}. - */ - DefaultWebSocketGraphQlClientBuilder(String url, WebSocketClient client, long keepalive) { - this(toURI(url), client, keepalive); - } - /** * Constructor to start via {@link WebSocketGraphQlClient#builder(URI, WebSocketClient)}. */ DefaultWebSocketGraphQlClientBuilder(URI url, WebSocketClient client) { - this(url, client, 0); - } - - /** - * Constructor to start via {@link WebSocketGraphQlClient#builder(URI, WebSocketClient, long)}. - */ - DefaultWebSocketGraphQlClientBuilder(URI url, WebSocketClient client, long keepalive) { this.url = url; this.webSocketClient = client; this.codecConfigurer = ClientCodecConfigurer.create(); - this.keepalive = keepalive; } /** @@ -91,7 +79,7 @@ final class DefaultWebSocketGraphQlClientBuilder this.headers.putAll(transport.getHeaders()); this.webSocketClient = transport.getWebSocketClient(); this.codecConfigurer = transport.getCodecConfigurer(); - this.keepalive = transport.getKeepAlive(); + this.keepAlive = transport.getKeepAlive(); } @@ -128,6 +116,12 @@ final class DefaultWebSocketGraphQlClientBuilder return this; } + @Override + public WebSocketGraphQlClient.Builder keepAlive(Duration keepalive) { + this.keepAlive = keepalive; + return this; + } + @Override public WebSocketGraphQlClient build() { @@ -136,18 +130,12 @@ final class DefaultWebSocketGraphQlClientBuilder CodecDelegate.findJsonDecoder(this.codecConfigurer)); WebSocketGraphQlTransport transport = new WebSocketGraphQlTransport( - this.url, this.headers, this.webSocketClient, this.codecConfigurer, getInterceptor(), this.keepalive); + this.url, this.headers, this.webSocketClient, this.codecConfigurer, getInterceptor(), this.keepAlive); GraphQlClient graphQlClient = super.buildGraphQlClient(transport); return new DefaultWebSocketGraphQlClient(graphQlClient, transport, getBuilderInitializer()); } - @Override - public WebSocketGraphQlClient.Builder keepalive(long keepalive) { - this.keepalive = keepalive; - return this; - } - private WebSocketGraphQlClientInterceptor getInterceptor() { List interceptors = getInterceptors().stream() diff --git a/spring-graphql/src/main/java/org/springframework/graphql/client/WebSocketGraphQlClient.java b/spring-graphql/src/main/java/org/springframework/graphql/client/WebSocketGraphQlClient.java index 10b8f427..b28178dd 100644 --- a/spring-graphql/src/main/java/org/springframework/graphql/client/WebSocketGraphQlClient.java +++ b/spring-graphql/src/main/java/org/springframework/graphql/client/WebSocketGraphQlClient.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2022 the original author or authors. + * Copyright 2002-2024 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. @@ -17,6 +17,7 @@ package org.springframework.graphql.client; import java.net.URI; +import java.time.Duration; import reactor.core.publisher.Mono; @@ -64,16 +65,6 @@ public interface WebSocketGraphQlClient extends WebGraphQlClient { return builder(url, webSocketClient).build(); } - /** - * Create a {@link WebSocketGraphQlClient}. - * @param url the GraphQL endpoint URL - * @param webSocketClient the underlying transport client to use - * @param keepalive the delay in seconds between sending ping messages, or 0 to disable - */ - static WebSocketGraphQlClient create(URI url, WebSocketClient webSocketClient, long keepalive) { - return builder(url, webSocketClient).keepalive(keepalive).build(); - } - /** * Return a builder for a {@link WebSocketGraphQlClient}. * @param url the GraphQL endpoint URL @@ -83,16 +74,6 @@ public interface WebSocketGraphQlClient extends WebGraphQlClient { return new DefaultWebSocketGraphQlClientBuilder(url, webSocketClient); } - /** - * Return a builder for a {@link WebSocketGraphQlClient}. - * @param url the GraphQL endpoint URL - * @param webSocketClient the underlying transport client to use - * @param keepalive the delay in seconds between sending ping messages, or 0 to disable - */ - static Builder builder(String url, WebSocketClient webSocketClient, long keepalive) { - return new DefaultWebSocketGraphQlClientBuilder(url, webSocketClient, keepalive); - } - /** * Return a builder for a {@link WebSocketGraphQlClient}. * @param url the GraphQL endpoint URL @@ -102,16 +83,6 @@ public interface WebSocketGraphQlClient extends WebGraphQlClient { return new DefaultWebSocketGraphQlClientBuilder(url, webSocketClient); } - /** - * Return a builder for a {@link WebSocketGraphQlClient}. - * @param url the GraphQL endpoint URL - * @param webSocketClient the underlying transport client to use - * @param keepalive the delay in seconds between sending ping messages, or 0 to disable - */ - static Builder builder(URI url, WebSocketClient webSocketClient, long keepalive) { - return new DefaultWebSocketGraphQlClientBuilder(url, webSocketClient, keepalive); - } - /** * Builder for a GraphQL over WebSocket client. @@ -119,14 +90,20 @@ public interface WebSocketGraphQlClient extends WebGraphQlClient { */ interface Builder> extends WebGraphQlClient.Builder { + /** + * Configure how frequently to send ping messages. + *

By default, this is not set, and ping messages are not sent. + * @param keepAlive the value to use + * @since 1.3 + */ + Builder keepAlive(Duration keepAlive); + /** * Build the {@code WebSocketGraphQlClient}. */ @Override WebSocketGraphQlClient build(); - Builder keepalive(long keepalive); - } } diff --git a/spring-graphql/src/main/java/org/springframework/graphql/client/WebSocketGraphQlTransport.java b/spring-graphql/src/main/java/org/springframework/graphql/client/WebSocketGraphQlTransport.java index 604c7503..54d8d0fe 100644 --- a/spring-graphql/src/main/java/org/springframework/graphql/client/WebSocketGraphQlTransport.java +++ b/spring-graphql/src/main/java/org/springframework/graphql/client/WebSocketGraphQlTransport.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2023 the original author or authors. + * Copyright 2002-2024 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. @@ -68,12 +68,13 @@ final class WebSocketGraphQlTransport implements GraphQlTransport { private final Mono graphQlSessionMono; - private final long keepalive; + @Nullable + private final Duration keepAlive; WebSocketGraphQlTransport( URI url, @Nullable HttpHeaders headers, WebSocketClient client, CodecConfigurer codecConfigurer, - WebSocketGraphQlClientInterceptor interceptor, long keepalive) { + WebSocketGraphQlClientInterceptor interceptor, @Nullable Duration keepAlive) { Assert.notNull(url, "URI is required"); Assert.notNull(client, "WebSocketClient is required"); @@ -83,9 +84,9 @@ final class WebSocketGraphQlTransport implements GraphQlTransport { this.url = url; this.headers.putAll((headers != null) ? headers : HttpHeaders.EMPTY); this.webSocketClient = client; - this.keepalive = keepalive; + this.keepAlive = keepAlive; - this.graphQlSessionHandler = new GraphQlSessionHandler(codecConfigurer, interceptor, keepalive); + this.graphQlSessionHandler = new GraphQlSessionHandler(codecConfigurer, interceptor, keepAlive); this.graphQlSessionMono = initGraphQlSession(this.url, this.headers, client, this.graphQlSessionHandler) .cacheInvalidateWhen(GraphQlSession::notifyWhenClosed); @@ -166,8 +167,9 @@ final class WebSocketGraphQlTransport implements GraphQlTransport { return this.graphQlSessionMono.flatMapMany((session) -> session.executeSubscription(request)); } - public long getKeepAlive() { - return keepalive; + @Nullable + Duration getKeepAlive() { + return this.keepAlive; } @@ -191,15 +193,18 @@ final class WebSocketGraphQlTransport implements GraphQlTransport { private final AtomicBoolean stopped = new AtomicBoolean(); - private final long keepalive; + @Nullable + private final Duration keepAlive; - GraphQlSessionHandler(CodecConfigurer codecConfigurer, WebSocketGraphQlClientInterceptor interceptor, - long keepalive) { + GraphQlSessionHandler( + CodecConfigurer codecConfigurer, WebSocketGraphQlClientInterceptor interceptor, + @Nullable Duration keepAlive) { + this.codecDelegate = new CodecDelegate(codecConfigurer); this.interceptor = interceptor; this.graphQlSessionSink = Sinks.unsafe().one(); - this.keepalive = keepalive; + this.keepAlive = keepAlive; } @@ -257,7 +262,7 @@ final class WebSocketGraphQlTransport implements GraphQlTransport { session.send(connectionInitMono.concatWith(graphQlSession.getRequestFlux()) .map((message) -> this.codecDelegate.encode(session, message))); - Flux receiveCompletion = session.receive() + Mono receiveCompletion = session.receive() .flatMap((webSocketMessage) -> { if (sessionNotInitialized()) { try { @@ -303,20 +308,22 @@ final class WebSocketGraphQlTransport implements GraphQlTransport { } } return Mono.empty(); - }); + }) + .mergeWith((this.keepAlive != null) ? + Flux.interval(this.keepAlive, this.keepAlive) + .filter((aLong) -> graphQlSession.checkSentOrReceivedMessagesAndClear()) + .doOnNext((aLong) -> graphQlSession.sendPing()) + .then() : + Flux.empty()) + .then(); - if (keepalive > 0) { - Duration keepAliveDuration = Duration.ofSeconds(keepalive); - receiveCompletion = receiveCompletion - .mergeWith(Flux.interval(keepAliveDuration, keepAliveDuration) - .flatMap(i -> { - graphQlSession.sendPing(null); - return Mono.empty(); - }) - ); + if (this.keepAlive != null) { + Flux.interval(this.keepAlive, this.keepAlive) + .filter((aLong) -> graphQlSession.checkSentOrReceivedMessagesAndClear()) + .doOnNext((aLong) -> graphQlSession.sendPing()) + .subscribe(); } - return Mono.zip(sendCompletion, receiveCompletion.then()).then(); } @@ -413,6 +420,8 @@ final class WebSocketGraphQlTransport implements GraphQlTransport { private final Map requestStateMap = new ConcurrentHashMap<>(); + private boolean hasReceivedMessages; + GraphQlSession(WebSocketSession webSocketSession) { this.connection = DisposableConnection.from(webSocketSession); @@ -483,11 +492,16 @@ final class WebSocketGraphQlTransport implements GraphQlTransport { this.requestSink.sendRequest(message); } - public void sendPing(@Nullable Map payload) { - GraphQlWebSocketMessage message = GraphQlWebSocketMessage.ping(payload); + void sendPing() { + GraphQlWebSocketMessage message = GraphQlWebSocketMessage.ping(null); this.requestSink.sendRequest(message); } + boolean checkSentOrReceivedMessagesAndClear() { + boolean received = this.hasReceivedMessages; + this.hasReceivedMessages = false; + return (this.requestSink.checkSentMessagesAndClear() || received); + } // Inbound messages @@ -504,6 +518,8 @@ final class WebSocketGraphQlTransport implements GraphQlTransport { return; } + this.hasReceivedMessages = true; + if (requestState instanceof SingleResponseRequestState) { this.requestStateMap.remove(id); } @@ -631,6 +647,8 @@ final class WebSocketGraphQlTransport implements GraphQlTransport { @Nullable private FluxSink requestSink; + private boolean hasSentMessages; + private final Flux requestFlux = Flux.create((sink) -> { Assert.state(this.requestSink == null, "Expected single subscriber only for outbound messages"); this.requestSink = sink; @@ -642,9 +660,16 @@ final class WebSocketGraphQlTransport implements GraphQlTransport { void sendRequest(GraphQlWebSocketMessage message) { Assert.state(this.requestSink != null, "Unexpected request before Flux is subscribed to"); + this.hasSentMessages = true; this.requestSink.next(message); } + boolean checkSentMessagesAndClear() { + boolean result = this.hasSentMessages; + this.hasSentMessages = false; + return result; + } + } 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 17a6e827..d086285d 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 @@ -1,5 +1,5 @@ /* - * Copyright 2002-2023 the original author or authors. + * Copyright 2002-2024 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. diff --git a/spring-graphql/src/test/java/org/springframework/graphql/client/WebSocketGraphQlTransportTests.java b/spring-graphql/src/test/java/org/springframework/graphql/client/WebSocketGraphQlTransportTests.java index e18f599e..209a3f1a 100644 --- a/spring-graphql/src/test/java/org/springframework/graphql/client/WebSocketGraphQlTransportTests.java +++ b/spring-graphql/src/test/java/org/springframework/graphql/client/WebSocketGraphQlTransportTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2023 the original author or authors. + * Copyright 2002-2024 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. @@ -60,7 +60,6 @@ public class WebSocketGraphQlTransportTests { private static final Duration TIMEOUT = Duration.ofSeconds(5); private static final CodecDelegate CODEC_DELEGATE = new CodecDelegate(ClientCodecConfigurer.create()); - public static final int KEEPALIVE = 1; private final MockGraphQlWebSocketServer mockServer = new MockGraphQlWebSocketServer(); @@ -187,18 +186,27 @@ public class WebSocketGraphQlTransportTests { } @Test - void pingSending() throws InterruptedException { + void pingSending() { - GraphQlRequest request = this.mockServer.expectOperation("{Sub1}").andStream(Flux.just(this.response1, response2)); + GraphQlRequest request = this.mockServer.expectOperation("{Sub1}").andStream(Flux.empty()); - StepVerifier.create(this.transport.executeSubscription(request)) - .expectNext(this.response1, response2).expectComplete() + WebSocketGraphQlTransport transport = new WebSocketGraphQlTransport( + URI.create("/"), HttpHeaders.EMPTY, this.webSocketClient, ClientCodecConfigurer.create(), + new WebSocketGraphQlClientInterceptor() { }, Duration.ofMillis(10)); + + StepVerifier.create(transport.executeSubscription(request)) + .thenAwait(Duration.ofMillis(50)) + .thenCancel() .verify(TIMEOUT); - Thread.sleep(KEEPALIVE*1000 + 50); // wait for ping - assertActualClientMessages( + List messages = + this.webSocketClient.getConnection(0).getClientMessages().stream() + .map(CODEC_DELEGATE::decode).toList().subList(0, 4); + + assertThat(messages).containsExactly( GraphQlWebSocketMessage.connectionInit(null), GraphQlWebSocketMessage.subscribe("1", request), + GraphQlWebSocketMessage.ping(null), GraphQlWebSocketMessage.ping(null)); } @@ -227,7 +235,7 @@ public class WebSocketGraphQlTransportTests { WebSocketGraphQlTransport transport = new WebSocketGraphQlTransport( - URI.create("/"), HttpHeaders.EMPTY, client, ClientCodecConfigurer.create(), interceptor, KEEPALIVE); + URI.create("/"), HttpHeaders.EMPTY, client, ClientCodecConfigurer.create(), interceptor, null); transport.start().block(TIMEOUT); @@ -341,7 +349,7 @@ public class WebSocketGraphQlTransportTests { private static WebSocketGraphQlTransport createTransport(WebSocketClient client) { return new WebSocketGraphQlTransport( URI.create("/"), HttpHeaders.EMPTY, client, ClientCodecConfigurer.create(), - new WebSocketGraphQlClientInterceptor() { }, KEEPALIVE); + new WebSocketGraphQlClientInterceptor() { }, null); } private void assertActualClientMessages(GraphQlWebSocketMessage... expectedMessages) {