From ced1a05df2b47daf629416b342d6eb00c7a11258 Mon Sep 17 00:00:00 2001 From: Gary Russell Date: Fri, 13 Dec 2013 20:49:54 +0200 Subject: [PATCH] INT-3233 NIO, Windows 7 and Java 7/8 JIRA: https://jira.springsource.org/browse/INT-3233 With Windows 7 and Java 7/8 closing a server channel does not close the underlying socket. Closing the Selector does close the socket. It is not clear why this is needed, given that a selector can be used for multiple sockets. However, we only use the server side selector for a single server socket so there is no detriment in closing it. Also close the selector on the client side (even though it is used for multiple sockets, because we are stopping the factory anyway and all sockets will be closed). However, closing the selector opens us up to 'ClosedSelectorException's in several places. Add catch blocks to deal with this exception, and only log an error if the factory is active. While debugging this issue, I found that a number of (older) tests left threads running, sockets open etc. INT-3233 Polishing - PR Comments --- build.gradle | 10 +- ...ternetProtocolReceivingChannelAdapter.java | 16 +- ...InternetProtocolSendingMessageHandler.java | 38 +++- .../connection/AbstractConnectionFactory.java | 4 +- .../CachingClientConnectionFactory.java | 5 + .../FailoverClientConnectionFactory.java | 9 + .../TcpNioClientConnectionFactory.java | 24 ++- .../TcpNioServerConnectionFactory.java | 29 ++- .../udp/UnicastReceivingChannelAdapter.java | 7 +- .../ip/udp/UnicastSendingMessageHandler.java | 38 +++- .../ip/tcp/TcpOutboundGatewayTests.java | 20 +- .../ConnectionFactoryShutDownTests.java | 5 + .../connection/TcpNioConnectionReadTests.java | 104 ++++++----- .../tcp/connection/TcpNioConnectionTests.java | 54 +++++- .../TcpNioConnectionWriteTests.java | 55 ++++-- .../tcp/serializer/DeserializationTests.java | 34 ++-- .../ip/tcp/serializer/SerializationTests.java | 45 +++-- .../DatagramPacketSendingHandlerTests.java | 18 +- .../integration/ip/udp/MultiClientTests.java | 46 ++++- .../integration/ip/udp/SyslogdTests.java | 4 +- .../ip/udp/UdpChannelAdapterTests.java | 29 ++- .../ip/udp/UdpMulticastEndToEndTests.java | 22 ++- .../ip/udp/UdpUnicastEndToEndTests.java | 9 +- .../ip/udp/testIp-out-multicast-context.xml | 4 +- .../integration/ip/util/SocketTestUtils.java | 172 ++++++++++++++---- 25 files changed, 611 insertions(+), 190 deletions(-) diff --git a/build.gradle b/build.gradle index c4ebc41d57..66b0fbe2f1 100644 --- a/build.gradle +++ b/build.gradle @@ -295,9 +295,11 @@ project('spring-integration-http') { compile ("net.java.dev.rome:rome:1.0.0", optional) testCompile project(":spring-integration-test") - // suppress deprecation warnings (@SuppressWarnings("deprecation") is not enough for javac) - compileJava.options.compilerArgs = ["${xLintArg},-deprecation"] } + + // suppress deprecation warnings (@SuppressWarnings("deprecation") is not enough for javac) + compileJava.options.compilerArgs = ["${xLintArg},-deprecation"] + } project('spring-integration-ip') { @@ -308,6 +310,10 @@ project('spring-integration-ip') { runtime project(":spring-integration-stream") testCompile project(":spring-integration-test") } + + // suppress deprecation warnings (@SuppressWarnings("deprecation") is not enough for javac) + compileJava.options.compilerArgs = ["${xLintArg},-deprecation"] + } project('spring-integration-jdbc') { diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/AbstractInternetProtocolReceivingChannelAdapter.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/AbstractInternetProtocolReceivingChannelAdapter.java index 8847124d53..062fe913fa 100644 --- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/AbstractInternetProtocolReceivingChannelAdapter.java +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/AbstractInternetProtocolReceivingChannelAdapter.java @@ -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; } /** diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/AbstractInternetProtocolSendingMessageHandler.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/AbstractInternetProtocolSendingMessageHandler.java index b107980e93..ca54bbfd04 100644 --- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/AbstractInternetProtocolSendingMessageHandler.java +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/AbstractInternetProtocolSendingMessageHandler.java @@ -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; + } + + } diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/AbstractConnectionFactory.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/AbstractConnectionFactory.java index 1fd86cc7cc..1efd1f2f9f 100644 --- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/AbstractConnectionFactory.java +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/AbstractConnectionFactory.java @@ -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 diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/CachingClientConnectionFactory.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/CachingClientConnectionFactory.java index c552b0fe90..9de5505964 100644 --- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/CachingClientConnectionFactory.java +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/CachingClientConnectionFactory.java @@ -44,6 +44,7 @@ public class CachingClientConnectionFactory extends AbstractClientConnectionFact this.targetConnectionFactory = target; pool = new SimplePool(poolSize, new SimplePool.PoolItemCallback() { + @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"); diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/FailoverClientConnectionFactory.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/FailoverClientConnectionFactory.java index 2b75454027..d68989383d 100644 --- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/FailoverClientConnectionFactory.java +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/FailoverClientConnectionFactory.java @@ -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) diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNioClientConnectionFactory.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNioClientConnectionFactory.java index 6935fbdd3b..877d4998e2 100644 --- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNioClientConnectionFactory.java +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNioClientConnectionFactory.java @@ -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); } diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNioServerConnectionFactory.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNioServerConnectionFactory.java index 6b97ac6bb0..53f2921e33 100644 --- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNioServerConnectionFactory.java +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNioServerConnectionFactory.java @@ -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; } diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/udp/UnicastReceivingChannelAdapter.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/udp/UnicastReceivingChannelAdapter.java index dab9fa892e..7cb919749c 100644 --- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/udp/UnicastReceivingChannelAdapter.java +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/udp/UnicastReceivingChannelAdapter.java @@ -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 message = null; try { @@ -222,6 +224,7 @@ public class UnicastReceivingChannelAdapter extends AbstractInternetProtocolRece } } + @Override public void setSoSendBufferSize(int soSendBufferSize) { this.soSendBufferSize = soSendBufferSize; } diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/udp/UnicastSendingMessageHandler.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/udp/UnicastSendingMessageHandler.java index 6310913cb1..bcc5da0d3b 100644 --- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/udp/UnicastSendingMessageHandler.java +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/udp/UnicastSendingMessageHandler.java @@ -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"; } diff --git a/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/TcpOutboundGatewayTests.java b/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/TcpOutboundGatewayTests.java index 21abfad147..d25585c6a3 100644 --- a/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/TcpOutboundGatewayTests.java +++ b/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/TcpOutboundGatewayTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2012 the original author or authors. + * Copyright 2002-2013 the original author or authors. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -86,10 +86,13 @@ public class TcpOutboundGatewayTests { AbstractConnectionFactory ccf = new TcpNetClientConnectionFactory("localhost", port); final CountDownLatch latch = new CountDownLatch(1); final AtomicBoolean done = new AtomicBoolean(); + final AtomicReference serverSocket = new AtomicReference(); Executors.newSingleThreadExecutor().execute(new Runnable() { + @Override public void run() { try { ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(port, 100); + serverSocket.set(server); latch.countDown(); List sockets = new ArrayList(); int i = 0; @@ -101,7 +104,8 @@ public class TcpOutboundGatewayTests { oos.writeObject("Reply" + (i++)); sockets.add(socket); } - } catch (Exception e) { + } + catch (Exception e) { if (!done.get()) { e.printStackTrace(); } @@ -141,6 +145,8 @@ public class TcpOutboundGatewayTests { for (int i = 0; i < 100; i++) { assertTrue(replies.remove("Reply" + i)); } + done.set(true); + serverSocket.get().close(); } @Test @@ -149,6 +155,7 @@ public class TcpOutboundGatewayTests { final CountDownLatch latch = new CountDownLatch(1); final AtomicBoolean done = new AtomicBoolean(); Executors.newSingleThreadExecutor().execute(new Runnable() { + @Override public void run() { try { ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(port, 10); @@ -202,6 +209,7 @@ public class TcpOutboundGatewayTests { final CountDownLatch latch = new CountDownLatch(1); final AtomicBoolean done = new AtomicBoolean(); Executors.newSingleThreadExecutor().execute(new Runnable() { + @Override public void run() { try { ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(port); @@ -240,6 +248,7 @@ public class TcpOutboundGatewayTests { for (int i = 0; i < 2; i++) { final int j = i; results[j] = (Executors.newSingleThreadExecutor().submit(new Callable(){ + @Override public Integer call() throws Exception { gateway.handleMessage(MessageBuilder.withPayload("Test" + j).build()); return 0; @@ -319,6 +328,7 @@ public class TcpOutboundGatewayTests { Executors.newSingleThreadExecutor().execute(new Runnable() { + @Override public void run() { try { ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(port); @@ -368,6 +378,7 @@ public class TcpOutboundGatewayTests { for (int i = 0; i < 2; i++) { final int j = i; results[j] = (Executors.newSingleThreadExecutor().submit(new Callable() { + @Override public Integer call() throws Exception { // increase the timeout after the first send if (j > 0) { @@ -419,6 +430,7 @@ public class TcpOutboundGatewayTests { Executors.newSingleThreadExecutor().execute(new Runnable() { + @Override public void run() { try { ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(port); @@ -498,6 +510,7 @@ public class TcpOutboundGatewayTests { Executors.newSingleThreadExecutor().execute(new Runnable() { + @Override public void run() { try { ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(port); @@ -633,14 +646,13 @@ public class TcpOutboundGatewayTests { Executors.newSingleThreadExecutor().execute(new Runnable() { + @Override public void run() { try { ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(port); latch.countDown(); - int i = 0; while (!done.get()) { Socket socket = server.accept(); - i++; while (!socket.isClosed()) { try { ObjectInputStream ois = new ObjectInputStream(socket.getInputStream()); diff --git a/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/ConnectionFactoryShutDownTests.java b/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/ConnectionFactoryShutDownTests.java index 4d90358adf..ae2ed88d46 100644 --- a/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/ConnectionFactoryShutDownTests.java +++ b/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/ConnectionFactoryShutDownTests.java @@ -22,6 +22,7 @@ import java.util.concurrent.Executor; import java.util.concurrent.TimeUnit; import org.junit.Test; + import org.springframework.integration.MessagingException; import org.springframework.util.StopWatch; @@ -36,13 +37,16 @@ public class ConnectionFactoryShutDownTests { public void testShutdownDoesntDeadlock() throws Exception { final AbstractConnectionFactory factory = new AbstractConnectionFactory(0) { + @Override public TcpConnection getConnection() throws Exception { return null; } @Override + @Deprecated public void close() { } + }; factory.setActive(true); Executor executor = factory.getTaskExecutor(); @@ -50,6 +54,7 @@ public class ConnectionFactoryShutDownTests { final CountDownLatch latch2 = new CountDownLatch(1); executor.execute(new Runnable() { + @Override public void run() { latch1.countDown(); try { diff --git a/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/TcpNioConnectionReadTests.java b/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/TcpNioConnectionReadTests.java index 1ab0f9d6e9..042931b6e8 100644 --- a/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/TcpNioConnectionReadTests.java +++ b/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/TcpNioConnectionReadTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2010 the original author or authors. + * Copyright 2002-2013 the original author or authors. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -20,9 +20,7 @@ import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertTrue; import static org.junit.Assert.fail; -import java.io.IOException; import java.net.Socket; -import java.net.UnknownHostException; import java.util.ArrayList; import java.util.List; import java.util.concurrent.CountDownLatch; @@ -32,6 +30,7 @@ import java.util.concurrent.TimeUnit; import javax.net.SocketFactory; import org.junit.Test; + import org.springframework.integration.Message; import org.springframework.integration.ip.tcp.serializer.AbstractByteArraySerializer; import org.springframework.integration.ip.tcp.serializer.ByteArrayCrLfSerializer; @@ -47,7 +46,7 @@ import org.springframework.integration.test.util.SocketUtils; */ public class TcpNioConnectionReadTests { - private CountDownLatch latch = new CountDownLatch(1); + private final CountDownLatch latch = new CountDownLatch(1); private AbstractServerConnectionFactory getConnectionFactory(int port, AbstractByteArraySerializer serializer, TcpListener listener) throws Exception { @@ -68,10 +67,6 @@ public class TcpNioConnectionReadTests { return scf; } - /** - * Test method for {@link org.springframework.integration.ip.tcp.NioSocketReader}. - */ - @SuppressWarnings("unchecked") @Test public void testReadLength() throws Exception { int port = SocketUtils.findAvailableServerSocket(); @@ -79,6 +74,7 @@ public class TcpNioConnectionReadTests { final List> responses = new ArrayList>(); final Semaphore semaphore = new Semaphore(0); AbstractServerConnectionFactory scf = getConnectionFactory(port, serializer,new TcpListener() { + @Override public boolean onMessage(Message message) { responses.add(message); semaphore.release(); @@ -88,7 +84,7 @@ public class TcpNioConnectionReadTests { // Fire up the sender. - SocketTestUtils.testSendLength(port, latch); + CountDownLatch done = SocketTestUtils.testSendLength(port, latch); latch.countDown(); assertTrue(semaphore.tryAcquire(1, 10000, TimeUnit.MILLISECONDS)); assertTrue(semaphore.tryAcquire(1, 10000, TimeUnit.MILLISECONDS)); @@ -97,7 +93,8 @@ public class TcpNioConnectionReadTests { new String((byte[]) responses.get(0).getPayload())); assertEquals("Data", SocketTestUtils.TEST_STRING + SocketTestUtils.TEST_STRING, new String((byte[]) responses.get(1).getPayload())); - scf.close(); + scf.stop(); + done.countDown(); } @@ -110,11 +107,15 @@ public class TcpNioConnectionReadTests { final List> responses = new ArrayList>(); 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) 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> responses = new ArrayList>(); 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) responses.get(0)).getPayload())); assertEquals("Data", SocketTestUtils.TEST_STRING + SocketTestUtils.TEST_STRING, new String(((Message) responses.get(1)).getPayload())); - scf.close(); + scf.stop(); + done.countDown(); } /** @@ -176,6 +177,7 @@ public class TcpNioConnectionReadTests { final List> responses = new ArrayList>(); 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) responses.get(0)).getPayload())); assertEquals("Data", SocketTestUtils.TEST_STRING + SocketTestUtils.TEST_STRING, new String(((Message) 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> responses = new ArrayList>(); final Semaphore semaphore = new Semaphore(0); final List added = new ArrayList(); final List removed = new ArrayList(); 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> responses = new ArrayList>(); final Semaphore semaphore = new Semaphore(0); final List added = new ArrayList(); final List removed = new ArrayList(); 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> responses = new ArrayList>(); final Semaphore semaphore = new Semaphore(0); final List added = new ArrayList(); final List removed = new ArrayList(); 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> responses = new ArrayList>(); final Semaphore semaphore = new Semaphore(0); final List added = new ArrayList(); final List removed = new ArrayList(); 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> responses = new ArrayList>(); final Semaphore semaphore = new Semaphore(0); final List added = new ArrayList(); final List removed = new ArrayList(); 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> responses = new ArrayList>(); final Semaphore semaphore = new Semaphore(0); final List added = new ArrayList(); final List removed = new ArrayList(); 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 added) diff --git a/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/TcpNioConnectionTests.java b/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/TcpNioConnectionTests.java index 0d8457eeef..6f9c0e276e 100644 --- a/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/TcpNioConnectionTests.java +++ b/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/TcpNioConnectionTests.java @@ -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 = new AtomicReference(); 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 = new AtomicReference(); 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 = new AtomicReference(); 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 fields = new ArrayList(); 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 future = exec.submit(new Callable() { + @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() { + @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 future = exec.submit(new Callable() { + @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() { + @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() { + @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() { + @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(); diff --git a/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/TcpNioConnectionWriteTests.java b/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/TcpNioConnectionWriteTests.java index 5aaaf763db..90cdcf63a2 100644 --- a/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/TcpNioConnectionWriteTests.java +++ b/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/TcpNioConnectionWriteTests.java @@ -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(); } /** diff --git a/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/serializer/DeserializationTests.java b/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/serializer/DeserializationTests.java index 7f9f85ae61..edcbfd4af9 100644 --- a/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/serializer/DeserializationTests.java +++ b/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/serializer/DeserializationTests.java @@ -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(); } } diff --git a/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/serializer/SerializationTests.java b/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/serializer/SerializationTests.java index e05374ee10..b51e1e38be 100644 --- a/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/serializer/SerializationTests.java +++ b/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/serializer/SerializationTests.java @@ -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(); } diff --git a/spring-integration-ip/src/test/java/org/springframework/integration/ip/udp/DatagramPacketSendingHandlerTests.java b/spring-integration-ip/src/test/java/org/springframework/integration/ip/udp/DatagramPacketSendingHandlerTests.java index 47aa20c135..c04f7aa611 100644 --- a/spring-integration-ip/src/test/java/org/springframework/integration/ip/udp/DatagramPacketSendingHandlerTests.java +++ b/spring-integration-ip/src/test/java/org/springframework/integration/ip/udp/DatagramPacketSendingHandlerTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2010 the original author or authors. + * Copyright 2002-2013 the original author or authors. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -33,6 +33,7 @@ import java.util.concurrent.TimeUnit; import org.apache.commons.logging.LogFactory; import org.junit.Ignore; import org.junit.Test; + import org.springframework.integration.Message; import org.springframework.integration.ip.IpHeaders; import org.springframework.integration.support.MessageBuilder; @@ -54,6 +55,7 @@ public class DatagramPacketSendingHandlerTests { final DatagramPacket receivedPacket = new DatagramPacket(buffer, buffer.length); final CountDownLatch latch = new CountDownLatch(1); Executors.newSingleThreadExecutor().execute(new Runnable() { + @Override public void run() { try { DatagramSocket socket = new DatagramSocket(testPort); @@ -78,7 +80,7 @@ public class DatagramPacketSendingHandlerTests { byte[] dest = new byte[length]; System.arraycopy(src, offset, dest, 0, length); assertEquals(payload, new String(dest)); - handler.shutDown(); + handler.stop(); } @Test @@ -97,7 +99,9 @@ public class DatagramPacketSendingHandlerTests { new UnicastSendingMessageHandler("localhost", testPort, true, true, "localhost", ackPort, 5000); handler.afterPropertiesSet(); + handler.start(); Executors.newSingleThreadExecutor().execute(new Runnable() { + @Override public void run() { try { DatagramSocket socket = new DatagramSocket(testPort); @@ -132,7 +136,7 @@ public class DatagramPacketSendingHandlerTests { byte[] dest = new byte[6]; System.arraycopy(src, offset+length-6, dest, 0, 6); assertEquals(payload, new String(dest)); - handler.shutDown(); + handler.stop(); } @Test @@ -144,6 +148,7 @@ public class DatagramPacketSendingHandlerTests { final CountDownLatch latch1 = new CountDownLatch(2); final CountDownLatch latch2 = new CountDownLatch(2); Runnable catcher = new Runnable() { + @Override public void run() { try { byte[] buffer = new byte[8]; @@ -183,7 +188,7 @@ public class DatagramPacketSendingHandlerTests { MulticastSendingMessageHandler handler = new MulticastSendingMessageHandler(multicastAddress, testPort); handler.handleMessage(MessageBuilder.withPayload(payload).build()); assertTrue(latch2.await(3000, TimeUnit.MILLISECONDS)); - handler.shutDown(); + handler.stop(); } @Test @@ -200,6 +205,7 @@ public class DatagramPacketSendingHandlerTests { final CountDownLatch latch1 = new CountDownLatch(2); final CountDownLatch latch2 = new CountDownLatch(2); Runnable catcher = new Runnable() { + @Override public void run() { try { byte[] buffer = new byte[1000]; @@ -249,9 +255,11 @@ public class DatagramPacketSendingHandlerTests { new MulticastSendingMessageHandler(multicastAddress, testPort, true, true, "localhost", ackPort, 500000); handler.setMinAcksForSuccess(2); + handler.afterPropertiesSet(); + handler.start(); handler.handleMessage(MessageBuilder.withPayload(payload).build()); assertTrue(latch2.await(10000, TimeUnit.MILLISECONDS)); - handler.shutDown(); + handler.stop(); } } diff --git a/spring-integration-ip/src/test/java/org/springframework/integration/ip/udp/MultiClientTests.java b/spring-integration-ip/src/test/java/org/springframework/integration/ip/udp/MultiClientTests.java index 6210aa1b32..b08726d171 100644 --- a/spring-integration-ip/src/test/java/org/springframework/integration/ip/udp/MultiClientTests.java +++ b/spring-integration-ip/src/test/java/org/springframework/integration/ip/udp/MultiClientTests.java @@ -16,10 +16,13 @@ package org.springframework.integration.ip.udp; import static org.junit.Assert.assertNotNull; -import org.junit.Assert; +import java.util.concurrent.atomic.AtomicBoolean; + +import org.junit.Assert; import org.junit.Ignore; import org.junit.Test; + import org.springframework.integration.Message; import org.springframework.integration.channel.QueueChannel; import org.springframework.integration.ip.util.SocketTestUtils; @@ -45,7 +48,8 @@ import org.springframework.integration.test.util.SocketUtils; public class MultiClientTests { @SuppressWarnings("unchecked") - @Test @Ignore + @Test + @Ignore public void testNoAck() throws Exception { final String payload = largePayload(1000); final UnicastReceivingChannelAdapter adapter = @@ -57,15 +61,24 @@ public class MultiClientTests { adapter.start(); final QueueChannel queueIn = new QueueChannel(1000); SocketTestUtils.waitListening(adapter); + + final AtomicBoolean done = new AtomicBoolean(); + for (int i = 0; i < drivers; i++) { Thread t = new Thread( new Runnable() { + @Override public void run() { UnicastSendingMessageHandler sender = new UnicastSendingMessageHandler( "localhost", adapter.getPort()); + sender.start(); while (true) { Message message = queueIn.receive(); sender.handleMessage(message); + if (done.get()) { + break; + } } + sender.stop(); }}); t.setDaemon(true); t.start(); @@ -79,12 +92,13 @@ public class MultiClientTests { Assert.assertEquals(payload, new String(messageOut.getPayload())); } adapter.stop(); + done.set(true); } @SuppressWarnings("unchecked") - @Test @Ignore + @Test + @Ignore public void testAck() throws Exception { - Thread.sleep(1000); final String payload = largePayload(1000); final UnicastReceivingChannelAdapter adapter = new UnicastReceivingChannelAdapter(SocketUtils.findAvailableUdpSocket(), false); @@ -95,19 +109,28 @@ public class MultiClientTests { adapter.start(); final QueueChannel queueIn = new QueueChannel(1000); SocketTestUtils.waitListening(adapter); + + final AtomicBoolean done = new AtomicBoolean(); + for (int i = 0; i < drivers; i++) { final int j = i; Thread t = new Thread( new Runnable() { + @Override public void run() { UnicastSendingMessageHandler sender = new UnicastSendingMessageHandler( "localhost", adapter.getPort(), false, true, "localhost", SocketUtils.findAvailableUdpSocket(adapter.getPort() + j + 1000), 10000); + sender.start(); while (true) { Message message = queueIn.receive(); sender.handleMessage(message); + if (done.get()) { + break; + } } + sender.stop(); }}); t.setDaemon(true); t.start(); @@ -121,12 +144,13 @@ public class MultiClientTests { Assert.assertEquals(payload, new String(messageOut.getPayload())); } adapter.stop(); + done.set(true); } @SuppressWarnings("unchecked") - @Test @Ignore + @Test + @Ignore public void testAckWithLength() throws Exception { - Thread.sleep(1000); final String payload = largePayload(1000); final UnicastReceivingChannelAdapter adapter = new UnicastReceivingChannelAdapter(SocketUtils.findAvailableUdpSocket(), true); @@ -137,19 +161,28 @@ public class MultiClientTests { adapter.start(); final QueueChannel queueIn = new QueueChannel(1000); SocketTestUtils.waitListening(adapter); + + final AtomicBoolean done = new AtomicBoolean(); + for (int i = 0; i < drivers; i++) { final int j = i; Thread t = new Thread( new Runnable() { + @Override public void run() { UnicastSendingMessageHandler sender = new UnicastSendingMessageHandler( "localhost", adapter.getPort(), true, true, "localhost", SocketUtils.findAvailableUdpSocket(adapter.getPort() + j + 1100), 10000); + sender.start(); while (true) { Message message = queueIn.receive(); sender.handleMessage(message); + if (done.get()) { + break; + } } + sender.stop(); }}); t.setDaemon(true); t.start(); @@ -163,6 +196,7 @@ public class MultiClientTests { Assert.assertEquals(payload, new String(messageOut.getPayload())); } adapter.stop(); + done.set(true); } private String largePayload(int n) { diff --git a/spring-integration-ip/src/test/java/org/springframework/integration/ip/udp/SyslogdTests.java b/spring-integration-ip/src/test/java/org/springframework/integration/ip/udp/SyslogdTests.java index edad25436e..30afa46a1b 100644 --- a/spring-integration-ip/src/test/java/org/springframework/integration/ip/udp/SyslogdTests.java +++ b/spring-integration-ip/src/test/java/org/springframework/integration/ip/udp/SyslogdTests.java @@ -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(); } } diff --git a/spring-integration-ip/src/test/java/org/springframework/integration/ip/udp/UdpChannelAdapterTests.java b/spring-integration-ip/src/test/java/org/springframework/integration/ip/udp/UdpChannelAdapterTests.java index 129a9969de..c1e7e11b22 100644 --- a/spring-integration-ip/src/test/java/org/springframework/integration/ip/udp/UdpChannelAdapterTests.java +++ b/spring-integration-ip/src/test/java/org/springframework/integration/ip/udp/UdpChannelAdapterTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2012 the original author or authors. + * Copyright 2002-2013 the original author or authors. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -32,6 +32,7 @@ import java.util.concurrent.atomic.AtomicReference; import org.apache.commons.logging.LogFactory; import org.junit.Ignore; import org.junit.Test; + import org.springframework.integration.Message; import org.springframework.integration.channel.DirectChannel; import org.springframework.integration.channel.QueueChannel; @@ -65,9 +66,12 @@ public class UdpChannelAdapterTests { DatagramPacketMessageMapper mapper = new DatagramPacketMessageMapper(); DatagramPacket packet = mapper.fromMessage(message); packet.setSocketAddress(new InetSocketAddress("localhost", port)); - new DatagramSocket(SocketUtils.findAvailableUdpSocket()).send(packet); + DatagramSocket datagramSocket = new DatagramSocket(SocketUtils.findAvailableUdpSocket()); + datagramSocket.send(packet); + datagramSocket.close(); Message receivedMessage = (Message) 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 message = MessageBuilder.withPayload("ABCD".getBytes()).build(); handler.handleMessage(message); Message receivedMessage = (Message) 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 receivedMessage = (Message) 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 receivedMessage = (Message) 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 { diff --git a/spring-integration-ip/src/test/java/org/springframework/integration/ip/udp/UdpMulticastEndToEndTests.java b/spring-integration-ip/src/test/java/org/springframework/integration/ip/udp/UdpMulticastEndToEndTests.java index de52414904..fe4212c3b7 100644 --- a/spring-integration-ip/src/test/java/org/springframework/integration/ip/udp/UdpMulticastEndToEndTests.java +++ b/spring-integration-ip/src/test/java/org/springframework/integration/ip/udp/UdpMulticastEndToEndTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2010 the original author or authors. + * Copyright 2002-2013 the original author or authors. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -41,12 +41,12 @@ import org.springframework.integration.support.channel.ChannelResolver; /** * Sends and receives a simple message through to the Udp channel adapters. * If run as a JUnit just sends one message and terminates (see console). - * - * If run from main(), + * + * If run from main(), * hangs around for a couple of minutes to allow console interaction (enter a message on the - * console and you should see it go through the outbound context, over UDP, and + * console and you should see it go through the outbound context, over UDP, and * received in the other context (and written back to the console). - * + * * @author Gary Russell * @since 2.0 */ @@ -58,13 +58,13 @@ public class UdpMulticastEndToEndTests implements Runnable { private CountDownLatch sentFirst = new CountDownLatch(1); - private CountDownLatch firstReceived = new CountDownLatch(1); + private final CountDownLatch firstReceived = new CountDownLatch(1); - private CountDownLatch doneProcessing = new CountDownLatch(1); + private final CountDownLatch doneProcessing = new CountDownLatch(1); private boolean okToRun = true; - private CountDownLatch readyToReceive = new CountDownLatch(1); + private final CountDownLatch readyToReceive = new CountDownLatch(1); private static long hangAroundFor = 0; @@ -77,7 +77,7 @@ public class UdpMulticastEndToEndTests implements Runnable { t.start(); // launch the receiver AbstractApplicationContext applicationContext = new ClassPathXmlApplicationContext( "testIp-out-multicast-context.xml", - UdpMulticastEndToEndTests.class); + UdpMulticastEndToEndTests.class); launcher.launchSender(applicationContext); applicationContext.stop(); } @@ -122,6 +122,7 @@ public class UdpMulticastEndToEndTests implements Runnable { /** * Instantiate the receiving context */ + @Override @SuppressWarnings("unchecked") public void run() { AbstractApplicationContext ctx = new ClassPathXmlApplicationContext( @@ -129,7 +130,7 @@ public class UdpMulticastEndToEndTests implements Runnable { UdpMulticastEndToEndTests.class); while (okToRun) { try { - readyToReceive.countDown(); + readyToReceive.countDown(); sentFirst.await(); } catch (InterruptedException e) { @@ -146,6 +147,7 @@ public class UdpMulticastEndToEndTests implements Runnable { } } ctx.stop(); + ctx.close(); } diff --git a/spring-integration-ip/src/test/java/org/springframework/integration/ip/udp/UdpUnicastEndToEndTests.java b/spring-integration-ip/src/test/java/org/springframework/integration/ip/udp/UdpUnicastEndToEndTests.java index a090202d13..23036a1d3c 100644 --- a/spring-integration-ip/src/test/java/org/springframework/integration/ip/udp/UdpUnicastEndToEndTests.java +++ b/spring-integration-ip/src/test/java/org/springframework/integration/ip/udp/UdpUnicastEndToEndTests.java @@ -16,8 +16,8 @@ package org.springframework.integration.ip.udp; -import static org.junit.Assert.assertNotNull; import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertNotNull; import static org.junit.Assert.assertTrue; import static org.junit.Assert.fail; @@ -28,6 +28,7 @@ import java.util.concurrent.TimeUnit; import org.junit.Before; import org.junit.Test; + import org.springframework.context.ApplicationContext; import org.springframework.context.support.AbstractApplicationContext; import org.springframework.context.support.ClassPathXmlApplicationContext; @@ -142,7 +143,9 @@ public class UdpUnicastEndToEndTests implements Runnable { * Instantiate the receiving context */ @SuppressWarnings("unchecked") + @Override public void run() { + @SuppressWarnings("resource") AbstractApplicationContext ctx = new ClassPathXmlApplicationContext( "testIp-in-context.xml", UdpUnicastEndToEndTests.class); UnicastReceivingChannelAdapter inbound = ctx.getBean(UnicastReceivingChannelAdapter.class); @@ -154,7 +157,8 @@ public class UdpUnicastEndToEndTests implements Runnable { throw new RuntimeException("Failed to start listening"); } } - } catch (Exception e) { } + } + catch (Exception e) { } while (okToRun) { try { readyToReceive.countDown(); @@ -178,6 +182,7 @@ public class UdpUnicastEndToEndTests implements Runnable { } } ctx.stop(); + ctx.close(); } diff --git a/spring-integration-ip/src/test/java/org/springframework/integration/ip/udp/testIp-out-multicast-context.xml b/spring-integration-ip/src/test/java/org/springframework/integration/ip/udp/testIp-out-multicast-context.xml index e449e17b22..96e938417a 100644 --- a/spring-integration-ip/src/test/java/org/springframework/integration/ip/udp/testIp-out-multicast-context.xml +++ b/spring-integration-ip/src/test/java/org/springframework/integration/ip/udp/testIp-out-multicast-context.xml @@ -14,9 +14,7 @@ http://www.springframework.org/schema/integration/ip/spring-integration-ip.xsd"> - - - + diff --git a/spring-integration-ip/src/test/java/org/springframework/integration/ip/util/SocketTestUtils.java b/spring-integration-ip/src/test/java/org/springframework/integration/ip/util/SocketTestUtils.java index 8efdca0271..4b7d029d06 100644 --- a/spring-integration-ip/src/test/java/org/springframework/integration/ip/util/SocketTestUtils.java +++ b/spring-integration-ip/src/test/java/org/springframework/integration/ip/util/SocketTestUtils.java @@ -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(