Spring/spring-amqp
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Update to amqp-client Version 4.0.0.M2

Browse Source 
Polishing

On linux, non-existent exchange still throws IOException.

* Add assertion to the expected exceptions in the `CachingConnectionFactoryIntegrationTests.testReceiveFromNonExistentVirtualHost()`

Conflicts:
	build.gradle
	src/reference/asciidoc/quick-tour.adoc

AMQP-673: 4.0.0.RC2 Client and auto-recovery

JIRA: https://jira.spring.io/browse/AMQP-673

Remove Use of Deprecated QueueingConsumer

- Make tests compatible with autorecovery which is now true by default.

- SimpleConnection: throw an exception in isOpen() if an auto recovery connection is currently closed.

Polishing - PR Comments

Return isOpen() = false if the SimpleConnection has been explicitly closed.

Fix RabbitAdminTests

When detecting if the broker supports x-delayed-exhange, and an auto-recovered
connection is being used, the connection is closed (and eventually reopens) if
the delayed exchange declaration fails.

Throw a new exception type which can be used to signal to the test that it should be ignored.

Remove log adjuster and SOUT

Polishing and Docs

Remove bogus @Rule

Polishing

Conflicts:
	build.gradle
	spring-rabbit/src/main/java/org/springframework/amqp/rabbit/core/RabbitTemplate.java
	spring-rabbit/src/test/java/org/springframework/amqp/rabbit/connection/CachingConnectionFactoryIntegrationTests.java
	src/reference/asciidoc/whats-new.adoc
Resolved.

Polishing

Fix Test

Fix `@since` to `1.7`
This commit is contained in:
Gary Russell
2016-10-17 15:43:01 -04:00
committed by Artem Bilan
parent 8258c2b647
commit 4efba7d5ee
21 changed files with 376 additions and 145 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
build.gradle
View File

@@ -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'

36
spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/AutoRecoverConnectionNotCurrentlyOpenException.java Normal file
View File

@@ -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);
}
}

52
spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/SimpleConnection.java
View File

@@ -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@"

2
spring-rabbit/src/main/java/org/springframework/amqp/rabbit/core/RabbitAdmin.java
View File

@@ -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)

20
spring-rabbit/src/main/java/org/springframework/amqp/rabbit/core/RabbitTemplate.java
View File

@@ -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;
/**
* <p>
@@ -861,10 +859,11 @@ public class RabbitTemplate extends RabbitAccessor implements BeanFactoryAware,
public Message receive(final String queueName, final long timeoutMillis) {
return execute(new ChannelCallback<Message>() {
@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 {

39
spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/BlockingQueueConsumer.java
View File

@@ -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 {

78
spring-rabbit/src/main/java/org/springframework/amqp/rabbit/support/Delivery.java Normal file
View File

@@ -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;
}
}

89
spring-rabbit/src/main/java/org/springframework/amqp/rabbit/support/PublisherCallbackChannelImpl.java
View File

@@ -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<String, Object> arguments)
@@ -286,6 +251,12 @@ public class PublisherCallbackChannelImpl
arguments);
}
@Override
public DeclareOk exchangeDeclare(String exchange, BuiltinExchangeType type, boolean durable, boolean autoDelete,
Map<String, Object> 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<String, Object> 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<String, Object> 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

3
spring-rabbit/src/main/java/org/springframework/amqp/rabbit/support/RabbitExceptionTranslator.java
View File

@@ -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);
}

89
spring-rabbit/src/test/java/org/springframework/amqp/rabbit/connection/CachingConnectionFactoryIntegrationTests.java
View File

@@ -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<Connection> connections = new ArrayList<Connection>();
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));

10
spring-rabbit/src/test/java/org/springframework/amqp/rabbit/connection/ClientRecoveryCompatibilityTests.java
View File

@@ -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();

4
spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/RabbitAdminIntegrationTests.java
View File

@@ -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);

2
spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/RabbitAdminTests.java
View File

@@ -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"));

1
spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/RabbitTemplateIntegrationTests.java
View File

@@ -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);

10
spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/RabbitTemplatePublisherCallbacksIntegrationTests2.java
View File

@@ -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();

6
spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/MessageListenerRecoveryCachingConnectionIntegrationTests.java
View File

@@ -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);

5
spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/SimpleMessageListenerContainerIntegration2Tests.java
View File

@@ -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();

43
spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/UnackedRawIntegrationTests.java
View File

@@ -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<Delivery> 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);
}
}
}
}

19
src/reference/asciidoc/amqp.adoc
View File

@@ -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 <<broker-configuration>>, 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 <<reply-timeout>> 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 <<auto-recovery>>.
[[retry]]
===== Failures in Synchronous Operations and Options for Retry

4
src/reference/asciidoc/quick-tour.adoc
View File

@@ -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

5
src/reference/asciidoc/whats-new.adoc
View File

@@ -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 <<auto-recovery>>.
===== Log4j2 upgrade
The minimum Log4j2 version (for the `AmqpAppender`) is now `2.7`.