INT-3178 Defer Publishing of TCP Open Event

Previously, the TcpConnectionOpenEvent was published as soon
as the socket was connected, but the TcpConnection was not yet
ready for use.

Defer publishing the event until the connection is fully
initialized and registered with the factory and thus available
for use.

Update the test cases to reflect this behavior.

Since the event is now published by the appropriate concrete
factory implementations, enhance the ConnectionToConnectionTests
to verify proper eventing with both Net and NIO implementations.

INT-3178: Add `theConnection` `ReadWriteLock`

INT-3178 Polishing

- Refactor obtainConnection() into obtainSharedConnection() and obtainNewConnection() for Net and NIO Client factories.
- obtainConnection is now common for these two factories, and overridden by the caching and failover factories.
- ReadWriteLock - double check for shared connection after lock obtained.

INT-3178: Polishing

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

@@ -17,17 +17,22 @@
 package org.springframework.integration.ip.tcp.connection;
 
 import java.net.Socket;
+import java.util.concurrent.locks.ReadWriteLock;
+import java.util.concurrent.locks.ReentrantReadWriteLock;
 
 /**
  * Abstract class for client connection factories; client connection factories
  * establish outgoing connections.
  * @author Gary Russell
+ * @author Artem Bilan
  * @since 2.0
  *
  */
 public abstract class AbstractClientConnectionFactory extends AbstractConnectionFactory {
 
-	private TcpConnectionSupport theConnection;
+	private final ReadWriteLock theConnectionLock = new ReentrantReadWriteLock();
+
+	private volatile TcpConnectionSupport theConnection;
 
 	/**
 	 * Constructs a factory that will established connections to the host and port.
@@ -45,18 +50,70 @@ public abstract class AbstractClientConnectionFactory extends AbstractConnection
 	 */
 	public TcpConnectionSupport getConnection() throws Exception {
 		this.checkActive();
-		if (this.isSingleUse()) {
-			return obtainConnection();
-		} else {
-			synchronized(this) {
-				TcpConnectionSupport connection = obtainConnection();
-				this.setTheConnection(connection);
+		return this.obtainConnection();
+	}
+
+	protected TcpConnectionSupport obtainConnection() throws Exception {
+		if (!this.isSingleUse()) {
+			TcpConnectionSupport connection = this.obtainSharedConnection();
+			if (connection != null) {
 				return connection;
 			}
 		}
+		return this.obtainNewConnection();
+	}
+
+	protected final TcpConnectionSupport obtainSharedConnection() throws InterruptedException {
+		this.theConnectionLock.readLock().lockInterruptibly();
+		try {
+			TcpConnectionSupport theConnection = this.getTheConnection();
+			if (theConnection != null && theConnection.isOpen()) {
+				return theConnection;
+			}
+		}
+		finally {
+			this.theConnectionLock.readLock().unlock();
+		}
+
+		return null;
+	}
+
+	protected final TcpConnectionSupport obtainNewConnection() throws Exception {
+		boolean singleUse = this.isSingleUse();
+		if (!singleUse) {
+			this.theConnectionLock.writeLock().lockInterruptibly();
+		}
+		try {
+			TcpConnectionSupport connection;
+			if (!singleUse) {
+				// Another write lock holder might have created a new one by now.
+				connection = this.obtainSharedConnection();
+				if (connection != null) {
+					return connection;
+				}
+			}
+
+			if (logger.isDebugEnabled()) {
+				logger.debug("Opening new socket connection to " + this.getHost() + ":" + this.getPort());
+			}
+
+			connection = this.buildNewConnection();
+			if (!singleUse) {
+				this.setTheConnection(connection);
+			}
+			connection.publishConnectionOpenEvent();
+			return connection;
+		}
+		finally {
+			if (!singleUse) {
+				this.theConnectionLock.writeLock().unlock();
+			}
+		}
 	}
 
-	protected abstract TcpConnectionSupport obtainConnection() throws Exception;
+	protected TcpConnectionSupport buildNewConnection() throws Exception {
+		throw new UnsupportedOperationException("Factories that don't override this class' obtainConnection() must implement this method");
+	}
 
 	/**
 	 * Transfers attributes such as (de)serializers, singleUse etc to a new connection.

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

@@ -127,7 +127,6 @@ public abstract class TcpConnectionSupport implements TcpConnection {
 		if (connectionFactoryName != null) {
 			this.connectionFactoryName = connectionFactoryName;
 		}
-		this.publishConnectionOpenEvent();
 		if (logger.isDebugEnabled()) {
 			logger.debug("New connection " + this.getConnectionId());
 		}

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

@@ -44,20 +44,8 @@ public class TcpNetClientConnectionFactory extends
 		super(host, port);
 	}
 
-	/**
-	 * @throws IOException
-	 * @throws SocketException
-	 * @throws Exception
-	 */
 	@Override
-	protected TcpConnectionSupport obtainConnection() throws Exception {
-		TcpConnectionSupport theConnection = this.getTheConnection();
-		if (theConnection != null && theConnection.isOpen()) {
-			return theConnection;
-		}
-		if (logger.isDebugEnabled()) {
-			logger.debug("Opening new socket connection to " + this.getHost() + ":" + this.getPort());
-		}
+	protected TcpConnectionSupport buildNewConnection() throws IOException, SocketException, Exception {
 		Socket socket = createSocket(this.getHost(), this.getPort());
 		setSocketAttributes(socket);
 		TcpConnectionSupport connection = new TcpNetConnection(socket, false, this.isLookupHost(),

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

@@ -64,7 +64,8 @@ public class TcpNetServerConnectionFactory extends AbstractServerConnectionFacto
 		try {
 			if (this.getLocalAddress() == null) {
 				theServerSocket = createServerSocket(this.getPort(), this.getBacklog(), null);
-			} else {
+			}
+			else {
 				InetAddress whichNic = InetAddress.getByName(this.getLocalAddress());
 				theServerSocket = createServerSocket(this.getPort(), this.getBacklog(), whichNic);
 			}
@@ -80,7 +81,8 @@ public class TcpNetServerConnectionFactory extends AbstractServerConnectionFacto
 				 */
 				try {
 					socket = serverSocket.accept();
-				} catch (SocketTimeoutException ste) {
+				}
+				catch (SocketTimeoutException ste) {
 					if (logger.isDebugEnabled()) {
 						logger.debug("Timed out on accept; continuing");
 					}
@@ -104,9 +106,11 @@ public class TcpNetServerConnectionFactory extends AbstractServerConnectionFacto
 					this.initializeConnection(connection, socket);
 					this.getTaskExecutor().execute(connection);
 					this.harvestClosedConnections();
+					connection.publishConnectionOpenEvent();
 				}
 			}
-		} catch (Exception e) {
+		}
+		catch (Exception e) {
 			// don't log an error if we had a good socket once and now it's closed
 			if (e instanceof SocketException && theServerSocket != null) {
 				logger.warn("Server Socket closed");

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

@@ -18,7 +18,6 @@ package org.springframework.integration.ip.tcp.connection;
 
 import java.io.IOException;
 import java.net.InetSocketAddress;
-import java.net.SocketException;
 import java.nio.ByteBuffer;
 import java.nio.channels.CancelledKeyException;
 import java.nio.channels.ClosedChannelException;
@@ -61,13 +60,9 @@ public class TcpNioClientConnectionFactory extends
 		super(host, port);
 	}
 
-	/**
-	 * @throws Exception
-	 * @throws IOException
-	 * @throws SocketException
-	 */
 	@Override
-	protected TcpConnectionSupport obtainConnection() throws Exception {
+	protected void checkActive() throws IOException {
+		super.checkActive();
 		int n = 0;
 		while (this.selector == null) {
 			try {
@@ -76,16 +71,13 @@ public class TcpNioClientConnectionFactory extends
 				Thread.currentThread().interrupt();
 			}
 			if (n++ > 600) {
-				throw new Exception("Factory failed to start");
+				throw new IOException("Factory failed to start");
 			}
 		}
-		TcpConnectionSupport theConnection = this.getTheConnection();
-		if (theConnection != null && theConnection.isOpen()) {
-			return theConnection;
-		}
-		if (logger.isDebugEnabled()) {
-			logger.debug("Opening new socket channel connection to " + this.getHost() + ":" + this.getPort());
-		}
+	}
+
+	@Override
+	protected TcpConnectionSupport buildNewConnection() throws Exception {
 		SocketChannel socketChannel = SocketChannel.open(new InetSocketAddress(this.getHost(), this.getPort()));
 		setSocketAttributes(socketChannel.socket());
 		TcpNioConnection connection = this.tcpNioConnectionSupport.createNewConnection(

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

@@ -168,6 +168,7 @@ public class TcpNioServerConnectionFactory extends AbstractServerConnectionFacto
 				connection.setLastRead(now);
 				this.channelMap.put(channel, connection);
 				channel.register(selector, SelectionKey.OP_READ, connection);
+				connection.publishConnectionOpenEvent();
 			}
 			catch (Exception e) {
 				logger.error("Exception accepting new connection", e);

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

@@ -13,9 +13,9 @@
 
 	<bean id="tcpIpUtils" class="org.springframework.integration.test.util.SocketUtils" />
 
-	<int-ip:tcp-connection-factory id="server"
+	<int-ip:tcp-connection-factory id="serverNet"
 		type="server"
-		using-nio="true"
+		using-nio="false"
 		single-use="true"
 		port="#{tcpIpUtils.findAvailableServerSocket(10000)}"
 		task-executor="exec"
@@ -23,18 +23,44 @@
 		so-timeout="20000"
 	/>
 
-	<int-ip:tcp-connection-factory id="client"
+	<int-ip:tcp-connection-factory id="clientNet"
 		type="client"
 		host="localhost"
-		port="#{server.port}"
+		port="#{serverNet.port}"
 		single-use="true"
 		lookup-host="false"
 		so-timeout="100000"
 	/>
 
-	<int-ip:tcp-inbound-gateway id="looper"
+	<int-ip:tcp-connection-factory id="serverNio"
+		type="server"
+		using-nio="true"
+		single-use="true"
+		port="#{tcpIpUtils.findAvailableServerSocket(20000)}"
+		task-executor="exec"
+		lookup-host="false"
+		so-timeout="20000"
+	/>
+
+	<int-ip:tcp-connection-factory id="clientNio"
+		type="client"
+		host="localhost"
+		using-nio="true"
+		port="#{serverNio.port}"
+		single-use="true"
+		lookup-host="false"
+		so-timeout="100000"
+	/>
+
+	<int-ip:tcp-inbound-gateway id="gwNet"
 		request-channel="serverSideChannel"
-		connection-factory="server"
+		connection-factory="serverNet"
+		reply-timeout="1"
+		/>
+
+	<int-ip:tcp-inbound-gateway id="gwNio"
+		request-channel="serverSideChannel"
+		connection-factory="serverNio"
 		reply-timeout="1"
 		/>
 

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

@@ -65,10 +65,16 @@ public class ConnectionToConnectionTests {
 	AbstractApplicationContext ctx;
 
 	@Autowired
-	private AbstractClientConnectionFactory client;
+	private AbstractClientConnectionFactory clientNet;
 
 	@Autowired
-	private AbstractServerConnectionFactory server;
+	private AbstractServerConnectionFactory serverNet;
+
+	@Autowired
+	private AbstractClientConnectionFactory clientNio;
+
+	@Autowired
+	private AbstractServerConnectionFactory serverNio;
 
 	@Autowired
 	private QueueChannel serverSideChannel;
@@ -90,9 +96,19 @@ public class ConnectionToConnectionTests {
 		ctx.close();
 	}
 
-	@SuppressWarnings("unchecked")
 	@Test
-	public void testConnect() throws Exception {
+	public void testConnectNet() throws Exception {
+		testConnectGuts(this.clientNet, this.serverNet, "gwNet", true);
+	}
+
+	@Test
+	public void testConnectNio() throws Exception {
+		testConnectGuts(this.clientNio, this.serverNio, "gwNio", false);
+	}
+
+	@SuppressWarnings("unchecked")
+	private void testConnectGuts(AbstractClientConnectionFactory client, AbstractServerConnectionFactory server,
+			String gatewayName, boolean expectExceptionOnClose) throws Exception {
 		TestingUtilities.waitListening(server, null);
 		client.start();
 		for (int i = 0; i < 100; i++) {
@@ -102,7 +118,7 @@ public class ConnectionToConnectionTests {
 			assertNotNull(message);
 			MessageHistory history = MessageHistory.read(message);
 			//org.springframework.integration.test.util.TestUtils
-			Properties componentHistoryRecord = TestUtils.locateComponentInHistory(history, "looper", 0);
+			Properties componentHistoryRecord = TestUtils.locateComponentInHistory(history, gatewayName, 0);
 			assertNotNull(componentHistoryRecord);
 			assertTrue(componentHistoryRecord.get("type").equals("ip:tcp-inbound-gateway"));
 			assertNotNull(message);
@@ -116,7 +132,7 @@ public class ConnectionToConnectionTests {
 		Message<TcpConnectionEvent> eventMessage;
 		while ((eventMessage = (Message<TcpConnectionEvent>) events.receive(1000)) != null) {
 			TcpConnectionEvent event = eventMessage.getPayload();
-			if ("client".equals(event.getConnectionFactoryName())) {
+			if (event.getConnectionFactoryName().startsWith("client")) {
 				if (event instanceof TcpConnectionOpenEvent) {
 					clientOpens++;
 				}
@@ -127,7 +143,7 @@ public class ConnectionToConnectionTests {
 					clientExceptions++;
 				}
 			}
-			else if ("server".equals(event.getConnectionFactoryName())) {
+			else if (event.getConnectionFactoryName().startsWith("server")) {
 				if (event instanceof TcpConnectionOpenEvent) {
 					serverOpens++;
 				}
@@ -138,7 +154,9 @@ public class ConnectionToConnectionTests {
 		}
 		assertEquals(100, clientOpens);
 		assertEquals(100, clientCloses);
-		assertEquals(100, clientExceptions);
+		if (expectExceptionOnClose) {
+			assertEquals(100, clientExceptions);
+		}
 		assertEquals(100, serverOpens);
 		assertEquals(100, serverCloses);
 	}
@@ -146,16 +164,16 @@ public class ConnectionToConnectionTests {
 	@Test
 	public void testConnectRaw() throws Exception {
 		ByteArrayRawSerializer serializer = new ByteArrayRawSerializer();
-		client.setSerializer(serializer);
-		server.setDeserializer(serializer);
-		client.start();
-		TcpConnection connection = client.getConnection();
+		clientNet.setSerializer(serializer);
+		serverNet.setDeserializer(serializer);
+		clientNet.start();
+		TcpConnection connection = clientNet.getConnection();
 		connection.send(MessageBuilder.withPayload("Test").build());
 		Message<?> message = serverSideChannel.receive(10000);
 		assertNotNull(message);
 		MessageHistory history = MessageHistory.read(message);
 		//org.springframework.integration.test.util.TestUtils
-		Properties componentHistoryRecord = TestUtils.locateComponentInHistory(history, "looper", 0);
+		Properties componentHistoryRecord = TestUtils.locateComponentInHistory(history, "gwNet", 0);
 		assertNotNull(componentHistoryRecord);
 		assertTrue(componentHistoryRecord.get("type").equals("ip:tcp-inbound-gateway"));
 		assertNotNull(message);
@@ -164,16 +182,16 @@ public class ConnectionToConnectionTests {
 
 	@Test
 	public void testLookup() throws Exception {
-		client.start();
-		TcpConnection connection = client.getConnection();
+		clientNet.start();
+		TcpConnection connection = clientNet.getConnection();
 		assertFalse(connection.getConnectionId().contains("localhost"));
 		connection.close();
-		client.setLookupHost(true);
-		connection = client.getConnection();
+		clientNet.setLookupHost(true);
+		connection = clientNet.getConnection();
 		assertTrue(connection.getConnectionId().contains("localhost"));
 		connection.close();
-		client.setLookupHost(false);
-		connection = client.getConnection();
+		clientNet.setLookupHost(false);
+		connection = clientNet.getConnection();
 		assertFalse(connection.getConnectionId().contains("localhost"));
 		connection.close();
 	}

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

@@ -52,10 +52,10 @@ public class ConnectionEventTests {
 				theEvent.add((TcpConnectionEvent) event);
 			}
 		}, "foo");
-		assertTrue(theEvent.size() > 0);
-		assertNotNull(theEvent.get(0));
-		assertTrue(theEvent.get(0) instanceof TcpConnectionOpenEvent);
-		assertTrue(theEvent.get(0).toString().endsWith("[factory=foo, connectionId=" + conn.getConnectionId() + "] **OPENED**"));
+		/*
+		 *  Open is not published by the connection itself; the factory publishes it after initialization.
+		 *  See ConnectionToConnectionTests.
+		 */
 		@SuppressWarnings("unchecked")
 		Serializer<Object> serializer = mock(Serializer.class);
 		RuntimeException toBeThrown = new RuntimeException("foo");
@@ -67,17 +67,17 @@ public class ConnectionEventTests {
 			fail("Expected exception");
 		}
 		catch (Exception e) {}
-		assertTrue(theEvent.size() > 1);
-		assertNotNull(theEvent.get(1));
-		assertTrue(theEvent.get(1) instanceof TcpConnectionExceptionEvent);
-		assertTrue(theEvent.get(1).toString().endsWith("[factory=foo, connectionId=" + conn.getConnectionId() + "]"));
-		assertTrue(theEvent.get(1).toString().contains("cause=java.lang.RuntimeException: foo]"));
-		TcpConnectionExceptionEvent event = (TcpConnectionExceptionEvent) theEvent.get(1);
+		assertTrue(theEvent.size() > 0);
+		assertNotNull(theEvent.get(0));
+		assertTrue(theEvent.get(0) instanceof TcpConnectionExceptionEvent);
+		assertTrue(theEvent.get(0).toString().endsWith("[factory=foo, connectionId=" + conn.getConnectionId() + "]"));
+		assertTrue(theEvent.get(0).toString().contains("cause=java.lang.RuntimeException: foo]"));
+		TcpConnectionExceptionEvent event = (TcpConnectionExceptionEvent) theEvent.get(0);
 		assertNotNull(event.getCause());
 		assertSame(toBeThrown, event.getCause());
-		assertTrue(theEvent.size() > 2);
-		assertNotNull(theEvent.get(2));
-		assertTrue(theEvent.get(2).toString().endsWith("[factory=foo, connectionId=" + conn.getConnectionId() + "] **CLOSED**"));
+		assertTrue(theEvent.size() > 1);
+		assertNotNull(theEvent.get(1));
+		assertTrue(theEvent.get(1).toString().endsWith("[factory=foo, connectionId=" + conn.getConnectionId() + "] **CLOSED**"));
 	}
 
 }