diff --git a/spring-rabbit-junit/src/main/java/org/springframework/amqp/rabbit/junit/BrokerRunning.java b/spring-rabbit-junit/src/main/java/org/springframework/amqp/rabbit/junit/BrokerRunning.java index 2220fcd1..51c1af49 100644 --- a/spring-rabbit-junit/src/main/java/org/springframework/amqp/rabbit/junit/BrokerRunning.java +++ b/spring-rabbit-junit/src/main/java/org/springframework/amqp/rabbit/junit/BrokerRunning.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2017 the original author or authors. + * Copyright 2002-2018 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. @@ -99,6 +99,8 @@ public final class BrokerRunning extends TestWatcher { private static final String DEFAULT_QUEUE_NAME = BrokerRunning.class.getName(); + private static final String GUEST = "guest"; + private static final Log logger = LogFactory.getLog(BrokerRunning.class); // Static so that we only test once on failure: speeds up test suite @@ -128,13 +130,13 @@ public final class BrokerRunning extends TestWatcher { private ConnectionFactory connectionFactory; - private String user = fromEnvironment(BROKER_USER, "guest"); + private String user = fromEnvironment(BROKER_USER, GUEST); - private String password = fromEnvironment(BROKER_PW, "guest"); + private String password = fromEnvironment(BROKER_PW, GUEST); - private String adminUser = fromEnvironment(BROKER_ADMIN_USER, "guest"); + private String adminUser = fromEnvironment(BROKER_ADMIN_USER, GUEST); - private String adminPassword = fromEnvironment(BROKER_ADMIN_PW, "guest"); + private String adminPassword = fromEnvironment(BROKER_ADMIN_PW, GUEST); private String fromEnvironment(String key, String defaultValue) { String environmentValue = environmentOverrides.get(key); @@ -363,13 +365,11 @@ public final class BrokerRunning extends TestWatcher { Assume.assumeTrue(brokerOffline.get(this.port)); } - ConnectionFactory connectionFactory = getConnectionFactory(); - Connection connection = null; // NOSONAR (closeResources()) Channel channel = null; try { - connection = getConnection(connectionFactory); + connection = getConnection(getConnectionFactory()); channel = createQueues(connection); } catch (Exception e) { @@ -391,7 +391,7 @@ public final class BrokerRunning extends TestWatcher { return super.apply(base, description); } - public void isUp() throws Exception { + public void isUp() throws IOException, TimeoutException, URISyntaxException { Connection connection = getConnectionFactory().newConnection(); // NOSONAR - closeResources() Channel channel = null; try { @@ -436,7 +436,7 @@ public final class BrokerRunning extends TestWatcher { if (this.management) { Client client = new Client(getAdminUri(), this.adminUser, this.adminPassword); if (!client.alivenessTest("/")) { - throw new RuntimeException("Aliveness test failed for localhost:15672 guest/quest; " + throw new BrokerNotAliveException("Aliveness test failed for localhost:15672 guest/quest; " + "management not available"); } } @@ -484,12 +484,11 @@ public final class BrokerRunning extends TestWatcher { queuesToRemove.addAll(Arrays.asList(additionalQueues)); } logger.debug("deleting test queues: " + queuesToRemove); - ConnectionFactory connectionFactory = getConnectionFactory(); Connection connection = null; // NOSONAR (closeResources()) Channel channel = null; try { - connection = getConnection(connectionFactory); + connection = getConnection(getConnectionFactory()); connection.setId(generateId() + ".queueDelete"); channel = connection.createChannel(); @@ -510,12 +509,11 @@ public final class BrokerRunning extends TestWatcher { * @param queues the queues to delete. */ public void deleteQueues(String... queues) { - ConnectionFactory connectionFactory = getConnectionFactory(); Connection connection = null; // NOSONAR (closeResources()) Channel channel = null; try { - connection = getConnection(connectionFactory); + connection = getConnection(getConnectionFactory()); connection.setId(generateId() + ".queueDelete"); channel = connection.createChannel(); @@ -536,12 +534,11 @@ public final class BrokerRunning extends TestWatcher { * @param exchanges the exchanges to delete. */ public void deleteExchanges(String... exchanges) { - ConnectionFactory connectionFactory = getConnectionFactory(); Connection connection = null; // NOSONAR (closeResources()) Channel channel = null; try { - connection = getConnection(connectionFactory); + connection = getConnection(getConnectionFactory()); connection.setId(generateId() + ".exchangeDelete"); channel = connection.createChannel(); @@ -614,4 +611,14 @@ public final class BrokerRunning extends TestWatcher { } } + public static class BrokerNotAliveException extends RuntimeException { + + private static final long serialVersionUID = 1L; + + BrokerNotAliveException(String message) { + super(message); + } + + } + } diff --git a/spring-rabbit-junit/src/main/java/org/springframework/amqp/rabbit/junit/RabbitAvailableCondition.java b/spring-rabbit-junit/src/main/java/org/springframework/amqp/rabbit/junit/RabbitAvailableCondition.java index caf4e648..1a9d0a35 100644 --- a/spring-rabbit-junit/src/main/java/org/springframework/amqp/rabbit/junit/RabbitAvailableCondition.java +++ b/spring-rabbit-junit/src/main/java/org/springframework/amqp/rabbit/junit/RabbitAvailableCondition.java @@ -45,6 +45,8 @@ import com.rabbitmq.client.ConnectionFactory; */ public class RabbitAvailableCondition implements ExecutionCondition, AfterAllCallback, ParameterResolver { + private static final String BROKER_RUNNING_BEAN = "brokerRunning"; + private static final ConditionEvaluationResult ENABLED = ConditionEvaluationResult.enabled( "@RabbitAvailable is not present"); @@ -57,7 +59,7 @@ public class RabbitAvailableCondition implements ExecutionCondition, AfterAllCal if (rabbit != null) { try { String[] queues = rabbit.queues(); - BrokerRunning brokerRunning = getStore(context).get("brokerRunning", BrokerRunning.class); + BrokerRunning brokerRunning = getStore(context).get(BROKER_RUNNING_BEAN, BrokerRunning.class); if (brokerRunning == null) { if (rabbit.management()) { brokerRunning = BrokerRunning.isBrokerAndManagementRunningWithEmptyQueues(queues); @@ -69,13 +71,13 @@ public class RabbitAvailableCondition implements ExecutionCondition, AfterAllCal brokerRunning.isUp(); brokerRunningHolder.set(brokerRunning); Store store = getStore(context); - store.put("brokerRunning", brokerRunning); + store.put(BROKER_RUNNING_BEAN, brokerRunning); store.put("queuesToDelete", queues); return ConditionEvaluationResult.enabled("RabbitMQ is available"); } catch (Exception e) { if (BrokerRunning.fatal()) { - throw new IllegalStateException("Required RabbitMQ is not available"); + throw new IllegalStateException("Required RabbitMQ is not available", e); } return ConditionEvaluationResult.disabled("RabbitMQ is not available"); } @@ -84,10 +86,10 @@ public class RabbitAvailableCondition implements ExecutionCondition, AfterAllCal } @Override - public void afterAll(ExtensionContext context) throws Exception { + public void afterAll(ExtensionContext context) { brokerRunningHolder.remove(); Store store = getStore(context); - BrokerRunning brokerRunning = store.remove("brokerRunning", BrokerRunning.class); + BrokerRunning brokerRunning = store.remove(BROKER_RUNNING_BEAN, BrokerRunning.class); if (brokerRunning != null) { brokerRunning.removeTestQueues(); } @@ -105,9 +107,9 @@ public class RabbitAvailableCondition implements ExecutionCondition, AfterAllCal throws ParameterResolutionException { // in parent for method injection, Composite key causes a store miss BrokerRunning brokerRunning = - getParentStore(context).get("brokerRunning", BrokerRunning.class) == null - ? getStore(context).get("brokerRunning", BrokerRunning.class) - : getParentStore(context).get("brokerRunning", BrokerRunning.class); + getParentStore(context).get(BROKER_RUNNING_BEAN, BrokerRunning.class) == null + ? getStore(context).get(BROKER_RUNNING_BEAN, BrokerRunning.class) + : getParentStore(context).get(BROKER_RUNNING_BEAN, BrokerRunning.class); Assert.state(brokerRunning != null, "Could not find brokerRunning instance"); Class type = parameterContext.getParameter().getType(); return type.equals(ConnectionFactory.class) ? brokerRunning.getConnectionFactory() diff --git a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/config/QueueParser.java b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/config/QueueParser.java index af677c28..c483f30c 100644 --- a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/config/QueueParser.java +++ b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/config/QueueParser.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2017 the original author or authors. + * Copyright 2002-2018 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. @@ -112,6 +112,13 @@ public class QueueParser extends AbstractSingleBeanDefinitionParser { } + parseArguments(element, parserContext, builder); + + NamespaceUtils.parseDeclarationControls(element, builder); + CURRENT_ELEMENT.set(element); + } + + private void parseArguments(Element element, ParserContext parserContext, BeanDefinitionBuilder builder) { String queueArguments = element.getAttribute(ARGUMENTS); Element argumentsElement = DomUtils.getChildElementByTagName(element, ARGUMENTS); @@ -141,9 +148,6 @@ public class QueueParser extends AbstractSingleBeanDefinitionParser { if (StringUtils.hasText(queueArguments)) { builder.addConstructorArgReference(queueArguments); } - - NamespaceUtils.parseDeclarationControls(element, builder); - CURRENT_ELEMENT.set(element); } private boolean attributeHasIllegalOverride(Element element, String name, String allowed) { diff --git a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/ConnectionFactoryUtils.java b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/ConnectionFactoryUtils.java index 5e37d120..8d347afc 100644 --- a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/ConnectionFactoryUtils.java +++ b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/ConnectionFactoryUtils.java @@ -162,7 +162,7 @@ public final class ConnectionFactoryUtils { } } - public static void releaseResources(RabbitResourceHolder resourceHolder) { + public static void releaseResources(@Nullable RabbitResourceHolder resourceHolder) { if (resourceHolder == null || resourceHolder.isSynchronizedWithTransaction()) { return; } diff --git a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/RabbitResourceHolder.java b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/RabbitResourceHolder.java index 0a588e50..cc2a0712 100644 --- a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/RabbitResourceHolder.java +++ b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/RabbitResourceHolder.java @@ -129,12 +129,12 @@ public class RabbitResourceHolder extends ResourceHolderSupport { if (!this.channels.contains(channel)) { this.channels.add(channel); if (connection != null) { - List channels = this.channelsPerConnection.get(connection); - if (channels == null) { - channels = new LinkedList(); - this.channelsPerConnection.put(connection, channels); + List channelsForConnection = this.channelsPerConnection.get(connection); + if (channelsForConnection == null) { + channelsForConnection = new LinkedList(); + this.channelsPerConnection.put(connection, channelsForConnection); } - channels.add(channel); + channelsForConnection.add(channel); } } } diff --git a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/RabbitUtils.java b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/RabbitUtils.java index abcc62da..d4f04e42 100644 --- a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/RabbitUtils.java +++ b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/RabbitUtils.java @@ -124,19 +124,7 @@ public abstract class RabbitUtils { } try { for (String consumerTag : consumerTags) { - try { - channel.basicCancel(consumerTag); - } - catch (IOException e) { - if (logger.isDebugEnabled()) { - logger.debug("Error performing 'basicCancel'", e); - } - } - catch (AlreadyClosedException e) { - if (logger.isTraceEnabled()) { - logger.trace(channel + " is already closed"); - } - } + cancel(channel, consumerTag); } if (transactional) { /* @@ -155,6 +143,22 @@ public abstract class RabbitUtils { } } + private static void cancel(Channel channel, String consumerTag) { + try { + channel.basicCancel(consumerTag); + } + catch (IOException e) { + if (logger.isDebugEnabled()) { + logger.debug("Error performing 'basicCancel'", e); + } + } + catch (AlreadyClosedException e) { + if (logger.isTraceEnabled()) { + logger.trace(channel + " is already closed"); + } + } + } + /** * Declare to that broker that a channel is going to be used transactionally, and convert exceptions that arise. * @param channel the channel to use @@ -229,7 +233,7 @@ public abstract class RabbitUtils { */ public static boolean isPassiveDeclarationChannelClose(ShutdownSignalException sig) { Method shutdownReason = sig.getReason(); - return shutdownReason instanceof AMQP.Channel.Close + return shutdownReason instanceof AMQP.Channel.Close // NOSONAR boolean complexity && AMQP.NOT_FOUND == ((AMQP.Channel.Close) shutdownReason).getReplyCode() && ((((AMQP.Channel.Close) shutdownReason).getClassId() == 40 // exchange || ((AMQP.Channel.Close) shutdownReason).getClassId() == 50) // queue @@ -245,7 +249,7 @@ public abstract class RabbitUtils { */ public static boolean isExclusiveUseChannelClose(ShutdownSignalException sig) { Method shutdownReason = sig.getReason(); - return shutdownReason instanceof AMQP.Channel.Close + return shutdownReason instanceof AMQP.Channel.Close // NOSONAR boolean complexity && AMQP.ACCESS_REFUSED == ((AMQP.Channel.Close) shutdownReason).getReplyCode() && ((AMQP.Channel.Close) shutdownReason).getClassId() == 60 // basic && ((AMQP.Channel.Close) shutdownReason).getMethodId() == 20 // consume diff --git a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/SimpleResourceHolder.java b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/SimpleResourceHolder.java index 678a671f..9db5b5a7 100644 --- a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/SimpleResourceHolder.java +++ b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/SimpleResourceHolder.java @@ -39,14 +39,19 @@ import org.springframework.util.Assert; * Use {@code TransactionSynchronizationManager} and {@code ResourceHolder} instead. * * @author Artem Bilan + * @author Gary Russell * @since 1.3 */ public final class SimpleResourceHolder { - private static final Log logger = LogFactory.getLog(SimpleResourceHolder.class); + private static final String FOR_KEY = "] for key ["; - private static final ThreadLocal> resources = new NamedThreadLocal>("Simple resources"); + private static final String BOUND_TO_THREAD = "] bound to thread ["; + private static final Log logger = LogFactory.getLog(SimpleResourceHolder.class); // NOSONAR lower case + + private static final ThreadLocal> resources = // NOSONAR lower case + new NamedThreadLocal>("Simple resources"); /** * Return all resources that are bound to the current thread. @@ -82,7 +87,8 @@ public final class SimpleResourceHolder { public static Object get(Object key) { Object value = doGet(key); if (value != null && logger.isTraceEnabled()) { - logger.trace("Retrieved value [" + value + "] for key [" + key + "] bound to thread [" + Thread.currentThread().getName() + "]"); + logger.trace("Retrieved value [" + value + FOR_KEY + key + BOUND_TO_THREAD + + Thread.currentThread().getName() + "]"); } return value; } @@ -116,10 +122,12 @@ public final class SimpleResourceHolder { resources.set(map); } Object oldValue = map.put(key, value); - Assert.isNull(oldValue, () -> "Already value [" + oldValue + "] for key [" + key + "] bound to thread [" + Thread.currentThread().getName() + "]"); + Assert.isNull(oldValue, () -> "Already value [" + oldValue + FOR_KEY + key + BOUND_TO_THREAD + + Thread.currentThread().getName() + "]"); if (logger.isTraceEnabled()) { - logger.trace("Bound value [" + value + "] for key [" + key + "] to thread [" + Thread.currentThread().getName() + "]"); + logger.trace( + "Bound value [" + value + FOR_KEY + key + "] to thread [" + Thread.currentThread().getName() + "]"); } } @@ -131,7 +139,8 @@ public final class SimpleResourceHolder { */ public static Object unbind(Object key) throws IllegalStateException { Object value = unbindIfPossible(key); - Assert.notNull(value, () -> "No value for key [" + key + "] bound to thread [" + Thread.currentThread().getName() + "]"); + Assert.notNull(value, + () -> "No value for key [" + key + BOUND_TO_THREAD + Thread.currentThread().getName() + "]"); return value; } @@ -153,7 +162,8 @@ public final class SimpleResourceHolder { } if (value != null && logger.isTraceEnabled()) { - logger.trace("Removed value [" + value + "] for key [" + key + "] from thread [" + Thread.currentThread().getName() + "]"); + logger.trace("Removed value [" + value + FOR_KEY + key + "] from thread [" + + Thread.currentThread().getName() + "]"); } return value; } diff --git a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/core/ChannelCallback.java b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/core/ChannelCallback.java index 8c0293ff..4b41c45a 100644 --- a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/core/ChannelCallback.java +++ b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/core/ChannelCallback.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2017 the original author or authors. + * Copyright 2002-2018 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,6 @@ public interface ChannelCallback { * @return The result. * @throws Exception Not sure what else Rabbit Throws */ - T doInRabbit(Channel channel) throws Exception; + T doInRabbit(Channel channel) throws Exception; // NOSONAR user code might throw anything; cannot change } diff --git a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/core/DeclarationExceptionEvent.java b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/core/DeclarationExceptionEvent.java index 0cd8dbe0..bf972957 100644 --- a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/core/DeclarationExceptionEvent.java +++ b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/core/DeclarationExceptionEvent.java @@ -30,7 +30,7 @@ public class DeclarationExceptionEvent extends RabbitAdminEvent { private static final long serialVersionUID = -8367796410619780665L; - private final Declarable declarable; + private final transient Declarable declarable; private final Throwable throwable; @@ -43,6 +43,7 @@ public class DeclarationExceptionEvent extends RabbitAdminEvent { /** * @return the declarable - if null, we were declaring a broker-named queue. */ + @Nullable public Declarable getDeclarable() { return this.declarable; } diff --git a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/core/RabbitAdmin.java b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/core/RabbitAdmin.java index 38441b2a..8fed706a 100644 --- a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/core/RabbitAdmin.java +++ b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/core/RabbitAdmin.java @@ -530,7 +530,7 @@ public class RabbitAdmin implements AmqpAdmin, ApplicationContextAware, Applicat * Declares all the exchanges, queues and bindings in the enclosing application context, if any. It should be safe * (but unnecessary) to call this method more than once. */ - @Override + @Override // NOSONAR complexity public void initialize() { if (this.applicationContext == null) { diff --git a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/core/RabbitTemplate.java b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/core/RabbitTemplate.java index 77becd37..0e1d6c3f 100644 --- a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/core/RabbitTemplate.java +++ b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/core/RabbitTemplate.java @@ -142,6 +142,8 @@ import com.rabbitmq.client.ShutdownListener; public class RabbitTemplate extends RabbitAccessor implements BeanFactoryAware, RabbitOperations, MessageListener, ListenerContainerAware, PublisherCallbackChannel.Listener, Lifecycle, BeanNameAware { + private static final String UNCHECKED = "unchecked"; + private static final String RETURN_CORRELATION_KEY = "spring_request_return_correlation"; /** Alias for amq.direct default exchange. */ @@ -1126,7 +1128,7 @@ public class RabbitTemplate extends RabbitAccessor implements BeanFactoryAware, } @Override - @SuppressWarnings("unchecked") + @SuppressWarnings(UNCHECKED) @Nullable public T receiveAndConvert(String queueName, long timeoutMillis, ParameterizedTypeReference type) throws AmqpException { Message response = timeoutMillis == 0 ? doReceiveNoWait(queueName) : receive(queueName, timeoutMillis); @@ -1143,7 +1145,7 @@ public class RabbitTemplate extends RabbitAccessor implements BeanFactoryAware, } @Override - @SuppressWarnings("unchecked") + @SuppressWarnings(UNCHECKED) @Nullable public boolean receiveAndReply(final String queueName, ReceiveAndReplyCallback callback) throws AmqpException { return receiveAndReply(queueName, callback, (ReplyToAddressCallback) this.defaultReplyToAddressCallback); @@ -1292,7 +1294,7 @@ public class RabbitTemplate extends RabbitAccessor implements BeanFactoryAware, } } - @SuppressWarnings("unchecked") + @SuppressWarnings(UNCHECKED) private Boolean sendReply(final ReceiveAndReplyCallback callback, final ReplyToAddressCallback replyToAddressCallback, Channel channel, Message receiveMessage) throws Exception { // NOSONAR TODO change to IOException in 2.2. @@ -1588,7 +1590,7 @@ public class RabbitTemplate extends RabbitAccessor implements BeanFactoryAware, } @Override - @SuppressWarnings("unchecked") + @SuppressWarnings(UNCHECKED) @Nullable public T convertSendAndReceiveAsType(final String exchange, final String routingKey, final Object message, @Nullable final MessagePostProcessor messagePostProcessor, @Nullable final CorrelationData correlationData, @@ -1916,7 +1918,7 @@ public class RabbitTemplate extends RabbitAccessor implements BeanFactoryAware, return execute(action, getConnectionFactory()); } - @SuppressWarnings("unchecked") + @SuppressWarnings(UNCHECKED) private T execute(final ChannelCallback action, final ConnectionFactory connectionFactory) { if (this.retryTemplate != null) { try { @@ -1937,7 +1939,7 @@ public class RabbitTemplate extends RabbitAccessor implements BeanFactoryAware, } - private T doExecute(ChannelCallback action, ConnectionFactory connectionFactory) { + private T doExecute(ChannelCallback action, ConnectionFactory connectionFactory) { // NOSONAR complexity Assert.notNull(action, "Callback object must not be null"); Channel channel = null; boolean invokeScope = false; @@ -2050,27 +2052,40 @@ public class RabbitTemplate extends RabbitAccessor implements BeanFactoryAware, throw e; } } + ConfirmListener listener = addConfirmListener(acks, nacks, channel); + try { + return action.doInRabbit(this); + } + finally { + cleanUpAfterAction(resourceHolder, connection, channel, listener); + } + } + + @Nullable + private ConfirmListener addConfirmListener(@Nullable com.rabbitmq.client.ConfirmCallback acks, + @Nullable com.rabbitmq.client.ConfirmCallback nacks, Channel channel) { ConfirmListener listener = null; if (acks != null && nacks != null && channel instanceof ChannelProxy && ((ChannelProxy) channel).isConfirmSelected()) { listener = channel.addConfirmListener(acks, nacks); } - try { - return action.doInRabbit(this); + return listener; + } + + private void cleanUpAfterAction(RabbitResourceHolder resourceHolder, Connection connection, Channel channel, + ConfirmListener listener) { + + if (listener != null) { + channel.removeConfirmListener(listener); } - finally { - if (listener != null) { - channel.removeConfirmListener(listener); - } - this.activeTemplateCallbacks.decrementAndGet(); - this.dedicatedChannels.remove(); - if (resourceHolder != null) { - ConnectionFactoryUtils.releaseResources(resourceHolder); - } - else { - RabbitUtils.closeChannel(channel); - RabbitUtils.closeConnection(connection); - } + this.activeTemplateCallbacks.decrementAndGet(); + this.dedicatedChannels.remove(); + if (resourceHolder != null) { + ConnectionFactoryUtils.releaseResources(resourceHolder); + } + else { + RabbitUtils.closeChannel(channel); + RabbitUtils.closeConnection(connection); } } @@ -2116,28 +2131,28 @@ public class RabbitTemplate extends RabbitAccessor implements BeanFactoryAware, * Send the given message to the specified exchange. * * @param channel The RabbitMQ Channel to operate within. - * @param exchange The name of the RabbitMQ exchange to send to. - * @param routingKey The routing key. + * @param exchangeArg The name of the RabbitMQ exchange to send to. + * @param routingKeyArg The routing key. * @param message The Message to send. * @param mandatory The mandatory flag. * @param correlationData The correlation data. * @throws Exception If thrown by RabbitMQ API methods */ - public void doSend(Channel channel, String exchange, String routingKey, Message message, + public void doSend(Channel channel, String exchangeArg, String routingKeyArg, Message message, // NOSONAR complexity boolean mandatory, @Nullable CorrelationData correlationData) throws Exception { // NOSONAR TODO: change to IOException in 2.2. - if (exchange == null) { - // try to send to configured exchange - exchange = this.exchange; + String exch = exchangeArg; + String rKey = routingKeyArg; + if (exch == null) { + exch = this.exchange; } - if (routingKey == null) { - // try to send to configured routing key - routingKey = this.routingKey; + if (rKey == null) { + rKey = this.routingKey; } if (logger.isDebugEnabled()) { logger.debug("Publishing message " + message - + "on exchange [" + exchange + "], routingKey = [" + routingKey + "]"); + + "on exchange [" + exch + "], routingKey = [" + rKey + "]"); } Message messageToUse = message; @@ -2157,7 +2172,7 @@ public class RabbitTemplate extends RabbitAccessor implements BeanFactoryAware, messageProperties.setUserId(userId); } } - sendToRabbit(channel, exchange, routingKey, mandatory, messageToUse); + sendToRabbit(channel, exch, rKey, mandatory, messageToUse); // Check if commit needed if (isChannelLocallyTransacted(channel)) { // Transacted channel created by this template -> commit. @@ -2172,12 +2187,13 @@ public class RabbitTemplate extends RabbitAccessor implements BeanFactoryAware, channel.basicPublish(exchange, routingKey, mandatory, convertedMessageProperties, message.getBody()); } - private void setupConfirm(Channel channel, Message message, @Nullable CorrelationData correlationData) { + private void setupConfirm(Channel channel, Message message, @Nullable CorrelationData correlationDataArg) { if ((this.publisherConfirms || this.confirmCallback != null) && channel instanceof PublisherCallbackChannel) { + PublisherCallbackChannel publisherCallbackChannel = (PublisherCallbackChannel) channel; - correlationData = this.correlationDataPostProcessor != null - ? this.correlationDataPostProcessor.postProcess(message, correlationData) - : correlationData; + CorrelationData correlationData = this.correlationDataPostProcessor != null + ? this.correlationDataPostProcessor.postProcess(message, correlationDataArg) + : correlationDataArg; long nextPublishSeqNo = channel.getNextPublishSeqNo(); message.getMessageProperties().setPublishSequenceNumber(nextPublishSeqNo); publisherCallbackChannel.addPendingConfirm(this, nextPublishSeqNo, @@ -2318,14 +2334,14 @@ public class RabbitTemplate extends RabbitAccessor implements BeanFactoryAware, byte[] body) throws IOException { - ReturnCallback returnCallback = this.returnCallback; - if (returnCallback == null) { + ReturnCallback callback = this.returnCallback; + if (callback == null) { Object messageTagHeader = properties.getHeaders().remove(RETURN_CORRELATION_KEY); if (messageTagHeader != null) { String messageTag = messageTagHeader.toString(); final PendingReply pendingReply = this.replyHolder.get(messageTag); if (pendingReply != null) { - returnCallback = (message, replyCode1, replyText1, exchange1, routingKey1) -> + callback = (message, replyCode1, replyText1, exchange1, routingKey1) -> pendingReply.returned(new AmqpMessageReturnedException("Message returned", message, replyCode1, replyText1, exchange1, routingKey1)); } @@ -2337,12 +2353,12 @@ public class RabbitTemplate extends RabbitAccessor implements BeanFactoryAware, logger.warn("Returned message but no callback available"); } } - if (returnCallback != null) { + if (callback != null) { properties.getHeaders().remove(PublisherCallbackChannel.RETURN_LISTENER_CORRELATION_KEY); MessageProperties messageProperties = this.messagePropertiesConverter.toMessageProperties( properties, null, this.encoding); Message returnedMessage = new Message(body, messageProperties); - returnCallback.returnedMessage(returnedMessage, + callback.returnedMessage(returnedMessage, replyCode, replyText, exchange, routingKey); } } @@ -2496,7 +2512,7 @@ public class RabbitTemplate extends RabbitAccessor implements BeanFactoryAware, return this.future.get(); } catch (ExecutionException e) { - throw RabbitExceptionTranslator.convertRabbitAccessException(e.getCause()); + throw RabbitExceptionTranslator.convertRabbitAccessException(e.getCause()); // NOSONAR lost stack trace } } @@ -2506,7 +2522,7 @@ public class RabbitTemplate extends RabbitAccessor implements BeanFactoryAware, return this.future.get(timeout, unit); } catch (ExecutionException e) { - throw RabbitExceptionTranslator.convertRabbitAccessException(e.getCause()); + throw RabbitExceptionTranslator.convertRabbitAccessException(e.getCause()); // NOSONAR lost stack trace } catch (TimeoutException e) { return null; diff --git a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/AbstractMessageListenerContainer.java b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/AbstractMessageListenerContainer.java index 930a2025..78eae818 100644 --- a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/AbstractMessageListenerContainer.java +++ b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/AbstractMessageListenerContainer.java @@ -1215,10 +1215,8 @@ public abstract class AbstractMessageListenerContainer extends RabbitAccessor * Register any invokers within this container. *

* Subclasses need to implement this method for their specific invoker management process. - * - * @throws Exception Any Exception. */ - protected abstract void doInitialize() throws Exception; + protected abstract void doInitialize(); /** * Close the registered invokers. @@ -1270,9 +1268,8 @@ public abstract class AbstractMessageListenerContainer extends RabbitAccessor /** * Start this container, and notify all invoker tasks. - * @throws Exception if thrown by Rabbit API methods */ - protected void doStart() throws Exception { + protected void doStart() { // Reschedule paused tasks, if any. synchronized (this.lifecycleMonitor) { this.active = true; @@ -1360,15 +1357,12 @@ public abstract class AbstractMessageListenerContainer extends RabbitAccessor /** * Execute the specified listener, committing or rolling back the transaction afterwards (if necessary). - * * @param channel the Rabbit Channel to operate on * @param messageIn the received Rabbit Message - * @throws Exception Any Exception. - * * @see #invokeListener * @see #handleListenerException */ - protected void executeListener(Channel channel, Message messageIn) throws Exception { + protected void executeListener(Channel channel, Message messageIn) { if (!isRunning()) { if (logger.isWarnEnabled()) { logger.warn("Rejecting received message because the listener container has been stopped: " + messageIn); @@ -1376,45 +1370,13 @@ public abstract class AbstractMessageListenerContainer extends RabbitAccessor throw new MessageRejectedWhileStoppingException(); } try { - Message message = messageIn; - if (this.afterReceivePostProcessors != null) { - for (MessagePostProcessor processor : this.afterReceivePostProcessors) { - message = processor.postProcessMessage(message); - if (message == null) { - throw new ImmediateAcknowledgeAmqpException( - "Message Post Processor returned 'null', discarding message"); - } - } - } - Object batchFormat = message.getMessageProperties().getHeaders().get(MessageProperties.SPRING_BATCH_FORMAT); - if (MessageProperties.BATCH_FORMAT_LENGTH_HEADER4.equals(batchFormat) && this.deBatchingEnabled) { - ByteBuffer byteBuffer = ByteBuffer.wrap(message.getBody()); - MessageProperties messageProperties = message.getMessageProperties(); - messageProperties.getHeaders().remove(MessageProperties.SPRING_BATCH_FORMAT); - while (byteBuffer.hasRemaining()) { - int length = byteBuffer.getInt(); - if (length < 0 || length > byteBuffer.remaining()) { - throw new ListenerExecutionFailedException("Bad batched message received", - new MessageConversionException("Insufficient batch data at offset " + byteBuffer.position()), - message); - } - byte[] body = new byte[length]; - byteBuffer.get(body); - messageProperties.setContentLength(length); - // Caveat - shared MessageProperties. - Message fragment = new Message(body, messageProperties); - invokeListener(channel, fragment); - } - } - else { - invokeListener(channel, message); - } + doExecuteListener(channel, messageIn); } - catch (Exception ex) { + catch (RuntimeException ex) { if (messageIn.getMessageProperties().isFinalRetryForMessageWithNoId()) { if (this.statefulRetryFatalWithNullMessageId) { throw new FatalListenerExecutionException( - "Illegal null id in message. Failed to manage retry for message: " + messageIn); + "Illegal null id in message. Failed to manage retry for message: " + messageIn, ex); } else { throw new ListenerExecutionFailedException("Cannot retry message more than once without an ID", @@ -1427,7 +1389,43 @@ public abstract class AbstractMessageListenerContainer extends RabbitAccessor } } - protected void invokeListener(Channel channel, Message message) throws Exception { + private void doExecuteListener(Channel channel, Message messageIn) { + Message message = messageIn; + if (this.afterReceivePostProcessors != null) { + for (MessagePostProcessor processor : this.afterReceivePostProcessors) { + message = processor.postProcessMessage(message); + if (message == null) { + throw new ImmediateAcknowledgeAmqpException( + "Message Post Processor returned 'null', discarding message"); + } + } + } + Object batchFormat = message.getMessageProperties().getHeaders().get(MessageProperties.SPRING_BATCH_FORMAT); + if (MessageProperties.BATCH_FORMAT_LENGTH_HEADER4.equals(batchFormat) && this.deBatchingEnabled) { + ByteBuffer byteBuffer = ByteBuffer.wrap(message.getBody()); + MessageProperties messageProperties = message.getMessageProperties(); + messageProperties.getHeaders().remove(MessageProperties.SPRING_BATCH_FORMAT); + while (byteBuffer.hasRemaining()) { + int length = byteBuffer.getInt(); + if (length < 0 || length > byteBuffer.remaining()) { + throw new ListenerExecutionFailedException("Bad batched message received", + new MessageConversionException("Insufficient batch data at offset " + byteBuffer.position()), + message); + } + byte[] body = new byte[length]; + byteBuffer.get(body); + messageProperties.setContentLength(length); + // Caveat - shared MessageProperties. + Message fragment = new Message(body, messageProperties); + invokeListener(channel, fragment); + } + } + else { + invokeListener(channel, message); + } + } + + protected void invokeListener(Channel channel, Message message) { this.proxy.invokeListener(channel, message); } @@ -1435,10 +1433,9 @@ public abstract class AbstractMessageListenerContainer extends RabbitAccessor * Invoke the specified listener: either as standard MessageListener or (preferably) as SessionAwareMessageListener. * @param channel the Rabbit Channel to operate on * @param message the received Rabbit Message - * @throws Exception if thrown by Rabbit API methods * @see #setMessageListener(MessageListener) */ - protected void actualInvokeListener(Channel channel, Message message) throws Exception { + protected void actualInvokeListener(Channel channel, Message message) { Object listener = getMessageListener(); if (listener instanceof ChannelAwareMessageListener) { doInvokeListener((ChannelAwareMessageListener) listener, channel, message); @@ -1473,17 +1470,14 @@ public abstract class AbstractMessageListenerContainer extends RabbitAccessor /** * Invoke the specified listener as Spring ChannelAwareMessageListener, exposing a new Rabbit Session (potentially * with its own transaction) to the listener if demanded. + * An exception thrown from the listener will be wrapped in a {@link ListenerExecutionFailedException}. * @param listener the Spring ChannelAwareMessageListener to invoke * @param channel the Rabbit Channel to operate on * @param message the received Rabbit Message - * @throws Exception if thrown by Rabbit API methods or listener itself. - *

- * Exception thrown from listener will be wrapped to {@link ListenerExecutionFailedException}. * @see ChannelAwareMessageListener * @see #setExposeListenerChannel(boolean) */ - protected void doInvokeListener(ChannelAwareMessageListener listener, Channel channel, Message message) - throws Exception { + protected void doInvokeListener(ChannelAwareMessageListener listener, Channel channel, Message message) { RabbitResourceHolder resourceHolder = null; Channel channelToUse = channel; @@ -1525,23 +1519,29 @@ public abstract class AbstractMessageListenerContainer extends RabbitAccessor } } finally { - if (resourceHolder != null && boundHere) { - // so the channel exposed (because exposeListenerChannel is false) will be closed - resourceHolder.setSynchronizedWithTransaction(false); - } - ConnectionFactoryUtils.releaseResources(resourceHolder); // NOSONAR - null check in method - if (boundHere) { - // unbind if we bound - TransactionSynchronizationManager.unbindResource(this.getConnectionFactory()); - if (!isExposeListenerChannel() && isChannelLocallyTransacted()) { - /* - * commit the temporary channel we exposed; the consumer's channel - * will be committed later. Note that when exposing a different channel - * when there's no transaction manager, the exposed channel is committed - * on each message, and not based on txSize. - */ - RabbitUtils.commitIfNecessary(channelToUse); - } + cleanUpAfterInvoke(resourceHolder, channelToUse, boundHere); + } + } + + private void cleanUpAfterInvoke(@Nullable RabbitResourceHolder resourceHolder, Channel channelToUse, + boolean boundHere) { + + if (resourceHolder != null && boundHere) { + // so the channel exposed (because exposeListenerChannel is false) will be closed + resourceHolder.setSynchronizedWithTransaction(false); + } + ConnectionFactoryUtils.releaseResources(resourceHolder); // NOSONAR - null check in method + if (boundHere) { + // unbind if we bound + TransactionSynchronizationManager.unbindResource(this.getConnectionFactory()); + if (!isExposeListenerChannel() && isChannelLocallyTransacted()) { + /* + * commit the temporary channel we exposed; the consumer's channel + * will be committed later. Note that when exposing a different channel + * when there's no transaction manager, the exposed channel is committed + * on each message, and not based on txSize. + */ + RabbitUtils.commitIfNecessary(channelToUse); } } } @@ -1555,11 +1555,10 @@ public abstract class AbstractMessageListenerContainer extends RabbitAccessor * * @param listener the Rabbit MessageListener to invoke * @param message the received Rabbit Message - * @throws Exception Any Exception. * * @see org.springframework.amqp.core.MessageListener#onMessage */ - protected void doInvokeListener(MessageListener listener, Message message) throws Exception { + protected void doInvokeListener(MessageListener listener, Message message) { try { listener.onMessage(message); } @@ -1607,12 +1606,14 @@ public abstract class AbstractMessageListenerContainer extends RabbitAccessor * @return If 'e' is of type {@link ListenerExecutionFailedException} - return 'e' as it is, otherwise wrap it to * {@link ListenerExecutionFailedException} and return. */ - protected Exception wrapToListenerExecutionFailedExceptionIfNeeded(Exception e, Message message) { + protected ListenerExecutionFailedException wrapToListenerExecutionFailedExceptionIfNeeded(Exception e, + Message message) { + if (!(e instanceof ListenerExecutionFailedException)) { // Wrap exception to ListenerExecutionFailedException. return new ListenerExecutionFailedException("Listener threw exception", e, message); } - return e; + return (ListenerExecutionFailedException) e; } protected void publishConsumerFailedEvent(String reason, boolean fatal, @Nullable Throwable t) { @@ -1707,34 +1708,37 @@ public abstract class AbstractMessageListenerContainer extends RabbitAccessor * fail with a fatal error if mismatches occur. */ protected synchronized void redeclareElementsIfNecessary() { - AmqpAdmin amqpAdmin = getAmqpAdmin(); - if (amqpAdmin == null || !isAutoDeclare()) { - return; + AmqpAdmin admin = getAmqpAdmin(); + if (admin != null && isAutoDeclare()) { + try { + attemptDeclarations(admin); + } + catch (Exception e) { + if (RabbitUtils.isMismatchedQueueArgs(e)) { + throw new FatalListenerStartupException("Mismatched queues", e); + } + logger.error("Failed to check/redeclare auto-delete queue(s).", e); + } } - try { - ApplicationContext applicationContext = this.getApplicationContext(); - if (applicationContext != null) { - Set queueNames = getQueueNamesAsSet(); - Map queueBeans = applicationContext.getBeansOfType(Queue.class); - for (Entry entry : queueBeans.entrySet()) { - Queue queue = entry.getValue(); - if (isMismatchedQueuesFatal() || (queueNames.contains(queue.getName()) && - amqpAdmin.getQueueProperties(queue.getName()) == null)) { - if (logger.isDebugEnabled()) { - logger.debug("Redeclaring context exchanges, queues, bindings."); - } - amqpAdmin.initialize(); - return; + } + + private void attemptDeclarations(AmqpAdmin amqpAdmin) { + ApplicationContext context = this.getApplicationContext(); + if (context != null) { + Set queueNames = getQueueNamesAsSet(); + Map queueBeans = context.getBeansOfType(Queue.class); + for (Entry entry : queueBeans.entrySet()) { + Queue queue = entry.getValue(); + if (isMismatchedQueuesFatal() || (queueNames.contains(queue.getName()) && + amqpAdmin.getQueueProperties(queue.getName()) == null)) { + if (logger.isDebugEnabled()) { + logger.debug("Redeclaring context exchanges, queues, bindings."); } + amqpAdmin.initialize(); + break; } } } - catch (Exception e) { - if (RabbitUtils.isMismatchedQueueArgs(e)) { - throw new FatalListenerStartupException("Mismatched queues", e); - } - logger.error("Failed to check/redeclare auto-delete queue(s).", e); - } } /** @@ -1779,17 +1783,17 @@ public abstract class AbstractMessageListenerContainer extends RabbitAccessor private void checkMissingQueuesFatalFromProperty() { if (!isMissingQueuesFatalSet()) { try { - ApplicationContext applicationContext = getApplicationContext(); - if (applicationContext != null) { - Properties properties = applicationContext.getBean("spring.amqp.global.properties", Properties.class); - String missingQueuesFatal = properties.getProperty("mlc.missing.queues.fatal"); + ApplicationContext context = getApplicationContext(); + if (context != null) { + Properties properties = context.getBean("spring.amqp.global.properties", Properties.class); + String missingQueuesFatalProperty = properties.getProperty("mlc.missing.queues.fatal"); - if (!StringUtils.hasText(missingQueuesFatal)) { - missingQueuesFatal = properties.getProperty("smlc.missing.queues.fatal"); + if (!StringUtils.hasText(missingQueuesFatalProperty)) { + missingQueuesFatalProperty = properties.getProperty("smlc.missing.queues.fatal"); } - if (StringUtils.hasText(missingQueuesFatal)) { - setMissingQueuesFatal(Boolean.parseBoolean(missingQueuesFatal)); + if (StringUtils.hasText(missingQueuesFatalProperty)) { + setMissingQueuesFatal(Boolean.parseBoolean(missingQueuesFatalProperty)); } } } @@ -1802,13 +1806,14 @@ public abstract class AbstractMessageListenerContainer extends RabbitAccessor private void checkPossibleAuthenticationFailureFatalFromProperty() { if (!isPossibleAuthenticationFailureFatal()) { try { - ApplicationContext applicationContext = getApplicationContext(); - if (applicationContext != null) { - Properties properties = applicationContext.getBean("spring.amqp.global.properties", Properties.class); - String possibleAuthenticationFailureFatal = + ApplicationContext context = getApplicationContext(); + if (context != null) { + Properties properties = context.getBean("spring.amqp.global.properties", Properties.class); + String possibleAuthenticationFailureFatalProperty = properties.getProperty("mlc.possible.authentication.failure.fatal"); - if (StringUtils.hasText(possibleAuthenticationFailureFatal)) { - setPossibleAuthenticationFailureFatal(Boolean.parseBoolean(possibleAuthenticationFailureFatal)); + if (StringUtils.hasText(possibleAuthenticationFailureFatalProperty)) { + setPossibleAuthenticationFailureFatal( + Boolean.parseBoolean(possibleAuthenticationFailureFatalProperty)); } } } @@ -1821,7 +1826,7 @@ public abstract class AbstractMessageListenerContainer extends RabbitAccessor @FunctionalInterface private interface ContainerDelegate { - void invokeListener(Channel channel, Message message) throws Exception; + void invokeListener(Channel channel, Message message); } diff --git a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/BlockingQueueConsumer.java b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/BlockingQueueConsumer.java index 6bb5b12e..a0932d0f 100644 --- a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/BlockingQueueConsumer.java +++ b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/BlockingQueueConsumer.java @@ -156,9 +156,9 @@ public class BlockingQueueConsumer { private volatile boolean normalCancel; - volatile Thread thread; + volatile Thread thread; // NOSONAR package protected - volatile boolean declaring; + volatile boolean declaring; // NOSONAR package protected /** * Create a consumer. The consumer must not attempt to use @@ -520,29 +520,29 @@ public class BlockingQueueConsumer { Iterator iterator = this.missingQueues.iterator(); while (iterator.hasNext()) { boolean available = true; - String queue = iterator.next(); + String queueToCheck = iterator.next(); Connection connection = null; // NOSONAR - RabbitUtils - Channel channel = null; + Channel channelForCheck = null; try { - channel = this.connectionFactory.createConnection().createChannel(false); - channel.queueDeclarePassive(queue); + channelForCheck = this.connectionFactory.createConnection().createChannel(false); + channelForCheck.queueDeclarePassive(queueToCheck); if (logger.isInfoEnabled()) { - logger.info("Queue '" + queue + "' is now available"); + logger.info("Queue '" + queueToCheck + "' is now available"); } } catch (IOException e) { available = false; if (logger.isWarnEnabled()) { - logger.warn("Queue '" + queue + "' is still not available"); + logger.warn("Queue '" + queueToCheck + "' is still not available"); } } finally { - RabbitUtils.closeChannel(channel); + RabbitUtils.closeChannel(channelForCheck); RabbitUtils.closeConnection(connection); } if (available) { try { - this.consumeFromQueue(queue); + this.consumeFromQueue(queueToCheck); iterator.remove(); } catch (IOException e) { @@ -589,34 +589,7 @@ public class BlockingQueueConsumer { passiveDeclareRetries = 0; } catch (DeclarationException e) { - if (passiveDeclareRetries > 0 && this.channel.isOpen()) { - if (logger.isWarnEnabled()) { - logger.warn("Queue declaration failed; retries left=" + (passiveDeclareRetries), e); - try { - Thread.sleep(this.failedDeclarationRetryInterval); - } - catch (InterruptedException e1) { - this.declaring = false; - Thread.currentThread().interrupt(); - this.activeObjectCounter.release(this); - throw RabbitExceptionTranslator.convertRabbitAccessException(e1); - } - } - } - else if (e.getFailedQueues().size() < this.queues.length) { - if (logger.isWarnEnabled()) { - logger.warn("Not all queues are available; only listening on those that are - configured: " - + Arrays.asList(this.queues) + "; not available: " + e.getFailedQueues()); - } - this.missingQueues.addAll(e.getFailedQueues()); - this.lastRetryDeclaration = System.currentTimeMillis(); - } - else { - this.declaring = false; - this.activeObjectCounter.release(this); - throw new QueuesNotAvailableException("Cannot prepare queue for listener. " - + "Either the queue doesn't exist or the broker will not allow us to use it.", e); - } + handleDeclarationException(passiveDeclareRetries, e); } } while (passiveDeclareRetries-- > 0 && !cancelled()); @@ -649,6 +622,37 @@ public class BlockingQueueConsumer { } } + private void handleDeclarationException(int passiveDeclareRetries, DeclarationException e) { + if (passiveDeclareRetries > 0 && this.channel.isOpen()) { + if (logger.isWarnEnabled()) { + logger.warn("Queue declaration failed; retries left=" + (passiveDeclareRetries), e); + try { + Thread.sleep(this.failedDeclarationRetryInterval); + } + catch (InterruptedException e1) { + this.declaring = false; + Thread.currentThread().interrupt(); + this.activeObjectCounter.release(this); + throw RabbitExceptionTranslator.convertRabbitAccessException(e1); // NOSONAR stack trace loss + } + } + } + else if (e.getFailedQueues().size() < this.queues.length) { + if (logger.isWarnEnabled()) { + logger.warn("Not all queues are available; only listening on those that are - configured: " + + Arrays.asList(this.queues) + "; not available: " + e.getFailedQueues()); + } + this.missingQueues.addAll(e.getFailedQueues()); + this.lastRetryDeclaration = System.currentTimeMillis(); + } + else { + this.declaring = false; + this.activeObjectCounter.release(this); + throw new QueuesNotAvailableException("Cannot prepare queue for listener. " + + "Either the queue doesn't exist or the broker will not allow us to use it.", e); + } + } + private void consumeFromQueue(String queue) throws IOException { InternalConsumer consumer = new InternalConsumer(this.channel, queue); String consumerTag = this.channel.basicConsume(queue, this.acknowledgeMode.isAutoAck(), @@ -756,7 +760,7 @@ public class BlockingQueueConsumer { } catch (Exception e) { logger.error("Application exception overridden by rollback exception", ex); - throw e; + throw RabbitExceptionTranslator.convertRabbitAccessException(e); // NOSONAR stack trace loss } finally { this.deliveryTags.clear(); @@ -890,13 +894,14 @@ public class BlockingQueueConsumer { new Delivery(consumerTag, envelope, properties, body, this.queue), BlockingQueueConsumer.this.shutdownTimeout, TimeUnit.MILLISECONDS)) { - RabbitUtils.setPhysicalCloseRequired(getChannel(), true); + Channel channelToClose = super.getChannel(); + RabbitUtils.setPhysicalCloseRequired(channelToClose, true); // Defensive - should never happen BlockingQueueConsumer.this.queue.clear(); - getChannel().basicNack(envelope.getDeliveryTag(), true, true); - getChannel().basicCancel(consumerTag); + channelToClose.basicNack(envelope.getDeliveryTag(), true, true); + channelToClose.basicCancel(consumerTag); try { - getChannel().close(); + channelToClose.close(); } catch (TimeoutException e) { // no-op diff --git a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/DirectMessageListenerContainer.java b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/DirectMessageListenerContainer.java index d5ff6c45..ee5a7098 100644 --- a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/DirectMessageListenerContainer.java +++ b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/DirectMessageListenerContainer.java @@ -350,7 +350,7 @@ public class DirectMessageListenerContainer extends AbstractMessageListenerConta } @Override - protected void doInitialize() throws Exception { + protected void doInitialize() { if (this.taskScheduler == null) { ThreadPoolTaskScheduler threadPoolTaskScheduler = new ThreadPoolTaskScheduler(); threadPoolTaskScheduler.setThreadNamePrefix(getBeanName() + "-consumerMonitor-"); @@ -363,7 +363,7 @@ public class DirectMessageListenerContainer extends AbstractMessageListenerConta } @Override - protected void doStart() throws Exception { + protected void doStart() { if (!this.started) { actualStart(); } @@ -378,7 +378,7 @@ public class DirectMessageListenerContainer extends AbstractMessageListenerConta } } - protected void actualStart() throws Exception { + protected void actualStart() { this.aborted = false; this.hasStopped = false; if (getPrefetchCount() < this.messagesPerAck) { diff --git a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/DirectReplyToMessageListenerContainer.java b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/DirectReplyToMessageListenerContainer.java index 30349640..3d4c0517 100644 --- a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/DirectReplyToMessageListenerContainer.java +++ b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/DirectReplyToMessageListenerContainer.java @@ -122,7 +122,7 @@ public class DirectReplyToMessageListenerContainer extends DirectMessageListener } @Override - protected void doStart() throws Exception { + protected void doStart() { if (!isRunning()) { this.consumerCount = 0; super.setConsumersPerQueue(0); diff --git a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/SimpleMessageListenerContainer.java b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/SimpleMessageListenerContainer.java index ed3648ba..3ca26241 100644 --- a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/SimpleMessageListenerContainer.java +++ b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/SimpleMessageListenerContainer.java @@ -53,6 +53,7 @@ import org.springframework.amqp.rabbit.listener.exception.FatalListenerStartupEx import org.springframework.amqp.rabbit.listener.exception.ListenerExecutionFailedException; import org.springframework.amqp.rabbit.support.ConsumerCancelledException; import org.springframework.amqp.rabbit.support.ListenerContainerAware; +import org.springframework.amqp.rabbit.support.RabbitExceptionTranslator; import org.springframework.jmx.export.annotation.ManagedMetric; import org.springframework.jmx.support.MetricType; import org.springframework.transaction.support.TransactionSynchronizationManager; @@ -462,7 +463,7 @@ public class SimpleMessageListenerContainer extends AbstractMessageListenerConta } @Override - protected void doInitialize() throws Exception { + protected void doInitialize() { } @@ -474,10 +475,9 @@ public class SimpleMessageListenerContainer extends AbstractMessageListenerConta /** * Re-initializes this container's Rabbit message consumers, if not initialized already. Then submits each consumer * to this container's task executor. - * @throws Exception Any Exception. */ @Override - protected void doStart() throws Exception { + protected void doStart() { if (getMessageListener() instanceof ListenerContainerAware) { Collection expectedQueueNames = ((ListenerContainerAware) getMessageListener()).expectedQueueNames(); if (expectedQueueNames != null) { @@ -526,7 +526,17 @@ public class SimpleMessageListenerContainer extends AbstractMessageListenerConta } } for (AsyncMessageProcessingConsumer processor : processors) { - FatalListenerStartupException startupException = processor.getStartupException(); + FatalListenerStartupException startupException = null; + try { + startupException = processor.getStartupException(); + } + catch (TimeoutException e) { + throw RabbitExceptionTranslator.convertRabbitAccessException(e); + } + catch (InterruptedException e) { + Thread.currentThread().interrupt(); + throw RabbitExceptionTranslator.convertRabbitAccessException(e); + } if (startupException != null) { throw new AmqpIllegalStateException("Fatal exception on listener startup", startupException); } @@ -976,8 +986,7 @@ public class SimpleMessageListenerContainer extends AbstractMessageListenerConta * @throws TimeoutException if the consumer hasn't started * @throws InterruptedException if the consumer startup is interrupted */ - private FatalListenerStartupException getStartupException() throws TimeoutException, - InterruptedException { + private FatalListenerStartupException getStartupException() throws TimeoutException, InterruptedException { if (!this.start.await( SimpleMessageListenerContainer.this.consumerStartTimeout, TimeUnit.MILLISECONDS)) { logger.error("Consumer failed to start in " diff --git a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/connection/CachingConnectionFactoryIntegrationTests.java b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/connection/CachingConnectionFactoryIntegrationTests.java index 80074011..7dc07d25 100644 --- a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/connection/CachingConnectionFactoryIntegrationTests.java +++ b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/connection/CachingConnectionFactoryIntegrationTests.java @@ -467,11 +467,4 @@ public class CachingConnectionFactoryIntegrationTests { connection.createChannel(false); } - private Log spyOnLogger(CachingConnectionFactory connectionFactory2) { - DirectFieldAccessor dfa = new DirectFieldAccessor(connectionFactory2); - Log logger = spy((Log) dfa.getPropertyValue("logger")); - dfa.setPropertyValue("logger", logger); - return logger; - } - } diff --git a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/connection/ConnectionFactoryLifecycleTests.java b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/connection/ConnectionFactoryLifecycleTests.java index b2e12ac3..a10eec35 100644 --- a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/connection/ConnectionFactoryLifecycleTests.java +++ b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/connection/ConnectionFactoryLifecycleTests.java @@ -128,6 +128,7 @@ public class ConnectionFactoryLifecycleTests { ConnectionUnblockedEvent connectionUnblockedEvent = unblockedConnectionEvent.get(); assertNotNull(connectionUnblockedEvent); assertSame(TestUtils.getPropertyValue(connection, "target"), connectionUnblockedEvent.getConnection()); + context.close(); } @Configuration diff --git a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/SimpleMessageListenerContainerTests.java b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/SimpleMessageListenerContainerTests.java index 0805a74a..046718c1 100644 --- a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/SimpleMessageListenerContainerTests.java +++ b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/SimpleMessageListenerContainerTests.java @@ -161,7 +161,7 @@ public class SimpleMessageListenerContainerTests { final SingleConnectionFactory singleConnectionFactory = new SingleConnectionFactory("localhost"); SimpleMessageListenerContainer container = new SimpleMessageListenerContainer(singleConnectionFactory) { @Override - protected void doStart() throws Exception { + protected void doStart() { // do nothing } }; @@ -536,7 +536,7 @@ public class SimpleMessageListenerContainerTests { class Container extends SimpleMessageListenerContainer { @Override - public void executeListener(Channel channel, Message messageIn) throws Exception { + public void executeListener(Channel channel, Message messageIn) { super.executeListener(channel, messageIn); }