diff --git a/build.gradle b/build.gradle index 0929868b..e26d4fb8 100644 --- a/build.gradle +++ b/build.gradle @@ -99,8 +99,8 @@ subprojects { subproject -> log4j2Version = '2.7' logbackVersion = '1.1.7' mockitoVersion = '1.10.19' - rabbitmqVersion = project.hasProperty('rabbitmqVersion') ? project.rabbitmqVersion : '3.6.5' - rabbitmqHttpClientVersion = '1.0.0.RELEASE' + rabbitmqVersion = project.hasProperty('rabbitmqVersion') ? project.rabbitmqVersion : '4.0.0' + rabbitmqHttpClientVersion = '1.1.0.RELEASE' springVersion = project.hasProperty('springVersion') ? project.springVersion : '4.3.4.RELEASE' diff --git a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/AutoRecoverConnectionNotCurrentlyOpenException.java b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/AutoRecoverConnectionNotCurrentlyOpenException.java new file mode 100644 index 00000000..9a269e82 --- /dev/null +++ b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/AutoRecoverConnectionNotCurrentlyOpenException.java @@ -0,0 +1,36 @@ +/* + * Copyright 2016 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. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.amqp.rabbit.connection; + +import org.springframework.amqp.AmqpException; + +/** + * An exception thrown if the connection is an auto recover connection + * that is not currently open. + * + * @author Gary Russell + * @since 1.7 + * + */ +@SuppressWarnings("serial") +public class AutoRecoverConnectionNotCurrentlyOpenException extends AmqpException { + + AutoRecoverConnectionNotCurrentlyOpenException(String message) { + super(message); + } + +} diff --git a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/SimpleConnection.java b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/SimpleConnection.java index 21d80956..255f574b 100755 --- a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/SimpleConnection.java +++ b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/SimpleConnection.java @@ -17,26 +17,32 @@ package org.springframework.amqp.rabbit.connection; import java.io.IOException; +import java.net.InetAddress; import org.springframework.amqp.rabbit.support.RabbitExceptionTranslator; import org.springframework.util.ObjectUtils; import com.rabbitmq.client.Channel; -import com.rabbitmq.client.impl.AMQConnection; +import com.rabbitmq.client.impl.NetworkConnection; +import com.rabbitmq.client.impl.recovery.AutorecoveringConnection; /** * Simply a Connection. + * * @author Dave Syer * @author Gary Russell + * * @since 1.0 * */ -public class SimpleConnection implements Connection { +public class SimpleConnection implements Connection, NetworkConnection { private final com.rabbitmq.client.Connection delegate; private final int closeTimeout; + private volatile boolean explicitlyClosed; + public SimpleConnection(com.rabbitmq.client.Connection delegate, int closeTimeout) { this.delegate = delegate; @@ -61,6 +67,7 @@ public class SimpleConnection implements Connection { @Override public void close() { try { + this.explicitlyClosed = true; // let the physical close time out if necessary this.delegate.close(this.closeTimeout); } @@ -69,21 +76,54 @@ public class SimpleConnection implements Connection { } } + /** + * True if the connection is open. + * @return true if the connection is open + * @throws AutoRecoverConnectionNotCurrentlyOpenException if the connection is an + * {@link AutorecoveringConnection} and is currently closed; this is required to + * prevent the {@link CachingConnectionFactory} from discarding this connection + * and opening a new one, in which case the "old" connection would eventually be recovered + * and orphaned - also any consumers belonging to it might be recovered too + * and the broker will deliver messages to them when there is no code actually running + * to deal with those messages (when using the {@code SimpleMessageListenerContainer}). + * If we have actually closed the connection + * (e.g. via {@link CachingConnectionFactory#resetConnection()}) this will return false. + */ @Override public boolean isOpen() { - return this.delegate != null - && (this.delegate.isOpen() || this.delegate.getClass().getSimpleName().contains("AutorecoveringConnection")); + if (!this.explicitlyClosed && this.delegate instanceof AutorecoveringConnection && !this.delegate.isOpen()) { + throw new AutoRecoverConnectionNotCurrentlyOpenException("Auto recovery connection is not currently open"); + } + return this.delegate != null && (this.delegate.isOpen()); } @Override public int getLocalPort() { - if (this.delegate instanceof AMQConnection) { - return ((AMQConnection) this.delegate).getLocalPort(); + if (this.delegate instanceof NetworkConnection) { + return ((NetworkConnection) this.delegate).getLocalPort(); } return 0; } + @Override + public InetAddress getLocalAddress() { + if (this.delegate instanceof NetworkConnection) { + return ((NetworkConnection) this.delegate).getLocalAddress(); + } + return null; + } + + @Override + public InetAddress getAddress() { + return this.delegate.getAddress(); + } + + @Override + public int getPort() { + return this.delegate.getPort(); + } + @Override public String toString() { return "SimpleConnection@" 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 6e6f2a50..29086a83 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 @@ -637,7 +637,7 @@ public class RabbitAdmin implements AmqpAdmin, ApplicationContextAware, Applicat if (this.applicationEventPublisher != null) { this.applicationEventPublisher.publishEvent(event); } - if (this.ignoreDeclarationExceptions || element.isIgnoreDeclarationExceptions()) { + if (this.ignoreDeclarationExceptions || (element != null && element.isIgnoreDeclarationExceptions())) { if (this.logger.isWarnEnabled()) { this.logger.warn("Failed to declare " + elementType + (element == null ? "broker-generated" : ": " + element) 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 a7b12bdd..a473a8e1 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 @@ -88,8 +88,6 @@ import com.rabbitmq.client.Channel; import com.rabbitmq.client.DefaultConsumer; import com.rabbitmq.client.Envelope; import com.rabbitmq.client.GetResponse; -import com.rabbitmq.client.QueueingConsumer; -import com.rabbitmq.client.QueueingConsumer.Delivery; /** *

@@ -861,10 +859,11 @@ public class RabbitTemplate extends RabbitAccessor implements BeanFactoryAware, public Message receive(final String queueName, final long timeoutMillis) { return execute(new ChannelCallback() { + @SuppressWarnings("deprecation") @Override public Message doInRabbit(Channel channel) throws Exception { - QueueingConsumer consumer = createQueueingConsumer(queueName, channel); - Delivery delivery; + com.rabbitmq.client.QueueingConsumer consumer = createQueueingConsumer(queueName, channel); + com.rabbitmq.client.QueueingConsumer.Delivery delivery; if (timeoutMillis < 0) { delivery = consumer.nextDelivery(); } @@ -991,8 +990,8 @@ public class RabbitTemplate extends RabbitAccessor implements BeanFactoryAware, } } else { - QueueingConsumer consumer = createQueueingConsumer(queueName, channel); - Delivery delivery; + com.rabbitmq.client.QueueingConsumer consumer = createQueueingConsumer(queueName, channel); + com.rabbitmq.client.QueueingConsumer.Delivery delivery; if (RabbitTemplate.this.receiveTimeout < 0) { delivery = consumer.nextDelivery(); } @@ -1514,7 +1513,8 @@ public class RabbitTemplate extends RabbitAccessor implements BeanFactoryAware, return isChannelTransacted() && !ConnectionFactoryUtils.isChannelTransactional(channel, getConnectionFactory()); } - private Message buildMessageFromDelivery(Delivery delivery) { + @SuppressWarnings("deprecation") + private Message buildMessageFromDelivery(com.rabbitmq.client.QueueingConsumer.Delivery delivery) { return buildMessage(delivery.getEnvelope(), delivery.getProperties(), delivery.getBody(), -1); } private Message buildMessageFromResponse(GetResponse response) { @@ -1739,10 +1739,12 @@ public class RabbitTemplate extends RabbitAccessor implements BeanFactoryAware, } } - private QueueingConsumer createQueueingConsumer(final String queueName, Channel channel) throws Exception { + @SuppressWarnings("deprecation") + private com.rabbitmq.client.QueueingConsumer createQueueingConsumer(final String queueName, Channel channel) + throws Exception { channel.basicQos(1); final CountDownLatch latch = new CountDownLatch(1); - QueueingConsumer consumer = new QueueingConsumer(channel) { + com.rabbitmq.client.QueueingConsumer consumer = new com.rabbitmq.client.QueueingConsumer(channel) { @Override public void handleCancel(String consumerTag) throws IOException { 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 1d193d31..a088e30e 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 @@ -51,6 +51,7 @@ import org.springframework.amqp.rabbit.connection.RabbitResourceHolder; import org.springframework.amqp.rabbit.connection.RabbitUtils; import org.springframework.amqp.rabbit.listener.exception.FatalListenerStartupException; import org.springframework.amqp.rabbit.support.ConsumerCancelledException; +import org.springframework.amqp.rabbit.support.Delivery; import org.springframework.amqp.rabbit.support.MessagePropertiesConverter; import org.springframework.amqp.rabbit.support.RabbitExceptionTranslator; import org.springframework.amqp.support.ConsumerTagStrategy; @@ -58,7 +59,6 @@ import org.springframework.util.ObjectUtils; import org.springframework.util.backoff.BackOffExecution; import com.rabbitmq.client.AMQP; -import com.rabbitmq.client.AMQP.BasicProperties; import com.rabbitmq.client.AlreadyClosedException; import com.rabbitmq.client.Channel; import com.rabbitmq.client.DefaultConsumer; @@ -819,43 +819,6 @@ public class BlockingQueueConsumer { } - /** - * Encapsulates an arbitrary message - simple "bean" holder structure. - */ - private static class Delivery { - - private final String consumerTag; - - private final Envelope envelope; - - private final AMQP.BasicProperties properties; - - private final byte[] body; - - Delivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) { //NOSONAR - this.consumerTag = consumerTag; - this.envelope = envelope; - this.properties = properties; - this.body = body; - } - - public String getConsumerTag() { - return this.consumerTag; - } - - public Envelope getEnvelope() { - return this.envelope; - } - - public BasicProperties getProperties() { - return this.properties; - } - - public byte[] getBody() { - return this.body; - } - } - @SuppressWarnings("serial") private static final class DeclarationException extends AmqpException { diff --git a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/support/Delivery.java b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/support/Delivery.java new file mode 100644 index 00000000..8d2391e0 --- /dev/null +++ b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/support/Delivery.java @@ -0,0 +1,78 @@ +/* + * Copyright 2016 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. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.amqp.rabbit.support; + +import com.rabbitmq.client.AMQP; +import com.rabbitmq.client.AMQP.BasicProperties; +import com.rabbitmq.client.Envelope; + +/** + * Encapsulates an arbitrary message - simple "bean" holder structure. + * + * @author Gary Russell + * @since 1.7 + */ +public class Delivery { + + private final String consumerTag; + + private final Envelope envelope; + + private final AMQP.BasicProperties properties; + + private final byte[] body; + + public Delivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) { //NOSONAR + this.consumerTag = consumerTag; + this.envelope = envelope; + this.properties = properties; + this.body = body; + } + + /** + * Retrieve the consumer tag. + * @return the consumer tag. + */ + public String getConsumerTag() { + return this.consumerTag; + } + + /** + * Retrieve the message envelope. + * @return the message envelope. + */ + public Envelope getEnvelope() { + return this.envelope; + } + + /** + * Retrieve the message properties. + * @return the message properties. + */ + public BasicProperties getProperties() { + return this.properties; + } + + /** + * Retrieve the message body. + * @return the message body. + */ + public byte[] getBody() { + return this.body; + } + +} diff --git a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/support/PublisherCallbackChannelImpl.java b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/support/PublisherCallbackChannelImpl.java index c1f133cb..af3c4f79 100644 --- a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/support/PublisherCallbackChannelImpl.java +++ b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/support/PublisherCallbackChannelImpl.java @@ -36,10 +36,6 @@ import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; import org.springframework.util.Assert; -import org.springframework.util.ObjectUtils; -import org.springframework.util.ReflectionUtils; -import org.springframework.util.ReflectionUtils.MethodCallback; -import org.springframework.util.ReflectionUtils.MethodFilter; import com.rabbitmq.client.AMQP; import com.rabbitmq.client.AMQP.Basic.RecoverOk; @@ -53,12 +49,12 @@ import com.rabbitmq.client.AMQP.Tx.CommitOk; import com.rabbitmq.client.AMQP.Tx.RollbackOk; import com.rabbitmq.client.AMQP.Tx.SelectOk; import com.rabbitmq.client.AlreadyClosedException; +import com.rabbitmq.client.BuiltinExchangeType; import com.rabbitmq.client.Channel; import com.rabbitmq.client.Command; import com.rabbitmq.client.ConfirmListener; import com.rabbitmq.client.Connection; import com.rabbitmq.client.Consumer; -import com.rabbitmq.client.FlowListener; import com.rabbitmq.client.GetResponse; import com.rabbitmq.client.Method; import com.rabbitmq.client.ReturnListener; @@ -76,24 +72,6 @@ import com.rabbitmq.client.ShutdownSignalException; public class PublisherCallbackChannelImpl implements PublisherCallbackChannel, ConfirmListener, ReturnListener, ShutdownListener { - private static final String[] METHODS_OF_INTEREST = - new String[] { "consumerCount", "messageCount" }; - - private static final MethodFilter METHOD_FILTER = new MethodFilter() { - - @Override - public boolean matches(java.lang.reflect.Method method) { - return ObjectUtils.containsElement(METHODS_OF_INTEREST, method.getName()); - } - - }; - - private static volatile java.lang.reflect.Method consumerCountMethod; - - private static volatile java.lang.reflect.Method messageCountMethod; - - private static volatile boolean conditionalMethodsChecked; - private final Log logger = LogFactory.getLog(this.getClass()); private final Channel delegate; @@ -108,29 +86,6 @@ public class PublisherCallbackChannelImpl public PublisherCallbackChannelImpl(Channel delegate) { delegate.addShutdownListener(this); this.delegate = delegate; - - if (!conditionalMethodsChecked) { - // The following reflection is required to maintain compatibility with pre 3.6.x clients. - ReflectionUtils.doWithMethods(delegate.getClass(), new MethodCallback() { - - @Override - public void doWith(java.lang.reflect.Method method) - throws IllegalArgumentException, IllegalAccessException { - if ("consumerCount".equals(method.getName()) && method.getParameterTypes().length == 1 - && String.class.equals(method.getParameterTypes()[0]) - && long.class.equals(method.getReturnType())) { - consumerCountMethod = method; - } - else if ("messageCount".equals(method.getName()) && method.getParameterTypes().length == 1 - && String.class.equals(method.getParameterTypes()[0]) - && long.class.equals(method.getReturnType())) { - messageCountMethod = method; - } - } - - }, METHOD_FILTER); - conditionalMethodsChecked = true; - } } ////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////// @@ -200,13 +155,13 @@ public class PublisherCallbackChannelImpl @Override @SuppressWarnings("deprecation") - public void addFlowListener(FlowListener listener) { + public void addFlowListener(com.rabbitmq.client.FlowListener listener) { this.delegate.addFlowListener(listener); } @Override @SuppressWarnings("deprecation") - public boolean removeFlowListener(FlowListener listener) { + public boolean removeFlowListener(com.rabbitmq.client.FlowListener listener) { return this.delegate.removeFlowListener(listener); } @@ -272,12 +227,22 @@ public class PublisherCallbackChannelImpl return this.delegate.exchangeDeclare(exchange, type); } + @Override + public DeclareOk exchangeDeclare(String exchange, BuiltinExchangeType type) throws IOException { + return this.delegate.exchangeDeclare(exchange, type); + } + @Override public DeclareOk exchangeDeclare(String exchange, String type, boolean durable) throws IOException { return this.delegate.exchangeDeclare(exchange, type, durable); } + @Override + public DeclareOk exchangeDeclare(String exchange, BuiltinExchangeType type, boolean durable) throws IOException { + return this.delegate.exchangeDeclare(exchange, type, durable); + } + @Override public DeclareOk exchangeDeclare(String exchange, String type, boolean durable, boolean autoDelete, Map arguments) @@ -286,6 +251,12 @@ public class PublisherCallbackChannelImpl arguments); } + @Override + public DeclareOk exchangeDeclare(String exchange, BuiltinExchangeType type, boolean durable, boolean autoDelete, + Map arguments) throws IOException { + return this.delegate.exchangeDeclare(exchange, type, durable, autoDelete, arguments); + } + @Override public DeclareOk exchangeDeclare(String exchange, String type, boolean durable, boolean autoDelete, boolean internal, @@ -294,6 +265,12 @@ public class PublisherCallbackChannelImpl internal, arguments); } + @Override + public DeclareOk exchangeDeclare(String exchange, BuiltinExchangeType type, boolean durable, boolean autoDelete, + boolean internal, Map arguments) throws IOException { + return this.delegate.exchangeDeclare(exchange, type, durable, autoDelete, internal, arguments); + } + @Override public DeclareOk exchangeDeclarePassive(String name) throws IOException { return this.delegate.exchangeDeclarePassive(name); @@ -577,6 +554,12 @@ public class PublisherCallbackChannelImpl this.delegate.exchangeDeclareNoWait(exchange, type, durable, autoDelete, internal, arguments); } + @Override + public void exchangeDeclareNoWait(String exchange, BuiltinExchangeType type, boolean durable, boolean autoDelete, + boolean internal, Map arguments) throws IOException { + this.delegate.exchangeDeclareNoWait(exchange, type, durable, autoDelete, internal, arguments); + } + @Override public void exchangeDeleteNoWait(String exchange, boolean ifUnused) throws IOException { this.delegate.exchangeDeleteNoWait(exchange, ifUnused); @@ -611,18 +594,12 @@ public class PublisherCallbackChannelImpl @Override public long consumerCount(String queue) throws IOException { - if (consumerCountMethod != null) { - return (Long) ReflectionUtils.invokeMethod(consumerCountMethod, this.delegate, new Object[] { queue }); - } - throw new UnsupportedOperationException("'consumerCount()' requires a 3.6+ client library"); + return this.delegate.consumerCount(queue); } @Override public long messageCount(String queue) throws IOException { - if (messageCountMethod != null) { - return (Long) ReflectionUtils.invokeMethod(messageCountMethod, this.delegate, new Object[] { queue }); - } - throw new UnsupportedOperationException("'messageCountMethod()' requires a 3.6+ client library"); + return this.delegate.messageCount(queue); } @Override diff --git a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/support/RabbitExceptionTranslator.java b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/support/RabbitExceptionTranslator.java index 094a4896..d416ec45 100644 --- a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/support/RabbitExceptionTranslator.java +++ b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/support/RabbitExceptionTranslator.java @@ -76,6 +76,9 @@ public final class RabbitExceptionTranslator { if (ex instanceof ConsumerCancelledException) { return new org.springframework.amqp.rabbit.support.ConsumerCancelledException(ex); } + if (ex instanceof org.springframework.amqp.rabbit.support.ConsumerCancelledException) { + throw (org.springframework.amqp.rabbit.support.ConsumerCancelledException) ex; + } // fallback return new UncategorizedAmqpException(ex); } 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 0b5013f5..fa8df18d 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 @@ -26,6 +26,7 @@ import static org.mockito.Mockito.never; import static org.mockito.Mockito.spy; import static org.mockito.Mockito.verify; +import java.io.IOException; import java.net.ServerSocket; import java.net.Socket; import java.util.ArrayList; @@ -43,6 +44,7 @@ import javax.net.SocketFactory; import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; +import org.hamcrest.Matchers; import org.junit.After; import org.junit.Before; import org.junit.Ignore; @@ -50,6 +52,8 @@ import org.junit.Rule; import org.junit.Test; import org.junit.rules.ExpectedException; +import org.springframework.amqp.AmqpAuthenticationException; +import org.springframework.amqp.AmqpException; import org.springframework.amqp.AmqpIOException; import org.springframework.amqp.AmqpTimeoutException; import org.springframework.amqp.core.Queue; @@ -64,8 +68,11 @@ import org.springframework.beans.DirectFieldAccessor; import com.rabbitmq.client.Channel; import com.rabbitmq.client.DefaultConsumer; +import com.rabbitmq.client.Recoverable; +import com.rabbitmq.client.RecoveryListener; import com.rabbitmq.client.ShutdownListener; import com.rabbitmq.client.ShutdownSignalException; +import com.rabbitmq.client.impl.recovery.AutorecoveringChannel; /** * @author Dave Syer @@ -99,7 +106,7 @@ public class CachingConnectionFactoryIntegrationTests { @After public void close() { if (!this.connectionFactory.getVirtualHost().equals("non-existent")) { - new RabbitAdmin(this.connectionFactory).deleteQueue(CF_INTEGRATION_TEST_QUEUE); + this.brokerIsRunning.getAdmin().deleteQueue(CF_INTEGRATION_TEST_QUEUE); } assertEquals("bar", connectionFactory.getRabbitConnectionFactory().getClientProperties().get("foo")); connectionFactory.destroy(); @@ -179,6 +186,8 @@ public class CachingConnectionFactoryIntegrationTests { connectionFactory.setCacheMode(CacheMode.CONNECTION); connectionFactory.setConnectionCacheSize(1); connectionFactory.setChannelCacheSize(3); + // the following is needed because we close the underlying connection below. + connectionFactory.getRabbitConnectionFactory().setAutomaticRecoveryEnabled(false); List connections = new ArrayList(); connections.add(connectionFactory.createConnection()); connections.add(connectionFactory.createConnection()); @@ -260,14 +269,13 @@ public class CachingConnectionFactoryIntegrationTests { @Test public void testReceiveFromNonExistentVirtualHost() throws Exception { - connectionFactory.setVirtualHost("non-existent"); RabbitTemplate template = new RabbitTemplate(connectionFactory); - // Wrong vhost is very unfriendly to client - the exception has no clue (just an EOF) - exception.expect(AmqpIOException.class); - String result = (String) template.receiveAndConvert("foo"); - assertEquals("message", result); + // Wrong vhost is very unfriendly to client - the exception has no clue (just an EOF) + exception.expect(Matchers.anyOf(Matchers.instanceOf(AmqpIOException.class), + Matchers.instanceOf(AmqpAuthenticationException.class))); + template.receiveAndConvert("foo"); } @Test @@ -319,8 +327,8 @@ public class CachingConnectionFactoryIntegrationTests { } @Test - public void testHardErrorAndReconnect() throws Exception { - + public void testHardErrorAndReconnectNoAuto() throws Exception { + this.connectionFactory.getRabbitConnectionFactory().setAutomaticRecoveryEnabled(false); RabbitTemplate template = new RabbitTemplate(connectionFactory); RabbitAdmin admin = new RabbitAdmin(connectionFactory); Queue queue = new Queue(CF_INTEGRATION_TEST_QUEUE); @@ -361,6 +369,71 @@ public class CachingConnectionFactoryIntegrationTests { assertEquals(null, result); } + @Test + public void testHardErrorAndReconnectAuto() throws Exception { + + RabbitTemplate template = new RabbitTemplate(connectionFactory); + RabbitAdmin admin = new RabbitAdmin(connectionFactory); + Queue queue = new Queue(CF_INTEGRATION_TEST_QUEUE); + admin.declareQueue(queue); + final String route = queue.getName(); + + final CountDownLatch latch = new CountDownLatch(1); + final CountDownLatch recoveryLatch = new CountDownLatch(1); + final RecoveryListener listener = new RecoveryListener() { + + @Override + public void handleRecoveryStarted(Recoverable recoverable) { + //NOSONAR + } + + @Override + public void handleRecovery(Recoverable recoverable) { + try { + ((Channel) recoverable).basicCancel("testHardErrorAndReconnect"); + } + catch (IOException e) { + } + recoveryLatch.countDown(); + } + + }; + try { + template.execute(channel -> { + channel.getConnection().addShutdownListener(cause -> { + logger.info("Error", cause); + latch.countDown(); + // This will be thrown on the Connection thread just before it dies, so basically ignored + throw new RuntimeException(cause); + }); + Channel targetChannel = ((ChannelProxy) channel).getTargetChannel(); + if (targetChannel instanceof AutorecoveringChannel) { + ((AutorecoveringChannel) targetChannel).addRecoveryListener(listener); + } + else { + fail("Expected an AutorecoveringChannel"); + } + String tag = channel.basicConsume(route, false, "testHardErrorAndReconnect", + new DefaultConsumer(channel)); + // Consume twice with the same tag is a hard error (connection will be reset) + String result = channel.basicConsume(route, false, tag, new DefaultConsumer(channel)); + fail("Expected IOException, got: " + result); + return null; + }); + fail("Expected AmqpIOException"); + } + catch (AmqpException e) { + // expected + } + assertTrue(recoveryLatch.await(10, TimeUnit.SECONDS)); + template.convertAndSend(route, "message"); + assertTrue(latch.await(10, TimeUnit.SECONDS)); + String result = (String) template.receiveAndConvert(route); + assertEquals("message", result); + result = (String) template.receiveAndConvert(route); + assertEquals(null, result); + } + @Test public void testConnectionCloseLog() { Log logger = spy(TestUtils.getPropertyValue(this.connectionFactory, "logger", Log.class)); diff --git a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/connection/ClientRecoveryCompatibilityTests.java b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/connection/ClientRecoveryCompatibilityTests.java index 52cecacc..f82e7eb2 100644 --- a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/connection/ClientRecoveryCompatibilityTests.java +++ b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/connection/ClientRecoveryCompatibilityTests.java @@ -16,7 +16,10 @@ package org.springframework.amqp.rabbit.connection; +import static org.hamcrest.Matchers.equalTo; import static org.junit.Assert.assertSame; +import static org.junit.Assert.assertThat; +import static org.junit.Assert.fail; import static org.mockito.Matchers.any; import static org.mockito.Mockito.doAnswer; import static org.mockito.Mockito.mock; @@ -74,6 +77,13 @@ public class ClientRecoveryCompatibilityTests { when(rabbitConn.isOpen()).thenReturn(false).thenReturn(true); when(channel1.isOpen()).thenReturn(false); conn2 = ccf.createConnection(); + try { + conn2.createChannel(false); + fail("Expected AutoRecoverConnectionNotCurrentlyOpenException"); + } + catch (AutoRecoverConnectionNotCurrentlyOpenException e) { + assertThat(e.getMessage(), equalTo("Auto recovery connection is not currently open")); + } channel = conn2.createChannel(false); verifyChannelIs(channel2, channel); channel.close(); diff --git a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/RabbitAdminIntegrationTests.java b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/RabbitAdminIntegrationTests.java index 0973994b..92d8aec5 100644 --- a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/RabbitAdminIntegrationTests.java +++ b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/RabbitAdminIntegrationTests.java @@ -46,6 +46,7 @@ import org.springframework.amqp.core.MessageBuilder; import org.springframework.amqp.core.MessagePostProcessor; import org.springframework.amqp.core.MessageProperties; import org.springframework.amqp.core.Queue; +import org.springframework.amqp.rabbit.connection.AutoRecoverConnectionNotCurrentlyOpenException; import org.springframework.amqp.rabbit.connection.CachingConnectionFactory; import org.springframework.amqp.rabbit.connection.RabbitUtils; import org.springframework.amqp.rabbit.test.BrokerRunning; @@ -386,6 +387,9 @@ public class RabbitAdminIntegrationTests { throw e; } } + catch (AutoRecoverConnectionNotCurrentlyOpenException e) { + Assume.assumeTrue("Broker does not have the delayed message exchange plugin installed", false); + } this.rabbitAdmin.declareQueue(queue); this.rabbitAdmin.declareBinding(binding); diff --git a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/RabbitAdminTests.java b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/RabbitAdminTests.java index 486b69af..7c08b93a 100644 --- a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/RabbitAdminTests.java +++ b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/RabbitAdminTests.java @@ -119,7 +119,7 @@ public class RabbitAdminTests { @Test public void testFailOnFirstUseWithMissingBroker() throws Exception { - SingleConnectionFactory connectionFactory = new SingleConnectionFactory("foo"); + SingleConnectionFactory connectionFactory = new SingleConnectionFactory("localhost"); connectionFactory.setPort(434343); GenericApplicationContext applicationContext = new GenericApplicationContext(); applicationContext.getBeanFactory().registerSingleton("foo", new Queue("queue")); diff --git a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/RabbitTemplateIntegrationTests.java b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/RabbitTemplateIntegrationTests.java index 86b6da4e..c150e747 100644 --- a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/RabbitTemplateIntegrationTests.java +++ b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/RabbitTemplateIntegrationTests.java @@ -1061,6 +1061,7 @@ public class RabbitTemplateIntegrationTests { assertTrue(received); Message receive = this.template.receive(); + assertNotNull(receive); assertEquals("bar", receive.getMessageProperties().getHeaders().get("foo")); this.template.convertAndSend(ROUTE, 1); diff --git a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/RabbitTemplatePublisherCallbacksIntegrationTests2.java b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/RabbitTemplatePublisherCallbacksIntegrationTests2.java index 6d2f2c44..b0fc097c 100644 --- a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/RabbitTemplatePublisherCallbacksIntegrationTests2.java +++ b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/RabbitTemplatePublisherCallbacksIntegrationTests2.java @@ -18,7 +18,6 @@ package org.springframework.amqp.rabbit.core; import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertTrue; -import static org.mockito.Mockito.mock; import java.io.IOException; import java.util.concurrent.CountDownLatch; @@ -26,15 +25,12 @@ import java.util.concurrent.TimeUnit; import org.junit.After; import org.junit.Before; -import org.junit.BeforeClass; import org.junit.Rule; import org.junit.Test; import org.springframework.amqp.rabbit.connection.CachingConnectionFactory; -import org.springframework.amqp.rabbit.support.PublisherCallbackChannelImpl; import org.springframework.amqp.rabbit.test.BrokerRunning; import org.springframework.amqp.rabbit.test.BrokerTestUtils; -import org.springframework.beans.DirectFieldAccessor; import com.rabbitmq.client.AMQP.BasicProperties; import com.rabbitmq.client.Channel; @@ -58,12 +54,6 @@ public class RabbitTemplatePublisherCallbacksIntegrationTests2 { @Rule public BrokerRunning brokerIsRunning = BrokerRunning.isRunningWithEmptyQueues(ROUTE); - @BeforeClass - public static void setup() { - new DirectFieldAccessor(new PublisherCallbackChannelImpl(mock(Channel.class))) - .setPropertyValue("conditionalMethodsChecked", false); - } - @Before public void create() { connectionFactoryWithConfirmsEnabled = new CachingConnectionFactory(); diff --git a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/MessageListenerRecoveryCachingConnectionIntegrationTests.java b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/MessageListenerRecoveryCachingConnectionIntegrationTests.java index 9e25b488..9af81a50 100644 --- a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/MessageListenerRecoveryCachingConnectionIntegrationTests.java +++ b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/MessageListenerRecoveryCachingConnectionIntegrationTests.java @@ -95,7 +95,7 @@ public class MessageListenerRecoveryCachingConnectionIntegrationTests { @Rule public BrokerRunning brokerIsRunning = BrokerRunning.isRunningWithEmptyQueues(queue, sendQueue); - protected ConnectionFactory createConnectionFactory() { + protected CachingConnectionFactory createConnectionFactory() { CachingConnectionFactory connectionFactory = new CachingConnectionFactory(); connectionFactory.setHost("localhost"); connectionFactory.setChannelCacheSize(concurrentConsumers); @@ -273,7 +273,9 @@ public class MessageListenerRecoveryCachingConnectionIntegrationTests { RabbitTemplate template = new RabbitTemplate(connectionFactory1); CountDownLatch latch = new CountDownLatch(messageCount); - ConnectionFactory connectionFactory2 = createConnectionFactory(); + CachingConnectionFactory connectionFactory2 = createConnectionFactory(); + // this test closes the underlying connection normally; it won't automatically recover. + connectionFactory2.getRabbitConnectionFactory().setAutomaticRecoveryEnabled(false); container = createContainer(queue.getName(), new CloseConnectionListener((ConnectionProxy) connectionFactory2.createConnection(), latch), connectionFactory2); diff --git a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/SimpleMessageListenerContainerIntegration2Tests.java b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/SimpleMessageListenerContainerIntegration2Tests.java index a2b76d43..70001a4e 100644 --- a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/SimpleMessageListenerContainerIntegration2Tests.java +++ b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/SimpleMessageListenerContainerIntegration2Tests.java @@ -446,7 +446,10 @@ public class SimpleMessageListenerContainerIntegration2Tests { public void testRestartConsumerOnConnectionLossDuringQueueDeclare() throws Exception { this.template.convertAndSend(queue.getName(), "foo"); - ConnectionFactory connectionFactory = new CachingConnectionFactory("localhost", BrokerTestUtils.getPort()); + CachingConnectionFactory connectionFactory = new CachingConnectionFactory("localhost", + BrokerTestUtils.getPort()); + // this test closes the underlying connection normally; it will never be recovered + connectionFactory.getRabbitConnectionFactory().setAutomaticRecoveryEnabled(false); final AtomicBoolean networkGlitch = new AtomicBoolean(); diff --git a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/UnackedRawIntegrationTests.java b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/UnackedRawIntegrationTests.java index 33d9cefc..cb159541 100644 --- a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/UnackedRawIntegrationTests.java +++ b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/UnackedRawIntegrationTests.java @@ -19,20 +19,25 @@ package org.springframework.amqp.rabbit.listener; import static org.junit.Assert.assertNotNull; import java.io.IOException; +import java.util.concurrent.BlockingQueue; +import java.util.concurrent.LinkedBlockingQueue; +import java.util.concurrent.TimeUnit; import org.junit.After; import org.junit.Before; import org.junit.Ignore; import org.junit.Test; +import org.springframework.amqp.rabbit.support.Delivery; import org.springframework.amqp.rabbit.test.BrokerTestUtils; +import com.rabbitmq.client.AMQP.BasicProperties; import com.rabbitmq.client.Channel; import com.rabbitmq.client.Connection; import com.rabbitmq.client.ConnectionFactory; +import com.rabbitmq.client.DefaultConsumer; +import com.rabbitmq.client.Envelope; import com.rabbitmq.client.GetResponse; -import com.rabbitmq.client.QueueingConsumer; -import com.rabbitmq.client.QueueingConsumer.Delivery; /** * Used to verify raw Rabbit Java Client behaviour for corner cases. @@ -95,9 +100,9 @@ public class UnackedRawIntegrationTests { noTxChannel.basicPublish("", "test.queue", null, "foo".getBytes()); - QueueingConsumer callback = new QueueingConsumer(txChannel); + BlockingConsumer callback = new BlockingConsumer(txChannel); txChannel.basicConsume("test.queue", callback); - Delivery next = callback.nextDelivery(1000L); + Delivery next = callback.nextDelivery(10_000L); assertNotNull(next); txChannel.basicReject(next.getEnvelope().getDeliveryTag(), true); txChannel.txRollback(); @@ -115,9 +120,9 @@ public class UnackedRawIntegrationTests { noTxChannel.basicPublish("", "test.queue", null, "one".getBytes()); noTxChannel.basicPublish("", "test.queue", null, "two".getBytes()); - QueueingConsumer callback = new QueueingConsumer(txChannel); + BlockingConsumer callback = new BlockingConsumer(txChannel); txChannel.basicConsume("test.queue", callback); - Delivery next = callback.nextDelivery(1000L); + Delivery next = callback.nextDelivery(10_000L); assertNotNull(next); txChannel.basicReject(next.getEnvelope().getDeliveryTag(), true); txChannel.txRollback(); @@ -127,4 +132,30 @@ public class UnackedRawIntegrationTests { } + public class BlockingConsumer extends DefaultConsumer { + + private final BlockingQueue queue = new LinkedBlockingQueue<>(); + + public BlockingConsumer(Channel channel) { + super(channel); + } + + public Delivery nextDelivery(long timeout) throws InterruptedException { + return this.queue.poll(timeout, TimeUnit.MILLISECONDS); + } + + @Override + public void handleDelivery(String consumerTag, Envelope envelope, BasicProperties properties, byte[] body) + throws IOException { + try { + this.queue.put(new Delivery(consumerTag, envelope, properties, body)); + } + catch (InterruptedException e) { + Thread.currentThread().interrupt(); + throw new IOException(e); + } + } + + } + } diff --git a/src/reference/asciidoc/amqp.adoc b/src/reference/asciidoc/amqp.adoc index a162a6e0..3cde2efd 100644 --- a/src/reference/asciidoc/amqp.adoc +++ b/src/reference/asciidoc/amqp.adoc @@ -644,6 +644,17 @@ The `cacheMode` property (`CHANNEL` or `CONNECTION` is also included). .JVisualVM Example image::images/cacheStats.png[align="center"] +[[auto-recovery]] +===== RabbitMQ Automatic Connection/Topology recovery + +Since the first version of Spring AMQP, the framework has provided its own connection and channel recovery in the event of a broker failure. +Also, as discussed in <>, the `RabbitAdmin` will re-declare any infrastructure beans (queues etc) when the connection is re-established. +It therefore does not rely on the https://www.rabbitmq.com/api-guide.html#recovery[Auto Recovery] that is now provided by the `amqp-client` library. +Spring AMQP now uses the `4.0.x` version of `amqp-client`, which has auto recovery enabled by default. +Spring AMQP can still use its own recovery mechanisms if you wish, disabling it in the client, (by setting the `automaticRecoveryEnabled` property on the underlying `RabbitMQ connectionFactory` to `false`). +However, the framework is completely compatible with auto recovery being enabled. +This means any consumers you create within your code (perhaps via `RabbitTemplate.execute()`) can be recovered automatically. + [[custom-client-props]] ==== Adding Custom Client Connection Properties @@ -2808,8 +2819,8 @@ See <> for more information. ===== Introduction The AMQP specification describes how the protocol can be used to configure Queues, Exchanges and Bindings on the broker. -These operations which are portable from the 0.8 specification and higher are present in the AmqpAdmin interface in the org.springframework.amqp.core package. -The RabbitMQ implementation of that class is RabbitAdmin located in the org.springframework.amqp.rabbit.core package. +These operations which are portable from the 0.8 specification and higher are present in the `AmqpAdmin` interface in the `org.springframework.amqp.core` package. +The RabbitMQ implementation of that class is `RabbitAdmin` located in the `org.springframework.amqp.rabbit.core` package. The AmqpAdmin interface is based on using the Spring AMQP domain abstractions and is shown below: @@ -4026,7 +4037,7 @@ It does this lazily, through a `ConnectionListener`, so if the broker is not pre The first time a `Connection` is used (e.g. by sending a message) the listener will fire and the admin features will be applied. A further benefit of doing the auto declarations in a listener is that if the connection is dropped for any reason (e.g. -broker death, network glitch, etc.) they will be applied again the next time they are needed. +broker death, network glitch, etc.) they will be applied again when the connection is re-established. NOTE: Queues declared this way must have fixed names; either explicitly declared, or generated by the framework for `AnonymousQueue` s. Anonymous queues are non-durable, exclusive, and auto-delete. @@ -4034,6 +4045,8 @@ Anonymous queues are non-durable, exclusive, and auto-delete. IMPORTANT: Automatic declaration is only performed when the `CachingConnectionFactory` cache mode is `CHANNEL` (the default). This limitation exists because exclusive and auto-delete queues are bound to the connection. +See also <>. + [[retry]] ===== Failures in Synchronous Operations and Options for Retry diff --git a/src/reference/asciidoc/quick-tour.adoc b/src/reference/asciidoc/quick-tour.adoc index d442b363..8f113dba 100644 --- a/src/reference/asciidoc/quick-tour.adoc +++ b/src/reference/asciidoc/quick-tour.adoc @@ -32,8 +32,8 @@ While the default Spring Framework version dependency is 4.3.x, Spring AMQP is g versions of Spring Framework. Annotation-based listeners and the `RabbitMessagingTemplate` require Spring Framework 4.1 or higher, however. -Similarly, the default `amqp-client` version is 3.6.x but the framework is compatible with versions 3.4.0 and above. -However, of course, features that rely on newer client versions will not be available. +The minimum `amqp-client` java client library version is 4.0.0. + Note the this refers to the java client library; generally, it will work with older broker versions. ===== Very, Very Quick diff --git a/src/reference/asciidoc/whats-new.adoc b/src/reference/asciidoc/whats-new.adoc index 2cac992e..1174a2e7 100644 --- a/src/reference/asciidoc/whats-new.adoc +++ b/src/reference/asciidoc/whats-new.adoc @@ -3,6 +3,11 @@ ==== Changes in 1.7 Since 1.6 +===== AMQP Client library + +Spring AMQP now uses the new 4.0.x version of the `amqp-client` library provided by the RabbitMQ team. +This client has auto recovery configured by default; see <>. + ===== Log4j2 upgrade The minimum Log4j2 version (for the `AmqpAppender`) is now `2.7`.