diff --git a/build.gradle b/build.gradle index 82ee9c39..b51164f9 100644 --- a/build.gradle +++ b/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' diff --git a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/AsyncRabbitTemplateTests.java b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/AsyncRabbitTemplateTests.java index 9427f460..75eddf50 100644 --- a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/AsyncRabbitTemplateTests.java +++ b/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 @@ package org.springframework.amqp.rabbit; 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 class AsyncRabbitTemplateTests { } 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 diff --git a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/annotation/ComplexTypeJsonIntegrationTests.java b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/annotation/ComplexTypeJsonIntegrationTests.java index a6551fdf..1c2a504f 100644 --- a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/annotation/ComplexTypeJsonIntegrationTests.java +++ b/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 class ComplexTypeJsonIntegrationTests { m.getMessageProperties().getHeaders().remove("__TypeId__"); return m; }); - Foo> foo = - this.rabbitTemplate.receiveAndConvert(new ParameterizedTypeReference>>() { }); - int n = 0; - while (n++ < 100 && foo == null) { - Thread.sleep(100); - foo = this.rabbitTemplate.receiveAndConvert(new ParameterizedTypeReference>>() { }); - } + Foo> foo = await().until( + () -> this.rabbitTemplate.receiveAndConvert(new ParameterizedTypeReference>>() { }), + msg -> msg != null); verifyFooBarBazQux(foo); } diff --git a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/annotation/ConsumerBatchingTests.java b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/annotation/ConsumerBatchingTests.java index d4925fb2..617fef6c 100644 --- a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/annotation/ConsumerBatchingTests.java +++ b/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 class ConsumerBatchingTests { 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 class ConsumerBatchingTests { .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") diff --git a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/annotation/EnableRabbitIntegrationTests.java b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/annotation/EnableRabbitIntegrationTests.java index 764457cc..0cdb2655 100644 --- a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/annotation/EnableRabbitIntegrationTests.java +++ b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/annotation/EnableRabbitIntegrationTests.java @@ -18,6 +18,7 @@ package org.springframework.amqp.rabbit.annotation; 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 class EnableRabbitIntegrationTests { 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 class EnableRabbitIntegrationTests { .timer(); } catch (@SuppressWarnings("unused") Exception e) { - Thread.sleep(100); + return null; } - } - assertThat(timer).isNotNull(); + }, tim -> tim != null); assertThat(timer.count()).isEqualTo(1L); } diff --git a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/config/QueueParserIntegrationTests.java b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/config/QueueParserIntegrationTests.java index 1139ec36..d00c8871 100644 --- a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/config/QueueParserIntegrationTests.java +++ b/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 final class QueueParserIntegrationTests { 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"); } diff --git a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/connection/PublisherCallbackChannelTests.java b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/connection/PublisherCallbackChannelTests.java index e04d1b8d..30e7c1d3 100644 --- a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/connection/PublisherCallbackChannelTests.java +++ b/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 @@ public class PublisherCallbackChannelTests { 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; + }); } } diff --git a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/BatchingRabbitTemplateTests.java b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/BatchingRabbitTemplateTests.java index d36b5152..5c16815d 100644 --- a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/BatchingRabbitTemplateTests.java +++ b/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 static org.mockito.Mockito.verify; 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.UnzipPostProcessor; 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 class BatchingRabbitTemplateTests { 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 diff --git a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/FixedReplyQueueDeadLetterTests.java b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/FixedReplyQueueDeadLetterTests.java index a02a7833..21237304 100644 --- a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/FixedReplyQueueDeadLetterTests.java +++ b/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 @@ public class FixedReplyQueueDeadLetterTests { 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 arguments = queue.getArguments(); assertThat(arguments.get("x-message-ttl")).isEqualTo(1000); assertThat(arguments.get("x-expires")).isEqualTo(200_000); @@ -124,13 +119,7 @@ public class FixedReplyQueueDeadLetterTests { 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 arguments = queue.getArguments(); assertThat(arguments.get("x-message-ttl")).isEqualTo(1000); assertThat(arguments.get("x-expires")).isEqualTo(200_000); @@ -148,13 +137,7 @@ public class FixedReplyQueueDeadLetterTests { 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 arguments = queue.getArguments(); assertThat(arguments.get("x-message-ttl")).isEqualTo(1000); assertThat(arguments.get("x-expires")).isEqualTo(200_000); @@ -178,13 +161,7 @@ public class FixedReplyQueueDeadLetterTests { 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 arguments = queue.getArguments(); assertThat(arguments.get("x-queue-type")).isEqualTo("quorum"); assertThat(arguments.get("x-delivery-limit")).isEqualTo(10); diff --git a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/RabbitAdminIntegrationTests.java b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/RabbitAdminIntegrationTests.java index c316d5b5..1c504a7c 100644 --- a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/RabbitAdminIntegrationTests.java +++ b/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 @@ package org.springframework.amqp.rabbit.core; 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 class RabbitAdminIntegrationTests { 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 class RabbitAdminIntegrationTests { 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); } /** diff --git a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/RabbitAdminTests.java b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/RabbitAdminTests.java index 31f294b7..cd55b385 100644 --- a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/RabbitAdminTests.java +++ b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/RabbitAdminTests.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. @@ -20,6 +20,7 @@ import static org.assertj.core.api.Assertions.assertThat; import static org.assertj.core.api.Assertions.assertThatIllegalArgumentException; import static org.assertj.core.api.Assertions.assertThatThrownBy; 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.anyBoolean; import static org.mockito.ArgumentMatchers.anyString; @@ -149,19 +150,11 @@ public class RabbitAdminTests { try { rabbitAdmin.declareQueue(new Queue(queueName)); new RabbitTemplate(connectionFactory).convertAndSend(queueName, "foo"); - int n = 0; - while (n++ < 100 && messageCount(rabbitAdmin, queueName) == 0) { - Thread.sleep(100); - } - assertThat(n < 100).as("Message count = 0").isTrue(); + await("Message count = 0").until(() -> messageCount(rabbitAdmin, queueName) > 0); Channel channel = connectionFactory.createConnection().createChannel(false); DefaultConsumer consumer = new DefaultConsumer(channel); channel.basicConsume(queueName, true, consumer); - n = 0; - while (n++ < 100 && messageCount(rabbitAdmin, queueName) > 0) { - Thread.sleep(100); - } - assertThat(n < 100).as("Message count > 0").isTrue(); + await("Message count > 0").until(() -> messageCount(rabbitAdmin, queueName) == 0); Properties props = rabbitAdmin.getQueueProperties(queueName); assertThat(props.get(RabbitAdmin.QUEUE_CONSUMER_COUNT)).isNotNull(); assertThat(props.get(RabbitAdmin.QUEUE_CONSUMER_COUNT)).isEqualTo(1); @@ -382,25 +375,15 @@ public class RabbitAdminTests { AnonymousQueue queue = new AnonymousQueue(); admin.declareQueue(queue); Client client = new Client("http://guest:guest@localhost:15672/api"); - QueueInfo info = client.getQueue("?", queue.getName()); - int n = 0; - while (n++ < 100 && info == null) { - Thread.sleep(100); - info = client.getQueue("/", queue.getName()); - } - assertThat(info).isNotNull(); + AnonymousQueue queue1 = queue; + QueueInfo info = await().until(() -> client.getQueue("/", queue1.getName()), inf -> inf != null); assertThat(info.getArguments().get(Queue.X_QUEUE_MASTER_LOCATOR)).isEqualTo("client-local"); queue = new AnonymousQueue(); queue.setMasterLocator(null); admin.declareQueue(queue); - info = client.getQueue("?", queue.getName()); - n = 0; - while (n++ < 100 && info == null) { - Thread.sleep(100); - info = client.getQueue("/", queue.getName()); - } - assertThat(info).isNotNull(); + AnonymousQueue queue2 = queue; + info = await().until(() -> client.getQueue("/", queue2.getName()), inf -> inf != null); assertThat(info.getArguments().get(Queue.X_QUEUE_MASTER_LOCATOR)).isNull(); cf.destroy(); } diff --git a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/RabbitBindingIntegrationTests.java b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/RabbitBindingIntegrationTests.java index b9c7cdc3..317d06f2 100644 --- a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/RabbitBindingIntegrationTests.java +++ b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/RabbitBindingIntegrationTests.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,6 +17,9 @@ package org.springframework.amqp.rabbit.core; import static org.assertj.core.api.Assertions.assertThat; +import static org.awaitility.Awaitility.await; + +import java.time.Duration; import org.junit.jupiter.api.AfterEach; import org.junit.jupiter.api.BeforeEach; @@ -268,18 +271,7 @@ public class RabbitBindingIntegrationTests { new ActiveObjectCounter(), AcknowledgeMode.AUTO, true, 1, QUEUE.getName()); consumer.start(); // wait for consumeOk... - int n = 0; - while (n++ < 100) { - if (consumer.getConsumerTags().size() == 0) { - try { - Thread.sleep(100); - } - catch (@SuppressWarnings("unused") InterruptedException e) { - Thread.currentThread().interrupt(); - break; - } - } - } + await().with().pollDelay(Duration.ZERO).until(() -> consumer.getConsumerTags().size() > 0); return consumer; } diff --git a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/RabbitRestApiTests.java b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/RabbitRestApiTests.java index 54fff80a..4a574ad8 100644 --- a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/RabbitRestApiTests.java +++ b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/RabbitRestApiTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2015-2019 the original author or authors. + * Copyright 2015-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; @@ -179,12 +180,8 @@ public class RabbitRestApiTests { admin.declareQueue(queue2); Channel channel = this.connectionFactory.createConnection().createChannel(false); String consumer = channel.basicConsume(queue1.getName(), false, "", false, true, null, new DefaultConsumer(channel)); - QueueInfo qi = this.rabbitRestClient.getQueue("/", queue1.getName()); - int n = 0; - while (n++ < 100 && (qi.getExclusiveConsumerTag() == null || qi.getExclusiveConsumerTag().equals(""))) { - Thread.sleep(100); - qi = this.rabbitRestClient.getQueue("/", queue1.getName()); - } + QueueInfo qi = await().until(() -> this.rabbitRestClient.getQueue("/", queue1.getName()), + info -> info.getExclusiveConsumerTag() != null && !"".equals(info.getExclusiveConsumerTag())); QueueInfo queueOut = this.rabbitRestClient.getQueue("/", queue1.getName()); assertThat(queueOut.isDurable()).isFalse(); assertThat(queueOut.isExclusive()).isFalse(); diff --git a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/RabbitTemplateIntegrationTests.java b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/RabbitTemplateIntegrationTests.java index 2523845a..a202064b 100644 --- a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/RabbitTemplateIntegrationTests.java +++ b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/RabbitTemplateIntegrationTests.java @@ -19,6 +19,7 @@ package org.springframework.amqp.rabbit.core; import static org.assertj.core.api.Assertions.assertThat; import static org.assertj.core.api.Assertions.assertThatThrownBy; 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.anyBoolean; import static org.mockito.ArgumentMatchers.anyString; @@ -205,11 +206,7 @@ public class RabbitTemplateIntegrationTests { this.template.setChannelTransacted(true); this.template.convertAndSend(ROUTE, "foo"); this.template.convertAndSend(UUID.randomUUID().toString(), ROUTE, "xxx"); // force channel close - int n = 0; - while (n++ < 100 && channel.isOpen()) { - Thread.sleep(100); - } - assertThat(channel.isOpen()).isFalse(); + await().until(() -> !channel.isOpen()); try { this.template.convertAndSend(ROUTE, "bar"); fail("Expected Exception"); @@ -265,13 +262,7 @@ public class RabbitTemplateIntegrationTests { @Test public void testReceiveNonBlocking() throws Exception { this.template.convertAndSend(ROUTE, "nonblock"); - int n = 0; - String out = (String) this.template.receiveAndConvert(ROUTE); - while (n++ < 100 && out == null) { - Thread.sleep(100); - out = (String) this.template.receiveAndConvert(ROUTE); - } - assertThat(out).isNotNull(); + String out = await().until(() -> (String) this.template.receiveAndConvert(ROUTE), str -> str != null); assertThat(out).isEqualTo("nonblock"); assertThat(this.template.receive(ROUTE)).isNull(); } @@ -1095,13 +1086,7 @@ public class RabbitTemplateIntegrationTests { this.template.convertAndSend(ROUTE, "test"); template.setReceiveTimeout(timeout); - boolean received = receiveAndReply(); - int n = 0; - while (timeout == 0 && !received && n++ < 100) { - Thread.sleep(100); - received = receiveAndReply(); - } - assertThat(received).isTrue(); + boolean received = await().until(() -> receiveAndReply(), b -> b); Message receive = this.template.receive(); assertThat(receive).isNotNull(); diff --git a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/RabbitTemplatePublisherCallbacksIntegrationTests.java b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/RabbitTemplatePublisherCallbacksIntegrationTests.java index 60371132..45ae136b 100644 --- a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/RabbitTemplatePublisherCallbacksIntegrationTests.java +++ b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/RabbitTemplatePublisherCallbacksIntegrationTests.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,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.anyBoolean; import static org.mockito.ArgumentMatchers.anyString; @@ -199,13 +200,9 @@ public class RabbitTemplatePublisherCallbacksIntegrationTests { assertThat(confirmCorrelation.get().getId()).isEqualTo("abc"); assertThat(templateWithConfirmsEnabled.getUnconfirmed(-1)).isNull(); this.templateWithConfirmsEnabled.execute(channel -> { - Map listenerMap = TestUtils.getPropertyValue(((ChannelProxy) channel).getTargetChannel(), "listenerForSeq", - Map.class); - int n = 0; - while (n++ < 100 && listenerMap.size() > 0) { - Thread.sleep(100); - } - assertThat(listenerMap).hasSize(0); + Map listenerMap = TestUtils.getPropertyValue(((ChannelProxy) channel).getTargetChannel(), + "listenerForSeq", Map.class); + await().until(() -> listenerMap.size() == 0); return null; }); @@ -415,11 +412,7 @@ public class RabbitTemplatePublisherCallbacksIntegrationTests { exec.shutdown(); assertThat(exec.awaitTermination(10, TimeUnit.SECONDS)).isTrue(); ccf.destroy(); - int n = 0; - while (n++ < 100 && pendingConfirms.size() > 0) { - Thread.sleep(100); - } - assertThat(pendingConfirms).hasSize(0); + await().until(() -> pendingConfirms.size() == 0); } @Test @@ -827,12 +820,8 @@ public class RabbitTemplatePublisherCallbacksIntegrationTests { assertThat(latch.await(10, TimeUnit.SECONDS)).isTrue(); - int n = 0; - while (n++ < 100 && TestUtils.getPropertyValue(channel, "pendingConfirms", Map.class).size() > 0) { - Thread.sleep(100); - } - assertThat(TestUtils.getPropertyValue(channel, "pendingConfirms", Map.class)).hasSize(0); - + Map pending = TestUtils.getPropertyValue(channel, "pendingConfirms", Map.class); + await().until(() -> pending.size() == 0); } @Test diff --git a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/ContainerInitializationTests.java b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/ContainerInitializationTests.java index 1002a98a..95dd3e2e 100644 --- a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/ContainerInitializationTests.java +++ b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/ContainerInitializationTests.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,7 +18,9 @@ package org.springframework.amqp.rabbit.listener; 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.concurrent.CountDownLatch; import java.util.concurrent.TimeUnit; @@ -89,11 +91,7 @@ public class ContainerInitializationTests { latches[2].countDown(); // let container thread continue to enable restart assertThat(latches[1].await(20, TimeUnit.SECONDS)).isTrue(); SimpleMessageListenerContainer container = context.getBean(SimpleMessageListenerContainer.class); - int n = 0; - while (n++ < 200 && container.isRunning()) { - Thread.sleep(100); - } - assertThat(container.isRunning()).isFalse(); + await().atMost(Duration.ofSeconds(20)).until(() -> !container.isRunning()); context.close(); } @@ -108,11 +106,7 @@ public class ContainerInitializationTests { latches[2].countDown(); // let container thread continue to enable restart assertThat(latches[1].await(20, TimeUnit.SECONDS)).isTrue(); SimpleMessageListenerContainer container = context.getBean(SimpleMessageListenerContainer.class); - int n = 0; - while (n++ < 200 && container.isRunning()) { - Thread.sleep(100); - } - assertThat(container.isRunning()).isFalse(); + await().atMost(Duration.ofSeconds(20)).until(() -> !container.isRunning()); context.close(); } diff --git a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/DirectMessageListenerContainerIntegrationTests.java b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/DirectMessageListenerContainerIntegrationTests.java index bc687625..d28ab00e 100644 --- a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/DirectMessageListenerContainerIntegrationTests.java +++ b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/DirectMessageListenerContainerIntegrationTests.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.listener; import static org.assertj.core.api.Assertions.assertThat; +import static org.awaitility.Awaitility.await; import static org.mockito.ArgumentMatchers.anyBoolean; import static org.mockito.ArgumentMatchers.anyMap; import static org.mockito.ArgumentMatchers.anyString; @@ -26,9 +27,9 @@ import static org.mockito.BDDMockito.willAnswer; import static org.mockito.Mockito.mock; import static org.mockito.Mockito.verify; +import java.time.Duration; import java.util.ArrayList; import java.util.List; -import java.util.Properties; import java.util.concurrent.CountDownLatch; import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; @@ -36,7 +37,6 @@ import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicReference; import org.aopalliance.intercept.MethodInterceptor; -import org.apache.commons.logging.LogFactory; import org.junit.jupiter.api.AfterAll; import org.junit.jupiter.api.BeforeAll; import org.junit.jupiter.api.BeforeEach; @@ -431,11 +431,7 @@ public class DirectMessageListenerContainerIntegrationTests { // Since backOff exhausting makes listenerContainer as invalid (calls stop()), // it is enough to check the listenerContainer activity - int n = 0; - while (container.isActive() && n++ < 100) { - Thread.sleep(100); - } - assertThat(container.isActive()).isFalse(); + await().until(() -> !container.isActive()); } @Test @@ -636,11 +632,7 @@ public class DirectMessageListenerContainerIntegrationTests { container.afterPropertiesSet(); container.start(); - int n = 0; - while (n++ < 100 && container.isRunning()) { - Thread.sleep(100); - } - assertThat(container.isRunning()).isFalse(); + await().until(() -> !container.isActive()); } @Test @@ -663,11 +655,7 @@ public class DirectMessageListenerContainerIntegrationTests { assertThat(latch.await(10, TimeUnit.SECONDS)).isTrue(); cf.onApplicationEvent(new ContextClosedEvent(context)); cf.destroy(); - int n = 0; - while (n++ < 100 && container.isRunning()) { - Thread.sleep(100); - } - assertThat(container.isRunning()).isFalse(); + await().until(() -> !container.isActive()); } @Test @@ -692,34 +680,24 @@ public class DirectMessageListenerContainerIntegrationTests { } private boolean consumersOnQueue(String queue, int expected) throws Exception { - int n = 0; - Properties queueProperties = admin.getQueueProperties(queue); - LogFactory.getLog(getClass()).debug(queue + " waiting for " + expected + " : " + queueProperties); - while (n++ < 600 - && (queueProperties == null || !queueProperties.get(RabbitAdmin.QUEUE_CONSUMER_COUNT).equals(expected))) { - Thread.sleep(100); - queueProperties = admin.getQueueProperties(queue); - LogFactory.getLog(getClass()).debug(queue + " waiting for " + expected + " : " + queueProperties); - } - return queueProperties.get(RabbitAdmin.QUEUE_CONSUMER_COUNT).equals(expected); + await().with().pollDelay(Duration.ZERO).atMost(Duration.ofSeconds(60)) + .until(() -> admin.getQueueProperties(queue), + props -> props != null && props.get(RabbitAdmin.QUEUE_CONSUMER_COUNT).equals(expected)); + return true; } private boolean activeConsumerCount(AbstractMessageListenerContainer container, int expected) throws Exception { - int n = 0; List consumers = TestUtils.getPropertyValue(container, "consumers", List.class); - while (n++ < 600 && consumers.size() != expected) { - Thread.sleep(100); - } - return consumers.size() == expected; + await().with().pollDelay(Duration.ZERO).atMost(Duration.ofSeconds(60)) + .until(() -> consumers.size() == expected); + return true; } private boolean restartConsumerCount(AbstractMessageListenerContainer container, int expected) throws Exception { - int n = 0; List consumers = TestUtils.getPropertyValue(container, "consumersToRestart", List.class); - while (n++ < 600 && consumers.size() != expected) { - Thread.sleep(100); - } - return consumers.size() == expected; + await().with().pollDelay(Duration.ZERO).atMost(Duration.ofSeconds(60)) + .until(() -> consumers.size() == expected); + return true; } public class Tag implements ConsumerTagStrategy { diff --git a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/DirectReplyToMessageListenerContainerTests.java b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/DirectReplyToMessageListenerContainerTests.java index 1339bd07..20bd2e6b 100644 --- a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/DirectReplyToMessageListenerContainerTests.java +++ b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/DirectReplyToMessageListenerContainerTests.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,7 +17,9 @@ package org.springframework.amqp.rabbit.listener; import static org.assertj.core.api.Assertions.assertThat; +import static org.awaitility.Awaitility.await; +import java.time.Duration; import java.util.Map; import java.util.concurrent.CountDownLatch; import java.util.concurrent.TimeUnit; @@ -83,12 +85,9 @@ public class DirectReplyToMessageListenerContainerTests { BasicProperties props = new BasicProperties().builder().replyTo(Address.AMQ_RABBITMQ_REPLY_TO).build(); channel1.getChannel().basicPublish("", TEST_RELEASE_CONSUMER_Q, props, "foo".getBytes()); Channel replyChannel = connectionFactory.createConnection().createChannel(false); - GetResponse request = replyChannel.basicGet(TEST_RELEASE_CONSUMER_Q, true); - int n = 0; - while (n++ < 100 && request == null) { - Thread.sleep(100); - request = replyChannel.basicGet(TEST_RELEASE_CONSUMER_Q, true); - } + GetResponse request = await() + .pollDelay(Duration.ZERO) + .until(() -> replyChannel.basicGet(TEST_RELEASE_CONSUMER_Q, true), req -> req != null); assertThat(request).isNotNull(); replyChannel.basicPublish("", request.getProps().getReplyTo(), new BasicProperties(), "bar".getBytes()); replyChannel.close(); diff --git a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/ListenFromAutoDeleteQueueTests.java b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/ListenFromAutoDeleteQueueTests.java index 35d7d130..5aa8cc8a 100644 --- a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/ListenFromAutoDeleteQueueTests.java +++ b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/ListenFromAutoDeleteQueueTests.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.listener; import static org.assertj.core.api.Assertions.assertThat; +import static org.awaitility.Awaitility.await; import static org.mockito.Mockito.atLeastOnce; import static org.mockito.Mockito.doThrow; import static org.mockito.Mockito.mock; @@ -115,12 +116,7 @@ public class ListenFromAutoDeleteQueueTests { listenerContainer.stop(); RabbitAdmin admin = spy(TestUtils.getPropertyValue(listenerContainer, "amqpAdmin", RabbitAdmin.class)); new DirectFieldAccessor(listenerContainer).setPropertyValue("amqpAdmin", admin); - int n = 0; - while (admin.getQueueProperties(this.expiringQueue.getName()) != null && n < 100) { - Thread.sleep(100); - n++; - } - assertThat(n < 100).isTrue(); + await().until(() -> admin.getQueueProperties(this.expiringQueue.getName()) == null); listenerContainer.start(); template.convertAndSend(this.expiringQueue.getName(), "foo"); assertThat(queue.poll(10, TimeUnit.SECONDS)).isNotNull(); diff --git a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/MessageListenerContainerErrorHandlerIntegrationTests.java b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/MessageListenerContainerErrorHandlerIntegrationTests.java index 6896fc81..5a994997 100644 --- a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/MessageListenerContainerErrorHandlerIntegrationTests.java +++ b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/MessageListenerContainerErrorHandlerIntegrationTests.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,6 +18,7 @@ package org.springframework.amqp.rabbit.listener; 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.ArgumentMatchers.any; import static org.mockito.ArgumentMatchers.contains; import static org.mockito.Mockito.doAnswer; @@ -247,16 +248,11 @@ public class MessageListenerContainerErrorHandlerIntegrationTests { .build(); template.send("", testQueueName, message); - Message rejected = template.receive(dlq.getName()); // can't use timed receive, queue will be deleted - int n = 0; - while (n++ < 100 && rejected == null) { - Thread.sleep(100); - rejected = template.receive(dlq.getName()); - } - assertThat(n < 100).as("Message did not arrive in DLQ").isTrue(); + // can't use timed receive, queue will be deleted + Message rejected = await("Message did not arrive in DLQ") + .until(() -> template.receive(dlq.getName()), msg -> msg != null); assertThat(new String(rejected.getBody())).isEqualTo("foo"); - // Verify that the exception strategy has access to the message final AtomicReference failed = new AtomicReference(); ConditionalRejectingErrorHandler eh = new ConditionalRejectingErrorHandler(t -> { @@ -270,13 +266,8 @@ public class MessageListenerContainerErrorHandlerIntegrationTests { template.send("", testQueueName, message); - rejected = template.receive(dlq.getName()); - n = 0; - while (n++ < 100 && rejected == null) { - Thread.sleep(100); - rejected = template.receive(dlq.getName()); - } - assertThat(n < 100).as("Message did not arrive in DLQ").isTrue(); + rejected = await("Message did not arrive in DLQ") + .until(() -> template.receive(dlq.getName()), msg -> msg != null); assertThat(new String(rejected.getBody())).isEqualTo("foo"); assertThat(failed.get()).isNotNull(); diff --git a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/MessageListenerContainerLifecycleIntegrationTests.java b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/MessageListenerContainerLifecycleIntegrationTests.java index f5ad5cb0..e3ec0683 100755 --- a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/MessageListenerContainerLifecycleIntegrationTests.java +++ b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/MessageListenerContainerLifecycleIntegrationTests.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,6 +18,7 @@ package org.springframework.amqp.rabbit.listener; 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.Mockito.doAnswer; import static org.mockito.Mockito.spy; import static org.mockito.Mockito.when; @@ -251,11 +252,7 @@ public class MessageListenerContainerLifecycleIntegrationTests { assertThat(container.getActiveConsumerCount()).isEqualTo(concurrentConsumers); container.stop(); - int n = 0; - while (n++ < 100 && container.getActiveConsumerCount() > 0) { - Thread.sleep(100); - } - assertThat(container.getActiveConsumerCount()).isEqualTo(0); + await().until(() -> container.getActiveConsumerCount() == 0); if (!transactional) { int messagesReceivedAfterStop = listener.getCount(); @@ -308,11 +305,7 @@ public class MessageListenerContainerLifecycleIntegrationTests { container.shutdown(); } - int n = 0; - while (n++ < 100 && container.getActiveConsumerCount() > 0) { - Thread.sleep(100); - } - assertThat(container.getActiveConsumerCount()).isEqualTo(0); + await().until(() -> container.getActiveConsumerCount() == 0); assertThat(template.receiveAndConvert(queue.getName())).isNull(); ((DisposableBean) template.getConnectionFactory()).destroy(); @@ -369,40 +362,32 @@ public class MessageListenerContainerLifecycleIntegrationTests { @SuppressWarnings("unchecked") Set consumers = (Set) TestUtils .getPropertyValue(container, "consumers"); - int n = 0; - while (n++ < 100) { - if (consumers.size() > 0) { - if (TestUtils.getPropertyValue(consumers.iterator().next(), "queue", BlockingQueue.class) - .size() > 3) { - prefetched.countDown(); - break; - } + await().until(() -> { + if (consumers.size() > 0 + && TestUtils.getPropertyValue(consumers.iterator().next(), "queue", BlockingQueue.class).size() > 3) { + prefetched.countDown(); + return true; } - Thread.sleep(100); - } + else { + return false; + } + }); Executors.newSingleThreadExecutor().execute(() -> container.stop()); - n = 0; - while (container.isActive() && n++ < 100) { - Thread.sleep(100); - } - assertThat(n < 100).isTrue(); - + await().until(() -> !container.isActive()); awaitStop.countDown(); - assertThat(awaitConsumeFirst.await(10, TimeUnit.SECONDS)).as("awaitConsumeFirst.count=" + awaitConsumeFirst.getCount()).isTrue(); - n = 0; + assertThat(awaitConsumeFirst.await(10, TimeUnit.SECONDS)) + .as("awaitConsumeFirst.count=" + awaitConsumeFirst.getCount()).isTrue(); DirectFieldAccessor dfa = new DirectFieldAccessor(container); - while (dfa.getPropertyValue("consumers") != null && n++ < 100) { - Thread.sleep(100); - } - assertThat(n < 100).isTrue(); + await().until(() -> dfa.getPropertyValue("consumers") == null); // make sure we stopped receiving after the prefetch was consumed assertThat(received.get()).isEqualTo(5); assertThat(awaitStart2.getCount()).isEqualTo(1); container.start(); assertThat(awaitStart2.await(10, TimeUnit.SECONDS)).isTrue(); - assertThat(awaitConsumeSecond.await(10, TimeUnit.SECONDS)).as("awaitConsumeSecond.count=" + awaitConsumeSecond.getCount()).isTrue(); + assertThat(awaitConsumeSecond.await(10, TimeUnit.SECONDS)) + .as("awaitConsumeSecond.count=" + awaitConsumeSecond.getCount()).isTrue(); container.stop(); ((DisposableBean) template.getConnectionFactory()).destroy(); } @@ -464,11 +449,7 @@ public class MessageListenerContainerLifecycleIntegrationTests { ActiveObjectCounter counter = TestUtils.getPropertyValue(container, "cancellationLock", ActiveObjectCounter.class); assertThat(counter.getCount() > 0).isTrue(); - int n = 0; - while (counter.getCount() > 0 && n++ < 10) { - Thread.sleep(500); - } - assertThat(n < 10).isTrue(); + await().until(() -> counter.getCount() == 0); ((DisposableBean) template.getConnectionFactory()).destroy(); } diff --git a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/MessageListenerRecoveryCachingConnectionIntegrationTests.java b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/MessageListenerRecoveryCachingConnectionIntegrationTests.java index 8e2669d4..8cf6ffff 100644 --- a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/MessageListenerRecoveryCachingConnectionIntegrationTests.java +++ b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/MessageListenerRecoveryCachingConnectionIntegrationTests.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,7 +18,9 @@ package org.springframework.amqp.rabbit.listener; import static org.assertj.core.api.Assertions.assertThat; import static org.assertj.core.api.Assertions.assertThatExceptionOfType; +import static org.awaitility.Awaitility.with; +import java.time.Duration; import java.util.Collections; import java.util.HashSet; import java.util.Properties; @@ -236,11 +238,8 @@ public class MessageListenerRecoveryCachingConnectionIntegrationTests { CountDownLatch latch = new CountDownLatch(messageCount); ConnectionFactory connectionFactory2 = createConnectionFactory(); container = createContainer(queue.getName(), new AbortChannelListener(latch), connectionFactory2); - int n = 0; - while (n++ < 100 && container.getActiveConsumerCount() != concurrentConsumers) { - Thread.sleep(50L); - } - assertThat(container.getActiveConsumerCount()).isEqualTo(concurrentConsumers); + with().pollInterval(Duration.ofMillis(50)) + .await().until(() -> container.getActiveConsumerCount() == this.concurrentConsumers); for (int i = 0; i < messageCount; i++) { template.convertAndSend(queue.getName(), i + "foo"); diff --git a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/SimpleMessageListenerContainerIntegration2Tests.java b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/SimpleMessageListenerContainerIntegration2Tests.java index 349b8eb2..de7441a0 100644 --- a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/SimpleMessageListenerContainerIntegration2Tests.java +++ b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/SimpleMessageListenerContainerIntegration2Tests.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,6 +17,8 @@ package org.springframework.amqp.rabbit.listener; import static org.assertj.core.api.Assertions.assertThat; +import static org.awaitility.Awaitility.await; +import static org.awaitility.Awaitility.with; import static org.mockito.ArgumentMatchers.any; import static org.mockito.ArgumentMatchers.anyBoolean; import static org.mockito.Mockito.atLeastOnce; @@ -27,6 +29,7 @@ import static org.mockito.Mockito.when; import java.io.IOException; import java.io.Serializable; +import java.time.Duration; import java.util.ArrayList; import java.util.List; import java.util.NoSuchElementException; @@ -249,38 +252,26 @@ public class SimpleMessageListenerContainerIntegration2Tests { } waited = latch.await(10, TimeUnit.SECONDS); assertThat(waited).as("Timed out waiting for message").isTrue(); - BlockingQueueConsumer newConsumer = consumer; - int n = 0; - while (n++ < 100) { + BlockingQueueConsumer newConsumer = await("Failed to restart consumer").until(() -> { try { - newConsumer = (BlockingQueueConsumer) consumers.iterator().next(); - if (newConsumer != consumer) { - break; - } + return (BlockingQueueConsumer) consumers.iterator().next(); } catch (NoSuchElementException e) { // race; hasNext() won't help + return null; } - Thread.sleep(100); - } - assertThat(n < 100).as("Failed to restart consumer").isTrue(); + }, newCon -> newCon != consumer); Set missingQueues = TestUtils.getPropertyValue(newConsumer, "missingQueues", Set.class); - n = 0; - while (n++ < 100 && missingQueues.size() == 0) { - Thread.sleep(200); - } - assertThat(n < 100).as("Failed to detect missing queue").isTrue(); + with().pollInterval(Duration.ofMillis(200)).await("Failed to detect missing queue") + .atMost(Duration.ofSeconds(20)) + .until(() -> missingQueues.size() > 0); assertThat(eventRef.get().getThrowable()).isInstanceOf(ConsumerCancelledException.class); assertThat(eventRef.get().isFatal()).isFalse(); DirectFieldAccessor dfa = new DirectFieldAccessor(newConsumer); dfa.setPropertyValue("lastRetryDeclaration", 0); dfa.setPropertyValue("retryDeclarationInterval", 100); admin.declareQueue(queue1); - n = 0; - while (n++ < 100 && missingQueues.size() > 0) { - Thread.sleep(100); - } - assertThat(n < 100).as("Failed to redeclare missing queue").isTrue(); + await("Failed to redeclare missing queue").until(() -> missingQueues.size() == 0); latch = new CountDownLatch(20); container.setMessageListener(new MessageListenerAdapter(new PojoListener(latch))); for (int i = 0; i < 10; i++) { @@ -636,11 +627,7 @@ public class SimpleMessageListenerContainerIntegration2Tests { this.container.start(); this.template.convertAndSend(this.queue.getName(), "foo"); assertThat(latch.await(10, TimeUnit.SECONDS)).isTrue(); - int n = 0; - while (n++ < 100 && this.container.isRunning()) { - Thread.sleep(100); - } - assertThat(this.container.isRunning()).isFalse(); + await().until(() -> !this.container.isRunning()); } @Test @@ -673,10 +660,7 @@ public class SimpleMessageListenerContainerIntegration2Tests { Log logger = spy(TestUtils.getPropertyValue(container, "logger", Log.class)); new DirectFieldAccessor(container).setPropertyValue("logger", logger); this.template.convertAndSend(queue.getName(), "foo"); - int n = 0; - while (n++ < 100 && this.container.isRunning()) { - Thread.sleep(100); - } + await().until(() -> !this.container.isRunning()); ArgumentCaptor captor = ArgumentCaptor.forClass(String.class); verify(logger).error(captor.capture()); assertThat(captor.getValue()).contains("Stopping container from aborted consumer"); diff --git a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/SimpleMessageListenerContainerLongTests.java b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/SimpleMessageListenerContainerLongTests.java index d32a1ddd..81faf4f8 100644 --- a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/SimpleMessageListenerContainerLongTests.java +++ b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/SimpleMessageListenerContainerLongTests.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,6 +18,7 @@ package org.springframework.amqp.rabbit.listener; 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.Set; @@ -131,11 +132,7 @@ public class SimpleMessageListenerContainerLongTests { } } - int n = 0; - while (n++ < 100 && container.getActiveConsumerCount() != 2) { - Thread.sleep(100); - } - assertThat(container.getActiveConsumerCount()).isEqualTo(2); + await().until(() -> container.getActiveConsumerCount() == 2); container.stop(); for (int i = 0; i < 20; i++) { admin.deleteQueue("testAddQueuesAndStartInCycle" + i); diff --git a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/SimpleMessageListenerContainerTests.java b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/SimpleMessageListenerContainerTests.java index 7381cd68..4350780e 100644 --- a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/SimpleMessageListenerContainerTests.java +++ b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/SimpleMessageListenerContainerTests.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. @@ -19,6 +19,8 @@ package org.springframework.amqp.rabbit.listener; import static org.assertj.core.api.Assertions.assertThat; import static org.assertj.core.api.Assertions.assertThatIllegalStateException; 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.anyBoolean; import static org.mockito.ArgumentMatchers.anyLong; @@ -38,6 +40,7 @@ import static org.mockito.Mockito.when; import java.io.IOException; import java.net.URL; import java.net.URLClassLoader; +import java.time.Duration; import java.util.ArrayList; import java.util.Collection; import java.util.Collections; @@ -528,11 +531,7 @@ public class SimpleMessageListenerContainerTests { // Since backOff exhausting makes listenerContainer as invalid (calls stop()), // it is enough to check the listenerContainer activity - int n = 0; - while (container.isActive() && n++ < 100) { - Thread.sleep(100); - } - assertThat(n).isLessThanOrEqualTo(100); + await().until(() -> !container.isActive()); } @Test @@ -664,16 +663,10 @@ public class SimpleMessageListenerContainerTests { } private void waitForConsumersToStop(Set consumers) throws Exception { - int n = 0; - boolean stillUp = true; - while (stillUp && n++ < 1000) { - stillUp = false; - for (Object consumer : consumers) { - stillUp |= TestUtils.getPropertyValue(consumer, "consumer") != null; - } - Thread.sleep(10); - } - assertThat(stillUp).isFalse(); + with().pollInterval(Duration.ofMillis(10)).atMost(Duration.ofSeconds(10)) + .until(() -> consumers.stream() + .map(consumer -> TestUtils.getPropertyValue(consumer, "consumer")) + .allMatch(c -> c == null)); } @SuppressWarnings("serial") diff --git a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/log4j2/AmqpAppenderTests.java b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/log4j2/AmqpAppenderTests.java index 9c2321b2..6f4afb57 100644 --- a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/log4j2/AmqpAppenderTests.java +++ b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/log4j2/AmqpAppenderTests.java @@ -17,6 +17,7 @@ package org.springframework.amqp.rabbit.log4j2; 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.anyBoolean; import static org.mockito.ArgumentMatchers.anyString; @@ -211,11 +212,7 @@ public class AmqpAppenderTests { assertThat(events.getClass()).isEqualTo(LinkedBlockingQueue.class); BlockingQueue queue = (BlockingQueue) events; - int n = 0; - while (n++ < 100 && queue.size() > 0) { - Thread.sleep(100); - } - assertThat(queue).hasSize(0); + await().until(() -> queue.size() == 0); } @Test diff --git a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/retry/MissingIdRetryTests.java b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/retry/MissingIdRetryTests.java index a8433f0d..dd783c13 100644 --- a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/retry/MissingIdRetryTests.java +++ b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/retry/MissingIdRetryTests.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,6 +17,7 @@ package org.springframework.amqp.rabbit.retry; import static org.assertj.core.api.Assertions.assertThat; +import static org.awaitility.Awaitility.await; import static org.mockito.ArgumentMatchers.any; import static org.mockito.Mockito.atLeast; import static org.mockito.Mockito.atMost; @@ -112,13 +113,9 @@ public class MissingIdRetryTests { try { assertThat(latch.await(30, TimeUnit.SECONDS)).isTrue(); Map map = (Map) new DirectFieldAccessor(cache).getPropertyValue("map"); - int n = 0; - while (n++ < 100 && map.size() != 0) { - Thread.sleep(100); - } + await().until(() -> map.size() == 0); verify(cache, never()).put(any(), any(RetryContext.class)); verify(cache, never()).remove(any()); - assertThat(map).as("Expected map.size() = 0, was: " + map.size()).hasSize(0); } finally { container.stop(); @@ -161,10 +158,7 @@ public class MissingIdRetryTests { try { assertThat(latch.await(30, TimeUnit.SECONDS)).isTrue(); Map map = (Map) new DirectFieldAccessor(cache).getPropertyValue("map"); - int n = 0; - while (n++ < 100 && map.size() != 0) { - Thread.sleep(100); - } + await().until(() -> map.size() == 0); ArgumentCaptor putCaptor = ArgumentCaptor.forClass(Object.class); ArgumentCaptor getCaptor = ArgumentCaptor.forClass(Object.class); ArgumentCaptor removeCaptor = ArgumentCaptor.forClass(Object.class); @@ -175,7 +169,6 @@ public class MissingIdRetryTests { logger.debug("puts:" + putCaptor.getAllValues()); logger.debug("gets:" + putCaptor.getAllValues()); logger.debug("removes:" + removeCaptor.getAllValues()); - assertThat(map).as("Expected map.size() = 0, was: " + map.size()).hasSize(0); } finally { container.stop(); @@ -226,10 +219,7 @@ public class MissingIdRetryTests { try { assertThat(cdl.await(30, TimeUnit.SECONDS)).isTrue(); Map map = (Map) new DirectFieldAccessor(cache).getPropertyValue("map"); - int n = 0; - while (n++ < 100 && map.size() != 0) { - Thread.sleep(100); - } + await().until(() -> map.size() == 0); ArgumentCaptor putCaptor = ArgumentCaptor.forClass(Object.class); ArgumentCaptor getCaptor = ArgumentCaptor.forClass(Object.class); ArgumentCaptor removeCaptor = ArgumentCaptor.forClass(Object.class); @@ -240,7 +230,6 @@ public class MissingIdRetryTests { logger.debug("puts:" + putCaptor.getAllValues()); logger.debug("gets:" + putCaptor.getAllValues()); logger.debug("removes:" + removeCaptor.getAllValues()); - assertThat(map).as("Expected map.size() = 0, was: " + map.size()).hasSize(0); } finally { container.stop(); diff --git a/src/checkstyle/checkstyle.xml b/src/checkstyle/checkstyle.xml index 59662a38..73ba6197 100644 --- a/src/checkstyle/checkstyle.xml +++ b/src/checkstyle/checkstyle.xml @@ -79,7 +79,9 @@ + value="org.assertj.core.api.Assertions.*, org.junit.Assert.*, org.junit.Assume.*, + org.awaitility.Awaitility.*, + org.junit.internal.matchers.ThrowableMessageMatcher.*, org.hamcrest.CoreMatchers.*, org.hamcrest.Matchers.*, org.hamcrest.collection.IsArrayContainingInOrder.*, org.springframework.boot.configurationprocessor.ConfigurationMetadataMatchers.*, org.springframework.boot.configurationprocessor.TestCompiler.*, org.mockito.Mockito.*, org.mockito.BDDMockito.*, org.mockito.ArgumentMatchers.*, org.mockito.AdditionalMatchers.*, org.springframework.test.web.servlet.request.MockMvcRequestBuilders.*, org.springframework.test.web.servlet.result.MockMvcResultMatchers.*, org.springframework.security.test.web.servlet.request.SecurityMockMvcRequestPostProcessors.*, org.springframework.security.test.web.servlet.setup.SecurityMockMvcConfigurers.*, org.springframework.hateoas.mvc.ControllerLinkBuilder.linkTo, org.springframework.amqp.rabbit.test.RabbitMatchers.*"/>