GH-3509: Fix regression in TcpNetServerConnectionFactory

Related to #3509

* Fix intercepted connection cleanup tests
* Add missing logic to call `setSenders()` on wrapped connection

**Cherry-pick to `5.4.x`**

Author: gigermocas

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

@@ -35,6 +35,7 @@
  *
  * @author Gary Russell
  * @author Artem Bilan
+ * @author Mário Dias
  *
  * @since 2.0
  *
@@ -180,7 +181,11 @@ private void acceptConnectionAndExecute() throws IOException {
 				setSocketAttributes(socket);
 				TcpConnectionSupport connection = this.tcpNetConnectionSupport.createNewConnection(socket, true,
 						isLookupHost(), getApplicationEventPublisher(), getComponentName());
-				connection = wrapConnection(connection);
+				TcpConnectionSupport wrapped = wrapConnection(connection);
+				if (!wrapped.equals(connection)) {
+					connection.setSenders(getSenders());
+					connection = wrapped;
+				}
 				initializeConnection(connection, socket);
 				getTaskExecutor().execute(connection);
 				harvestClosedConnections();

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

@@ -1199,7 +1199,6 @@ public void testConnectionException() throws Exception {
 		}
 	}
 
-	@SuppressWarnings("unchecked")
 	@Test
 	public void testInterceptedConnection() throws Exception {
 		final CountDownLatch latch = new CountDownLatch(1);
@@ -1229,7 +1228,7 @@ public void testInterceptedConnection() throws Exception {
 		socket.close();
 		assertThat(latch.await(10, TimeUnit.SECONDS)).isTrue();
 		assertThat(connection.get()).isInstanceOf(HelloWorldInterceptor.class);
-		await().untilAsserted(() -> handler.getConnections().isEmpty());
+		await().untilAsserted(() -> assertThat(handler.getConnections()).isEmpty());
 		scf.stop();
 	}
 
@@ -1258,7 +1257,7 @@ public void testInterceptedCleanup() throws Exception {
 		Socket socket = SocketFactory.getDefault().createSocket("localhost", port);
 		socket.close();
 		assertThat(latch.await(10, TimeUnit.SECONDS)).isTrue();
-		await().untilAsserted(() -> handler.getConnections().isEmpty());
+		await().untilAsserted(() -> assertThat(handler.getConnections()).isEmpty());
 		scf.stop();
 	}