diff --git a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/annotation/MultiRabbitListenerAnnotationBeanPostProcessor.java b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/annotation/MultiRabbitListenerAnnotationBeanPostProcessor.java index 9b9e5446..fbbee0bc 100644 --- a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/annotation/MultiRabbitListenerAnnotationBeanPostProcessor.java +++ b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/annotation/MultiRabbitListenerAnnotationBeanPostProcessor.java @@ -20,7 +20,6 @@ import java.lang.reflect.Method; import java.util.Collection; import org.springframework.amqp.core.Declarable; -import org.springframework.amqp.rabbit.core.RabbitAdmin; import org.springframework.util.StringUtils; /** @@ -28,10 +27,12 @@ import org.springframework.util.StringUtils; * proper RabbitAdmin to the beans of Exchanges, Queues, and Bindings after they are * created. *

- * This processing restricts the {@link RabbitAdmin} according to the related + * This processing restricts the {@link org.springframework.amqp.rabbit.core.RabbitAdmin} according to the related * configuration, preventing the server from automatic binding non-related structures. * * @author Wander Costa + * + * @since 2.3 */ public class MultiRabbitListenerAnnotationBeanPostProcessor extends RabbitListenerAnnotationBeanPostProcessor { @@ -59,7 +60,6 @@ public class MultiRabbitListenerAnnotationBeanPostProcessor extends RabbitListen /** * Resolves the name of the RabbitAdmin bean based on the RabbitListener, or falls back to * the default RabbitAdmin name provided by MultiRabbit. - * * @param rabbitListener The RabbitListener to process the name from. * @return The name of the RabbitAdmin bean. */ @@ -74,4 +74,5 @@ public class MultiRabbitListenerAnnotationBeanPostProcessor extends RabbitListen } return admin; } + } 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 0307a653..43a5f001 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 @@ -252,7 +252,7 @@ public abstract class AbstractConnectionFactory implements ConnectionFactory, Di * @since 2.1.15 */ public void setAddressResolver(AddressResolver addressResolver) { - this.addressResolver = addressResolver; + this.addressResolver = addressResolver; // NOSONAR - sync inconsistency } /** @@ -507,7 +507,7 @@ public abstract class AbstractConnectionFactory implements ConnectionFactory, Di */ public void setAddressShuffleMode(AddressShuffleMode addressShuffleMode) { Assert.notNull(addressShuffleMode, "'addressShuffleMode' cannot be null"); - this.addressShuffleMode = addressShuffleMode; + this.addressShuffleMode = addressShuffleMode; // NOSONAR - sync inconsistency } public boolean hasPublisherConnectionFactory() { diff --git a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/ConnectionFactoryContextWrapper.java b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/ConnectionFactoryContextWrapper.java index 5643bd00..57b5c2da 100644 --- a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/ConnectionFactoryContextWrapper.java +++ b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/ConnectionFactoryContextWrapper.java @@ -18,70 +18,73 @@ package org.springframework.amqp.rabbit.connection; import java.util.concurrent.Callable; +import org.springframework.lang.Nullable; +import org.springframework.util.Assert; import org.springframework.util.StringUtils; /** * Helper class to handle {@link ConnectionFactory} context binding and unbinding when executing instructions. * * @author Wander Costa + * @author Artem Bilan + * + * @since 2.3 */ public class ConnectionFactoryContextWrapper { private final ConnectionFactory connectionFactory; public ConnectionFactoryContextWrapper(ConnectionFactory connectionFactory) { + Assert.notNull(connectionFactory, "'connectionFactory' must not be null"); this.connectionFactory = connectionFactory; } /** - * Executes a {@link Callable} binding to the default {@link ConnectionFactory} and finally unbinding it. - * + * Execute a {@link Callable} binding to the default {@link ConnectionFactory} and finally unbinding it. * @param callable the {@link Callable} object to be executed. * @param the return type. * @return the result of the {@link Callable}. - * @throws Exception when an Exception is thrown by the {@link Callable}. */ - public T call(final Callable callable) throws Exception { + public T call(final Callable callable) { return call(null, callable); } /** - * Executes a {@link Callable} binding the given {@link ConnectionFactory} and finally unbinding it. - * + * Execute a {@link Callable} binding the given {@link ConnectionFactory} and finally unbinding it. * @param contextName the name of the context. In null, empty or blank, default context is bound. * @param callable the {@link Callable} object to be executed. * @param the return type. * @return the result of the {@link Callable}. - * @throws Exception when an Exception is thrown by the {@link Callable}. */ - public T call(final String contextName, final Callable callable) throws Exception { + public T call(@Nullable String contextName, Callable callable) { try { bind(contextName); return callable.call(); } + catch (Exception ex) { + throw new IllegalStateException(ex); + } finally { unbind(contextName); } } /** - * Executes a {@link Runnable} binding to the default {@link ConnectionFactory} and finally unbinding it. - * + * Execute a {@link Runnable} binding to the default {@link ConnectionFactory} and finally unbinding it. * @param runnable the {@link Runnable} object to be executed. * @throws RuntimeException when a RuntimeException is thrown by the {@link Runnable}. */ - public void run(final Runnable runnable) { + public void run(Runnable runnable) { run(null, runnable); } /** - * Executes a {@link Runnable} binding the given {@link ConnectionFactory} and finally unbinding it. - * + * Execute a {@link Runnable} binding the given {@link ConnectionFactory} and finally unbinding it. * @param contextName the name of the context. In null, empty or blank, default context is bound. * @param runnable the {@link Runnable} object to be executed. * @throws RuntimeException when a RuntimeException is thrown by the {@link Runnable}. */ - public void run(final String contextName, final Runnable runnable) { + public void run(@Nullable String contextName, Runnable runnable) { try { bind(contextName); runnable.run(); @@ -92,22 +95,20 @@ public class ConnectionFactoryContextWrapper { } /** - * Binds the context. - * + * Bind the context. * @param contextName the name of the context for the connection factory. */ - private void bind(final String contextName) { + private void bind(@Nullable String contextName) { if (StringUtils.hasText(contextName)) { SimpleResourceHolder.bind(this.connectionFactory, contextName); } } /** - * Unbinds the context. - * + * Unbind the context. * @param contextName the name of the context for the connection factory. */ - private void unbind(final String contextName) { + private void unbind(@Nullable String contextName) { if (StringUtils.hasText(contextName)) { SimpleResourceHolder.unbind(this.connectionFactory); } 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 056ecee2..14e5aecb 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 @@ -22,8 +22,6 @@ import java.util.function.BiConsumer; import org.aopalliance.aop.Advice; import org.aopalliance.intercept.MethodInterceptor; -import org.aopalliance.intercept.MethodInvocation; -import org.apache.commons.logging.Log; import org.apache.commons.pool2.ObjectPool; import org.apache.commons.pool2.PooledObject; import org.apache.commons.pool2.PooledObjectFactory; @@ -46,6 +44,7 @@ import com.rabbitmq.client.ConnectionFactory; * a callback. * * @author Gary Russell + * * @since 2.3 * */ @@ -59,7 +58,6 @@ public class PooledChannelConnectionFactory extends AbstractConnectionFactory { /** * Construct an instance. - * * @param rabbitConnectionFactory the rabbitmq connection factory. */ public PooledChannelConnectionFactory(ConnectionFactory rabbitConnectionFactory) { @@ -68,7 +66,6 @@ public class PooledChannelConnectionFactory extends AbstractConnectionFactory { /** * Construct an instance. - * * @param rabbitConnectionFactory the rabbitmq connection factory. * @param isPublisher true if we are creating a publisher connection factory. */ @@ -86,7 +83,7 @@ public class PooledChannelConnectionFactory extends AbstractConnectionFactory { */ public void setPoolConfigurer(BiConsumer, Boolean> poolConfigurer) { Assert.notNull(poolConfigurer, "'poolConfigurer' cannot be null"); - this.poolConfigurer = poolConfigurer; + this.poolConfigurer = poolConfigurer; // NOSONAR - sync inconsistency } @Override @@ -105,9 +102,9 @@ public class PooledChannelConnectionFactory extends AbstractConnectionFactory { @Override public synchronized Connection createConnection() throws AmqpException { if (this.connection == null || !this.connection.isOpen()) { - Connection bareConnection = createBareConnection(); + Connection bareConnection = createBareConnection(); // NOSONAR - see destroy() this.connection = new ConnectionWrapper(bareConnection.getDelegate(), getCloseTimeout(), - this.simplePublisherConfirms, this.logger, this.poolConfigurer); + this.simplePublisherConfirms, this.poolConfigurer); } return this.connection; } @@ -123,8 +120,6 @@ public class PooledChannelConnectionFactory extends AbstractConnectionFactory { private final static class ConnectionWrapper extends SimpleConnection { - private final Log logger; - private final ObjectPool channels; private final ObjectPool txChannels; @@ -132,7 +127,7 @@ public class PooledChannelConnectionFactory extends AbstractConnectionFactory { private final boolean simplePublisherConfirms; ConnectionWrapper(com.rabbitmq.client.Connection delegate, int closeTimeout, boolean simplePublisherConfirms, - Log logger, BiConsumer, Boolean> configurer) { + BiConsumer, Boolean> configurer) { super(delegate, closeTimeout); GenericObjectPool pool = new GenericObjectPool<>(new ChannelFactory()); @@ -142,7 +137,6 @@ public class PooledChannelConnectionFactory extends AbstractConnectionFactory { configurer.accept(pool, true); this.txChannels = pool; this.simplePublisherConfirms = simplePublisherConfirms; - this.logger = logger; } @Override @@ -158,20 +152,16 @@ public class PooledChannelConnectionFactory extends AbstractConnectionFactory { private Channel createProxy(Channel channel, boolean transacted) { ProxyFactory pf = new ProxyFactory(channel); AtomicReference proxy = new AtomicReference<>(); - Advice advice = new MethodInterceptor() { - - @Override - public Object invoke(MethodInvocation invocation) throws Throwable { - if (transacted) { - ConnectionWrapper.this.txChannels.returnObject(proxy.get()); - } - else { - ConnectionWrapper.this.channels.returnObject(proxy.get()); - } - return null; - } - - }; + Advice advice = + (MethodInterceptor) invocation -> { + if (transacted) { + ConnectionWrapper.this.txChannels.returnObject(proxy.get()); + } + else { + ConnectionWrapper.this.channels.returnObject(proxy.get()); + } + return null; + }; NameMatchMethodPointcutAdvisor advisor = new NameMatchMethodPointcutAdvisor(advice); advisor.addMethodName("close"); pf.addAdvisor(advisor); @@ -192,7 +182,7 @@ public class PooledChannelConnectionFactory extends AbstractConnectionFactory { private class ChannelFactory implements PooledObjectFactory { @Override - public PooledObject makeObject() throws Exception { + public PooledObject makeObject() { Channel channel = ConnectionWrapper.super.createChannel(false); if (ConnectionWrapper.this.simplePublisherConfirms) { try { @@ -228,7 +218,7 @@ public class PooledChannelConnectionFactory extends AbstractConnectionFactory { private final class TxChannelFactory extends ChannelFactory { @Override - public PooledObject makeObject() throws Exception { + public PooledObject makeObject() { Channel channel = ConnectionWrapper.super.createChannel(true); try { channel.txSelect(); 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 19bb0094..5b3244e8 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 @@ -21,7 +21,6 @@ import java.util.concurrent.TimeoutException; 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; @@ -37,6 +36,7 @@ import com.rabbitmq.client.ConnectionFactory; * {@link #closeThreadChannel()}. * * @author Gary Russell + * * @since 2.3 * */ @@ -48,7 +48,6 @@ public class ThreadChannelConnectionFactory extends AbstractConnectionFactory { /** * Construct an instance. - * * @param rabbitConnectionFactory the rabbitmq connection factory. */ public ThreadChannelConnectionFactory(ConnectionFactory rabbitConnectionFactory) { @@ -57,7 +56,6 @@ public class ThreadChannelConnectionFactory extends AbstractConnectionFactory { /** * Construct an instance. - * * @param rabbitConnectionFactory the rabbitmq connection factory. * @param isPublisher true if we are creating a publisher connection factory. */ @@ -84,7 +82,7 @@ public class ThreadChannelConnectionFactory extends AbstractConnectionFactory { @Override public synchronized Connection createConnection() throws AmqpException { if (this.connection == null || !this.connection.isOpen()) { - Connection bareConnection = createBareConnection(); + Connection bareConnection = createBareConnection(); // NOSONAR - see destroy() this.connection = new ConnectionWrapper(bareConnection.getDelegate(), getCloseTimeout()); } return this.connection; @@ -156,19 +154,15 @@ public class ThreadChannelConnectionFactory extends AbstractConnectionFactory { private Channel createProxy(Channel channel) { ProxyFactory pf = new ProxyFactory(channel); - Advice advice = new MethodInterceptor() { - - @Override - public Object invoke(MethodInvocation invocation) throws Throwable { - if (ConnectionWrapper.this.channels.get() == null) { - return invocation.proceed(); - } - else { - return null; - } - } - - }; + Advice advice = + (MethodInterceptor) invocation -> { + if (ConnectionWrapper.this.channels.get() == null) { + return invocation.proceed(); + } + else { + return null; + } + }; NameMatchMethodPointcutAdvisor advisor = new NameMatchMethodPointcutAdvisor(advice); advisor.addMethodName("close"); pf.addAdvisor(advisor);