From 99ccb1bfb5459426a0694ffe7b61d6eed3d03710 Mon Sep 17 00:00:00 2001 From: Gary Russell Date: Mon, 30 Nov 2020 13:41:12 -0500 Subject: [PATCH] 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. --- .../connection/AbstractConnectionFactory.java | 41 +++++++++- .../connection/CachingConnectionFactory.java | 81 ++++++++----------- .../PooledChannelConnectionFactory.java | 21 ++++- .../ThreadChannelConnectionFactory.java | 16 ++++ .../MessagingMessageListenerAdapter.java | 14 ++-- .../CachingConnectionFactoryTests.java | 10 +++ src/reference/asciidoc/amqp.adoc | 5 +- 7 files changed, 128 insertions(+), 60 deletions(-) diff --git a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/AbstractConnectionFactory.java b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/AbstractConnectionFactory.java index b35ec95c..f4fd8f14 100644 --- a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/AbstractConnectionFactory.java +++ b/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; diff --git a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/CachingConnectionFactory.java b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/CachingConnectionFactory.java index 6c155a65..9412619e 100644 --- a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/CachingConnectionFactory.java +++ b/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(); } diff --git a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/PooledChannelConnectionFactory.java b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/PooledChannelConnectionFactory.java index 35115a28..3e4ef0fa 100644 --- a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/PooledChannelConnectionFactory.java +++ b/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, 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, 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; diff --git a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/ThreadChannelConnectionFactory.java b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/ThreadChannelConnectionFactory.java index bac34811..f527ecb8 100644 --- a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/ThreadChannelConnectionFactory.java +++ b/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 diff --git a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/adapter/MessagingMessageListenerAdapter.java b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/adapter/MessagingMessageListenerAdapter.java index a614715e..d58efe5d 100644 --- a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/adapter/MessagingMessageListenerAdapter.java +++ b/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) { diff --git a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/connection/CachingConnectionFactoryTests.java b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/connection/CachingConnectionFactoryTests.java index f15a404a..380994f8 100644 --- a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/connection/CachingConnectionFactoryTests.java +++ b/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 diff --git a/src/reference/asciidoc/amqp.adoc b/src/reference/asciidoc/amqp.adoc index b230ebc9..9b89d226 100644 --- a/src/reference/asciidoc/amqp.adoc +++ b/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 <>, 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`.