Use Awaitility in tests

Author: garyrussell

build.gradle

@@ -40,6 +40,7 @@ ext {
 
 	assertjVersion = '3.15.0'
 	assertkVersion = '0.20'
+	awaitilityVersion = '4.0.3'
 	commonsHttpClientVersion = '4.5.10'
 	commonsPoolVersion = '2.8.0'
 	googleJsr305Version = '3.0.2'
@@ -55,9 +56,9 @@ ext {
 	rabbitmqVersion = project.hasProperty('rabbitmqVersion') ? project.rabbitmqVersion : '5.9.0'
 	rabbitmqHttpClientVersion = '3.2.0.RELEASE'
 	reactorVersion = 'Dysprosium-SR7'
-	springDataCommonsVersion = '2.3.0.RC1'
-	springVersion = project.hasProperty('springVersion') ? project.springVersion : '5.2.5.RELEASE'
-	springRetryVersion = '1.2.5.RELEASE'
+	springDataCommonsVersion = '2.3.0.RELEASE'
+	springVersion = project.hasProperty('springVersion') ? project.springVersion : '5.2.6.RELEASE'
+	springRetryVersion = '1.3.0'
 }
 
 nohttp {
@@ -161,10 +162,14 @@ subprojects { subproject ->
 		testImplementation 'org.junit.jupiter:junit-jupiter-params'
 		testImplementation 'org.junit.jupiter:junit-jupiter-engine'
 		testImplementation 'org.junit.platform:junit-platform-launcher'
+		testImplementation("org.awaitility:awaitility:$awaitilityVersion") {
+			exclude group: 'org.hamcrest'
+		}
 
 		// To avoid compiler warnings about @API annotations in JUnit code
 		testCompileOnly 'org.apiguardian:apiguardian-api:1.0.0'
-
+		
+		testCompileOnly "com.google.code.findbugs:jsr305:$googleJsr305Version"
 		testImplementation 'org.jetbrains.kotlin:kotlin-reflect'
 		testImplementation 'org.jetbrains.kotlin:kotlin-stdlib-jdk8'
 

spring-rabbit/src/test/java/org/springframework/amqp/rabbit/AsyncRabbitTemplateTests.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2016-2019 the original author or authors.
+ * Copyright 2016-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 @@
 
 import static org.assertj.core.api.Assertions.assertThat;
 import static org.assertj.core.api.Assertions.fail;
+import static org.awaitility.Awaitility.await;
 
 import java.util.Map;
 import java.util.UUID;
@@ -175,13 +176,9 @@ public void testMessage1ArgDirect() throws Exception {
 	}
 
 	private void waitForZeroInUseConsumers() throws InterruptedException {
-		int n = 0;
 		Map<?, ?> inUseConsumers = TestUtils
 				.getPropertyValue(this.asyncDirectTemplate, "directReplyToContainer.inUseConsumerChannels", Map.class);
-		while (n++ < 100 && inUseConsumers.size() > 0) {
-			Thread.sleep(100);
-		}
-		assertThat(inUseConsumers).hasSize(0);
+		await().until(() -> inUseConsumers.size() == 0);
 	}
 
 	@Test

spring-rabbit/src/test/java/org/springframework/amqp/rabbit/annotation/ComplexTypeJsonIntegrationTests.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2016-2019 the original author or authors.
+ * Copyright 2016-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,6 +17,7 @@
 package org.springframework.amqp.rabbit.annotation;
 
 import static org.assertj.core.api.Assertions.assertThat;
+import static org.awaitility.Awaitility.await;
 
 import java.util.concurrent.TimeUnit;
 
@@ -111,13 +112,9 @@ public void testReceiveNoWait() throws Exception {
 			m.getMessageProperties().getHeaders().remove("__TypeId__");
 			return m;
 		});
-		Foo<Bar<Baz, Qux>> foo =
-				this.rabbitTemplate.receiveAndConvert(new ParameterizedTypeReference<Foo<Bar<Baz, Qux>>>() { });
-		int n = 0;
-		while (n++ < 100 && foo == null) {
-			Thread.sleep(100);
-			foo = this.rabbitTemplate.receiveAndConvert(new ParameterizedTypeReference<Foo<Bar<Baz, Qux>>>() { });
-		}
+		Foo<Bar<Baz, Qux>> foo = await().until(
+				() -> this.rabbitTemplate.receiveAndConvert(new ParameterizedTypeReference<Foo<Bar<Baz, Qux>>>() { }),
+				msg -> msg != null);
 		verifyFooBarBazQux(foo);
 	}
 

spring-rabbit/src/test/java/org/springframework/amqp/rabbit/annotation/ConsumerBatchingTests.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2019 the original author or authors.
+ * Copyright 2019-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,6 +17,7 @@
 package org.springframework.amqp.rabbit.annotation;
 
 import static org.assertj.core.api.Assertions.assertThat;
+import static org.awaitility.Awaitility.await;
 
 import java.io.IOException;
 import java.io.Serializable;
@@ -82,11 +83,9 @@ public void replayWholeBatch() throws InterruptedException {
 		assertThat(this.listener.foos)
 			.extracting(foo -> foo.getBar())
 			.contains("foo", "bar", "baz", "qux", "foo", "bar", "baz", "qux");
-		Timer timer = null;
-		int n = 0;
-		while (timer == null && n++ < 100) {
+		Timer timer = await().until(() -> {
 			try {
-				timer = this.meterRegistry.get("spring.rabbitmq.listener")
+				return this.meterRegistry.get("spring.rabbitmq.listener")
 						.tag("listener.id", "batch.1")
 						.tag("queue", "[c.batch.1]")
 						.tag("result", "success")
@@ -95,9 +94,9 @@ public void replayWholeBatch() throws InterruptedException {
 						.timer();
 			}
 			catch (@SuppressWarnings("unused") Exception e) {
-				Thread.sleep(100);
+				return null;
 			}
-		}
+		}, tim -> tim != null);
 		assertThat(timer).isNotNull();
 		assertThat(timer.count()).isEqualTo(1L);
 		timer = this.meterRegistry.get("spring.rabbitmq.listener")

spring-rabbit/src/test/java/org/springframework/amqp/rabbit/annotation/EnableRabbitIntegrationTests.java

@@ -18,6 +18,7 @@
 
 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.mockito.Mockito.doAnswer;
 import static org.mockito.Mockito.mock;
 
@@ -872,11 +873,9 @@ public void messagingMessageReturned() throws InterruptedException {
 		assertThat(message).isNotNull();
 		assertThat(new String(message.getBody())).isEqualTo("{\"field\":\"MESSAGING\"}");
 		assertThat(message.getMessageProperties().getHeaders().get("foo")).isEqualTo("bar");
-		Timer timer = null;
-		int n = 0;
-		while (timer == null && n++ < 100) {
+		Timer timer = await().until(() -> {
 			try {
-				timer = this.meterRegistry.get("spring.rabbitmq.listener")
+				return this.meterRegistry.get("spring.rabbitmq.listener")
 						.tag("listener.id", "list.of.messages")
 						.tag("queue", "test.messaging.message")
 						.tag("result", "success")
@@ -885,10 +884,9 @@ public void messagingMessageReturned() throws InterruptedException {
 						.timer();
 			}
 			catch (@SuppressWarnings("unused") Exception e) {
-				Thread.sleep(100);
+				return null;
 			}
-		}
-		assertThat(timer).isNotNull();
+		}, tim -> tim != null);
 		assertThat(timer.count()).isEqualTo(1L);
 	}
 

spring-rabbit/src/test/java/org/springframework/amqp/rabbit/config/QueueParserIntegrationTests.java

@@ -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,8 +17,9 @@
 package org.springframework.amqp.rabbit.config;
 
 import static org.assertj.core.api.Assertions.assertThat;
+import static org.awaitility.Awaitility.await;
 
-import java.util.Properties;
+import java.time.Duration;
 
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
@@ -66,16 +67,9 @@ public void testArgumentsQueue() throws Exception {
 
 		assertThat(queue.getArguments().get("x-message-ttl")).isEqualTo(100L);
 		template.convertAndSend(queue.getName(), "message");
-		Properties props = rabbitAdmin.getQueueProperties("arguments");
-		if (props != null) {
-			int n = 0;
-			while (n++ < 200 && (Integer) props.get(RabbitAdmin.QUEUE_MESSAGE_COUNT) > 0) {
-				Thread.sleep(50);
-				props = rabbitAdmin.getQueueProperties("arguments");
-			}
-			assertThat((Integer) props.get(RabbitAdmin.QUEUE_MESSAGE_COUNT)).isEqualTo(0);
-		}
-
+		await().with().pollInterval(Duration.ofMillis(50))
+				.until(() -> rabbitAdmin.getQueueProperties("arguments")
+						.get(RabbitAdmin.QUEUE_MESSAGE_COUNT).equals(0));
 		connectionFactory.destroy();
 		RabbitAvailableCondition.getBrokerRunning().deleteQueues("arguments");
 	}

spring-rabbit/src/test/java/org/springframework/amqp/rabbit/connection/PublisherCallbackChannelTests.java

@@ -17,6 +17,7 @@
 package org.springframework.amqp.rabbit.connection;
 
 import static org.assertj.core.api.Assertions.assertThat;
+import static org.awaitility.Awaitility.await;
 import static org.mockito.ArgumentMatchers.any;
 import static org.mockito.BDDMockito.willAnswer;
 import static org.mockito.Mockito.mock;
@@ -130,12 +131,10 @@ void testNotCached() throws Exception {
 		assertThat(confirmLatch.await(10, TimeUnit.SECONDS)).isTrue();
 		assertThat(closedLatch.await(10, TimeUnit.SECONDS)).isTrue();
 		cacheProperties = cf.getCacheProperties();
-		int n = 0;
-		while (n++ < 100 && Integer.parseInt(cacheProperties.getProperty("idleChannelsNotTx")) < 2) {
-			Thread.sleep(100);
-			cacheProperties = cf.getCacheProperties();
-		}
-		assertThat(cacheProperties.getProperty("idleChannelsNotTx")).isEqualTo("2");
+		await().until(() -> {
+			Properties props = cf.getCacheProperties();
+			return Integer.parseInt(props.getProperty("idleChannelsNotTx")) == 2;
+		});
 	}
 
 }

spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/BatchingRabbitTemplateTests.java

@@ -17,6 +17,7 @@
 package org.springframework.amqp.rabbit.core;
 
 import static org.assertj.core.api.Assertions.assertThat;
+import static org.awaitility.Awaitility.await;
 import static org.mockito.ArgumentMatchers.any;
 import static org.mockito.ArgumentMatchers.anyString;
 import static org.mockito.Mockito.doNothing;
@@ -27,6 +28,7 @@
 
 import java.io.OutputStream;
 import java.lang.reflect.Method;
+import java.time.Duration;
 import java.util.ArrayList;
 import java.util.List;
 import java.util.concurrent.CountDownLatch;
@@ -69,6 +71,7 @@
 import org.springframework.amqp.support.postprocessor.ZipPostProcessor;
 import org.springframework.amqp.utils.test.TestUtils;
 import org.springframework.beans.DirectFieldAccessor;
+import org.springframework.lang.Nullable;
 import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler;
 import org.springframework.util.ReflectionUtils;
 import org.springframework.util.StopWatch;
@@ -615,15 +618,10 @@ public void testSimpleBatchDeflaterWithEncoding() throws Exception {
 		assertThat(new String(message.getBody())).isEqualTo("\u0000\u0000\u0000\u0003foo\u0000\u0000\u0000\u0003bar");
 	}
 
+	@Nullable
 	private Message receive(BatchingRabbitTemplate template) throws InterruptedException {
-		Message message = template.receive(ROUTE);
-		int n = 0;
-		while (n++ < 200 && message == null) {
-			Thread.sleep(50);
-			message = template.receive(ROUTE);
-		}
-		assertThat(message).isNotNull();
-		return message;
+		return await().with().pollInterval(Duration.ofMillis(50))
+				.until(() -> template.receive(ROUTE), msg -> msg != null);
 	}
 
 	@Test

spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/FixedReplyQueueDeadLetterTests.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2014-2019 the original author or authors.
+ * Copyright 2014-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,6 +17,7 @@
 package org.springframework.amqp.rabbit.core;
 
 import static org.assertj.core.api.Assertions.assertThat;
+import static org.awaitility.Awaitility.await;
 
 import java.net.MalformedURLException;
 import java.net.URISyntaxException;
@@ -99,13 +100,7 @@ void test() throws Exception {
 	void testQueueArgs1() throws MalformedURLException, URISyntaxException, InterruptedException {
 		Client client = new Client(brokerRunning.getAdminUri(), brokerRunning.getAdminUser(),
 				brokerRunning.getAdminPassword());
-		QueueInfo queue = client.getQueue("/", "all.args.1");
-		int n = 0;
-		while (n++ < 100 && queue == null) {
-			Thread.sleep(100);
-			queue = client.getQueue("/", "all.args.1");
-		}
-		assertThat(n).isLessThan(100);
+		QueueInfo queue = await().until(() -> client.getQueue("/", "all.args.1"), que -> que != null);
 		Map<String, Object> arguments = queue.getArguments();
 		assertThat(arguments.get("x-message-ttl")).isEqualTo(1000);
 		assertThat(arguments.get("x-expires")).isEqualTo(200_000);
@@ -124,13 +119,7 @@ void testQueueArgs1() throws MalformedURLException, URISyntaxException, Interrup
 	void testQueueArgs2() throws MalformedURLException, URISyntaxException, InterruptedException {
 		Client client = new Client(brokerRunning.getAdminUri(), brokerRunning.getAdminUser(),
 				brokerRunning.getAdminPassword());
-		QueueInfo queue = client.getQueue("/", "all.args.2");
-		int n = 0;
-		while (n++ < 100 && queue == null) {
-			Thread.sleep(100);
-			queue = client.getQueue("/", "all.args.1");
-		}
-		assertThat(n).isLessThan(100);
+		QueueInfo queue = await().until(() -> client.getQueue("/", "all.args.2"), que -> que != null);
 		Map<String, Object> arguments = queue.getArguments();
 		assertThat(arguments.get("x-message-ttl")).isEqualTo(1000);
 		assertThat(arguments.get("x-expires")).isEqualTo(200_000);
@@ -148,13 +137,7 @@ void testQueueArgs2() throws MalformedURLException, URISyntaxException, Interrup
 	void testQueueArgs3() throws MalformedURLException, URISyntaxException, InterruptedException {
 		Client client = new Client(brokerRunning.getAdminUri(), brokerRunning.getAdminUser(),
 				brokerRunning.getAdminPassword());
-		QueueInfo queue = client.getQueue("/", "all.args.3");
-		int n = 0;
-		while (n++ < 100 && queue == null) {
-			Thread.sleep(100);
-			queue = client.getQueue("/", "all.args.1");
-		}
-		assertThat(n).isLessThan(100);
+		QueueInfo queue = await().until(() -> client.getQueue("/", "all.args.3"), que -> que != null);
 		Map<String, Object> arguments = queue.getArguments();
 		assertThat(arguments.get("x-message-ttl")).isEqualTo(1000);
 		assertThat(arguments.get("x-expires")).isEqualTo(200_000);
@@ -178,13 +161,7 @@ void testQueueArgs3() throws MalformedURLException, URISyntaxException, Interrup
 	void testQuorumArgs() throws MalformedURLException, URISyntaxException, InterruptedException {
 		Client client = new Client(brokerRunning.getAdminUri(), brokerRunning.getAdminUser(),
 				brokerRunning.getAdminPassword());
-		QueueInfo queue = client.getQueue("/", "test.quorum");
-		int n = 0;
-		while (n++ < 100 && queue == null) {
-			Thread.sleep(100);
-			queue = client.getQueue("/", "test.quorum");
-		}
-		assertThat(n).isLessThan(100);
+		QueueInfo queue = await().until(() -> client.getQueue("/", "test.quorum"), que -> que != null);
 		Map<String, Object> arguments = queue.getArguments();
 		assertThat(arguments.get("x-queue-type")).isEqualTo("quorum");
 		assertThat(arguments.get("x-delivery-limit")).isEqualTo(10);

spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/RabbitAdminIntegrationTests.java

@@ -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,8 +18,10 @@
 
 import static org.assertj.core.api.Assertions.assertThat;
 import static org.assertj.core.api.Assertions.assertThatThrownBy;
+import static org.awaitility.Awaitility.await;
 
 import java.io.IOException;
+import java.time.Duration;
 import java.util.UUID;
 
 import org.junit.jupiter.api.AfterEach;
@@ -412,7 +414,6 @@ public void testDeclareDelayedExchange() throws Exception {
 		assertThat(System.currentTimeMillis() - t1).isGreaterThan(950L);
 
 		ExchangeInfo exchange2 = getExchange(exchangeName);
-		assertThat(exchange2).isNotNull();
 		assertThat(exchange2.getArguments().get("x-delayed-type")).isEqualTo(ExchangeTypes.DIRECT);
 		assertThat(exchange2.getType()).isEqualTo("x-delayed-message");
 
@@ -422,13 +423,8 @@ public void testDeclareDelayedExchange() throws Exception {
 
 	private ExchangeInfo getExchange(String exchangeName) throws Exception {
 		Client rabbitRestClient = new Client("http://localhost:15672/api/", "guest", "guest");
-		int n = 0;
-		ExchangeInfo exchange = rabbitRestClient.getExchange("/", exchangeName);
-		while (n++ < 100 && exchange == null) {
-			Thread.sleep(100);
-			exchange = rabbitRestClient.getExchange("/", exchangeName);
-		}
-		return exchange;
+		return await().pollDelay(Duration.ZERO)
+				.until(() -> rabbitRestClient.getExchange("/", exchangeName), exch -> exch != null);
 	}
 
 	/**