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 5b5566f2b6..a6f2c4e690 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 @@ -477,11 +477,11 @@ public abstract class AbstractConnectionFactory while (iterator.hasNext()) { final SelectionKey key = iterator.next(); iterator.remove(); - if (!key.isValid()) { - logger.debug("Selection key no longer valid"); - } - else if (key.isReadable()) { - try { + try { + 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(); @@ -505,23 +505,23 @@ public abstract class AbstractConnectionFactory selector.wakeup(); } }}); - } catch (Exception e) { - if (e instanceof CancelledKeyException) { - logger.debug("Exception on readable key", e); - continue; + } + else if (key.isAcceptable()) { + try { + doAccept(selector, server, now); + } catch (Exception e) { + logger.error("Exception accepting new connection", e); } - logger.error("Exception on readable key", e); } - } - else if (key.isAcceptable()) { - try { - doAccept(selector, server, now); - } catch (Exception e) { - logger.error("Exception accepting new connection", e); + else { + logger.error("Unexpected key: " + key); } - } - else { - logger.error("Unexpected key: " + key); + } catch (CancelledKeyException e) { + if (logger.isDebugEnabled()) { + logger.debug("Selection key " + key + " cancelled"); + } + } catch (Exception e) { + logger.error("Exception on selection key " + key, e); } } } 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 e782e41e15..d9306ade90 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,6 +18,7 @@ package org.springframework.integration.ip.tcp.connection; import java.net.InetSocketAddress; import java.nio.ByteBuffer; +import java.nio.channels.ClosedChannelException; import java.nio.channels.SelectionKey; import java.nio.channels.Selector; import java.nio.channels.SocketChannel; @@ -119,7 +120,13 @@ public class TcpNioClientConnectionFactory extends SocketChannel newChannel; int selectionCount = selector.select(this.soTimeout); while ((newChannel = newChannels.poll()) != null) { - newChannel.register(this.selector, SelectionKey.OP_READ, connections.get(newChannel)); + try { + newChannel.register(this.selector, SelectionKey.OP_READ, connections.get(newChannel)); + } catch (ClosedChannelException cce) { + if (logger.isDebugEnabled()) { + logger.debug("Channel closed before registering with selector for reading"); + } + } } this.processNioSelections(selectionCount, selector, null, this.connections); } 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 ab72f628b4..0dd691322e 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 @@ -303,6 +303,9 @@ public class TcpNioConnection extends AbstractTcpConnection { try { doRead(); } catch (ClosedChannelException cce) { + if (logger.isDebugEnabled()) { + logger.debug(this.getConnectionId() + " Channel is closed"); + } this.closeConnection(); } catch (Exception e) { logger.error("Exception on Read " + 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 b4b21dbdee..f7589ead2c 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 @@ -43,7 +43,6 @@ import javax.net.ServerSocketFactory; import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; import org.junit.Test; - import org.springframework.core.serializer.DefaultDeserializer; import org.springframework.core.serializer.DefaultSerializer; import org.springframework.integration.Message; @@ -495,13 +494,15 @@ public class TcpSendingMessageHandlerTests { try { ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(port); latch.countDown(); - while (true) { + for (int i = 0; i < 2; i++) { Socket socket = server.accept(); semaphore.release(); byte[] b = new byte[6]; readFully(socket.getInputStream(), b); semaphore.release(); + socket.close(); } + server.close(); } catch (Exception e) { if (!done.get()) { e.printStackTrace(); @@ -523,6 +524,7 @@ public class TcpSendingMessageHandlerTests { handler.handleMessage(MessageBuilder.withPayload("Test").build()); assertTrue(semaphore.tryAcquire(4, 10000, TimeUnit.MILLISECONDS)); done.set(true); + ccf.stop(); } @Test @@ -536,13 +538,15 @@ public class TcpSendingMessageHandlerTests { try { ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(port); latch.countDown(); - while (true) { + for (int i = 0; i < 2; i++) { Socket socket = server.accept(); semaphore.release(); - byte[] b = new byte[6]; + byte[] b = new byte[8]; readFully(socket.getInputStream(), b); semaphore.release(); + socket.close(); } + server.close(); } catch (Exception e) { if (!done.get()) { e.printStackTrace(); @@ -554,16 +558,17 @@ public class TcpSendingMessageHandlerTests { ByteArrayCrLfSerializer serializer = new ByteArrayCrLfSerializer(); ccf.setSerializer(serializer); ccf.setDeserializer(serializer); - ccf.setSoTimeout(10000); + ccf.setSoTimeout(5000); ccf.start(); ccf.setSingleUse(true); TcpSendingMessageHandler handler = new TcpSendingMessageHandler(); handler.setConnectionFactory(ccf); assertTrue(latch.await(10, TimeUnit.SECONDS)); - handler.handleMessage(MessageBuilder.withPayload("Test").build()); - handler.handleMessage(MessageBuilder.withPayload("Test").build()); + handler.handleMessage(MessageBuilder.withPayload("Test.1").build()); + handler.handleMessage(MessageBuilder.withPayload("Test.2").build()); assertTrue(semaphore.tryAcquire(4, 10000, TimeUnit.MILLISECONDS)); done.set(true); + ccf.stop(); } @Test @@ -577,15 +582,16 @@ public class TcpSendingMessageHandlerTests { try { ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(port); latch.countDown(); - int i = 0; - while (true) { + for (int i = 1; i < 3; i++) { Socket socket = server.accept(); semaphore.release(); byte[] b = new byte[6]; readFully(socket.getInputStream(), b); - b = ("Reply" + (++i) + "\r\n").getBytes(); + b = ("Reply" + i + "\r\n").getBytes(); socket.getOutputStream().write(b); + socket.close(); } + server.close(); } catch (Exception e) { if (!done.get()) { e.printStackTrace(); @@ -619,6 +625,7 @@ public class TcpSendingMessageHandlerTests { assertTrue(replies.remove("Reply1")); assertTrue(replies.remove("Reply2")); done.set(true); + ccf.stop(); } @Test @@ -632,15 +639,16 @@ public class TcpSendingMessageHandlerTests { try { ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(port); latch.countDown(); - int i = 0; - while (true) { + for (int i = 1; i < 3; i++) { Socket socket = server.accept(); semaphore.release(); byte[] b = new byte[6]; readFully(socket.getInputStream(), b); - b = ("Reply" + (++i) + "\r\n").getBytes(); + b = ("Reply" + i + "\r\n").getBytes(); socket.getOutputStream().write(b); + socket.close(); } + server.close(); } catch (Exception e) { if (!done.get()) { e.printStackTrace(); @@ -674,6 +682,7 @@ public class TcpSendingMessageHandlerTests { assertTrue(replies.remove("Reply1")); assertTrue(replies.remove("Reply2")); done.set(true); + ccf.stop(); } @Test @@ -686,18 +695,19 @@ public class TcpSendingMessageHandlerTests { Executors.newSingleThreadExecutor().execute(new Runnable() { public void run() { try { - ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(port); + ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(port, 100); latch.countDown(); - int i = 0; - while (true) { + for (int i = 0; i < 100; i++) { Socket socket = server.accept(); serverSockets.add(socket); semaphore.release(); byte[] b = new byte[9]; readFully(socket.getInputStream(), b); - b = ("Reply" + (i++) + "\r\n").getBytes(); + b = ("Reply" + i + "\r\n").getBytes(); socket.getOutputStream().write(b); + socket.close(); } + server.close(); } catch (Exception e) { if (!done.get()) { e.printStackTrace(); @@ -740,6 +750,7 @@ public class TcpSendingMessageHandlerTests { assertTrue("Reply" + i + " missing", replies.remove("Reply" + i)); } done.set(true); + ccf.stop(); } @Test