GH-8609: Fix TcpConnectorInterceptor chain propagation

Fixes https://github.com/spring-projects/spring-integration/issues/8609

* Changed to passed the self instance to `addNewConnection()` instead of argument's connection
in a `TcpConnectionInterceptorSupport` to compose a chain of interceptors from top to down.
This way the target `Sender` get the last interceptor in a chain as its connection.

**Cherry-pick to `6.0.x` & `5.5.x`**

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2022 the original author or authors.
+ * Copyright 2002-2023 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.
@@ -33,6 +33,7 @@ import org.springframework.messaging.support.ErrorMessage;
  * to the underlying {@link TcpConnection}.
  *
  * @author Gary Russell
+ * @author Kazuki Shimizu
  *
  * @since 2.0
  */
@@ -232,7 +233,7 @@ public abstract class TcpConnectionInterceptorSupport extends TcpConnectionSuppo
 	@Override
 	public void addNewConnection(TcpConnection connection) {
 		if (this.interceptedSenders != null) {
-			this.interceptedSenders.forEach(sender -> sender.addNewConnection(connection));
+			this.interceptedSenders.forEach(sender -> sender.addNewConnection(this));
 		}
 	}
 

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2022 the original author or authors.
+ * Copyright 2022-2023 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.
@@ -18,7 +18,9 @@ package org.springframework.integration.ip.tcp.connection;
 
 import java.util.ArrayList;
 import java.util.Collections;
+import java.util.HashMap;
 import java.util.List;
+import java.util.Map;
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicInteger;
@@ -30,6 +32,7 @@ import static org.assertj.core.api.Assertions.assertThat;
 
 /**
  * @author Gary Russell
+ * @author Kazuki Shimizu
  * @since 5.3.10
  *
  */
@@ -81,11 +84,14 @@ public class TcpSenderTests {
 		List<Integer> addOrder = Collections.synchronizedList(new ArrayList<>());
 		List<Integer> remOrder = Collections.synchronizedList(new ArrayList<>());
 		AtomicReference<Thread> thread = new AtomicReference<>();
+		Map<Integer, TcpConnection> interceptorsPerInstance = new HashMap<>();
+		List<TcpConnection> passedConnectionsToSenderViaAddNewConnection = new ArrayList<>();
 		class InterceptorFactory extends HelloWorldInterceptorFactory {
 
 			@Override
 			public TcpConnectionInterceptorSupport getInterceptor() {
-				return new TcpConnectionInterceptorSupport() {
+
+				TcpConnectionInterceptorSupport interceptor = new TcpConnectionInterceptorSupport() {
 
 					private final int instance = instances.incrementAndGet();
 
@@ -107,6 +113,8 @@ public class TcpSenderTests {
 					}
 
 				};
+				interceptorsPerInstance.put(instances.get(), interceptor);
+				return interceptor;
 			}
 
 		}
@@ -118,6 +126,7 @@ public class TcpSenderTests {
 			@Override
 			public void addNewConnection(TcpConnection connection) {
 				addOrder.add(99);
+				passedConnectionsToSenderViaAddNewConnection.add(connection);
 				adds.countDown();
 			}
 
@@ -146,6 +155,9 @@ public class TcpSenderTests {
 		assertThat(remOrder).containsExactly(1, 2, 99, 3, 4, 5, 99, 6);
 		assertThat(interceptorAddCalled.await(10, TimeUnit.SECONDS)).isTrue();
 		assertThat(interceptorRemCalled.await(10, TimeUnit.SECONDS)).isTrue();
+		// should be passed the last interceptor to the real sender via addNewConnection method
+		assertThat(passedConnectionsToSenderViaAddNewConnection.get(0)).isSameAs(interceptorsPerInstance.get(3));
+		assertThat(passedConnectionsToSenderViaAddNewConnection.get(1)).isSameAs(interceptorsPerInstance.get(6));
 	}
 
 }