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(