Fix Sonar issues and other refactoring

- Allow the publishing connection factory to be explicitly set and be
  any type
- Do not propagate properties when the factory is explicitly configured

* Reset flag when this is a publisher CF.

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

@@ -113,6 +113,8 @@ public abstract class AbstractConnectionFactory implements ConnectionFactory, Di
 
 	private final AtomicInteger defaultConnectionNameStrategyCounter = new AtomicInteger();
 
+	private ConditionalExceptionLogger closeExceptionLogger = new DefaultChannelCloseLogger();
+
 	private AbstractConnectionFactory publisherConnectionFactory;
 
 	private RecoveryListener recoveryListener = new RecoveryListener() {
@@ -156,8 +158,6 @@ public abstract class AbstractConnectionFactory implements ConnectionFactory, Di
 
 	private volatile boolean contextStopped;
 
-	protected ConditionalExceptionLogger closeExceptionLogger = new DefaultChannelCloseLogger();
-
 	/**
 	 * Create a new AbstractConnectionFactory for the given target ConnectionFactory,
 	 * with no publisher connection factory.
@@ -168,8 +168,21 @@ public abstract class AbstractConnectionFactory implements ConnectionFactory, Di
 		this.rabbitConnectionFactory = rabbitConnectionFactory;
 	}
 
-	protected final void setPublisherConnectionFactory(
-			AbstractConnectionFactory publisherConnectionFactory) {
+	/**
+	 * Set a custom publisher connection factory; the type does not need to be the same
+	 * as this factory.
+	 * @param publisherConnectionFactory the factory.
+	 * @since 2.3.2
+	 */
+	public void setPublisherConnectionFactory(
+			@Nullable AbstractConnectionFactory publisherConnectionFactory) {
+
+		doSetPublisherConnectionFactory(publisherConnectionFactory);
+	}
+
+	protected final void doSetPublisherConnectionFactory(
+			@Nullable AbstractConnectionFactory publisherConnectionFactory) {
+
 		this.publisherConnectionFactory = publisherConnectionFactory;
 	}
 
@@ -472,6 +485,26 @@ public abstract class AbstractConnectionFactory implements ConnectionFactory, Di
 		}
 	}
 
+	/**
+	 * Set the strategy for logging close exceptions; by default, if a channel is closed due to a failed
+	 * passive queue declaration, it is logged at debug level. Normal channel closes (200 OK) are not
+	 * logged. All others are logged at ERROR level (unless access is refused due to an exclusive consumer
+	 * condition, in which case, it is logged at INFO level).
+	 * @param closeExceptionLogger the {@link ConditionalExceptionLogger}.
+	 * @since 1.5
+	 */
+	public void setCloseExceptionLogger(ConditionalExceptionLogger closeExceptionLogger) {
+		Assert.notNull(closeExceptionLogger, "'closeExceptionLogger' cannot be null");
+		this.closeExceptionLogger = closeExceptionLogger;
+		if (this.publisherConnectionFactory != null) {
+			this.publisherConnectionFactory.setCloseExceptionLogger(closeExceptionLogger);
+		}
+	}
+
+	protected ConnectionNameStrategy getConnectionNameStrategy() {
+		return this.connectionNameStrategy;
+	}
+
 	@Override
 	public void setBeanName(String name) {
 		this.beanName = name;

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

@@ -49,7 +49,6 @@ import org.springframework.amqp.AmqpApplicationContextClosedException;
 import org.springframework.amqp.AmqpException;
 import org.springframework.amqp.AmqpTimeoutException;
 import org.springframework.amqp.rabbit.support.ActiveObjectCounter;
-import org.springframework.amqp.support.ConditionalExceptionLogger;
 import org.springframework.beans.factory.InitializingBean;
 import org.springframework.jmx.export.annotation.ManagedAttribute;
 import org.springframework.jmx.export.annotation.ManagedResource;
@@ -184,8 +183,6 @@ public class CachingConnectionFactory extends AbstractConnectionFactory
 
 	private final AtomicInteger connectionHighWaterMark = new AtomicInteger();
 
-	private final CachingConnectionFactory publisherConnectionFactory;
-
 	/** Synchronization monitor for the shared Connection. */
 	private final Object connectionMonitor = new Object();
 
@@ -207,6 +204,8 @@ public class CachingConnectionFactory extends AbstractConnectionFactory
 
 	private PublisherCallbackChannelFactory publisherChannelFactory = PublisherCallbackChannelImpl.factory();
 
+	private boolean defaultPublisherFactory = true;
+
 	private volatile boolean active = true;
 
 	private volatile boolean initialized;
@@ -257,8 +256,7 @@ public class CachingConnectionFactory extends AbstractConnectionFactory
 		}
 		setHost(hostname);
 		setPort(port);
