Spring/spring-integration
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

INT-2419 Thread Starvation Detection

Browse Source 
Previously, it was possible to run out of threads in
a fixed thread pool, and this could cause an indefinite
deadlock.

With this change, the deadlock is detected causing the
current message to fail, but freeing up the stuck
threads.

This was only seen with very small thread pools but
the potential was there for the condition to occur
under extreme conditions.

INT-2419 Deprecate poolSize Property

INT-2419 Polishing

Add note about OOM possibility now that an
unbounded task executor is used by default.
This commit is contained in:
Gary Russell
2012-05-22 20:24:31 -04:00
committed by Oleg Zhurakousky
parent 01225b8a48
commit 90fe92ae29
17 changed files with 420 additions and 183 deletions
Download Patch File Download Diff File Expand all files Collapse all files

18
spring-integration-ip/src/main/java/org/springframework/integration/ip/config/IpAdapterParserUtils.java
View File

@@ -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) {

17
spring-integration-ip/src/main/java/org/springframework/integration/ip/config/TcpConnectionFactoryFactoryBean.java
View File

@@ -84,7 +84,7 @@ public class TcpConnectionFactoryFactoryBean extends AbstractFactoryBean<Abstrac
private volatile boolean singleUse;
private volatile int poolSize = 5;
private volatile int backlog = 5;
private volatile TcpConnectionInterceptorFactoryChain interceptorFactoryChain;
@@ -156,7 +156,6 @@ public class TcpConnectionFactoryFactoryBean extends AbstractFactoryBean<Abstrac
factory.setLookupHost(this.lookupHost);
this.mapper.setApplySequence(this.applySequence);
factory.setMapper(this.mapper);
factory.setPoolSize(this.poolSize);
factory.setSerializer(this.serializer);
factory.setSingleUse(this.singleUse);
factory.setSoKeepAlive(this.soKeepAlive);
@@ -173,6 +172,7 @@ public class TcpConnectionFactoryFactoryBean extends AbstractFactoryBean<Abstrac
private void setServerAttributes(AbstractServerConnectionFactory factory) {
factory.setLocalAddress(this.localAddress);
factory.setBacklog(this.backlog);
}
private TcpSocketFactorySupport obtainSocketFactorySupport() {
@@ -349,9 +349,20 @@ public class TcpConnectionFactoryFactoryBean extends AbstractFactoryBean<Abstrac
/**
* @param poolSize
* @see org.springframework.integration.ip.tcp.connection.AbstractConnectionFactory#setPoolSize(int)
* @deprecated
*/
@Deprecated
public void setPoolSize(int poolSize) {
this.poolSize = poolSize;
logger.warn("poolSize is deprecated; use backlog instead");
this.backlog = poolSize;
}
/**
* @param backlog
* @see AbstractServerConnectionFactory#setBacklog(int)
*/
public void setBacklog(int backlog) {
this.backlog = backlog;
}
/**

34
spring-integration-ip/src/main/java/org/springframework/integration/ip/config/TcpConnectionParser.java
View File

@@ -30,51 +30,53 @@ import org.w3c.dom.Element;
*
*/
public class TcpConnectionParser extends AbstractBeanDefinitionParser {
@Override
protected AbstractBeanDefinition parseInternal(Element element,
ParserContext parserContext) {
BeanDefinitionBuilder builder = null;
String type = element.getAttribute(IpAdapterParserUtils.TCP_CONNECTION_TYPE);
if (!StringUtils.hasText(type)) {
parserContext.getReaderContext().error(IpAdapterParserUtils.TCP_CONNECTION_TYPE +
parserContext.getReaderContext().error(IpAdapterParserUtils.TCP_CONNECTION_TYPE +
" is required for a tcp connection", element);
} else if (!"server".equals(type) && !"client".equals(type)) {
parserContext.getReaderContext().error(IpAdapterParserUtils.TCP_CONNECTION_TYPE +
parserContext.getReaderContext().error(IpAdapterParserUtils.TCP_CONNECTION_TYPE +
" must be 'client' or 'server' for a TCP Connection Factory", element);
}
}
builder = BeanDefinitionBuilder.genericBeanDefinition(TcpConnectionFactoryFactoryBean.class);
IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "type");
IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element,
IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element,
IpAdapterParserUtils.HOST);
IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element,
IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element,
IpAdapterParserUtils.PORT);
IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element,
IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element,
IpAdapterParserUtils.USING_NIO);
IpAdapterParserUtils.addCommonSocketOptions(builder, element);
IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element,
IpAdapterParserUtils.RECEIVE_BUFFER_SIZE);
IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element,
IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element,
IpAdapterParserUtils.USING_DIRECT_BUFFERS);
IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element,
IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element,
IpAdapterParserUtils.SO_KEEP_ALIVE);
IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element,
IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element,
IpAdapterParserUtils.SO_LINGER);
IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element,
IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element,
IpAdapterParserUtils.SO_TCP_NODELAY);
IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element,
IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element,
IpAdapterParserUtils.SO_TRAFFIC_CLASS);
IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element,
IpAdapterParserUtils.POOL_SIZE);
IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element,
IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element,
IpAdapterParserUtils.BACKLOG);
IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element,
IpAdapterParserUtils.TASK_EXECUTOR);
IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element,
IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element,
IpAdapterParserUtils.SERIALIZER);
IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element,
IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element,
IpAdapterParserUtils.DESERIALIZER);
IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element,
IpAdapterParserUtils.SINGLE_USE);
IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element,
IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element,
IpAdapterParserUtils.INTERCEPTOR_FACTORY_CHAIN);
IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element,
IpAdapterParserUtils.LOOKUP_HOST);

