spring-projectsGH-8609: Fix bug that cannot applied correctly TcpConnectorInterceptor

Fixes spring-projects#8609

* add getLastWrapper() in TcpConnectionSupport
* change to passed the instance that geted via getLastWrapper() instead of self instance to TcpSender#addNewConnection

Author: kazuki43zoo

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2001-2022 the original author or authors.
+ * Copyright 2001-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.
@@ -22,6 +22,7 @@
 import java.util.Collections;
 import java.util.List;
 import java.util.Map;
+import java.util.Optional;
 import java.util.UUID;
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.TimeUnit;
@@ -48,6 +49,7 @@
  *
  * @author Gary Russell
  * @author Artem Bilan
+ * @author Kazuki Shimizu
  *
  * @since 2.0
  *
@@ -313,7 +315,7 @@ public void enableManualListenerRegistration() {
 	public void registerSender(@Nullable TcpSender senderToRegister) {
 		if (senderToRegister != null) {
 			this.senders.add(senderToRegister);
-			senderToRegister.addNewConnection(this);
+			senderToRegister.addNewConnection(getLastWrapper());
 		}
 	}
 
@@ -327,7 +329,7 @@ public void registerSender(@Nullable TcpSender senderToRegister) {
 	public void registerSenders(List<TcpSender> sendersToRegister) {
 		this.senders.addAll(sendersToRegister);
 		for (TcpSender sender : sendersToRegister) {
-			sender.addNewConnection(this);
+			sender.addNewConnection(getLastWrapper());
 		}
 	}
 
@@ -433,6 +435,18 @@ public void setWrapper(TcpConnectionSupport wrapper) {
 		this.wrapper = wrapper;
 	}
 
+	/**
+	 * Get the {@link TcpConnectionSupport} that wrapping at last.
+	 * <p>
+	 * This method add for fixing the <a href="https://github.com/spring-projects/spring-integration/issues/8609">gh-8609</a>.
+	 * </p>
+	 *
+	 * @return the {@link TcpConnectionSupport} that wrapping at last
+	 */
+	TcpConnectionSupport getLastWrapper() {
+		return Optional.ofNullable(this.wrapper).map(TcpConnectionSupport::getLastWrapper).orElse(this);
+	}
+
 	public String getConnectionFactoryName() {
 		return this.connectionFactoryName;
 	}

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 @@
 
 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 @@
 
 /**
  * @author Gary Russell
+ * @author Kazuki Shimizu
  * @since 5.3.10
  *
  */
@@ -81,11 +84,14 @@ private void senderCalledForDeadConnectionClient(AbstractClientConnectionFactory
 		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 synchronized void removeDeadConnection(TcpConnection connection) {
 					}
 
 				};
+				interceptorsPerInstance.put(instances.get(), interceptor);
+				return interceptor;
 			}
 
 		}
@@ -118,6 +126,7 @@ public synchronized void removeDeadConnection(TcpConnection connection) {
 			@Override
 			public void addNewConnection(TcpConnection connection) {
 				addOrder.add(99);
+				passedConnectionsToSenderViaAddNewConnection.add(connection);
 				adds.countDown();
 			}
 
@@ -146,6 +155,9 @@ public synchronized void removeDeadConnection(TcpConnection connection) {
 		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));
 	}
 
 }