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

Catch async MessageTimeoutException in IP test

Browse Source 
- fix async exec thread names
This commit is contained in:
Gary Russell
2021-03-12 13:36:26 -05:00
parent 84ea4422a2
commit 2522c79cb3
11 changed files with 26 additions and 19 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -99,7 +99,7 @@ public class TcpOutboundGatewayTests {
private static final Log logger = LogFactory.getLog(TcpOutboundGatewayTests.class);
private final AsyncTaskExecutor executor = new SimpleAsyncTaskExecutor("TcpOutboundGatewayTests");
private final AsyncTaskExecutor executor = new SimpleAsyncTaskExecutor("TcpOutboundGatewayTests-");
@Test
void testGoodNetSingle() throws Exception {

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

@@ -100,7 +100,7 @@ public class TcpReceivingChannelAdapterTests extends AbstractTcpChannelAdapterTe
final CountDownLatch latch1 = new CountDownLatch(1);
final CountDownLatch latch2 = new CountDownLatch(1);
final AtomicBoolean done = new AtomicBoolean();
new SimpleAsyncTaskExecutor("testNetClientMode").execute(() -> {
new SimpleAsyncTaskExecutor("testNetClientMode-").execute(() -> {
try {
ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(0, 10);
serverSocket.set(server);
@@ -412,7 +412,7 @@ public class TcpReceivingChannelAdapterTests extends AbstractTcpChannelAdapterTe
handler.setConnectionFactory(scf);
TcpReceivingChannelAdapter adapter = new TcpReceivingChannelAdapter();
adapter.setConnectionFactory(scf);
Executor te = new SimpleAsyncTaskExecutor("testNioSingleSharedMany");
Executor te = new SimpleAsyncTaskExecutor("testNioSingleSharedMany-");
scf.setTaskExecutor(te);
scf.start();
QueueChannel channel = new QueueChannel();

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

@@ -84,7 +84,7 @@ public class TcpSendingMessageHandlerTests extends AbstractTcpChannelAdapterTest
private static final Log logger = LogFactory.getLog(TcpSendingMessageHandlerTests.class);
private final AsyncTaskExecutor executor = new SimpleAsyncTaskExecutor("TcpSendingMessageHandlerTests");
private final AsyncTaskExecutor executor = new SimpleAsyncTaskExecutor("TcpSendingMessageHandlerTests-");
private void readFully(InputStream is, byte[] buff) throws IOException {
for (int i = 0; i < buff.length; i++) {

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

@@ -65,6 +65,7 @@ import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.context.ApplicationEventPublisher;
import org.springframework.core.log.LogAccessor;
import org.springframework.core.task.SimpleAsyncTaskExecutor;
import org.springframework.integration.MessageTimeoutException;
import org.springframework.integration.channel.QueueChannel;
import org.springframework.integration.ip.IpHeaders;
import org.springframework.integration.ip.tcp.TcpOutboundGateway;
@@ -751,13 +752,19 @@ public class CachingClientConnectionFactoryTests {
invocation.callRealMethod();
String log = ((Supplier<String>) invocation.getArgument(0)).get();
if (log.startsWith("Response")) {
new SimpleAsyncTaskExecutor("testGatewayRelease")
.execute(() -> gate.handleMessage(new GenericMessage<>("bar")));
new SimpleAsyncTaskExecutor("testGatewayRelease-")
.execute(() -> {
try {
gate.handleMessage(new GenericMessage<>("bar"));
}
catch (MessageTimeoutException e) {
}
});
// hold up the first thread until the second has added its pending reply
latch.await(20, TimeUnit.SECONDS);
this.latch.await(20, TimeUnit.SECONDS);
}
else if (log.startsWith("Added")) {
latch.countDown();
this.latch.countDown();
}
return null;
}

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

@@ -228,7 +228,7 @@ public class ConnectionFactoryTests {
factory.start();
assertThat(latch1.await(10, TimeUnit.SECONDS)).as("missing info log").isTrue();
// stop on a different thread because it waits for the executor
new SimpleAsyncTaskExecutor("testEarlyClose")
new SimpleAsyncTaskExecutor("testEarlyClose-")
.execute(factory::stop);
int n = 0;
DirectFieldAccessor accessor = new DirectFieldAccessor(factory);

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

@@ -609,7 +609,7 @@ public class FailoverClientConnectionFactoryTests {
private Holder setupAndStartServers(AbstractServerConnectionFactory server1,
AbstractServerConnectionFactory server2) {
Executor exec = new SimpleAsyncTaskExecutor("FailoverClientConnectionFactoryTests");
Executor exec = new SimpleAsyncTaskExecutor("FailoverClientConnectionFactoryTests-");
server1.setTaskExecutor(exec);
server2.setTaskExecutor(exec);
server1.setBeanName("server1");

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

@@ -109,7 +109,7 @@ public class TcpNioConnectionTests {
private final ApplicationEventPublisher nullPublisher = mock(ApplicationEventPublisher.class);
private final AsyncTaskExecutor executor = new SimpleAsyncTaskExecutor("TcpNioConnectionTests");
private final AsyncTaskExecutor executor = new SimpleAsyncTaskExecutor("TcpNioConnectionTests-");
@Test
public void testWriteTimeout(TestInfo testInfo) throws Exception {

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

@@ -407,7 +407,7 @@ public class DeserializationTests {
// eat SocketTimeoutException. Doesn't matter for this test
}
};
Executor exec = new SimpleAsyncTaskExecutor("testTimeoutWhileDecoding");
Executor exec = new SimpleAsyncTaskExecutor("-");
Message<?> message;
@@ -459,7 +459,7 @@ public class DeserializationTests {
// eat SocketTimeoutException. Doesn't matter for this test
}
};
Executor exec = new SimpleAsyncTaskExecutor("testTimeoutWithRawDeserializerEofIsTerminator");
Executor exec = new SimpleAsyncTaskExecutor("testTimeoutWithRawDeserializerEofIsTerminator-");
Message<?> message;

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

@@ -90,7 +90,7 @@ public class DatagramPacketMulticastSendingHandlerTests {
e.printStackTrace();
}
};
Executor executor = new SimpleAsyncTaskExecutor("verifySendMulticast");
Executor executor = new SimpleAsyncTaskExecutor("verifySendMulticast-");
executor.execute(catcher);
executor.execute(catcher);
assertThat(listening.await(10000, TimeUnit.MILLISECONDS)).isTrue();
@@ -191,7 +191,7 @@ public class DatagramPacketMulticastSendingHandlerTests {
e.printStackTrace();
}
};
Executor executor = new SimpleAsyncTaskExecutor("verifySendMulticastWithAcks");
Executor executor = new SimpleAsyncTaskExecutor("verifySendMulticastWithAcks-");
executor.execute(catcher);
executor.execute(catcher);
assertThat(listening.await(10000, TimeUnit.MILLISECONDS)).isTrue();

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

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2020 the original author or authors.
* Copyright 2002-2021 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.
@@ -53,7 +53,7 @@ public class DatagramPacketSendingHandlerTests {
final CountDownLatch received = new CountDownLatch(1);
final AtomicInteger testPort = new AtomicInteger();
final CountDownLatch listening = new CountDownLatch(1);
new SimpleAsyncTaskExecutor()
new SimpleAsyncTaskExecutor("verifySend-")
.execute(() -> {
try {
DatagramSocket socket = new DatagramSocket();
@@ -93,7 +93,7 @@ public class DatagramPacketSendingHandlerTests {
final CountDownLatch listening = new CountDownLatch(1);
final CountDownLatch ackListening = new CountDownLatch(1);
final CountDownLatch ackSent = new CountDownLatch(1);
new SimpleAsyncTaskExecutor()
new SimpleAsyncTaskExecutor("verifySendWithAck-")
.execute(() -> {
try {
DatagramSocket socket = new DatagramSocket();

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

@@ -190,7 +190,7 @@ 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()
new SimpleAsyncTaskExecutor("testUnicastReceiverWithReply-")
.execute(() -> {
DatagramPacket answer = new DatagramPacket(new byte[2000], 2000);
try {