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

Shutdown executors in TCP/IP tests

Browse Source
This commit is contained in:
Artem Bilan
2018-01-22 16:17:20 -05:00
parent 2736de99de
commit 170292acf3
15 changed files with 298 additions and 219 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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);

49
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();
@@ -158,6 +161,7 @@ public class TcpOutboundGatewayTests {
assertTrue(replies.remove("Reply" + i));
}
done.set(true);
ccf.stop();
serverSocket.get().close();
}
@@ -165,8 +169,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);
@@ -213,6 +217,7 @@ public class TcpOutboundGatewayTests {
}
done.set(true);
gateway.stop();
ccf.stop();
serverSocket.get().close();
}
@@ -220,8 +225,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 +265,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 {
@@ -294,6 +299,7 @@ public class TcpOutboundGatewayTests {
}
done.set(true);
gateway.stop();
ccf.stop();
serverSocket.get().close();
}
@@ -344,7 +350,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 +404,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;
}));
@@ -433,6 +439,7 @@ public class TcpOutboundGatewayTests {
done.set(true);
assertEquals(0, TestUtils.getPropertyValue(gateway, "pendingReplies", Map.class).size());
gateway.stop();
ccf.stop();
}
@Test
@@ -442,7 +449,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);
@@ -512,17 +519,18 @@ public class TcpOutboundGatewayTests {
done.set(true);
gateway.stop();
verify(mockConn1).send(Mockito.any(Message.class));
factory2.stop();
serverSocket.get().close();
}
@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);
@@ -595,6 +603,7 @@ public class TcpOutboundGatewayTests {
done.set(true);
gateway.stop();
verify(mockConn1).send(Mockito.any(Message.class));
factory2.stop();
serverSocket.get().close();
}
@@ -667,11 +676,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 +802,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");

69
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);
@@ -207,6 +209,7 @@ public class TcpSendingMessageHandlerTests extends AbstractTcpChannelAdapterTest
handler.start();
handler.stop();
adapter.stop();
ccf.stop();
serverSocket.get().close();
}
@@ -215,7 +218,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 +274,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 +327,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 +380,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 +439,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 +498,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 +547,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 +601,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 +648,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 +695,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 +755,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 +815,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 +830,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 +845,8 @@ public class TcpSendingMessageHandlerTests extends AbstractTcpChannelAdapterTest
try {
socket.close();
}
catch (IOException e2) { }
catch (IOException e2) {
}
}
});
}
@@ -864,7 +867,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 +905,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 +976,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 +1013,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 +1045,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 +1102,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);

