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 ace9c238f5..4c6c62c9aa 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 @@ -16,14 +16,21 @@ package org.springframework.integration.ip.tcp.connection; +import java.io.IOException; import java.net.Socket; import java.net.SocketException; +import java.nio.channels.SelectionKey; +import java.nio.channels.Selector; +import java.nio.channels.ServerSocketChannel; +import java.nio.channels.SocketChannel; +import java.util.Iterator; +import java.util.Map; +import java.util.Set; import java.util.concurrent.Executor; import java.util.concurrent.Executors; import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; - import org.springframework.context.SmartLifecycle; import org.springframework.core.serializer.Deserializer; import org.springframework.core.serializer.Serializer; @@ -346,6 +353,92 @@ public abstract class AbstractConnectionFactory return connection; } + /** + * + * Times out any expired connections then, if selectionCount > 0, processes the selected keys. + * + * @param selectionCount + * @param selector + * @param connections + * @throws IOException + */ + protected void processNioSelections(int selectionCount, final Selector selector, ServerSocketChannel server, + Map connections) throws IOException { + long now = 0; + if (this.soTimeout > 0) { + Iterator it = connections.keySet().iterator(); + now = System.currentTimeMillis(); + while (it.hasNext()) { + SocketChannel channel = it.next(); + if (!channel.isOpen()) { + logger.debug("Removing closed channel"); + it.remove(); + } else { + TcpNioConnection connection = connections.get(channel); + if (now - connection.getLastRead() > this.soTimeout) { + logger.warn("Timing out TcpNioConnection " + + this.port + " : " + + connection.getConnectionId()); + connection.timeout(); + } + } + } + } + if (logger.isTraceEnabled()) + logger.trace("Host" + this.host + " port " + this.port + " SelectionCount: " + selectionCount); + if (selectionCount > 0) { + Set keys = selector.selectedKeys(); + Iterator iterator = keys.iterator(); + while (iterator.hasNext()) { + final SelectionKey key = iterator.next(); + iterator.remove(); + if (!key.isValid()) { + logger.debug("Selection key no longer valid"); + } + else if (key.isReadable()) { + key.interestOps(key.interestOps() - key.readyOps()); + final TcpNioConnection connection; + connection = (TcpNioConnection) key.attachment(); + connection.setLastRead(System.currentTimeMillis()); + this.taskExecutor.execute(new Runnable() { + public void run() { + try { + connection.readPacket(); + } catch (Exception e) { + if (connection.isOpen()) { + logger.error("Exception on read " + + connection.getConnectionId() + " " + + e.getMessage()); + connection.close(); + } else { + logger.debug("Connection closed"); + } + } + if (key.channel().isOpen()) { + key.interestOps(SelectionKey.OP_READ); + selector.wakeup(); + } + }}); + } + else if (key.isAcceptable()) { + doAccept(selector, server, now); + } + else { + logger.error("Unexpected key: " + key); + } + } + } + } + + /** + * @param selector + * @param now + * @throws IOException + */ + protected void doAccept(final Selector selector, ServerSocketChannel server, long now) throws IOException { + throw new UnsupportedOperationException("Nio server factory must override this method"); + } + public int getPhase() { return 0; } 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 ecc858a115..e43e2f3217 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 @@ -21,9 +21,7 @@ import java.nio.ByteBuffer; import java.nio.channels.SelectionKey; import java.nio.channels.Selector; import java.nio.channels.SocketChannel; -import java.util.Iterator; import java.util.Map; -import java.util.Set; import java.util.concurrent.BlockingQueue; import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.Executors; @@ -120,70 +118,12 @@ public class TcpNioClientConnectionFactory extends try { this.selector = Selector.open(); while (this.active) { - int selectionCount = selector.select(this.soTimeout); SocketChannel newChannel; + int selectionCount = selector.select(this.soTimeout); while ((newChannel = newChannels.poll()) != null) { newChannel.register(this.selector, SelectionKey.OP_READ, connections.get(newChannel)); } - if (logger.isTraceEnabled()) - logger.trace("Connection " + host + ":" + port + " SelectionCount: " + selectionCount); - long now = 0; - if (this.soTimeout > 0) { - Iterator it = connections.keySet().iterator(); - now = System.currentTimeMillis(); - while (it.hasNext()) { - SocketChannel channel = it.next(); - if (!channel.isOpen()) { - logger.debug("Removing closed channel"); - it.remove(); - } else { - TcpNioConnection connection = this.connections.get(channel); - if (now - connection.getLastRead() > this.soTimeout) { - logger.warn("Timing out TcpNioConnection " + connection.getConnectionId()); - connection.timeout(); - } - } - } - } - if (selectionCount > 0) { - Set keys = selector.selectedKeys(); - Iterator iterator = keys.iterator(); - while (iterator.hasNext()) { - final SelectionKey key = iterator.next(); - iterator.remove(); - if (!key.isValid()) { - logger.debug("Selection key no longer valid"); - } - else if (key.isReadable()) { - key.interestOps(key.interestOps() - key.readyOps()); - final TcpNioConnection connection; - connection = (TcpNioConnection) key.attachment(); - connection.setLastRead(System.currentTimeMillis()); - this.taskExecutor.execute(new Runnable() { - public void run() { - try { - connection.readPacket(); - } catch (Exception e) { - if (connection.isOpen()) { - logger.error("Exception on read " + - connection.getConnectionId() + " " + - e.getMessage()); - connection.close(); - } else { - logger.debug("Connection closed"); - } - } - if (key.channel().isOpen()) { - key.interestOps(SelectionKey.OP_READ); - selector.wakeup(); - } - }}); - } - else { - logger.error("Unexpected key: " + key); - } - } - } + this.processNioSelections(selectionCount, selector, null, this.connections); } } catch (Exception e) { logger.error("Exception in reader thread", e); 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 c4d159873f..9fa9611cd4 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 @@ -22,6 +22,7 @@ import java.io.PipedInputStream; import java.io.PipedOutputStream; import java.net.SocketTimeoutException; import java.nio.ByteBuffer; +import java.nio.channels.ClosedChannelException; import java.nio.channels.SelectionKey; import java.nio.channels.Selector; import java.nio.channels.SocketChannel; @@ -277,7 +278,6 @@ public class TcpNioConnection extends AbstractTcpConnection { int len = socketChannel.read(rawBuffer); if (len < 0) { this.closeConnection(); - throw new IOException("Channel closed"); } rawBuffer.flip(); if (logger.isDebugEnabled()) { @@ -305,8 +305,11 @@ public class TcpNioConnection extends AbstractTcpConnection { * Invoked by the factory when there is data to be read. */ public void readPacket() { + logger.debug("Reading..."); try { doRead(); + } catch (ClosedChannelException cce) { + this.closeConnection(); } catch (Exception e) { logger.error("Exception on Read " + this.getConnectionId() + " " + 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 8e513eb0c6..6574d3e081 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 @@ -27,9 +27,7 @@ import java.nio.channels.Selector; import java.nio.channels.ServerSocketChannel; import java.nio.channels.SocketChannel; import java.util.HashMap; -import java.util.Iterator; import java.util.Map; -import java.util.Set; /** /** @@ -112,84 +110,36 @@ public class TcpNioServerConnectionFactory extends AbstractServerConnectionFacto throws IOException, ClosedChannelException, SocketException { while (this.active) { int selectionCount = selector.select(this.soTimeout); - if (logger.isTraceEnabled()) - logger.trace("Port " + port + " SelectionCount: " + selectionCount); - long now = 0; - if (this.soTimeout > 0) { - Iterator it = connections.keySet().iterator(); - now = System.currentTimeMillis(); - while (it.hasNext()) { - SocketChannel channel = it.next(); - if (!channel.isOpen()) { - logger.debug("Removing closed channel"); - it.remove(); - } else { - TcpNioConnection connection = this.connections.get(channel); - if (now - connection.getLastRead() > this.soTimeout) { - logger.warn("Timing out TcpNioConnection " + - this.port + " : " + - connection.getConnectionId()); - connection.timeout(); - } - } - } - } - if (selectionCount > 0) { - Set keys = selector.selectedKeys(); - Iterator iterator = keys.iterator(); - SocketChannel channel = null; - while (iterator.hasNext()) { - final SelectionKey key = iterator.next(); - iterator.remove(); - if (!key.isValid()) { - logger.debug("Selection key no longer valid"); - } - else if (key.isAcceptable()) { - logger.debug("New accept"); - channel = server.accept(); - channel.configureBlocking(false); - Socket socket = channel.socket(); - setSocketAttributes(socket); - TcpNioConnection connection = createTcpNioConnection(channel); - if (connection == null) { - continue; - } - connection.setTaskExecutor(this.taskExecutor); - connection.setLastRead(now); - connections.put(channel, connection); - channel.register(selector, SelectionKey.OP_READ, connection); - } - else if (key.isReadable()) { - key.interestOps(key.interestOps() - key.readyOps()); - final TcpNioConnection connection; - connection = (TcpNioConnection) key.attachment(); - connection.setLastRead(System.currentTimeMillis()); - this.taskExecutor.execute(new Runnable() { - public void run() { - try { - connection.readPacket(); - } catch (Exception e) { - if (connection.isOpen()) { - logger.error("Exception on read " + e.getMessage()); - connection.close(); - } else { - logger.debug("Connection closed"); - } - } - if (key.channel().isOpen()) { - key.interestOps(SelectionKey.OP_READ); - selector.wakeup(); - } - }}); - } - else { - logger.error("Unexpected key: " + key); - } - } - } + this.processNioSelections(selectionCount, selector, server, this.connections); } } - + + /** + * @param selector + * @param connections + * @param server + * @param now + * @throws IOException + * @throws SocketException + * @throws ClosedChannelException + */ + @Override + protected void doAccept(final Selector selector, ServerSocketChannel server, long now) throws IOException { + logger.debug("New accept"); + SocketChannel channel = server.accept(); + channel.configureBlocking(false); + Socket socket = channel.socket(); + setSocketAttributes(socket); + TcpNioConnection connection = createTcpNioConnection(channel); + if (connection == null) { + return; + } + connection.setTaskExecutor(this.taskExecutor); + connection.setLastRead(now); + connections.put(channel, connection); + channel.register(selector, SelectionKey.OP_READ, connection); + } + private TcpNioConnection createTcpNioConnection(SocketChannel socketChannel) { try { TcpNioConnection connection = new TcpNioConnection(socketChannel, true); diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/serializer/AbstractByteArraySerializer.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/serializer/AbstractByteArraySerializer.java index 749acb6bcd..3bdd8d41fd 100644 --- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/serializer/AbstractByteArraySerializer.java +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/serializer/AbstractByteArraySerializer.java @@ -59,8 +59,8 @@ public abstract class AbstractByteArraySerializer implements protected void checkClosure(int bite) throws IOException { if (bite < 0) { - logger.debug("Socket closed"); - throw new IOException("Socket closed"); + logger.debug("Socket closed during message assembly"); + throw new IOException("Socket closed during message assembly"); } } 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 9d68a4cdab..4bdacd9ae9 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,6 +13,8 @@ diff --git a/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/TcpNioConnectionReadTests.java b/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/TcpNioConnectionReadTests.java index 8a27d3ea56..8e01d4fcf9 100644 --- a/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/TcpNioConnectionReadTests.java +++ b/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/TcpNioConnectionReadTests.java @@ -357,6 +357,46 @@ public class TcpNioConnectionReadTests { scf.close(); } + /** + * Tests socket closure when no data received. + * + * @throws Exception + */ + @Test + public void testCloseCleanupPartialData() throws Exception { + int port = SocketTestUtils.findAvailableServerSocket(); + ByteArrayCrLfSerializer serializer = new ByteArrayCrLfSerializer(); + serializer.setMaxMessageSize(1024); + final List> responses = new ArrayList>(); + final Semaphore semaphore = new Semaphore(0); + final List added = new ArrayList(); + final List removed = new ArrayList(); + AbstractServerConnectionFactory scf = getConnectionFactory(port, serializer,new TcpListener() { + public boolean onMessage(Message message) { + responses.add(message); + semaphore.release(); + return false; + } + }, new TcpSender() { + public void addNewConnection(TcpConnection connection) { + added.add(connection); + semaphore.release(); + } + public void removeDeadConnection(TcpConnection connection) { + removed.add(connection); + semaphore.release(); + } + }); + Socket socket = SocketFactory.getDefault().createSocket("localhost", port); + socket.getOutputStream().write("partial".getBytes()); + socket.close(); + whileOpen(semaphore, added); + assertEquals(1, added.size()); + assertTrue(semaphore.tryAcquire(10000, TimeUnit.MILLISECONDS)); + assertTrue(removed.size() > 0); + scf.close(); + } + /** * Tests socket closure when mid-message *