Reconnect failures delegated to TcpConnectionHandler

rstoyanchev · rstoyanchev · commit 698c885e0601 · 2016-12-28T21:37:07.000-05:00
When connecting with a ReconnectStrategy we can only report the outcome
of the first connect to the ListenableFuture&lt;Void&gt; return value.

Failures for all subsequent attempts to reconnect however must be
channeled to TcpConnectHandler#afterConnectFailure which is used in
the STOMP broker relay for example to publish
BroadcastAvailability(true/false) events.
diff --git a/spring-messaging/src/main/java/org/springframework/messaging/tcp/reactor/ReactorNettyTcpClient.java b/spring-messaging/src/main/java/org/springframework/messaging/tcp/reactor/ReactorNettyTcpClient.java
@@ -116,12 +116,15 @@ public ListenableFuture<Void> connect(TcpConnectionHandler<P> handler, Reconnect
 			return handleShuttingDownConnectFailure(handler);
 		}
 
+		// Report first connect to the ListenableFuture
 		MonoProcessor<Void> connectMono = MonoProcessor.create();
+
 		this.tcpClient
 				.newHandler(new ReactorNettyHandler(handler))
-				.doOnNext(connectFailureConsumer(connectMono))
-				.doOnError(connectFailureConsumer(connectMono))
-				.then(NettyContext::onClose)
+				.doOnNext(updateConnectMono(connectMono))
+				.doOnError(updateConnectMono(connectMono))
+				.doOnError(handler::afterConnectFailure)    // report all connect failures to the handler
+				.then(NettyContext::onClose)                // post-connect issues
 				.retryWhen(reconnectFunction(strategy))
 				.repeatWhen(reconnectFunction(strategy))
 				.subscribe();
@@ -135,7 +138,7 @@ private ListenableFuture<Void> handleShuttingDownConnectFailure(TcpConnectionHan
 		return new MonoToListenableFutureAdapter<>(Mono.error(ex));
 	}
 
-	private <T> Consumer<T> connectFailureConsumer(MonoProcessor<Void> connectMono) {
+	private <T> Consumer<T> updateConnectMono(MonoProcessor<Void> connectMono) {
 		return o -> {
 			if (!connectMono.isTerminated()) {
 				if (o instanceof Throwable) {
