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

INT-3233 NIO, Windows 7 and Java 7/8

Browse Source 
JIRA: https://jira.springsource.org/browse/INT-3233

With Windows 7 and Java 7/8 closing a server channel does
not close the underlying socket. Closing the Selector does
close the socket.

It is not clear why this is needed, given that a selector can
be used for multiple sockets.

However, we only use the server side selector for a single
server socket so there is no detriment in closing it.

Also close the selector on the client side (even though it is
used for multiple sockets, because we are stopping the factory
anyway and all sockets will be closed).

However, closing the selector opens us up to 'ClosedSelectorException's
in several places. Add catch blocks to deal with this exception, and only
log an error if the factory is active.

While debugging this issue, I found that a number of (older) tests
left threads running, sockets open etc.

INT-3233 Polishing - PR Comments
This commit is contained in:
Gary Russell
2013-12-13 20:49:54 +02:00
committed by Artem Bilan
parent 50af153298
commit ced1a05df2
25 changed files with 611 additions and 190 deletions
Download Patch File Download Diff File Expand all files Collapse all files

10
build.gradle
View File

@@ -295,9 +295,11 @@ project('spring-integration-http') {
compile ("net.java.dev.rome:rome:1.0.0", optional)
testCompile project(":spring-integration-test")
// suppress deprecation warnings (@SuppressWarnings("deprecation") is not enough for javac)
compileJava.options.compilerArgs = ["${xLintArg},-deprecation"]
}
// suppress deprecation warnings (@SuppressWarnings("deprecation") is not enough for javac)
compileJava.options.compilerArgs = ["${xLintArg},-deprecation"]
}
project('spring-integration-ip') {
@@ -308,6 +310,10 @@ project('spring-integration-ip') {
runtime project(":spring-integration-stream")
testCompile project(":spring-integration-test")
}
// suppress deprecation warnings (@SuppressWarnings("deprecation") is not enough for javac)
compileJava.options.compilerArgs = ["${xLintArg},-deprecation"]
}
project('spring-integration-jdbc') {

16
spring-integration-ip/src/main/java/org/springframework/integration/ip/AbstractInternetProtocolReceivingChannelAdapter.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2010 the original author or authors.
* Copyright 2002-2013 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.
@@ -17,10 +17,12 @@
package org.springframework.integration.ip;
import java.util.concurrent.Executor;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.ThreadFactory;
import org.springframework.integration.endpoint.MessageProducerSupport;
import org.springframework.util.Assert;
/**
* Base class for inbound TCP/UDP Channel Adapters.
@@ -48,6 +50,8 @@ public abstract class AbstractInternetProtocolReceivingChannelAdapter
private volatile Executor taskExecutor;
private volatile boolean taskExecutorSet;
private volatile int poolSize = 5;
@@ -63,6 +67,7 @@ public abstract class AbstractInternetProtocolReceivingChannelAdapter
return port;
}
@Override
public void setSoTimeout(int soTimeout) {
this.soTimeout = soTimeout;
}
@@ -74,6 +79,7 @@ public abstract class AbstractInternetProtocolReceivingChannelAdapter
return soTimeout;
}
@Override
public void setSoReceiveBufferSize(int soReceiveBufferSize) {
this.soReceiveBufferSize = soReceiveBufferSize;
}
@@ -117,6 +123,7 @@ public abstract class AbstractInternetProtocolReceivingChannelAdapter
protected void checkTaskExecutor(final String threadName) {
if (this.active && this.taskExecutor == null) {
Executor executor = Executors.newFixedThreadPool(this.poolSize, new ThreadFactory() {
@Override
public Thread newThread(Runnable runner) {
Thread thread = new Thread(runner);
thread.setName(threadName);
@@ -131,6 +138,10 @@ public abstract class AbstractInternetProtocolReceivingChannelAdapter
@Override
protected void doStop() {
this.active = false;
if (!this.taskExecutorSet && this.taskExecutor != null) {
((ExecutorService) this.taskExecutor).shutdown();
this.taskExecutor = null;
}
}
public boolean isListening() {
@@ -148,6 +159,7 @@ public abstract class AbstractInternetProtocolReceivingChannelAdapter
return localAddress;
}
@Override
public void setLocalAddress(String localAddress) {
this.localAddress = localAddress;
}
@@ -157,7 +169,9 @@ public abstract class AbstractInternetProtocolReceivingChannelAdapter
}
public void setTaskExecutor(Executor taskExecutor) {
Assert.notNull(taskExecutor, "'taskExecutor' cannot be null");
this.taskExecutor = taskExecutor;
this.taskExecutorSet = true;
}
/**

38
spring-integration-ip/src/main/java/org/springframework/integration/ip/AbstractInternetProtocolSendingMessageHandler.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2010 the original author or authors.
* Copyright 2002-2013 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.
@@ -21,6 +21,7 @@ import java.net.InetSocketAddress;
import java.net.Socket;
import java.net.SocketAddress;
import org.springframework.context.Lifecycle;
import org.springframework.integration.handler.AbstractMessageHandler;
import org.springframework.util.Assert;
@@ -30,7 +31,8 @@ import org.springframework.util.Assert;
* @author Gary Russell
* @since 2.0
*/
public abstract class AbstractInternetProtocolSendingMessageHandler extends AbstractMessageHandler implements CommonSocketOptions {
public abstract class AbstractInternetProtocolSendingMessageHandler extends AbstractMessageHandler implements CommonSocketOptions,
Lifecycle {
private final SocketAddress destinationAddress;
@@ -42,6 +44,8 @@ public abstract class AbstractInternetProtocolSendingMessageHandler extends Abst
private volatile int soTimeout = -1;
private volatile boolean running;
public AbstractInternetProtocolSendingMessageHandler(String host, int port) {
Assert.notNull(host, "host must not be null");
this.destinationAddress = new InetSocketAddress(host, port);
@@ -55,6 +59,7 @@ public abstract class AbstractInternetProtocolSendingMessageHandler extends Abst
* @see DatagramSocket#setSoTimeout(int)
* @param timeout
*/
@Override
public void setSoTimeout(int timeout) {
this.soTimeout = timeout;
}
@@ -64,6 +69,7 @@ public abstract class AbstractInternetProtocolSendingMessageHandler extends Abst
* @see DatagramSocket#setReceiveBufferSize(int)
* @param size
*/
@Override
public void setSoReceiveBufferSize(int size) {
}
@@ -72,6 +78,7 @@ public abstract class AbstractInternetProtocolSendingMessageHandler extends Abst
* @see DatagramSocket#setSendBufferSize(int)
* @param size
*/
@Override
public void setSoSendBufferSize(int size) {
this.soSendBufferSize = size;
}
@@ -115,4 +122,31 @@ public abstract class AbstractInternetProtocolSendingMessageHandler extends Abst
return soSendBufferSize;
}
@Override
public synchronized void start() {
if (!this.running) {
this.doStart();
this.running = true;
}
}
protected abstract void doStart();
@Override
public synchronized void stop() {
if (this.running) {
this.doStop();
this.running = false;
}
}
protected abstract void doStop();
@Override
public boolean isRunning() {
return this.running;
}
}

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

@@ -410,8 +410,10 @@ public abstract class AbstractConnectionFactory extends IntegrationObjectSupport
}
/**
* Closes the server.
* Closes the factory.
* @deprecated As of 3.0; use {@link #stop()}.
*/
@Deprecated
public abstract void close();
@Override

5
spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/CachingClientConnectionFactory.java
View File

@@ -44,6 +44,7 @@ public class CachingClientConnectionFactory extends AbstractClientConnectionFact
this.targetConnectionFactory = target;
pool = new SimplePool<TcpConnectionSupport>(poolSize, new SimplePool.PoolItemCallback<TcpConnectionSupport>() {
@Override
public TcpConnectionSupport createForPool() {
try {
return targetConnectionFactory.getConnection();
@@ -52,10 +53,12 @@ public class CachingClientConnectionFactory extends AbstractClientConnectionFact
}
}
@Override
public boolean isStale(TcpConnectionSupport connection) {
return !connection.isOpen();
}
@Override
public void removedFromPool(TcpConnectionSupport connection) {
connection.close();
}
@@ -167,6 +170,7 @@ public class CachingClientConnectionFactory extends AbstractClientConnectionFact
return targetConnectionFactory.isRunning();
}
@SuppressWarnings("deprecation")
@Override
public void close() {
targetConnectionFactory.close();
@@ -317,6 +321,7 @@ public class CachingClientConnectionFactory extends AbstractClientConnectionFact
public void registerListener(TcpListener listener) {
super.registerListener(listener);
targetConnectionFactory.registerListener(new TcpListener() {
@Override
public boolean onMessage(Message<?> message) {
if (!(message instanceof ErrorMessage)) {
throw new UnsupportedOperationException("This should never be called");

9
spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/FailoverClientConnectionFactory.java
View File

@@ -82,6 +82,7 @@ public class FailoverClientConnectionFactory extends AbstractClientConnectionFac
super.registerListener(listener);
for (AbstractClientConnectionFactory factory : this.factories) {
factory.registerListener(new TcpListener() {
@Override
public boolean onMessage(Message<?> message) {
if (!(message instanceof ErrorMessage)) {
throw new UnsupportedOperationException("This should never be called");
@@ -112,6 +113,7 @@ public class FailoverClientConnectionFactory extends AbstractClientConnectionFac
return failoverTcpConnection;
}
@SuppressWarnings("deprecation")
@Override
public void close() {
for (AbstractClientConnectionFactory factory : this.factories) {
@@ -229,6 +231,7 @@ public class FailoverClientConnectionFactory extends AbstractClientConnectionFac
this.open = false;
}
@Override
public boolean isOpen() {
return this.open;
}
@@ -238,6 +241,7 @@ public class FailoverClientConnectionFactory extends AbstractClientConnectionFac
* send to a new connection obtained from {@link #findAConnection()}.
* If send fails on a connection from every factory, we give up.
*/
@Override
public synchronized void send(Message<?> message) throws Exception {
boolean success = false;
AbstractClientConnectionFactory lastFactoryToTry = this.currentFactory;
@@ -268,10 +272,12 @@ public class FailoverClientConnectionFactory extends AbstractClientConnectionFac
}
}
@Override
public Object getPayload() throws Exception {
return this.delegate.getPayload();
}
@Override
public void run() {
throw new UnsupportedOperationException("Not supported on FailoverTcpConnection");
}
@@ -286,10 +292,12 @@ public class FailoverClientConnectionFactory extends AbstractClientConnectionFac
return this.delegate.getHostAddress();
}
@Override
public int getPort() {
return this.delegate.getPort();
}
@Override
public Object getDeserializerStateKey() {
return this.delegate.getDeserializerStateKey();
}
@@ -350,6 +358,7 @@ public class FailoverClientConnectionFactory extends AbstractClientConnectionFac
* the actual connectionId in another header for convenience and tracing
* purposes.
*/
@Override
public boolean onMessage(Message<?> message) {
if (this.delegate.getConnectionId().equals(message.getHeaders().get(IpHeaders.CONNECTION_ID))) {
MessageBuilder<?> messageBuilder = MessageBuilder.fromMessage(message)

24
spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNioClientConnectionFactory.java
View File

@@ -21,6 +21,7 @@ import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.CancelledKeyException;
import java.nio.channels.ClosedChannelException;
import java.nio.channels.ClosedSelectorException;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.SocketChannel;
@@ -111,10 +112,16 @@ public class TcpNioClientConnectionFactory extends
this.tcpNioConnectionSupport = tcpNioSupport;
}
@Deprecated
@Override
public void close() {
if (this.selector != null) {
this.selector.wakeup();
try {
this.selector.close();
}
catch (IOException e) {
logger.error("Error closing selector", e);
}
}
}
@@ -129,6 +136,7 @@ public class TcpNioClientConnectionFactory extends
super.start();
}
@Override
public void run() {
if (logger.isDebugEnabled()) {
logger.debug("Read selector running for connections to " + this.getHost() + ":" + this.getPort());
@@ -141,7 +149,8 @@ public class TcpNioClientConnectionFactory extends
int selectionCount = 0;
try {
selectionCount = selector.select(soTimeout < 0 ? 0 : soTimeout);
} catch (CancelledKeyException cke) {
}
catch (CancelledKeyException cke) {
if (logger.isDebugEnabled()) {
logger.debug("CancelledKeyException during Selector.select()");
}
@@ -149,7 +158,8 @@ public class TcpNioClientConnectionFactory extends
while ((newChannel = newChannels.poll()) != null) {
try {
newChannel.register(this.selector, SelectionKey.OP_READ, channelMap.get(newChannel));
} catch (ClosedChannelException cce) {
}
catch (ClosedChannelException cce) {
if (logger.isDebugEnabled()) {
logger.debug("Channel closed before registering with selector for reading");
}
@@ -157,7 +167,13 @@ public class TcpNioClientConnectionFactory extends
}
this.processNioSelections(selectionCount, selector, null, this.channelMap);
}
} catch (Exception e) {
}
catch (ClosedSelectorException cse) {
if (this.isActive()) {
logger.error("Selector closed", cse);
}
}
catch (Exception e) {
logger.error("Exception in read selector thread", e);
this.setActive(false);
}

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

@@ -23,6 +23,7 @@ import java.net.Socket;
import java.net.SocketException;
import java.nio.channels.CancelledKeyException;
import java.nio.channels.ClosedChannelException;
import java.nio.channels.ClosedSelectorException;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.ServerSocketChannel;
@@ -67,6 +68,7 @@ public class TcpNioServerConnectionFactory extends AbstractServerConnectionFacto
* connection {@link TcpConnection#run()} using the task executor.
* I/O errors on the server socket/channel are logged and the factory is stopped.
*/
@Override
public void run() {
if (this.getListener() == null) {
logger.info("No listener bound to server connection factory; will not read; exiting...");
@@ -83,7 +85,8 @@ public class TcpNioServerConnectionFactory extends AbstractServerConnectionFacto
if (this.getLocalAddress() == null) {
this.serverChannel.socket().bind(new InetSocketAddress(port),
Math.abs(this.getBacklog()));
} else {
}
else {
InetAddress whichNic = InetAddress.getByName(this.getLocalAddress());
this.serverChannel.socket().bind(new InetSocketAddress(whichNic, port),
Math.abs(this.getBacklog()));
@@ -94,7 +97,8 @@ public class TcpNioServerConnectionFactory extends AbstractServerConnectionFacto
this.selector = selector;
doSelect(this.serverChannel, selector);
} catch (IOException e) {
}
catch (IOException e) {
this.close();
if (this.isActive()) {
logger.error("Error on ServerSocketChannel", e);
@@ -127,12 +131,19 @@ public class TcpNioServerConnectionFactory extends AbstractServerConnectionFacto
int selectionCount = 0;
try {
selectionCount = selector.select(soTimeout < 0 ? 0 : soTimeout);
} catch (CancelledKeyException cke) {
this.processNioSelections(selectionCount, selector, server, this.channelMap);
}
catch (CancelledKeyException cke) {
if (logger.isDebugEnabled()) {
logger.debug("CancelledKeyException during Selector.select()");
}
}
this.processNioSelections(selectionCount, selector, server, this.channelMap);
catch (ClosedSelectorException cse) {
if (this.isActive()) {
logger.error("Selector closed", cse);
break;
}
}
}
}
@@ -195,14 +206,20 @@ public class TcpNioServerConnectionFactory extends AbstractServerConnectionFacto
@Override
public void close() {
if (this.selector != null) {
this.selector.wakeup();
try {
this.selector.close();
}
catch (IOException e) {
logger.error("Error closing selector", e);
}
}
if (this.serverChannel == null) {
return;
}
try {
this.serverChannel.close();
} catch (IOException e) {}
}
catch (IOException e) {}
this.serverChannel = null;
}

7
spring-integration-ip/src/main/java/org/springframework/integration/ip/udp/UnicastReceivingChannelAdapter.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2010 the original author or authors.
* Copyright 2002-2013 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.
@@ -73,6 +73,7 @@ public class UnicastReceivingChannelAdapter extends AbstractInternetProtocolRece
}
@Override
public void run() {
if (logger.isDebugEnabled()) {
logger.debug("UDP Receiver running on port:" + this.getPort());
@@ -90,7 +91,7 @@ public class UnicastReceivingChannelAdapter extends AbstractInternetProtocolRece
// continue
}
catch (SocketException e) {
doStop();
this.stop();
}
catch (Exception e) {
if (e instanceof MessagingException) {
@@ -133,6 +134,7 @@ public class UnicastReceivingChannelAdapter extends AbstractInternetProtocolRece
protected boolean asyncSendMessage(final DatagramPacket packet) {
this.getTaskExecutor().execute(new Runnable(){
@Override
public void run() {
Message<byte[]> message = null;
try {
@@ -222,6 +224,7 @@ public class UnicastReceivingChannelAdapter extends AbstractInternetProtocolRece
}
}
@Override
public void setSoSendBufferSize(int soSendBufferSize) {
this.soSendBufferSize = soSendBufferSize;
}

38
spring-integration-ip/src/main/java/org/springframework/integration/ip/udp/UnicastSendingMessageHandler.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2001-2011 the original author or authors.
* Copyright 2001-2013 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,6 +26,7 @@ import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.Executor;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.TimeUnit;
@@ -51,7 +52,7 @@ import org.springframework.util.Assert;
* @since 2.0
*/
public class UnicastSendingMessageHandler extends
AbstractInternetProtocolSendingMessageHandler implements Runnable{
AbstractInternetProtocolSendingMessageHandler implements Runnable {
private final DatagramPacketMessageMapper mapper = new DatagramPacketMessageMapper();
@@ -86,6 +87,8 @@ public class UnicastSendingMessageHandler extends
private volatile Executor taskExecutor;
private volatile boolean taskExecutorSet;
/**
* Basic constructor; no reliability; no acknowledgment.
* @param host Destination host.
@@ -167,12 +170,14 @@ public class UnicastSendingMessageHandler extends
}
}
public void onInit() {
@Override
public void doStart() {
if (this.acknowledge) {
if (this.taskExecutor == null) {
Executor executor = Executors
.newSingleThreadExecutor(new ThreadFactory() {
private AtomicInteger n = new AtomicInteger();
private final AtomicInteger n = new AtomicInteger();
@Override
public Thread newThread(Runnable runner) {
Thread thread = new Thread(runner);
thread.setName("UDP-Ack-Handler-" + n.getAndIncrement());
@@ -185,10 +190,21 @@ public class UnicastSendingMessageHandler extends
}
}
@Override
protected void doStop() {
this.closeSocketIfNeeded();
if (!this.taskExecutorSet && this.taskExecutor != null) {
((ExecutorService) this.taskExecutor).shutdown();
this.taskExecutor = null;
}
}
@Override
public void handleMessageInternal(Message<?> message)
throws MessageRejectedException, MessageHandlingException,
MessageDeliveryException {
if (this.acknowledge) {
Assert.state(this.isRunning(), "When 'acknowlege' is enabled, adapter must be running");
if (!this.ackThreadRunning) {
synchronized(this) {
if (!this.ackThreadRunning) {
@@ -294,6 +310,7 @@ public class UnicastSendingMessageHandler extends
/**
* Process acknowledgments, if requested.
*/
@Override
public void run() {
try {
this.ackThreadRunning = true;
@@ -333,7 +350,15 @@ public class UnicastSendingMessageHandler extends
this.taskExecutor.execute(this);
}
/**
* @deprecated Use stop() instead.
*/
@Deprecated
public void shutDown() {
this.stop();
}
private void closeSocketIfNeeded() {
if (socket != null) {
socket.close();
socket = null;
@@ -344,16 +369,20 @@ public class UnicastSendingMessageHandler extends
* @see java.net.Socket#setReceiveBufferSize(int)
* @see DatagramSocket#setReceiveBufferSize(int)
*/
@Override
public void setSoReceiveBufferSize(int size) {
this.soReceiveBufferSize = size;
}
@Override
public void setLocalAddress(String localAddress) {
this.localAddress = localAddress;
}
public void setTaskExecutor(Executor taskExecutor) {
Assert.notNull(taskExecutor, "'taskExecutor' cannot be null");
this.taskExecutor = taskExecutor;
this.taskExecutorSet = true;
}
/**
@@ -363,6 +392,7 @@ public class UnicastSendingMessageHandler extends
this.ackCounter = ackCounter;
}
@Override
public String getComponentType(){
return "ip:udp-outbound-channel-adapter";
}

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

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2012 the original author or authors.
* Copyright 2002-2013 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.
@@ -86,10 +86,13 @@ public class TcpOutboundGatewayTests {
AbstractConnectionFactory ccf = new TcpNetClientConnectionFactory("localhost", port);
final CountDownLatch latch = new CountDownLatch(1);
final AtomicBoolean done = new AtomicBoolean();
final AtomicReference<ServerSocket> serverSocket = new AtomicReference<ServerSocket>();
Executors.newSingleThreadExecutor().execute(new Runnable() {
@Override
public void run() {
try {
ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(port, 100);
serverSocket.set(server);
latch.countDown();
List<Socket> sockets = new ArrayList<Socket>();
int i = 0;
@@ -101,7 +104,8 @@ public class TcpOutboundGatewayTests {
oos.writeObject("Reply" + (i++));
sockets.add(socket);
}
} catch (Exception e) {
}
catch (Exception e) {
if (!done.get()) {
e.printStackTrace();
}
@@ -141,6 +145,8 @@ public class TcpOutboundGatewayTests {
for (int i = 0; i < 100; i++) {
assertTrue(replies.remove("Reply" + i));
}
done.set(true);
serverSocket.get().close();
}
@Test
@@ -149,6 +155,7 @@ public class TcpOutboundGatewayTests {
final CountDownLatch latch = new CountDownLatch(1);
final AtomicBoolean done = new AtomicBoolean();
Executors.newSingleThreadExecutor().execute(new Runnable() {
@Override
public void run() {
try {
ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(port, 10);
@@ -202,6 +209,7 @@ public class TcpOutboundGatewayTests {
final CountDownLatch latch = new CountDownLatch(1);
final AtomicBoolean done = new AtomicBoolean();
Executors.newSingleThreadExecutor().execute(new Runnable() {
@Override
public void run() {
try {
ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(port);
@@ -240,6 +248,7 @@ public class TcpOutboundGatewayTests {
for (int i = 0; i < 2; i++) {
final int j = i;
results[j] = (Executors.newSingleThreadExecutor().submit(new Callable<Integer>(){
@Override
public Integer call() throws Exception {
gateway.handleMessage(MessageBuilder.withPayload("Test" + j).build());
return 0;
@@ -319,6 +328,7 @@ public class TcpOutboundGatewayTests {
Executors.newSingleThreadExecutor().execute(new Runnable() {
@Override
public void run() {
try {
ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(port);
@@ -368,6 +378,7 @@ public class TcpOutboundGatewayTests {
for (int i = 0; i < 2; i++) {
final int j = i;
results[j] = (Executors.newSingleThreadExecutor().submit(new Callable<Integer>() {
@Override
public Integer call() throws Exception {
// increase the timeout after the first send
if (j > 0) {
@@ -419,6 +430,7 @@ public class TcpOutboundGatewayTests {
Executors.newSingleThreadExecutor().execute(new Runnable() {
@Override
public void run() {
try {
ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(port);
@@ -498,6 +510,7 @@ public class TcpOutboundGatewayTests {
Executors.newSingleThreadExecutor().execute(new Runnable() {
@Override
public void run() {
try {
ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(port);
@@ -633,14 +646,13 @@ public class TcpOutboundGatewayTests {
Executors.newSingleThreadExecutor().execute(new Runnable() {
@Override
public void run() {
try {
ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(port);
latch.countDown();
int i = 0;
while (!done.get()) {
Socket socket = server.accept();
i++;
while (!socket.isClosed()) {
try {
ObjectInputStream ois = new ObjectInputStream(socket.getInputStream());

5
spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/ConnectionFactoryShutDownTests.java
View File

@@ -22,6 +22,7 @@ import java.util.concurrent.Executor;
import java.util.concurrent.TimeUnit;
import org.junit.Test;
import org.springframework.integration.MessagingException;
import org.springframework.util.StopWatch;
@@ -36,13 +37,16 @@ public class ConnectionFactoryShutDownTests {
public void testShutdownDoesntDeadlock() throws Exception {
final AbstractConnectionFactory factory = new AbstractConnectionFactory(0) {
@Override
public TcpConnection getConnection() throws Exception {
return null;
}
@Override
@Deprecated
public void close() {
}
};
factory.setActive(true);
Executor executor = factory.getTaskExecutor();
@@ -50,6 +54,7 @@ public class ConnectionFactoryShutDownTests {
final CountDownLatch latch2 = new CountDownLatch(1);
executor.execute(new Runnable() {
@Override
public void run() {
latch1.countDown();
try {

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

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2010 the original author or authors.
* Copyright 2002-2013 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.
@@ -20,9 +20,7 @@ import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertTrue;
import static org.junit.Assert.fail;
import java.io.IOException;
import java.net.Socket;
import java.net.UnknownHostException;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.CountDownLatch;
@@ -32,6 +30,7 @@ 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,7 +46,7 @@ import org.springframework.integration.test.util.SocketUtils;
*/
public class TcpNioConnectionReadTests {
private CountDownLatch latch = new CountDownLatch(1);
private final CountDownLatch latch = new CountDownLatch(1);
private AbstractServerConnectionFactory getConnectionFactory(int port,
AbstractByteArraySerializer serializer, TcpListener listener) throws Exception {
@@ -68,10 +67,6 @@ public class TcpNioConnectionReadTests {
return scf;
}
/**
* Test method for {@link org.springframework.integration.ip.tcp.NioSocketReader}.
*/
@SuppressWarnings("unchecked")
@Test
public void testReadLength() throws Exception {
int port = SocketUtils.findAvailableServerSocket();
@@ -79,6 +74,7 @@ public class TcpNioConnectionReadTests {
final List<Message<?>> responses = new ArrayList<Message<?>>();
final Semaphore semaphore = new Semaphore(0);
AbstractServerConnectionFactory scf = getConnectionFactory(port, serializer,new TcpListener() {
@Override
public boolean onMessage(Message<?> message) {
responses.add(message);
semaphore.release();
@@ -88,7 +84,7 @@ public class TcpNioConnectionReadTests {
// Fire up the sender.
SocketTestUtils.testSendLength(port, latch);
CountDownLatch done = SocketTestUtils.testSendLength(port, latch);
latch.countDown();
assertTrue(semaphore.tryAcquire(1, 10000, TimeUnit.MILLISECONDS));
assertTrue(semaphore.tryAcquire(1, 10000, TimeUnit.MILLISECONDS));
@@ -97,7 +93,8 @@ public class TcpNioConnectionReadTests {
new String((byte[]) responses.get(0).getPayload()));
assertEquals("Data", SocketTestUtils.TEST_STRING + SocketTestUtils.TEST_STRING,
new String((byte[]) responses.get(1).getPayload()));
scf.close();
scf.stop();
done.countDown();
}
@@ -110,11 +107,15 @@ public class TcpNioConnectionReadTests {
final List<Message<?>> responses = new ArrayList<Message<?>>();
final Semaphore semaphore = new Semaphore(0);
AbstractServerConnectionFactory scf = getConnectionFactory(port, serializer,new TcpListener() {
@Override
public boolean onMessage(Message<?> message) {
responses.add(message);
try {
Thread.sleep(1000);
} catch (InterruptedException e) { }
}
catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
semaphore.release();
return false;
}
@@ -123,19 +124,17 @@ public class TcpNioConnectionReadTests {
int howMany = 2;
scf.setBacklog(howMany + 5);
// Fire up the sender.
SocketTestUtils.testSendFragmented(port, howMany, false);
CountDownLatch done = 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",
new String(((Message<byte[]>) responses.get(0)).getPayload()));
}
scf.close();
scf.stop();
done.countDown();
}
/**
* Test method for {@link org.springframework.integration.ip.tcp.NioSocketReader}.
*/
@SuppressWarnings("unchecked")
@Test
public void testReadStxEtx() throws Exception {
@@ -144,6 +143,7 @@ public class TcpNioConnectionReadTests {
final List<Message<?>> responses = new ArrayList<Message<?>>();
final Semaphore semaphore = new Semaphore(0);
AbstractServerConnectionFactory scf = getConnectionFactory(port, serializer,new TcpListener() {
@Override
public boolean onMessage(Message<?> message) {
responses.add(message);
semaphore.release();
@@ -153,7 +153,7 @@ public class TcpNioConnectionReadTests {
// Fire up the sender.
SocketTestUtils.testSendStxEtx(port, latch);
CountDownLatch done = SocketTestUtils.testSendStxEtx(port, latch);
latch.countDown();
assertTrue(semaphore.tryAcquire(1, 10000, TimeUnit.MILLISECONDS));
assertTrue(semaphore.tryAcquire(1, 10000, TimeUnit.MILLISECONDS));
@@ -162,7 +162,8 @@ public class TcpNioConnectionReadTests {
new String(((Message<byte[]>) responses.get(0)).getPayload()));
assertEquals("Data", SocketTestUtils.TEST_STRING + SocketTestUtils.TEST_STRING,
new String(((Message<byte[]>) responses.get(1)).getPayload()));
scf.close();
scf.stop();
done.countDown();
}
/**
@@ -176,6 +177,7 @@ public class TcpNioConnectionReadTests {
final List<Message<?>> responses = new ArrayList<Message<?>>();
final Semaphore semaphore = new Semaphore(0);
AbstractServerConnectionFactory scf = getConnectionFactory(port, serializer,new TcpListener() {
@Override
public boolean onMessage(Message<?> message) {
responses.add(message);
semaphore.release();
@@ -185,7 +187,7 @@ public class TcpNioConnectionReadTests {
// Fire up the sender.
SocketTestUtils.testSendCrLf(port, latch);
CountDownLatch done = SocketTestUtils.testSendCrLf(port, latch);
latch.countDown();
assertTrue(semaphore.tryAcquire(1, 10000, TimeUnit.MILLISECONDS));
assertTrue(semaphore.tryAcquire(1, 10000, TimeUnit.MILLISECONDS));
@@ -194,31 +196,30 @@ public class TcpNioConnectionReadTests {
new String(((Message<byte[]>) responses.get(0)).getPayload()));
assertEquals("Data", SocketTestUtils.TEST_STRING + SocketTestUtils.TEST_STRING,
new String(((Message<byte[]>) responses.get(1)).getPayload()));
scf.close();
scf.stop();
done.countDown();
}
/**
* Test method for {@link org.springframework.integration.ip.tcp.NioSocketReader}.
*/
@Test
public void testReadLengthOverflow() throws Exception {
int port = SocketUtils.findAvailableServerSocket();
ByteArrayLengthHeaderSerializer serializer = new ByteArrayLengthHeaderSerializer();
final List<Message<?>> responses = new ArrayList<Message<?>>();
final Semaphore semaphore = new Semaphore(0);
final List<TcpConnection> added = new ArrayList<TcpConnection>();
final List<TcpConnection> removed = new ArrayList<TcpConnection>();
AbstractServerConnectionFactory scf = getConnectionFactory(port, serializer,new TcpListener() {
@Override
public boolean onMessage(Message<?> message) {
responses.add(message);
semaphore.release();
return false;
}
}, new TcpSender() {
@Override
public void addNewConnection(TcpConnection connection) {
added.add(connection);
semaphore.release();
}
@Override
public void removeDeadConnection(TcpConnection connection) {
removed.add(connection);
semaphore.release();
@@ -227,37 +228,36 @@ public class TcpNioConnectionReadTests {
// Fire up the sender.
SocketTestUtils.testSendLengthOverflow(port);
CountDownLatch done = SocketTestUtils.testSendLengthOverflow(port);
whileOpen(semaphore, added);
assertEquals(1, added.size());
assertTrue(semaphore.tryAcquire(10000, TimeUnit.MILLISECONDS));
assertTrue(removed.size() > 0);
scf.close();
scf.stop();
done.countDown();
}
/**
* Test method for {@link org.springframework.integration.ip.tcp.NioSocketReader}.
*/
@Test
public void testReadStxEtxOverflow() throws Exception {
int port = SocketUtils.findAvailableServerSocket();
ByteArrayStxEtxSerializer serializer = new ByteArrayStxEtxSerializer();
serializer.setMaxMessageSize(1024);
final List<Message<?>> responses = new ArrayList<Message<?>>();
final Semaphore semaphore = new Semaphore(0);
final List<TcpConnection> added = new ArrayList<TcpConnection>();
final List<TcpConnection> removed = new ArrayList<TcpConnection>();
AbstractServerConnectionFactory scf = getConnectionFactory(port, serializer,new TcpListener() {
@Override
public boolean onMessage(Message<?> message) {
responses.add(message);
semaphore.release();
return false;
}
}, new TcpSender() {
@Override
public void addNewConnection(TcpConnection connection) {
added.add(connection);
semaphore.release();
}
@Override
public void removeDeadConnection(TcpConnection connection) {
removed.add(connection);
semaphore.release();
@@ -266,37 +266,36 @@ public class TcpNioConnectionReadTests {
// Fire up the sender.
SocketTestUtils.testSendStxEtxOverflow(port);
CountDownLatch done = SocketTestUtils.testSendStxEtxOverflow(port);
whileOpen(semaphore, added);
assertEquals(1, added.size());
assertTrue(semaphore.tryAcquire(10000, TimeUnit.MILLISECONDS));
assertTrue(removed.size() > 0);
scf.close();
scf.stop();
done.countDown();
}
/**
* Test method for {@link org.springframework.integration.ip.tcp.NioSocketReader}.
*/
@Test
public void testReadCrLfOverflow() throws Exception {
int port = SocketUtils.findAvailableServerSocket();
ByteArrayCrLfSerializer serializer = new ByteArrayCrLfSerializer();
serializer.setMaxMessageSize(1024);
final List<Message<?>> responses = new ArrayList<Message<?>>();
final Semaphore semaphore = new Semaphore(0);
final List<TcpConnection> added = new ArrayList<TcpConnection>();
final List<TcpConnection> removed = new ArrayList<TcpConnection>();
AbstractServerConnectionFactory scf = getConnectionFactory(port, serializer,new TcpListener() {
@Override
public boolean onMessage(Message<?> message) {
responses.add(message);
semaphore.release();
return false;
}
}, new TcpSender() {
@Override
public void addNewConnection(TcpConnection connection) {
added.add(connection);
semaphore.release();
}
@Override
public void removeDeadConnection(TcpConnection connection) {
removed.add(connection);
semaphore.release();
@@ -305,12 +304,13 @@ public class TcpNioConnectionReadTests {
// Fire up the sender.
SocketTestUtils.testSendCrLfOverflow(port);
CountDownLatch done = SocketTestUtils.testSendCrLfOverflow(port);
whileOpen(semaphore, added);
assertEquals(1, added.size());
assertTrue(semaphore.tryAcquire(10000, TimeUnit.MILLISECONDS));
assertTrue(removed.size() > 0);
scf.close();
scf.stop();
done.countDown();
}
/**
@@ -323,21 +323,22 @@ public class TcpNioConnectionReadTests {
int port = SocketUtils.findAvailableServerSocket();
ByteArrayCrLfSerializer serializer = new ByteArrayCrLfSerializer();
serializer.setMaxMessageSize(1024);
final List<Message<?>> responses = new ArrayList<Message<?>>();
final Semaphore semaphore = new Semaphore(0);
final List<TcpConnection> added = new ArrayList<TcpConnection>();
final List<TcpConnection> removed = new ArrayList<TcpConnection>();
AbstractServerConnectionFactory scf = getConnectionFactory(port, serializer,new TcpListener() {
@Override
public boolean onMessage(Message<?> message) {
responses.add(message);
semaphore.release();
return false;
}
}, new TcpSender() {
@Override
public void addNewConnection(TcpConnection connection) {
added.add(connection);
semaphore.release();
}
@Override
public void removeDeadConnection(TcpConnection connection) {
removed.add(connection);
semaphore.release();
@@ -349,7 +350,7 @@ public class TcpNioConnectionReadTests {
assertEquals(1, added.size());
assertTrue(semaphore.tryAcquire(10000, TimeUnit.MILLISECONDS));
assertTrue(removed.size() > 0);
scf.close();
scf.stop();
}
/**
@@ -362,21 +363,22 @@ public class TcpNioConnectionReadTests {
int port = SocketUtils.findAvailableServerSocket();
ByteArrayCrLfSerializer serializer = new ByteArrayCrLfSerializer();
serializer.setMaxMessageSize(1024);
final List<Message<?>> responses = new ArrayList<Message<?>>();
final Semaphore semaphore = new Semaphore(0);
final List<TcpConnection> added = new ArrayList<TcpConnection>();
final List<TcpConnection> removed = new ArrayList<TcpConnection>();
AbstractServerConnectionFactory scf = getConnectionFactory(port, serializer,new TcpListener() {
@Override
public boolean onMessage(Message<?> message) {
responses.add(message);
semaphore.release();
return false;
}
}, new TcpSender() {
@Override
public void addNewConnection(TcpConnection connection) {
added.add(connection);
semaphore.release();
}
@Override
public void removeDeadConnection(TcpConnection connection) {
removed.add(connection);
semaphore.release();
@@ -389,7 +391,7 @@ public class TcpNioConnectionReadTests {
assertEquals(1, added.size());
assertTrue(semaphore.tryAcquire(10000, TimeUnit.MILLISECONDS));
assertTrue(removed.size() > 0);
scf.close();
scf.stop();
}
/**
@@ -428,23 +430,25 @@ public class TcpNioConnectionReadTests {
}
private void testClosureMidMessageGuts(AbstractByteArraySerializer serializer, String shortMessage)
throws Exception, IOException, UnknownHostException,
InterruptedException {
throws Exception {
final int port = SocketUtils.findAvailableServerSocket();
final List<Message<?>> responses = new ArrayList<Message<?>>();
final Semaphore semaphore = new Semaphore(0);
final List<TcpConnection> added = new ArrayList<TcpConnection>();
final List<TcpConnection> removed = new ArrayList<TcpConnection>();
AbstractServerConnectionFactory scf = getConnectionFactory(port, serializer,new TcpListener() {
@Override
public boolean onMessage(Message<?> message) {
responses.add(message);
return false;
}
}, new TcpSender() {
@Override
public void addNewConnection(TcpConnection connection) {
added.add(connection);
semaphore.release();
}
@Override
public void removeDeadConnection(TcpConnection connection) {
removed.add(connection);
semaphore.release();
@@ -457,7 +461,7 @@ public class TcpNioConnectionReadTests {
assertEquals(1, added.size());
assertTrue(semaphore.tryAcquire(10000, TimeUnit.MILLISECONDS));
assertTrue(removed.size() > 0);
scf.close();
scf.stop();
}
private void whileOpen(Semaphore semaphore, final List<TcpConnection> added)

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

@@ -50,6 +50,7 @@ import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicReference;
import javax.net.ServerSocketFactory;
import org.junit.Test;
@@ -81,6 +82,7 @@ import org.springframework.util.ReflectionUtils.FieldFilter;
public class TcpNioConnectionTests {
private final ApplicationEventPublisher nullPublisher = new ApplicationEventPublisher() {
@Override
public void publishEvent(ApplicationEvent event) {
}
};
@@ -92,16 +94,21 @@ public class TcpNioConnectionTests {
factory.setSoTimeout(1000);
factory.start();
final CountDownLatch latch = new CountDownLatch(1);
final CountDownLatch done = new CountDownLatch(1);
final AtomicReference<ServerSocket> serverSocket = new AtomicReference<ServerSocket>();
Executors.newSingleThreadExecutor().execute(new Runnable() {
@Override
@SuppressWarnings("unused")
public void run() {
try {
ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(port);
serverSocket.set(server);
latch.countDown();
Socket s = server.accept();
// block so we fill the buffer
server.accept();
} catch (Exception e) {
done.await(10, TimeUnit.SECONDS);
}
catch (Exception e) {
e.printStackTrace();
}
}
@@ -110,10 +117,13 @@ public class TcpNioConnectionTests {
try {
TcpConnection connection = factory.getConnection();
connection.send(MessageBuilder.withPayload(new byte[1000000]).build());
} catch (Exception e) {
}
catch (Exception e) {
assertTrue("Expected SocketTimeoutException, got " + e.getClass().getSimpleName() +
":" + e.getMessage(), e instanceof SocketTimeoutException);
}
done.countDown();
serverSocket.get().close();
}
@Test
@@ -123,17 +133,22 @@ public class TcpNioConnectionTests {
factory.setSoTimeout(1000);
factory.start();
final CountDownLatch latch = new CountDownLatch(1);
final CountDownLatch done = new CountDownLatch(1);
final AtomicReference<ServerSocket> serverSocket = new AtomicReference<ServerSocket>();
Executors.newSingleThreadExecutor().execute(new Runnable() {
@Override
public void run() {
try {
ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(port);
serverSocket.set(server);
latch.countDown();
Socket socket = server.accept();
byte[] b = new byte[6];
readFully(socket.getInputStream(), b);
// block to cause timeout on read.
server.accept();
} catch (Exception e) {
done.await(10, TimeUnit.SECONDS);
}
catch (Exception e) {
e.printStackTrace();
}
}
@@ -150,9 +165,12 @@ public class TcpNioConnectionTests {
}
}
assertTrue(!connection.isOpen());
} catch (Exception e) {
}
catch (Exception e) {
fail("Unexpected exception " + e);
}
done.countDown();
serverSocket.get().close();
}
@Test
@@ -162,15 +180,19 @@ public class TcpNioConnectionTests {
factory.setNioHarvestInterval(100);
factory.start();
final CountDownLatch latch = new CountDownLatch(1);
final AtomicReference<ServerSocket> serverSocket = new AtomicReference<ServerSocket>();
Executors.newSingleThreadExecutor().execute(new Runnable() {
@Override
public void run() {
try {
ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(port);
serverSocket.set(server);
latch.countDown();
Socket socket = server.accept();
byte[] b = new byte[6];
readFully(socket.getInputStream(), b);
} catch (Exception e) {
}
catch (Exception e) {
e.printStackTrace();
}
}
@@ -182,8 +204,7 @@ public class TcpNioConnectionTests {
assertEquals(1, connections.size());
connection.close();
assertTrue(!connection.isOpen());
// force a wakeup of the selector
factory.close();
TestUtils.getPropertyValue(factory, "selector", Selector.class).wakeup();
int n = 0;
while (connections.size() > 0) {
Thread.sleep(100);
@@ -192,11 +213,13 @@ public class TcpNioConnectionTests {
}
}
assertEquals(0, connections.size());
} catch (Exception e) {
}
catch (Exception e) {
e.printStackTrace();
fail("Unexpected exception " + e);
}
factory.stop();
serverSocket.get().close();
}
@Test
@@ -216,6 +239,7 @@ public class TcpNioConnectionTests {
final List<Field> fields = new ArrayList<Field>();
ReflectionUtils.doWithFields(SocketChannel.class, new FieldCallback() {
@Override
public void doWith(Field field) throws IllegalArgumentException,
IllegalAccessException {
field.setAccessible(true);
@@ -223,6 +247,7 @@ public class TcpNioConnectionTests {
}
}, new FieldFilter() {
@Override
public boolean matches(Field field) {
return field.getName().equals("open");
}});
@@ -265,11 +290,13 @@ public class TcpNioConnectionTests {
public void testInsufficientThreads() throws Exception {
final ExecutorService exec = Executors.newFixedThreadPool(2);
Future<Object> future = exec.submit(new Callable<Object>() {
@Override
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>() {
@Override
public Integer answer(InvocationOnMock invocation) throws Throwable {
ByteBuffer buffer = (ByteBuffer) invocation.getArguments()[0];
buffer.position(1);
@@ -311,11 +338,13 @@ public class TcpNioConnectionTests {
final ExecutorService exec = Executors.newFixedThreadPool(3);
final CountDownLatch messageLatch = new CountDownLatch(1);
Future<Object> future = exec.submit(new Callable<Object>() {
@Override
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>() {
@Override
public Integer answer(InvocationOnMock invocation) throws Throwable {
ByteBuffer buffer = (ByteBuffer) invocation.getArguments()[0];
buffer.position(1025);
@@ -327,6 +356,7 @@ public class TcpNioConnectionTests {
final TcpNioConnection connection = new TcpNioConnection(channel, false, false, null, null);
connection.setTaskExecutor(exec);
connection.registerListener(new TcpListener(){
@Override
public boolean onMessage(Message<?> message) {
messageLatch.countDown();
return false;
@@ -438,6 +468,7 @@ public class TcpNioConnectionTests {
final byte[] out = new byte[4];
ExecutorService exec = Executors.newSingleThreadExecutor();
exec.execute(new Runnable(){
@Override
public void run() {
try {
stream.read(out);
@@ -468,6 +499,7 @@ public class TcpNioConnectionTests {
inboundConnection.setMapper(inMapper);
final ByteArrayOutputStream written = new ByteArrayOutputStream();
doAnswer(new Answer<Integer>() {
@Override
public Integer answer(InvocationOnMock invocation) throws Throwable {
ByteBuffer buff = (ByteBuffer) invocation.getArguments()[0];
byte[] bytes = written.toByteArray();
@@ -481,6 +513,7 @@ public class TcpNioConnectionTests {
when(outChannel.socket()).thenReturn(outSocket);
TcpNioConnection outboundConnection = new TcpNioConnection(outChannel, true, false, nullPublisher, null);
doAnswer(new Answer<Object>() {
@Override
public Object answer(InvocationOnMock invocation) throws Throwable {
ByteBuffer buff = (ByteBuffer) invocation.getArguments()[0];
byte[] bytes = new byte[buff.limit()];
@@ -505,6 +538,7 @@ public class TcpNioConnectionTests {
final CountDownLatch latch = new CountDownLatch(1);
TcpListener listener = new TcpListener() {
@Override
public boolean onMessage(Message<?> message) {
inboundMessage.set(message);
latch.countDown();

55
spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/TcpNioConnectionWriteTests.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2012 the original author or authors.
* Copyright 2002-2013 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.
@@ -23,10 +23,13 @@ import java.io.InputStream;
import java.net.ServerSocket;
import java.net.Socket;
import java.nio.ByteBuffer;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import javax.net.ServerSocketFactory;
import org.junit.Test;
import org.springframework.integration.ip.tcp.serializer.AbstractByteArraySerializer;
import org.springframework.integration.ip.tcp.serializer.ByteArrayCrLfSerializer;
import org.springframework.integration.ip.tcp.serializer.ByteArrayLengthHeaderSerializer;
@@ -58,15 +61,18 @@ public class TcpNioConnectionWriteTests {
ServerSocket server = ServerSocketFactory.getDefault()
.createServerSocket(port);
server.setSoTimeout(10000);
final CountDownLatch latch = new CountDownLatch(1);
Thread t = new Thread(new Runnable() {
@Override
public void run() {
try {
ByteArrayLengthHeaderSerializer serializer = new ByteArrayLengthHeaderSerializer();
AbstractConnectionFactory ccf = getClientConnectionFactory(false, port, serializer);
TcpConnection connection = ccf.getConnection();
connection.send(MessageBuilder.withPayload(testString.getBytes()).build());
Thread.sleep(1000000000L);
} catch (Exception e) {
latch.await(10, TimeUnit.SECONDS);
}
catch (Exception e) {
e.printStackTrace();
}
}
@@ -82,6 +88,7 @@ public class TcpNioConnectionWriteTests {
assertEquals(testString.length(), buffer.getInt());
assertEquals(testString, new String(buff, 4, testString.length()));
server.close();
latch.countDown();
}
@Test
@@ -91,15 +98,18 @@ public class TcpNioConnectionWriteTests {
ServerSocket server = ServerSocketFactory.getDefault()
.createServerSocket(port);
server.setSoTimeout(10000);
final CountDownLatch latch = new CountDownLatch(1);
Thread t = new Thread(new Runnable() {
@Override
public void run() {
try {
ByteArrayStxEtxSerializer serializer = new ByteArrayStxEtxSerializer();
AbstractConnectionFactory ccf = getClientConnectionFactory(false, port, serializer);
TcpConnection connection = ccf.getConnection();
connection.send(MessageBuilder.withPayload(testString.getBytes()).build());
Thread.sleep(1000000000L);
} catch (Exception e) {
latch.await(10, TimeUnit.SECONDS);
}
catch (Exception e) {
e.printStackTrace();
}
}
@@ -115,6 +125,7 @@ public class TcpNioConnectionWriteTests {
assertEquals(testString, new String(buff, 1, testString.length()));
assertEquals(ByteArrayStxEtxSerializer.ETX, buff[testString.length() + 1]);
server.close();
latch.countDown();
}
@Test
@@ -124,15 +135,18 @@ public class TcpNioConnectionWriteTests {
ServerSocket server = ServerSocketFactory.getDefault()
.createServerSocket(port);
server.setSoTimeout(10000);
final CountDownLatch latch = new CountDownLatch(1);
Thread t = new Thread(new Runnable() {
@Override
public void run() {
try {
ByteArrayCrLfSerializer serializer = new ByteArrayCrLfSerializer();
AbstractConnectionFactory ccf = getClientConnectionFactory(false, port, serializer);
TcpConnection connection = ccf.getConnection();
connection.send(MessageBuilder.withPayload(testString.getBytes()).build());
Thread.sleep(1000000000L);
} catch (Exception e) {
latch.await(10, TimeUnit.SECONDS);
}
catch (Exception e) {
e.printStackTrace();
}
}
@@ -148,6 +162,7 @@ public class TcpNioConnectionWriteTests {
assertEquals('\r', buff[testString.length()]);
assertEquals('\n', buff[testString.length() + 1]);
server.close();
latch.countDown();
}
@Test
@@ -157,15 +172,18 @@ public class TcpNioConnectionWriteTests {
ServerSocket server = ServerSocketFactory.getDefault()
.createServerSocket(port);
server.setSoTimeout(10000);
final CountDownLatch latch = new CountDownLatch(1);
Thread t = new Thread(new Runnable() {
@Override
public void run() {
try {
ByteArrayLengthHeaderSerializer serializer = new ByteArrayLengthHeaderSerializer();
AbstractConnectionFactory ccf = getClientConnectionFactory(true, port, serializer);
TcpConnection connection = ccf.getConnection();
connection.send(MessageBuilder.withPayload(testString.getBytes()).build());
Thread.sleep(1000000000L);
} catch (Exception e) {
latch.await(10, TimeUnit.SECONDS);
}
catch (Exception e) {
e.printStackTrace();
}
}
@@ -181,6 +199,7 @@ public class TcpNioConnectionWriteTests {
assertEquals(testString.length(), buffer.getInt());
assertEquals(testString, new String(buff, 4, testString.length()));
server.close();
latch.countDown();
}
@Test
@@ -190,16 +209,18 @@ public class TcpNioConnectionWriteTests {
ServerSocket server = ServerSocketFactory.getDefault()
.createServerSocket(port);
server.setSoTimeout(10000);
final CountDownLatch latch = new CountDownLatch(1);
Thread t = new Thread(new Runnable() {
@Override
public void run() {
try {
ByteArrayStxEtxSerializer serializer = new ByteArrayStxEtxSerializer();
AbstractConnectionFactory ccf = getClientConnectionFactory(true, port, serializer);
TcpConnection connection = ccf.getConnection();
connection.send(MessageBuilder.withPayload(testString.getBytes()).build());
Thread.sleep(1000000000L);
Thread.sleep(1000000000L);
} catch (Exception e) {
latch.await(10, TimeUnit.SECONDS);
}
catch (Exception e) {
e.printStackTrace();
}
}
@@ -215,6 +236,7 @@ public class TcpNioConnectionWriteTests {
assertEquals(testString, new String(buff, 1, testString.length()));
assertEquals(ByteArrayStxEtxSerializer.ETX, buff[testString.length() + 1]);
server.close();
latch.countDown();
}
@Test
@@ -224,16 +246,18 @@ public class TcpNioConnectionWriteTests {
ServerSocket server = ServerSocketFactory.getDefault()
.createServerSocket(port);
server.setSoTimeout(10000);
final CountDownLatch latch = new CountDownLatch(1);
Thread t = new Thread(new Runnable() {
@Override
public void run() {
try {
ByteArrayCrLfSerializer serializer = new ByteArrayCrLfSerializer();
AbstractConnectionFactory ccf = getClientConnectionFactory(true, port, serializer);
TcpConnection connection = ccf.getConnection();
connection.send(MessageBuilder.withPayload(testString.getBytes()).build());
Thread.sleep(1000000000L);
Thread.sleep(1000000000L);
} catch (Exception e) {
latch.await(10, TimeUnit.SECONDS);
}
catch (Exception e) {
e.printStackTrace();
}
}
@@ -249,6 +273,7 @@ public class TcpNioConnectionWriteTests {
assertEquals('\r', buff[testString.length()]);
assertEquals('\n', buff[testString.length() + 1]);
server.close();
latch.countDown();
}
/**

34
spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/serializer/DeserializationTests.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2012 the original author or authors.
* Copyright 2002-2013 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.
@@ -22,10 +22,12 @@ import static org.junit.Assert.fail;
import java.io.IOException;
import java.net.ServerSocket;
import java.net.Socket;
import java.util.concurrent.CountDownLatch;
import javax.net.ServerSocketFactory;
import org.junit.Test;
import org.springframework.core.serializer.DefaultDeserializer;
import org.springframework.integration.ip.util.SocketTestUtils;
import org.springframework.integration.test.util.SocketUtils;
@@ -41,7 +43,7 @@ public class DeserializationTests {
int port = SocketUtils.findAvailableServerSocket();
ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(port);
server.setSoTimeout(10000);
SocketTestUtils.testSendLength(port, null);
CountDownLatch done = SocketTestUtils.testSendLength(port, null);
Socket socket = server.accept();
socket.setSoTimeout(5000);
ByteArrayLengthHeaderSerializer serializer = new ByteArrayLengthHeaderSerializer();
@@ -52,6 +54,7 @@ public class DeserializationTests {
assertEquals("Data", SocketTestUtils.TEST_STRING + SocketTestUtils.TEST_STRING,
new String(out));
server.close();
done.countDown();
}
@Test
@@ -59,7 +62,7 @@ public class DeserializationTests {
int port = SocketUtils.findAvailableServerSocket();
ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(port);
server.setSoTimeout(10000);
SocketTestUtils.testSendStxEtx(port, null);
CountDownLatch done = SocketTestUtils.testSendStxEtx(port, null);
Socket socket = server.accept();
socket.setSoTimeout(5000);
ByteArrayStxEtxSerializer serializer = new ByteArrayStxEtxSerializer();
@@ -70,6 +73,7 @@ public class DeserializationTests {
assertEquals("Data", SocketTestUtils.TEST_STRING + SocketTestUtils.TEST_STRING,
new String(out));
server.close();
done.countDown();
}
@Test
@@ -77,7 +81,7 @@ public class DeserializationTests {
int port = SocketUtils.findAvailableServerSocket();
ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(port);
server.setSoTimeout(10000);
SocketTestUtils.testSendCrLf(port, null);
CountDownLatch done = SocketTestUtils.testSendCrLf(port, null);
Socket socket = server.accept();
socket.setSoTimeout(5000);
ByteArrayCrLfSerializer serializer = new ByteArrayCrLfSerializer();
@@ -88,6 +92,7 @@ public class DeserializationTests {
assertEquals("Data", SocketTestUtils.TEST_STRING + SocketTestUtils.TEST_STRING,
new String(out));
server.close();
done.countDown();
}
@Test
@@ -110,7 +115,7 @@ public class DeserializationTests {
int port = SocketUtils.findAvailableServerSocket();
ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(port);
server.setSoTimeout(10000);
SocketTestUtils.testSendSerialized(port);
CountDownLatch done = SocketTestUtils.testSendSerialized(port);
Socket socket = server.accept();
socket.setSoTimeout(5000);
DefaultDeserializer deserializer = new DefaultDeserializer();
@@ -119,6 +124,7 @@ public class DeserializationTests {
out = deserializer.deserialize(socket.getInputStream());
assertEquals("Data", SocketTestUtils.TEST_STRING, out);
server.close();
done.countDown();
}
@Test
@@ -126,7 +132,7 @@ public class DeserializationTests {
int port = SocketUtils.findAvailableServerSocket();
ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(port);
server.setSoTimeout(10000);
SocketTestUtils.testSendLengthOverflow(port);
CountDownLatch done = SocketTestUtils.testSendLengthOverflow(port);
Socket socket = server.accept();
socket.setSoTimeout(5000);
ByteArrayLengthHeaderSerializer serializer = new ByteArrayLengthHeaderSerializer();
@@ -140,6 +146,7 @@ public class DeserializationTests {
}
}
server.close();
done.countDown();
}
@Test
@@ -147,7 +154,7 @@ public class DeserializationTests {
int port = SocketUtils.findAvailableServerSocket();
ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(port);
server.setSoTimeout(10000);
SocketTestUtils.testSendStxEtxOverflow(port);
CountDownLatch done = SocketTestUtils.testSendStxEtxOverflow(port);
Socket socket = server.accept();
socket.setSoTimeout(500);
ByteArrayStxEtxSerializer serializer = new ByteArrayStxEtxSerializer();
@@ -161,6 +168,7 @@ public class DeserializationTests {
}
}
server.close();
done.countDown();
}
@Test
@@ -168,7 +176,7 @@ public class DeserializationTests {
int port = SocketUtils.findAvailableServerSocket();
ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(port);
server.setSoTimeout(10000);
SocketTestUtils.testSendStxEtxOverflow(port);
CountDownLatch done = SocketTestUtils.testSendStxEtxOverflow(port);
Socket socket = server.accept();
socket.setSoTimeout(5000);
ByteArrayStxEtxSerializer serializer = new ByteArrayStxEtxSerializer();
@@ -183,6 +191,7 @@ public class DeserializationTests {
}
}
server.close();
done.countDown();
}
@Test
@@ -190,7 +199,7 @@ public class DeserializationTests {
int port = SocketUtils.findAvailableServerSocket();
ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(port);
server.setSoTimeout(10000);
SocketTestUtils.testSendCrLfOverflow(port);
CountDownLatch latch = SocketTestUtils.testSendCrLfOverflow(port);
Socket socket = server.accept();
socket.setSoTimeout(500);
ByteArrayCrLfSerializer serializer = new ByteArrayCrLfSerializer();
@@ -204,6 +213,7 @@ public class DeserializationTests {
}
}
server.close();
latch.countDown();
}
@Test
@@ -211,7 +221,7 @@ public class DeserializationTests {
int port = SocketUtils.findAvailableServerSocket();
ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(port);
server.setSoTimeout(10000);
SocketTestUtils.testSendCrLfOverflow(port);
CountDownLatch latch = SocketTestUtils.testSendCrLfOverflow(port);
Socket socket = server.accept();
socket.setSoTimeout(5000);
ByteArrayCrLfSerializer serializer = new ByteArrayCrLfSerializer();
@@ -219,13 +229,15 @@ public class DeserializationTests {
try {
serializer.deserialize(socket.getInputStream());
fail("Expected message length exceeded exception");
} catch (IOException e) {
}
catch (IOException e) {
if (!e.getMessage().startsWith("CRLF not found")) {
e.printStackTrace();
fail("Unexpected IO Error:" + e.getMessage());
}
}
server.close();
latch.countDown();
}
}

45
spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/serializer/SerializationTests.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2012 the original author or authors.
* Copyright 2002-2013 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.
@@ -24,11 +24,14 @@ import java.io.ObjectInputStream;
import java.net.ServerSocket;
import java.net.Socket;
import java.nio.ByteBuffer;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import javax.net.ServerSocketFactory;
import javax.net.SocketFactory;
import org.junit.Test;
import org.springframework.core.serializer.DefaultSerializer;
import org.springframework.integration.test.util.SocketUtils;
@@ -44,7 +47,9 @@ public class SerializationTests {
final String testString = "abcdef";
ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(port);
server.setSoTimeout(10000);
final CountDownLatch latch = new CountDownLatch(1);
Thread t = new Thread(new Runnable() {
@Override
public void run() {
try {
Socket socket = SocketFactory.getDefault().createSocket("localhost", port);
@@ -52,8 +57,9 @@ public class SerializationTests {
buffer.put(testString.getBytes());
ByteArrayLengthHeaderSerializer serializer = new ByteArrayLengthHeaderSerializer();
serializer.serialize(buffer.array(), socket.getOutputStream());
Thread.sleep(1000000000L);
} catch (Exception e) {
latch.await(10, TimeUnit.SECONDS);
}
catch (Exception e) {
e.printStackTrace();
}
}
@@ -69,6 +75,7 @@ public class SerializationTests {
assertEquals(testString.length(), buffer.getInt());
assertEquals(testString, new String(buff, 4, testString.length()));
server.close();
latch.countDown();
}
@Test
@@ -77,7 +84,9 @@ public class SerializationTests {
final String testString = "abcdef";
ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(port);
server.setSoTimeout(10000);
final CountDownLatch latch = new CountDownLatch(1);
Thread t = new Thread(new Runnable() {
@Override
public void run() {
try {
Socket socket = SocketFactory.getDefault().createSocket("localhost", port);
@@ -85,8 +94,9 @@ public class SerializationTests {
buffer.put(testString.getBytes());
ByteArrayStxEtxSerializer serializer = new ByteArrayStxEtxSerializer();
serializer.serialize(buffer.array(), socket.getOutputStream());
Thread.sleep(1000000000L);
} catch (Exception e) {
latch.await(10, TimeUnit.SECONDS);
}
catch (Exception e) {
e.printStackTrace();
}
}
@@ -102,6 +112,7 @@ public class SerializationTests {
assertEquals(testString, new String(buff, 1, testString.length()));
assertEquals(ByteArrayStxEtxSerializer.ETX, buff[testString.length() + 1]);
server.close();
latch.countDown();
}
@Test
@@ -110,7 +121,9 @@ public class SerializationTests {
final String testString = "abcdef";
ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(port);
server.setSoTimeout(10000);
final CountDownLatch latch = new CountDownLatch(1);
Thread t = new Thread(new Runnable() {
@Override
public void run() {
try {
Socket socket = SocketFactory.getDefault().createSocket("localhost", port);
@@ -118,8 +131,9 @@ public class SerializationTests {
buffer.put(testString.getBytes());
ByteArrayCrLfSerializer serializer = new ByteArrayCrLfSerializer();
serializer.serialize(buffer.array(), socket.getOutputStream());
Thread.sleep(1000000000L);
} catch (Exception e) {
latch.await(10, TimeUnit.SECONDS);
}
catch (Exception e) {
e.printStackTrace();
}
}
@@ -135,6 +149,7 @@ public class SerializationTests {
assertEquals('\r', buff[testString.length()]);
assertEquals('\n', buff[testString.length() + 1]);
server.close();
latch.countDown();
}
@Test
@@ -143,7 +158,9 @@ public class SerializationTests {
final String testString = "abcdef";
ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(port);
server.setSoTimeout(10000);
final CountDownLatch latch = new CountDownLatch(1);
Thread t = new Thread(new Runnable() {
@Override
public void run() {
try {
Socket socket = SocketFactory.getDefault().createSocket("localhost", port);
@@ -152,8 +169,9 @@ public class SerializationTests {
ByteArrayRawSerializer serializer = new ByteArrayRawSerializer();
serializer.serialize(buffer.array(), socket.getOutputStream());
socket.close();
Thread.sleep(1000000000L);
} catch (Exception e) {
latch.await(10, TimeUnit.SECONDS);
}
catch (Exception e) {
e.printStackTrace();
}
}
@@ -167,6 +185,7 @@ public class SerializationTests {
readFully(is, buff);
assertEquals(testString, new String(buff, 0, testString.length()));
assertEquals(-1, buff[testString.length()]);
latch.countDown();
server.close();
}
@@ -176,15 +195,18 @@ public class SerializationTests {
final String testString = "abcdef";
ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(port);
server.setSoTimeout(10000);
final CountDownLatch latch = new CountDownLatch(1);
Thread t = new Thread(new Runnable() {
@Override
public void run() {
try {
Socket socket = SocketFactory.getDefault().createSocket("localhost", port);
DefaultSerializer serializer = new DefaultSerializer();
serializer.serialize(testString, socket.getOutputStream());
serializer.serialize(testString, socket.getOutputStream());
Thread.sleep(1000000000L);
} catch (Exception e) {
latch.await(10, TimeUnit.SECONDS);
}
catch (Exception e) {
e.printStackTrace();
}
}
@@ -198,6 +220,7 @@ public class SerializationTests {
assertEquals(testString, ois.readObject());
ois = new ObjectInputStream(is);
assertEquals(testString, ois.readObject());
latch.countDown();
server.close();
}

18
spring-integration-ip/src/test/java/org/springframework/integration/ip/udp/DatagramPacketSendingHandlerTests.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2010 the original author or authors.
* Copyright 2002-2013 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.
@@ -33,6 +33,7 @@ import java.util.concurrent.TimeUnit;
import org.apache.commons.logging.LogFactory;
import org.junit.Ignore;
import org.junit.Test;
import org.springframework.integration.Message;
import org.springframework.integration.ip.IpHeaders;
import org.springframework.integration.support.MessageBuilder;
@@ -54,6 +55,7 @@ public class DatagramPacketSendingHandlerTests {
final DatagramPacket receivedPacket = new DatagramPacket(buffer, buffer.length);
final CountDownLatch latch = new CountDownLatch(1);
Executors.newSingleThreadExecutor().execute(new Runnable() {
@Override
public void run() {
try {
DatagramSocket socket = new DatagramSocket(testPort);
@@ -78,7 +80,7 @@ public class DatagramPacketSendingHandlerTests {
byte[] dest = new byte[length];
System.arraycopy(src, offset, dest, 0, length);
assertEquals(payload, new String(dest));
handler.shutDown();
handler.stop();
}
@Test
@@ -97,7 +99,9 @@ public class DatagramPacketSendingHandlerTests {
new UnicastSendingMessageHandler("localhost", testPort, true,
true, "localhost", ackPort, 5000);
handler.afterPropertiesSet();
handler.start();
Executors.newSingleThreadExecutor().execute(new Runnable() {
@Override
public void run() {
try {
DatagramSocket socket = new DatagramSocket(testPort);
@@ -132,7 +136,7 @@ public class DatagramPacketSendingHandlerTests {
byte[] dest = new byte[6];
System.arraycopy(src, offset+length-6, dest, 0, 6);
assertEquals(payload, new String(dest));
handler.shutDown();
handler.stop();
}
@Test
@@ -144,6 +148,7 @@ public class DatagramPacketSendingHandlerTests {
final CountDownLatch latch1 = new CountDownLatch(2);
final CountDownLatch latch2 = new CountDownLatch(2);
Runnable catcher = new Runnable() {
@Override
public void run() {
try {
byte[] buffer = new byte[8];
@@ -183,7 +188,7 @@ public class DatagramPacketSendingHandlerTests {
MulticastSendingMessageHandler handler = new MulticastSendingMessageHandler(multicastAddress, testPort);
handler.handleMessage(MessageBuilder.withPayload(payload).build());
assertTrue(latch2.await(3000, TimeUnit.MILLISECONDS));
handler.shutDown();
handler.stop();
}
@Test
@@ -200,6 +205,7 @@ public class DatagramPacketSendingHandlerTests {
final CountDownLatch latch1 = new CountDownLatch(2);
final CountDownLatch latch2 = new CountDownLatch(2);
Runnable catcher = new Runnable() {
@Override
public void run() {
try {
byte[] buffer = new byte[1000];
@@ -249,9 +255,11 @@ public class DatagramPacketSendingHandlerTests {
new MulticastSendingMessageHandler(multicastAddress, testPort, true,
true, "localhost", ackPort, 500000);
handler.setMinAcksForSuccess(2);
handler.afterPropertiesSet();
handler.start();
handler.handleMessage(MessageBuilder.withPayload(payload).build());
assertTrue(latch2.await(10000, TimeUnit.MILLISECONDS));
handler.shutDown();
handler.stop();
}
}

46
spring-integration-ip/src/test/java/org/springframework/integration/ip/udp/MultiClientTests.java
View File

@@ -16,10 +16,13 @@
package org.springframework.integration.ip.udp;
import static org.junit.Assert.assertNotNull;
import org.junit.Assert;
import java.util.concurrent.atomic.AtomicBoolean;
import org.junit.Assert;
import org.junit.Ignore;
import org.junit.Test;
import org.springframework.integration.Message;
import org.springframework.integration.channel.QueueChannel;
import org.springframework.integration.ip.util.SocketTestUtils;
@@ -45,7 +48,8 @@ import org.springframework.integration.test.util.SocketUtils;
public class MultiClientTests {
@SuppressWarnings("unchecked")
@Test @Ignore
@Test
@Ignore
public void testNoAck() throws Exception {
final String payload = largePayload(1000);
final UnicastReceivingChannelAdapter adapter =
@@ -57,15 +61,24 @@ public class MultiClientTests {
adapter.start();
final QueueChannel queueIn = new QueueChannel(1000);
SocketTestUtils.waitListening(adapter);
final AtomicBoolean done = new AtomicBoolean();
for (int i = 0; i < drivers; i++) {
Thread t = new Thread( new Runnable() {
@Override
public void run() {
UnicastSendingMessageHandler sender = new UnicastSendingMessageHandler(
"localhost", adapter.getPort());
sender.start();
while (true) {
Message<?> message = queueIn.receive();
sender.handleMessage(message);
if (done.get()) {
break;
}
}
sender.stop();
}});
t.setDaemon(true);
t.start();
@@ -79,12 +92,13 @@ public class MultiClientTests {
Assert.assertEquals(payload, new String(messageOut.getPayload()));
}
adapter.stop();
done.set(true);
}
@SuppressWarnings("unchecked")
@Test @Ignore
@Test
@Ignore
public void testAck() throws Exception {
Thread.sleep(1000);
final String payload = largePayload(1000);
final UnicastReceivingChannelAdapter adapter =
new UnicastReceivingChannelAdapter(SocketUtils.findAvailableUdpSocket(), false);
@@ -95,19 +109,28 @@ public class MultiClientTests {
adapter.start();
final QueueChannel queueIn = new QueueChannel(1000);
SocketTestUtils.waitListening(adapter);
final AtomicBoolean done = new AtomicBoolean();
for (int i = 0; i < drivers; i++) {
final int j = i;
Thread t = new Thread( new Runnable() {
@Override
public void run() {
UnicastSendingMessageHandler sender = new UnicastSendingMessageHandler(
"localhost", adapter.getPort(),
false, true, "localhost",
SocketUtils.findAvailableUdpSocket(adapter.getPort() + j + 1000),
10000);
sender.start();
while (true) {
Message<?> message = queueIn.receive();
sender.handleMessage(message);
if (done.get()) {
break;
}
}
sender.stop();
}});
t.setDaemon(true);
t.start();
@@ -121,12 +144,13 @@ public class MultiClientTests {
Assert.assertEquals(payload, new String(messageOut.getPayload()));
}
adapter.stop();
done.set(true);
}
@SuppressWarnings("unchecked")
@Test @Ignore
@Test
@Ignore
public void testAckWithLength() throws Exception {
Thread.sleep(1000);
final String payload = largePayload(1000);
final UnicastReceivingChannelAdapter adapter =
new UnicastReceivingChannelAdapter(SocketUtils.findAvailableUdpSocket(), true);
@@ -137,19 +161,28 @@ public class MultiClientTests {
adapter.start();
final QueueChannel queueIn = new QueueChannel(1000);
SocketTestUtils.waitListening(adapter);
final AtomicBoolean done = new AtomicBoolean();
for (int i = 0; i < drivers; i++) {
final int j = i;
Thread t = new Thread( new Runnable() {
@Override
public void run() {
UnicastSendingMessageHandler sender = new UnicastSendingMessageHandler(
"localhost", adapter.getPort(),
true, true, "localhost",
SocketUtils.findAvailableUdpSocket(adapter.getPort() + j + 1100),
10000);
sender.start();
while (true) {
Message<?> message = queueIn.receive();
sender.handleMessage(message);
if (done.get()) {
break;
}
}
sender.stop();
}});
t.setDaemon(true);
t.start();
@@ -163,6 +196,7 @@ public class MultiClientTests {
Assert.assertEquals(payload, new String(messageOut.getPayload()));
}
adapter.stop();
done.set(true);
}
private String largePayload(int n) {

4
spring-integration-ip/src/test/java/org/springframework/integration/ip/udp/SyslogdTests.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2012 the original author or authors.
* Copyright 2002-2013 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.
@@ -29,6 +29,6 @@ public class SyslogdTests {
AbstractApplicationContext ctx = new ClassPathXmlApplicationContext("SyslogdTests-context.xml", SyslogdTests.class);
System.out.println("Hit enter to terminate");
System.in.read();
ctx.destroy();
ctx.close();
}
}

29
spring-integration-ip/src/test/java/org/springframework/integration/ip/udp/UdpChannelAdapterTests.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2012 the original author or authors.
* Copyright 2002-2013 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.
@@ -32,6 +32,7 @@ import java.util.concurrent.atomic.AtomicReference;
import org.apache.commons.logging.LogFactory;
import org.junit.Ignore;
import org.junit.Test;
import org.springframework.integration.Message;
import org.springframework.integration.channel.DirectChannel;
import org.springframework.integration.channel.QueueChannel;
@@ -65,9 +66,12 @@ public class UdpChannelAdapterTests {
DatagramPacketMessageMapper mapper = new DatagramPacketMessageMapper();
DatagramPacket packet = mapper.fromMessage(message);
packet.setSocketAddress(new InetSocketAddress("localhost", port));
new DatagramSocket(SocketUtils.findAvailableUdpSocket()).send(packet);
DatagramSocket datagramSocket = new DatagramSocket(SocketUtils.findAvailableUdpSocket());
datagramSocket.send(packet);
datagramSocket.close();
Message<byte[]> receivedMessage = (Message<byte[]>) channel.receive(2000);
assertEquals(new String(message.getPayload()), new String(receivedMessage.getPayload()));
adapter.stop();
}
@SuppressWarnings("unchecked")
@@ -91,6 +95,7 @@ public class UdpChannelAdapterTests {
final CountDownLatch replyReceivedLatch = new CountDownLatch(1);
//main thread sends the reply using the headers, this thread will receive it
Executors.newSingleThreadExecutor().execute(new Runnable() {
@Override
public void run() {
DatagramPacket answer = new DatagramPacket(new byte[2000], 2000);
try {
@@ -112,11 +117,15 @@ public class UdpChannelAdapterTests {
(String) receivedMessage.getHeaders().get(IpHeaders.IP_ADDRESS),
(Integer) receivedMessage.getHeaders().get(IpHeaders.PORT)));
assertTrue(receiverReadyLatch.await(10, TimeUnit.SECONDS));
new DatagramSocket().send(reply);
DatagramSocket datagramSocket = new DatagramSocket();
datagramSocket.send(reply);
assertTrue(replyReceivedLatch.await(10, TimeUnit.SECONDS));
DatagramPacket answerPacket = theAnswer.get();
assertNotNull(answerPacket);
assertEquals(replyString, new String(answerPacket.getData(), 0, answerPacket.getLength()));
datagramSocket.close();
socket.close();
adapter.stop();
}
@SuppressWarnings("unchecked")
@@ -139,10 +148,13 @@ public class UdpChannelAdapterTests {
SocketUtils.findAvailableUdpSocket(), 5000);
// handler.setLocalAddress(whichNic);
handler.afterPropertiesSet();
handler.start();
Message<byte[]> message = MessageBuilder.withPayload("ABCD".getBytes()).build();
handler.handleMessage(message);
Message<byte[]> receivedMessage = (Message<byte[]>) channel.receive(2000);
assertEquals(new String(message.getPayload()), new String(receivedMessage.getPayload()));
adapter.stop();
handler.stop();
}
@SuppressWarnings("unchecked")
@@ -165,11 +177,14 @@ public class UdpChannelAdapterTests {
DatagramPacketMessageMapper mapper = new DatagramPacketMessageMapper();
DatagramPacket packet = mapper.fromMessage(message);
packet.setSocketAddress(new InetSocketAddress("225.6.7.8", port));
new DatagramSocket(0, Inet4Address.getByName(nic)).send(packet);
DatagramSocket datagramSocket = new DatagramSocket(0, Inet4Address.getByName(nic));
datagramSocket.send(packet);
datagramSocket.close();
Message<byte[]> receivedMessage = (Message<byte[]>) channel.receive(2000);
assertNotNull(receivedMessage);
assertEquals(new String(message.getPayload()), new String(receivedMessage.getPayload()));
adapter.stop();
}
@SuppressWarnings("unchecked")
@@ -196,6 +211,7 @@ public class UdpChannelAdapterTests {
Message<byte[]> receivedMessage = (Message<byte[]>) channel.receive(2000);
assertNotNull(receivedMessage);
assertEquals(new String(message.getPayload()), new String(receivedMessage.getPayload()));
adapter.stop();
}
@Test
@@ -217,10 +233,13 @@ public class UdpChannelAdapterTests {
DatagramPacketMessageMapper mapper = new DatagramPacketMessageMapper();
DatagramPacket packet = mapper.fromMessage(message);
packet.setSocketAddress(new InetSocketAddress("localhost", port));
new DatagramSocket(SocketUtils.findAvailableUdpSocket()).send(packet);
DatagramSocket datagramSocket = new DatagramSocket(SocketUtils.findAvailableUdpSocket());
datagramSocket.send(packet);
datagramSocket.close();
Message<?> receivedMessage = errorChannel.receive(2000);
assertNotNull(receivedMessage);
assertEquals("Failed", ((Exception) receivedMessage.getPayload()).getCause().getMessage());
adapter.stop();
}
private class FailingService {

22
spring-integration-ip/src/test/java/org/springframework/integration/ip/udp/UdpMulticastEndToEndTests.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2010 the original author or authors.
* Copyright 2002-2013 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.
@@ -41,12 +41,12 @@ import org.springframework.integration.support.channel.ChannelResolver;
/**
* Sends and receives a simple message through to the Udp channel adapters.
* If run as a JUnit just sends one message and terminates (see console).
*
* If run from main(),
*
* If run from main(),
* hangs around for a couple of minutes to allow console interaction (enter a message on the
* console and you should see it go through the outbound context, over UDP, and
* console and you should see it go through the outbound context, over UDP, and
* received in the other context (and written back to the console).
*
*
* @author Gary Russell
* @since 2.0
*/
@@ -58,13 +58,13 @@ public class UdpMulticastEndToEndTests implements Runnable {
private CountDownLatch sentFirst = new CountDownLatch(1);
private CountDownLatch firstReceived = new CountDownLatch(1);
private final CountDownLatch firstReceived = new CountDownLatch(1);
private CountDownLatch doneProcessing = new CountDownLatch(1);
private final CountDownLatch doneProcessing = new CountDownLatch(1);
private boolean okToRun = true;
private CountDownLatch readyToReceive = new CountDownLatch(1);
private final CountDownLatch readyToReceive = new CountDownLatch(1);
private static long hangAroundFor = 0;
@@ -77,7 +77,7 @@ public class UdpMulticastEndToEndTests implements Runnable {
t.start(); // launch the receiver
AbstractApplicationContext applicationContext = new ClassPathXmlApplicationContext(
"testIp-out-multicast-context.xml",
UdpMulticastEndToEndTests.class);
UdpMulticastEndToEndTests.class);
launcher.launchSender(applicationContext);
applicationContext.stop();
}
@@ -122,6 +122,7 @@ public class UdpMulticastEndToEndTests implements Runnable {
/**
* Instantiate the receiving context
*/
@Override
@SuppressWarnings("unchecked")
public void run() {
AbstractApplicationContext ctx = new ClassPathXmlApplicationContext(
@@ -129,7 +130,7 @@ public class UdpMulticastEndToEndTests implements Runnable {
UdpMulticastEndToEndTests.class);
while (okToRun) {
try {
readyToReceive.countDown();
readyToReceive.countDown();
sentFirst.await();
}
catch (InterruptedException e) {
@@ -146,6 +147,7 @@ public class UdpMulticastEndToEndTests implements Runnable {
}
}
ctx.stop();
ctx.close();
}

9
spring-integration-ip/src/test/java/org/springframework/integration/ip/udp/UdpUnicastEndToEndTests.java
View File

@@ -16,8 +16,8 @@
package org.springframework.integration.ip.udp;
import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertTrue;
import static org.junit.Assert.fail;
@@ -28,6 +28,7 @@ import java.util.concurrent.TimeUnit;
import org.junit.Before;
import org.junit.Test;
import org.springframework.context.ApplicationContext;
import org.springframework.context.support.AbstractApplicationContext;
import org.springframework.context.support.ClassPathXmlApplicationContext;
@@ -142,7 +143,9 @@ public class UdpUnicastEndToEndTests implements Runnable {
* Instantiate the receiving context
*/
@SuppressWarnings("unchecked")
@Override
public void run() {
@SuppressWarnings("resource")
AbstractApplicationContext ctx = new ClassPathXmlApplicationContext(
"testIp-in-context.xml", UdpUnicastEndToEndTests.class);
UnicastReceivingChannelAdapter inbound = ctx.getBean(UnicastReceivingChannelAdapter.class);
@@ -154,7 +157,8 @@ public class UdpUnicastEndToEndTests implements Runnable {
throw new RuntimeException("Failed to start listening");
}
}
} catch (Exception e) { }
}
catch (Exception e) { }
while (okToRun) {
try {
readyToReceive.countDown();
@@ -178,6 +182,7 @@ public class UdpUnicastEndToEndTests implements Runnable {
}
}
ctx.stop();
ctx.close();
}

4
spring-integration-ip/src/test/java/org/springframework/integration/ip/udp/testIp-out-multicast-context.xml
View File

@@ -14,9 +14,7 @@
http://www.springframework.org/schema/integration/ip/spring-integration-ip.xsd">
<stream:stdin-channel-adapter id="stdin" channel="mcOutputChannel" >
<poller>
<interval-trigger interval="100" time-unit="MILLISECONDS"/>
</poller>
<poller fixed-delay="100" />
</stream:stdin-channel-adapter>
<channel id="mcInputChannel"/>

172
spring-integration-ip/src/test/java/org/springframework/integration/ip/util/SocketTestUtils.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2012 the original author or authors.
* Copyright 2002-2013 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.
@@ -16,6 +16,7 @@
package org.springframework.integration.ip.util;
import java.io.IOException;
import java.io.ObjectOutputStream;
import java.io.OutputStream;
import java.net.InetAddress;
@@ -23,6 +24,7 @@ import java.net.Socket;
import java.net.UnknownHostException;
import java.nio.ByteBuffer;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
@@ -46,11 +48,14 @@ public class SocketTestUtils {
* Sends a message in two chunks with a preceding length. Two such messages are sent.
* @param latch If not null, await until counted down before sending second chunk.
*/
public static void testSendLength(final int port, final CountDownLatch latch) {
public static CountDownLatch testSendLength(final int port, final CountDownLatch latch) {
final CountDownLatch testCompleteLatch = new CountDownLatch(1);
Thread thread = new Thread(new Runnable() {
@Override
public void run() {
Socket socket = null;
try {
Socket socket = new Socket(InetAddress.getByName("localhost"), port);
socket = new Socket(InetAddress.getByName("localhost"), port);
for (int i = 0; i < 2; i++) {
byte[] len = new byte[4];
ByteBuffer.wrap(len).putInt(TEST_STRING.length() * 2);
@@ -65,48 +70,74 @@ public class SocketTestUtils {
socket.getOutputStream().write(TEST_STRING.getBytes());
logger.debug(i + " Wrote second part");
}
Thread.sleep(1000000000L); // wait forever, but we're a daemon
} catch (Exception e) {
testCompleteLatch.await(10, TimeUnit.SECONDS);
}
catch (Exception e) {
e.printStackTrace();
}
finally {
if (socket != null) {
try {
socket.close();
}
catch (IOException e) { }
}
}
}
});
thread.setDaemon(true);
thread.start();
return testCompleteLatch;
}
/**
* Sends a message with a bad length part, causing an overflow on the receiver.
*/
public static void testSendLengthOverflow(final int port) {
public static CountDownLatch testSendLengthOverflow(final int port) {
final CountDownLatch testCompleteLatch = new CountDownLatch(1);
Thread thread = new Thread(new Runnable() {
@Override
public void run() {
Socket socket = null;
try {
Socket socket = new Socket(InetAddress.getByName("localhost"), port);
socket = new Socket(InetAddress.getByName("localhost"), port);
byte[] len = new byte[4];
ByteBuffer.wrap(len).putInt(Integer.MAX_VALUE);
socket.getOutputStream().write(len);
socket.getOutputStream().write(TEST_STRING.getBytes());
Thread.sleep(1000000000L); // wait forever, but we're a daemon
} catch (Exception e) {
testCompleteLatch.await(10, TimeUnit.SECONDS);
}
catch (Exception e) {
e.printStackTrace();
}
finally {
if (socket != null) {
try {
socket.close();
}
catch (IOException e) { }
}
}
}
});
thread.setDaemon(true);
thread.start();
return testCompleteLatch;
}
/**
* Test for reassembly of completely fragmented message; sends
* 6 bytes 500ms apart.
*/
public static void testSendFragmented(final int port, final int howMany, final boolean noDelay) {
public static CountDownLatch testSendFragmented(final int port, final int howMany, final boolean noDelay) {
final CountDownLatch testCompleteLatch = new CountDownLatch(1);
Thread thread = new Thread(new Runnable() {
@Override
public void run() {
Socket socket = null;
try {
logger.debug("Connecting to " + port);
Socket socket = new Socket(InetAddress.getByName("localhost"), port);
socket = new Socket(InetAddress.getByName("localhost"), port);
OutputStream os = socket.getOutputStream();
for (int i = 0; i < howMany; i++) {
writeByte(os, 0, noDelay);
@@ -116,14 +147,24 @@ public class SocketTestUtils {
writeByte(os, 'x', noDelay);
writeByte(os, 'x', noDelay);
}
Thread.sleep(1000000000L); // wait forever, but we're a daemon
} catch (Exception e) {
testCompleteLatch.await(10, TimeUnit.SECONDS);
}
catch (Exception e) {
e.printStackTrace();
}
finally {
if (socket != null) {
try {
socket.close();
}
catch (IOException e) { }
}
}
}
});
thread.setDaemon(true);
thread.start();
return testCompleteLatch;
}
private static void writeByte(OutputStream os, int b, boolean noDelay) throws Exception {
@@ -139,11 +180,14 @@ public class SocketTestUtils {
* Sends a STX/ETX message in two chunks. Two such messages are sent.
* @param latch If not null, await until counted down before sending second chunk.
*/
public static void testSendStxEtx(final int port, final CountDownLatch latch) {
public static CountDownLatch testSendStxEtx(final int port, final CountDownLatch latch) {
final CountDownLatch testCompleteLatch = new CountDownLatch(1);
Thread thread = new Thread(new Runnable() {
@Override
public void run() {
Socket socket = null;
try {
Socket socket = new Socket(InetAddress.getByName("localhost"), port);
socket = new Socket(InetAddress.getByName("localhost"), port);
OutputStream outputStream = socket.getOutputStream();
for (int i = 0; i < 2; i++) {
writeByte(outputStream, 0x02, true);
@@ -158,46 +202,74 @@ public class SocketTestUtils {
logger.debug(i + " Wrote second part");
writeByte(outputStream, 0x03, true);
}
Thread.sleep(1000000000L); // wait forever, but we're a daemon
} catch (Exception e) {
testCompleteLatch.await(10, TimeUnit.SECONDS);
}
catch (Exception e) {
e.printStackTrace();
}
finally {
if (socket != null) {
try {
socket.close();
}
catch (IOException e) { }
}
}
}
});
thread.setDaemon(true);
thread.start();
return testCompleteLatch;
}
/**
* Sends a large STX/ETX message with no ETX
*/
public static void testSendStxEtxOverflow(final int port) {
public static CountDownLatch testSendStxEtxOverflow(final int port) {
final CountDownLatch testCompleteLatch = new CountDownLatch(1);
Thread thread = new Thread(new Runnable() {
@Override
public void run() {
Socket socket = null;
try {
Socket socket = new Socket(InetAddress.getByName("localhost"), port);
socket = new Socket(InetAddress.getByName("localhost"), port);
OutputStream outputStream = socket.getOutputStream();
writeByte(outputStream, 0x02, true);
for (int i = 0; i < 1500; i++) {
writeByte(outputStream, 'x', true);
}
Thread.sleep(1000000000L); // wait forever, but we're a daemon
} catch (Exception e) { }
testCompleteLatch.await(10, TimeUnit.SECONDS);
}
catch (Exception e) {
e.printStackTrace();
}
finally {
if (socket != null) {
try {
socket.close();
}
catch (IOException e) { }
}
}
}
});
thread.setDaemon(true);
thread.start();
return testCompleteLatch;
}
/**
* Sends a message +CRLF in two chunks. Two such messages are sent.
* @param latch If not null, await until counted down before sending second chunk.
*/
public static void testSendCrLf(final int port, final CountDownLatch latch) {
public static CountDownLatch testSendCrLf(final int port, final CountDownLatch latch) {
final CountDownLatch testCompleteLatch = new CountDownLatch(1);
Thread thread = new Thread(new Runnable() {
@Override
public void run() {
Socket socket = null;
try {
Socket socket = new Socket(InetAddress.getByName("localhost"), port);
socket = new Socket(InetAddress.getByName("localhost"), port);
OutputStream outputStream = socket.getOutputStream();
for (int i = 0; i < 2; i++) {
outputStream.write(TEST_STRING.getBytes());
@@ -212,14 +284,24 @@ public class SocketTestUtils {
writeByte(outputStream, '\r', true);
writeByte(outputStream, '\n', true);
}
Thread.sleep(1000000000L); // wait forever, but we're a daemon
} catch (Exception e) {
testCompleteLatch.await(10, TimeUnit.SECONDS);
}
catch (Exception e) {
e.printStackTrace();
}
finally {
if (socket != null) {
try {
socket.close();
}
catch (IOException e) { }
}
}
}
});
thread.setDaemon(true);
thread.start();
return testCompleteLatch;
}
/**
@@ -228,6 +310,7 @@ public class SocketTestUtils {
*/
public static void testSendCrLfSingle(final int port, final CountDownLatch latch) {
Thread thread = new Thread(new Runnable() {
@Override
public void run() {
try {
Socket socket = new Socket(InetAddress.getByName("localhost"), port);
@@ -240,7 +323,8 @@ public class SocketTestUtils {
latch.await();
}
socket.close();
} catch (Exception e) {
}
catch (Exception e) {
e.printStackTrace();
}
}
@@ -254,6 +338,7 @@ public class SocketTestUtils {
*/
public static void testSendRaw(final int port) {
Thread thread = new Thread(new Runnable() {
@Override
public void run() {
try {
Socket socket = new Socket(InetAddress.getByName("localhost"), port);
@@ -261,7 +346,8 @@ public class SocketTestUtils {
outputStream.write(TEST_STRING.getBytes());
outputStream.write(TEST_STRING.getBytes());
socket.close();
} catch (Exception e) {
}
catch (Exception e) {
e.printStackTrace();
}
}
@@ -273,11 +359,14 @@ public class SocketTestUtils {
* Sends two serialized objects over the same socket.
* @param port
*/
public static void testSendSerialized(final int port) {
public static CountDownLatch testSendSerialized(final int port) {
final CountDownLatch testCompleteLatch = new CountDownLatch(1);
Thread thread = new Thread(new Runnable() {
@Override
public void run() {
Socket socket = null;
try {
Socket socket = new Socket(InetAddress.getByName("localhost"), port);
socket = new Socket(InetAddress.getByName("localhost"), port);
OutputStream outputStream = socket.getOutputStream();
ObjectOutputStream oos = new ObjectOutputStream(outputStream);
oos.writeObject(TEST_STRING);
@@ -285,21 +374,33 @@ public class SocketTestUtils {
oos = new ObjectOutputStream(outputStream);
oos.writeObject(TEST_STRING);
oos.flush();
Thread.sleep(1000000000L); // wait forever, but we're a daemon
} catch (Exception e) {
testCompleteLatch.await(10, TimeUnit.SECONDS);
}
catch (Exception e) {
e.printStackTrace();
}
finally {
if (socket != null) {
try {
socket.close();
}
catch (IOException e) { }
}
}
}
});
thread.setDaemon(true);
thread.start();
return testCompleteLatch;
}
/**
* Sends a large CRLF message with no CRLF.
*/
public static void testSendCrLfOverflow(final int port) {
public static CountDownLatch testSendCrLfOverflow(final int port) {
final CountDownLatch testCompleteLatch = new CountDownLatch(1);
Thread thread = new Thread(new Runnable() {
@Override
public void run() {
try {
Socket socket = new Socket(InetAddress.getByName("localhost"), port);
@@ -307,12 +408,15 @@ public class SocketTestUtils {
for (int i = 0; i < 1500; i++) {
writeByte(outputStream, 'x', true);
}
Thread.sleep(1000000000L); // wait forever, but we're a daemon
} catch (Exception e) { }
testCompleteLatch.await(10, TimeUnit.SECONDS);
socket.close();
}
catch (Exception e) { }
}
});
thread.setDaemon(true);
thread.start();
return testCompleteLatch;
}
public static void setLocalNicIfPossible(