diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/config/IpAdapterParserUtils.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/config/IpAdapterParserUtils.java index 57c878a100..4e6d9f9173 100644 --- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/config/IpAdapterParserUtils.java +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/config/IpAdapterParserUtils.java @@ -25,7 +25,7 @@ import org.w3c.dom.Element; /** * Utility methods and constants for IP adapter parsers. - * + * * @author Gary Russell * @since 2.0 */ @@ -119,13 +119,15 @@ public abstract class IpAdapterParserUtils { public static final String SOCKET_FACTORY_SUPPORT = "socket-factory-support"; + public static final String BACKLOG = "backlog"; + private IpAdapterParserUtils() {} /** - * Adds a constructor-arg to the provided bean definition builder + * Adds a constructor-arg to the provided bean definition builder * with the value of the attribute whose name is provided if that * attribute is defined in the given element. - * + * * @param builder the bean definition builder to be configured * @param element the XML element where the attribute should be defined * @param attributeName the name of the attribute whose value will be @@ -142,7 +144,7 @@ public abstract class IpAdapterParserUtils { /** * @param element * @param builder - * @param parserContext + * @param parserContext */ public static void addHostAndPortToConstructor(Element element, BeanDefinitionBuilder builder, ParserContext parserContext) { @@ -159,18 +161,18 @@ public abstract class IpAdapterParserUtils { /** * @param element * @param builder - * @param parserContext + * @param parserContext */ public static void addPortToConstructor(Element element, BeanDefinitionBuilder builder, ParserContext parserContext) { String port = IpAdapterParserUtils.getPort(element, parserContext); builder.addConstructorArgValue(port); } - + /** * Asserts that a port attribute is supplied. * @param element - * @param parserContext + * @param parserContext * @return The value of the attribute. * @throws BeanCreationException if attribute is not provided. */ @@ -199,7 +201,7 @@ public abstract class IpAdapterParserUtils { /** * Sets the common port attributes on the bean being built (timeout, receive buffer size, * send buffer size). - * @param builder + * @param builder * @param element */ static void addCommonSocketOptions(BeanDefinitionBuilder builder, Element element) { diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/config/TcpConnectionFactoryFactoryBean.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/config/TcpConnectionFactoryFactoryBean.java index f56ed05bb7..89a259050c 100644 --- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/config/TcpConnectionFactoryFactoryBean.java +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/config/TcpConnectionFactoryFactoryBean.java @@ -84,7 +84,7 @@ public class TcpConnectionFactoryFactoryBean extends AbstractFactoryBean= 0, "You cannot set backlog negative"); + this.backlog = backlog; + } + + /** + * @deprecated Default task executor is now a cached rather + * than a fixed pool executor. + * Use {@link #setBacklog(int)} to set the connection backlog. + */ + @Override + @Deprecated + public void setPoolSize(int poolSize) { + this.setBacklog(poolSize); + } } 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 b0df719676..1cc46020c3 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 @@ -52,8 +52,8 @@ public class TcpNetServerConnectionFactory extends AbstractServerConnectionFacto /** * If no listener registers, exits. - * Accepts incoming connections and creates TcpConnections for each new connection. - * Invokes {{@link #initializeConnection(TcpConnection, Socket)} and executes the + * Accepts incoming connections and creates TcpConnections for each new connection. + * Invokes {{@link #initializeConnection(TcpConnection, Socket)} and executes the * connection {@link TcpConnection#run()} using the task executor. * I/O errors on the server socket/channel are logged and the factory is stopped. */ @@ -65,10 +65,10 @@ public class TcpNetServerConnectionFactory extends AbstractServerConnectionFacto } try { if (this.getLocalAddress() == null) { - theServerSocket = createServerSocket(this.getPort(), this.getPoolSize(), null); + theServerSocket = createServerSocket(this.getPort(), this.getBacklog(), null); } else { InetAddress whichNic = InetAddress.getByName(this.getLocalAddress()); - theServerSocket = createServerSocket(this.getPort(), this.getPoolSize(), whichNic); + theServerSocket = createServerSocket(this.getPort(), this.getBacklog(), whichNic); } this.getTcpSocketSupport().postProcessServerSocket(theServerSocket); this.serverSocket = theServerSocket; @@ -132,6 +132,7 @@ public class TcpNetServerConnectionFactory extends AbstractServerConnectionFacto } } + @Override public void close() { if (this.serverSocket == null) { return; @@ -158,5 +159,5 @@ public class TcpNetServerConnectionFactory extends AbstractServerConnectionFacto Assert.notNull(tcpSocketFactorySupport, "TcpSocketFactorySupport may not be null"); this.tcpSocketFactorySupport = tcpSocketFactorySupport; } - + } 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 afb76ab58f..5a5823b15d 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 @@ -1,5 +1,5 @@ /* - * Copyright 2002-2011 the original author or authors. + * Copyright 2002-2012 the original author or authors. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -26,12 +26,15 @@ import java.nio.channels.ClosedChannelException; import java.nio.channels.SelectionKey; import java.nio.channels.Selector; import java.nio.channels.SocketChannel; +import java.util.concurrent.CountDownLatch; import java.util.concurrent.Executor; import java.util.concurrent.Executors; +import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicInteger; import org.springframework.core.serializer.Serializer; import org.springframework.integration.Message; +import org.springframework.integration.MessagingException; import org.springframework.integration.ip.tcp.serializer.SoftEndOfStreamException; import org.springframework.util.Assert; @@ -44,6 +47,8 @@ import org.springframework.util.Assert; */ public class TcpNioConnection extends AbstractTcpConnection { + private static final long DEFAULT_PIPE_TIMEOUT = 60000; + private final SocketChannel socketChannel; private final ChannelOutputStream channelOutputStream; @@ -66,6 +71,8 @@ public class TcpNioConnection extends AbstractTcpConnection { private volatile boolean writingToPipe; + private volatile long pipeTimeout = DEFAULT_PIPE_TIMEOUT; + /** * Constructs a TcpNetConnection for the SocketChannel. * @param socketChannel the socketChannel @@ -80,6 +87,10 @@ public class TcpNioConnection extends AbstractTcpConnection { this.channelOutputStream = new ChannelOutputStream(); } + public void setPipeTimeout(long pipeTimeout) { + this.pipeTimeout = pipeTimeout; + } + @Override public void close() { doClose(); @@ -265,7 +276,7 @@ public class TcpNioConnection extends AbstractTcpConnection { this.writingToPipe = true; try { if (this.taskExecutor == null) { - this.taskExecutor = Executors.newSingleThreadExecutor(); + this.taskExecutor = Executors.newCachedThreadPool(); } // If there is no assembler running, start one checkForAssembler(); @@ -287,7 +298,28 @@ public class TcpNioConnection extends AbstractTcpConnection { if (logger.isDebugEnabled()) { logger.debug("Read " + rawBuffer.limit() + " into raw buffer"); } - this.sendToPipe(this.rawBuffer); + final CountDownLatch latch = new CountDownLatch(1); + /* + * If there are insufficient threads, either to run the + * write to the pipe, or to assemble the data, we need + * avoid a deadlock (block on the write to the pipe). + * Hence the count down latch. + */ + this.taskExecutor.execute(new Runnable() { + public void run() { + try { + TcpNioConnection.this.sendToPipe(rawBuffer); + latch.countDown(); + } + catch (Exception e) { + logger.error(getConnectionId() + " Failed to write to pipe", e); + } + } + }); + if (!latch.await(this.pipeTimeout , TimeUnit.MILLISECONDS)) { + throw new MessagingException("Timed out writing to pipe, probably due to insufficient threads in " + + "a fixed thread pool; consider increasing this task executor pool size"); + } } finally { this.writingToPipe = false; } @@ -295,6 +327,9 @@ public class TcpNioConnection extends AbstractTcpConnection { protected void sendToPipe(ByteBuffer rawBuffer) throws IOException { Assert.notNull(rawBuffer, "rawBuffer cannot be null"); + if (logger.isTraceEnabled()) { + logger.trace("Sending " + rawBuffer.limit() + " to pipe"); + } this.pipedOutputStream.write(rawBuffer.array(), 0, rawBuffer.limit()); this.pipedOutputStream.flush(); rawBuffer.clear(); 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 74842dc0f7..a055e3c0b4 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 @@ -64,8 +64,8 @@ public class TcpNioServerConnectionFactory extends AbstractServerConnectionFacto /** * If no listener registers, exits. - * Accepts incoming connections and creates TcpConnections for each new connection. - * Invokes {{@link #initializeConnection(TcpConnection, Socket)} and executes the + * Accepts incoming connections and creates TcpConnections for each new connection. + * Invokes {{@link #initializeConnection(TcpConnection, Socket)} and executes the * connection {@link TcpConnection#run()} using the task executor. * I/O errors on the server socket/channel are logged and the factory is stopped. */ @@ -84,11 +84,11 @@ public class TcpNioServerConnectionFactory extends AbstractServerConnectionFacto this.serverChannel.configureBlocking(false); if (this.getLocalAddress() == null) { this.serverChannel.socket().bind(new InetSocketAddress(port), - Math.abs(this.getPoolSize())); + Math.abs(this.getBacklog())); } else { InetAddress whichNic = InetAddress.getByName(this.getLocalAddress()); this.serverChannel.socket().bind(new InetSocketAddress(whichNic, port), - Math.abs(this.getPoolSize())); + Math.abs(this.getBacklog())); } final Selector selector = Selector.open(); this.serverChannel.register(selector, SelectionKey.OP_ACCEPT); @@ -109,11 +109,11 @@ public class TcpNioServerConnectionFactory extends AbstractServerConnectionFacto /** * Listens for incoming connections and for notifications that a connected * socket is ready for reading. - * Accepts incoming connections, registers the new socket with the + * Accepts incoming connections, registers the new socket with the * selector for reading. * When a socket is ready for reading, unregisters the read interest and * schedules a call to doRead which reads all available data. When the read - * is complete, the socket is again registered for read interest. + * is complete, the socket is again registered for read interest. * @param server * @param selector * @throws IOException @@ -176,6 +176,7 @@ public class TcpNioServerConnectionFactory extends AbstractServerConnectionFacto } } + @Override public void close() { if (this.selector != null) { this.selector.wakeup(); diff --git a/spring-integration-ip/src/main/resources/org/springframework/integration/ip/config/spring-integration-ip-2.2.xsd b/spring-integration-ip/src/main/resources/org/springframework/integration/ip/config/spring-integration-ip-2.2.xsd index 2b8ac02be4..c397062daf 100644 --- a/spring-integration-ip/src/main/resources/org/springframework/integration/ip/config/spring-integration-ip-2.2.xsd +++ b/spring-integration-ip/src/main/resources/org/springframework/integration/ip/config/spring-integration-ip-2.2.xsd @@ -456,13 +456,29 @@ apply to TCP outbound adapters and gateways. - + + + +A task executor for managing connections; if not specified a +cached thread pool task executor is used. + + + -The number of threads that will be used for socket/channel handling. Only applies -if an external task-executor is NOT being used. When using an external task executor, -its configuration specifies the number of threads. +Deprecated since 2.2; previously it specified the thread pool size if +an external task executor was not provided; it was also used to set +the connection backlog for server factories. That setting is now +specified by the backlog attribute. + + + + + + +Specifies the connection backlog for server sockets. Does not +apply to client factories. diff --git a/spring-integration-ip/src/test/java/org/springframework/integration/ip/config/ParserUnitTests-context.xml b/spring-integration-ip/src/test/java/org/springframework/integration/ip/config/ParserUnitTests-context.xml index 0ba6b8ae23..15d10543a6 100644 --- a/spring-integration-ip/src/test/java/org/springframework/integration/ip/config/ParserUnitTests-context.xml +++ b/spring-integration-ip/src/test/java/org/springframework/integration/ip/config/ParserUnitTests-context.xml @@ -206,7 +206,6 @@ using-nio="#{props['use.nio']}" single-use="true" task-executor="externalTE" - pool-size="321" using-direct-buffers="true" interceptor-factory-chain="interceptors" /> @@ -232,11 +231,18 @@ using-nio="true" single-use="true" task-executor="externalTE" - pool-size="123" + backlog="123" using-direct-buffers="true" interceptor-factory-chain="interceptors" /> + + @@ -275,7 +280,7 @@ using-nio="false" single-use="true" task-executor="externalTE" - pool-size="123" + backlog="123" interceptor-factory-chain="interceptors" /> diff --git a/spring-integration-ip/src/test/java/org/springframework/integration/ip/config/ParserUnitTests.java b/spring-integration-ip/src/test/java/org/springframework/integration/ip/config/ParserUnitTests.java index f655ce71d8..66faa95ad2 100644 --- a/spring-integration-ip/src/test/java/org/springframework/integration/ip/config/ParserUnitTests.java +++ b/spring-integration-ip/src/test/java/org/springframework/integration/ip/config/ParserUnitTests.java @@ -154,6 +154,9 @@ public class ParserUnitTests { @Autowired AbstractConnectionFactory server1; + @Autowired + AbstractConnectionFactory serverBackwardsCompatible; + @Autowired AbstractConnectionFactory server2; @@ -404,7 +407,6 @@ public class ParserUnitTests { assertEquals(true, dfa.getPropertyValue("soTcpNoDelay")); assertEquals(true, dfa.getPropertyValue("singleUse")); assertSame(taskExecutor, dfa.getPropertyValue("taskExecutor")); - assertEquals(321, dfa.getPropertyValue("poolSize")); assertEquals(true, dfa.getPropertyValue("usingDirectBuffers")); assertNotNull(dfa.getPropertyValue("interceptorFactoryChain")); } @@ -424,11 +426,18 @@ public class ParserUnitTests { assertEquals(true, dfa.getPropertyValue("soTcpNoDelay")); assertEquals(true, dfa.getPropertyValue("singleUse")); assertSame(taskExecutor, dfa.getPropertyValue("taskExecutor")); - assertEquals(123, dfa.getPropertyValue("poolSize")); + assertEquals(123, dfa.getPropertyValue("backlog")); assertEquals(true, dfa.getPropertyValue("usingDirectBuffers")); assertNotNull(dfa.getPropertyValue("interceptorFactoryChain")); } + @Test + public void testConnDeprecatedPoolSize() { + assertTrue(serverBackwardsCompatible instanceof TcpNetServerConnectionFactory); + DirectFieldAccessor dfa = new DirectFieldAccessor(serverBackwardsCompatible); + assertEquals(123, dfa.getPropertyValue("backlog")); + } + @Test public void testConnClient2() { assertTrue(client2 instanceof TcpNetClientConnectionFactory); @@ -445,7 +454,6 @@ public class ParserUnitTests { assertEquals(true, dfa.getPropertyValue("soTcpNoDelay")); assertEquals(true, dfa.getPropertyValue("singleUse")); assertSame(taskExecutor, dfa.getPropertyValue("taskExecutor")); - assertEquals(321, dfa.getPropertyValue("poolSize")); assertNotNull(dfa.getPropertyValue("interceptorFactoryChain")); } @@ -464,7 +472,7 @@ public class ParserUnitTests { assertEquals(true, dfa.getPropertyValue("soTcpNoDelay")); assertEquals(true, dfa.getPropertyValue("singleUse")); assertSame(taskExecutor, dfa.getPropertyValue("taskExecutor")); - assertEquals(123, dfa.getPropertyValue("poolSize")); + assertEquals(123, dfa.getPropertyValue("backlog")); assertNotNull(dfa.getPropertyValue("interceptorFactoryChain")); } 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 ce4a61e270..f0f3bdecd6 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 @@ -37,7 +37,6 @@ import java.util.concurrent.atomic.AtomicBoolean; import javax.net.ServerSocketFactory; import org.junit.Test; - import org.springframework.core.serializer.DefaultDeserializer; import org.springframework.core.serializer.DefaultSerializer; import org.springframework.integration.Message; @@ -54,7 +53,7 @@ import org.springframework.integration.support.MessageBuilder; */ public class TcpOutboundGatewayTests { - @Test + @Test public void testGoodNetSingle() throws Exception { final int port = SocketTestUtils.findAvailableServerSocket(); AbstractConnectionFactory ccf = new TcpNetClientConnectionFactory("localhost", port); @@ -84,7 +83,6 @@ public class TcpOutboundGatewayTests { ccf.setDeserializer(new DefaultDeserializer()); ccf.setSoTimeout(10000); ccf.setSingleUse(true); - ccf.setPoolSize(10); ccf.start(); assertTrue(latch.await(10000, TimeUnit.MILLISECONDS)); TcpOutboundGateway gateway = new TcpOutboundGateway(); @@ -98,7 +96,7 @@ public class TcpOutboundGatewayTests { gateway.handleMessage(MessageBuilder.withPayload("Test" + i).build()); } Set replies = new HashSet(); - for (int i = 100; i < 200; i++) { + for (int i = 100; i < 200; i++) { Message m = replyChannel.receive(10000); assertNotNull(m); replies.add((String) m.getPayload()); @@ -108,7 +106,7 @@ public class TcpOutboundGatewayTests { } } - @Test + @Test public void testGoodNetMultiplex() throws Exception { final int port = SocketTestUtils.findAvailableServerSocket(); final CountDownLatch latch = new CountDownLatch(1); @@ -149,7 +147,7 @@ public class TcpOutboundGatewayTests { gateway.handleMessage(MessageBuilder.withPayload("Test" + i).build()); } Set replies = new HashSet(); - for (int i = 100; i < 110; i++) { + for (int i = 100; i < 110; i++) { Message m = replyChannel.receive(10000); assertNotNull(m); replies.add((String) m.getPayload()); @@ -160,7 +158,7 @@ public class TcpOutboundGatewayTests { done.set(true); } - @Test + @Test public void testGoodNetTimeout() throws Exception { final int port = SocketTestUtils.findAvailableServerSocket(); final CountDownLatch latch = new CountDownLatch(1); @@ -212,11 +210,11 @@ public class TcpOutboundGatewayTests { } Set replies = new HashSet(); int timeouts = 0; - for (int i = 0; i < 2; i++) { + for (int i = 0; i < 2; i++) { try { results[i].get(); } catch (InterruptedException e) { - + } catch (ExecutionException e) { if (timeouts > 0) { fail("Unexpected " + e.getMessage()); diff --git a/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/TcpReceivingChannelAdapterTests.java b/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/TcpReceivingChannelAdapterTests.java index 93e43ca9ab..cfb6fb562e 100644 --- a/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/TcpReceivingChannelAdapterTests.java +++ b/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/TcpReceivingChannelAdapterTests.java @@ -151,7 +151,7 @@ public class TcpReceivingChannelAdapterTests { TcpNioServerConnectionFactory scf = new TcpNioServerConnectionFactory(port); ByteArrayCrLfSerializer serializer = new ByteArrayCrLfSerializer(); scf.setSerializer(serializer); - scf.setDeserializer(serializer); + scf.setDeserializer(serializer); scf.setSoTimeout(5000); TcpReceivingChannelAdapter adapter = new TcpReceivingChannelAdapter(); adapter.setConnectionFactory(scf); @@ -168,7 +168,7 @@ public class TcpReceivingChannelAdapterTests { Socket socket = SocketFactory.getDefault().createSocket("localhost", port); for (int i = 0; i < 1000; i++) { socket.getOutputStream().write(("Test" + i + "\r\n").getBytes()); - } + } Set results = new HashSet(); for (int i = 0; i < 1000; i++) { Message message = channel.receive(10000); @@ -210,14 +210,14 @@ public class TcpReceivingChannelAdapterTests { handler.handleMessage(message); message = channel.receive(10000); assertNotNull(message); - handler.handleMessage(message); + handler.handleMessage(message); byte[] b = new byte[6]; readFully(socket.getInputStream(), b); assertEquals("Test\r\n", new String(b)); readFully(socket.getInputStream(), b); assertEquals("Test\r\n", new String(b)); } - + @Test public void testNioShared() throws Exception { final int port = SocketTestUtils.findAvailableServerSocket(); @@ -248,14 +248,14 @@ public class TcpReceivingChannelAdapterTests { handler.handleMessage(message); message = channel.receive(10000); assertNotNull(message); - handler.handleMessage(message); + handler.handleMessage(message); byte[] b = new byte[6]; readFully(socket.getInputStream(), b); assertEquals("Test\r\n", new String(b)); readFully(socket.getInputStream(), b); assertEquals("Test\r\n", new String(b)); } - + @Test public void testNetSingleNoOutbound() throws Exception { final int port = SocketTestUtils.findAvailableServerSocket(); @@ -316,7 +316,7 @@ public class TcpReceivingChannelAdapterTests { socket.getOutputStream().write("Test1\r\n".getBytes()); socket = SocketFactory.getDefault().createSocket("localhost", port); socket.getOutputStream().write("Test2\r\n".getBytes()); - Message message = channel.receive(10000); + Message message = channel.receive(60000); assertNotNull(message); // with single use, results may come back in a different order Set results = new HashSet(); @@ -371,15 +371,15 @@ public class TcpReceivingChannelAdapterTests { handler.handleMessage(message); message = channel.receive(10000); assertNotNull(message); - handler.handleMessage(message); + handler.handleMessage(message); byte[] b = new byte[7]; readFully(socket1.getInputStream(), b); assertEquals("Test1\r\n", new String(b)); readFully(socket2.getInputStream(), b); assertEquals("Test2\r\n", new String(b)); } - - @Test + + @Test public void testNioSingleShared() throws Exception { final int port = SocketTestUtils.findAvailableServerSocket(); TcpNioServerConnectionFactory scf = new TcpNioServerConnectionFactory(port); @@ -412,15 +412,15 @@ public class TcpReceivingChannelAdapterTests { handler.handleMessage(message); message = channel.receive(10000); assertNotNull(message); - handler.handleMessage(message); + handler.handleMessage(message); byte[] b = new byte[7]; readFully(socket1.getInputStream(), b); assertEquals("Test1\r\n", new String(b)); readFully(socket2.getInputStream(), b); assertEquals("Test2\r\n", new String(b)); } - - @Test + + @Test public void testNioSingleSharedMany() throws Exception { final int port = SocketTestUtils.findAvailableServerSocket(); TcpNioServerConnectionFactory scf = new TcpNioServerConnectionFactory(port); @@ -428,12 +428,12 @@ public class TcpReceivingChannelAdapterTests { scf.setSerializer(serializer); scf.setDeserializer(serializer); scf.setSingleUse(true); - scf.setPoolSize(100); + scf.setBacklog(100); TcpSendingMessageHandler handler = new TcpSendingMessageHandler(); handler.setConnectionFactory(scf); TcpReceivingChannelAdapter adapter = new TcpReceivingChannelAdapter(); adapter.setConnectionFactory(scf); - Executor te = Executors.newFixedThreadPool(10); + Executor te = Executors.newCachedThreadPool(); scf.setTaskExecutor(te); scf.start(); QueueChannel channel = new QueueChannel(); @@ -453,17 +453,17 @@ public class TcpReceivingChannelAdapterTests { sockets.add(socket1); } for (int i = 100; i < 200; i++) { - Message message = channel.receive(10000); + Message message = channel.receive(60000); assertNotNull(message); handler.handleMessage(message); } byte[] b = new byte[9]; - for (int i = 100; i < 200; i++) { + for (int i = 100; i < 200; i++) { readFully(sockets.remove(0).getInputStream(), b); assertEquals("Test" + i + "\r\n", new String(b)); } } - + @Test public void testNetInterceptors() throws Exception { final int port = SocketTestUtils.findAvailableServerSocket(); @@ -484,7 +484,7 @@ public class TcpReceivingChannelAdapterTests { AbstractServerConnectionFactory scf = new TcpNetServerConnectionFactory(port); singleSharedInterceptorsGuts(port, scf); } - + @Test public void testNioInterceptors() throws Exception { final int port = SocketTestUtils.findAvailableServerSocket(); @@ -509,11 +509,11 @@ public class TcpReceivingChannelAdapterTests { private void interceptorsGuts(final int port, AbstractServerConnectionFactory scf) throws Exception { scf.setSerializer(new DefaultSerializer()); scf.setDeserializer(new DefaultDeserializer()); - scf.setSingleUse(false); + scf.setSingleUse(false); TcpReceivingChannelAdapter adapter = new TcpReceivingChannelAdapter(); adapter.setConnectionFactory(scf); TcpConnectionInterceptorFactoryChain fc = new TcpConnectionInterceptorFactoryChain(); - fc.setInterceptors(new TcpConnectionInterceptorFactory[] + fc.setInterceptors(new TcpConnectionInterceptorFactory[] {new HelloWorldInterceptorFactory(), new HelloWorldInterceptorFactory()}); scf.setInterceptorFactoryChain(fc); @@ -553,7 +553,7 @@ public class TcpReceivingChannelAdapterTests { scf.setSingleUse(true); scf.setSoTimeout(10000); TcpConnectionInterceptorFactoryChain fc = new TcpConnectionInterceptorFactoryChain(); - fc.setInterceptors(new TcpConnectionInterceptorFactory[] + fc.setInterceptors(new TcpConnectionInterceptorFactory[] {new HelloWorldInterceptorFactory(), new HelloWorldInterceptorFactory()}); scf.setInterceptorFactoryChain(fc); @@ -576,7 +576,7 @@ public class TcpReceivingChannelAdapterTests { new ObjectOutputStream(socket.getOutputStream()).writeObject("Hello"); assertEquals("world!", new ObjectInputStream(socket.getInputStream()).readObject()); new ObjectOutputStream(socket.getOutputStream()).writeObject("Test1"); - + socket = SocketFactory.getDefault().createSocket("localhost", port); new ObjectOutputStream(socket.getOutputStream()).writeObject("Hello"); assertEquals("world!", new ObjectInputStream(socket.getInputStream()).readObject()); @@ -601,7 +601,7 @@ public class TcpReceivingChannelAdapterTests { scf.setSingleUse(true); scf.setSoTimeout(60000); TcpConnectionInterceptorFactoryChain fc = new TcpConnectionInterceptorFactoryChain(); - fc.setInterceptors(new TcpConnectionInterceptorFactory[] + fc.setInterceptors(new TcpConnectionInterceptorFactory[] {new HelloWorldInterceptorFactory(), new HelloWorldInterceptorFactory()}); scf.setInterceptorFactoryChain(fc); @@ -641,7 +641,7 @@ public class TcpReceivingChannelAdapterTests { message = channel.receive(10000); assertNotNull(message); handler.handleMessage(message); - + assertEquals("Test1", new ObjectInputStream(socket1.getInputStream()).readObject()); assertEquals("Test2", new ObjectInputStream(socket2.getInputStream()).readObject()); } 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 597292f9fe..55c2b4c395 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 @@ -51,7 +51,6 @@ import org.springframework.integration.Message; import org.springframework.integration.MessageChannel; import org.springframework.integration.channel.QueueChannel; import org.springframework.integration.core.PollableChannel; -import org.springframework.integration.ip.IpHeaders; import org.springframework.integration.ip.tcp.connection.AbstractConnectionFactory; import org.springframework.integration.ip.tcp.connection.HelloWorldInterceptorFactory; import org.springframework.integration.ip.tcp.connection.TcpConnectionInterceptorFactory; @@ -80,7 +79,7 @@ public class TcpSendingMessageHandlerTests { buff[i] = (byte) is.read(); } } - + @Test public void testNetCrLf() throws Exception { final int port = SocketTestUtils.findAvailableServerSocket(); @@ -111,7 +110,7 @@ public class TcpSendingMessageHandlerTests { ByteArrayCrLfSerializer serializer = new ByteArrayCrLfSerializer(); ccf.setSerializer(serializer); ccf.setDeserializer(serializer); - ccf.setSoTimeout(10000); + ccf.setSoTimeout(10000); ccf.start(); TcpSendingMessageHandler handler = new TcpSendingMessageHandler(); handler.setConnectionFactory(ccf); @@ -119,7 +118,7 @@ public class TcpSendingMessageHandlerTests { adapter.setConnectionFactory(ccf); QueueChannel channel = new QueueChannel(); adapter.setOutputChannel(channel); - assertTrue(latch.await(10, TimeUnit.SECONDS)); + assertTrue(latch.await(10, TimeUnit.SECONDS)); handler.handleMessage(MessageBuilder.withPayload("Test").build()); handler.handleMessage(MessageBuilder.withPayload("Test").build()); Message mOut = channel.receive(10000); @@ -239,7 +238,7 @@ public class TcpSendingMessageHandlerTests { assertTrue(results.remove("Reply2")); done.set(true); } - + @Test public void testNetStxEtx() throws Exception { final int port = SocketTestUtils.findAvailableServerSocket(); @@ -277,7 +276,7 @@ public class TcpSendingMessageHandlerTests { adapter.setConnectionFactory(ccf); QueueChannel channel = new QueueChannel(); adapter.setOutputChannel(channel); - assertTrue(latch.await(10, TimeUnit.SECONDS)); + assertTrue(latch.await(10, TimeUnit.SECONDS)); handler.handleMessage(MessageBuilder.withPayload("Test").build()); handler.handleMessage(MessageBuilder.withPayload("Test").build()); Message mOut = channel.receive(10000); @@ -381,7 +380,7 @@ public class TcpSendingMessageHandlerTests { adapter.setConnectionFactory(ccf); QueueChannel channel = new QueueChannel(); adapter.setOutputChannel(channel); - assertTrue(latch.await(10, TimeUnit.SECONDS)); + assertTrue(latch.await(10, TimeUnit.SECONDS)); handler.handleMessage(MessageBuilder.withPayload("Test").build()); handler.handleMessage(MessageBuilder.withPayload("Test").build()); Message mOut = channel.receive(10000); @@ -484,7 +483,7 @@ public class TcpSendingMessageHandlerTests { adapter.setConnectionFactory(ccf); QueueChannel channel = new QueueChannel(); adapter.setOutputChannel(channel); - assertTrue(latch.await(10, TimeUnit.SECONDS)); + assertTrue(latch.await(10, TimeUnit.SECONDS)); handler.handleMessage(MessageBuilder.withPayload("Test").build()); handler.handleMessage(MessageBuilder.withPayload("Test").build()); Message mOut = channel.receive(10000); @@ -679,7 +678,7 @@ public class TcpSendingMessageHandlerTests { assertTrue(latch.await(10, TimeUnit.SECONDS)); handler.handleMessage(MessageBuilder.withPayload("Test").build()); handler.handleMessage(MessageBuilder.withPayload("Test").build()); - assertTrue(semaphore.tryAcquire(2, 10000, TimeUnit.MILLISECONDS)); + assertTrue(semaphore.tryAcquire(2, 10000, TimeUnit.MILLISECONDS)); Set replies = new HashSet(); for (int i = 0; i < 2; i++) { Message mOut = channel.receive(10000); @@ -736,7 +735,7 @@ public class TcpSendingMessageHandlerTests { assertTrue(latch.await(10, TimeUnit.SECONDS)); handler.handleMessage(MessageBuilder.withPayload("Test").build()); handler.handleMessage(MessageBuilder.withPayload("Test").build()); - assertTrue(semaphore.tryAcquire(2, 10000, TimeUnit.MILLISECONDS)); + assertTrue(semaphore.tryAcquire(2, 10000, TimeUnit.MILLISECONDS)); Set replies = new HashSet(); for (int i = 0; i < 2; i++) { Message mOut = channel.receive(10000); @@ -859,9 +858,9 @@ public class TcpSendingMessageHandlerTests { AbstractConnectionFactory ccf = new TcpNetClientConnectionFactory("localhost", port); ccf.setSerializer(new DefaultSerializer()); ccf.setDeserializer(new DefaultDeserializer()); - ccf.setSoTimeout(10000); + ccf.setSoTimeout(10000); TcpConnectionInterceptorFactoryChain fc = new TcpConnectionInterceptorFactoryChain(); - fc.setInterceptors(new TcpConnectionInterceptorFactory[] + fc.setInterceptors(new TcpConnectionInterceptorFactory[] {new HelloWorldInterceptorFactory(), new HelloWorldInterceptorFactory()}); ccf.setInterceptorFactoryChain(fc); @@ -872,7 +871,7 @@ public class TcpSendingMessageHandlerTests { adapter.setConnectionFactory(ccf); QueueChannel channel = new QueueChannel(); adapter.setOutputChannel(channel); - assertTrue(latch.await(10, TimeUnit.SECONDS)); + assertTrue(latch.await(10, TimeUnit.SECONDS)); handler.handleMessage(MessageBuilder.withPayload("Test").build()); handler.handleMessage(MessageBuilder.withPayload("Test").build()); Message mOut = channel.receive(10000); @@ -921,7 +920,7 @@ public class TcpSendingMessageHandlerTests { AbstractConnectionFactory ccf = new TcpNioClientConnectionFactory("localhost", port); ccf.setSerializer(new DefaultSerializer()); ccf.setDeserializer(new DefaultDeserializer()); - ccf.setSoTimeout(10000); + ccf.setSoTimeout(10000); TcpConnectionInterceptorFactoryChain fc = new TcpConnectionInterceptorFactoryChain(); fc.setInterceptors(new TcpConnectionInterceptorFactory[] {new HelloWorldInterceptorFactory()}); ccf.setInterceptorFactoryChain(fc); @@ -932,7 +931,7 @@ public class TcpSendingMessageHandlerTests { adapter.setConnectionFactory(ccf); QueueChannel channel = new QueueChannel(); adapter.setOutputChannel(channel); - assertTrue(latch.await(10, TimeUnit.SECONDS)); + assertTrue(latch.await(10, TimeUnit.SECONDS)); for (int i = 0; i < 1000; i++) { handler.handleMessage(MessageBuilder.withPayload("Test").build()); } @@ -991,9 +990,9 @@ public class TcpSendingMessageHandlerTests { AbstractConnectionFactory ccf = new TcpNetClientConnectionFactory("localhost", port); ccf.setSerializer(new DefaultSerializer()); ccf.setDeserializer(new DefaultDeserializer()); - ccf.setSoTimeout(10000); + ccf.setSoTimeout(10000); TcpConnectionInterceptorFactoryChain fc = new TcpConnectionInterceptorFactoryChain(); - fc.setInterceptors(new TcpConnectionInterceptorFactory[] + fc.setInterceptors(new TcpConnectionInterceptorFactory[] {new HelloWorldInterceptorFactory(), new HelloWorldInterceptorFactory()}); ccf.setInterceptorFactoryChain(fc); @@ -1001,7 +1000,7 @@ public class TcpSendingMessageHandlerTests { ccf.start(); TcpSendingMessageHandler handler = new TcpSendingMessageHandler(); handler.setConnectionFactory(ccf); - assertTrue(latch.await(10, TimeUnit.SECONDS)); + assertTrue(latch.await(10, TimeUnit.SECONDS)); handler.handleMessage(MessageBuilder.withPayload("Test").build()); done.set(true); } @@ -1047,9 +1046,9 @@ public class TcpSendingMessageHandlerTests { AbstractConnectionFactory ccf = new TcpNioClientConnectionFactory("localhost", port); ccf.setSerializer(new DefaultSerializer()); ccf.setDeserializer(new DefaultDeserializer()); - ccf.setSoTimeout(10000); + ccf.setSoTimeout(10000); TcpConnectionInterceptorFactoryChain fc = new TcpConnectionInterceptorFactoryChain(); - fc.setInterceptors(new TcpConnectionInterceptorFactory[] + fc.setInterceptors(new TcpConnectionInterceptorFactory[] {new HelloWorldInterceptorFactory(), new HelloWorldInterceptorFactory()}); ccf.setInterceptorFactoryChain(fc); @@ -1057,7 +1056,7 @@ public class TcpSendingMessageHandlerTests { ccf.start(); TcpSendingMessageHandler handler = new TcpSendingMessageHandler(); handler.setConnectionFactory(ccf); - assertTrue(latch.await(10, TimeUnit.SECONDS)); + assertTrue(latch.await(10, TimeUnit.SECONDS)); handler.handleMessage(MessageBuilder.withPayload("Test").build()); done.set(true); } 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 8e01d4fcf9..02924d26a0 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 @@ -32,7 +32,6 @@ import java.util.concurrent.TimeUnit; import javax.net.SocketFactory; import org.junit.Test; - import org.springframework.integration.Message; import org.springframework.integration.ip.tcp.serializer.AbstractByteArraySerializer; import org.springframework.integration.ip.tcp.serializer.ByteArrayCrLfSerializer; @@ -47,12 +46,12 @@ import org.springframework.integration.ip.util.SocketTestUtils; public class TcpNioConnectionReadTests { private CountDownLatch latch = new CountDownLatch(1); - + private AbstractServerConnectionFactory getConnectionFactory(int port, AbstractByteArraySerializer serializer, TcpListener listener) throws Exception { return getConnectionFactory(port, serializer, listener, null); } - + private AbstractServerConnectionFactory getConnectionFactory(int port, AbstractByteArraySerializer serializer, TcpListener listener, TcpSender sender) throws Exception { AbstractServerConnectionFactory scf = new TcpNioServerConnectionFactory(port); @@ -72,7 +71,7 @@ public class TcpNioConnectionReadTests { } return scf; } - + /** * Test method for {@link org.springframework.integration.ip.tcp.NioSocketReader}. */ @@ -90,17 +89,17 @@ public class TcpNioConnectionReadTests { return false; } }); - + // Fire up the sender. - + SocketTestUtils.testSendLength(port, latch); latch.countDown(); assertTrue(semaphore.tryAcquire(1, 10000, TimeUnit.MILLISECONDS)); assertTrue(semaphore.tryAcquire(1, 10000, TimeUnit.MILLISECONDS)); assertEquals("Did not receive data", 2, responses.size()); - assertEquals("Data", SocketTestUtils.TEST_STRING + SocketTestUtils.TEST_STRING, + assertEquals("Data", SocketTestUtils.TEST_STRING + SocketTestUtils.TEST_STRING, new String(((Message) responses.get(0)).getPayload())); - assertEquals("Data", SocketTestUtils.TEST_STRING + SocketTestUtils.TEST_STRING, + assertEquals("Data", SocketTestUtils.TEST_STRING + SocketTestUtils.TEST_STRING, new String(((Message) responses.get(1)).getPayload())); scf.close(); } @@ -124,20 +123,20 @@ public class TcpNioConnectionReadTests { return false; } }); - + int howMany = 2; - scf.setPoolSize(howMany + 5); + scf.setBacklog(howMany + 5); // Fire up the sender. SocketTestUtils.testSendFragmented(port, howMany, false); assertTrue(semaphore.tryAcquire(howMany, 20000, TimeUnit.MILLISECONDS)); assertEquals("Expected", howMany, responses.size()); for (int i = 0; i < howMany; i++) { - assertEquals("Data", "xx", + assertEquals("Data", "xx", new String(((Message) responses.get(0)).getPayload())); } scf.close(); } - + /** * Test method for {@link org.springframework.integration.ip.tcp.NioSocketReader}. */ @@ -155,17 +154,17 @@ public class TcpNioConnectionReadTests { return false; } }); - + // Fire up the sender. - + SocketTestUtils.testSendStxEtx(port, latch); latch.countDown(); assertTrue(semaphore.tryAcquire(1, 10000, TimeUnit.MILLISECONDS)); assertTrue(semaphore.tryAcquire(1, 10000, TimeUnit.MILLISECONDS)); assertEquals("Did not receive data", 2, responses.size()); - assertEquals("Data", SocketTestUtils.TEST_STRING + SocketTestUtils.TEST_STRING, + assertEquals("Data", SocketTestUtils.TEST_STRING + SocketTestUtils.TEST_STRING, new String(((Message) responses.get(0)).getPayload())); - assertEquals("Data", SocketTestUtils.TEST_STRING + SocketTestUtils.TEST_STRING, + assertEquals("Data", SocketTestUtils.TEST_STRING + SocketTestUtils.TEST_STRING, new String(((Message) responses.get(1)).getPayload())); scf.close(); } @@ -187,17 +186,17 @@ public class TcpNioConnectionReadTests { return false; } }); - + // Fire up the sender. - + SocketTestUtils.testSendCrLf(port, latch); latch.countDown(); assertTrue(semaphore.tryAcquire(1, 10000, TimeUnit.MILLISECONDS)); assertTrue(semaphore.tryAcquire(1, 10000, TimeUnit.MILLISECONDS)); assertEquals("Did not receive data", 2, responses.size()); - assertEquals("Data", SocketTestUtils.TEST_STRING + SocketTestUtils.TEST_STRING, + assertEquals("Data", SocketTestUtils.TEST_STRING + SocketTestUtils.TEST_STRING, new String(((Message) responses.get(0)).getPayload())); - assertEquals("Data", SocketTestUtils.TEST_STRING + SocketTestUtils.TEST_STRING, + assertEquals("Data", SocketTestUtils.TEST_STRING + SocketTestUtils.TEST_STRING, new String(((Message) responses.get(1)).getPayload())); scf.close(); } @@ -229,9 +228,9 @@ public class TcpNioConnectionReadTests { semaphore.release(); } }); - + // Fire up the sender. - + SocketTestUtils.testSendLengthOverflow(port); whileOpen(semaphore, added); assertEquals(1, added.size()); @@ -268,9 +267,9 @@ public class TcpNioConnectionReadTests { semaphore.release(); } }); - + // Fire up the sender. - + SocketTestUtils.testSendStxEtxOverflow(port); whileOpen(semaphore, added); assertEquals(1, added.size()); @@ -307,9 +306,9 @@ public class TcpNioConnectionReadTests { semaphore.release(); } }); - + // Fire up the sender. - + SocketTestUtils.testSendCrLfOverflow(port); whileOpen(semaphore, added); assertEquals(1, added.size()); @@ -320,7 +319,7 @@ public class TcpNioConnectionReadTests { /** * Tests socket closure when no data received. - * + * * @throws Exception */ @Test @@ -359,7 +358,7 @@ public class TcpNioConnectionReadTests { /** * Tests socket closure when no data received. - * + * * @throws Exception */ @Test @@ -399,7 +398,7 @@ public class TcpNioConnectionReadTests { /** * Tests socket closure when mid-message - * + * * @throws Exception */ @Test @@ -407,10 +406,10 @@ public class TcpNioConnectionReadTests { ByteArrayCrLfSerializer serializer = new ByteArrayCrLfSerializer(); testClosureMidMessageGuts(serializer, "xx"); } - + /** * Tests socket closure when mid-message - * + * * @throws Exception */ @@ -422,7 +421,7 @@ public class TcpNioConnectionReadTests { /** * Tests socket closure when mid-message - * + * * @throws Exception */ diff --git a/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/TcpNioConnectionTests.java b/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/TcpNioConnectionTests.java index 8b60b3e8ae..03d5f82c28 100644 --- a/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/TcpNioConnectionTests.java +++ b/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/TcpNioConnectionTests.java @@ -19,15 +19,19 @@ package org.springframework.integration.ip.tcp.connection; import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertTrue; import static org.junit.Assert.fail; +import static org.mockito.Mockito.doAnswer; import static org.mockito.Mockito.mock; import static org.mockito.Mockito.when; import java.io.IOException; import java.io.InputStream; import java.lang.reflect.Field; +import java.lang.reflect.InvocationTargetException; +import java.lang.reflect.Method; import java.net.ServerSocket; import java.net.Socket; import java.net.SocketTimeoutException; +import java.nio.ByteBuffer; import java.nio.channels.SelectionKey; import java.nio.channels.Selector; import java.nio.channels.SocketChannel; @@ -36,19 +40,30 @@ import java.util.HashMap; import java.util.HashSet; import java.util.List; import java.util.Map; +import java.util.concurrent.Callable; import java.util.concurrent.CountDownLatch; +import java.util.concurrent.ExecutionException; +import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; +import java.util.concurrent.Future; import java.util.concurrent.TimeUnit; import javax.net.ServerSocketFactory; import org.junit.Test; +import org.mockito.Mockito; +import org.mockito.invocation.InvocationOnMock; +import org.mockito.stubbing.Answer; +import org.springframework.integration.Message; +import org.springframework.integration.ip.tcp.serializer.ByteArrayCrLfSerializer; import org.springframework.integration.ip.util.SocketTestUtils; import org.springframework.integration.support.MessageBuilder; import org.springframework.integration.test.util.TestUtils; import org.springframework.util.ReflectionUtils; import org.springframework.util.ReflectionUtils.FieldCallback; import org.springframework.util.ReflectionUtils.FieldFilter; +import org.springframework.util.ReflectionUtils.MethodCallback; +import org.springframework.util.ReflectionUtils.MethodFilter; /** @@ -84,7 +99,7 @@ public class TcpNioConnectionTests { TcpConnection connection = factory.getConnection(); connection.send(MessageBuilder.withPayload(new byte[1000000]).build()); } catch (Exception e) { - assertTrue("Expected SocketTimeoutException, got " + e.getClass().getSimpleName() + + assertTrue("Expected SocketTimeoutException, got " + e.getClass().getSimpleName() + ":" + e.getMessage(), e instanceof SocketTimeoutException); } } @@ -102,7 +117,7 @@ public class TcpNioConnectionTests { ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(port); latch.countDown(); Socket socket = server.accept(); - byte[] b = new byte[6]; + byte[] b = new byte[6]; readFully(socket.getInputStream(), b); // block to cause timeout on read. server.accept(); @@ -127,7 +142,7 @@ public class TcpNioConnectionTests { fail("Unexpected exception " + e); } } - + @Test public void testMemoryLeak() throws Exception { final int port = SocketTestUtils.findAvailableServerSocket(); @@ -234,10 +249,108 @@ public class TcpNioConnectionTests { assertEquals(0, TestUtils.getPropertyValue(factory, "connections", List.class).size()); } + @Test + public void testInsufficientThreads() throws Exception { + final ExecutorService exec = Executors.newFixedThreadPool(2); + Future future = exec.submit(new Callable() { + public Object call() throws Exception { + SocketChannel channel = mock(SocketChannel.class); + Socket socket = mock(Socket.class); + Mockito.when(channel.socket()).thenReturn(socket); + doAnswer(new Answer() { + public Integer answer(InvocationOnMock invocation) throws Throwable { + ByteBuffer buffer = (ByteBuffer) invocation.getArguments()[0]; + buffer.position(1025); + return 1025; + } + }).when(channel).read(Mockito.any(ByteBuffer.class)); + final TcpNioConnection connection = new TcpNioConnection(channel, false, false); + connection.setTaskExecutor(exec); + connection.setPipeTimeout(200); + ReflectionUtils.doWithMethods(TcpNioConnection.class, new MethodCallback() { + public void doWith(Method method) throws IllegalArgumentException, IllegalAccessException { + method.setAccessible(true); + try { + method.invoke(connection, (Object[]) null); + } + catch (InvocationTargetException e) { + throw (RuntimeException) e.getCause(); + } + } + }, new MethodFilter() { + public boolean matches(Method method) { + return method.getName().equals("doRead"); + } + }); + return null; + } + }); + try { + Object o = future.get(10, TimeUnit.SECONDS); + fail("Expected exception, got " + o); + } + catch (ExecutionException e) { + assertEquals("Timed out writing to pipe, probably due to insufficient threads in " + + "a fixed thread pool; consider increasing this task executor pool size", e.getCause() + .getMessage()); + } + } + + @Test + public void testSufficientThreads() throws Exception { + final ExecutorService exec = Executors.newFixedThreadPool(3); + final CountDownLatch messageLatch = new CountDownLatch(1); + Future future = exec.submit(new Callable() { + public Object call() throws Exception { + SocketChannel channel = mock(SocketChannel.class); + Socket socket = mock(Socket.class); + Mockito.when(channel.socket()).thenReturn(socket); + doAnswer(new Answer() { + public Integer answer(InvocationOnMock invocation) throws Throwable { + ByteBuffer buffer = (ByteBuffer) invocation.getArguments()[0]; + buffer.position(1025); + buffer.put((byte) '\r'); + buffer.put((byte) '\n'); + return 1027; + } + }).when(channel).read(Mockito.any(ByteBuffer.class)); + final TcpNioConnection connection = new TcpNioConnection(channel, false, false); + connection.setTaskExecutor(exec); + connection.registerListener(new TcpListener(){ + public boolean onMessage(Message message) { + System.out.println(message); + messageLatch.countDown(); + return false; + } + }); + connection.setMapper(new TcpMessageMapper()); + connection.setDeserializer(new ByteArrayCrLfSerializer()); + ReflectionUtils.doWithMethods(TcpNioConnection.class, new MethodCallback() { + public void doWith(Method method) throws IllegalArgumentException, IllegalAccessException { + method.setAccessible(true); + try { + method.invoke(connection, (Object[]) null); + } + catch (InvocationTargetException e) { + throw (RuntimeException) e.getCause(); + } + } + }, new MethodFilter() { + public boolean matches(Method method) { + return method.getName().equals("doRead"); + } + }); + return null; + } + }); + future.get(60, TimeUnit.SECONDS); + assertTrue(messageLatch.await(10, TimeUnit.SECONDS)); + } + private void readFully(InputStream is, byte[] buff) throws IOException { for (int i = 0; i < buff.length; i++) { buff[i] = (byte) is.read(); } } - + } diff --git a/src/reference/docbook/ip.xml b/src/reference/docbook/ip.xml index 1f29dc42c1..ae961be75b 100644 --- a/src/reference/docbook/ip.xml +++ b/src/reference/docbook/ip.xml @@ -798,30 +798,30 @@ Pool Size - When using NIO, it is important to understand how threads are used, in order set the pool-size - appropriately. One thread from the pool is used to handle all socket events (e.g. data is - ready to be read). This thread is not available for other tasks. When data are ready to be - read, this thread dispatches the actual I/O to another thread from the pool, which reads - from the channel and writes the data to a temporary buffer; if this read is the start of - a new message, a third thread is used to read from that buffer to assemble the data into - a message. + The pool size attribute is no longer used; previously, it specified the size + of the default thread pool when a task-executor was not specified. It was also + used to set the connection backlog on server sockets. The first function is + no longer needed (see below); the second function is replaced by the + backlog attribute. - If there is not enough room in the temporary buffer to receive the newly read data, the - reader thread will block until the assembler thread consumes some data. If the pool is - exhausted, this will cause a deadlock, until another thread becomes available. The temporary - buffer is currently 1024 bytes. In the simplest case, with one connection, and data greater - than 1024 bytes, a pool-size of 2 will cause this deadlock to occur because a thread will - never be made available. - - - For this reason, when using NIO, the pool-size should be set to a minimum of 3. This does not - mean you have to reserve 2 threads for each socket because, aside from the selector thread, - the threads in the pool are shared across all the connections. The actual pool-size needed - will depend on a number of factors including the number of active connections, how - much utilization there is on those connections, and how long message processing takes when - a new message is received. + Previously, when using a fixed thread pool task executor (which was the default), with NIO, it + was possible to get a deadlock and processing would stop. The problem occurred when + a buffer was full, a thread reading from the socket was trying to add more data + to the buffer, and there were no threads available to make space in the buffer. + This only occurred with a very small pool size, but it could be possible under + extreme conditions. Since 2.2, two changes have eliminated this problem. First, + the default task executor is a cached thread pool executor. Second, deadlock + detection logic has been added such that if thread starvation occurs, instead of + deadlocking, an exception is thrown, thus releasing the deadlocked resources. + + Now that the default task executor is unbounded, it is possible that an out of + memory condition might occur with high rates of incoming messages, if message + processing takes extended time. If your application exhibits this type of + behavior, you are advised to use a pooled task executor with an appropriate + pool size. +
SSL/TLS Support @@ -1136,9 +1136,8 @@ Specifies a specific Executor to be used for socket handling. If not supplied, an internal - pooled executor will be used. Needed on some platforms that require the use of specific - task executors such as a WorkManagerTaskExecutor. See pool-size for thread - requirements, depending on other options. + cached thread executor will be used. Needed on some platforms that require the use of specific + task executors such as a WorkManagerTaskExecutor. single-use @@ -1150,16 +1149,20 @@ pool-size - Y + N + N + + This attribute is no longer used. For backward + compatibility, it sets the backlog but users should + use backlog to specify the + connection backlog in server factories + + + backlog + N Y - Specifies the concurrency. For tcp, not using nio, specifies the - number of concurrent connections supported by the adapter. For tcp, - using nio, it should be set to a minimum of 3; see 'Pool Size' in - . - It only applies in this sense if task-executor is not configured. - However, pool-size is also used for the server socket backlog, - regardless of whether an external task executor is used. Defaults to 5. + Sets the connection backlog for server factories. lookup-host