-		this.publisherConnectionFactory = new CachingConnectionFactory(getRabbitConnectionFactory(), true);
-		setPublisherConnectionFactory(this.publisherConnectionFactory);
+		doSetPublisherConnectionFactory(new CachingConnectionFactory(getRabbitConnectionFactory(), true));
 	}
 
 	/**
@@ -269,8 +267,7 @@ public class CachingConnectionFactory extends AbstractConnectionFactory
 	public CachingConnectionFactory(URI uri) {
 		super(newRabbitConnectionFactory());
 		setUri(uri);
-		this.publisherConnectionFactory = new CachingConnectionFactory(getRabbitConnectionFactory(), true);
-		setPublisherConnectionFactory(this.publisherConnectionFactory);
+		doSetPublisherConnectionFactory(new CachingConnectionFactory(getRabbitConnectionFactory(), true));
 	}
 
 	/**
@@ -288,6 +285,7 @@ public class CachingConnectionFactory extends AbstractConnectionFactory
 	 */
 	private CachingConnectionFactory(com.rabbitmq.client.ConnectionFactory rabbitConnectionFactory,
 			boolean isPublisherFactory) {
+
 		super(rabbitConnectionFactory);
 		if (!isPublisherFactory) {
 			if (rabbitConnectionFactory.isAutomaticRecoveryEnabled()) {
@@ -301,11 +299,11 @@ public class CachingConnectionFactory extends AbstractConnectionFactory
 						+ "'getRabbitConnectionFactory().setAutomaticRecoveryEnabled(true)',\n"
 						+ "but this is discouraged.");
 			}
-			this.publisherConnectionFactory = new CachingConnectionFactory(getRabbitConnectionFactory(), true);
-			setPublisherConnectionFactory(this.publisherConnectionFactory);
+			super.setPublisherConnectionFactory(new CachingConnectionFactory(getRabbitConnectionFactory(), true));
 		}
 		else {
-			this.publisherConnectionFactory = null;
+			super.setPublisherConnectionFactory(null);
+			this.defaultPublisherFactory = false;
 		}
 	}
 
@@ -315,6 +313,12 @@ public class CachingConnectionFactory extends AbstractConnectionFactory
 		return connectionFactory;
 	}
 
