From 931896d4f03ce4804c1263af74cdc2dca6f2d7d2 Mon Sep 17 00:00:00 2001
From: Tim Bq <code.timbq@gmail.com>
Date: Tue, 7 Feb 2023 15:53:47 +0100
Subject: [PATCH]  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`**
---
 .../SimpleMessageListenerContainer.java       | 15 ++++++++---
 .../SimpleMessageListenerContainerTests.java  | 26 ++++++++++++++++++-
 2 files changed, 36 insertions(+), 5 deletions(-)

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 023a7a50..9f933202 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
@@ -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) {
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 1b7d7d12..3cabcfe3 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-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 =