GH-1561: Alwways run stop callback

Fixes https://github.com/spring-projects/spring-amqp/issues/1561

* GH-1561 run callback when container is stopping for abort too
* GH-1561 add author information

**Cherry-pick to `2.4.x`**
This commit is contained in:
Tim Bq
2023-02-07 15:53:47 +01:00
committed by GitHub
parent b3a4b2589e
commit 931896d4f0
2 changed files with 36 additions and 5 deletions

View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2022 the original author or authors.
* Copyright 2002-2023 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.
@@ -79,6 +79,7 @@ import com.rabbitmq.client.ShutdownSignalException;
* @author Alex Panchenko
* @author Mat Jaggard
* @author Yansong Ren
* @author Tim Bourquin
*
* @since 1.0
*/
@@ -622,6 +623,7 @@ public class SimpleMessageListenerContainer extends AbstractMessageListenerConta
Thread thread = this.containerStoppingForAbort.get();
if (thread != null && !thread.equals(Thread.currentThread())) {
logger.info("Shutdown ignored - container is stopping due to an aborted consumer");
runCallbackIfNotNull(callback);
return;
}
@@ -641,6 +643,7 @@ public class SimpleMessageListenerContainer extends AbstractMessageListenerConta
}
else {
logger.info("Shutdown ignored - container is already stopped");
runCallbackIfNotNull(callback);
return;
}
}
@@ -674,9 +677,7 @@ public class SimpleMessageListenerContainer extends AbstractMessageListenerConta
this.cancellationLock.deactivate();
}
if (callback != null) {
callback.run();
}
runCallbackIfNotNull(callback);
};
if (callback == null) {
awaitShutdown.run();
@@ -686,6 +687,12 @@ public class SimpleMessageListenerContainer extends AbstractMessageListenerConta
}
}
private void runCallbackIfNotNull(@Nullable Runnable callback) {
if (callback != null) {
callback.run();
}
}
private boolean isActive(BlockingQueueConsumer consumer) {
boolean consumerActive;
synchronized (this.consumersMonitor) {

View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2022 the original author or authors.
* Copyright 2002-2023 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.
@@ -108,6 +108,7 @@ import com.rabbitmq.client.PossibleAuthenticationFailureException;
* @author Artem Bilan
* @author Mohammad Hewedy
* @author Yansong Ren
* @author Tim Bourquin
*/
public class SimpleMessageListenerContainerTests {
@@ -431,6 +432,29 @@ public class SimpleMessageListenerContainerTests {
}).given(channel).basicCancel(anyString());
}
@Test
public void testCallbackIsRunOnStopAlsoWhenNoConsumerIsActive() throws InterruptedException {
ConnectionFactory connectionFactory = mock(ConnectionFactory.class);
SimpleMessageListenerContainer container = new SimpleMessageListenerContainer(connectionFactory);
final CountDownLatch countDownLatch = new CountDownLatch(1);
container.stop(countDownLatch::countDown);
assertThat(countDownLatch.await(100, TimeUnit.MILLISECONDS)).isTrue();
}
@Test
public void testCallbackIsRunOnStopAlsoWhenContainerIsStoppingForAbort() throws InterruptedException {
ConnectionFactory connectionFactory = mock(ConnectionFactory.class);
SimpleMessageListenerContainer container = new SimpleMessageListenerContainer(connectionFactory);
ReflectionTestUtils.setField(container, "containerStoppingForAbort", new AtomicReference<>(new Thread()));
final CountDownLatch countDownLatch = new CountDownLatch(1);
container.stop(countDownLatch::countDown);
assertThat(countDownLatch.await(100, TimeUnit.MILLISECONDS)).isTrue();
}
@Test
public void testWithConnectionPerListenerThread() throws Exception {
com.rabbitmq.client.ConnectionFactory mockConnectionFactory =