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

INT-4366: Fix MulticastSendingMessageHandler (#2329)

Browse Source 
* INT-4366: Fix MulticastSendingMessageHandler

JIRA: https://jira.spring.io/browse/INT-4366

Fix race condition in the `MulticastSendingMessageHandler` around
`multicastSocket` and super `socket` properties.

* Synchronize around `this` and check for the `multicastSocket == null`.
This let the `MulticastSendingMessageHandler` to fully configure and
prepare the socket for use.
* Remove `socket.setInterface(whichNic)` since it is populated by the
`InetSocketAddress` ctor before

**Cherry-pick to 4.3.x**

* Fix thread leaks in TCP/IP tests
This commit is contained in:
Artem Bilan
2018-01-19 12:34:57 -05:00
committed by Gary Russell
parent 8aa91d1db0
commit c3b64dc1ac
13 changed files with 262 additions and 236 deletions
Download Patch File Download Diff File Expand all files Collapse all files

66
spring-integration-ip/src/main/java/org/springframework/integration/ip/udp/MulticastSendingMessageHandler.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2001-2016 the original author or authors.
* Copyright 2001-2018 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.
@@ -38,6 +38,8 @@ import org.springframework.messaging.MessageHandler;
* determine success.
*
* @author Gary Russell
* @author Artem Bilan
*
* @since 2.0
*/
public class MulticastSendingMessageHandler extends UnicastSendingMessageHandler {
@@ -126,49 +128,45 @@ public class MulticastSendingMessageHandler extends UnicastSendingMessageHandler
@Override
protected DatagramSocket getSocket() throws IOException {
if (this.getTheSocket() == null) {
if (this.multicastSocket == null) {
synchronized (this) {
createSocket();
if (this.multicastSocket == null) {
createSocket();
}
}
}
return this.getTheSocket();
return getTheSocket();
}
private void createSocket() throws IOException {
if (this.getTheSocket() == null) {
MulticastSocket socket;
if (this.isAcknowledge()) {
int ackPort = this.getAckPort();
if (this.localAddress == null) {
socket = ackPort == 0 ? new MulticastSocket() : new MulticastSocket(ackPort);
}
else {
InetAddress whichNic = InetAddress.getByName(this.localAddress);
socket = new MulticastSocket(new InetSocketAddress(whichNic, ackPort));
}
if (getSoReceiveBufferSize() > 0) {
socket.setReceiveBufferSize(this.getSoReceiveBufferSize());
}
if (logger.isDebugEnabled()) {
logger.debug("Listening for acks on port: " + socket.getLocalPort());
}
setSocket(socket);
updateAckAddress();
MulticastSocket socket;
if (isAcknowledge()) {
int ackPort = getAckPort();
if (this.localAddress == null) {
socket = ackPort == 0 ? new MulticastSocket() : new MulticastSocket(ackPort);
}
else {
socket = new MulticastSocket();
setSocket(socket);
}
if (this.timeToLive >= 0) {
socket.setTimeToLive(this.timeToLive);
}
setSocketAttributes(socket);
if (this.localAddress != null) {
InetAddress whichNic = InetAddress.getByName(this.localAddress);
socket.setInterface(whichNic);
socket = new MulticastSocket(new InetSocketAddress(whichNic, ackPort));
}
this.multicastSocket = socket;
if (getSoReceiveBufferSize() > 0) {
socket.setReceiveBufferSize(getSoReceiveBufferSize());
}
if (logger.isDebugEnabled()) {
logger.debug("Listening for acks on port: " + socket.getLocalPort());
}
setSocket(socket);
updateAckAddress();
}
else {
socket = new MulticastSocket();
setSocket(socket);
}
if (this.timeToLive >= 0) {
socket.setTimeToLive(this.timeToLive);
}
setSocketAttributes(socket);
this.multicastSocket = socket;
}
@@ -178,7 +176,7 @@ public class MulticastSendingMessageHandler extends UnicastSendingMessageHandler
* @param minAcksForSuccess The minimum number of acks that will represent success.
*/
public void setMinAcksForSuccess(int minAcksForSuccess) {
this.setAckCounter(minAcksForSuccess);
setAckCounter(minAcksForSuccess);
}
/**

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

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2016 the original author or authors.
* Copyright 2002-2018 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.
@@ -28,7 +28,6 @@ import java.net.Socket;
import java.util.HashSet;
import java.util.Set;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;
@@ -39,6 +38,7 @@ import javax.net.SocketFactory;
import org.junit.Test;
import org.springframework.beans.factory.BeanFactory;
import org.springframework.core.task.SimpleAsyncTaskExecutor;
import org.springframework.integration.channel.DirectChannel;
import org.springframework.integration.channel.QueueChannel;
import org.springframework.integration.handler.ServiceActivatingHandler;
@@ -56,6 +56,8 @@ import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler;
/**
* @author Gary Russell
* @author Artem Bilan
*
* @since 2.0
*/
public class TcpInboundGatewayTests {
@@ -119,30 +121,31 @@ public class TcpInboundGatewayTests {
final CountDownLatch latch2 = new CountDownLatch(1);
final CountDownLatch latch3 = new CountDownLatch(1);
final AtomicBoolean done = new AtomicBoolean();
Executors.newSingleThreadExecutor().execute(() -> {
try {
ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(0, 10);
port.set(server.getLocalPort());
latch1.countDown();
Socket socket = server.accept();
socket.getOutputStream().write("Test1\r\nTest2\r\n".getBytes());
byte[] bytes = new byte[12];
readFully(socket.getInputStream(), bytes);
assertEquals("Echo:Test1\r\n", new String(bytes));
readFully(socket.getInputStream(), bytes);
assertEquals("Echo:Test2\r\n", new String(bytes));
latch2.await();
socket.close();
server.close();
done.set(true);
latch3.countDown();
}
catch (Exception e) {
if (!done.get()) {
e.printStackTrace();
}
}
});
new SimpleAsyncTaskExecutor()
.execute(() -> {
try {
ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(0, 10);
port.set(server.getLocalPort());
latch1.countDown();
Socket socket = server.accept();
socket.getOutputStream().write("Test1\r\nTest2\r\n".getBytes());
byte[] bytes = new byte[12];
readFully(socket.getInputStream(), bytes);
assertEquals("Echo:Test1\r\n", new String(bytes));
readFully(socket.getInputStream(), bytes);
assertEquals("Echo:Test2\r\n", new String(bytes));
latch2.await();
socket.close();
server.close();
done.set(true);
latch3.countDown();
}
catch (Exception e) {
if (!done.get()) {
e.printStackTrace();
}
}
});
assertTrue(latch1.await(10, TimeUnit.SECONDS));
AbstractClientConnectionFactory ccf = new TcpNetClientConnectionFactory("localhost", port.get());
ccf.setSingleUse(false);

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

@@ -43,7 +43,6 @@ import java.util.Map;
import java.util.Set;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
@@ -61,6 +60,8 @@ import org.mockito.Mockito;
import org.springframework.beans.factory.BeanFactory;
import org.springframework.core.serializer.DefaultDeserializer;
import org.springframework.core.serializer.DefaultSerializer;
import org.springframework.core.task.AsyncTaskExecutor;
import org.springframework.core.task.SimpleAsyncTaskExecutor;
import org.springframework.expression.EvaluationContext;
import org.springframework.expression.Expression;
import org.springframework.expression.spel.standard.SpelExpressionParser;
@@ -90,6 +91,8 @@ public class TcpOutboundGatewayTests {
private static final Log logger = LogFactory.getLog(TcpOutboundGatewayTests.class);
private AsyncTaskExecutor executor = new SimpleAsyncTaskExecutor();
@ClassRule
public static LongRunningIntegrationTest longTests = new LongRunningIntegrationTest();
@@ -101,13 +104,13 @@ public class TcpOutboundGatewayTests {
public void testGoodNetSingle() throws Exception {
final CountDownLatch latch = new CountDownLatch(1);
final AtomicBoolean done = new AtomicBoolean();
final AtomicReference<ServerSocket> serverSocket = new AtomicReference<ServerSocket>();
Executors.newSingleThreadExecutor().execute(() -> {
final AtomicReference<ServerSocket> serverSocket = new AtomicReference<>();
this.executor.execute(() -> {
try {
ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(0, 100);
serverSocket.set(server);
latch.countDown();
List<Socket> sockets = new ArrayList<Socket>();
List<Socket> sockets = new ArrayList<>();
int i = 0;
while (true) {
Socket socket = server.accept();
@@ -165,8 +168,8 @@ public class TcpOutboundGatewayTests {
public void testGoodNetMultiplex() throws Exception {
final CountDownLatch latch = new CountDownLatch(1);
final AtomicBoolean done = new AtomicBoolean();
final AtomicReference<ServerSocket> serverSocket = new AtomicReference<ServerSocket>();
Executors.newSingleThreadExecutor().execute(() -> {
final AtomicReference<ServerSocket> serverSocket = new AtomicReference<>();
this.executor.execute(() -> {
try {
ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(0, 10);
serverSocket.set(server);
@@ -220,8 +223,8 @@ public class TcpOutboundGatewayTests {
public void testGoodNetTimeout() throws Exception {
final CountDownLatch latch = new CountDownLatch(1);
final AtomicBoolean done = new AtomicBoolean();
final AtomicReference<ServerSocket> serverSocket = new AtomicReference<ServerSocket>();
Executors.newSingleThreadExecutor().execute(() -> {
final AtomicReference<ServerSocket> serverSocket = new AtomicReference<>();
this.executor.execute(() -> {
try {
ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(0);
serverSocket.set(server);
@@ -260,12 +263,12 @@ public class TcpOutboundGatewayTests {
Future<Integer>[] results = (Future<Integer>[]) new Future<?>[2];
for (int i = 0; i < 2; i++) {
final int j = i;
results[j] = (Executors.newSingleThreadExecutor().submit(() -> {
results[j] = (this.executor.submit(() -> {
gateway.handleMessage(MessageBuilder.withPayload("Test" + j).build());
return 0;
}));
}
Set<String> replies = new HashSet<String>();
Set<String> replies = new HashSet<>();
int timeouts = 0;
for (int i = 0; i < 2; i++) {
try {
@@ -344,7 +347,7 @@ public class TcpOutboundGatewayTests {
final AtomicReference<String> lastReceived = new AtomicReference<String>();
final CountDownLatch serverLatch = new CountDownLatch(2);
Executors.newSingleThreadExecutor().execute(() -> {
this.executor.execute(() -> {
try {
latch.countDown();
int i = 0;
@@ -398,7 +401,7 @@ public class TcpOutboundGatewayTests {
for (int i = 0; i < 2; i++) {
final int j = i;
results[j] = (Executors.newSingleThreadExecutor().submit(() -> {
results[j] = (this.executor.submit(() -> {
gateway.handleMessage(MessageBuilder.withPayload("Test" + j).build());
return j;
}));
@@ -442,7 +445,7 @@ public class TcpOutboundGatewayTests {
final AtomicBoolean done = new AtomicBoolean();
final CountDownLatch serverLatch = new CountDownLatch(1);
Executors.newSingleThreadExecutor().execute(() -> {
this.executor.execute(() -> {
try {
ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(0);
serverSocket.set(server);
@@ -517,12 +520,12 @@ public class TcpOutboundGatewayTests {
@Test
public void testFailoverCached() throws Exception {
final AtomicReference<ServerSocket> serverSocket = new AtomicReference<ServerSocket>();
final AtomicReference<ServerSocket> serverSocket = new AtomicReference<>();
final CountDownLatch latch = new CountDownLatch(1);
final AtomicBoolean done = new AtomicBoolean();
final CountDownLatch serverLatch = new CountDownLatch(1);
Executors.newSingleThreadExecutor().execute(() -> {
this.executor.execute(() -> {
try {
ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(0);
serverSocket.set(server);
@@ -667,11 +670,11 @@ public class TcpOutboundGatewayTests {
final ServerSocket server) throws Exception {
final CountDownLatch latch = new CountDownLatch(1);
final AtomicBoolean done = new AtomicBoolean();
final AtomicReference<String> lastReceived = new AtomicReference<String>();
final AtomicReference<String> lastReceived = new AtomicReference<>();
final CountDownLatch serverLatch = new CountDownLatch(1);
Executors.newSingleThreadExecutor().execute(() -> {
List<Socket> sockets = new ArrayList<Socket>();
this.executor.execute(() -> {
List<Socket> sockets = new ArrayList<>();
try {
latch.countDown();
while (!done.get()) {
@@ -793,8 +796,8 @@ public class TcpOutboundGatewayTests {
final CountDownLatch latch = new CountDownLatch(1);
final AtomicBoolean done = new AtomicBoolean();
Executors.newSingleThreadExecutor().execute(() -> {
List<Socket> sockets = new ArrayList<Socket>();
this.executor.execute(() -> {
List<Socket> sockets = new ArrayList<>();
try {
latch.countDown();
while (!done.get()) {

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

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2015 the original author or authors.
* Copyright 2002-2018 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,7 +33,6 @@ import java.util.List;
import java.util.Set;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.Executor;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicReference;
@@ -46,6 +45,7 @@ import org.junit.Test;
import org.springframework.beans.factory.BeanFactory;
import org.springframework.core.serializer.DefaultDeserializer;
import org.springframework.core.serializer.DefaultSerializer;
import org.springframework.core.task.SimpleAsyncTaskExecutor;
import org.springframework.integration.channel.DirectChannel;
import org.springframework.integration.channel.QueueChannel;
import org.springframework.integration.handler.ServiceActivatingHandler;
@@ -64,6 +64,7 @@ import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler;
/**
* @author Gary Russell
* @author Artem Bilan
*/
public class TcpReceivingChannelAdapterTests extends AbstractTcpChannelAdapterTests {
@@ -97,27 +98,24 @@ public class TcpReceivingChannelAdapterTests extends AbstractTcpChannelAdapterTe
@Test
public void testNetClientMode() throws Exception {
final AtomicReference<ServerSocket> serverSocket = new AtomicReference<ServerSocket>();
final AtomicReference<ServerSocket> serverSocket = new AtomicReference<>();
final CountDownLatch latch1 = new CountDownLatch(1);
final CountDownLatch latch2 = new CountDownLatch(1);
final AtomicBoolean done = new AtomicBoolean();
Executors.newSingleThreadExecutor().execute(new Runnable() {
@Override
public void run() {
try {
ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(0, 10);
serverSocket.set(server);
latch1.countDown();
Socket socket = server.accept();
socket.getOutputStream().write("Test1\r\nTest2\r\n".getBytes());
latch2.await();
socket.close();
server.close();
}
catch (Exception e) {
if (!done.get()) {
e.printStackTrace();
}
new SimpleAsyncTaskExecutor().execute(() -> {
try {
ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(0, 10);
serverSocket.set(server);
latch1.countDown();
Socket socket = server.accept();
socket.getOutputStream().write("Test1\r\nTest2\r\n".getBytes());
latch2.await();
socket.close();
server.close();
}
catch (Exception e) {
if (!done.get()) {
e.printStackTrace();
}
}
});
@@ -416,7 +414,7 @@ public class TcpReceivingChannelAdapterTests extends AbstractTcpChannelAdapterTe
handler.setConnectionFactory(scf);
TcpReceivingChannelAdapter adapter = new TcpReceivingChannelAdapter();
adapter.setConnectionFactory(scf);
Executor te = Executors.newCachedThreadPool();
Executor te = new SimpleAsyncTaskExecutor();
scf.setTaskExecutor(te);
scf.start();
QueueChannel channel = new QueueChannel();
@@ -650,6 +648,7 @@ public class TcpReceivingChannelAdapterTests extends AbstractTcpChannelAdapterTe
}
private class FailingService {
@SuppressWarnings("unused")
public String serviceMethod(byte[] bytes) {
throw new RuntimeException("Failed");

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

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2016 the original author or authors.
* Copyright 2002-2018 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.
@@ -35,8 +35,6 @@ import java.util.List;
import java.util.Set;
import java.util.TreeSet;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Semaphore;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
@@ -54,6 +52,8 @@ import org.springframework.context.support.AbstractApplicationContext;
import org.springframework.context.support.ClassPathXmlApplicationContext;
import org.springframework.core.serializer.DefaultDeserializer;
import org.springframework.core.serializer.DefaultSerializer;
import org.springframework.core.task.AsyncTaskExecutor;
import org.springframework.core.task.SimpleAsyncTaskExecutor;
import org.springframework.integration.channel.QueueChannel;
import org.springframework.integration.config.ConsumerEndpointFactoryBean;
import org.springframework.integration.ip.tcp.connection.AbstractClientConnectionFactory;
@@ -79,12 +79,14 @@ import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler;
/**
* @author Gary Russell
* @author Artem Bilan
*
* @since 2.0
*/
public class TcpSendingMessageHandlerTests extends AbstractTcpChannelAdapterTests {
private static final Log logger = LogFactory.getLog(TcpSendingMessageHandlerTests.class);
private AsyncTaskExecutor executor = new SimpleAsyncTaskExecutor();
private void readFully(InputStream is, byte[] buff) throws IOException {
for (int i = 0; i < buff.length; i++) {
@@ -97,7 +99,7 @@ public class TcpSendingMessageHandlerTests extends AbstractTcpChannelAdapterTest
final AtomicReference<ServerSocket> serverSocket = new AtomicReference<ServerSocket>();
final CountDownLatch latch = new CountDownLatch(1);
final AtomicBoolean done = new AtomicBoolean();
Executors.newSingleThreadExecutor().execute(() -> {
this.executor.execute(() -> {
try {
ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(0);
serverSocket.set(server);
@@ -150,7 +152,7 @@ public class TcpSendingMessageHandlerTests extends AbstractTcpChannelAdapterTest
final AtomicReference<ServerSocket> serverSocket = new AtomicReference<ServerSocket>();
final CountDownLatch latch = new CountDownLatch(1);
final AtomicBoolean done = new AtomicBoolean();
Executors.newSingleThreadExecutor().execute(() -> {
this.executor.execute(() -> {
try {
ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(0);
serverSocket.set(server);
@@ -215,7 +217,7 @@ public class TcpSendingMessageHandlerTests extends AbstractTcpChannelAdapterTest
final AtomicReference<ServerSocket> serverSocket = new AtomicReference<ServerSocket>();
final CountDownLatch latch = new CountDownLatch(1);
final AtomicBoolean done = new AtomicBoolean();
Executors.newSingleThreadExecutor().execute(() -> {
this.executor.execute(() -> {
try {
ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(0);
serverSocket.set(server);
@@ -271,7 +273,7 @@ public class TcpSendingMessageHandlerTests extends AbstractTcpChannelAdapterTest
final AtomicReference<ServerSocket> serverSocket = new AtomicReference<ServerSocket>();
final CountDownLatch latch = new CountDownLatch(1);
final AtomicBoolean done = new AtomicBoolean();
Executors.newSingleThreadExecutor().execute(() -> {
this.executor.execute(() -> {
try {
ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(0);
serverSocket.set(server);
@@ -324,7 +326,7 @@ public class TcpSendingMessageHandlerTests extends AbstractTcpChannelAdapterTest
final AtomicReference<ServerSocket> serverSocket = new AtomicReference<ServerSocket>();
final CountDownLatch latch = new CountDownLatch(1);
final AtomicBoolean done = new AtomicBoolean();
Executors.newSingleThreadExecutor().execute(() -> {
this.executor.execute(() -> {
try {
ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(0);
serverSocket.set(server);
@@ -377,10 +379,10 @@ public class TcpSendingMessageHandlerTests extends AbstractTcpChannelAdapterTest
@Test
public void testNetLength() throws Exception {
final AtomicReference<ServerSocket> serverSocket = new AtomicReference<ServerSocket>();
final AtomicReference<ServerSocket> serverSocket = new AtomicReference<>();
final CountDownLatch latch = new CountDownLatch(1);
final AtomicBoolean done = new AtomicBoolean();
Executors.newSingleThreadExecutor().execute(() -> {
this.executor.execute(() -> {
try {
ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(0);
serverSocket.set(server);
@@ -436,7 +438,7 @@ public class TcpSendingMessageHandlerTests extends AbstractTcpChannelAdapterTest
final AtomicReference<ServerSocket> serverSocket = new AtomicReference<ServerSocket>();
final CountDownLatch latch = new CountDownLatch(1);
final AtomicBoolean done = new AtomicBoolean();
Executors.newSingleThreadExecutor().execute(() -> {
this.executor.execute(() -> {
try {
ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(0);
serverSocket.set(server);
@@ -495,7 +497,7 @@ public class TcpSendingMessageHandlerTests extends AbstractTcpChannelAdapterTest
final AtomicReference<ServerSocket> serverSocket = new AtomicReference<ServerSocket>();
final CountDownLatch latch = new CountDownLatch(1);
final AtomicBoolean done = new AtomicBoolean();
Executors.newSingleThreadExecutor().execute(() -> {
this.executor.execute(() -> {
try {
ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(0);
serverSocket.set(server);
@@ -544,10 +546,10 @@ public class TcpSendingMessageHandlerTests extends AbstractTcpChannelAdapterTest
@Test
public void testNioSerial() throws Exception {
final AtomicReference<ServerSocket> serverSocket = new AtomicReference<ServerSocket>();
final AtomicReference<ServerSocket> serverSocket = new AtomicReference<>();
final CountDownLatch latch = new CountDownLatch(1);
final AtomicBoolean done = new AtomicBoolean();
Executors.newSingleThreadExecutor().execute(() -> {
this.executor.execute(() -> {
try {
ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(0);
serverSocket.set(server);
@@ -598,12 +600,12 @@ public class TcpSendingMessageHandlerTests extends AbstractTcpChannelAdapterTest
}
@Test
public void testNetSingleUseNoInbound() throws Exception {
public void testNetSingleUseNoInbound() throws Exception {
final AtomicReference<ServerSocket> serverSocket = new AtomicReference<ServerSocket>();
final CountDownLatch latch = new CountDownLatch(1);
final Semaphore semaphore = new Semaphore(0);
final AtomicBoolean done = new AtomicBoolean();
Executors.newSingleThreadExecutor().execute(() -> {
this.executor.execute(() -> {
try {
ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(0);
serverSocket.set(server);
@@ -645,12 +647,12 @@ public class TcpSendingMessageHandlerTests extends AbstractTcpChannelAdapterTest
}
@Test
public void testNioSingleUseNoInbound() throws Exception {
public void testNioSingleUseNoInbound() throws Exception {
final AtomicReference<ServerSocket> serverSocket = new AtomicReference<ServerSocket>();
final CountDownLatch latch = new CountDownLatch(1);
final Semaphore semaphore = new Semaphore(0);
final AtomicBoolean done = new AtomicBoolean();
Executors.newSingleThreadExecutor().execute(() -> {
this.executor.execute(() -> {
try {
ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(0);
serverSocket.set(server);
@@ -692,12 +694,12 @@ public class TcpSendingMessageHandlerTests extends AbstractTcpChannelAdapterTest
}
@Test
public void testNetSingleUseWithInbound() throws Exception {
public void testNetSingleUseWithInbound() throws Exception {
final AtomicReference<ServerSocket> serverSocket = new AtomicReference<ServerSocket>();
final CountDownLatch latch = new CountDownLatch(1);
final Semaphore semaphore = new Semaphore(0);
final AtomicBoolean done = new AtomicBoolean();
Executors.newSingleThreadExecutor().execute(() -> {
this.executor.execute(() -> {
try {
ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(0);
serverSocket.set(server);
@@ -752,12 +754,12 @@ public class TcpSendingMessageHandlerTests extends AbstractTcpChannelAdapterTest
}
@Test
public void testNioSingleUseWithInbound() throws Exception {
public void testNioSingleUseWithInbound() throws Exception {
final AtomicReference<ServerSocket> serverSocket = new AtomicReference<ServerSocket>();
final CountDownLatch latch = new CountDownLatch(1);
final Semaphore semaphore = new Semaphore(0);
final AtomicBoolean done = new AtomicBoolean();
Executors.newSingleThreadExecutor().execute(() -> {
this.executor.execute(() -> {
try {
ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(0);
serverSocket.set(server);
@@ -812,14 +814,13 @@ public class TcpSendingMessageHandlerTests extends AbstractTcpChannelAdapterTest
}
@Test
public void testNioSingleUseWithInboundMany() throws Exception {
public void testNioSingleUseWithInboundMany() throws Exception {
final AtomicReference<ServerSocket> serverSocket = new AtomicReference<ServerSocket>();
final CountDownLatch latch = new CountDownLatch(1);
final Semaphore semaphore = new Semaphore(0);
final AtomicBoolean done = new AtomicBoolean();
final List<Socket> serverSockets = new ArrayList<Socket>();
final ExecutorService exec = Executors.newCachedThreadPool();
exec.execute(() -> {
this.executor.execute(() -> {
try {
ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(0, 100);
serverSocket.set(server);
@@ -828,7 +829,7 @@ public class TcpSendingMessageHandlerTests extends AbstractTcpChannelAdapterTest
final Socket socket = server.accept();
serverSockets.add(socket);
final int j = i;
exec.execute(() -> {
this.executor.execute(() -> {
semaphore.release();
byte[] b = new byte[9];
try {
@@ -843,7 +844,8 @@ public class TcpSendingMessageHandlerTests extends AbstractTcpChannelAdapterTest
try {
socket.close();
}
catch (IOException e2) { }
catch (IOException e2) {
}
}
});
}
@@ -864,7 +866,7 @@ public class TcpSendingMessageHandlerTests extends AbstractTcpChannelAdapterTest
ccf.setDeserializer(serializer);
ccf.setSoTimeout(10000);
ccf.setSingleUse(true);
ccf.setTaskExecutor(Executors.newCachedThreadPool());
ccf.setTaskExecutor(this.executor);
ccf.start();
TcpSendingMessageHandler handler = new TcpSendingMessageHandler();
handler.setConnectionFactory(ccf);
@@ -902,7 +904,7 @@ public class TcpSendingMessageHandlerTests extends AbstractTcpChannelAdapterTest
final AtomicReference<ServerSocket> serverSocket = new AtomicReference<ServerSocket>();
final CountDownLatch latch = new CountDownLatch(1);
final AtomicBoolean done = new AtomicBoolean();
Executors.newSingleThreadExecutor().execute(() -> {
this.executor.execute(() -> {
try {
ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(0);
serverSocket.set(server);
@@ -973,7 +975,7 @@ public class TcpSendingMessageHandlerTests extends AbstractTcpChannelAdapterTest
final AtomicReference<ServerSocket> serverSocket = new AtomicReference<ServerSocket>();
final CountDownLatch latch = new CountDownLatch(1);
final AtomicBoolean done = new AtomicBoolean();
Executors.newSingleThreadExecutor().execute(() -> {
this.executor.execute(() -> {
try {
ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(0);
serverSocket.set(server);
@@ -1010,7 +1012,7 @@ public class TcpSendingMessageHandlerTests extends AbstractTcpChannelAdapterTest
ccf.setDeserializer(new DefaultDeserializer());
ccf.setSoTimeout(10000);
TcpConnectionInterceptorFactoryChain fc = new TcpConnectionInterceptorFactoryChain();
fc.setInterceptors(new TcpConnectionInterceptorFactory[] {newInterceptorFactory()});
fc.setInterceptors(new TcpConnectionInterceptorFactory[] { newInterceptorFactory() });
ccf.setInterceptorFactoryChain(fc);
ccf.start();
TcpSendingMessageHandler handler = new TcpSendingMessageHandler();
@@ -1042,7 +1044,7 @@ public class TcpSendingMessageHandlerTests extends AbstractTcpChannelAdapterTest
final AtomicReference<ServerSocket> serverSocket = new AtomicReference<ServerSocket>();
final CountDownLatch latch = new CountDownLatch(1);
final AtomicBoolean done = new AtomicBoolean();
Executors.newSingleThreadExecutor().execute(() -> {
this.executor.execute(() -> {
try {
ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(0);
serverSocket.set(server);
@@ -1099,7 +1101,7 @@ public class TcpSendingMessageHandlerTests extends AbstractTcpChannelAdapterTest
final AtomicReference<ServerSocket> serverSocket = new AtomicReference<ServerSocket>();
final CountDownLatch latch = new CountDownLatch(1);
final AtomicBoolean done = new AtomicBoolean();
Executors.newSingleThreadExecutor().execute(() -> {
this.executor.execute(() -> {
int i = 0;
try {
ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(0);

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

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2017 the original author or authors.
* Copyright 2002-2018 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.
@@ -68,6 +68,7 @@ import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.context.ApplicationEvent;
import org.springframework.context.ApplicationEventPublisher;
import org.springframework.core.task.SimpleAsyncTaskExecutor;
import org.springframework.integration.channel.QueueChannel;
import org.springframework.integration.ip.IpHeaders;
import org.springframework.integration.ip.tcp.TcpOutboundGateway;
@@ -782,7 +783,7 @@ public class CachingClientConnectionFactoryTests {
invocation.callRealMethod();
String log = invocation.getArgument(0);
if (log.startsWith("Response")) {
Executors.newSingleThreadScheduledExecutor()
new SimpleAsyncTaskExecutor()
.execute(() -> gate.handleMessage(new GenericMessage<>("bar")));
// hold up the first thread until the second has added its pending reply
latch.await(10, TimeUnit.SECONDS);

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

@@ -39,7 +39,6 @@ import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import org.apache.commons.logging.Log;
@@ -52,6 +51,7 @@ import org.springframework.beans.DirectFieldAccessor;
import org.springframework.beans.factory.BeanFactory;
import org.springframework.context.ApplicationEvent;
import org.springframework.context.ApplicationEventPublisher;
import org.springframework.core.task.SimpleAsyncTaskExecutor;
import org.springframework.integration.channel.NullChannel;
import org.springframework.integration.context.IntegrationContextUtils;
import org.springframework.integration.ip.config.TcpConnectionFactoryFactoryBean;
@@ -233,7 +233,8 @@ public class ConnectionFactoryTests {
factory.start();
assertTrue("missing info log", latch1.await(10, TimeUnit.SECONDS));
// stop on a different thread because it waits for the executor
Executors.newSingleThreadExecutor().execute(() -> factory.stop());
new SimpleAsyncTaskExecutor()
.execute(factory::stop);
int n = 0;
DirectFieldAccessor accessor = new DirectFieldAccessor(factory);
while (n++ < 200 && accessor.getPropertyValue(property) != null) {

7
spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/FailoverClientConnectionFactoryTests.java
View File

@@ -35,7 +35,6 @@ import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.Executor;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;
@@ -48,6 +47,7 @@ import org.mockito.Mockito;
import org.springframework.beans.factory.BeanFactory;
import org.springframework.context.ApplicationEvent;
import org.springframework.context.ApplicationEventPublisher;
import org.springframework.core.task.SimpleAsyncTaskExecutor;
import org.springframework.integration.channel.DirectChannel;
import org.springframework.integration.channel.QueueChannel;
import org.springframework.integration.handler.BridgeHandler;
@@ -548,8 +548,9 @@ public class FailoverClientConnectionFactoryTests {
}
private Holder setupAndStartServers(AbstractServerConnectionFactory server1,
AbstractServerConnectionFactory server2) throws Exception {
Executor exec = Executors.newCachedThreadPool();
AbstractServerConnectionFactory server2) {
Executor exec = new SimpleAsyncTaskExecutor();
server1.setTaskExecutor(exec);
server2.setTaskExecutor(exec);
server1.setBeanName("server1");

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

@@ -72,8 +72,11 @@ import org.mockito.invocation.InvocationOnMock;
import org.mockito.stubbing.Answer;
import org.springframework.beans.DirectFieldAccessor;
import org.springframework.beans.factory.DisposableBean;
import org.springframework.context.ApplicationEvent;
import org.springframework.context.ApplicationEventPublisher;
import org.springframework.core.task.AsyncTaskExecutor;
import org.springframework.core.task.SimpleAsyncTaskExecutor;
import org.springframework.integration.ip.tcp.connection.TcpNioConnection.ChannelInputStream;
import org.springframework.integration.ip.tcp.serializer.ByteArrayCrLfSerializer;
import org.springframework.integration.ip.tcp.serializer.MapJsonSerializer;
@@ -111,12 +114,14 @@ public class TcpNioConnectionTests {
private final ApplicationEventPublisher nullPublisher = mock(ApplicationEventPublisher.class);
private final AsyncTaskExecutor executor = new SimpleAsyncTaskExecutor();
@Test
public void testWriteTimeout() throws Exception {
final CountDownLatch latch = new CountDownLatch(1);
final CountDownLatch done = new CountDownLatch(1);
final AtomicReference<ServerSocket> serverSocket = new AtomicReference<ServerSocket>();
Executors.newSingleThreadExecutor().execute(() -> {
final AtomicReference<ServerSocket> serverSocket = new AtomicReference<>();
this.executor.execute(() -> {
try {
ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(0);
logger.debug(testName.getMethodName() + " starting server for " + server.getLocalPort());
@@ -134,7 +139,7 @@ public class TcpNioConnectionTests {
TcpNioClientConnectionFactory factory = new TcpNioClientConnectionFactory("localhost",
serverSocket.get().getLocalPort());
factory.setApplicationEventPublisher(nullPublisher);
factory.setSoTimeout(1000);
factory.setSoTimeout(100);
factory.start();
try {
TcpConnection connection = factory.getConnection();
@@ -152,8 +157,8 @@ public class TcpNioConnectionTests {
public void testReadTimeout() throws Exception {
final CountDownLatch latch = new CountDownLatch(1);
final CountDownLatch done = new CountDownLatch(1);
final AtomicReference<ServerSocket> serverSocket = new AtomicReference<ServerSocket>();
Executors.newSingleThreadExecutor().execute(() -> {
final AtomicReference<ServerSocket> serverSocket = new AtomicReference<>();
this.executor.execute(() -> {
try {
ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(0);
logger.debug(testName.getMethodName() + " starting server for " + server.getLocalPort());
@@ -173,14 +178,14 @@ public class TcpNioConnectionTests {
TcpNioClientConnectionFactory factory = new TcpNioClientConnectionFactory("localhost",
serverSocket.get().getLocalPort());
factory.setApplicationEventPublisher(nullPublisher);
factory.setSoTimeout(1000);
factory.setSoTimeout(100);
factory.start();
try {
TcpConnection connection = factory.getConnection();
connection.send(MessageBuilder.withPayload("Test").build());
int n = 0;
while (connection.isOpen()) {
Thread.sleep(100);
Thread.sleep(10);
if (n++ > 200) {
break;
}
@@ -197,8 +202,8 @@ public class TcpNioConnectionTests {
@Test
public void testMemoryLeak() throws Exception {
final CountDownLatch latch = new CountDownLatch(1);
final AtomicReference<ServerSocket> serverSocket = new AtomicReference<ServerSocket>();
Executors.newSingleThreadExecutor().execute(() -> {
final AtomicReference<ServerSocket> serverSocket = new AtomicReference<>();
this.executor.execute(() -> {
try {
ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(0);
logger.debug(testName.getMethodName() + " starting server for " + server.getLocalPort());
@@ -247,7 +252,7 @@ public class TcpNioConnectionTests {
TcpNioClientConnectionFactory factory = new TcpNioClientConnectionFactory("localhost", 0);
factory.setApplicationEventPublisher(nullPublisher);
factory.setNioHarvestInterval(100);
Map<SocketChannel, TcpNioConnection> connections = new HashMap<SocketChannel, TcpNioConnection>();
Map<SocketChannel, TcpNioConnection> connections = new HashMap<>();
SocketChannel chan1 = mock(SocketChannel.class);
SocketChannel chan2 = mock(SocketChannel.class);
SocketChannel chan3 = mock(SocketChannel.class);
@@ -334,6 +339,9 @@ public class TcpNioConnectionTests {
catch (ExecutionException e) {
assertEquals("Timed out waiting for buffer space", e.getCause().getMessage());
}
finally {
exec.shutdownNow();
}
}
@Test
@@ -375,6 +383,8 @@ public class TcpNioConnectionTests {
});
future.get(60, TimeUnit.SECONDS);
assertTrue(messageLatch.await(10, TimeUnit.SECONDS));
exec.shutdownNow();
}
@Test
@@ -463,19 +473,14 @@ public class TcpNioConnectionTests {
.getPropertyValue("channelInputStream");
final CountDownLatch latch = new CountDownLatch(1);
final byte[] out = new byte[4];
ExecutorService exec = Executors.newSingleThreadExecutor();
exec.execute(new Runnable() {
@Override
public void run() {
try {
stream.read(out);
}
catch (IOException e) {
e.printStackTrace();
}
latch.countDown();
this.executor.execute(() -> {
try {
stream.read(out);
}
catch (IOException e) {
e.printStackTrace();
}
latch.countDown();
});
Thread.sleep(1000);
assertEquals(0x00, out[0]);
@@ -599,11 +604,18 @@ public class TcpNioConnectionTests {
assertThat(threadName.get(), containsString("assembler"));
factory.stop();
cleanupCompositeExecutor(compositeExec);
}
private void cleanupCompositeExecutor(CompositeExecutor compositeExec) throws Exception {
TestUtils.getPropertyValue(compositeExec, "primaryTaskExecutor", DisposableBean.class).destroy();
TestUtils.getPropertyValue(compositeExec, "secondaryTaskExecutor", DisposableBean.class).destroy();
}
@Test
public void testAllMessagesDelivered() throws Exception {
final int numberOfSockets = 100;
final int numberOfSockets = 10;
TcpNioServerConnectionFactory factory = new TcpNioServerConnectionFactory(0);
factory.setApplicationEventPublisher(nullPublisher);
@@ -611,16 +623,11 @@ public class TcpNioConnectionTests {
factory.setTaskExecutor(compositeExec);
final CountDownLatch latch = new CountDownLatch(numberOfSockets * 4);
factory.registerListener(new TcpListener() {
@Override
public boolean onMessage(Message<?> message) {
if (!(message instanceof ErrorMessage)) {
latch.countDown();
}
return false;
factory.registerListener(message -> {
if (!(message instanceof ErrorMessage)) {
latch.countDown();
}
return false;
});
factory.start();
TestingUtilities.waitListening(factory, null);
@@ -637,7 +644,7 @@ public class TcpNioConnectionTests {
}
catch (ConnectException e) {
}
Thread.sleep(100);
Thread.sleep(1);
}
assertTrue("Could not open socket to localhost:" + port, n < 100);
sockets[i] = socket;
@@ -646,7 +653,7 @@ public class TcpNioConnectionTests {
sockets[i].getOutputStream().write("foo1 and...".getBytes());
sockets[i].getOutputStream().flush();
}
Thread.sleep(100);
Thread.sleep(1);
for (int i = 0; i < numberOfSockets; i++) {
sockets[i].getOutputStream().write(("...foo2\r\nbar1 and...").getBytes());
sockets[i].getOutputStream().flush();
@@ -659,7 +666,7 @@ public class TcpNioConnectionTests {
sockets[i].getOutputStream().write("foo3 and...".getBytes());
sockets[i].getOutputStream().flush();
}
Thread.sleep(100);
Thread.sleep(1);
for (int i = 0; i < numberOfSockets; i++) {
sockets[i].getOutputStream().write(("...foo4\r\nbar3 and...").getBytes());
sockets[i].getOutputStream().flush();
@@ -672,6 +679,8 @@ public class TcpNioConnectionTests {
assertTrue("latch is still " + latch.getCount(), latch.await(60, TimeUnit.SECONDS));
factory.stop();
cleanupCompositeExecutor(compositeExec);
}
private CompositeExecutor compositeExecutor() {
@@ -791,6 +800,8 @@ public class TcpNioConnectionTests {
assertThat(Arrays.asList(stackTrace).toString(), not(containsString("ChannelInputStream.getNextBuffer")));
socket.close();
factory.stop();
te.shutdown();
}
private void readFully(InputStream is, byte[] buff) throws IOException {

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

@@ -33,8 +33,7 @@ import java.net.ServerSocket;
import java.net.Socket;
import java.nio.ByteBuffer;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Executor;
import java.util.concurrent.atomic.AtomicReference;
import javax.net.ServerSocketFactory;
@@ -46,6 +45,7 @@ import org.springframework.beans.factory.BeanFactory;
import org.springframework.context.ApplicationEvent;
import org.springframework.context.ApplicationEventPublisher;
import org.springframework.core.serializer.DefaultDeserializer;
import org.springframework.core.task.SimpleAsyncTaskExecutor;
import org.springframework.integration.channel.QueueChannel;
import org.springframework.integration.ip.tcp.TcpInboundGateway;
import org.springframework.integration.ip.tcp.TcpOutboundGateway;
@@ -62,6 +62,7 @@ import org.springframework.messaging.support.GenericMessage;
/**
* @author Gary Russell
* @author Gavin Gray
*
* @since 2.0
*/
public class DeserializationTests {
@@ -419,7 +420,7 @@ public class DeserializationTests {
// eat SocketTimeoutException. Doesn't matter for this test
}
};
ExecutorService exec = Executors.newSingleThreadExecutor();
Executor exec = new SimpleAsyncTaskExecutor();
Message<?> message;

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

@@ -27,7 +27,6 @@ import java.net.InetSocketAddress;
import java.net.MulticastSocket;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.Executor;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
@@ -35,6 +34,7 @@ import org.junit.Rule;
import org.junit.Test;
import org.springframework.beans.factory.BeanFactory;
import org.springframework.core.task.SimpleAsyncTaskExecutor;
import org.springframework.integration.ip.IpHeaders;
import org.springframework.integration.support.MessageBuilder;
import org.springframework.messaging.Message;
@@ -89,7 +89,7 @@ public class DatagramPacketMulticastSendingHandlerTests {
e.printStackTrace();
}
};
Executor executor = Executors.newFixedThreadPool(2);
Executor executor = new SimpleAsyncTaskExecutor();
executor.execute(catcher);
executor.execute(catcher);
assertTrue(listening.await(10000, TimeUnit.MILLISECONDS));
@@ -159,7 +159,7 @@ public class DatagramPacketMulticastSendingHandlerTests {
e.printStackTrace();
}
};
Executor executor = Executors.newFixedThreadPool(2);
Executor executor = new SimpleAsyncTaskExecutor();
executor.execute(catcher);
executor.execute(catcher);
assertTrue(listening.await(10000, TimeUnit.MILLISECONDS));

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

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2016 the original author or authors.
* Copyright 2002-2018 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,13 +24,13 @@ import java.net.DatagramPacket;
import java.net.DatagramSocket;
import java.net.InetSocketAddress;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import org.junit.Test;
import org.springframework.beans.factory.BeanFactory;
import org.springframework.core.task.SimpleAsyncTaskExecutor;
import org.springframework.integration.ip.IpHeaders;
import org.springframework.integration.support.MessageBuilder;
import org.springframework.messaging.Message;
@@ -39,6 +39,8 @@ import org.springframework.messaging.Message;
* @author Mark Fisher
* @author Gary Russell
* @author Marcin Pilaczynski
* @author Artem Bilan
*
* @since 2.0
*/
public class DatagramPacketSendingHandlerTests {
@@ -50,19 +52,20 @@ public class DatagramPacketSendingHandlerTests {
final CountDownLatch received = new CountDownLatch(1);
final AtomicInteger testPort = new AtomicInteger();
final CountDownLatch listening = new CountDownLatch(1);
Executors.newSingleThreadExecutor().execute(() -> {
try {
DatagramSocket socket = new DatagramSocket();
testPort.set(socket.getLocalPort());
listening.countDown();
socket.receive(receivedPacket);
received.countDown();
socket.close();
}
catch (Exception e) {
e.printStackTrace();
}
});
new SimpleAsyncTaskExecutor()
.execute(() -> {
try {
DatagramSocket socket = new DatagramSocket();
testPort.set(socket.getLocalPort());
listening.countDown();
socket.receive(receivedPacket);
received.countDown();
socket.close();
}
catch (Exception e) {
e.printStackTrace();
}
});
assertTrue(listening.await(10, TimeUnit.SECONDS));
UnicastSendingMessageHandler handler =
new UnicastSendingMessageHandler("localhost", testPort.get());
@@ -89,31 +92,32 @@ public class DatagramPacketSendingHandlerTests {
final CountDownLatch listening = new CountDownLatch(1);
final CountDownLatch ackListening = new CountDownLatch(1);
final CountDownLatch ackSent = new CountDownLatch(1);
Executors.newSingleThreadExecutor().execute(() -> {
try {
DatagramSocket socket = new DatagramSocket();
testPort.set(socket.getLocalPort());
listening.countDown();
assertTrue(ackListening.await(10, TimeUnit.SECONDS));
socket.receive(receivedPacket);
socket.close();
DatagramPacketMessageMapper mapper = new DatagramPacketMessageMapper();
mapper.setAcknowledge(true);
mapper.setLengthCheck(true);
Message<byte[]> message = mapper.toMessage(receivedPacket);
Object id = message.getHeaders().get(IpHeaders.ACK_ID);
byte[] ack = id.toString().getBytes();
DatagramPacket ackPack = new DatagramPacket(ack, ack.length,
new InetSocketAddress("localHost", ackPort.get()));
DatagramSocket out = new DatagramSocket();
out.send(ackPack);
out.close();
ackSent.countDown();
}
catch (Exception e) {
e.printStackTrace();
}
});
new SimpleAsyncTaskExecutor()
.execute(() -> {
try {
DatagramSocket socket = new DatagramSocket();
testPort.set(socket.getLocalPort());
listening.countDown();
assertTrue(ackListening.await(10, TimeUnit.SECONDS));
socket.receive(receivedPacket);
socket.close();
DatagramPacketMessageMapper mapper = new DatagramPacketMessageMapper();
mapper.setAcknowledge(true);
mapper.setLengthCheck(true);
Message<byte[]> message = mapper.toMessage(receivedPacket);
Object id = message.getHeaders().get(IpHeaders.ACK_ID);
byte[] ack = id.toString().getBytes();
DatagramPacket ackPack = new DatagramPacket(ack, ack.length,
new InetSocketAddress("localHost", ackPort.get()));
DatagramSocket out = new DatagramSocket();
out.send(ackPack);
out.close();
ackSent.countDown();
}
catch (Exception e) {
e.printStackTrace();
}
});
listening.await(10000, TimeUnit.MILLISECONDS);
UnicastSendingMessageHandler handler =
new UnicastSendingMessageHandler("localhost", testPort.get(), true, true, "localhost", 0, 5000);

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

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2016 the original author or authors.
* Copyright 2002-2018 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.
@@ -31,7 +31,6 @@ import java.net.InetSocketAddress;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.Executor;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicReference;
@@ -42,6 +41,7 @@ import org.junit.Test;
import org.springframework.beans.factory.BeanFactory;
import org.springframework.context.ConfigurableApplicationContext;
import org.springframework.context.support.ClassPathXmlApplicationContext;
import org.springframework.core.task.SimpleAsyncTaskExecutor;
import org.springframework.integration.channel.DirectChannel;
import org.springframework.integration.channel.QueueChannel;
import org.springframework.integration.handler.ServiceActivatingHandler;
@@ -56,6 +56,7 @@ import org.springframework.messaging.SubscribableChannel;
* @author Gary Russell
* @author Artem Bilan
* @author Marcin Pilaczynski
*
* @since 2.0
*
*/
@@ -185,18 +186,19 @@ public class UdpChannelAdapterTests {
final CountDownLatch receiverReadyLatch = new CountDownLatch(1);
final CountDownLatch replyReceivedLatch = new CountDownLatch(1);
//main thread sends the reply using the headers, this thread will receive it
Executors.newSingleThreadExecutor().execute(() -> {
DatagramPacket answer = new DatagramPacket(new byte[2000], 2000);
try {
receiverReadyLatch.countDown();
socket.receive(answer);
theAnswer.set(answer);
replyReceivedLatch.countDown();
}
catch (IOException e) {
e.printStackTrace();
}
});
new SimpleAsyncTaskExecutor()
.execute(() -> {
DatagramPacket answer = new DatagramPacket(new byte[2000], 2000);
try {
receiverReadyLatch.countDown();
socket.receive(answer);
theAnswer.set(answer);
replyReceivedLatch.countDown();
}
catch (IOException e) {
e.printStackTrace();
}
});
Message<byte[]> receivedMessage = (Message<byte[]>) channel.receive(10000);
assertEquals(new String(message.getPayload()), new String(receivedMessage.getPayload()));
String replyString = "reply:" + System.currentTimeMillis();