From 0faa63beeaef0a722df28c0aeab57a49d30dd031 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Iv=C3=A1n=20Rodr=C3=ADguez=20Murillo?= Date: Mon, 10 Oct 2022 17:07:04 +0100 Subject: [PATCH] Allow use of loadbalanced RSocketRequester See gh-498 --- .../DefaultRSocketGraphQlClientBuilder.java | 38 +++++++++++++++++-- .../graphql/client/RSocketGraphQlClient.java | 18 +++++++++ 2 files changed, 52 insertions(+), 4 deletions(-) diff --git a/spring-graphql/src/main/java/org/springframework/graphql/client/DefaultRSocketGraphQlClientBuilder.java b/spring-graphql/src/main/java/org/springframework/graphql/client/DefaultRSocketGraphQlClientBuilder.java index ee652bbe..1d4933e4 100644 --- a/spring-graphql/src/main/java/org/springframework/graphql/client/DefaultRSocketGraphQlClientBuilder.java +++ b/spring-graphql/src/main/java/org/springframework/graphql/client/DefaultRSocketGraphQlClientBuilder.java @@ -17,11 +17,15 @@ package org.springframework.graphql.client; import java.net.URI; +import java.util.List; import java.util.function.Consumer; +import io.rsocket.loadbalance.LoadbalanceStrategy; +import io.rsocket.loadbalance.LoadbalanceTarget; import io.rsocket.transport.ClientTransport; import io.rsocket.transport.netty.client.TcpClientTransport; import io.rsocket.transport.netty.client.WebsocketClientTransport; +import org.reactivestreams.Publisher; import reactor.core.publisher.Mono; import org.springframework.lang.Nullable; @@ -45,6 +49,10 @@ final class DefaultRSocketGraphQlClientBuilder private final RSocketRequester.Builder requesterBuilder; + private Publisher> targetPublisher; + + private LoadbalanceStrategy loadbalanceStrategy; + @Nullable private ClientTransport clientTransport; @@ -111,6 +119,13 @@ final class DefaultRSocketGraphQlClientBuilder return this; } + @Override + public DefaultRSocketGraphQlClientBuilder transports(Publisher> targetPublisher, LoadbalanceStrategy loadbalanceStrategy) { + this.targetPublisher = targetPublisher; + this.loadbalanceStrategy = loadbalanceStrategy; + return this; + } + @Override public RSocketGraphQlClient build() { @@ -120,13 +135,20 @@ final class DefaultRSocketGraphQlClientBuilder 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); + RSocketRequester requester; + + if (this.targetPublisher != null && this.loadbalanceStrategy != null) { + requester = this.requesterBuilder.transports(this.targetPublisher, this.loadbalanceStrategy); + } else { + Assert.state(this.clientTransport != null, "Neither WebSocket nor TCP networking configured"); + requester = this.requesterBuilder.transport(this.clientTransport); + } RSocketGraphQlTransport graphQlTransport = new RSocketGraphQlTransport(this.route, requester, getJsonDecoder()); return new DefaultRSocketGraphQlClient( super.buildGraphQlClient(graphQlTransport), requester, - this.requesterBuilder, this.clientTransport, this.route, getBuilderInitializer()); + this.requesterBuilder, this.clientTransport, this.targetPublisher, this.loadbalanceStrategy, + this.route, getBuilderInitializer()); } @@ -141,19 +163,26 @@ final class DefaultRSocketGraphQlClientBuilder private final ClientTransport clientTransport; + private final Publisher> targetPublisher; + + private final LoadbalanceStrategy loadbalanceStrategy; + private final String route; private final Consumer> builderInitializer; DefaultRSocketGraphQlClient( GraphQlClient graphQlClient, RSocketRequester requester, RSocketRequester.Builder requesterBuilder, - ClientTransport clientTransport, String route, Consumer> builderInitializer) { + ClientTransport clientTransport, Publisher> targetPublisher, LoadbalanceStrategy loadbalanceStrategy, + String route, Consumer> builderInitializer) { super(graphQlClient); this.requester = requester; this.requesterBuilder = requesterBuilder; this.clientTransport = clientTransport; + this.targetPublisher = targetPublisher; + this.loadbalanceStrategy = loadbalanceStrategy; this.route = route; this.builderInitializer = builderInitializer; } @@ -174,6 +203,7 @@ final class DefaultRSocketGraphQlClientBuilder public RSocketGraphQlClient.Builder mutate() { DefaultRSocketGraphQlClientBuilder builder = new DefaultRSocketGraphQlClientBuilder(this.requesterBuilder); builder.clientTransport(this.clientTransport); + builder.transports(this.targetPublisher, this.loadbalanceStrategy); builder.route(this.route); this.builderInitializer.accept(builder); return builder; diff --git a/spring-graphql/src/main/java/org/springframework/graphql/client/RSocketGraphQlClient.java b/spring-graphql/src/main/java/org/springframework/graphql/client/RSocketGraphQlClient.java index 448a57af..6c15c0b7 100644 --- a/spring-graphql/src/main/java/org/springframework/graphql/client/RSocketGraphQlClient.java +++ b/spring-graphql/src/main/java/org/springframework/graphql/client/RSocketGraphQlClient.java @@ -17,10 +17,14 @@ package org.springframework.graphql.client; import java.net.URI; +import java.util.List; import java.util.function.Consumer; import io.rsocket.core.RSocketClient; +import io.rsocket.loadbalance.LoadbalanceStrategy; +import io.rsocket.loadbalance.LoadbalanceTarget; import io.rsocket.transport.ClientTransport; +import org.reactivestreams.Publisher; import reactor.core.publisher.Mono; import org.springframework.messaging.rsocket.RSocketRequester; @@ -129,6 +133,20 @@ public interface RSocketGraphQlClient extends GraphQlClient { */ B rsocketRequester(Consumer requester); + /** + * Build an {@link RSocketRequester} with an + * {@link io.rsocket.loadbalance.LoadbalanceRSocketClient} that will + * connect to one of the given targets selected through the given + * {@link io.rsocket.loadbalance.LoadbalanceRSocketClient}. + * @param targetPublisher a {@code Publisher} that supplies a list of + * target transports to loadbalance against; the given list may be + * periodically updated by the {@code Publisher}. + * @param loadbalanceStrategy the strategy to use for selecting from + * the list of loadbalance targets. + * @return the same builder instance + */ + B transports(Publisher> targetPublisher, LoadbalanceStrategy loadbalanceStrategy); + /** * Build the {@code RSocketGraphQlClient} instance. */