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 0e2cf1d7b2..d6cd48f64d 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 @@ -58,9 +58,6 @@ public abstract class AbstractClientConnectionFactory extends AbstractConnection connection.registerListener(this.listener); } if (this.listener != null || this.singleUse) { - if (this.taskExecutor == null) { - this.taskExecutor = Executors.newFixedThreadPool(this.poolSize); - } if (this.soTimeout <= 0) { try { socket.setSoTimeout(DEFAULT_REPLY_TIMEOUT); diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/AbstractConnectionFactory.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/AbstractConnectionFactory.java index db89441c0f..7faa1edc07 100644 --- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/AbstractConnectionFactory.java +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/AbstractConnectionFactory.java @@ -25,10 +25,15 @@ import java.nio.channels.Selector; import java.nio.channels.ServerSocketChannel; import java.nio.channels.SocketChannel; import java.util.Iterator; +import java.util.LinkedList; +import java.util.List; import java.util.Map; import java.util.Set; +import java.util.concurrent.ConcurrentSkipListSet; import java.util.concurrent.Executor; +import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; +import java.util.concurrent.TimeUnit; import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; @@ -74,8 +79,10 @@ public abstract class AbstractConnectionFactory private int soTrafficClass = -1; // don't set by default - protected Executor taskExecutor; + private Executor taskExecutor; + private boolean privateExecutor; + protected Deserializer deserializer = new ByteArrayCrLfSerializer(); protected Serializer serializer = new ByteArrayCrLfSerializer(); @@ -91,6 +98,8 @@ public abstract class AbstractConnectionFactory protected TcpConnectionInterceptorFactoryChain interceptorFactoryChain; private boolean lookupHost = true; + + private List connections = new LinkedList(); /** * Sets socket attributes on the socket. @@ -329,15 +338,22 @@ public abstract class AbstractConnectionFactory public abstract void close(); /** - * Creates a taskExecutor (if one was not provided) and starts - * the listening process on one of its threads. + * Starts the listening process. */ public void start() { + this.active = true; + this.getTaskExecutor().execute(this); + } + + /** + * Creates a taskExecutor (if one was not provided). + */ + protected Executor getTaskExecutor() { if (this.taskExecutor == null) { + this.privateExecutor = true; this.taskExecutor = Executors.newFixedThreadPool(this.poolSize); } - this.active = true; - this.taskExecutor.execute(this); + return this.taskExecutor; } /** @@ -346,30 +362,58 @@ public abstract class AbstractConnectionFactory public void stop() { this.active = false; this.close(); + synchronized (this.connections) { + Iterator iterator = this.connections.iterator(); + while (iterator.hasNext()) { + TcpConnection connection = iterator.next(); + connection.close(); + iterator.remove(); + } + } + if (this.privateExecutor) { + ExecutorService executorService = (ExecutorService) this.taskExecutor; + executorService.shutdown(); + try { + if (!executorService.awaitTermination(10, TimeUnit.SECONDS)) { + logger.debug("Forcing executor shutdown"); + executorService.shutdownNow(); + if (!executorService.awaitTermination(10, TimeUnit.SECONDS)) { + logger.debug("Executor failed to shutdown"); + } + } + } catch (InterruptedException e) { + executorService.shutdownNow(); + Thread.currentThread().interrupt(); + } + } } protected TcpConnection wrapConnection(TcpConnection connection) throws Exception { - if (this.interceptorFactoryChain == null) { - return connection; - } - TcpConnectionInterceptorFactory[] interceptorFactories = - this.interceptorFactoryChain.getInterceptorFactories(); - if (interceptorFactories == null) { - return connection; - } - for (TcpConnectionInterceptorFactory factory : interceptorFactories) { - TcpConnectionInterceptor wrapper = factory.getInterceptor(); - wrapper.setTheConnection(connection); - // if no ultimate listener or sender, register each wrapper in turn - if (this.listener == null) { - connection.registerListener(wrapper); + try { + if (this.interceptorFactoryChain == null) { + return connection; } - if (this.sender == null) { - connection.registerSender(wrapper); + TcpConnectionInterceptorFactory[] interceptorFactories = + this.interceptorFactoryChain.getInterceptorFactories(); + if (interceptorFactories == null) { + return connection; } - connection = wrapper; + for (TcpConnectionInterceptorFactory factory : interceptorFactories) { + TcpConnectionInterceptor wrapper = factory.getInterceptor(); + wrapper.setTheConnection(connection); + // if no ultimate listener or sender, register each wrapper in turn + if (this.listener == null) { + connection.registerListener(wrapper); + } + if (this.sender == null) { + connection.registerSender(wrapper); + } + connection = wrapper; + } + return connection; + } finally { + this.addConnection(connection); } - return connection; } /** @@ -403,6 +447,7 @@ public abstract class AbstractConnectionFactory } } } + this.harvestClosedConnections(); if (logger.isTraceEnabled()) { if (host == null) { logger.trace("Port " + this.port + " SelectionCount: " + selectionCount); @@ -488,4 +533,21 @@ public abstract class AbstractConnectionFactory callback.run(); } + protected void addConnection(TcpConnection connection) { + synchronized (this.connections) { + this.connections.add(connection); + } + } + + protected void harvestClosedConnections() { + synchronized (this.connections) { + Iterator iterator = this.connections.iterator(); + while (iterator.hasNext()) { + TcpConnection connection = iterator.next(); + if (!connection.isOpen()) { + iterator.remove(); + } + } + } + } } 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 0b9d92d2e9..4aac272f00 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 @@ -53,10 +53,11 @@ public class TcpNetClientConnectionFactory extends TcpConnection connection = new TcpNetConnection(socket, false, this.isLookupHost()); connection = wrapConnection(connection); initializeConnection(connection, socket); - this.taskExecutor.execute(connection); + this.getTaskExecutor().execute(connection); if (!this.singleUse) { this.theConnection = connection; } + this.harvestClosedConnections(); return connection; } 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 1f17431935..d75c35ed26 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 @@ -75,7 +75,8 @@ public class TcpNetServerConnectionFactory extends AbstractServerConnectionFacto TcpConnection connection = new TcpNetConnection(socket, true, this.isLookupHost()); connection = wrapConnection(connection); this.initializeConnection(connection, socket); - this.taskExecutor.execute(connection); + this.getTaskExecutor().execute(connection); + this.harvestClosedConnections(); } } catch (Exception e) { this.listening = false; 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 ad6df77813..18ce974e42 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 @@ -24,7 +24,6 @@ import java.nio.channels.SocketChannel; import java.util.Map; import java.util.concurrent.BlockingQueue; import java.util.concurrent.ConcurrentHashMap; -import java.util.concurrent.Executors; import java.util.concurrent.LinkedBlockingQueue; @@ -80,11 +79,7 @@ public class TcpNioClientConnectionFactory extends setSocketAttributes(socketChannel.socket()); TcpNioConnection connection = new TcpNioConnection(socketChannel, false, this.isLookupHost()); connection.setUsingDirectBuffers(this.usingDirectBuffers); - if (this.taskExecutor == null) { - connection.setTaskExecutor(Executors.newSingleThreadExecutor()); - } else { - connection.setTaskExecutor(this.taskExecutor); - } + connection.setTaskExecutor(this.getTaskExecutor()); TcpConnection wrappedConnection = wrapConnection(connection); initializeConnection(wrappedConnection, socketChannel.socket()); socketChannel.configureBlocking(false); @@ -111,6 +106,9 @@ public class TcpNioClientConnectionFactory extends } public void close() { + if (this.selector != null) { + this.selector.wakeup(); + } } public void run() { diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNioConnection.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNioConnection.java index 40ae936a30..cdcb3d0b09 100644 --- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNioConnection.java +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNioConnection.java @@ -162,8 +162,17 @@ public class TcpNioConnection extends AbstractTcpConnection { } else { this.executionControl.decrementAndGet(); } - } catch (IOException e) { - logger.error("Unexpected exception, exiting...", e); + } catch (Exception e) { + if (logger.isTraceEnabled()) { + logger.error("Read exception " + + this.getConnectionId(), e); + } else { + logger.error("Read exception " + + this.getConnectionId() + " " + + e.getClass().getSimpleName() + + ":" + e.getCause() + ":" + e.getMessage()); + } + this.closeConnection(); return; } } finally { @@ -172,7 +181,7 @@ public class TcpNioConnection extends AbstractTcpConnection { // timing was such that we were the last assembler and // a new one wasn't run try { - if (dataAvailable()) { + if (this.isOpen() && dataAvailable()) { checkForAssembler(); } } catch (IOException e) { @@ -191,7 +200,7 @@ public class TcpNioConnection extends AbstractTcpConnection { * @return The Message or null if no data is available. * @throws IOException */ - private synchronized Message convert() throws IOException { + private synchronized Message convert() throws Exception { if (!dataAvailable()) { return null; } @@ -206,15 +215,7 @@ public class TcpNioConnection extends AbstractTcpConnection { } } else { if (!(e instanceof SoftEndOfStreamException)) { - if (logger.isTraceEnabled()) { - logger.error("Read exception " + - this.getConnectionId(), e); - } else { - logger.error("Read exception " + - this.getConnectionId() + " " + - e.getClass().getSimpleName() + - ":" + e.getCause() + ":" + e.getMessage()); - } + throw e; } } return null; 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 a6d826f1d2..e61f51e1a1 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 @@ -44,6 +44,8 @@ public class TcpNioServerConnectionFactory extends AbstractServerConnectionFacto protected boolean usingDirectBuffers; protected Map connections = new HashMap(); + + private Selector selector; /** * Listens for incoming connections on the port. @@ -80,6 +82,7 @@ public class TcpNioServerConnectionFactory extends AbstractServerConnectionFacto final Selector selector = Selector.open(); this.serverChannel.register(selector, SelectionKey.OP_ACCEPT); this.listening = true; + this.selector = selector; doSelect(this.serverChannel, selector); } catch (IOException e) { @@ -116,7 +119,6 @@ public class TcpNioServerConnectionFactory extends AbstractServerConnectionFacto /** * @param selector - * @param connections * @param server * @param now * @throws IOException @@ -134,7 +136,7 @@ public class TcpNioServerConnectionFactory extends AbstractServerConnectionFacto if (connection == null) { return; } - connection.setTaskExecutor(this.taskExecutor); + connection.setTaskExecutor(this.getTaskExecutor()); connection.setLastRead(now); connections.put(channel, connection); channel.register(selector, SelectionKey.OP_READ, connection); @@ -158,6 +160,9 @@ public class TcpNioServerConnectionFactory extends AbstractServerConnectionFacto } public void close() { + if (this.selector != null) { + this.selector.wakeup(); + } if (this.serverChannel == null) { return; } 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 84635eef01..906d145b8d 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 @@ -26,7 +26,9 @@ import java.util.Properties; import org.junit.Test; import org.junit.runner.RunWith; import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.context.ConfigurableApplicationContext; import org.springframework.context.support.AbstractApplicationContext; +import org.springframework.context.support.ClassPathXmlApplicationContext; import org.springframework.integration.Message; import org.springframework.integration.channel.QueueChannel; import org.springframework.integration.history.MessageHistory; @@ -60,6 +62,20 @@ public class ConnectionToConnectionTests { @Autowired private QueueChannel serverSideChannel; + + // Test jvm shutdown + public static void main(String[] args) { + ConfigurableApplicationContext ctx = new ClassPathXmlApplicationContext( + ConnectionToConnectionTests.class.getPackage().getName() + .replaceAll("\\.", "/") + + "/common-context.xml"); + ctx.close(); + ctx = new ClassPathXmlApplicationContext( + ConnectionToConnectionTests.class.getPackage().getName() + .replaceAll("\\.", "/") + + "/ConnectionToConnectionTests-context.xml"); + ctx.close(); + } @Test public void testConnect() throws Exception { @@ -115,4 +131,5 @@ public class ConnectionToConnectionTests { assertFalse(connection.getConnectionId().contains("localhost")); connection.close(); } + } diff --git a/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/TcpInboundGatewayTests.java b/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/TcpInboundGatewayTests.java index e4c92e41b8..f79e914119 100644 --- a/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/TcpInboundGatewayTests.java +++ b/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/TcpInboundGatewayTests.java @@ -231,7 +231,7 @@ public class TcpInboundGatewayTests { private class FailingService { @SuppressWarnings("unused") public String serviceMethod(byte[] bytes) { - throw new RuntimeException("Failed"); + throw new RuntimeException("Planned Failure For Tests"); } } diff --git a/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/TcpOutboundGatewayTests.java b/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/TcpOutboundGatewayTests.java index cbc3c9b217..e139753ff7 100644 --- a/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/TcpOutboundGatewayTests.java +++ b/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/TcpOutboundGatewayTests.java @@ -157,6 +157,7 @@ public class TcpOutboundGatewayTests { for (int i = 0; i < 10; i++) { assertTrue(replies.remove("Reply" + i)); } + done.set(true); } @Test @@ -236,6 +237,7 @@ public class TcpOutboundGatewayTests { for (int i = 0; i < 1; i++) { assertTrue(replies.remove("Reply" + i)); } + done.set(true); } diff --git a/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/TcpSendingMessageHandlerTests.java b/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/TcpSendingMessageHandlerTests.java index a1c818044a..79634b6407 100644 --- a/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/TcpSendingMessageHandlerTests.java +++ b/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/TcpSendingMessageHandlerTests.java @@ -937,11 +937,11 @@ public class TcpSendingMessageHandlerTests { final AtomicBoolean done = new AtomicBoolean(); Executors.newSingleThreadExecutor().execute(new Runnable() { public void run() { + int i = 0; try { ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(port); latch.countDown(); Socket socket = server.accept(); - int i = 0; while (true) { ObjectInputStream ois = new ObjectInputStream(socket.getInputStream()); Object in = null; @@ -962,7 +962,7 @@ public class TcpSendingMessageHandlerTests { oos.writeObject("Reply" + (++i)); } } catch (Exception e) { - if (!done.get()) { + if (i == 0) { e.printStackTrace(); } } diff --git a/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/SOLingerTests.java b/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/SOLingerTests.java index 695befd7df..0be55b0049 100644 --- a/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/SOLingerTests.java +++ b/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/SOLingerTests.java @@ -22,7 +22,7 @@ import static org.junit.Assert.fail; import java.io.IOException; import java.io.InputStream; import java.net.Socket; -import java.net.SocketTimeoutException; +import java.net.SocketException; import javax.net.SocketFactory; @@ -139,7 +139,7 @@ public class SOLingerTests { } try { Socket socket = SocketFactory.getDefault().createSocket("localhost", port); - socket.setSoTimeout(200); + socket.setSoTimeout(10000); String test = "Test\r\n"; socket.getOutputStream().write(test.getBytes()); byte[] buff = new byte[test.length() + 5]; @@ -149,7 +149,7 @@ public class SOLingerTests { n = socket.getInputStream().read(); fail("Expected IOException"); } catch (IOException ioe) { - assertTrue(ioe instanceof SocketTimeoutException); + assertTrue(ioe instanceof SocketException); } } catch (Exception e) { e.printStackTrace(); diff --git a/spring-integration-ip/src/test/java/org/springframework/integration/ip/util/SocketTestUtils.java b/spring-integration-ip/src/test/java/org/springframework/integration/ip/util/SocketTestUtils.java index cbc8ea5909..14df3c59d0 100644 --- a/spring-integration-ip/src/test/java/org/springframework/integration/ip/util/SocketTestUtils.java +++ b/spring-integration-ip/src/test/java/org/springframework/integration/ip/util/SocketTestUtils.java @@ -131,7 +131,7 @@ public class SocketTestUtils { private static void writeByte(OutputStream os, int b, boolean noDelay) throws Exception { os.write(b); - logger.debug("Wrote 0x" + Integer.toHexString(b)); + logger.trace("Wrote 0x" + Integer.toHexString(b)); if (noDelay) { return; }