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**")); } }