Reactor2TcpClient constructor with address supplier

Issue: SPR-12452

Author: rstoyanchev

spring-messaging/src/main/java/org/springframework/messaging/simp/config/StompBrokerRelayRegistration.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2013 the original author or authors.
+ * Copyright 2002-2018 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.
@@ -19,6 +19,8 @@
 import org.springframework.messaging.MessageChannel;
 import org.springframework.messaging.SubscribableChannel;
 import org.springframework.messaging.simp.stomp.StompBrokerRelayMessageHandler;
+import org.springframework.messaging.tcp.TcpOperations;
+import org.springframework.messaging.tcp.reactor.Reactor2TcpClient;
 import org.springframework.util.Assert;
 
 /**
@@ -47,6 +49,8 @@ public class StompBrokerRelayRegistration extends AbstractBrokerRegistration {
 
 	private String virtualHost;
 
+	private TcpOperations<byte[]> tcpClient;
+
 	private boolean autoStartup = true;
 
 	private String userDestinationBroadcast;
@@ -160,6 +164,18 @@ public StompBrokerRelayRegistration setVirtualHost(String virtualHost) {
 		return this;
 	}
 
+	/**
+	 * Configure a TCP client for managing TCP connections to the STOMP broker.
+	 * By default {@link Reactor2TcpClient} is used.
+	 * <p><strong>Note:</strong> when this property is used, any
+	 * {@link #setRelayHost(String) host} or {@link #setRelayPort(int) port}
+	 * specified are effectively ignored.
+	 * @since 4.3.15
+	 */
+	public void setTcpClient(TcpOperations<byte[]> tcpClient) {
+		this.tcpClient = tcpClient;
+	}
+
 	/**
 	 * Configure whether the {@link StompBrokerRelayMessageHandler} should start
 	 * automatically when the Spring ApplicationContext is refreshed.
@@ -231,6 +247,9 @@ protected StompBrokerRelayMessageHandler getMessageHandler(SubscribableChannel b
 		if (this.virtualHost != null) {
 			handler.setVirtualHost(this.virtualHost);
 		}
+		if (this.tcpClient != null) {
+			handler.setTcpClient(this.tcpClient);
+		}
 
 		handler.setAutoStartup(this.autoStartup);
 

spring-messaging/src/main/java/org/springframework/messaging/simp/stomp/StompBrokerRelayMessageHandler.java

@@ -334,6 +334,9 @@ public String getVirtualHost() {
 	/**
 	 * Configure a TCP client for managing TCP connections to the STOMP broker.
 	 * By default {@link Reactor2TcpClient} is used.
+	 * <p><strong>Note:</strong> when this property is used, any
+	 * {@link #setRelayHost(String) host} or {@link #setRelayPort(int) port}
+	 * specified are effectively ignored.
 	 */
 	public void setTcpClient(TcpOperations<byte[]> tcpClient) {
 		this.tcpClient = tcpClient;

spring-messaging/src/main/java/org/springframework/messaging/tcp/reactor/Reactor2TcpClient.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2016 the original author or authors.
+ * Copyright 2002-2018 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.
@@ -35,6 +35,7 @@
 import reactor.core.support.NamedDaemonThreadFactory;
 import reactor.fn.Consumer;
 import reactor.fn.Function;
+import reactor.fn.Supplier;
 import reactor.fn.tuple.Tuple;
 import reactor.fn.tuple.Tuple2;
 import reactor.io.buffer.Buffer;
@@ -65,10 +66,10 @@
 
 /**
  * An implementation of {@link org.springframework.messaging.tcp.TcpOperations}
- * based on the TCP client support of the Reactor project.
+ * based on the TCP client support of project Reactor.
  *
- * <p>This implementation wraps N (Reactor) clients for N {@link #connect} calls,
- * i.e. a separate (Reactor) client instance for each connection.
+ * <p>This implementation wraps N Reactor {@code TcpClient} instances created
+ * for N {@link #connect} calls, i.e. once instance per connection.
  *
  * @author Rossen Stoyanchev
  * @author Stephane Maldini
@@ -100,13 +101,28 @@ public class Reactor2TcpClient<P> implements TcpOperations<P> {
 	 * relying on Netty threads. The number of Netty threads can be tweaked with
 	 * the {@code reactor.tcp.ioThreadCount} System property. The network I/O
 	 * threads will be shared amongst the active clients.
-	 * <p>Also see the constructor accepting a ready Reactor
-	 * {@link TcpClientSpec} {@link Function} factory.
+	 *
 	 * @param host the host to connect to
 	 * @param port the port to connect to
 	 * @param codec the codec to use for encoding and decoding the TCP stream
 	 */
 	public Reactor2TcpClient(final String host, final int port, final Codec<Buffer, Message<P>, Message<P>> codec) {
+		this(new FixedAddressSupplier(host, port), codec);
+	}
+
+	/**
+	 * A variant of {@link #Reactor2TcpClient(String, int, Codec)} that takes a
+	 * supplier of any number of addresses instead of just one host and port.
+	 * This can be used to {@link #connect(TcpConnectionHandler, ReconnectStrategy)
+	 * reconnect} to a different address after the current host becomes unavailable.
+	 *
+	 * @param addressSupplier supplier of addresses to use for connecting
+	 * @param codec the codec to use for encoding and decoding the TCP stream
+	 * @since 4.3.15
+	 */
+	public Reactor2TcpClient(final Supplier<InetSocketAddress> addressSupplier,
+			final Codec<Buffer, Message<P>, Message<P>> codec) {
+
 		// Reactor 2.0.5 requires NioEventLoopGroup vs 2.0.6+ requires EventLoopGroup
 		final NioEventLoopGroup nioEventLoopGroup = initEventLoopGroup();
 		this.eventLoopGroup = nioEventLoopGroup;
@@ -118,7 +134,7 @@ public TcpClientSpec<Message<P>, Message<P>> apply(TcpClientSpec<Message<P>, Mes
 				return spec
 						.env(environment)
 						.codec(codec)
-						.connect(host, port)
+						.connect(addressSupplier)
 						.options(createClientSocketOptions());
 			}
 
@@ -133,10 +149,13 @@ private ClientSocketOptions createClientSocketOptions() {
 	 * A constructor with a pre-configured {@link TcpClientSpec} {@link Function}
 	 * factory. This might be used to add SSL or specific network parameters to
 	 * the generated client configuration.
+	 *
 	 * <p><strong>NOTE:</strong> if the client is configured with a thread-creating
-	 * dispatcher, you are responsible for cleaning them, e.g. using
+	 * dispatcher, you are responsible for cleaning them, e.g. via
 	 * {@link reactor.core.Dispatcher#shutdown}.
-	 * @param tcpClientSpecFactory the TcpClientSpec {@link Function} to use for each client creation
+	 *
+	 * @param tcpClientSpecFactory the TcpClientSpec {@link Function} to use for
+	 * each client creation
 	 */
 	public Reactor2TcpClient(TcpClientFactory<Message<P>, Message<P>> tcpClientSpecFactory) {
 		Assert.notNull(tcpClientSpecFactory, "'tcpClientClientFactory' must not be null");
@@ -295,6 +314,21 @@ private static Method initEventLoopGroupMethod() {
 	}
 
 
+	private static class FixedAddressSupplier implements Supplier<InetSocketAddress> {
+
+		private final InetSocketAddress address;
+
+		FixedAddressSupplier(String host, int port) {
+			this.address = new InetSocketAddress(host, port);
+		}
+
+		@Override
+		public InetSocketAddress get() {
+			return this.address;
+		}
+	}
+
+
 	private static class SynchronousDispatcherConfigReader implements ConfigurationReader {
 
 		@Override

src/asciidoc/web-websocket.adoc

@@ -1615,8 +1615,8 @@ values ``guest``/``guest``.
 ====
 The STOMP broker relay always sets the `login` and `passcode` headers on every `CONNECT`
 frame that it forwards to the broker on behalf of clients. Therefore WebSocket clients
-need not set those headers; they will be ignored. As the following section explains,
-instead WebSocket clients should rely on HTTP authentication to protect the WebSocket
+need not set those headers; they will be ignored. As the <<websocket-stomp-authentication>>
+explains, instead WebSocket clients should rely on HTTP authentication to protect the WebSocket
 endpoint and establish the client identity.
 ====
 
@@ -1626,13 +1626,47 @@ and receiving heartbeats (10 seconds each by default). If connectivity to the br
 is lost, the broker relay will continue to try to reconnect, every 5 seconds,
 until it succeeds.
 
-[NOTE]
-====
-A Spring bean can implement `ApplicationListener<BrokerAvailabilityEvent>` in order
+Any Spring bean can implement `ApplicationListener<BrokerAvailabilityEvent>` in order
 to receive notifications when the "system" connection to the broker is lost and
 re-established. For example a Stock Quote service broadcasting stock quotes can
 stop trying to send messages when there is no active "system" connection.
-====
+
+By default, the STOMP broker relay always connects, and reconnects as needed if
+connectivity is lost, to the same host and port. If you wish to supply multiple addresses,
+on each attempt to connect, you can configure a supplier of addresses, instead of a
+fixed host and port. For example:
+
+[source,java,indent=0]
+[subs="verbatim,quotes"]
+----
+@Configuration
+@EnableWebSocketMessageBroker
+public class WebSocketConfig extends AbstractWebSocketMessageBrokerConfigurer {
+
+	// ...
+
+	@Override
+	public void configureMessageBroker(MessageBrokerRegistry registry) {
+		registry.enableStompBrokerRelay("/queue/", "/topic/").setTcpClient(createTcpClient());
+		registry.setApplicationDestinationPrefixes("/app");
+	}
+
+	private Reactor2TcpClient<byte[]> createTcpClient() {
+
+		Supplier<InetSocketAddress> addressSupplier = new Supplier<InetSocketAddress>() {
+			@Override
+			public InetSocketAddress get() {
+				// Select address to connect to ...
+			}
+		};
+
+		StompDecoder decoder = new StompDecoder();
+		Reactor2StompCodec codec = new Reactor2StompCodec(new StompEncoder(), decoder);
+		return new Reactor2TcpClient<>(addressSupplier, codec);
+	}
+
+}
+----
 
 The STOMP broker relay can also be configured with a `virtualHost` property.
 The value of this property will be set as the `host` header of every `CONNECT` frame