GH-3523: TcpConnectionEvent Fixes

Resolves https://github.com/spring-projects/spring-integration/issues/3523

- `TcpNetConnection` - publish open event on reader thread to avoid race with first read.
- Intercepted connections - Ensure that the event source is always the outermost interceptor.
- Also reduce delays during NIO client connect.

**cherry-pick to 5.4.x, 5.3.x**

5.3.x will require cherry picking GH-3509 commits.

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

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2020 the original author or authors.
+ * Copyright 2002-2021 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.
@@ -158,7 +158,6 @@ public abstract class AbstractClientConnectionFactory extends AbstractConnection
 					return connection;
 				}
 			}
-
 			return doObtain(singleUse);
 		}
 		catch (RuntimeException e) {
@@ -190,7 +189,6 @@ public abstract class AbstractClientConnectionFactory extends AbstractConnection
 		if (!singleUse) {
 			setTheConnection(connection);
 		}
-		connection.publishConnectionOpenEvent();
 		return connection;
 	}
 

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

@@ -616,6 +616,7 @@ public abstract class AbstractConnectionFactory extends IntegrationObjectSupport
 					connection.registerSender(wrapper);
 				}
 				connection.setWrapped(true);
+				connection.setWrapper(wrapper);
 				connection = wrapper;
 			}
 			return connection;

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