+	@Override
+	public void setPublisherConnectionFactory(@Nullable AbstractConnectionFactory publisherConnectionFactory) {
+		super.setPublisherConnectionFactory(publisherConnectionFactory);
+		this.defaultPublisherFactory = false;
+	}
+
 	/**
 	 * The number of channels to maintain in the cache. By default, channels are allocated on
 	 * demand (unbounded) and this represents the maximum cache size. To limit the available
@@ -325,8 +329,8 @@ public class CachingConnectionFactory extends AbstractConnectionFactory
 	public void setChannelCacheSize(int sessionCacheSize) {
 		Assert.isTrue(sessionCacheSize >= 1, "Channel cache size must be 1 or higher");
 		this.channelCacheSize = sessionCacheSize;
-		if (this.publisherConnectionFactory != null) {
-			this.publisherConnectionFactory.setChannelCacheSize(sessionCacheSize);
+		if (this.defaultPublisherFactory) {
+			((CachingConnectionFactory) getPublisherConnectionFactory()).setChannelCacheSize(sessionCacheSize);
 		}
 	}
 
@@ -342,8 +346,8 @@ public class CachingConnectionFactory extends AbstractConnectionFactory
 		Assert.isTrue(!this.initialized, "'cacheMode' cannot be changed after initialization.");
 		Assert.notNull(cacheMode, "'cacheMode' must not be null.");
 		this.cacheMode = cacheMode;
-		if (this.publisherConnectionFactory != null) {
-			this.publisherConnectionFactory.setCacheMode(cacheMode);
+		if (this.defaultPublisherFactory) {
+			((CachingConnectionFactory) getPublisherConnectionFactory()).setCacheMode(cacheMode);
 		}
 	}
 
@@ -354,8 +358,8 @@ public class CachingConnectionFactory extends AbstractConnectionFactory
 	public void setConnectionCacheSize(int connectionCacheSize) {
 		Assert.isTrue(connectionCacheSize >= 1, "Connection cache size must be 1 or higher.");
 		this.connectionCacheSize = connectionCacheSize;
-		if (this.publisherConnectionFactory != null) {
-			this.publisherConnectionFactory.setConnectionCacheSize(connectionCacheSize);
+		if (this.defaultPublisherFactory) {
+			((CachingConnectionFactory) getPublisherConnectionFactory()).setConnectionCacheSize(connectionCacheSize);
 		}
 	}
 
@@ -370,8 +374,8 @@ public class CachingConnectionFactory extends AbstractConnectionFactory
 	public void setConnectionLimit(int connectionLimit) {
 		Assert.isTrue(connectionLimit >= 1, "Connection limit must be 1 or higher.");
 		this.connectionLimit = connectionLimit;
-		if (this.publisherConnectionFactory != null) {
-			this.publisherConnectionFactory.setConnectionLimit(connectionLimit);
+		if (this.defaultPublisherFactory) {
+			((CachingConnectionFactory) getPublisherConnectionFactory()).setConnectionLimit(connectionLimit);
 		}
 	}
 
@@ -387,8 +391,8 @@ public class CachingConnectionFactory extends AbstractConnectionFactory
 
 	public void setPublisherReturns(boolean publisherReturns) {
 		this.publisherReturns = publisherReturns;
-		if (this.publisherConnectionFactory != null) {
-			this.publisherConnectionFactory.setPublisherReturns(publisherReturns);
+		if (this.defaultPublisherFactory) {
+			((CachingConnectionFactory) getPublisherConnectionFactory()).setPublisherReturns(publisherReturns);
 		}
 	}
 
@@ -444,8 +448,8 @@ public class CachingConnectionFactory extends AbstractConnectionFactory
 	public void setPublisherConfirmType(ConfirmType confirmType) {
 		Assert.notNull(confirmType, "'confirmType' cannot be null");
 		this.confirmType = confirmType;
-		if (this.publisherConnectionFactory != null) {
-			this.publisherConnectionFactory.setPublisherConfirmType(confirmType);
+		if (this.defaultPublisherFactory) {
+			((CachingConnectionFactory) getPublisherConnectionFactory()).setPublisherConfirmType(confirmType);
 		}
 	}
 
@@ -463,24 +467,9 @@ public class CachingConnectionFactory extends AbstractConnectionFactory
 	 */
 	public void setChannelCheckoutTimeout(long channelCheckoutTimeout) {
 		this.channelCheckoutTimeout = channelCheckoutTimeout;
-		if (this.publisherConnectionFactory != null) {
-			this.publisherConnectionFactory.setChannelCheckoutTimeout(channelCheckoutTimeout);
-		}
-	}
-
-	/**
-	 * Set the strategy for logging close exceptions; by default, if a channel is closed due to a failed
-	 * passive queue declaration, it is logged at debug level. Normal channel closes (200 OK) are not
-	 * logged. All others are logged at ERROR level (unless access is refused due to an exclusive consumer
-	 * condition, in which case, it is logged at INFO level).
-	 * @param closeExceptionLogger the {@link ConditionalExceptionLogger}.
-	 * @since 1.5
-	 */
-	public void setCloseExceptionLogger(ConditionalExceptionLogger closeExceptionLogger) {
-		Assert.notNull(closeExceptionLogger, "'closeExceptionLogger' cannot be null");
-		this.closeExceptionLogger = closeExceptionLogger;
-		if (this.publisherConnectionFactory != null) {
-			this.publisherConnectionFactory.setCloseExceptionLogger(closeExceptionLogger);
+		if (this.defaultPublisherFactory) {
+			((CachingConnectionFactory) getPublisherConnectionFactory())
+					.setChannelCheckoutTimeout(channelCheckoutTimeout);
 		}
 	}
 
@@ -502,8 +491,8 @@ public class CachingConnectionFactory extends AbstractConnectionFactory
 					"When the cache mode is 'CHANNEL', the connection cache size cannot be configured.");
 		}
 		initCacheWaterMarks();
