GH-3713: TCP: Fix Intercepted Sender List

Resolves #3713

#3326 added support
for multiple `TcpSenders`. However, when connections are intercepted, the sender
list was not properly chained through the interceptors.

- override `registerSenders` and properly capture the real senders in the last
  interceptor and intermediate interceptors
- this ensures that `addNewConnection` is called on each interceptor
- when removing dead connections, use the (possibly intercepted) connection sender
  list insted of the factory's raw sender list.

**cherry-pick to 5.5.x**

Author: garyrussell

spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/AbstractConnectionFactory.java

@@ -563,10 +563,10 @@ public void stop() {
 		synchronized (this.connections) {
 			Iterator<Entry<String, TcpConnectionSupport>> iterator = this.connections.entrySet().iterator();
 			while (iterator.hasNext()) {
-				TcpConnection connection = iterator.next().getValue();
+				TcpConnectionSupport connection = iterator.next().getValue();
 				connection.close();
 				iterator.remove();
-				getSenders().forEach(sender -> sender.removeDeadConnection(connection));
+				connection.getSenders().forEach(sender -> sender.removeDeadConnection(connection));
 			}
 		}
 		synchronized (this.lifecycleMonitor) {
@@ -866,7 +866,7 @@ private List<String> removeClosedConnectionsAndReturnOpenConnectionIds() {
 				TcpConnectionSupport connection = entry.getValue();
 				if (!connection.isOpen()) {
 					iterator.remove();
-					getSenders().forEach(sender -> sender.removeDeadConnection(connection));
+					connection.getSenders().forEach(sender -> sender.removeDeadConnection(connection));
 					logger.debug(() -> getComponentName() + ": Removed closed connection: " +
 							connection.getConnectionId());
 				}
@@ -944,7 +944,7 @@ public boolean closeConnection(String connectionId) {
 				try {
 					connection.close();
 					closed = true;
-					getSenders().forEach(sender -> sender.removeDeadConnection(connection));
+					connection.getSenders().forEach(sender -> sender.removeDeadConnection(connection));
 				}
 				catch (Exception ex) {
 					logger.debug(ex, () -> "Failed to close connection " + connectionId);

spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpConnectionInterceptorSupport.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2021 the original author or authors.
+ * Copyright 2002-2022 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.
@@ -16,11 +16,14 @@
 
 package org.springframework.integration.ip.tcp.connection;
 
+import java.util.List;
+
 import javax.net.ssl.SSLSession;
 
 import org.springframework.context.ApplicationEventPublisher;
 import org.springframework.core.serializer.Deserializer;
 import org.springframework.core.serializer.Serializer;
+import org.springframework.lang.Nullable;
 import org.springframework.messaging.Message;
 import org.springframework.messaging.support.ErrorMessage;
 
@@ -38,9 +41,10 @@ public abstract class TcpConnectionInterceptorSupport extends TcpConnectionSuppo
 
 	private TcpListener tcpListener;
 
-	private TcpSender tcpSender;
+	private boolean realSender;
+
+	private List<TcpSender> interceptedSenders;
 
-	private Boolean realSender;
 
 	public TcpConnectionInterceptorSupport() {
 	}
@@ -92,10 +96,24 @@ public void registerListener(TcpListener listener) {
 
 	@Override
 	public void registerSender(TcpSender sender) {
-		this.tcpSender = sender;
 		this.theConnection.registerSender(this);
 	}
 
+	@Override
+	public void registerSenders(List<TcpSender> sendersToRegister) {
+		this.interceptedSenders = sendersToRegister;
+		if (sendersToRegister.size() > 0) {
+			if (!(sendersToRegister.get(0) instanceof TcpConnectionInterceptorSupport)) {
+				this.realSender = true;
+			}
+			else {
+				this.realSender = ((TcpConnectionInterceptorSupport) this.interceptedSenders.get(0))
+						.hasRealSender();
+			}
+		}
+		super.registerSender(this);
+	}
+
 	/**
 	 * {@inheritDoc}
 	 * <p>
@@ -198,21 +216,22 @@ public void setTheConnection(TcpConnectionSupport theConnection) {
 	 * @return the listener
 	 */
 	@Override
+	@Nullable
 	public TcpListener getListener() {
 		return this.tcpListener;
 	}
 
 	@Override
 	public void addNewConnection(TcpConnection connection) {
-		if (this.tcpSender != null) {
-			this.tcpSender.addNewConnection(this);
+		if (this.interceptedSenders != null) {
+			this.interceptedSenders.forEach(sender -> sender.addNewConnection(connection));
 		}
 	}
 
 	@Override
 	public void removeDeadConnection(TcpConnection connection) {
-		if (this.tcpSender != null) {
-			this.tcpSender.removeDeadConnection(this);
+		if (this.interceptedSenders != null) {
+			this.interceptedSenders.forEach(sender -> sender.removeDeadConnection(connection));
 		}
 	}
 
@@ -222,19 +241,14 @@ public long incrementAndGetConnectionSequence() {
 	}
 
 	@Override
+	@Nullable
 	public TcpSender getSender() {
-		return this.tcpSender;
+		return this.interceptedSenders != null && this.interceptedSenders.size() > 0
+				? this.interceptedSenders.get(0)
+				: null;
 	}
 
 	protected boolean hasRealSender() {
-		if (this.realSender != null) {
-			return this.realSender;
-		}
-		TcpSender sender = getSender();
-		while (sender instanceof TcpConnectionInterceptorSupport) {
-			sender = ((TcpConnectionInterceptorSupport) sender).getSender();
-		}
-		this.realSender = sender != null;
 		return this.realSender;
 	}
 

spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpConnectionSupport.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2001-2021 the original author or authors.
+ * Copyright 2001-2022 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.
@@ -337,6 +337,7 @@ public void registerSenders(List<TcpSender> sendersToRegister) {
 	 * @return the listener
 	 */
 	@Override
+	@Nullable
 	public TcpListener getListener() {
 		if (this.needsTest && this.testListener != null) {
 			this.needsTest = false;

spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/TcpSenderTests.java

@@ -69,12 +69,27 @@ void senderCalledForDeadConnectionClientNio() throws InterruptedException {
 	private void senderCalledForDeadConnectionClient(AbstractClientConnectionFactory client) throws InterruptedException {
 		CountDownLatch adds = new CountDownLatch(2);
 		CountDownLatch removes = new CountDownLatch(2);
+		CountDownLatch interceptorAddCalled = new CountDownLatch(2);
+		CountDownLatch interceptorRemCalled = new CountDownLatch(2);
 		TcpConnectionInterceptorFactoryChain chain = new TcpConnectionInterceptorFactoryChain();
 		chain.setInterceptor(new HelloWorldInterceptorFactory() {
 
 			@Override
 			public TcpConnectionInterceptorSupport getInterceptor() {
 				return new TcpConnectionInterceptorSupport() {
+
+					@Override
+					public void addNewConnection(TcpConnection connection) {
+						interceptorAddCalled.countDown();
+						super.addNewConnection(connection);
+					}
+
+					@Override
+					public void removeDeadConnection(TcpConnection connection) {
+						interceptorRemCalled.countDown();
+						super.removeDeadConnection(connection);
+					}
+
 				};
 			}
 
@@ -97,12 +112,15 @@ public void removeDeadConnection(TcpConnection connection) {
 		client.afterPropertiesSet();
 		client.start();
 		TcpConnectionSupport conn = client.getConnection();
+		assertThat(((TcpConnectionInterceptorSupport) conn).hasRealSender()).isTrue();
 		conn.close();
 		conn = client.getConnection();
 		assertThat(adds.await(10, TimeUnit.SECONDS)).isTrue();
 		conn.close();
 		client.stop();
 		assertThat(removes.await(10, TimeUnit.SECONDS)).isTrue();
+		assertThat(interceptorAddCalled.await(10, TimeUnit.SECONDS)).isTrue();
+		assertThat(interceptorRemCalled.await(10, TimeUnit.SECONDS)).isTrue();
 	}
 
 }