@@ -98,6 +98,8 @@ public abstract class TcpConnectionSupport implements TcpConnection {
 
 	private boolean wrapped;
 
+	private TcpConnectionSupport wrapper;
+
 	/*
 	 * This boolean is to avoid looking for a temporary listener when not needed
 	 * to avoid a CPU cache flush. This does not have to be volatile because it
@@ -415,11 +417,21 @@ public abstract class TcpConnectionSupport implements TcpConnection {
 	/**
 	 * Set to true if intercepted.
 	 * @param wrapped true if wrapped.
+	 * @since 5.4.5
 	 */
 	public void setWrapped(boolean wrapped) {
 		this.wrapped = wrapped;
 	}
 
+	/**
+	 * Set the wrapper.
+	 * @param wrapper the wrapper.
+	 * @since 5.4.6
+	 */
+	public void setWrapper(TcpConnectionSupport wrapper) {
+		this.wrapper = wrapper;
+	}
+
 	public String getConnectionFactoryName() {
 		return this.connectionFactoryName;
 	}
@@ -443,15 +455,30 @@ public abstract class TcpConnectionSupport implements TcpConnection {
 	}
 
 	protected void publishConnectionOpenEvent() {
-		doPublish(new TcpConnectionOpenEvent(this, getConnectionFactoryName()));
+		if (this.wrapper != null) {
+			this.wrapper.publishConnectionOpenEvent();
+		}
+		else {
+			doPublish(new TcpConnectionOpenEvent(this, getConnectionFactoryName()));
+		}
 	}
 
 	protected void publishConnectionCloseEvent() {
-		doPublish(new TcpConnectionCloseEvent(this, getConnectionFactoryName()));
+		if (this.wrapper != null) {
+			this.wrapper.publishConnectionCloseEvent();
+		}
+		else {
+			doPublish(new TcpConnectionCloseEvent(this, getConnectionFactoryName()));
+		}
 	}
 
 	protected void publishConnectionExceptionEvent(Throwable t) {
-		doPublish(new TcpConnectionExceptionEvent(this, getConnectionFactoryName(), t));
+		if (this.wrapper != null) {
+			this.wrapper.publishConnectionExceptionEvent(t);
+		}
+		else {
+			doPublish(new TcpConnectionExceptionEvent(this, getConnectionFactoryName(), t));
+		}
 	}
 
 	/**

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2001-2020 the original author or authors.
+ * Copyright 2001-2021 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.
@@ -189,9 +189,10 @@ public class TcpNetConnection extends TcpConnectionSupport implements Scheduling
 	 */
 	@Override
 	public void run() {
-		if (logger.isDebugEnabled()) {
-			logger.debug(getConnectionId() + " Reading...");
+		if (this.logger.isDebugEnabled()) {
+			this.logger.debug(getConnectionId() + " Reading...");
 		}
+		publishConnectionOpenEvent();
 		while (true) {
 			if (!receiveAndProcessMessage()) {
 				break;

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

@@ -112,59 +112,7 @@ public class TcpNetServerConnectionFactory extends AbstractServerConnectionFacto
 		try {
 			setupServerSocket();
 			while (true) {
-				final Socket socket;
-				/*
-				 *  User hooks in the TcpSocketSupport may have set the server socket SO_TIMEOUT.
-				 *  Not fatal.
-				 */
-				try {
-					if (this.serverSocket == null) {
-						logger.debug(() -> this + " stopped before accept");
-						throw new IOException(this + " stopped before accept");
-					}
-					else {
-						socket = this.serverSocket.accept();
-					}
-				}
-				catch (@SuppressWarnings("unused") SocketTimeoutException ste) {
-					logger.debug("Timed out on accept; continuing");
-					continue;
-				}
-				if (isShuttingDown()) {
-					logger.info(() -> "New connection from " + socket.getInetAddress().getHostAddress()
-							+ ":" + socket.getPort()
-							+ " rejected; the server is in the process of shutting down.");
-					socket.close();
-				}
-				else {
-					logger.debug(() -> "Accepted connection from " + socket.getInetAddress().getHostAddress()
-							+ ":" + socket.getPort());
-					try {
-						setSocketAttributes(socket);
-						TcpConnectionSupport connection = this.tcpNetConnectionSupport.createNewConnection(socket, true,
-								isLookupHost(), getApplicationEventPublisher(), getComponentName());
-						TcpConnectionSupport wrapped = wrapConnection(connection);
-						if (!wrapped.equals(connection)) {
-							connection.setSenders(getSenders());
-							connection = wrapped;
-						}
-						initializeConnection(connection, socket);
-						getTaskExecutor().execute(connection);
-						harvestClosedConnections();
-						connection.publishConnectionOpenEvent();
-					}
-					catch (RuntimeException ex) {
-						this.logger.error(ex, () ->
-								"Failed to create and configure a TcpConnection for the new socket: "
-										+ socket.getInetAddress().getHostAddress() + ":" + socket.getPort());
-						try {
-							socket.close();
-						}
-						catch (@SuppressWarnings("unused") IOException e1) { // NOSONAR - exception as flow control
-							// empty
-						}
-					}
-				}
+				acceptConnectionAndExecute();
 			}
 		}
 		catch (IOException ex) { // NOSONAR flow control via exceptions
@@ -200,6 +148,57 @@ public class TcpNetServerConnectionFactory extends AbstractServerConnectionFacto
 		publishServerListeningEvent(getPort());
 	}
 
+	private void acceptConnectionAndExecute() throws IOException {
+		final Socket socket;
+		/*
+		 *  User hooks in the TcpSocketSupport may have set the server socket SO_TIMEOUT.
+		 *  Not fatal.
+		 */
+		try {
+			if (this.serverSocket == null) {
+				logger.debug(() -> this + " stopped before accept");
+				throw new IOException(this + " stopped before accept");
+			}
+			else {
+				socket = this.serverSocket.accept();
+			}
+		}
+		catch (@SuppressWarnings("unused") SocketTimeoutException ste) {
+			logger.debug("Timed out on accept; continuing");
+			return;
+		}
+		if (isShuttingDown()) {
+			logger.info(() -> "New connection from " + socket.getInetAddress().getHostAddress()
+					+ ":" + socket.getPort()
+					+ " rejected; the server is in the process of shutting down.");
+			socket.close();
+		}
+		else {
+			logger.debug(() -> "Accepted connection from " + socket.getInetAddress().getHostAddress()
+					+ ":" + socket.getPort());
+			try {
+				setSocketAttributes(socket);
+				TcpConnectionSupport connection = this.tcpNetConnectionSupport.createNewConnection(socket, true,
+						isLookupHost(), getApplicationEventPublisher(), getComponentName());
+				connection = wrapConnection(connection);
+				initializeConnection(connection, socket);
+				getTaskExecutor().execute(connection);
+				harvestClosedConnections();
+			}
+			catch (RuntimeException ex) {
+				this.logger.error(ex, () ->
+						"Failed to create and configure a TcpConnection for the new socket: "
+								+ socket.getInetAddress().getHostAddress() + ":" + socket.getPort());
+				try {
+					socket.close();
+				}
+				catch (@SuppressWarnings("unused") IOException e1) { // NOSONAR - exception as flow control
+					// empty
+				}
+			}
+		}
+	}
+
 	/**
 	 * Create a new {@link ServerSocket}. This default implementation uses the default
 	 * {@link ServerSocketFactory}. Override to use some other mechanism

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

@@ -106,6 +106,7 @@ public class TcpNioClientConnectionFactory extends
 				connection.setLastRead(System.currentTimeMillis());
 			}
 			this.channelMap.put(socketChannel, connection);
+			wrappedConnection.publishConnectionOpenEvent();
 			this.newChannels.add(socketChannel);
 			this.selector.wakeup();
 			return wrappedConnection;
@@ -125,9 +126,9 @@ public class TcpNioClientConnectionFactory extends
 		boolean connected = socketChannel.finishConnect();
 		long timeLeft = getConnectTimeout().toMillis();
 		while (!connected && timeLeft > 0) {
-			Thread.sleep(50); // NOSONAR Magic #
+			Thread.sleep(5); // NOSONAR Magic #
 			connected = socketChannel.finishConnect();
-			timeLeft -= 50; // NOSONAR Magic #
+			timeLeft -= 5; // NOSONAR Magic #
 		}
 		if (!connected) {
 			throw new IOException("Not connected after connectTimeout");

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

@@ -258,7 +258,6 @@ public class TcpNioServerConnectionFactory extends AbstractServerConnectionFacto
 				}
 				this.channelMap.put(channel, connection);
 				channel.register(selectorForNewSocket, SelectionKey.OP_READ, connection);
-				connection.publishConnectionOpenEvent();
 			}
 		}
 		catch (IOException ex) {
@@ -281,6 +280,7 @@ public class TcpNioServerConnectionFactory extends AbstractServerConnectionFacto
 				connection.setSenders(getSenders());
 			}
 			initializeConnection(wrappedConnection, socketChannel.socket());
+			wrappedConnection.publishConnectionOpenEvent();
 			return connection;
 		}
 		catch (Exception ex) {

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2019 the original author or authors.
+ * Copyright 2002-2021 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.
@@ -108,6 +108,7 @@ public class ConnectionToConnectionTests {
 	@SuppressWarnings("unchecked")
 	private void testConnectGuts(AbstractClientConnectionFactory client, AbstractServerConnectionFactory server,
 			String gatewayName, boolean expectExceptionOnClose) throws Exception {
+
 		TestingUtilities.waitListening(server, null);
 		client.setPort(server.getPort());
 		client.start();

spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/InterceptedSharedConnectionTests-context.xml

@@ -45,6 +45,28 @@
 		interceptor-factory-chain="helloWorldInterceptors"
 	/>
 
+	<int-ip:tcp-connection-factory id="netServer"
+		type="server"
+		port="0"
+		serializer="serializer"
+		deserializer="deserializer"
+		using-nio="false"
+		single-use="true"
+		interceptor-factory-chain="helloWorldInterceptors"
+	/>
+
+	<int-ip:tcp-connection-factory id="netClient"
+		type="client"
+		host="localhost"
+		port="0"
+		single-use="true"
+		so-timeout="100000"
+		using-nio="false"
+		serializer="serializer"
+		deserializer="deserializer"
+		interceptor-factory-chain="helloWorldInterceptors"
+	/>
+
 	<int:channel id="input" />
 
 	<int:channel id="replies">
@@ -71,6 +93,16 @@
 
 	<int:channel id="loop" />
 
+	<int-ip:tcp-inbound-channel-adapter id="inboundNetServer"
+		channel="loop2"
+		connection-factory="netServer"/>
+
+	<int-ip:tcp-outbound-channel-adapter id="outboundNetServer"
+		channel="loop2"
+		connection-factory="netServer"/>
+
+	<int:channel id="loop2" />
+
 	<bean class="org.springframework.integration.ip.tcp.InterceptedSharedConnectionTests$Listener" />
 
 </beans>

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2019 the original author or authors.
+ * Copyright 2002-2021 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,8 +18,10 @@ package org.springframework.integration.ip.tcp;
 
 import static org.assertj.core.api.Assertions.assertThat;
 
-import org.junit.Test;
-import org.junit.runner.RunWith;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+
+import org.junit.jupiter.api.Test;
 
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.context.ApplicationListener;
@@ -30,20 +32,20 @@ import org.springframework.integration.ip.tcp.connection.AbstractClientConnectio
 import org.springframework.integration.ip.tcp.connection.AbstractServerConnectionFactory;
 import org.springframework.integration.ip.tcp.connection.HelloWorldInterceptor;
 import org.springframework.integration.ip.tcp.connection.TcpConnectionOpenEvent;
+import org.springframework.integration.ip.tcp.connection.TcpConnectionSupport;
 import org.springframework.integration.ip.util.TestingUtilities;
 import org.springframework.integration.support.MessageBuilder;
 import org.springframework.messaging.Message;
 import org.springframework.messaging.MessageChannel;
+import org.springframework.messaging.support.GenericMessage;
 import org.springframework.test.annotation.DirtiesContext;
-import org.springframework.test.context.ContextConfiguration;
-import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
+import org.springframework.test.context.junit.jupiter.SpringJUnitConfig;
 
 /**
  * @author Gary Russell
  * @since 2.0
  */
-@ContextConfiguration
-@RunWith(SpringJUnit4ClassRunner.class)
+@SpringJUnitConfig
 @DirtiesContext
 public class InterceptedSharedConnectionTests {
 
@@ -56,6 +58,12 @@ public class InterceptedSharedConnectionTests {
 	@Autowired
 	AbstractClientConnectionFactory client;
 
+	@Autowired
+	AbstractServerConnectionFactory netServer;
+
+	@Autowired
+	AbstractClientConnectionFactory netClient;
+
 	@Autowired
 	Listener listener;
 
@@ -70,7 +78,7 @@ public class InterceptedSharedConnectionTests {
 	 * @throws Exception
 	 */
 	@Test
-	public void test1() throws Exception {
+	void test1() throws Exception {
 		TestingUtilities.waitListening(this.server, null);
 		this.client.setPort(this.server.getPort());
 		this.ctx.getBeansOfType(ConsumerEndpointFactoryBean.class).values().forEach(c -> c.start());
@@ -82,18 +90,49 @@ public class InterceptedSharedConnectionTests {
 			assertThat(message).isNotNull();
 			assertThat(message.getPayload()).isEqualTo("Test");
 		}
-		assertThat(this.listener.openEvent).isNotNull();
-		assertThat(this.listener.openEvent.getConnectionFactoryName()).isEqualTo("client");
+		assertThat(this.listener.clientOpenEvent).isNotNull();
+		assertThat(this.listener.clientOpenEvent.getConnectionFactoryName()).isEqualTo("client");
+		assertThat(this.listener.serverOpenEvent).isNotNull();
+		assertThat(this.listener.serverOpenEvent.getConnectionFactoryName()).isEqualTo("server");
+	}
+
+	@Test
+	void correctOpenNetEventPublished() throws InterruptedException {
+		TestingUtilities.waitListening(this.netServer, null);
+		this.listener.clientOpenEvent = null;
+		this.listener.serverOpenEvent = null;
+		this.netClient.setPort(this.netServer.getPort());
+		this.netClient.start();
+		TcpConnectionSupport conn = this.netClient.getConnection();
+		conn.send(new GenericMessage<>("foo"));
+		conn.close();
+		assertThat(this.listener.latch.await(10, TimeUnit.SECONDS)).isTrue();
+		assertThat(this.listener.clientOpenEvent).isNotNull();
+		assertThat(this.listener.clientOpenEvent.getConnectionFactoryName()).isEqualTo("netClient");
+		assertThat(this.listener.serverOpenEvent).isNotNull();
+		assertThat(this.listener.serverOpenEvent.getConnectionFactoryName()).isEqualTo("netServer");
 	}
 
 	public static class Listener implements ApplicationListener<TcpConnectionOpenEvent> {
 
-		private volatile TcpConnectionOpenEvent openEvent;
+		final CountDownLatch latch = new CountDownLatch(2);
+
+		volatile TcpConnectionOpenEvent clientOpenEvent;
+
+		volatile TcpConnectionOpenEvent serverOpenEvent;
 
 		@Override
 		public void onApplicationEvent(TcpConnectionOpenEvent event) {
 			if (event.getSource() instanceof HelloWorldInterceptor) {
-				this.openEvent = event;
+				if (event.getConnectionFactoryName().startsWith("net")) {
+					this.latch.countDown();
+				}
+				if (event.getConnectionFactoryName().contains("lient")) {
+					this.clientOpenEvent = event;
+				}
+				else {
+					this.serverOpenEvent = event;
+				}
 			}
 		}
 

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

@@ -29,7 +29,9 @@ import static org.mockito.Mockito.spy;
 import static org.mockito.Mockito.verify;
 import static org.mockito.Mockito.when;
 
+import java.io.IOException;
 import java.net.InetSocketAddress;
+import java.net.Socket;
 import java.net.SocketAddress;
 import java.time.Duration;
 import java.util.ArrayList;
@@ -43,6 +45,8 @@ import java.util.concurrent.atomic.AtomicInteger;
 import java.util.concurrent.atomic.AtomicReference;
 import java.util.stream.Collectors;
 
+import javax.net.SocketFactory;
+
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
 import org.junit.jupiter.api.Test;
@@ -78,6 +82,36 @@ import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler;
  */
 public class ConnectionFactoryTests {
 
+	@Test
+	void netOpenEventOnReadThread() throws InterruptedException, IOException {
+		TcpNetServerConnectionFactory server = new TcpNetServerConnectionFactory(0);
+		AtomicReference<Thread> readThread = new AtomicReference<>();
+		AtomicReference<Thread> openEventThread = new AtomicReference<>();
+		CountDownLatch latch1 = new CountDownLatch(1);
+		CountDownLatch latch2 = new CountDownLatch(1);
+		server.registerListener(msg -> {
+			readThread.set(Thread.currentThread());
+			latch2.countDown();
+			return false;
+		});
+		server.setApplicationEventPublisher(event -> {
+			if (event instanceof TcpConnectionServerListeningEvent) {
+				latch1.countDown();
+			}
+			if (event instanceof TcpConnectionOpenEvent) {
+				openEventThread.set(Thread.currentThread());
+			}
+		});
+		server.afterPropertiesSet();
+		server.start();
+		assertThat(latch1.await(10, TimeUnit.SECONDS)).isTrue();
+		Socket socket = SocketFactory.getDefault().createSocket("localhost", server.getPort());
+		socket.getOutputStream().write("test\r\n".getBytes());
+		socket.close();
+		assertThat(latch2.await(10, TimeUnit.SECONDS)).isTrue();
+		assertThat(readThread.get()).isSameAs(openEventThread.get());
+	}
+
 	@Test
 	public void factoryBeanTests() {
 		TcpConnectionFactoryFactoryBean fb = new TcpConnectionFactoryFactoryBean("client");