Merge branch '1.0.x'

Author: rstoyanchev

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<List<LoadbalanceTarget>> targetPublisher;
+
+	@Nullable
+	private LoadbalanceStrategy loadbalanceStrategy;
+
 	@Nullable
 	private ClientTransport clientTransport;
 
@@ -87,8 +97,17 @@ public DefaultRSocketGraphQlClientBuilder webSocket(URI uri) {
 	}
 
 	@Override
-	public DefaultRSocketGraphQlClientBuilder clientTransport(ClientTransport clientTransport) {
-		this.clientTransport = clientTransport;
+	public DefaultRSocketGraphQlClientBuilder clientTransport(ClientTransport transport) {
+		this.clientTransport = transport;
+		return this;
+	}
+
+	@Override
+	public DefaultRSocketGraphQlClientBuilder clientTransports(
+			Publisher<List<LoadbalanceTarget>> publisher, LoadbalanceStrategy strategy) {
+
+		this.targetPublisher = publisher;
+		this.loadbalanceStrategy = strategy;
 		return this;
 	}
 
@@ -106,8 +125,8 @@ public DefaultRSocketGraphQlClientBuilder route(String route) {
 	}
 
 	@Override
-	public DefaultRSocketGraphQlClientBuilder rsocketRequester(Consumer<RSocketRequester.Builder> requesterConsumer) {
-		requesterConsumer.accept(this.requesterBuilder);
+	public DefaultRSocketGraphQlClientBuilder rsocketRequester(Consumer<RSocketRequester.Builder> consumer) {
+		consumer.accept(this.requesterBuilder);
 		return this;
 	}
 
@@ -120,13 +139,25 @@ public RSocketGraphQlClient build() {
 			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 @@ private static class DefaultRSocketGraphQlClient extends AbstractDelegatingGraph
 
 		private final RSocketRequester.Builder requesterBuilder;
 
+		@Nullable
 		private final ClientTransport clientTransport;
 
+		@Nullable
+		private final Publisher<List<LoadbalanceTarget>> targetPublisher;
+
+		@Nullable
+		private final LoadbalanceStrategy loadbalanceStrategy;
+
 		private final String route;
 
 		private final Consumer<AbstractGraphQlClientBuilder<?>> builderInitializer;
 
 		DefaultRSocketGraphQlClient(
-				GraphQlClient graphQlClient, RSocketRequester requester, RSocketRequester.Builder requesterBuilder,
-				ClientTransport clientTransport, String route, Consumer<AbstractGraphQlClientBuilder<?>> builderInitializer) {
+				GraphQlClient graphQlClient,
+				RSocketRequester requester, RSocketRequester.Builder requesterBuilder,
+				@Nullable ClientTransport clientTransport,
+				@Nullable Publisher<List<LoadbalanceTarget>> targetPublisher, @Nullable LoadbalanceStrategy strategy,
+				String route, Consumer<AbstractGraphQlClientBuilder<?>> 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 @@ public Mono<Void> stop() {
 		@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;

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,27 +82,48 @@ static Builder<?> builder(RSocketRequester.Builder requesterBuilder) {
 	interface Builder<B extends Builder<B>> extends GraphQlClient.Builder<B> {
 
 		/**
-		 * 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
 		 */
 		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<List<LoadbalanceTarget>> targetPublisher, LoadbalanceStrategy loadbalanceStrategy);
+
 		/**
 		 * Customize the format of data payloads for the connection.
 		 * <p>By default, this is set to {@code "application/graphql+json"} but