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

Merge remote-tracking branch 'upstream/master' into 4.0.0-WIP

Browse Source 
Conflicts:
	build.gradle
	gradle.properties
	spring-integration-core/src/main/java/org/springframework/integration/config/RouterFactoryBean.java
	spring-integration-core/src/test/java/org/springframework/integration/channel/config/DispatchingChannelParserTests.java
	spring-integration-core/src/test/java/org/springframework/integration/gateway/GatewayProxyFactoryBeanTests.java
	spring-integration-core/src/test/java/org/springframework/integration/router/config/PayloadTypeRouterParserTests.java
	spring-integration-core/src/test/java/org/springframework/integration/transformer/MessageTransformingChannelInterceptorTests.java
	spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/ConnectionFactoryShutDownTests.java
	spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/TcpNioConnectionReadTests.java
	spring-integration-ip/src/test/java/org/springframework/integration/ip/udp/DatagramPacketSendingHandlerTests.java
	spring-integration-ip/src/test/java/org/springframework/integration/ip/udp/MultiClientTests.java
	spring-integration-ip/src/test/java/org/springframework/integration/ip/udp/UdpChannelAdapterTests.java
	spring-integration-jdbc/src/main/java/org/springframework/integration/jdbc/StoredProcMessageHandler.java
	spring-integration-jdbc/src/main/java/org/springframework/integration/jdbc/StoredProcOutboundGateway.java
	spring-integration-jdbc/src/main/java/org/springframework/integration/jdbc/StoredProcPollingChannelAdapter.java
	spring-integration-xmpp/src/main/java/org/springframework/integration/xmpp/core/AbstractXmppConnectionAwareEndpoint.java

Resolved.
This commit is contained in:
Gary Russell
2013-12-16 11:02:40 -05:00
parent 5ddecd13e7 5ed110c2a0
commit 2baea786b1
77 changed files with 1626 additions and 1685 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

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