10
spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/ConnectionEventTests.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.
@@ -73,7 +73,7 @@ import org.springframework.messaging.support.MessageBuilder;
/**
* @author Gary Russell
* @author Artyem Bilan
* @author Artem Bilan
*
* @since 3.0
*
@@ -216,6 +216,8 @@ public class ConnectionEventTests {
TcpConnectionFailedCorrelationEvent event = (TcpConnectionFailedCorrelationEvent) theEvent.get();
assertEquals("bar", event.getConnectionId());
assertSame(message, ((MessagingException) event.getCause()).getFailedMessage());
gw.stop();
scf.stop();
}
@Test
@@ -260,6 +262,8 @@ public class ConnectionEventTests {
messagingException = (MessagingException) event.getCause();
assertSame(message, messagingException.getFailedMessage());
assertEquals("Cannot correlate response - no connection id", messagingException.getMessage());
gw.stop();
ccf.stop();
}
private void testServerExceptionGuts(AbstractServerConnectionFactory factory) throws Exception {
@@ -336,6 +340,8 @@ public class ConnectionEventTests {
TcpConnectionFailedEvent event = (TcpConnectionFailedEvent) failEvent.get();
assertSame(e, event.getCause());
}
ccf.stop();
}
}

10
spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/ConnectionFactoryShutDownTests.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.
@@ -29,6 +29,8 @@ import org.springframework.util.StopWatch;
/**
* @author Gary Russell
* @author Artem Bilan
*
* @since 2.2
*
*/
@@ -39,7 +41,7 @@ public class ConnectionFactoryShutDownTests {
final AbstractConnectionFactory factory = new AbstractConnectionFactory(0) {
@Override
public TcpConnection getConnection() throws Exception {
public TcpConnection getConnection() {
return null;
}
@@ -53,7 +55,7 @@ public class ConnectionFactoryShutDownTests {
try {
while (true) {
factory.getTaskExecutor();
Thread.sleep(100);
Thread.sleep(10);
}
}
catch (MessagingException e1) {
@@ -68,7 +70,7 @@ public class ConnectionFactoryShutDownTests {
watch.start();
factory.stop();
watch.stop();
assertTrue("Expected < 10000, was:" + watch.getLastTaskTimeMillis(), watch.getLastTaskTimeMillis() < 10000);
assertTrue("Expected < 10000, was: " + watch.getLastTaskTimeMillis(), watch.getLastTaskTimeMillis() < 10000);
assertTrue(latch1.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");

83
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();
@@ -145,6 +150,7 @@ public class TcpNioConnectionTests {
":" + e.getMessage(), e instanceof SocketTimeoutException);
}
done.countDown();
factory.stop();
serverSocket.get().close();
}
@@ -152,8 +158,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 +179,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;
}
@@ -191,14 +197,15 @@ public class TcpNioConnectionTests {
fail("Unexpected exception " + e);
}
done.countDown();
factory.stop();
serverSocket.get().close();
}
@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 +254,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 +341,9 @@ public class TcpNioConnectionTests {
catch (ExecutionException e) {
assertEquals("Timed out waiting for buffer space", e.getCause().getMessage());
}
finally {
exec.shutdownNow();
}
}
@Test
@@ -375,6 +385,8 @@ public class TcpNioConnectionTests {
});
future.get(60, TimeUnit.SECONDS);
assertTrue(messageLatch.await(10, TimeUnit.SECONDS));
exec.shutdownNow();
}
@Test
@@ -463,19 +475,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 +606,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 +625,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 +646,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 +655,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 +668,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 +681,8 @@ public class TcpNioConnectionTests {
assertTrue("latch is still " + latch.getCount(), latch.await(60, TimeUnit.SECONDS));
factory.stop();
cleanupCompositeExecutor(compositeExec);
}
private CompositeExecutor compositeExecutor() {
@@ -791,6 +802,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 {

56
spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/TcpNioConnectionWriteTests.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.
@@ -38,6 +38,8 @@ import org.springframework.integration.support.MessageBuilder;
/**
* @author Gary Russell
* @author Artem Bilan
*
* @since 2.0
*/
public class TcpNioConnectionWriteTests {
@@ -61,9 +63,10 @@ public class TcpNioConnectionWriteTests {
server.setSoTimeout(10000);
final CountDownLatch latch = new CountDownLatch(1);
Thread t = new Thread(() -> {
AbstractConnectionFactory ccf = null;
try {
ByteArrayLengthHeaderSerializer serializer = new ByteArrayLengthHeaderSerializer();
AbstractConnectionFactory ccf = getClientConnectionFactory(false, port, serializer);
ccf = getClientConnectionFactory(false, port, serializer);
TcpConnection connection = ccf.getConnection();
connection.send(MessageBuilder.withPayload(testString.getBytes()).build());
latch.await(10, TimeUnit.SECONDS);
@@ -71,6 +74,11 @@ public class TcpNioConnectionWriteTests {
catch (Exception e) {
e.printStackTrace();
}
finally {
if (ccf != null) {
ccf.stop();
}
}
});
t.setDaemon(true);
t.start();
@@ -94,9 +102,10 @@ public class TcpNioConnectionWriteTests {
server.setSoTimeout(10000);
final CountDownLatch latch = new CountDownLatch(1);
Thread t = new Thread(() -> {
AbstractConnectionFactory ccf = null;
try {
ByteArrayStxEtxSerializer serializer = new ByteArrayStxEtxSerializer();
AbstractConnectionFactory ccf = getClientConnectionFactory(false, port, serializer);
ccf = getClientConnectionFactory(false, port, serializer);
TcpConnection connection = ccf.getConnection();
connection.send(MessageBuilder.withPayload(testString.getBytes()).build());
latch.await(10, TimeUnit.SECONDS);
@@ -104,6 +113,11 @@ public class TcpNioConnectionWriteTests {
catch (Exception e) {
e.printStackTrace();
}
finally {
if (ccf != null) {
ccf.stop();
}
}
});
t.setDaemon(true);
t.start();
@@ -127,9 +141,10 @@ public class TcpNioConnectionWriteTests {
server.setSoTimeout(10000);
final CountDownLatch latch = new CountDownLatch(1);
Thread t = new Thread(() -> {
AbstractConnectionFactory ccf = null;
try {
ByteArrayCrLfSerializer serializer = new ByteArrayCrLfSerializer();
AbstractConnectionFactory ccf = getClientConnectionFactory(false, port, serializer);
ccf = getClientConnectionFactory(false, port, serializer);
TcpConnection connection = ccf.getConnection();
connection.send(MessageBuilder.withPayload(testString.getBytes()).build());
latch.await(10, TimeUnit.SECONDS);
@@ -137,6 +152,11 @@ public class TcpNioConnectionWriteTests {
catch (Exception e) {
e.printStackTrace();
}
finally {
if (ccf != null) {
ccf.stop();
}
}
});
t.setDaemon(true);
t.start();
@@ -160,9 +180,10 @@ public class TcpNioConnectionWriteTests {
server.setSoTimeout(10000);
final CountDownLatch latch = new CountDownLatch(1);
Thread t = new Thread(() -> {
AbstractConnectionFactory ccf = null;
try {
ByteArrayLengthHeaderSerializer serializer = new ByteArrayLengthHeaderSerializer();
AbstractConnectionFactory ccf = getClientConnectionFactory(true, port, serializer);
ccf = getClientConnectionFactory(true, port, serializer);
TcpConnection connection = ccf.getConnection();
connection.send(MessageBuilder.withPayload(testString.getBytes()).build());
latch.await(10, TimeUnit.SECONDS);
@@ -170,6 +191,11 @@ public class TcpNioConnectionWriteTests {
catch (Exception e) {
e.printStackTrace();
}
finally {
if (ccf != null) {
ccf.stop();
}
}
});
t.setDaemon(true);
t.start();
@@ -193,9 +219,10 @@ public class TcpNioConnectionWriteTests {
server.setSoTimeout(10000);
final CountDownLatch latch = new CountDownLatch(1);
Thread t = new Thread(() -> {
AbstractConnectionFactory ccf = null;
try {
ByteArrayStxEtxSerializer serializer = new ByteArrayStxEtxSerializer();
AbstractConnectionFactory ccf = getClientConnectionFactory(true, port, serializer);
ccf = getClientConnectionFactory(true, port, serializer);
TcpConnection connection = ccf.getConnection();
connection.send(MessageBuilder.withPayload(testString.getBytes()).build());
latch.await(10, TimeUnit.SECONDS);
@@ -203,6 +230,11 @@ public class TcpNioConnectionWriteTests {
catch (Exception e) {
e.printStackTrace();
}
finally {
if (ccf != null) {
ccf.stop();
}
}
});
t.setDaemon(true);
t.start();
@@ -226,9 +258,10 @@ public class TcpNioConnectionWriteTests {
server.setSoTimeout(10000);
final CountDownLatch latch = new CountDownLatch(1);
Thread t = new Thread(() -> {
AbstractConnectionFactory ccf = null;
try {
ByteArrayCrLfSerializer serializer = new ByteArrayCrLfSerializer();
AbstractConnectionFactory ccf = getClientConnectionFactory(true, port, serializer);
ccf = getClientConnectionFactory(true, port, serializer);
TcpConnection connection = ccf.getConnection();
connection.send(MessageBuilder.withPayload(testString.getBytes()).build());
latch.await(10, TimeUnit.SECONDS);
@@ -236,6 +269,11 @@ public class TcpNioConnectionWriteTests {
catch (Exception e) {
e.printStackTrace();
}
finally {
if (ccf != null) {
ccf.stop();
}
}
});
t.setDaemon(true);
t.start();
@@ -251,10 +289,6 @@ public class TcpNioConnectionWriteTests {
latch.countDown();
}
/**
* @param is
* @param buff
*/
private void readFully(InputStream is, byte[] buff) throws IOException {
for (int i = 0; i < buff.length; i++) {
buff[i] = (byte) is.read();

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();