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

AMQP-208 Default Exchange and Bindings

Browse Source 
https://jira.springsource.org/browse/AMQP-208

Don't attempt to declare
the default exchange or bindings to it in RabbitAdmin.

Applied same logic to delete exchange and remove binding.

Refined existing auto delete queue tests to enable re-use of
native checking to see if queue exists and correct intent in names.
This commit is contained in:
Ed Scriven
2012-01-20 12:37:40 +00:00
committed by Gary Russell
parent 8332a77b7d
commit 75208ca8e5
2 changed files with 224 additions and 87 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -41,9 +41,12 @@ import com.rabbitmq.client.Channel;
* @author Mark Pollack
* @author Mark Fisher
* @author Dave Syer
* @author Ed Scriven
*/
public class RabbitAdmin implements AmqpAdmin, ApplicationContextAware, InitializingBean {
protected static final String DEFAULT_EXCHANGE_NAME = "";
/** Logger available to subclasses */
protected final Log logger = LogFactory.getLog(getClass());
@@ -92,6 +95,10 @@ public class RabbitAdmin implements AmqpAdmin, ApplicationContextAware, Initiali
public boolean deleteExchange(final String exchangeName) {
return this.rabbitTemplate.execute(new ChannelCallback<Boolean>() {
public Boolean doInRabbit(Channel channel) throws Exception {
if (isDeletingDefaultExchange(exchangeName)) {
return true;
}
try {
channel.exchangeDelete(exchangeName);
} catch (IOException e) {
@@ -178,6 +185,10 @@ public class RabbitAdmin implements AmqpAdmin, ApplicationContextAware, Initiali
rabbitTemplate.execute(new ChannelCallback<Object>() {
public Object doInRabbit(Channel channel) throws Exception {
if (binding.isDestinationQueue()) {
if (isRemovingImplicitQueueBinding(binding)) {
return null;
}
channel.queueUnbind(binding.getDestination(), binding.getExchange(), binding.getRoutingKey(),
binding.getArguments());
} else {
@@ -314,6 +325,11 @@ public class RabbitAdmin implements AmqpAdmin, ApplicationContextAware, Initiali
if (logger.isDebugEnabled()) {
logger.debug("declaring Exchange '" + exchange.getName() + "'");
}
if (isDeclaringDefaultExchange(exchange)) {
continue;
}
channel.exchangeDeclare(exchange.getName(), exchange.getType(), exchange.isDurable(),
exchange.isAutoDelete(), exchange.getArguments());
}
@@ -340,7 +356,12 @@ public class RabbitAdmin implements AmqpAdmin, ApplicationContextAware, Initiali
+ ")] to exchange [" + binding.getExchange() + "] with routing key [" + binding.getRoutingKey()
+ "]");
}
if (binding.isDestinationQueue()) {
if (isDeclaringImplicitQueueBinding(binding)) {
continue;
}
channel.queueBind(binding.getDestination(), binding.getExchange(), binding.getRoutingKey(),
binding.getArguments());
} else {
@@ -350,4 +371,51 @@ public class RabbitAdmin implements AmqpAdmin, ApplicationContextAware, Initiali
}
}
private boolean isDeclaringDefaultExchange(Exchange exchange) {
if (isDefaultExchange(exchange.getName())) {
if (logger.isDebugEnabled()) {
logger.debug("Default exchange is pre-declared by server.");
}
return true;
}
return false;
}
private boolean isDeletingDefaultExchange(String exchangeName) {
if (isDefaultExchange(exchangeName)) {
if (logger.isDebugEnabled()) {
logger.debug("Default exchange cannot be deleted.");
}
return true;
}
return false;
}
private boolean isDefaultExchange(String exchangeName) {
return DEFAULT_EXCHANGE_NAME.equals(exchangeName);
}
private boolean isDeclaringImplicitQueueBinding(Binding binding) {
if (isImplicitQueueBinding(binding)) {
if (logger.isDebugEnabled()) {
logger.debug("The default exchange is implicitly bound to every queue, with a routing key equal to the queue name.");
}
return true;
}
return false;
}
private boolean isRemovingImplicitQueueBinding(Binding binding) {
if (isImplicitQueueBinding(binding)) {
if (logger.isDebugEnabled()) {
logger.debug("Cannot remove implicit default exchange binding to queue.");
}
return true;
}
return false;
}
private boolean isImplicitQueueBinding(Binding binding) {
return isDefaultExchange(binding.getExchange()) && binding.getDestination().equals(binding.getRoutingKey());
}
}

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

@@ -4,13 +4,16 @@ import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertTrue;
import java.io.IOException;
import java.util.concurrent.atomic.AtomicReference;
import org.junit.After;
import org.junit.Before;
import org.junit.Rule;
import org.junit.Test;
import org.springframework.amqp.AmqpIOException;
import org.springframework.amqp.core.Binding;
import org.springframework.amqp.core.Binding.DestinationType;
import org.springframework.amqp.core.DirectExchange;
import org.springframework.amqp.core.Exchange;
import org.springframework.amqp.core.Queue;
import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
import org.springframework.amqp.rabbit.test.BrokerRunning;
@@ -101,114 +104,183 @@ public class RabbitAdminIntegrationTests {
}
@Test
public void testStartupWithAutodelete() throws Exception {
public void testQueueWithAutoDelete() throws Exception {
final Queue queue = new Queue("test.queue", false, true, true);
context.getBeanFactory().registerSingleton("foo", queue);
rabbitAdmin.afterPropertiesSet();
final AtomicReference<Connection> connectionHolder = new AtomicReference<Connection>();
RabbitTemplate rabbitTemplate = new RabbitTemplate(connectionFactory);
boolean exists = rabbitTemplate.execute(new ChannelCallback<Boolean>() {
public Boolean doInRabbit(Channel channel) throws Exception {
DeclareOk result = channel.queueDeclarePassive(queue.getName());
connectionHolder.set(channel.getConnection());
return result != null;
}
});
assertTrue("Expected Queue to exist", exists);
assertTrue(queueExists(connectionHolder.get(), queue));
exists = rabbitTemplate.execute(new ChannelCallback<Boolean>() {
public Boolean doInRabbit(Channel channel) throws Exception {
DeclareOk result = channel.queueDeclarePassive(queue.getName());
connectionHolder.set(channel.getConnection());
return result != null;
}
});
assertTrue("Expected Queue to exist", exists);
// Queue created on spring startup
rabbitAdmin.initialize();
assertTrue(queueExists(queue));
// Stop and broker deletes queue (only verifiable in native API)
connectionFactory.destroy();
// Broker now deletes queue (only verifiable in native API)
assertFalse(queueExists(null, queue));
assertFalse(queueExists(queue));
// Broker auto-deleted queue, but it is re-created by the connection listener
exists = rabbitTemplate.execute(new ChannelCallback<Boolean>() {
public Boolean doInRabbit(Channel channel) throws Exception {
DeclareOk result = channel.queueDeclarePassive(queue.getName());
connectionHolder.set(channel.getConnection());
return result != null;
}
});
assertTrue("Expected Queue to exist", exists);
// Start and queue re-created by the connection listener
connectionFactory.createConnection();
assertTrue(queueExists(queue));
assertTrue(queueExists(connectionHolder.get(), queue));
// Queue manually deleted
assertTrue(rabbitAdmin.deleteQueue(queue.getName()));
assertFalse(queueExists(null, queue));
assertFalse(queueExists(queue));
}
@Test
public void testStartupWithNonDurable() throws Exception {
public void testQueueWithoutAutoDelete() throws Exception {
final Queue queue = new Queue("test.queue", false, false, false);
context.getBeanFactory().registerSingleton("foo", queue);
rabbitAdmin.afterPropertiesSet();
final AtomicReference<Connection> connectionHolder = new AtomicReference<Connection>();
// Queue created on Spring startup
rabbitAdmin.initialize();
assertTrue(queueExists(queue));
RabbitTemplate rabbitTemplate = new RabbitTemplate(connectionFactory);
// Force RabbitAdmin to initialize the queue
boolean exists = rabbitTemplate.execute(new ChannelCallback<Boolean>() {
public Boolean doInRabbit(Channel channel) throws Exception {
DeclareOk result = channel.queueDeclarePassive(queue.getName());
connectionHolder.set(channel.getConnection());
return result != null;
}
});
assertTrue("Expected Queue to exist", exists);
assertTrue(queueExists(connectionHolder.get(), queue));
// simulate broker going down and coming back up...
rabbitAdmin.deleteQueue(queue.getName());
// Stop and broker retains queue (only verifiable in native API)
connectionFactory.destroy();
assertFalse(queueExists(null, queue));
assertTrue(queueExists(queue));
// Broker auto-deleted queue, but it is re-created by the connection listener
exists = rabbitTemplate.execute(new ChannelCallback<Boolean>() {
public Boolean doInRabbit(Channel channel) throws Exception {
DeclareOk result = channel.queueDeclarePassive(queue.getName());
connectionHolder.set(channel.getConnection());
return result != null;
}
});
assertTrue("Expected Queue to exist", exists);
// Start and queue still exists
connectionFactory.createConnection();
assertTrue(queueExists(queue));
assertTrue(queueExists(connectionHolder.get(), queue));
// Queue manually deleted
assertTrue(rabbitAdmin.deleteQueue(queue.getName()));
assertFalse(queueExists(null, queue));
assertFalse(queueExists(queue));
}
@Test
public void testDeclareExchangeWithDefaultExchange() throws Exception {
Exchange exchange = new DirectExchange(RabbitAdmin.DEFAULT_EXCHANGE_NAME);
rabbitAdmin.declareExchange(exchange);
// Pass by virtue of RabbitMQ not firing a 403 reply code
}
/**
* Use native Rabbit API to test queue, bypassing all the connection and channel caching and callbacks in Spring
* AMQP.
*
* @param connection the raw connection to use
* @param queue the Queue to test
* @return true if the queue exists
*/
private boolean queueExists(Connection connection, Queue queue) throws Exception {
Connection target = connection;
if (target == null) {
ConnectionFactory connectionFactory = new ConnectionFactory();
connectionFactory.setPort(BrokerTestUtils.getPort());
target = connectionFactory.newConnection();
@Test
public void testSpringWithDefaultExchange() throws Exception {
Exchange exchange = new DirectExchange(RabbitAdmin.DEFAULT_EXCHANGE_NAME);
context.getBeanFactory().registerSingleton("foo", exchange);
rabbitAdmin.afterPropertiesSet();
rabbitAdmin.initialize();
// Pass by virtue of RabbitMQ not firing a 403 reply code
}
@Test
public void testDeleteExchangeWithDefaultExchange() throws Exception {
boolean result = rabbitAdmin.deleteExchange(RabbitAdmin.DEFAULT_EXCHANGE_NAME);
assertTrue(result);
}
@Test
public void testDeclareBindingWithDefaultExchangeImplicitBinding() throws Exception {
Exchange exchange = new DirectExchange(RabbitAdmin.DEFAULT_EXCHANGE_NAME);
String queueName = "test.queue";
final Queue queue = new Queue(queueName, false, false, false);
rabbitAdmin.declareQueue(queue);
Binding binding = new Binding(queueName, DestinationType.QUEUE, exchange.getName(), queueName, null);
rabbitAdmin.declareBinding(binding);
// Pass by virtue of RabbitMQ not firing a 403 reply code for both exchange and binding declaration
assertTrue(queueExists(queue));
}
@Test
public void testSpringWithDefaultExchangeImplicitBinding() throws Exception {
Exchange exchange = new DirectExchange(RabbitAdmin.DEFAULT_EXCHANGE_NAME);
context.getBeanFactory().registerSingleton("foo", exchange);
String queueName = "test.queue";
final Queue queue = new Queue(queueName, false, false, false);
context.getBeanFactory().registerSingleton("bar", queue);
Binding binding = new Binding(queueName, DestinationType.QUEUE, exchange.getName(), queueName, null);
context.getBeanFactory().registerSingleton("baz", binding);
rabbitAdmin.afterPropertiesSet();
rabbitAdmin.initialize();
// Pass by virtue of RabbitMQ not firing a 403 reply code for both exchange and binding declaration
assertTrue(queueExists(queue));
}
@Test
public void testRemoveBindingWithDefaultExchangeImplicitBinding() throws Exception {
String queueName = "test.queue";
final Queue queue = new Queue(queueName, false, false, false);
rabbitAdmin.declareQueue(queue);
Binding binding = new Binding(queueName, DestinationType.QUEUE, RabbitAdmin.DEFAULT_EXCHANGE_NAME, queueName, null);
rabbitAdmin.removeBinding(binding);
// Pass by virtue of RabbitMQ not firing a 403 reply code
}
@Test
public void testDeclareBindingWithDefaultExchangeNonImplicitBinding() throws Exception {
Exchange exchange = new DirectExchange(RabbitAdmin.DEFAULT_EXCHANGE_NAME);
String queueName = "test.queue";
final Queue queue = new Queue(queueName, false, false, false);
rabbitAdmin.declareQueue(queue);
Binding binding = new Binding(queueName, DestinationType.QUEUE, exchange.getName(), "test.routingKey", null);
try {
rabbitAdmin.declareBinding(binding);
} catch (AmqpIOException ex) {
Throwable cause = ex;
Throwable rootCause = null;
while (cause != null) {
rootCause = cause;
cause = cause.getCause();
}
assertTrue(rootCause.getMessage().contains("reply-code=403"));
assertTrue(rootCause.getMessage().contains("operation not permitted on the default exchange"));
}
Channel channel = target.createChannel();
}
@Test
public void testSpringWithDefaultExchangeNonImplicitBinding() throws Exception {
Exchange exchange = new DirectExchange(RabbitAdmin.DEFAULT_EXCHANGE_NAME);
context.getBeanFactory().registerSingleton("foo", exchange);
String queueName = "test.queue";
final Queue queue = new Queue(queueName, false, false, false);
context.getBeanFactory().registerSingleton("bar", queue);
Binding binding = new Binding(queueName, DestinationType.QUEUE, exchange.getName(), "test.routingKey", null);
context.getBeanFactory().registerSingleton("baz", binding);
rabbitAdmin.afterPropertiesSet();
try {
rabbitAdmin.declareBinding(binding);
} catch (AmqpIOException ex) {
Throwable cause = ex;
Throwable rootCause = null;
while (cause != null) {
rootCause = cause;
cause = cause.getCause();
}
assertTrue(rootCause.getMessage().contains("reply-code=403"));
assertTrue(rootCause.getMessage().contains("operation not permitted on the default exchange"));
}
}
/**
* Verify that a queue exists using the native Rabbit API to bypass all the connection and
* channel caching and callbacks in Spring AMQP.
*
* @param Queue The queue to verify
* @return True if the queue exists
*/
private boolean queueExists(final Queue queue) throws Exception {
ConnectionFactory connectionFactory = new ConnectionFactory();
connectionFactory.setPort(BrokerTestUtils.getPort());
Connection connection = connectionFactory.newConnection();
Channel channel = connection.createChannel();
try {
DeclareOk result = channel.queueDeclarePassive(queue.getName());
return result != null;
@@ -218,10 +290,7 @@ public class RabbitAdminIntegrationTests {
}
return false;
} finally {
if (connection==null) {
target.close();
}
connection.close();
}
}
}
}