diff --git a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/SimpleMessageListenerContainer.java b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/SimpleMessageListenerContainer.java index 49152ac8..fedd3fd1 100644 --- a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/SimpleMessageListenerContainer.java +++ b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/SimpleMessageListenerContainer.java @@ -358,8 +358,7 @@ public class SimpleMessageListenerContainer extends AbstractMessageListenerConta */ @Override public void addQueueNames(String... queueName) { - super.addQueueNames(queueName); - queuesChanged(); + super.addQueueNames(queueName); // calls addQueues() which will cycle consumers } /** @@ -756,7 +755,13 @@ public class SimpleMessageListenerContainer extends AbstractMessageListenerConta consumerIterator.remove(); count++; } - this.addAndStartConsumers(count); + try { + this.cancellationLock.await(getShutdownTimeout(), TimeUnit.MILLISECONDS); + } + catch (InterruptedException e) { + Thread.currentThread().interrupt(); + } + addAndStartConsumers(count); } } } 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 843345c3..3b0f85b9 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-2018 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. @@ -30,6 +30,7 @@ import static org.junit.Assert.assertThat; import static org.junit.Assert.assertTrue; import static org.mockito.ArgumentMatchers.any; import static org.mockito.ArgumentMatchers.anyBoolean; +import static org.mockito.BDDMockito.willAnswer; import static org.mockito.Mockito.atLeastOnce; import static org.mockito.Mockito.doReturn; import static org.mockito.Mockito.spy; @@ -149,14 +150,21 @@ public class SimpleMessageListenerContainerIntegration2Tests { @Test public void testChangeQueues() throws Exception { CountDownLatch latch = new CountDownLatch(30); - container = - createContainer(new MessageListenerAdapter(new PojoListener(latch)), queue.getName(), queue1.getName()); + AtomicInteger restarts = new AtomicInteger(); + container = spy(createContainer(new MessageListenerAdapter(new PojoListener(latch)), false, queue.getName(), + queue1.getName())); + willAnswer(invocation -> { + restarts.incrementAndGet(); + invocation.callRealMethod(); + return null; + }).given(container).addAndStartConsumers(1); final CountDownLatch consumerLatch = new CountDownLatch(1); this.container.setApplicationEventPublisher(e -> { if (e instanceof AsyncConsumerStoppedEvent) { consumerLatch.countDown(); } }); + this.container.start(); for (int i = 0; i < 10; i++) { template.convertAndSend(queue.getName(), i + "foo"); template.convertAndSend(queue1.getName(), i + "foo"); @@ -170,6 +178,7 @@ public class SimpleMessageListenerContainerIntegration2Tests { assertTrue("Timed out waiting for message", waited); assertNull(template.receiveAndConvert(queue.getName())); assertNull(template.receiveAndConvert(queue1.getName())); + assertThat(restarts.get(), equalTo(1)); } @Test