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

Revert "INT-4366: Fix MulticastSendingMH

Browse Source 
This reverts commit c3b64dc1ac.
This commit is contained in:
Gary Russell
2018-01-19 17:19:27 -05:00
parent c3b64dc1ac
commit 2d6984b3cb
13 changed files with 236 additions and 262 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-2018 the original author or authors.
* Copyright 2001-2016 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,8 +38,6 @@ import org.springframework.messaging.MessageHandler;
* determine success.
*
* @author Gary Russell
* @author Artem Bilan
*
* @since 2.0
*/
public class MulticastSendingMessageHandler extends UnicastSendingMessageHandler {
@@ -128,45 +126,49 @@ public class MulticastSendingMessageHandler extends UnicastSendingMessageHandler
@Override
protected DatagramSocket getSocket() throws IOException {
if (this.multicastSocket == null) {
if (this.getTheSocket() == null) {
synchronized (this) {
if (this.multicastSocket == null) {
createSocket();
}
createSocket();
}
}
return getTheSocket();
return this.getTheSocket();
}
private void createSocket() throws IOException {
MulticastSocket socket;
if (isAcknowledge()) {
int ackPort = getAckPort();
if (this.localAddress == null) {
socket = ackPort == 0 ? new MulticastSocket() : new MulticastSocket(ackPort);
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();
}
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 = new MulticastSocket(new InetSocketAddress(whichNic, ackPort));
socket.setInterface(whichNic);
}
if (getSoReceiveBufferSize() > 0) {
socket.setReceiveBufferSize(getSoReceiveBufferSize());
}
if (logger.isDebugEnabled()) {
logger.debug("Listening for acks on port: " + socket.getLocalPort());
}
setSocket(socket);
updateAckAddress();
this.multicastSocket = socket;
}
else {
socket = new MulticastSocket();
setSocket(socket);
}
if (this.timeToLive >= 0) {
socket.setTimeToLive(this.timeToLive);
}
setSocketAttributes(socket);
this.multicastSocket = socket;
}
@@ -176,7 +178,7 @@ public class MulticastSendingMessageHandler extends UnicastSendingMessageHandler
* @param minAcksForSuccess The minimum number of acks that will represent success.
*/
public void setMinAcksForSuccess(int minAcksForSuccess) {
setAckCounter(minAcksForSuccess);
this.setAckCounter(minAcksForSuccess);
}
/**

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

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2018 the original author or authors.
* Copyright 2002-2016 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,6 +28,7 @@ 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;
@@ -38,7 +39,6 @@ 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,8 +56,6 @@ import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler;
/**
* @author Gary Russell
* @author Artem Bilan
*
* @since 2.0
*/
public class TcpInboundGatewayTests {
@@ -121,31 +119,30 @@ public class TcpInboundGatewayTests {
final CountDownLatch latch2 = new CountDownLatch(1);
final CountDownLatch latch3 = new CountDownLatch(1);
final AtomicBoolean done = new AtomicBoolean();
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();
}
}
});
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();
}
}
});
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,6 +43,7 @@ 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;
@@ -60,8 +61,6 @@ 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;
@@ -91,8 +90,6 @@ public class TcpOutboundGatewayTests {
private static final Log logger = LogFactory.getLog(TcpOutboundGatewayTests.class);
private AsyncTaskExecutor executor = new SimpleAsyncTaskExecutor();
@ClassRule
public static LongRunningIntegrationTest longTests = new LongRunningIntegrationTest();
@@ -104,13 +101,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<>();
this.executor.execute(() -> {
final AtomicReference<ServerSocket> serverSocket = new AtomicReference<ServerSocket>();
Executors.newSingleThreadExecutor().execute(() -> {
try {
ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(0, 100);
serverSocket.set(server);
latch.countDown();
List<Socket> sockets = new ArrayList<>();
List<Socket> sockets = new ArrayList<Socket>();
int i = 0;
while (true) {
Socket socket = server.accept();
@@ -168,8 +165,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<>();
this.executor.execute(() -> {
final AtomicReference<ServerSocket> serverSocket = new AtomicReference<ServerSocket>();
Executors.newSingleThreadExecutor().execute(() -> {
try {
ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(0, 10);
serverSocket.set(server);
@@ -223,8 +220,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<>();
this.executor.execute(() -> {
final AtomicReference<ServerSocket> serverSocket = new AtomicReference<ServerSocket>();
Executors.newSingleThreadExecutor().execute(() -> {
try {
ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(0);
serverSocket.set(server);
@@ -263,12 +260,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] = (this.executor.submit(() -> {
results[j] = (Executors.newSingleThreadExecutor().submit(() -> {
gateway.handleMessage(MessageBuilder.withPayload("Test" + j).build());
return 0;
}));
}
Set<String> replies = new HashSet<>();
Set<String> replies = new HashSet<String>();
int timeouts = 0;
for (int i = 0; i < 2; i++) {
try {
@@ -347,7 +344,7 @@ public class TcpOutboundGatewayTests {
final AtomicReference<String> lastReceived = new AtomicReference<String>();
final CountDownLatch serverLatch = new CountDownLatch(2);
this.executor.execute(() -> {
Executors.newSingleThreadExecutor().execute(() -> {
try {
latch.countDown();
int i = 0;
@@ -401,7 +398,7 @@ public class TcpOutboundGatewayTests {
for (int i = 0; i < 2; i++) {
final int j = i;
results[j] = (this.executor.submit(() -> {
results[j] = (Executors.newSingleThreadExecutor().submit(() -> {
gateway.handleMessage(MessageBuilder.withPayload("Test" + j).build());
return j;
}));
@@ -445,7 +442,7 @@ public class TcpOutboundGatewayTests {
final AtomicBoolean done = new AtomicBoolean();
final CountDownLatch serverLatch = new CountDownLatch(1);
this.executor.execute(() -> {
Executors.newSingleThreadExecutor().execute(() -> {
try {
ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(0);
serverSocket.set(server);
@@ -520,12 +517,12 @@ public class TcpOutboundGatewayTests {
@Test
public void testFailoverCached() throws Exception {
final AtomicReference<ServerSocket> serverSocket = new AtomicReference<>();
final AtomicReference<ServerSocket> serverSocket = new AtomicReference<ServerSocket>();
final CountDownLatch latch = new CountDownLatch(1);
final AtomicBoolean done = new AtomicBoolean();
final CountDownLatch serverLatch = new CountDownLatch(1);
this.executor.execute(() -> {
Executors.newSingleThreadExecutor().execute(() -> {
try {
ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(0);
serverSocket.set(server);
@@ -670,11 +667,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<>();
final AtomicReference<String> lastReceived = new AtomicReference<String>();
final CountDownLatch serverLatch = new CountDownLatch(1);
this.executor.execute(() -> {
List<Socket> sockets = new ArrayList<>();
Executors.newSingleThreadExecutor().execute(() -> {
List<Socket> sockets = new ArrayList<Socket>();
try {
latch.countDown();
while (!done.get()) {
@@ -796,8 +793,8 @@ public class TcpOutboundGatewayTests {
final CountDownLatch latch = new CountDownLatch(1);
final AtomicBoolean done = new AtomicBoolean();
this.executor.execute(() -> {
List<Socket> sockets = new ArrayList<>();
Executors.newSingleThreadExecutor().execute(() -> {
List<Socket> sockets = new ArrayList<Socket>();
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-2018 the original author or authors.
* Copyright 2002-2015 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.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;
@@ -45,7 +46,6 @@ 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,7 +64,6 @@ import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler;
/**
* @author Gary Russell
* @author Artem Bilan
*/
public class TcpReceivingChannelAdapterTests extends AbstractTcpChannelAdapterTests {
@@ -98,24 +97,27 @@ public class TcpReceivingChannelAdapterTests extends AbstractTcpChannelAdapterTe
@Test
public void testNetClientMode() throws Exception {
final AtomicReference<ServerSocket> serverSocket = new AtomicReference<>();
final AtomicReference<ServerSocket> serverSocket = new AtomicReference<ServerSocket>();
final CountDownLatch latch1 = new CountDownLatch(1);
final CountDownLatch latch2 = new CountDownLatch(1);
final AtomicBoolean done = new AtomicBoolean();
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();
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();
}
}
}
});
@@ -414,7 +416,7 @@ public class TcpReceivingChannelAdapterTests extends AbstractTcpChannelAdapterTe
handler.setConnectionFactory(scf);
TcpReceivingChannelAdapter adapter = new TcpReceivingChannelAdapter();
adapter.setConnectionFactory(scf);
Executor te = new SimpleAsyncTaskExecutor();
Executor te = Executors.newCachedThreadPool();
scf.setTaskExecutor(te);
scf.start();
QueueChannel channel = new QueueChannel();
@@ -648,7 +650,6 @@ 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-2018 the original author or authors.
* Copyright 2002-2016 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,6 +35,8 @@ 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;
@@ -52,8 +54,6 @@ 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,14 +79,12 @@ 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++) {
@@ -99,7 +97,7 @@ public class TcpSendingMessageHandlerTests extends AbstractTcpChannelAdapterTest
final AtomicReference<ServerSocket> serverSocket = new AtomicReference<ServerSocket>();
final CountDownLatch latch = new CountDownLatch(1);
final AtomicBoolean done = new AtomicBoolean();
this.executor.execute(() -> {
Executors.newSingleThreadExecutor().execute(() -> {
try {
ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(0);
serverSocket.set(server);
@@ -152,7 +150,7 @@ public class TcpSendingMessageHandlerTests extends AbstractTcpChannelAdapterTest
final AtomicReference<ServerSocket> serverSocket = new AtomicReference<ServerSocket>();
final CountDownLatch latch = new CountDownLatch(1);
final AtomicBoolean done = new AtomicBoolean();
this.executor.execute(() -> {
Executors.newSingleThreadExecutor().execute(() -> {
try {
ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(0);
serverSocket.set(server);
@@ -217,7 +215,7 @@ public class TcpSendingMessageHandlerTests extends AbstractTcpChannelAdapterTest
final AtomicReference<ServerSocket> serverSocket = new AtomicReference<ServerSocket>();
final CountDownLatch latch = new CountDownLatch(1);
final AtomicBoolean done = new AtomicBoolean();
this.executor.execute(() -> {
Executors.newSingleThreadExecutor().execute(() -> {
try {
ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(0);
serverSocket.set(server);
@@ -273,7 +271,7 @@ public class TcpSendingMessageHandlerTests extends AbstractTcpChannelAdapterTest
final AtomicReference<ServerSocket> serverSocket = new AtomicReference<ServerSocket>();
final CountDownLatch latch = new CountDownLatch(1);
final AtomicBoolean done = new AtomicBoolean();
this.executor.execute(() -> {
Executors.newSingleThreadExecutor().execute(() -> {
try {
ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(0);
serverSocket.set(server);
@@ -326,7 +324,7 @@ public class TcpSendingMessageHandlerTests extends AbstractTcpChannelAdapterTest
final AtomicReference<ServerSocket> serverSocket = new AtomicReference<ServerSocket>();
final CountDownLatch latch = new CountDownLatch(1);
final AtomicBoolean done = new AtomicBoolean();
this.executor.execute(() -> {
Executors.newSingleThreadExecutor().execute(() -> {
try {
ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(0);
serverSocket.set(server);
@@ -379,10 +377,10 @@ public class TcpSendingMessageHandlerTests extends AbstractTcpChannelAdapterTest
@Test
public void testNetLength() throws Exception {
final AtomicReference<ServerSocket> serverSocket = new AtomicReference<>();
final AtomicReference<ServerSocket> serverSocket = new AtomicReference<ServerSocket>();
final CountDownLatch latch = new CountDownLatch(1);
final AtomicBoolean done = new AtomicBoolean();
this.executor.execute(() -> {
Executors.newSingleThreadExecutor().execute(() -> {
try {
ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(0);
serverSocket.set(server);
@@ -438,7 +436,7 @@ public class TcpSendingMessageHandlerTests extends AbstractTcpChannelAdapterTest
final AtomicReference<ServerSocket> serverSocket = new AtomicReference<ServerSocket>();
final CountDownLatch latch = new CountDownLatch(1);
final AtomicBoolean done = new AtomicBoolean();
this.executor.execute(() -> {
Executors.newSingleThreadExecutor().execute(() -> {
try {
ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(0);
serverSocket.set(server);
@@ -497,7 +495,7 @@ public class TcpSendingMessageHandlerTests extends AbstractTcpChannelAdapterTest
final AtomicReference<ServerSocket> serverSocket = new AtomicReference<ServerSocket>();
final CountDownLatch latch = new CountDownLatch(1);
final AtomicBoolean done = new AtomicBoolean();
this.executor.execute(() -> {
Executors.newSingleThreadExecutor().execute(() -> {
try {
ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(0);
serverSocket.set(server);
@@ -546,10 +544,10 @@ public class TcpSendingMessageHandlerTests extends AbstractTcpChannelAdapterTest
@Test
public void testNioSerial() throws Exception {
final AtomicReference<ServerSocket> serverSocket = new AtomicReference<>();
final AtomicReference<ServerSocket> serverSocket = new AtomicReference<ServerSocket>();
final CountDownLatch latch = new CountDownLatch(1);
final AtomicBoolean done = new AtomicBoolean();
this.executor.execute(() -> {
Executors.newSingleThreadExecutor().execute(() -> {
try {
ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(0);
serverSocket.set(server);
@@ -600,12 +598,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();
this.executor.execute(() -> {
Executors.newSingleThreadExecutor().execute(() -> {
try {
ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(0);
serverSocket.set(server);
@@ -647,12 +645,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();
this.executor.execute(() -> {
Executors.newSingleThreadExecutor().execute(() -> {
try {
ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(0);
serverSocket.set(server);
@@ -694,12 +692,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();
this.executor.execute(() -> {
Executors.newSingleThreadExecutor().execute(() -> {
try {
ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(0);
serverSocket.set(server);
@@ -754,12 +752,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();
this.executor.execute(() -> {
Executors.newSingleThreadExecutor().execute(() -> {
try {
ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(0);
serverSocket.set(server);
@@ -814,13 +812,14 @@ 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>();
this.executor.execute(() -> {
final ExecutorService exec = Executors.newCachedThreadPool();
exec.execute(() -> {
try {
ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(0, 100);
serverSocket.set(server);
@@ -829,7 +828,7 @@ public class TcpSendingMessageHandlerTests extends AbstractTcpChannelAdapterTest
final Socket socket = server.accept();
serverSockets.add(socket);
final int j = i;
this.executor.execute(() -> {
exec.execute(() -> {
semaphore.release();
byte[] b = new byte[9];
try {
@@ -844,8 +843,7 @@ public class TcpSendingMessageHandlerTests extends AbstractTcpChannelAdapterTest
try {
socket.close();
}
catch (IOException e2) {
}
catch (IOException e2) { }
}
});
}
@@ -866,7 +864,7 @@ public class TcpSendingMessageHandlerTests extends AbstractTcpChannelAdapterTest
ccf.setDeserializer(serializer);
ccf.setSoTimeout(10000);
ccf.setSingleUse(true);
ccf.setTaskExecutor(this.executor);
ccf.setTaskExecutor(Executors.newCachedThreadPool());
ccf.start();
TcpSendingMessageHandler handler = new TcpSendingMessageHandler();
handler.setConnectionFactory(ccf);
@@ -904,7 +902,7 @@ public class TcpSendingMessageHandlerTests extends AbstractTcpChannelAdapterTest
final AtomicReference<ServerSocket> serverSocket = new AtomicReference<ServerSocket>();
final CountDownLatch latch = new CountDownLatch(1);
final AtomicBoolean done = new AtomicBoolean();
this.executor.execute(() -> {
Executors.newSingleThreadExecutor().execute(() -> {
try {
ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(0);
serverSocket.set(server);
@@ -975,7 +973,7 @@ public class TcpSendingMessageHandlerTests extends AbstractTcpChannelAdapterTest
final AtomicReference<ServerSocket> serverSocket = new AtomicReference<ServerSocket>();
final CountDownLatch latch = new CountDownLatch(1);
final AtomicBoolean done = new AtomicBoolean();
this.executor.execute(() -> {
Executors.newSingleThreadExecutor().execute(() -> {
try {
ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(0);
serverSocket.set(server);
@@ -1012,7 +1010,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();
@@ -1044,7 +1042,7 @@ public class TcpSendingMessageHandlerTests extends AbstractTcpChannelAdapterTest
final AtomicReference<ServerSocket> serverSocket = new AtomicReference<ServerSocket>();
final CountDownLatch latch = new CountDownLatch(1);
final AtomicBoolean done = new AtomicBoolean();
this.executor.execute(() -> {
Executors.newSingleThreadExecutor().execute(() -> {
try {
ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(0);
serverSocket.set(server);
@@ -1101,7 +1099,7 @@ public class TcpSendingMessageHandlerTests extends AbstractTcpChannelAdapterTest
final AtomicReference<ServerSocket> serverSocket = new AtomicReference<ServerSocket>();
final CountDownLatch latch = new CountDownLatch(1);
final AtomicBoolean done = new AtomicBoolean();
this.executor.execute(() -> {
Executors.newSingleThreadExecutor().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-2018 the original author or authors.
* Copyright 2002-2017 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,7 +68,6 @@ 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;
@@ -783,7 +782,7 @@ public class CachingClientConnectionFactoryTests {
invocation.callRealMethod();
String log = invocation.getArgument(0);
if (log.startsWith("Response")) {
new SimpleAsyncTaskExecutor()
Executors.newSingleThreadScheduledExecutor()
.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,6 +39,7 @@ 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;
@@ -51,7 +52,6 @@ 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,8 +233,7 @@ 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
new SimpleAsyncTaskExecutor()
.execute(factory::stop);
Executors.newSingleThreadExecutor().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,6 +35,7 @@ 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;
@@ -47,7 +48,6 @@ 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,9 +548,8 @@ public class FailoverClientConnectionFactoryTests {
}
private Holder setupAndStartServers(AbstractServerConnectionFactory server1,
AbstractServerConnectionFactory server2) {
Executor exec = new SimpleAsyncTaskExecutor();
AbstractServerConnectionFactory server2) throws Exception {
Executor exec = Executors.newCachedThreadPool();
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,11 +72,8 @@ 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;
@@ -114,14 +111,12 @@ 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<>();
this.executor.execute(() -> {
final AtomicReference<ServerSocket> serverSocket = new AtomicReference<ServerSocket>();
Executors.newSingleThreadExecutor().execute(() -> {
try {
ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(0);
logger.debug(testName.getMethodName() + " starting server for " + server.getLocalPort());
@@ -139,7 +134,7 @@ public class TcpNioConnectionTests {
TcpNioClientConnectionFactory factory = new TcpNioClientConnectionFactory("localhost",
serverSocket.get().getLocalPort());
factory.setApplicationEventPublisher(nullPublisher);
factory.setSoTimeout(100);
factory.setSoTimeout(1000);
factory.start();
try {
TcpConnection connection = factory.getConnection();
@@ -157,8 +152,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<>();
this.executor.execute(() -> {
final AtomicReference<ServerSocket> serverSocket = new AtomicReference<ServerSocket>();
Executors.newSingleThreadExecutor().execute(() -> {
try {
ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(0);
logger.debug(testName.getMethodName() + " starting server for " + server.getLocalPort());
@@ -178,14 +173,14 @@ public class TcpNioConnectionTests {
TcpNioClientConnectionFactory factory = new TcpNioClientConnectionFactory("localhost",
serverSocket.get().getLocalPort());
factory.setApplicationEventPublisher(nullPublisher);
factory.setSoTimeout(100);
factory.setSoTimeout(1000);
factory.start();
try {
TcpConnection connection = factory.getConnection();
connection.send(MessageBuilder.withPayload("Test").build());
int n = 0;
while (connection.isOpen()) {
Thread.sleep(10);
Thread.sleep(100);
if (n++ > 200) {
break;
}
@@ -202,8 +197,8 @@ public class TcpNioConnectionTests {
@Test
public void testMemoryLeak() throws Exception {
final CountDownLatch latch = new CountDownLatch(1);
final AtomicReference<ServerSocket> serverSocket = new AtomicReference<>();
this.executor.execute(() -> {
final AtomicReference<ServerSocket> serverSocket = new AtomicReference<ServerSocket>();
Executors.newSingleThreadExecutor().execute(() -> {
try {
ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(0);
logger.debug(testName.getMethodName() + " starting server for " + server.getLocalPort());
@@ -252,7 +247,7 @@ public class TcpNioConnectionTests {
TcpNioClientConnectionFactory factory = new TcpNioClientConnectionFactory("localhost", 0);
factory.setApplicationEventPublisher(nullPublisher);
factory.setNioHarvestInterval(100);
Map<SocketChannel, TcpNioConnection> connections = new HashMap<>();
Map<SocketChannel, TcpNioConnection> connections = new HashMap<SocketChannel, TcpNioConnection>();
SocketChannel chan1 = mock(SocketChannel.class);
SocketChannel chan2 = mock(SocketChannel.class);
SocketChannel chan3 = mock(SocketChannel.class);
@@ -339,9 +334,6 @@ public class TcpNioConnectionTests {
catch (ExecutionException e) {
assertEquals("Timed out waiting for buffer space", e.getCause().getMessage());
}
finally {
exec.shutdownNow();
}
}
@Test
@@ -383,8 +375,6 @@ public class TcpNioConnectionTests {
});
future.get(60, TimeUnit.SECONDS);
assertTrue(messageLatch.await(10, TimeUnit.SECONDS));
exec.shutdownNow();
}
@Test
@@ -473,14 +463,19 @@ public class TcpNioConnectionTests {
.getPropertyValue("channelInputStream");
final CountDownLatch latch = new CountDownLatch(1);
final byte[] out = new byte[4];
this.executor.execute(() -> {
try {
stream.read(out);
ExecutorService exec = Executors.newSingleThreadExecutor();
exec.execute(new Runnable() {
@Override
public void run() {
try {
stream.read(out);
}
catch (IOException e) {
e.printStackTrace();
}
latch.countDown();
}
catch (IOException e) {
e.printStackTrace();
}
latch.countDown();
});
Thread.sleep(1000);
assertEquals(0x00, out[0]);
@@ -604,18 +599,11 @@ 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 = 10;
final int numberOfSockets = 100;
TcpNioServerConnectionFactory factory = new TcpNioServerConnectionFactory(0);
factory.setApplicationEventPublisher(nullPublisher);
@@ -623,11 +611,16 @@ public class TcpNioConnectionTests {
factory.setTaskExecutor(compositeExec);
final CountDownLatch latch = new CountDownLatch(numberOfSockets * 4);
factory.registerListener(message -> {
if (!(message instanceof ErrorMessage)) {
latch.countDown();
factory.registerListener(new TcpListener() {
@Override
public boolean onMessage(Message<?> message) {
if (!(message instanceof ErrorMessage)) {
latch.countDown();
}
return false;
}
return false;
});
factory.start();
TestingUtilities.waitListening(factory, null);
@@ -644,7 +637,7 @@ public class TcpNioConnectionTests {
}
catch (ConnectException e) {
}
Thread.sleep(1);
Thread.sleep(100);
}
assertTrue("Could not open socket to localhost:" + port, n < 100);
sockets[i] = socket;
@@ -653,7 +646,7 @@ public class TcpNioConnectionTests {
sockets[i].getOutputStream().write("foo1 and...".getBytes());
sockets[i].getOutputStream().flush();
}
Thread.sleep(1);
Thread.sleep(100);
for (int i = 0; i < numberOfSockets; i++) {
sockets[i].getOutputStream().write(("...foo2\r\nbar1 and...").getBytes());
sockets[i].getOutputStream().flush();
@@ -666,7 +659,7 @@ public class TcpNioConnectionTests {
sockets[i].getOutputStream().write("foo3 and...".getBytes());
sockets[i].getOutputStream().flush();
}
Thread.sleep(1);
Thread.sleep(100);
for (int i = 0; i < numberOfSockets; i++) {
sockets[i].getOutputStream().write(("...foo4\r\nbar3 and...").getBytes());
sockets[i].getOutputStream().flush();
@@ -679,8 +672,6 @@ public class TcpNioConnectionTests {
assertTrue("latch is still " + latch.getCount(), latch.await(60, TimeUnit.SECONDS));
factory.stop();
cleanupCompositeExecutor(compositeExec);
}
private CompositeExecutor compositeExecutor() {
@@ -800,8 +791,6 @@ 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,7 +33,8 @@ import java.net.ServerSocket;
import java.net.Socket;
import java.nio.ByteBuffer;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.Executor;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.atomic.AtomicReference;
import javax.net.ServerSocketFactory;
@@ -45,7 +46,6 @@ 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,7 +62,6 @@ import org.springframework.messaging.support.GenericMessage;
/**
* @author Gary Russell
* @author Gavin Gray
*
* @since 2.0
*/
public class DeserializationTests {
@@ -420,7 +419,7 @@ public class DeserializationTests {
// eat SocketTimeoutException. Doesn't matter for this test
}
};
Executor exec = new SimpleAsyncTaskExecutor();
ExecutorService exec = Executors.newSingleThreadExecutor();
Message<?> message;

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

@@ -27,6 +27,7 @@ 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;
@@ -34,7 +35,6 @@ 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 = new SimpleAsyncTaskExecutor();
Executor executor = Executors.newFixedThreadPool(2);
executor.execute(catcher);
executor.execute(catcher);
assertTrue(listening.await(10000, TimeUnit.MILLISECONDS));
@@ -159,7 +159,7 @@ public class DatagramPacketMulticastSendingHandlerTests {
e.printStackTrace();
}
};
Executor executor = new SimpleAsyncTaskExecutor();
Executor executor = Executors.newFixedThreadPool(2);
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-2018 the original author or authors.
* Copyright 2002-2016 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,8 +39,6 @@ import org.springframework.messaging.Message;
* @author Mark Fisher
* @author Gary Russell
* @author Marcin Pilaczynski
* @author Artem Bilan
*
* @since 2.0
*/
public class DatagramPacketSendingHandlerTests {
@@ -52,20 +50,19 @@ public class DatagramPacketSendingHandlerTests {
final CountDownLatch received = new CountDownLatch(1);
final AtomicInteger testPort = new AtomicInteger();
final CountDownLatch listening = new CountDownLatch(1);
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();
}
});
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();
}
});
assertTrue(listening.await(10, TimeUnit.SECONDS));
UnicastSendingMessageHandler handler =
new UnicastSendingMessageHandler("localhost", testPort.get());
@@ -92,32 +89,31 @@ public class DatagramPacketSendingHandlerTests {
final CountDownLatch listening = new CountDownLatch(1);
final CountDownLatch ackListening = new CountDownLatch(1);
final CountDownLatch ackSent = new CountDownLatch(1);
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();
}
});
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();
}
});
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-2018 the original author or authors.
* Copyright 2002-2016 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,6 +31,7 @@ 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;
@@ -41,7 +42,6 @@ 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,7 +56,6 @@ import org.springframework.messaging.SubscribableChannel;
* @author Gary Russell
* @author Artem Bilan
* @author Marcin Pilaczynski
*
* @since 2.0
*
*/
@@ -186,19 +185,18 @@ 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
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();
}
});
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();
}
});
Message<byte[]> receivedMessage = (Message<byte[]>) channel.receive(10000);
assertEquals(new String(message.getPayload()), new String(receivedMessage.getPayload()));
String replyString = "reply:" + System.currentTimeMillis();