-		if (this.publisherConnectionFactory != null) {
-			this.publisherConnectionFactory.afterPropertiesSet();
+		if (this.defaultPublisherFactory) {
+			((CachingConnectionFactory) getPublisherConnectionFactory()).afterPropertiesSet();
 		}
 	}
 
@@ -901,8 +890,8 @@ public class CachingConnectionFactory extends AbstractConnectionFactory
 			this.channelHighWaterMarks.values().forEach(count -> count.set(0));
 			this.connectionHighWaterMark.set(0);
 		}
-		if (this.publisherConnectionFactory != null) {
-			this.publisherConnectionFactory.resetConnection();
+		if (this.defaultPublisherFactory) {
+			((CachingConnectionFactory) getPublisherConnectionFactory()).resetConnection();
 		}
 	}
 
@@ -995,8 +984,8 @@ public class CachingConnectionFactory extends AbstractConnectionFactory
 	 */
 	@ManagedAttribute
 	public Properties getPublisherConnectionFactoryCacheProperties() {
-		if (this.publisherConnectionFactory != null) {
-			return this.publisherConnectionFactory.getCacheProperties();
+		if (this.defaultPublisherFactory) {
+			return ((CachingConnectionFactory) getPublisherConnectionFactory()).getCacheProperties();
 		}
 		return new Properties();
 	}

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

@@ -36,6 +36,7 @@ import org.springframework.amqp.AmqpException;
 import org.springframework.amqp.rabbit.support.RabbitExceptionTranslator;
 import org.springframework.aop.framework.ProxyFactory;
 import org.springframework.aop.support.NameMatchMethodPointcutAdvisor;
+import org.springframework.lang.Nullable;
 import org.springframework.util.Assert;
 
 import com.rabbitmq.client.Channel;
@@ -60,6 +61,8 @@ public class PooledChannelConnectionFactory extends AbstractConnectionFactory im
 
 	private BiConsumer<GenericObjectPool<Channel>, Boolean> poolConfigurer = (pool, tx) -> { };
 
+	private boolean defaultPublisherFactory = true;
+
 	/**
 	 * Construct an instance.
 	 * @param rabbitConnectionFactory the rabbitmq connection factory.
@@ -78,6 +81,15 @@ public class PooledChannelConnectionFactory extends AbstractConnectionFactory im
 		if (!isPublisher) {
 			setPublisherConnectionFactory(new PooledChannelConnectionFactory(rabbitConnectionFactory, true));
 		}
+		else {
+			this.defaultPublisherFactory = false;
+		}
+	}
+
+	@Override
+	public void setPublisherConnectionFactory(@Nullable AbstractConnectionFactory publisherConnectionFactory) {
+		super.setPublisherConnectionFactory(publisherConnectionFactory);
+		this.defaultPublisherFactory = false;
 	}
 
 	/**
@@ -88,6 +100,9 @@ public class PooledChannelConnectionFactory extends AbstractConnectionFactory im
 	public void setPoolConfigurer(BiConsumer<GenericObjectPool<Channel>, Boolean> poolConfigurer) {
 		Assert.notNull(poolConfigurer, "'poolConfigurer' cannot be null");
 		this.poolConfigurer = poolConfigurer; // NOSONAR - sync inconsistency
+		if (this.defaultPublisherFactory) {
+			((PooledChannelConnectionFactory) getPublisherConnectionFactory()).setPoolConfigurer(poolConfigurer);
+		}
 	}
 
 	@Override
@@ -101,6 +116,10 @@ public class PooledChannelConnectionFactory extends AbstractConnectionFactory im
 	 */
 	public void setSimplePublisherConfirms(boolean simplePublisherConfirms) {
 		this.simplePublisherConfirms = simplePublisherConfirms;
+		if (this.defaultPublisherFactory) {
+			((ThreadChannelConnectionFactory) getPublisherConnectionFactory())
+				.setSimplePublisherConfirms(simplePublisherConfirms);
+		}
 	}
 
 	@Override
@@ -117,7 +136,7 @@ public class PooledChannelConnectionFactory extends AbstractConnectionFactory im
 		if (this.connection == null || !this.connection.isOpen()) {
 			Connection bareConnection = createBareConnection(); // NOSONAR - see destroy()
 			this.connection = new ConnectionWrapper(bareConnection.getDelegate(), getCloseTimeout(), // NOSONAR
-					this.simplePublisherConfirms, this.poolConfigurer, getChannelListener());
+					this.simplePublisherConfirms, this.poolConfigurer, getChannelListener()); // NOSONAR
 			getConnectionListener().onCreate(this.connection);
 		}
 		return this.connection;

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

