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

Use Awaitility in tests

Browse Source
This commit is contained in:
Gary Russell
2020-05-19 19:55:18 -04:00
committed by Artem Bilan
parent 095b2df3c4
commit 7cc3f0a5ee
28 changed files with 183 additions and 377 deletions
Download Patch File Download Diff File Expand all files Collapse all files

13
build.gradle
View File

@@ -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'

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

@@ -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

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

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

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

@@ -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")

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

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

18
spring-rabbit/src/test/java/org/springframework/amqp/rabbit/config/QueueParserIntegrationTests.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,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");
}

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

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

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

@@ -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

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

@@ -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<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 @@ 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<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 @@ 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<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 @@ 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<String, Object> arguments = queue.getArguments();
assertThat(arguments.get("x-queue-type")).isEqualTo("quorum");
assertThat(arguments.get("x-delivery-limit")).isEqualTo(10);

14
spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/RabbitAdminIntegrationTests.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,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);
}
/**

33
spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/RabbitAdminTests.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.
@@ -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();
}

18
spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/RabbitBindingIntegrationTests.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,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<BlockingQueueConsumer>(), 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;
}

11
spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/RabbitRestApiTests.java
View File

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

23
spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/RabbitTemplateIntegrationTests.java
View File

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

27
spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/RabbitTemplatePublisherCallbacksIntegrationTests.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,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

16
spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/ContainerInitializationTests.java
View File

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

54
spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/DirectMessageListenerContainerIntegrationTests.java
View File

@@ -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 {

13
spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/DirectReplyToMessageListenerContainerTests.java
View File

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

10
spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/ListenFromAutoDeleteQueueTests.java
View File

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

23
spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/MessageListenerContainerErrorHandlerIntegrationTests.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.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<Message> failed = new AtomicReference<Message>();
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();

59
spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/MessageListenerContainerLifecycleIntegrationTests.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.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<BlockingQueueConsumer> consumers = (Set<BlockingQueueConsumer>) 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();
}

11
spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/MessageListenerRecoveryCachingConnectionIntegrationTests.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.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");

44
spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/SimpleMessageListenerContainerIntegration2Tests.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,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<String> captor = ArgumentCaptor.forClass(String.class);
verify(logger).error(captor.capture());
assertThat(captor.getValue()).contains("Stopping container from aborted consumer");

9
spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/SimpleMessageListenerContainerLongTests.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.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);

25
spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/SimpleMessageListenerContainerTests.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.
@@ -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")

7
spring-rabbit/src/test/java/org/springframework/amqp/rabbit/log4j2/AmqpAppenderTests.java
View File

@@ -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

21
spring-rabbit/src/test/java/org/springframework/amqp/rabbit/retry/MissingIdRetryTests.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,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();

4
src/checkstyle/checkstyle.xml
View File

@@ -79,7 +79,9 @@
<module name="AvoidStarImport"/>
<module name="AvoidStaticImport">
<property name="excludes"
value="org.assertj.core.api.Assertions.*, org.junit.Assert.*, org.junit.Assume.*, 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.*"/>
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.*"/>
</module>
<module name="IllegalImport"/>
<module name="RedundantImport"/>