GH-2504: Add SmartLifecycle to Conn. Factories

Resolves #2504

Author: garyrussell

spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/CachingConnectionFactory.java

@@ -50,6 +50,7 @@
 import org.springframework.amqp.AmqpTimeoutException;
 import org.springframework.amqp.rabbit.support.ActiveObjectCounter;
 import org.springframework.beans.factory.InitializingBean;
+import org.springframework.context.SmartLifecycle;
 import org.springframework.jmx.export.annotation.ManagedAttribute;
 import org.springframework.jmx.export.annotation.ManagedResource;
 import org.springframework.lang.Nullable;
@@ -99,7 +100,7 @@
  */
 @ManagedResource
 public class CachingConnectionFactory extends AbstractConnectionFactory
-		implements InitializingBean, ShutdownListener {
+		implements InitializingBean, ShutdownListener, SmartLifecycle {
 
 	private static final String UNUSED = "unused";
 
@@ -190,6 +191,8 @@ public enum ConfirmType {
 
 	private final ActiveObjectCounter<Channel> inFlightAsyncCloses = new ActiveObjectCounter<>();
 
+	private final AtomicBoolean running = new AtomicBoolean();
+
 	private long channelCheckoutTimeout = 0;
 
 	private CacheMode cacheMode = CacheMode.CHANNEL;
@@ -446,6 +449,11 @@ public void setPublisherChannelFactory(PublisherCallbackChannelFactory publisher
 		this.publisherChannelFactory = publisherChannelFactory;
 	}
 
+	@Override
+	public int getPhase() {
+		return Integer.MIN_VALUE;
+	}
+
 	@Override
 	public void afterPropertiesSet() {
 		this.initialized = true;
@@ -459,6 +467,22 @@ public void afterPropertiesSet() {
 		}
 	}
 
+	@Override
+	public void start() {
+		this.running.set(true);
+	}
+
+	@Override
+	public void stop() {
+		this.running.set(false);
+		resetConnection();
+	}
+
+	@Override
+	public boolean isRunning() {
+		return this.running.get();
+	}
+
 	private void initCacheWaterMarks() {
 		this.channelHighWaterMarks.put(ObjectUtils.getIdentityHexString(this.cachedChannelsNonTransactional),
 				new AtomicInteger());

spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/LocalizedQueueConnectionFactory.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2015-2022 the original author or authors.
+ * Copyright 2015-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.
@@ -20,6 +20,7 @@
 import java.util.Arrays;
 import java.util.HashMap;
 import java.util.Map;
+import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.stream.Collectors;
 import java.util.stream.IntStream;
 
@@ -29,6 +30,7 @@
 import org.springframework.amqp.AmqpException;
 import org.springframework.amqp.rabbit.support.RabbitExceptionTranslator;
 import org.springframework.beans.factory.DisposableBean;
+import org.springframework.context.SmartLifecycle;
 import org.springframework.core.io.Resource;
 import org.springframework.lang.Nullable;
 import org.springframework.util.Assert;
@@ -49,7 +51,8 @@
  * @author Gary Russell
  * @since 1.2
  */
-public class LocalizedQueueConnectionFactory implements ConnectionFactory, RoutingConnectionFactory, DisposableBean {
+public class LocalizedQueueConnectionFactory implements ConnectionFactory, RoutingConnectionFactory, DisposableBean,
+		SmartLifecycle {
 
 	private final Log logger = LogFactory.getLog(getClass());
 
@@ -79,6 +82,8 @@ public class LocalizedQueueConnectionFactory implements ConnectionFactory, Routi
 
 	private final String trustStorePassPhrase;
 
+	private final AtomicBoolean running = new AtomicBoolean();
+
 	private NodeLocator<?> nodeLocator;
 
 	/**
@@ -233,6 +238,27 @@ public String getUsername() {
 		return this.username;
 	}
 
+	@Override
+	public int getPhase() {
+		return Integer.MIN_VALUE;
+	}
+
+	@Override
+	public void start() {
+		this.running.set(true);
+	}
+
+	@Override
+	public void stop() {
+		this.running.set(false);
+		resetConnection();
+	}
+
+	@Override
+	public boolean isRunning() {
+		return this.running.get();
+	}
+
 	@Override
 	public void addConnectionListener(ConnectionListener listener) {
 		this.defaultConnectionFactory.addConnectionListener(listener);

spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/PooledChannelConnectionFactory.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2020-2022 the original author or authors.
+ * Copyright 2020-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.
@@ -36,6 +36,7 @@
 import org.springframework.amqp.rabbit.support.RabbitExceptionTranslator;
 import org.springframework.aop.framework.ProxyFactory;
 import org.springframework.aop.support.NameMatchMethodPointcutAdvisor;
+import org.springframework.context.SmartLifecycle;
 import org.springframework.lang.Nullable;
 import org.springframework.util.Assert;
 
@@ -54,7 +55,10 @@
  * @since 2.3
  *
  */
-public class PooledChannelConnectionFactory extends AbstractConnectionFactory implements ShutdownListener {
+public class PooledChannelConnectionFactory extends AbstractConnectionFactory
+		implements ShutdownListener, SmartLifecycle {
+
+	private final AtomicBoolean running = new AtomicBoolean();
 
 	private volatile ConnectionWrapper connection;
 
@@ -132,6 +136,27 @@ public void addConnectionListener(ConnectionListener listener) {
 		}
 	}
 
+	@Override
+	public int getPhase() {
+		return Integer.MIN_VALUE;
+	}
+
+	@Override
+	public void start() {
+		this.running.set(true);
+	}
+
+	@Override
+	public void stop() {
+		this.running.set(false);
+		resetConnection();
+	}
+
+	@Override
+	public boolean isRunning() {
+		return this.running.get();
+	}
+
 	@Override
 	public synchronized Connection createConnection() throws AmqpException {
 		if (this.connection == null || !this.connection.isOpen()) {
@@ -180,16 +205,20 @@ private static final class ConnectionWrapper extends SimpleConnection {
 				BiConsumer<GenericObjectPool<Channel>, Boolean> configurer, ChannelListener channelListener) {
 
 			super(delegate, closeTimeout);
-			GenericObjectPool<Channel> pool = new GenericObjectPool<>(new ChannelFactory());
-			configurer.accept(pool, false);
-			this.channels = pool;
-			pool = new GenericObjectPool<>(new TxChannelFactory());
-			configurer.accept(pool, true);
-			this.txChannels = pool;
+			this.channels = createPool(new ChannelFactory(), configurer, false);
+			this.txChannels = createPool(new TxChannelFactory(), configurer, true);
 			this.simplePublisherConfirms = simplePublisherConfirms;
 			this.channelListener = channelListener;
 		}
 
+		private GenericObjectPool<Channel> createPool(ChannelFactory channelFactory,
+				BiConsumer<GenericObjectPool<Channel>, Boolean> configurer, boolean tx) {
+
+			GenericObjectPool<Channel> pool = new GenericObjectPool<>(channelFactory);
+			configurer.accept(pool, tx);
+			return pool;
+		}
+
 		@Override
 		public Channel createChannel(boolean transactional) {
 			try {

spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/ThreadChannelConnectionFactory.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2020-2022 the original author or authors.
+ * Copyright 2020-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.
@@ -31,6 +31,7 @@
 import org.springframework.amqp.rabbit.support.RabbitExceptionTranslator;
 import org.springframework.aop.framework.ProxyFactory;
 import org.springframework.aop.support.NameMatchMethodPointcutAdvisor;
+import org.springframework.context.SmartLifecycle;
 import org.springframework.lang.Nullable;
 import org.springframework.util.Assert;
 
@@ -48,12 +49,15 @@
  * @since 2.3
  *
  */
-public class ThreadChannelConnectionFactory extends AbstractConnectionFactory implements ShutdownListener {
+public class ThreadChannelConnectionFactory extends AbstractConnectionFactory
+		implements ShutdownListener, SmartLifecycle {
 
 	private final Map<UUID, Context> contextSwitches = new ConcurrentHashMap<>();
 
 	private final Map<UUID, Thread> switchesInProgress = new ConcurrentHashMap<>();
 
+	private final AtomicBoolean running = new AtomicBoolean();
+
 	private volatile ConnectionWrapper connection;
 
 	private boolean simplePublisherConfirms;
@@ -106,6 +110,27 @@ public void setSimplePublisherConfirms(boolean simplePublisherConfirms) {
 		}
 	}
 
+	@Override
+	public int getPhase() {
+		return Integer.MIN_VALUE;
+	}
+
+	@Override
+	public void start() {
+		this.running.set(true);
+	}
+
+	@Override
+	public void stop() {
+		this.running.set(false);
+		resetConnection();
+	}
+
+	@Override
+	public boolean isRunning() {
+		return this.running.get();
+	}
+
 	@Override
 	public void addConnectionListener(ConnectionListener listener) {
 		super.addConnectionListener(listener); // handles publishing sub-factory

spring-rabbit/src/test/java/org/springframework/amqp/rabbit/connection/CachingConnectionFactoryIntegrationTests.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2019 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.
@@ -291,7 +291,7 @@ public void testSendAndReceiveFromVolatileQueueAfterImplicitRemoval() throws Exc
 		template.convertAndSend(queue.getName(), "message");
 
 		// Force a physical close of the channel
-		this.connectionFactory.resetConnection();
+		this.connectionFactory.stop();
 
 		// The queue was removed when the channel was closed
 		assertThatThrownBy(() -> template.receiveAndConvert(queue.getName()))

spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/BrokerDeclaredQueueNameTests.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2018-2019 the original author or authors.
+ * Copyright 2018-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.
@@ -124,7 +124,7 @@ private void testBrokerNamedQueue(AbstractMessageListenerContainer container,
 		assertThat(this.message.get().getBody()).isEqualTo("foo".getBytes());
 		final CountDownLatch newConnectionLatch = new CountDownLatch(2);
 		this.cf.addConnectionListener(c -> newConnectionLatch.countDown());
-		this.cf.resetConnection();
+		this.cf.stop();
 		assertThat(newConnectionLatch.await(10, TimeUnit.SECONDS)).isTrue();
 		String secondActualName = queue.getActualName();
 		assertThat(secondActualName).isNotEqualTo(firstActualName);