New Conn. Factories - Honor Physical Close Request

The new connection factories did not honor `RabbitUtils.isPhysicalCloseRequired()`.

Author: garyrussell

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

@@ -1338,6 +1338,7 @@ private void physicalClose(Object proxy) throws IOException, TimeoutException {
 			if (logger.isDebugEnabled()) {
 				logger.debug("Closing cached Channel: " + this.target);
 			}
+			RabbitUtils.clearPhysicalCloseRequired();
 			if (this.target == null) {
 				return;
 			}

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

@@ -17,11 +17,15 @@
 package org.springframework.amqp.rabbit.connection;
 
 import java.io.IOException;
+import java.util.concurrent.TimeoutException;
+import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicReference;
 import java.util.function.BiConsumer;
 
 import org.aopalliance.aop.Advice;
 import org.aopalliance.intercept.MethodInterceptor;
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
 import org.apache.commons.pool2.ObjectPool;
 import org.apache.commons.pool2.PooledObject;
 import org.apache.commons.pool2.PooledObjectFactory;
@@ -120,6 +124,8 @@ public synchronized void destroy() {
 
 	private static final class ConnectionWrapper extends SimpleConnection {
 
+		private static final Log logger = LogFactory.getLog(ConnectionWrapper.class);
+
 		private final ObjectPool<Channel> channels;
 
 		private final ObjectPool<Channel> txChannels;
@@ -152,23 +158,55 @@ public Channel createChannel(boolean transactional) {
 		private Channel createProxy(Channel channel, boolean transacted) {
 			ProxyFactory pf = new ProxyFactory(channel);
 			AtomicReference<Channel> proxy = new AtomicReference<>();
+			AtomicBoolean confirmSelected = new AtomicBoolean();
 			Advice advice =
 					(MethodInterceptor) invocation -> {
-						if (transacted) {
-							ConnectionWrapper.this.txChannels.returnObject(proxy.get());
-						}
-						else {
-							ConnectionWrapper.this.channels.returnObject(proxy.get());
+						String method = invocation.getMethod().getName();
+						switch (method) {
+						case "close":
+							return handleClose(channel, transacted, proxy);
+						case "getTargetChannel":
+							return channel;
+						case "isTransactional":
+							return transacted;
+						case "confirmSelect":
+							confirmSelected.set(true);
+							return channel.confirmSelect();
+						case "isConfirmSelected":
+							return confirmSelected.get();
 						}
 						return null;
 					};
 			NameMatchMethodPointcutAdvisor advisor = new NameMatchMethodPointcutAdvisor(advice);
 			advisor.addMethodName("close");
+			advisor.addMethodName("getTargetChannel");
+			advisor.addMethodName("isTransactional");
+			advisor.addMethodName("confirmSelect");
+			advisor.addMethodName("isConfirmSelected");
 			pf.addAdvisor(advisor);
+			pf.addInterface(ChannelProxy.class);
 			proxy.set((Channel) pf.getProxy());
 			return proxy.get();
 		}
 
+		private Object handleClose(Channel channel, boolean transacted, AtomicReference<Channel> proxy)
+				throws Exception {
+
+			if (!RabbitUtils.isPhysicalCloseRequired()) {
+				if (transacted) {
+					ConnectionWrapper.this.txChannels.returnObject(proxy.get());
+				}
+				else {
+					ConnectionWrapper.this.channels.returnObject(proxy.get());
+				}
+				return null;
+			}
+			else {
+				physicalClose(channel);
+			}
+			return null;
+		}
+
 		@Override
 		public void close() {
 		}
@@ -179,11 +217,23 @@ void forceClose() {
 			this.txChannels.close();
 		}
 
+		private void physicalClose(Channel channel) {
+			RabbitUtils.clearPhysicalCloseRequired();
+			if (channel.isOpen()) {
+				try {
+					channel.close();
+				}
+				catch (IOException | TimeoutException e) {
+					logger.debug("Error on close", e);
+				}
+			}
+		}
+
 		private class ChannelFactory implements PooledObjectFactory<Channel> {
 
 			@Override
 			public PooledObject<Channel> makeObject() {
-				Channel channel = ConnectionWrapper.super.createChannel(false);
+				Channel channel = createProxy(ConnectionWrapper.super.createChannel(false), false);
 				if (ConnectionWrapper.this.simplePublisherConfirms) {
 					try {
 						channel.confirmSelect();
@@ -192,7 +242,7 @@ public PooledObject<Channel> makeObject() {
 						throw RabbitExceptionTranslator.convertRabbitAccessException(e);
 					}
 				}
-				return new DefaultPooledObject<>(createProxy(channel, false));
+				return new DefaultPooledObject<>(channel);
 			}
 
 			@Override
@@ -219,14 +269,14 @@ private final class TxChannelFactory extends ChannelFactory {
 
 			@Override
 			public PooledObject<Channel> makeObject() {
-				Channel channel = ConnectionWrapper.super.createChannel(true);
+				Channel channel = createProxy(ConnectionWrapper.super.createChannel(true), true);
 				try {
 					channel.txSelect();
 				}
 				catch (IOException e) {
 					throw RabbitExceptionTranslator.convertRabbitAccessException(e);
 				}
-				return new DefaultPooledObject<>(createProxy(channel, true));
+				return new DefaultPooledObject<>(channel);
 			}
 
 		}

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

@@ -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.
@@ -229,12 +229,18 @@ public static void setPhysicalCloseRequired(Channel channel, boolean b) {
 	public static boolean isPhysicalCloseRequired() {
 		Boolean mustClose = physicalCloseRequired.get();
 		if (mustClose == null) {
-			mustClose = Boolean.FALSE;
+			return false;
 		}
 		else {
-			physicalCloseRequired.remove();
+			return mustClose;
 		}
-		return mustClose;
+	}
+
+	/**
+	 * Clear the physicalCloseRequired flag.
+	 */
+	public static void clearPhysicalCloseRequired() {
+		physicalCloseRequired.remove();
 	}
 
 	/**

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

@@ -18,9 +18,11 @@
 
 import java.io.IOException;
 import java.util.concurrent.TimeoutException;
+import java.util.concurrent.atomic.AtomicBoolean;
 
 import org.aopalliance.aop.Advice;
 import org.aopalliance.intercept.MethodInterceptor;
+import org.aopalliance.intercept.MethodInvocation;
 
 import org.springframework.amqp.AmqpException;
 import org.springframework.amqp.rabbit.support.RabbitExceptionTranslator;
@@ -125,15 +127,14 @@ private final class ConnectionWrapper extends SimpleConnection {
 		public Channel createChannel(boolean transactional) {
 			Channel channel = transactional ? this.txChannels.get() : this.channels.get();
 			if (channel == null || !channel.isOpen()) {
-				channel = super.createChannel(transactional);
+				channel = createProxy(super.createChannel(transactional), transactional);
 				if (transactional) {
 					try {
 						channel.txSelect();
 					}
 					catch (IOException e) {
 						throw RabbitExceptionTranslator.convertRabbitAccessException(e);
 					}
-					channel = createProxy(channel);
 					this.txChannels.set(channel);
 				}
 				else {
@@ -145,30 +146,65 @@ public Channel createChannel(boolean transactional) {
 							throw RabbitExceptionTranslator.convertRabbitAccessException(e);
 						}
 					}
-					channel = createProxy(channel);
 					this.channels.set(channel);
 				}
 			}
 			return channel;
 		}
 
-		private Channel createProxy(Channel channel) {
+		private Channel createProxy(Channel channel, boolean transactional) {
 			ProxyFactory pf = new ProxyFactory(channel);
+			AtomicBoolean confirmSelected = new AtomicBoolean();
 			Advice advice =
 					(MethodInterceptor) invocation -> {
-						if (ConnectionWrapper.this.channels.get() == null) {
-							return invocation.proceed();
-						}
-						else {
-							return null;
+						String method = invocation.getMethod().getName();
+						switch (method) {
+							case "close":
+								return handleClose(channel, transactional, invocation);
+							case "getTargetChannel":
+								return channel;
+							case "isTransactional":
+								return transactional;
+							case "confirmSelect":
+								confirmSelected.set(true);
+								return channel.confirmSelect();
+							case "isConfirmSelected":
+								return confirmSelected.get();
 						}
+						return null;
 					};
 			NameMatchMethodPointcutAdvisor advisor = new NameMatchMethodPointcutAdvisor(advice);
 			advisor.addMethodName("close");
+			advisor.addMethodName("getTargetChannel");
+			advisor.addMethodName("isTransactional");
+			advisor.addMethodName("confirmSelect");
+			advisor.addMethodName("isConfirmSelected");
 			pf.addAdvisor(advisor);
+			pf.addInterface(ChannelProxy.class);
 			return (Channel) pf.getProxy();
 		}
 
+		private Object handleClose(Channel channel, boolean transactional, MethodInvocation invocation)
+				throws Throwable {
+
+			if (ConnectionWrapper.this.channels.get() == null) {
+				return invocation.proceed();
+			}
+			else {
+				if (RabbitUtils.isPhysicalCloseRequired()) {
+					physicalClose(channel);
+					if (transactional) {
+						this.txChannels.remove();
+					}
+					else {
+						this.channels.remove();
+					}
+					RabbitUtils.clearPhysicalCloseRequired();
+				}
+				return null;
+			}
+		}
+
 		@Override
 		public void close() {
 		}
@@ -183,13 +219,17 @@ private void doClose(ThreadLocal<Channel> channelsTL) {
 			Channel channel = channelsTL.get();
 			if (channel != null) {
 				channelsTL.remove();
-				if (channel.isOpen()) {
-					try {
-						channel.close();
-					}
-					catch (IOException | TimeoutException e) {
-						logger.debug("Error on close", e);
-					}
+				physicalClose(channel);
+			}
+		}
+
+		private void physicalClose(Channel channel) {
+			if (channel.isOpen()) {
+				try {
+					channel.close();
+				}
+				catch (IOException | TimeoutException e) {
+					logger.debug("Error on close", e);
 				}
 			}
 		}

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

@@ -50,10 +50,12 @@ void testBasic() throws Exception {
 				nonTxConfiged.set(true);
 			}
 		});
+		pcf.setSimplePublisherConfirms(true);
 		Connection conn = pcf.createConnection();
 		assertThat(txConfiged.get()).isTrue();
 		assertThat(nonTxConfiged.get()).isTrue();
 		Channel chann1 = conn.createChannel(false);
+		assertThat(((ChannelProxy) chann1).isConfirmSelected()).isTrue();
 		chann1.close();
 		Channel chann2 = conn.createChannel(false);
 		assertThat(chann2).isSameAs(chann1);
@@ -63,6 +65,11 @@ void testBasic() throws Exception {
 		chann1.close();
 		chann2 = conn.createChannel(true);
 		assertThat(chann2).isSameAs(chann1);
+		RabbitUtils.setPhysicalCloseRequired(chann2, true);
+		chann2.close();
+		chann1 = conn.createChannel(true);
+		assertThat(chann1).isNotSameAs(chann2);
+		chann1.close();
 		pcf.destroy();
 	}
 

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

@@ -66,7 +66,13 @@ void testBasic() throws Exception {
 				.isTrue();
 		chann2.close();
 		chann2 = conn.createChannel(false);
+		RabbitUtils.setPhysicalCloseRequired(chann2, true);
 		chann2.close();
+		scf.setSimplePublisherConfirms(true);
+		chann1 = conn.createChannel(false);
+		assertThat(chann1).isNotSameAs(chann2);
+		assertThat(((ChannelProxy) chann1).isConfirmSelected()).isTrue();
+		chann1.close();
 		scf.destroy();
 		assertThat(((Channel) TestUtils.getPropertyValue(conn, "channels", ThreadLocal.class).get()).isOpen())
 				.isFalse();