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

Modify Binding to support more generic contracts

Browse Source
This commit is contained in:
Dave Syer
2011-04-18 14:31:43 +01:00
parent 390b8d8847
commit a708c6ce36
5 changed files with 185 additions and 112 deletions
Download Patch File Download Diff File Expand all files Collapse all files

80
spring-amqp-core/src/main/java/org/springframework/amqp/core/Binding.java
View File

@@ -1,17 +1,14 @@
/*
* Copyright 2002-2010 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.
*
* 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.core;
@@ -19,57 +16,48 @@ package org.springframework.amqp.core;
import java.util.Map;
/**
* Simple container collecting information to describe a queue binding. Takes Queue and Exchange
* instances as arguments to facilitate wiring using @Bean code based configuration.
* Used in conjunction with AmqpAdmin.
* 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}.
*
* @author Mark Pollack
* @author Mark Fisher
* @author Dave Syer
*
* @see AmqpAdmin
*/
public class Binding {
private String queue;
public static final String QUEUE_TYPE = "queue";
private String exchange;
private final String destination;
private String routingKey;
private final String exchange;
private Map<String, Object> arguments;
private final String routingKey;
public Binding(Queue queue, FanoutExchange exchange) {
this.queue = queue.getName();
this.exchange = exchange.getName();
this.routingKey = "";
}
private final Map<String, Object> arguments;
public Binding(Queue queue, HeadersExchange exchange, Map<String, Object> arguments) {
this.queue = queue.getName();
this.exchange = exchange.getName();
this.routingKey = "";
private final String destinationType;
// public Binding(String destination, String exchange, String routingKey, Map<String, Object> arguments) {
// this(destination, QUEUE_TYPE, exchange, routingKey, arguments);
// }
//
public Binding(String destination, String destinationType, String exchange, String routingKey, Map<String, Object> arguments) {
this.destination = destination;
this.destinationType = destinationType;
this.exchange = exchange;
this.routingKey = routingKey;
this.arguments = arguments;
}
public Binding(Queue queue, DirectExchange exchange) {
this.queue = queue.getName();
this.exchange = exchange.getName();
this.routingKey = queue.getName();
public String getDestination() {
return this.destination;
}
public Binding(Queue queue, DirectExchange exchange, String routingKey) {
this.queue = queue.getName();
this.exchange = exchange.getName();
this.routingKey = routingKey;
}
public Binding(Queue queue, TopicExchange exchange, String routingPattern) {
this.queue = queue.getName();
this.exchange = exchange.getName();
this.routingKey = routingPattern;
}
public String getQueue() {
return this.queue;
public String getDestinationType() {
return this.destinationType;
}
public String getExchange() {

145
spring-amqp-core/src/main/java/org/springframework/amqp/core/BindingBuilder.java
View File

@@ -1,21 +1,19 @@
/*
* Copyright 2002-2010 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.
*
* 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.core;
import java.util.Collections;
import java.util.HashMap;
import java.util.Map;
@@ -26,47 +24,57 @@ import org.springframework.util.Assert;
*
* @author Mark Pollack
* @author Mark Fisher
* @author Dave Syer
*/
public final class BindingBuilder {
public final class BindingBuilder {
public static ExchangeConfigurer bind(Queue queue) {
return new ExchangeConfigurer(queue);
public static DestinationConfigurer bind(Queue queue) {
return new DestinationConfigurer(queue.getName(), "queue");
}
public static class ExchangeConfigurer {
public static DestinationConfigurer bind(Exchange exchange) {
return new DestinationConfigurer(exchange.getName(), "exchange");
}
private final Queue queue;
public static class DestinationConfigurer {
private ExchangeConfigurer(Queue queue) {
this.queue = queue;
protected final String name;
protected final String type;
private DestinationConfigurer(String name, String type) {
this.name = name;
this.type = type;
}
public Binding to(FanoutExchange exchange) {
return new Binding(this.queue, exchange);
return new Binding(this.name, this.type, exchange.getName(), "", new HashMap<String, Object>());
}
public HeadersExchangeMapConfigurer to(HeadersExchange exchange) {
return new HeadersExchangeMapConfigurer(this.queue, exchange);
return new HeadersExchangeMapConfigurer(this, exchange);
}
public DirectExchangeRoutingKeyConfigurer to(DirectExchange exchange) {
return new DirectExchangeRoutingKeyConfigurer(this.queue, exchange);
return new DirectExchangeRoutingKeyConfigurer(this, exchange);
}
public TopicExchangeRoutingKeyConfigurer to(TopicExchange exchange) {
return new TopicExchangeRoutingKeyConfigurer(this.queue, exchange);
return new TopicExchangeRoutingKeyConfigurer(this, exchange);
}
public GenericExchangeRoutingKeyConfigurer to(Exchange exchange) {
return new GenericExchangeRoutingKeyConfigurer(this, exchange);
}
}
public static class HeadersExchangeMapConfigurer {
protected final Queue queue;
protected final DestinationConfigurer destination;
protected final HeadersExchange exchange;
private HeadersExchangeMapConfigurer(Queue queue, HeadersExchange exchange) {
this.queue = queue;
private HeadersExchangeMapConfigurer(DestinationConfigurer destination, HeadersExchange exchange) {
this.destination = destination;
this.exchange = exchange;
}
@@ -90,7 +98,6 @@ public final class BindingBuilder {
return new HeadersExchangeMapBindingCreator(headerValues, true);
}
public class HeadersExchangeSingleValueBindingCreator {
private final String key;
@@ -101,17 +108,17 @@ public final class BindingBuilder {
}
public Binding exists() {
return new Binding(queue, exchange, createMapForKeys(this.key));
return new Binding(destination.name, destination.type, exchange.getName(), "",
createMapForKeys(this.key));
}
public Binding matches(Object value) {
Map<String, Object> map = new HashMap<String, Object>();
map.put(key, value);
return new Binding(queue, exchange, map);
return new Binding(destination.name, destination.type, exchange.getName(), "", map);
}
}
public class HeadersExchangeKeysBindingCreator {
private final Map<String, Object> headerMap;
@@ -123,11 +130,10 @@ public final class BindingBuilder {
}
public Binding exist() {
return new Binding(queue, exchange, this.headerMap);
return new Binding(destination.name, destination.type, exchange.getName(), "", this.headerMap);
}
}
public class HeadersExchangeMapBindingCreator {
private final Map<String, Object> headerMap;
@@ -139,61 +145,100 @@ public final class BindingBuilder {
}
public Binding match() {
return new Binding(queue, exchange, this.headerMap);
return new Binding(destination.name, destination.type, exchange.getName(), "", this.headerMap);
}
}
}
private static abstract class AbstractRoutingKeyConfigurer<E extends Exchange> {
protected final Queue queue;
protected final DestinationConfigurer destination;
protected final E exchange;
protected final String exchange;
private AbstractRoutingKeyConfigurer(Queue queue, E exchange) {
this.queue = queue;
private AbstractRoutingKeyConfigurer(DestinationConfigurer destination, String exchange) {
this.destination = destination;
this.exchange = exchange;
}
}
public static class TopicExchangeRoutingKeyConfigurer extends AbstractRoutingKeyConfigurer<TopicExchange> {
private TopicExchangeRoutingKeyConfigurer(Queue queue, TopicExchange exchange) {
super(queue, exchange);
private TopicExchangeRoutingKeyConfigurer(DestinationConfigurer destination, TopicExchange exchange) {
super(destination, exchange.getName());
}
public Binding with(String routingKey) {
return new Binding(this.queue, this.exchange, routingKey);
return new Binding(destination.name, destination.type, exchange, routingKey,
Collections.<String, Object> emptyMap());
}
public Binding with(Enum<?> routingKeyEnum) {
return new Binding(this.queue, this.exchange, routingKeyEnum.toString());
return new Binding(destination.name, destination.type, exchange, routingKeyEnum.toString(),
Collections.<String, Object> emptyMap());
}
}
public static class GenericExchangeRoutingKeyConfigurer extends AbstractRoutingKeyConfigurer<TopicExchange> {
private GenericExchangeRoutingKeyConfigurer(DestinationConfigurer destination, Exchange exchange) {
super(destination, exchange.getName());
}
public GenericArgumentsConfigurer with(String routingKey) {
return new GenericArgumentsConfigurer(this, routingKey);
}
public GenericArgumentsConfigurer with(Enum<?> routingKeyEnum) {
return new GenericArgumentsConfigurer(this, routingKeyEnum.toString());
}
}
public static class GenericArgumentsConfigurer {
private final GenericExchangeRoutingKeyConfigurer configurer;
private final String routingKey;
public GenericArgumentsConfigurer(GenericExchangeRoutingKeyConfigurer configurer, String routingKey) {
this.configurer = configurer;
this.routingKey = routingKey;
}
public Binding and(Map<String, Object> map) {
return new Binding(configurer.destination.name, configurer.destination.type, configurer.exchange,
routingKey, map);
}
public Binding noargs() {
return new Binding(configurer.destination.name, configurer.destination.type, configurer.exchange,
routingKey, Collections.<String, Object> emptyMap());
}
}
public static class DirectExchangeRoutingKeyConfigurer extends AbstractRoutingKeyConfigurer<DirectExchange> {
private DirectExchangeRoutingKeyConfigurer(Queue queue, DirectExchange exchange) {
super(queue, exchange);
private DirectExchangeRoutingKeyConfigurer(DestinationConfigurer destination, DirectExchange exchange) {
super(destination, exchange.getName());
}
public Binding with(String routingKey) {
return new Binding(this.queue, this.exchange, routingKey);
return new Binding(destination.name, destination.type, exchange, routingKey,
Collections.<String, Object> emptyMap());
}
public Binding with(Enum<?> routingKeyEnum) {
return new Binding(this.queue, this.exchange, routingKeyEnum.toString());
return new Binding(destination.name, destination.type, exchange, routingKeyEnum.toString(),
Collections.<String, Object> emptyMap());
}
public Binding withQueueName() {
return new Binding(this.queue, this.exchange, this.queue.getName());
return new Binding(destination.name, destination.type, exchange, destination.name,
Collections.<String, Object> emptyMap());
}
}
private static Map<String, Object> createMapForKeys(String... keys) {
Map<String, Object> map = new HashMap<String, Object>();
for (String key : keys) {

6
spring-amqp-core/src/main/java/org/springframework/amqp/core/Exchange.java
View File

@@ -48,6 +48,12 @@ public interface Exchange {
*/
boolean isAutoDelete();
/**
* A map of arguments used to declare the exchange. These are stored by the broker, but do not necessarily have any
* meaning to the broker (depending on the exchange type).
*
* @return the arguments
*/
Map<String, Object> getArguments();
}

60
spring-amqp-core/src/test/java/org/springframework/amqp/core/BindingBuilderTests.java
View File

@@ -1,23 +1,22 @@
/*
* Copyright 2002-2010 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.
*
* 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.core;
import static org.junit.Assert.assertNotNull;
import java.util.Collections;
import org.junit.Test;
/**
@@ -32,7 +31,7 @@ public class BindingBuilderTests {
}
@Test
public void directBinding() {
public void directBinding() {
Binding binding = BindingBuilder.bind(new Queue("q")).to(new DirectExchange("d")).with("r");
assertNotNull(binding);
}
@@ -49,4 +48,39 @@ public class BindingBuilderTests {
assertNotNull(binding);
}
@Test
public void customBinding() {
class CustomExchange extends AbstractExchange {
public CustomExchange(String name) {
super(name);
}
@Override
public String getType() {
return "x-custom";
}
}
;
Binding binding = BindingBuilder.bind(new Queue("q")).to(new CustomExchange("f")).with("r")
.and(Collections.<String, Object> 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"));
assertNotNull(binding);
}
}

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

@@ -177,7 +177,7 @@ public class RabbitAdmin implements AmqpAdmin, ApplicationContextAware, Initiali
public void removeBinding(final Binding binding) {
rabbitTemplate.execute(new ChannelCallback<Object>() {
public Object doInRabbit(Channel channel) throws Exception {
channel.queueUnbind(binding.getQueue(), binding.getExchange(), binding.getRoutingKey(),
channel.queueUnbind(binding.getDestination(), binding.getExchange(), binding.getRoutingKey(),
binding.getArguments());
return null;
}
@@ -331,10 +331,10 @@ 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.getQueue() + "] to exchange [" + binding.getExchange()
logger.debug("Binding queue [" + binding.getDestination() + "] to exchange [" + binding.getExchange()
+ "] with routing key [" + binding.getRoutingKey() + "]");
}
channel.queueBind(binding.getQueue(), binding.getExchange(), binding.getRoutingKey(),
channel.queueBind(binding.getDestination(), binding.getExchange(), binding.getRoutingKey(),
binding.getArguments());
}
}