18
spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/AbstractConnectionFactory.java
View File

@@ -92,8 +92,6 @@ public abstract class AbstractConnectionFactory extends IntegrationObjectSupport
private volatile boolean singleUse;
private volatile int poolSize = 5;
private volatile boolean active;
private volatile TcpConnectionInterceptorFactoryChain interceptorFactoryChain;
@@ -296,10 +294,14 @@ public abstract class AbstractConnectionFactory extends IntegrationObjectSupport
}
/**
* @deprecated This property is no longer used. If you wish
* to use a fixed thread pool, provide your own Executor
* in {@link #setTaskExecutor(Executor)}.
* @return the poolSize
*/
@Deprecated
public int getPoolSize() {
return poolSize;
return 0;
}
/**
@@ -372,8 +374,14 @@ public abstract class AbstractConnectionFactory extends IntegrationObjectSupport
}
/**
* @deprecated Default task executor is now a cached rather
* than a fixed pool executor. To use a pool, supply an
* appropriate Executor in {@link AbstractConnectionFactory#setTaskExecutor(Executor)}.
* Use {@link AbstractServerConnectionFactory#setBacklog(int)} to set the connection backlog.
*/
@Deprecated
public void setPoolSize(int poolSize) {
this.poolSize = poolSize;
}
public void setInterceptorFactoryChain(TcpConnectionInterceptorFactoryChain interceptorFactoryChain) {
@@ -438,7 +446,7 @@ public abstract class AbstractConnectionFactory extends IntegrationObjectSupport
}
if (this.taskExecutor == null) {
this.privateExecutor = true;
this.taskExecutor = Executors.newFixedThreadPool(this.poolSize);
this.taskExecutor = Executors.newCachedThreadPool();
}
return this.taskExecutor;
}

40
spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/AbstractServerConnectionFactory.java
View File

@@ -20,6 +20,8 @@ import java.net.ServerSocket;
import java.net.Socket;
import java.net.SocketException;
import org.springframework.util.Assert;
/**
* Base class for all server connection factories. Server connection factories
* listen on a port for incoming connections and create new TcpConnection objects
@@ -30,9 +32,13 @@ import java.net.SocketException;
*/
public abstract class AbstractServerConnectionFactory extends AbstractConnectionFactory {
private boolean listening;
private static final int DEFAULT_BACKLOG = 5;
private String localAddress;
private volatile boolean listening;
private volatile String localAddress;
private volatile int backlog = DEFAULT_BACKLOG;
/**
@@ -120,4 +126,34 @@ public abstract class AbstractServerConnectionFactory extends AbstractConnection
public void setLocalAddress(String localAddress) {
this.localAddress = localAddress;
}
/**
* The number of sockets in the server connection backlog.
* @return The backlog.
*/
public int getBacklog() {
return backlog;
}
/**
* The number of sockets in the connection backlog. Default 5;
* increase if you expect high connection rates.
* @param backlog
*/
public void setBacklog(int backlog) {
Assert.isTrue(backlog >= 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);
}
}

11
spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNetServerConnectionFactory.java
View File

@@ -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;
}
}

41
spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNioConnection.java
View File

@@ -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();

13
spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNioServerConnectionFactory.java
View File

@@ -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();

24
spring-integration-ip/src/main/resources/org/springframework/integration/ip/config/spring-integration-ip-2.2.xsd
View File

