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

AMQP-266/296/297 Allow Admin to Continue After Err

Browse Source 
* AMQP-296 Update Spring Framework dependency version to 3.1.3
* AMQP-266 Add option to allow Admin to continue after exception.

Currently, if the Rabbit Admin hits an exception while declaring Queues etc, it aborts the entire process. An exception can be caused if, say, the 'x-message-ttl` attribute is mismatched with that currently configured on the broker.

* Add an option that allows declarations to continue after exceptions, while logging (WARN) any exceptions encountered.
* AMQP-297 Add RabbitAdmin.getQueueProperties()
* Used in the above test to verify presence, or otherwise of queues during Admin declarations.
* Returns message and consumer counts for the queue.
* Bump rabbit-client dependency version to 3.0.4
This commit is contained in:
Gary Russell
2013-03-07 18:51:24 -05:00
committed by Gunnar Hillert
parent e2fc633d69
commit 0a60a902bd
10 changed files with 320 additions and 18 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
build.gradle
View File

@@ -45,9 +45,9 @@ subprojects { subproject ->
junitVersion = '4.8.2'
log4jVersion = '1.2.15'
mockitoVersion = '1.8.4'
rabbitmqVersion = '3.0.3'
rabbitmqVersion = '3.0.4'
springVersion = '3.0.7.RELEASE'
springVersion = '3.1.4.RELEASE'
}
eclipse {

13
spring-amqp/src/main/java/org/springframework/amqp/core/AmqpAdmin.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2010 the original author or authors.
* Copyright 2002-2013 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
@@ -13,11 +13,15 @@
package org.springframework.amqp.core;
import java.util.Properties;
/**
* Specifies a basic set of portable AMQP administrative operations for AMQP > 0.8
*
* @author Mark Pollack
* @author Dave Syer
* @author Gary Russell
*/
public interface AmqpAdmin {
@@ -88,4 +92,11 @@ public interface AmqpAdmin {
*/
void removeBinding(Binding binding);
/**
* Returns an implementation-specific Map of properties if the queue exists.
* @param queueName the name of the queue.
* @return the properties or null if the queue doesn't exist
*/
Properties getQueueProperties(String queueName);
}

6
spring-amqp/src/main/java/org/springframework/amqp/support/converter/MarshallingMessageConverter.java
View File

@@ -39,8 +39,8 @@ import org.springframework.util.Assert;
* @author Arjen Poutsma
* @author Juergen Hoeller
* @author James Carr
* @see org.springframework.amqp.rabbit.core.RabbitTemplate#convertAndSend
* @see org.springframework.amqp.rabbit.core.RabbitTemplate#receiveAndConvert
* @see org.springframework.amqp.core.AmqpTemplate#convertAndSend(Object)
* @see org.springframework.amqp.core.AmqpTemplate#receiveAndConvert()
*/
public class MarshallingMessageConverter extends AbstractMessageConverter implements InitializingBean {
private volatile Marshaller marshaller;
@@ -128,6 +128,7 @@ public class MarshallingMessageConverter extends AbstractMessageConverter implem
/**
* Marshals the given object to a {@link Message}.
*/
@Override
protected Message createMessage(Object object, MessageProperties messageProperties) throws MessageConversionException {
try {
if (contentType != null) {
@@ -149,6 +150,7 @@ public class MarshallingMessageConverter extends AbstractMessageConverter implem
/**
* Unmarshals the given {@link Message} into an object.
*/
@Override
public Object fromMessage(Message message) throws MessageConversionException {
try {
ByteArrayInputStream bis = new ByteArrayInputStream(message.getBody());

4
spring-rabbit/src/main/java/org/springframework/amqp/rabbit/config/AdminParser.java
View File

@@ -29,6 +29,8 @@ class AdminParser extends AbstractSingleBeanDefinitionParser {
private static final String AUTO_STARTUP_ATTRIBUTE = "auto-startup";
private static final String IGNORE_DECLARATION_EXCEPTIONS = "ignore-declaration-exceptions";
@Override
protected String getBeanClassName(Element element) {
return "org.springframework.amqp.rabbit.core.RabbitAdmin";
@@ -64,5 +66,7 @@ class AdminParser extends AbstractSingleBeanDefinitionParser {
if (StringUtils.hasText(attributeValue)) {
builder.addPropertyValue("autoStartup", attributeValue);
}
NamespaceUtils.setValueIfAttributeDefined(builder, element, IGNORE_DECLARATION_EXCEPTIONS);
}
}

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

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2010 the original author or authors.
* Copyright 2002-2013 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
@@ -15,6 +15,7 @@ package org.springframework.amqp.rabbit.core;
import java.io.IOException;
import java.util.Collection;
import java.util.Properties;
import java.util.concurrent.atomic.AtomicBoolean;
import org.apache.commons.logging.Log;
@@ -42,11 +43,18 @@ import com.rabbitmq.client.Channel;
* @author Mark Fisher
* @author Dave Syer
* @author Ed Scriven
* @author Gary Russell
*/
public class RabbitAdmin implements AmqpAdmin, ApplicationContextAware, InitializingBean {
protected static final String DEFAULT_EXCHANGE_NAME = "";
protected static final Object QUEUE_NAME = "QUEUE_NAME";
protected static final Object QUEUE_MESSAGE_COUNT = "QUEUE_MESSAGE_COUNT";
protected static final Object QUEUE_CONSUMER_COUNT = "QUEUE_CONSUMER_COUNT";
/** Logger available to subclasses */
protected final Log logger = LogFactory.getLog(getClass());
@@ -58,6 +66,8 @@ public class RabbitAdmin implements AmqpAdmin, ApplicationContextAware, Initiali
private volatile ApplicationContext applicationContext;
private volatile boolean ignoreDeclarationExceptions;
private final Object lifecycleMonitor = new Object();
private final ConnectionFactory connectionFactory;
@@ -76,6 +86,10 @@ public class RabbitAdmin implements AmqpAdmin, ApplicationContextAware, Initiali
this.applicationContext = applicationContext;
}
public void setIgnoreDeclarationExceptions(boolean ignoreDeclarationExceptions) {
this.ignoreDeclarationExceptions = ignoreDeclarationExceptions;
}
public RabbitTemplate getRabbitTemplate() {
return this.rabbitTemplate;
}
@@ -200,6 +214,32 @@ public class RabbitAdmin implements AmqpAdmin, ApplicationContextAware, Initiali
});
}
/**
* Returns 3 properties {@link #QUEUE_NAME}, {@link #QUEUE_MESSAGE_COUNT},
* {@link #QUEUE_CONSUMER_COUNT}, or null if the queue doesn't exist.
*/
public Properties getQueueProperties(final String queueName) {
Assert.hasText(queueName, "'queueName' cannot be null or empty");
return this.rabbitTemplate.execute(new ChannelCallback<Properties>() {
public Properties doInRabbit(Channel channel) throws Exception {
try {
DeclareOk declareOk = channel.queueDeclarePassive(queueName);
Properties props = new Properties();
props.put(QUEUE_NAME, declareOk.getQueue());
props.put(QUEUE_MESSAGE_COUNT, declareOk.getMessageCount());
props.put(QUEUE_CONSUMER_COUNT, declareOk.getConsumerCount());
return props;
}
catch (Exception e) {
if (logger.isDebugEnabled()) {
logger.debug("Queue '" + queueName + "' does not exist");
}
return null;
}
}
});
}
// Lifecycle implementation
public boolean isAutoStartup() {
@@ -226,7 +266,7 @@ public class RabbitAdmin implements AmqpAdmin, ApplicationContextAware, Initiali
connectionFactory.addConnectionListener(new ConnectionListener() {
// Prevent stack overflow...
private AtomicBoolean initializing = new AtomicBoolean(false);
private final AtomicBoolean initializing = new AtomicBoolean(false);
public void onCreate(Connection connection) {
if (!initializing.compareAndSet(false, true)) {
@@ -327,8 +367,20 @@ public class RabbitAdmin implements AmqpAdmin, ApplicationContextAware, Initiali
}
if (!isDeclaringDefaultExchange(exchange)) {
channel.exchangeDeclare(exchange.getName(), exchange.getType(), exchange.isDurable(),
try {
channel.exchangeDeclare(exchange.getName(), exchange.getType(), exchange.isDurable(),
exchange.isAutoDelete(), exchange.getArguments());
}
catch (IOException e) {
if (this.ignoreDeclarationExceptions) {
if (logger.isWarnEnabled()) {
logger.warn("Failed to declare exchange:" + exchange + ", continuing...", e);
}
}
else {
throw e;
}
}
}
}
}
@@ -339,8 +391,20 @@ public class RabbitAdmin implements AmqpAdmin, ApplicationContextAware, Initiali
if (logger.isDebugEnabled()) {
logger.debug("declaring Queue '" + queue.getName() + "'");
}
channel.queueDeclare(queue.getName(), queue.isDurable(), queue.isExclusive(), queue.isAutoDelete(),
queue.getArguments());
try {
channel.queueDeclare(queue.getName(), queue.isDurable(), queue.isExclusive(), queue.isAutoDelete(),
queue.getArguments());
}
catch (IOException e) {
if (this.ignoreDeclarationExceptions) {
if (logger.isWarnEnabled()) {
logger.warn("Failed to declare queue:" + queue + ", continuing...", e);
}
}
else {
throw e;
}
}
} else if (logger.isDebugEnabled()) {
logger.debug("Queue with name that starts with 'amq.' cannot be declared.");
}
@@ -355,14 +419,26 @@ public class RabbitAdmin implements AmqpAdmin, ApplicationContextAware, Initiali
+ "]");
}
if (binding.isDestinationQueue()) {
if (!isDeclaringImplicitQueueBinding(binding)) {
channel.queueBind(binding.getDestination(), binding.getExchange(), binding.getRoutingKey(),
try {
if (binding.isDestinationQueue()) {
if (!isDeclaringImplicitQueueBinding(binding)) {
channel.queueBind(binding.getDestination(), binding.getExchange(), binding.getRoutingKey(),
binding.getArguments());
}
} else {
channel.exchangeBind(binding.getDestination(), binding.getExchange(), binding.getRoutingKey(),
binding.getArguments());
}
} else {
channel.exchangeBind(binding.getDestination(), binding.getExchange(), binding.getRoutingKey(),
binding.getArguments());
}
catch (IOException e) {
if (this.ignoreDeclarationExceptions) {
if (logger.isWarnEnabled()) {
logger.warn("Failed to declare binding:" + binding + ", continuing...", e);
}
}
else {
throw e;
}
}
}
}

3
spring-rabbit/src/main/resources/META-INF/spring.schemas
View File

@@ -1,3 +1,4 @@
http\://www.springframework.org/schema/rabbit/spring-rabbit-1.2.xsd=org/springframework/amqp/rabbit/config/spring-rabbit-1.2.xsd
http\://www.springframework.org/schema/rabbit/spring-rabbit-1.1.xsd=org/springframework/amqp/rabbit/config/spring-rabbit-1.1.xsd
http\://www.springframework.org/schema/rabbit/spring-rabbit-1.0.xsd=org/springframework/amqp/rabbit/config/spring-rabbit-1.0.xsd
http\://www.springframework.org/schema/rabbit/spring-rabbit.xsd=org/springframework/amqp/rabbit/config/spring-rabbit-1.1.xsd
http\://www.springframework.org/schema/rabbit/spring-rabbit.xsd=org/springframework/amqp/rabbit/config/spring-rabbit-1.2.xsd

17
spring-rabbit/src/main/resources/org/springframework/amqp/rabbit/config/spring-rabbit-1.2.xsd
View File

@@ -768,12 +768,27 @@
</xsd:appinfo>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="auto-startup" type="xsd:boolean">
<xsd:attribute name="auto-startup" default="true">
<xsd:annotation>
<xsd:documentation><![CDATA[
Specifies if the queues, exchanges and bindings in the context should be automatically declared (lazily on first connection to the broker). Default value is 'true'.
]]></xsd:documentation>
</xsd:annotation>
<xsd:simpleType>
<xsd:union memberTypes="xsd:boolean xsd:string" />
</xsd:simpleType>
</xsd:attribute>
<xsd:attribute name="ignore-declaration-exceptions" default="false">
<xsd:annotation>
<xsd:documentation><![CDATA[
If automatic declaration is enabled (see 'auto-startup'), if this is set to 'true', exceptions will
be logged (WARNings) but declaration of other elements will continue. If false, declarations will
cease when an exception occurs. Default value is 'false'.
]]></xsd:documentation>
</xsd:annotation>
<xsd:simpleType>
<xsd:union memberTypes="xsd:boolean xsd:string" />
</xsd:simpleType>
</xsd:attribute>
</xsd:complexType>
</xsd:element>

125
spring-rabbit/src/test/java/org/springframework/amqp/rabbit/config/MismatchedQueueDeclarationTests.java Normal file
View File

@@ -0,0 +1,125 @@
/*
* Copyright 2002-2013 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.config;
import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertNull;
import static org.junit.Assert.assertTrue;
import static org.junit.Assert.fail;
import org.junit.After;
import org.junit.Before;
import org.junit.Rule;
import org.junit.Test;
import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
import org.springframework.amqp.rabbit.connection.SingleConnectionFactory;
import org.springframework.amqp.rabbit.core.RabbitAdmin;
import org.springframework.amqp.rabbit.test.BrokerRunning;
import org.springframework.context.support.ClassPathXmlApplicationContext;
import org.springframework.core.env.StandardEnvironment;
import com.rabbitmq.client.Channel;
/**
* @author Gary Russell
* @since 1.2
*
*/
public class MismatchedQueueDeclarationTests {
@Rule
public BrokerRunning brokerIsRunning = BrokerRunning.isRunning();
private SingleConnectionFactory connectionFactory;
private RabbitAdmin admin;
@Before
public void setup() {
connectionFactory = new SingleConnectionFactory();
this.admin = new RabbitAdmin(this.connectionFactory);
deleteQueues();
}
@After
public void deleteQueues() {
this.admin.deleteQueue("mismatch.foo");
this.admin.deleteQueue("mismatch.bar");
}
@Test
public void testAdminFailsWithMismatchedQueue() throws Exception {
ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext();
context.setConfigLocation("org/springframework/amqp/rabbit/config/MismatchedQueueDeclarationTests-context.xml");
StandardEnvironment env = new StandardEnvironment();
env.addActiveProfile("basicAdmin");
env.addActiveProfile("basic");
context.setEnvironment(env);
context.refresh();
context.getBean(CachingConnectionFactory.class).createConnection();
context.destroy();
Channel channel = this.connectionFactory.createConnection().createChannel(false);
channel.queueDeclarePassive("mismatch.bar");
this.admin.deleteQueue("mismatch.bar");
assertNotNull(this.admin.getQueueProperties("mismatch.foo"));
assertNull(this.admin.getQueueProperties("mismatch.bar"));
env = new StandardEnvironment();
env.addActiveProfile("basicAdmin");
env.addActiveProfile("ttl");
context.setEnvironment(env);
context.refresh();
channel = this.connectionFactory.createConnection().createChannel(false);
try {
context.getBean(CachingConnectionFactory.class).createConnection();
fail("Expected exception - basic admin fails with mismatched declarations");
}
catch (Exception e) {
assertTrue(e.getCause().getCause().getMessage().contains("inequivalent arg 'x-message-ttl'"));
}
assertNotNull(this.admin.getQueueProperties("mismatch.foo"));
assertNull(this.admin.getQueueProperties("mismatch.bar"));
}
@Test
public void testAdminSkipsMismatchedQueue() throws Exception {
ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext();
context.setConfigLocation("org/springframework/amqp/rabbit/config/MismatchedQueueDeclarationTests-context.xml");
StandardEnvironment env = new StandardEnvironment();
env.addActiveProfile("advancedAdmin");
env.addActiveProfile("basic");
context.setEnvironment(env);
context.refresh();
context.getBean(CachingConnectionFactory.class).createConnection();
context.destroy();
Channel channel = this.connectionFactory.createConnection().createChannel(false);
channel.queueDeclarePassive("mismatch.bar");
this.admin.deleteQueue("mismatch.bar");
assertNotNull(this.admin.getQueueProperties("mismatch.foo"));
assertNull(this.admin.getQueueProperties("mismatch.bar"));
env = new StandardEnvironment();
env.addActiveProfile("advancedAdmin");
env.addActiveProfile("ttl");
context.setEnvironment(env);
context.refresh();
channel = this.connectionFactory.createConnection().createChannel(false);
context.getBean(CachingConnectionFactory.class).createConnection();
assertNotNull(this.admin.getQueueProperties("mismatch.foo"));
assertNotNull(this.admin.getQueueProperties("mismatch.bar"));
}
}

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

@@ -1,7 +1,11 @@
package org.springframework.amqp.rabbit.core;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.fail;
import java.util.Properties;
import org.junit.Rule;
import org.junit.Test;
import org.junit.rules.ExpectedException;
@@ -10,6 +14,9 @@ import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.connection.SingleConnectionFactory;
import org.springframework.context.support.GenericApplicationContext;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.DefaultConsumer;
public class RabbitAdminTests {
@Rule
@@ -52,4 +59,33 @@ public class RabbitAdminTests {
rabbitAdmin.declareQueue();
}
@Test
public void testProperties() throws Exception {
SingleConnectionFactory connectionFactory = new SingleConnectionFactory();
RabbitAdmin rabbitAdmin = new RabbitAdmin(connectionFactory);
String queueName = "test.properties." + System.currentTimeMillis();
try {
rabbitAdmin.declareQueue(new Queue(queueName));
new RabbitTemplate(connectionFactory).convertAndSend(queueName, "foo");
Properties props = rabbitAdmin.getQueueProperties(queueName);
assertNotNull(props);
assertNotNull(props.get(RabbitAdmin.QUEUE_MESSAGE_COUNT));
assertEquals(1, props.get(RabbitAdmin.QUEUE_MESSAGE_COUNT));
Channel channel = connectionFactory.createConnection().createChannel(false);
DefaultConsumer consumer = new DefaultConsumer(channel);
channel.basicConsume(queueName, true, consumer);
props = rabbitAdmin.getQueueProperties(queueName);
assertNotNull(props);
assertNotNull(props.get(RabbitAdmin.QUEUE_MESSAGE_COUNT));
assertEquals(0, props.get(RabbitAdmin.QUEUE_MESSAGE_COUNT));
assertNotNull(props.get(RabbitAdmin.QUEUE_CONSUMER_COUNT));
assertEquals(1, props.get(RabbitAdmin.QUEUE_CONSUMER_COUNT));
channel.close();
}
finally {
rabbitAdmin.deleteQueue(queueName);
connectionFactory.destroy();
}
}
}

32
spring-rabbit/src/test/resources/org/springframework/amqp/rabbit/config/MismatchedQueueDeclarationTests-context.xml Normal file
View File

@@ -0,0 +1,32 @@
<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xmlns:rabbit="http://www.springframework.org/schema/rabbit"
xsi:schemaLocation="http://www.springframework.org/schema/rabbit http://www.springframework.org/schema/rabbit/spring-rabbit.xsd
http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd">
<rabbit:connection-factory id="connectionFactory" />
<beans profile="basicAdmin">
<rabbit:admin connection-factory="connectionFactory" />
</beans>
<beans profile="advancedAdmin">
<rabbit:admin connection-factory="connectionFactory" ignore-declaration-exceptions="true" />
</beans>
<beans profile="basic">
<rabbit:queue name="mismatch.foo" durable="true" />
<rabbit:queue name="mismatch.bar" durable="true" />
</beans>
<beans profile="ttl">
<rabbit:queue name="mismatch.foo" durable="true">
<rabbit:queue-arguments>
<entry key="x-message-ttl" value="1000" />
</rabbit:queue-arguments>
</rabbit:queue>
<rabbit:queue name="mismatch.bar" durable="true" />
</beans>
</beans>