@@ -27,6 +27,7 @@ import org.springframework.amqp.AmqpException;
 import org.springframework.amqp.rabbit.support.RabbitExceptionTranslator;
 import org.springframework.aop.framework.ProxyFactory;
 import org.springframework.aop.support.NameMatchMethodPointcutAdvisor;
+import org.springframework.lang.Nullable;
 
 import com.rabbitmq.client.Channel;
 import com.rabbitmq.client.ConnectionFactory;
@@ -47,6 +48,8 @@ public class ThreadChannelConnectionFactory extends AbstractConnectionFactory im
 
 	private boolean simplePublisherConfirms;
 
+	private boolean defaultPublisherFactory = true;
+
 	/**
 	 * Construct an instance.
 	 * @param rabbitConnectionFactory the rabbitmq connection factory.
@@ -65,6 +68,15 @@ public class ThreadChannelConnectionFactory extends AbstractConnectionFactory im
 		if (!isPublisher) {
 			setPublisherConnectionFactory(new ThreadChannelConnectionFactory(rabbitConnectionFactory, true));
 		}
+		else {
+			this.defaultPublisherFactory = false;
+		}
+	}
+
+	@Override
+	public void setPublisherConnectionFactory(@Nullable AbstractConnectionFactory publisherConnectionFactory) {
+		super.setPublisherConnectionFactory(publisherConnectionFactory);
+		this.defaultPublisherFactory = false;
 	}
 
 	@Override
@@ -78,6 +90,10 @@ public class ThreadChannelConnectionFactory extends AbstractConnectionFactory im
 	 */
 	public void setSimplePublisherConfirms(boolean simplePublisherConfirms) {
 		this.simplePublisherConfirms = simplePublisherConfirms;
+		if (this.defaultPublisherFactory) {
+			((ThreadChannelConnectionFactory) getPublisherConnectionFactory())
+				.setSimplePublisherConfirms(simplePublisherConfirms);
+		}
 	}
 
 	@Override

spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/adapter/MessagingMessageListenerAdapter.java

@@ -324,7 +324,7 @@ public class MessagingMessageListenerAdapter extends AbstractAdaptableMessageLis
 			return extractMessage(message);
 		}
 
-		private Type determineInferredType() {
+		private Type determineInferredType() { // NOSONAR - complexity
 			if (this.method == null) {
 				return null;
 			}
@@ -339,14 +339,12 @@ public class MessagingMessageListenerAdapter extends AbstractAdaptableMessageLis
 				 */
 				boolean isHeader = methodParameter.hasParameterAnnotation(Header.class);
 				boolean isPayload = methodParameter.hasParameterAnnotation(Payload.class);
-				if (isHeader && isPayload) {
-					if (MessagingMessageListenerAdapter.this.logger.isWarnEnabled()) {
-						MessagingMessageListenerAdapter.this.logger.warn(this.method.getName()
-							+ ": Cannot annotate a parameter with both @Header and @Payload; "
-							+ "ignored for payload conversion");
-					}
+				if (isHeader && isPayload && MessagingMessageListenerAdapter.this.logger.isWarnEnabled()) {
+					MessagingMessageListenerAdapter.this.logger.warn(this.method.getName()
+						+ ": Cannot annotate a parameter with both @Header and @Payload; "
+						+ "ignored for payload conversion");
 				}
-				if (isEligibleParameter(methodParameter)
+				if (isEligibleParameter(methodParameter) // NOSONAR
 						&& (!isHeader || isPayload) && !(isHeader && isPayload)) {
 
 					if (genericParameterType == null) {

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

@@ -1795,6 +1795,16 @@ public class CachingConnectionFactoryTests extends AbstractConnectionFactoryTest
 		assertThat(closeLatch.await(10, TimeUnit.SECONDS)).isTrue();
 		assertThat(rejected.get()).isFalse();
 		closeExec.shutdownNow();
+
+		ccf.setChannelCacheSize(42);
+		assertThat(ccf.getChannelCacheSize()).isEqualTo(42);
+		assertThat(((CachingConnectionFactory) ccf.getPublisherConnectionFactory()).getChannelCacheSize())
+				.isEqualTo(42);
+		ccf.setPublisherConnectionFactory(new CachingConnectionFactory());
+		ccf.setChannelCacheSize(42);
+		assertThat(ccf.getChannelCacheSize()).isEqualTo(42);
+		assertThat(((CachingConnectionFactory) ccf.getPublisherConnectionFactory()).getChannelCacheSize())
+				.isEqualTo(25);
 	}
 
 	@Test

src/reference/asciidoc/amqp.adoc

@@ -219,6 +219,9 @@ The `CachingConnectionFactory` should be used if you want to use correlated publ
 
 Simple publisher confirmations are supported by all three factories.
 
+When configuring a `RabbitTemplate` to use a <<separate-connection, separate connection>>, you can now, starting with version 2.3.2, configure the publishing connection factory to be a different type.
+By default, the publishing factory is the same type and any properties set on the main factory are also propagated to the publishing factory.
+
 ====== `PooledChannelConnectionFactory`
 
 This factory manages a single connection and two pools of channels, based on the Apache Pool2.
@@ -1327,7 +1330,7 @@ The second obtains the `username` property from a connection factory bean in the
 
 Starting with version 2.0.2, you can set the `usePublisherConnection` property to `true` to use a different connection to that used by listener containers, when possible.
 This is to avoid consumers being blocked when a producer is blocked for any reason.
-The `CachingConnectionFactory` now maintains a second internal connection factory for this purpose.
+The connection factories maintain a second internal connection factory for this purpose; by default it is the same type as the main factory, but can be set explicity if you wish to use a different factory type for publishing.
 If the rabbit template is running in a transaction started by the listener container, the container's channel is used, regardless of this setting.
 
 IMPORTANT: In general, you should not use a `RabbitAdmin` with a template that has this set to `true`.