@@ -456,13 +456,29 @@ apply to TCP outbound adapters and gateways.
</xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="task-executor" type="xsd:string" />
<xsd:attribute name="task-executor" type="xsd:string">
<xsd:annotation>
<xsd:documentation>
A task executor for managing connections; if not specified a
cached thread pool task executor is used.
</xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="pool-size" type="xsd:string" >
<xsd:annotation>
<xsd:documentation>
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.
</xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="backlog" type="xsd:string" >
<xsd:annotation>
<xsd:documentation>
Specifies the connection backlog for server sockets. Does not
apply to client factories.
</xsd:documentation>
</xsd:annotation>
</xsd:attribute>

13
spring-integration-ip/src/test/java/org/springframework/integration/ip/config/ParserUnitTests-context.xml
View File

@@ -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"
/>
<ip:tcp-connection-factory
id="serverBackwardsCompatible"
type="server"
port="#{client1.port}"
pool-size="123"
/>
<ip:tcp-connection-factory
id="client2"
type="client"
@@ -254,7 +260,6 @@
using-nio="false"
single-use="true"
task-executor="externalTE"
pool-size="321"
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"
/>

16
spring-integration-ip/src/test/java/org/springframework/integration/ip/config/ParserUnitTests.java
View File

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

16
spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/TcpOutboundGatewayTests.java
View File

@@ -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<String> replies = new HashSet<String>();
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<String> replies = new HashSet<String>();
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<String> replies = new HashSet<String>();
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());

50
spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/TcpReceivingChannelAdapterTests.java
View File

@@ -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<String> results = new HashSet<String>();
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<String> results = new HashSet<String>();
@@ -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());
}

41
spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/TcpSendingMessageHandlerTests.java
View File

@@ -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<String> replies = new HashSet<String>();
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<String> replies = new HashSet<String>();
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);
}

63
spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/TcpNioConnectionReadTests.java
View File

@@ -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<byte[]>) responses.get(0)).getPayload()));
assertEquals("Data", SocketTestUtils.TEST_STRING + SocketTestUtils.TEST_STRING,
assertEquals("Data", SocketTestUtils.TEST_STRING + SocketTestUtils.TEST_STRING,
new String(((Message<byte[]>) 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<byte[]>) 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<byte[]>) responses.get(0)).getPayload()));
assertEquals("Data", SocketTestUtils.TEST_STRING + SocketTestUtils.TEST_STRING,
assertEquals("Data", SocketTestUtils.TEST_STRING + SocketTestUtils.TEST_STRING,
new String(((Message<byte[]>) 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<byte[]>) responses.get(0)).getPayload()));
assertEquals("Data", SocketTestUtils.TEST_STRING + SocketTestUtils.TEST_STRING,
assertEquals("Data", SocketTestUtils.TEST_STRING + SocketTestUtils.TEST_STRING,
new String(((Message<byte[]>) 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
*/

121
spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/TcpNioConnectionTests.java
View File

@@ -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<Object> future = exec.submit(new Callable<Object>() {
public Object call() throws Exception {
SocketChannel channel = mock(SocketChannel.class);
Socket socket = mock(Socket.class);
Mockito.when(channel.socket()).thenReturn(socket);
doAnswer(new Answer<Integer>() {
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<Object> future = exec.submit(new Callable<Object>() {
public Object call() throws Exception {
SocketChannel channel = mock(SocketChannel.class);
Socket socket = mock(Socket.class);
Mockito.when(channel.socket()).thenReturn(socket);
doAnswer(new Answer<Integer>() {
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();
}
}
}

67
src/reference/docbook/ip.xml
View File

@@ -798,30 +798,30 @@
</para>
<para><emphasis>Pool Size</emphasis></para>
<para>
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
<emphasis>backlog</emphasis> attribute.
</para>
<para>
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.
</para>
<para>
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.
</para>
<note>
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.
</note>
</section>
<section id="ssl-tls">
<title>SSL/TLS Support</title>
@@ -1136,9 +1136,8 @@
<entry></entry>
<entry>
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.</entry>
cached thread executor will be used. Needed on some platforms that require the use of specific
task executors such as a WorkManagerTaskExecutor.</entry>
</row>
<row>
<entry>single-use</entry>
@@ -1150,16 +1149,20 @@
</row>
<row>
<entry>pool-size</entry>
<entry>Y</entry>
<entry>N</entry>
<entry>N</entry>
<entry></entry>
<entry>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</entry>
</row>
<row>
<entry>backlog</entry>
<entry>N</entry>
<entry>Y</entry>
<entry></entry>
<entry>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
<xref linkend="note_nio" />.
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.</entry>
<entry>Sets the connection backlog for server factories.</entry>
</row>
<row>
<entry>lookup-host</entry>