From 0cab85b9279393fe7e7c4616551f8ee731ec2372 Mon Sep 17 00:00:00 2001 From: Dave Syer Date: Mon, 18 Apr 2011 14:46:17 +0100 Subject: [PATCH] AMQP-160, AMQP-161: Custom exchanges and exchange bindings --- .../springframework/amqp/core/Binding.java | 25 +++++++++++-------- .../amqp/core/BindingBuilder.java | 9 ++++--- .../amqp/core/BindingBuilderTests.java | 21 +++++----------- .../amqp/rabbit/core/RabbitAdmin.java | 23 ++++++++++++----- .../log4j/AmqpAppenderConfiguration.java | 3 ++- 5 files changed, 44 insertions(+), 37 deletions(-) diff --git a/spring-amqp-core/src/main/java/org/springframework/amqp/core/Binding.java b/spring-amqp-core/src/main/java/org/springframework/amqp/core/Binding.java index 14d16c8c..3e369a4e 100644 --- a/spring-amqp-core/src/main/java/org/springframework/amqp/core/Binding.java +++ b/spring-amqp-core/src/main/java/org/springframework/amqp/core/Binding.java @@ -16,9 +16,9 @@ package org.springframework.amqp.core; import java.util.Map; /** - * Simple container collecting information to describe a binding. Takes String destination and Exchange source instances - * as arguments to facilitate wiring using code based configuration. Can be used in conjunction with {@link AmqpAdmin}, - * or created via a {@link BindingBuilder}. + * Simple container collecting information to describe a binding. Takes String destination and exchange names as + * arguments to facilitate wiring using code based configuration. Can be used in conjunction with {@link AmqpAdmin}, or + * created via a {@link BindingBuilder}. * * @author Mark Pollack * @author Mark Fisher @@ -28,7 +28,9 @@ import java.util.Map; */ public class Binding { - public static final String QUEUE_TYPE = "queue"; + public static enum DestinationType { + QUEUE, EXCHANGE; + } private final String destination; @@ -38,13 +40,10 @@ public class Binding { private final Map arguments; - private final String destinationType; + private final DestinationType destinationType; -// public Binding(String destination, String exchange, String routingKey, Map arguments) { -// this(destination, QUEUE_TYPE, exchange, routingKey, arguments); -// } -// - public Binding(String destination, String destinationType, String exchange, String routingKey, Map arguments) { + public Binding(String destination, DestinationType destinationType, String exchange, String routingKey, + Map arguments) { this.destination = destination; this.destinationType = destinationType; this.exchange = exchange; @@ -56,7 +55,7 @@ public class Binding { return this.destination; } - public String getDestinationType() { + public DestinationType getDestinationType() { return this.destinationType; } @@ -72,4 +71,8 @@ public class Binding { return this.arguments; } + public boolean isDestinationQueue() { + return DestinationType.QUEUE.equals(destinationType); + } + } diff --git a/spring-amqp-core/src/main/java/org/springframework/amqp/core/BindingBuilder.java b/spring-amqp-core/src/main/java/org/springframework/amqp/core/BindingBuilder.java index e24c8041..2d3c081c 100644 --- a/spring-amqp-core/src/main/java/org/springframework/amqp/core/BindingBuilder.java +++ b/spring-amqp-core/src/main/java/org/springframework/amqp/core/BindingBuilder.java @@ -17,6 +17,7 @@ import java.util.Collections; import java.util.HashMap; import java.util.Map; +import org.springframework.amqp.core.Binding.DestinationType; import org.springframework.util.Assert; /** @@ -29,19 +30,19 @@ import org.springframework.util.Assert; public final class BindingBuilder { public static DestinationConfigurer bind(Queue queue) { - return new DestinationConfigurer(queue.getName(), "queue"); + return new DestinationConfigurer(queue.getName(), DestinationType.QUEUE); } public static DestinationConfigurer bind(Exchange exchange) { - return new DestinationConfigurer(exchange.getName(), "exchange"); + return new DestinationConfigurer(exchange.getName(), DestinationType.EXCHANGE); } public static class DestinationConfigurer { protected final String name; - protected final String type; + protected final DestinationType type; - private DestinationConfigurer(String name, String type) { + private DestinationConfigurer(String name, DestinationType type) { this.name = name; this.type = type; } diff --git a/spring-amqp-core/src/test/java/org/springframework/amqp/core/BindingBuilderTests.java b/spring-amqp-core/src/test/java/org/springframework/amqp/core/BindingBuilderTests.java index 9782612c..04daab7f 100644 --- a/spring-amqp-core/src/test/java/org/springframework/amqp/core/BindingBuilderTests.java +++ b/spring-amqp-core/src/test/java/org/springframework/amqp/core/BindingBuilderTests.java @@ -60,26 +60,17 @@ public class BindingBuilderTests { return "x-custom"; } } - ; - Binding binding = BindingBuilder.bind(new Queue("q")).to(new CustomExchange("f")).with("r") - .and(Collections. singletonMap("k", new Object())); + Binding binding = BindingBuilder.// + bind(new Queue("q")).// + to(new CustomExchange("f")).// + with("r").// + and(Collections. singletonMap("k", new Object())); assertNotNull(binding); } @Test public void exchangeBinding() { - class CustomExchange extends AbstractExchange { - public CustomExchange(String name) { - super(name); - } - - @Override - public String getType() { - return "x-custom"; - } - } - ; - Binding binding = BindingBuilder.bind(new CustomExchange("q")).to(new FanoutExchange("f")); + Binding binding = BindingBuilder.bind(new DirectExchange("q")).to(new FanoutExchange("f")); assertNotNull(binding); } 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 4e243694..ffdc0314 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 @@ -177,8 +177,13 @@ public class RabbitAdmin implements AmqpAdmin, ApplicationContextAware, Initiali public void removeBinding(final Binding binding) { rabbitTemplate.execute(new ChannelCallback() { public Object doInRabbit(Channel channel) throws Exception { - channel.queueUnbind(binding.getDestination(), binding.getExchange(), binding.getRoutingKey(), - binding.getArguments()); + if (binding.isDestinationQueue()) { + channel.queueUnbind(binding.getDestination(), binding.getExchange(), binding.getRoutingKey(), + binding.getArguments()); + } else { + channel.exchangeUnbind(binding.getDestination(), binding.getExchange(), binding.getRoutingKey(), + binding.getArguments()); + } return null; } }); @@ -331,11 +336,17 @@ public class RabbitAdmin implements AmqpAdmin, ApplicationContextAware, Initiali private void declareBindings(final Channel channel, final Binding... bindings) throws IOException { for (Binding binding : bindings) { if (logger.isDebugEnabled()) { - logger.debug("Binding queue [" + binding.getDestination() + "] to exchange [" + binding.getExchange() - + "] with routing key [" + binding.getRoutingKey() + "]"); + logger.debug("Binding destination [" + binding.getDestination() + " (" + binding.getDestinationType() + + ")] to exchange [" + binding.getExchange() + "] with routing key [" + binding.getRoutingKey() + + "]"); + } + if (binding.isDestinationQueue()) { + channel.queueBind(binding.getDestination(), binding.getExchange(), binding.getRoutingKey(), + binding.getArguments()); + } else { + channel.exchangeBind(binding.getDestination(), binding.getExchange(), binding.getRoutingKey(), + binding.getArguments()); } - channel.queueBind(binding.getDestination(), binding.getExchange(), binding.getRoutingKey(), - binding.getArguments()); } } diff --git a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/log4j/AmqpAppenderConfiguration.java b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/log4j/AmqpAppenderConfiguration.java index 35ddbcdf..1addbc6a 100644 --- a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/log4j/AmqpAppenderConfiguration.java +++ b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/log4j/AmqpAppenderConfiguration.java @@ -18,6 +18,7 @@ package org.springframework.amqp.rabbit.log4j; import org.springframework.amqp.core.AcknowledgeMode; import org.springframework.amqp.core.Binding; +import org.springframework.amqp.core.BindingBuilder; import org.springframework.amqp.core.Queue; import org.springframework.amqp.core.TopicExchange; import org.springframework.amqp.rabbit.connection.SingleConnectionFactory; @@ -59,7 +60,7 @@ public class AmqpAppenderConfiguration { @Bean public Binding testBinding() { - return new Binding(testQueue(), testExchange(), ROUTING_KEY); + return BindingBuilder.bind(testQueue()).to(testExchange()).with(ROUTING_KEY); } @Bean