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

Use Awaitility in TCP/UDP Tests

Browse Source
This commit is contained in:
Gary Russell
2020-05-19 14:14:02 -04:00
committed by Artem Bilan
parent b8f0fb215a
commit 75021beae4
13 changed files with 66 additions and 112 deletions
Download Patch File Download Diff File Expand all files Collapse all files

5
build.gradle
View File

@@ -47,11 +47,11 @@ ext {
activeMqVersion = '5.15.12'
apacheSshdVersion = '2.4.0'
avroVersion = '1.9.2'
aspectjVersion = '1.9.5'
assertjVersion = '3.16.1'
assertkVersion = '0.22'
awaitilityVersion = '4.0.2'
avroVersion = '1.9.2'
awaitilityVersion = '4.0.3'
commonsDbcp2Version = '2.7.0'
commonsIoVersion = '2.6'
commonsNetVersion = '3.6'
@@ -524,6 +524,7 @@ project('spring-integration-ip') {
api project(':spring-integration-core')
testImplementation project(':spring-integration-stream')
testImplementation project(':spring-integration-event')
testImplementation "org.hamcrest:hamcrest-core:$hamcrestVersion"
testRuntimeOnly "com.esotericsoftware:kryo-shaded:$kryoShadedVersion"
}
}

16
spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/ClientModeControlBusTests.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2019 the original author or authors.
* Copyright 2002-2020 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
@@ -17,7 +17,9 @@
package org.springframework.integration.ip.tcp;
import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.fail;
import static org.awaitility.Awaitility.await;
import java.time.Duration;
import org.junit.Before;
import org.junit.Test;
@@ -66,14 +68,8 @@ public class ClientModeControlBusTests {
@Test
public void test() throws Exception {
assertThat(controlBus.boolResult("@tcpIn.isClientMode()")).isTrue();
int n = 0;
while (!controlBus.boolResult("@tcpIn.isClientModeConnected()")) {
Thread.sleep(100);
n += 100;
if (n > 10000) {
fail("Connection never established");
}
}
await("Connection never established").atMost(Duration.ofSeconds(10))
.until(() -> controlBus.boolResult("@tcpIn.isClientModeConnected()"));
assertThat(controlBus.boolResult("@tcpIn.isRunning()")).isTrue();
assertThat(TestUtils.getPropertyValue(tcpIn, "taskScheduler")).isSameAs(taskScheduler);
controlBus.voidResult("@tcpIn.retryConnection()");

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

@@ -19,6 +19,7 @@ package org.springframework.integration.ip.tcp;
import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.catchThrowable;
import static org.assertj.core.api.Assertions.fail;
import static org.awaitility.Awaitility.await;
import static org.mockito.Mockito.doThrow;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.verify;
@@ -34,6 +35,7 @@ import java.io.UncheckedIOException;
import java.net.ServerSocket;
import java.net.Socket;
import java.net.SocketTimeoutException;
import java.time.Duration;
import java.util.ArrayList;
import java.util.Collections;
import java.util.HashSet;
@@ -890,10 +892,7 @@ public class TcpOutboundGatewayTests {
int n = 0;
@SuppressWarnings("unchecked")
Map<String, ?> pending = TestUtils.getPropertyValue(gateway, "pendingReplies", Map.class);
while (n++ < 100 && pending.size() == 0) {
Thread.sleep(100);
}
assertThat(pending.size() > 0).isTrue();
await().atMost(Duration.ofSeconds(10)).until(() -> pending.size() > 0);
String connectionId = pending.keySet().iterator().next();
this.executor.execute(() -> gateway.onMessage(new ErrorMessage(new RuntimeException(),
Collections.singletonMap(IpHeaders.CONNECTION_ID, connectionId))));

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

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2019 the original author or authors.
* Copyright 2002-2020 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.
@@ -18,6 +18,7 @@ package org.springframework.integration.ip.tcp.connection;
import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.assertThatExceptionOfType;
import static org.awaitility.Awaitility.await;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.ArgumentMatchers.anyInt;
import static org.mockito.ArgumentMatchers.anyString;
@@ -36,6 +37,7 @@ import java.io.UncheckedIOException;
import java.net.Socket;
import java.nio.ByteBuffer;
import java.nio.channels.SocketChannel;
import java.time.Duration;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.BlockingQueue;
@@ -489,10 +491,7 @@ public class CachingClientConnectionFactoryTests {
BlockingQueue<?> connections = TestUtils
.getPropertyValue(this.gatewayCF, "pool.available", BlockingQueue.class);
// wait until the connection is returned to the pool
int n = 0;
while (n++ < 100 && connections.size() == 0) {
Thread.sleep(100);
}
await().atMost(Duration.ofSeconds(10)).until(() -> connections.size() > 0);
// assert we use the same connection from the pool
toGateway.send(new GenericMessage<String>("Hello, world2!"));
@@ -525,11 +524,7 @@ public class CachingClientConnectionFactoryTests {
CachingClientConnectionFactory cccf = new CachingClientConnectionFactory(cf, 1);
cccf.start();
TcpConnection connection = cccf.getConnection();
int n = 0;
while (n++ < 100 && connection.isOpen()) {
Thread.sleep(100);
}
assertThat(connection.isOpen()).isFalse();
await().atMost(Duration.ofSeconds(10)).until(() -> !connection.isOpen());
cccf.stop();
}

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

@@ -19,6 +19,7 @@ package org.springframework.integration.ip.tcp.connection;
import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.assertThatExceptionOfType;
import static org.assertj.core.api.Assertions.fail;
import static org.awaitility.Awaitility.await;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.ArgumentMatchers.contains;
import static org.mockito.Mockito.atLeast;
@@ -30,6 +31,7 @@ import static org.mockito.Mockito.when;
import java.net.InetSocketAddress;
import java.net.SocketAddress;
import java.time.Duration;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
@@ -151,12 +153,8 @@ public class ConnectionFactoryTests {
assertThat(serverFactory.closeConnection(servers.get(0))).isTrue();
servers = serverFactory.getOpenConnectionIds();
assertThat(servers.size()).isEqualTo(0);
int n = 0;
await().atMost(Duration.ofSeconds(10)).until(() -> clientFactory.getOpenConnectionIds().size() == 0);
clients = clientFactory.getOpenConnectionIds();
while (n++ < 100 && clients.size() > 0) {
Thread.sleep(100);
clients = clientFactory.getOpenConnectionIds();
}
assertThat(clients.size()).isEqualTo(0);
assertThat(eventLatch.await(10, TimeUnit.SECONDS)).isTrue();
assertThat(events.size())
@@ -229,10 +227,8 @@ public class ConnectionFactoryTests {
.execute(factory::stop);
int n = 0;
DirectFieldAccessor accessor = new DirectFieldAccessor(factory);
while (n++ < 200 && accessor.getPropertyValue(property) != null) {
Thread.sleep(100);
}
assertThat(n < 200).as("Stop was not invoked in time").isTrue();
await("Stop was not invoked in time").atMost(Duration.ofSeconds(20))
.until(() -> accessor.getPropertyValue(property) == null);
latch2.countDown();
assertThat(latch3.await(10, TimeUnit.SECONDS)).as("missing debug log").isTrue();
String expected = "bean 'foo', port=" + factory.getPort() + message;

14
spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/PushbackTcpTests.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2017-2019 the original author or authors.
* Copyright 2017-2020 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
@@ -17,10 +17,12 @@
package org.springframework.integration.ip.tcp.connection;
import static org.assertj.core.api.Assertions.assertThat;
import static org.awaitility.Awaitility.await;
import java.io.IOException;
import java.io.InputStream;
import java.io.PushbackInputStream;
import java.time.Duration;
import java.util.Collections;
import org.junit.Test;
@@ -138,14 +140,8 @@ public class PushbackTcpTests {
}
private int waitForPort(AbstractServerConnectionFactory serverCF) throws InterruptedException {
int port = serverCF.getPort();
int n = 0;
while (n++ < 200 && port == 0) {
Thread.sleep(100);
port = serverCF.getPort();
}
assertThat(n < 200).isTrue();
return port;
await().atMost(Duration.ofSeconds(20)).until(() -> serverCF.getPort() > 0);
return serverCF.getPort();
}
@Configuration

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

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2019 the original author or authors.
* Copyright 2002-2020 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
@@ -17,9 +17,10 @@
package org.springframework.integration.ip.tcp.connection;
import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.fail;
import static org.awaitility.Awaitility.with;
import java.net.Socket;
import java.time.Duration;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.CountDownLatch;
@@ -532,14 +533,10 @@ public class TcpNioConnectionReadTests {
private void whileOpen(Semaphore semaphore, final List<TcpConnection> added)
throws InterruptedException {
int n = 0;
assertThat(semaphore.tryAcquire(10000, TimeUnit.MILLISECONDS)).isTrue();
while (added.get(0).isOpen()) {
Thread.sleep(50);
if (n++ > 400) {
fail("Failed to close socket");
}
}
with().pollInterval(Duration.ofMillis(50)).await("Failed to close socket")
.atMost(Duration.ofSeconds(20))
.until(() -> !added.get(0).isOpen());
}
}

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

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2019 the original author or authors.
* Copyright 2002-2020 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.
@@ -18,6 +18,8 @@ package org.springframework.integration.ip.tcp.connection;
import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.fail;
import static org.awaitility.Awaitility.await;
import static org.awaitility.Awaitility.with;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.ArgumentMatchers.contains;
import static org.mockito.Mockito.doAnswer;
@@ -39,6 +41,7 @@ import java.nio.ByteBuffer;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.SocketChannel;
import java.time.Duration;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
@@ -186,14 +189,10 @@ public class TcpNioConnectionTests {
try {
TcpConnection connection = factory.getConnection();
connection.send(MessageBuilder.withPayload("Test").build());
int n = 0;
while (connection.isOpen()) {
Thread.sleep(10);
if (n++ > 200) {
break;
}
}
assertThat(!connection.isOpen()).isTrue();
with().pollInterval(Duration.ofMillis(10))
.await()
.atMost(Duration.ofSeconds(10))
.until(() -> !connection.isOpen());
}
catch (Exception e) {
fail("Unexpected exception " + e);
@@ -234,14 +233,7 @@ public class TcpNioConnectionTests {
connection.close();
assertThat(!connection.isOpen()).isTrue();
TestUtils.getPropertyValue(factory, "selector", Selector.class).wakeup();
int n = 0;
while (connections.size() > 0) {
Thread.sleep(100);
if (n++ > 100) {
break;
}
}
assertThat(connections.size()).isEqualTo(0);
await().atMost(Duration.ofSeconds(10)).until(() -> connections.size() == 0);
}
catch (Exception e) {
e.printStackTrace();

20
spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/ThreadAffinityClientConnectionFactoryTests.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2017-2019 the original author or authors.
* Copyright 2017-2020 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
@@ -17,7 +17,9 @@
package org.springframework.integration.ip.tcp.connection;
import static org.assertj.core.api.Assertions.assertThat;
import static org.awaitility.Awaitility.await;
import java.time.Duration;
import java.util.Collections;
import org.junit.Test;
@@ -68,14 +70,8 @@ public class ThreadAffinityClientConnectionFactoryTests {
}
private int waitForPort(AbstractServerConnectionFactory serverCF) throws InterruptedException {
int port = serverCF.getPort();
int n = 0;
while (n++ < 200 && port == 0) {
Thread.sleep(100);
port = serverCF.getPort();
}
assertThat(n < 200).isTrue();
return port;
await().atMost(Duration.ofSeconds(20)).until(() -> serverCF.getPort() > 0);
return serverCF.getPort();
}
protected void doTest(AnnotationConfigApplicationContext server, AbstractServerConnectionFactory serverCF,
@@ -102,11 +98,7 @@ public class ThreadAffinityClientConnectionFactoryTests {
assertThat(replyC.getPayload()).isEqualTo(replyD.getPayload());
assertThat(replyC.getPayload()).isNotEqualTo(replyA.getPayload());
System.getProperties().remove(PORT);
int n = 0;
while (n++ < 200 && serverCF.getOpenConnectionIds().size() > 0) {
Thread.sleep(100);
}
assertThat(n).isLessThan(200);
await().atMost(Duration.ofSeconds(20)).until(() -> serverCF.getOpenConnectionIds().size() == 0);
client.close();
server.close();
}

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

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2019 the original author or authors.
* Copyright 2002-2020 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
@@ -17,11 +17,13 @@
package org.springframework.integration.ip.udp;
import static org.assertj.core.api.Assertions.assertThat;
import static org.awaitility.Awaitility.await;
import static org.mockito.Mockito.mock;
import java.net.DatagramPacket;
import java.net.DatagramSocket;
import java.net.InetSocketAddress;
import java.time.Duration;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
@@ -138,12 +140,8 @@ public class DatagramPacketSendingHandlerTests {
handler.stop();
}
public void waitAckListening(UnicastSendingMessageHandler handler) throws InterruptedException {
int n = 0;
while (n++ < 100 && handler.getAckPort() == 0) {
Thread.sleep(100);
}
assertThat(n < 100).isTrue();
public void waitAckListening(UnicastSendingMessageHandler handler) {
await("Handler not listening").atMost(Duration.ofSeconds(10)).until(() -> handler.getAckPort() > 0);
}
}

10
spring-integration-ip/src/test/java/org/springframework/integration/ip/udp/UdpMulticastEndToEndTests.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2019 the original author or authors.
* Copyright 2002-2020 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.
@@ -18,7 +18,9 @@ package org.springframework.integration.ip.udp;
import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.fail;
import static org.awaitility.Awaitility.await;
import java.time.Duration;
import java.util.Date;
import java.util.Properties;
import java.util.concurrent.CountDownLatch;
@@ -85,11 +87,7 @@ public class UdpMulticastEndToEndTests implements Runnable {
UdpMulticastEndToEndTests launcher = new UdpMulticastEndToEndTests();
Thread t = new Thread(launcher);
t.start(); // launch the receiver
int n = 0;
while (n++ < 100 && launcher.getReceiverPort() == 0) {
Thread.sleep(100);
}
assertThat(n < 100).as("Receiver failed to listen").isTrue();
await("Receiver failed to listen").atMost(Duration.ofSeconds(10)).until(() -> launcher.getReceiverPort() > 0);
ClassPathXmlApplicationContext applicationContext = createContext(launcher, location);
launcher.launchSender(applicationContext);

10
spring-integration-ip/src/test/java/org/springframework/integration/ip/udp/UdpUnicastEndToEndTests.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2019 the original author or authors.
* Copyright 2002-2020 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.
@@ -18,7 +18,9 @@ package org.springframework.integration.ip.udp;
import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.fail;
import static org.awaitility.Awaitility.await;
import java.time.Duration;
import java.util.Date;
import java.util.Properties;
import java.util.concurrent.CountDownLatch;
@@ -102,11 +104,7 @@ public class UdpUnicastEndToEndTests implements Runnable {
UdpUnicastEndToEndTests launcher = new UdpUnicastEndToEndTests();
Thread t = new Thread(launcher);
t.start(); // launch the receiver
int n = 0;
while (n++ < 100 && launcher.getReceiverPort() == 0) {
Thread.sleep(100);
}
assertThat(n < 100).as("Receiver failed to listen").isTrue();
await("Receiver failed to listen").atMost(Duration.ofSeconds(10)).until(() -> launcher.getReceiverPort() > 0);
ClassPathXmlApplicationContext applicationContext = createContext(launcher, location);
launcher.launchSender(applicationContext);

14
spring-integration-ip/src/test/java/org/springframework/integration/ip/util/SocketTestUtils.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2019 the original author or authors.
* Copyright 2002-2020 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
@@ -16,6 +16,8 @@
package org.springframework.integration.ip.util;
import static org.awaitility.Awaitility.await;
import java.io.IOException;
import java.io.ObjectOutputStream;
import java.io.OutputStream;
@@ -25,6 +27,7 @@ import java.net.NetworkInterface;
import java.net.Socket;
import java.net.UnknownHostException;
import java.nio.ByteBuffer;
import java.time.Duration;
import java.util.Enumeration;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
@@ -443,14 +446,7 @@ public class SocketTestUtils {
}
public static void waitListening(AbstractInternetProtocolReceivingChannelAdapter adapter) throws Exception {
int n = 0;
while (!adapter.isListening()) {
Thread.sleep(100);
if (n++ > 100) {
throw new Exception("Gateway failed to listen");
}
}
await("Adapter not listening").atMost(Duration.ofSeconds(10)).until(() -> adapter.isListening());
}
}