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..0dc645d4 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,12 @@ final class DefaultRSocketGraphQlClientBuilder private final RSocketRequester.Builder requesterBuilder; + @Nullable + private Publisher> targetPublisher; + + @Nullable + private LoadbalanceStrategy loadbalanceStrategy; + @Nullable private ClientTransport clientTransport; @@ -87,8 +97,17 @@ final class DefaultRSocketGraphQlClientBuilder } @Override - public DefaultRSocketGraphQlClientBuilder clientTransport(ClientTransport clientTransport) { - this.clientTransport = clientTransport; + public DefaultRSocketGraphQlClientBuilder clientTransport(ClientTransport transport) { + this.clientTransport = transport; + return this; + } + + @Override + public DefaultRSocketGraphQlClientBuilder clientTransports( + Publisher> publisher, LoadbalanceStrategy strategy) { + + this.targetPublisher = publisher; + this.loadbalanceStrategy = strategy; return this; } @@ -106,8 +125,8 @@ final class DefaultRSocketGraphQlClientBuilder } @Override - public DefaultRSocketGraphQlClientBuilder rsocketRequester(Consumer requesterConsumer) { - requesterConsumer.accept(this.requesterBuilder); + public DefaultRSocketGraphQlClientBuilder rsocketRequester(Consumer consumer) { + consumer.accept(this.requesterBuilder); return this; } @@ -120,13 +139,25 @@ 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); - RSocketGraphQlTransport graphQlTransport = new RSocketGraphQlTransport(this.route, requester, getJsonDecoder()); + RSocketRequester requester; + + if (this.clientTransport != null) { + requester = this.requesterBuilder.transport(this.clientTransport); + } + else if (this.targetPublisher != null && this.loadbalanceStrategy != null) { + requester = this.requesterBuilder.transports(this.targetPublisher, this.loadbalanceStrategy); + } + else { + throw new IllegalStateException("Neither ClientTransport, nor Loadbalance targets and strategy"); + } + + 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()); } @@ -139,21 +170,33 @@ final class DefaultRSocketGraphQlClientBuilder private final RSocketRequester.Builder requesterBuilder; + @Nullable private final ClientTransport clientTransport; + @Nullable + private final Publisher> targetPublisher; + + @Nullable + 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) { + GraphQlClient graphQlClient, + RSocketRequester requester, RSocketRequester.Builder requesterBuilder, + @Nullable ClientTransport clientTransport, + @Nullable Publisher> targetPublisher, @Nullable LoadbalanceStrategy strategy, + String route, Consumer> builderInitializer) { super(graphQlClient); this.requester = requester; this.requesterBuilder = requesterBuilder; this.clientTransport = clientTransport; + this.targetPublisher = targetPublisher; + this.loadbalanceStrategy = strategy; this.route = route; this.builderInitializer = builderInitializer; } @@ -173,7 +216,12 @@ final class DefaultRSocketGraphQlClientBuilder @Override public RSocketGraphQlClient.Builder mutate() { DefaultRSocketGraphQlClientBuilder builder = new DefaultRSocketGraphQlClientBuilder(this.requesterBuilder); - builder.clientTransport(this.clientTransport); + if (this.clientTransport != null) { + builder.clientTransport(this.clientTransport); + } + if (this.targetPublisher != null && this.loadbalanceStrategy != null) { + builder.clientTransports(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..133975d8 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; @@ -78,7 +82,9 @@ public interface RSocketGraphQlClient extends GraphQlClient { interface Builder> extends GraphQlClient.Builder { /** - * Select TCP as the underlying network protocol. + * Select TCP as the underlying network protocol. This delegates to + * {@link RSocketRequester.Builder#tcp(String, int)} to create the + * {@code RSocketRequester} instance. * @param host the remote host to connect to * @param port the remote port to connect to * @return the same builder instance @@ -86,19 +92,38 @@ public interface RSocketGraphQlClient extends GraphQlClient { B tcp(String host, int port); /** - * Select WebSocket as the underlying network protocol. + * Select WebSocket as the underlying network protocol. This delegates to + * {@link RSocketRequester.Builder#websocket(URI)} to create the + * {@code RSocketRequester} instance. * @param uri the URL for the WebSocket handshake * @return the same builder instance */ B webSocket(URI uri); /** - * Use a given {@link ClientTransport} to communicate with the remote server. + * Use a given {@link ClientTransport} to communicate with the remote + * server. This delegates to + * {@link RSocketRequester.Builder#transport(ClientTransport)} to create + * the {@code RSocketRequester} instance. * @param clientTransport the transport to use * @return the same builder instance */ B clientTransport(ClientTransport clientTransport); + /** + * Use a {@link Publisher} of {@link LoadbalanceTarget}s, each of which + * contains a {@link ClientTransport}. This delegates to + * {@link RSocketRequester.Builder#transports(Publisher, LoadbalanceStrategy)} + * to create the {@code RSocketRequester} instance. + * @param targetPublisher supplies list of targets to loadbalance against; + * the targets are replaced when the given {@code Publisher} emits again. + * @param loadbalanceStrategy the strategy to use for selecting from + * the list of targets. + * @return the same builder instance + * @since 1.0.3 + */ + B clientTransports(Publisher> targetPublisher, LoadbalanceStrategy loadbalanceStrategy); + /** * Customize the format of data payloads for the connection. *

By default, this is set to {@code "application/graphql+json"} but