From 2bf8196353f77e6e65d2f972c197e24252583bd7 Mon Sep 17 00:00:00 2001 From: Gary Russell Date: Wed, 30 Oct 2013 13:57:24 +0200 Subject: [PATCH] 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 --- .../AbstractClientConnectionFactory.java | 73 +++++++++++++++++-- .../tcp/connection/TcpConnectionSupport.java | 1 - .../TcpNetClientConnectionFactory.java | 14 +--- .../TcpNetServerConnectionFactory.java | 10 ++- .../TcpNioClientConnectionFactory.java | 22 ++---- .../TcpNioServerConnectionFactory.java | 1 + .../ConnectionToConnectionTests-context.xml | 38 ++++++++-- .../ip/tcp/ConnectionToConnectionTests.java | 56 +++++++++----- .../tcp/connection/ConnectionEventTests.java | 26 +++---- 9 files changed, 163 insertions(+), 78 deletions(-) diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/AbstractClientConnectionFactory.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/AbstractClientConnectionFactory.java index e3f2c91868..700b54c865 100644 --- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/AbstractClientConnectionFactory.java +++ b/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. diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpConnectionSupport.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpConnectionSupport.java index 01d2ed28b0..05f2a675be 100644 --- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpConnectionSupport.java +++ b/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()); } diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNetClientConnectionFactory.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNetClientConnectionFactory.java index 6acaee17cf..7c02f261a2 100644 --- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNetClientConnectionFactory.java +++ b/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(), diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNetServerConnectionFactory.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNetServerConnectionFactory.java index 631628ca67..0504f2f693 100644 --- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNetServerConnectionFactory.java +++ b/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"); diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNioClientConnectionFactory.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNioClientConnectionFactory.java index ff2ec2c4ea..6935fbdd3b 100644 --- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNioClientConnectionFactory.java +++ b/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( diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNioServerConnectionFactory.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNioServerConnectionFactory.java index aa882c5e46..6b97ac6bb0 100644 --- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNioServerConnectionFactory.java +++ b/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); diff --git a/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/ConnectionToConnectionTests-context.xml b/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/ConnectionToConnectionTests-context.xml index dc8857782c..e762803a6e 100644 --- a/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/ConnectionToConnectionTests-context.xml +++ b/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/ConnectionToConnectionTests-context.xml @@ -13,9 +13,9 @@ - - - + + + + + + diff --git a/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/ConnectionToConnectionTests.java b/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/ConnectionToConnectionTests.java index 31032d8649..6b4d849c40 100644 --- a/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/ConnectionToConnectionTests.java +++ b/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 eventMessage; while ((eventMessage = (Message) 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(); } diff --git a/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/ConnectionEventTests.java b/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/ConnectionEventTests.java index 7bfe7b1735..2f2034ccb4 100644 --- a/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/ConnectionEventTests.java +++ b/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 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**")); } }