@@ -677,7 +677,8 @@ public class ParserUnitTests {
TcpConnectionSupport connection = mock(TcpConnectionSupport.class);
TcpConnectionEvent event = new TcpConnectionOpenEvent(connection, "foo");
this.eventAdapter.setEventTypes(new Class[] {TcpConnectionEvent.class});
Class<TcpConnectionEvent>[] types = (Class<TcpConnectionEvent>[]) new Class<?>[]{TcpConnectionEvent.class};
this.eventAdapter.setEventTypes(types);
this.eventAdapter.onApplicationEvent(event);
assertNull(this.eventChannel.receive(0));
this.eventAdapter.start();

24
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.
@@ -85,10 +85,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;
@@ -100,7 +103,8 @@ public class TcpOutboundGatewayTests {
oos.writeObject("Reply" + (i++));
sockets.add(socket);
}
} catch (Exception e) {
}
catch (Exception e) {
if (!done.get()) {
e.printStackTrace();
}
@@ -140,6 +144,8 @@ public class TcpOutboundGatewayTests {
for (int i = 0; i < 100; i++) {
assertTrue(replies.remove("Reply" + i));
}
done.set(true);
serverSocket.get().close();
}
@Test
@@ -148,6 +154,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);
@@ -201,6 +208,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);
@@ -235,10 +243,11 @@ public class TcpOutboundGatewayTests {
gateway.setRequiresReply(true);
gateway.setOutputChannel(replyChannel);
@SuppressWarnings("unchecked")
Future<Integer>[] results = new Future[2];
Future<Integer>[] results = (Future<Integer>[]) new Future<?>[2];
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;
@@ -318,6 +327,7 @@ public class TcpOutboundGatewayTests {
Executors.newSingleThreadExecutor().execute(new Runnable() {
@Override
public void run() {
try {
ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(port);
@@ -363,10 +373,11 @@ public class TcpOutboundGatewayTests {
gateway.setOutputChannel(replyChannel);
gateway.setRemoteTimeout(500);
@SuppressWarnings("unchecked")
Future<Integer>[] results = new Future[2];
Future<Integer>[] results = (Future<Integer>[]) new Future<?>[2];
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) {
@@ -418,6 +429,7 @@ public class TcpOutboundGatewayTests {
Executors.newSingleThreadExecutor().execute(new Runnable() {
@Override
public void run() {
try {
ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(port);
@@ -497,6 +509,7 @@ public class TcpOutboundGatewayTests {
Executors.newSingleThreadExecutor().execute(new Runnable() {
@Override
public void run() {
try {
ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(port);
@@ -632,14 +645,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.messaging.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 {

3
spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/TcpConnectionEventListenerTests.java
View File

@@ -65,7 +65,8 @@ public class TcpConnectionEventListenerTests {
TcpConnectionEventListeningMessageProducer eventProducer = new TcpConnectionEventListeningMessageProducer();
QueueChannel outputChannel = new QueueChannel();
eventProducer.setOutputChannel(outputChannel);
eventProducer.setEventTypes(new Class[] {FooEvent.class, BarEvent.class});
Class<?>[] eventTypes = new Class<?>[]{FooEvent.class, BarEvent.class};
eventProducer.setEventTypes((Class<? extends TcpConnectionEvent>[]) eventTypes);
eventProducer.afterPropertiesSet();
eventProducer.start();
TcpConnectionSupport connection = Mockito.mock(TcpConnectionSupport.class);

110
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,7 +30,7 @@ import java.util.concurrent.TimeUnit;
import javax.net.SocketFactory;
import org.junit.Test;
import org.springframework.messaging.Message;
import org.springframework.integration.ip.tcp.serializer.AbstractByteArraySerializer;
import org.springframework.integration.ip.tcp.serializer.ByteArrayCrLfSerializer;
import org.springframework.integration.ip.tcp.serializer.ByteArrayLengthHeaderSerializer;
@@ -40,6 +38,7 @@ import org.springframework.integration.ip.tcp.serializer.ByteArrayStxEtxSerializ
import org.springframework.integration.ip.util.SocketTestUtils;
import org.springframework.integration.ip.util.TestingUtilities;
import org.springframework.integration.test.util.SocketUtils;
import org.springframework.messaging.Message;
/**
* @author Gary Russell
@@ -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,16 +84,17 @@ 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));
assertEquals("Did not receive data", 2, responses.size());
assertEquals("Data", SocketTestUtils.TEST_STRING + SocketTestUtils.TEST_STRING,
new String(((Message<byte[]>) responses.get(0)).getPayload()));
new String((byte[]) responses.get(0).getPayload()));
assertEquals("Data", SocketTestUtils.TEST_STRING + SocketTestUtils.TEST_STRING,
new String(((Message<byte[]>) responses.get(1)).getPayload()));
scf.close();
new String((byte[]) responses.get(1).getPayload()));
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();
}

20
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,10 +33,11 @@ import java.util.concurrent.TimeUnit;
import org.apache.commons.logging.LogFactory;
import org.junit.Ignore;
import org.junit.Test;
import org.springframework.messaging.Message;
import org.springframework.integration.ip.IpHeaders;
import org.springframework.integration.support.MessageBuilder;
import org.springframework.integration.test.util.SocketUtils;
import org.springframework.messaging.Message;
/**
* @author Mark Fisher
@@ -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();
}
}

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

@@ -16,15 +16,18 @@
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.messaging.Message;
import org.springframework.integration.channel.QueueChannel;
import org.springframework.integration.ip.util.SocketTestUtils;
import org.springframework.integration.support.MessageBuilder;
import org.springframework.integration.test.util.SocketUtils;
import org.springframework.messaging.Message;
/**
@@ -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();
}
}

33
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,15 +32,16 @@ import java.util.concurrent.atomic.AtomicReference;
import org.apache.commons.logging.LogFactory;
import org.junit.Ignore;
import org.junit.Test;
import org.springframework.messaging.Message;
import org.springframework.integration.channel.DirectChannel;
import org.springframework.integration.channel.QueueChannel;
import org.springframework.messaging.SubscribableChannel;
import org.springframework.integration.handler.ServiceActivatingHandler;
import org.springframework.integration.ip.IpHeaders;
import org.springframework.integration.ip.util.SocketTestUtils;
import org.springframework.integration.support.MessageBuilder;
import org.springframework.integration.test.util.SocketUtils;
import org.springframework.messaging.Message;
import org.springframework.messaging.SubscribableChannel;
/**
*
@@ -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 {

4
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.
@@ -122,6 +122,7 @@ public class UdpMulticastEndToEndTests implements Runnable {
/**
* Instantiate the receiving context
*/
@Override
@SuppressWarnings("unchecked")
public void run() {
AbstractApplicationContext ctx = new ClassPathXmlApplicationContext(
@@ -146,6 +147,7 @@ public class UdpMulticastEndToEndTests implements Runnable {
}
}
ctx.stop();
ctx.close();
}

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

@@ -143,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);
@@ -155,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();
@@ -179,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(