diff --git a/spring-amqp-core/src/main/java/org/springframework/amqp/core/AbstractExchange.java b/spring-amqp-core/src/main/java/org/springframework/amqp/core/AbstractExchange.java index a53d8f73..96e24599 100644 --- a/spring-amqp-core/src/main/java/org/springframework/amqp/core/AbstractExchange.java +++ b/spring-amqp-core/src/main/java/org/springframework/amqp/core/AbstractExchange.java @@ -16,6 +16,7 @@ package org.springframework.amqp.core; +import java.util.HashMap; import java.util.Map; @@ -67,7 +68,12 @@ public abstract class AbstractExchange implements Exchange { this.name = name; this.durable = durable; this.autoDelete = autoDelete; - this.arguments = arguments; + if (arguments != null) { + this.arguments = arguments; + } + else { + this.arguments = new HashMap(); + } } public abstract String getType(); @@ -84,6 +90,9 @@ public abstract class AbstractExchange implements Exchange { return autoDelete; } + protected synchronized void addArgument(String argName, Object argValue) { + this.arguments.put(argName, argValue); + } /** * Return the collection of arbitrary arguments to use when declaring an exchange. * @return the collection of arbitrary arguments to use when declaring an exchange. diff --git a/spring-amqp-core/src/main/java/org/springframework/amqp/core/ExchangeTypes.java b/spring-amqp-core/src/main/java/org/springframework/amqp/core/ExchangeTypes.java index 2848dba1..35e4b854 100644 --- a/spring-amqp-core/src/main/java/org/springframework/amqp/core/ExchangeTypes.java +++ b/spring-amqp-core/src/main/java/org/springframework/amqp/core/ExchangeTypes.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2010 the original author or authors. + * Copyright 2002-2012 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. @@ -20,6 +20,7 @@ package org.springframework.amqp.core; * Constants for the standard Exchange type names. * * @author Mark Fisher + * @author Gary Russell */ public abstract class ExchangeTypes { @@ -33,4 +34,6 @@ public abstract class ExchangeTypes { public static final String SYSTEM = "system"; + public static final String FEDERATED = "x-federation"; + } diff --git a/spring-amqp-core/src/main/java/org/springframework/amqp/core/FederatedExchange.java b/spring-amqp-core/src/main/java/org/springframework/amqp/core/FederatedExchange.java new file mode 100644 index 00000000..cf014edc --- /dev/null +++ b/spring-amqp-core/src/main/java/org/springframework/amqp/core/FederatedExchange.java @@ -0,0 +1,60 @@ +/* + * Copyright 2002-2012 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.core; + +import java.util.Map; + +/** + * + * @see AmqpAdmin + * + * @author Gary Russell + */ +public class FederatedExchange extends AbstractExchange { + + public static final FederatedExchange DEFAULT = new FederatedExchange(""); + + private static final String BACKING_TYPE_ARG = "type"; + + private static final String UPSTREAM_SET_ARG = "upstream-set"; + + public FederatedExchange(String name) { + super(name); + } + + public FederatedExchange(String name, boolean durable, boolean autoDelete) { + super(name, durable, autoDelete); + } + + public FederatedExchange(String name, boolean durable, boolean autoDelete, Map arguments) { + super(name, durable, autoDelete, arguments); + } + + public void setBackingType(String backingType) { + this.addArgument(BACKING_TYPE_ARG, backingType); + } + + public void setUpstreamSet(String upstreamSet) { + this.addArgument(UPSTREAM_SET_ARG, upstreamSet); + } + + @Override + public final String getType() { + return ExchangeTypes.FEDERATED; + } + +} diff --git a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/config/AbstractExchangeParser.java b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/config/AbstractExchangeParser.java index 5cb0ac02..13b82e96 100644 --- a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/config/AbstractExchangeParser.java +++ b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/config/AbstractExchangeParser.java @@ -51,15 +51,7 @@ public abstract class AbstractExchangeParser extends AbstractSingleBeanDefinitio protected void doParse(Element element, ParserContext parserContext, BeanDefinitionBuilder builder) { String exchangeName = element.getAttribute(NAME_ATTRIBUTE); builder.addConstructorArgValue(new TypedStringValue(exchangeName)); - Element bindings = DomUtils.getChildElementByTagName(element, BINDINGS_ELE); - if (bindings != null) { - for (Element binding : DomUtils.getChildElementsByTagName(bindings, BINDING_ELE)) { - AbstractBeanDefinition beanDefinition = parseBinding(exchangeName, binding, - parserContext); - registerBeanDefinition(new BeanDefinitionHolder(beanDefinition, parserContext.getReaderContext() - .generateBeanName(beanDefinition)), parserContext.getRegistry()); - } - } + parseBindings(element, parserContext, builder, exchangeName); NamespaceUtils.addConstructorArgBooleanValueIfAttributeDefined(builder, element, DURABLE_ATTRIBUTE, true); NamespaceUtils.addConstructorArgBooleanValueIfAttributeDefined(builder, element, AUTO_DELETE_ATTRIBUTE, @@ -74,6 +66,24 @@ public abstract class AbstractExchangeParser extends AbstractSingleBeanDefinitio } + protected void parseBindings(Element element, ParserContext parserContext, BeanDefinitionBuilder builder, + String exchangeName) { + Element bindings = DomUtils.getChildElementByTagName(element, BINDINGS_ELE); + doParseBindings(parserContext, exchangeName, bindings, this); + } + + protected void doParseBindings(ParserContext parserContext, + String exchangeName, Element bindings, AbstractExchangeParser parser) { + if (bindings != null) { + for (Element binding : DomUtils.getChildElementsByTagName(bindings, BINDING_ELE)) { + AbstractBeanDefinition beanDefinition = parser.parseBinding(exchangeName, binding, + parserContext); + registerBeanDefinition(new BeanDefinitionHolder(beanDefinition, parserContext.getReaderContext() + .generateBeanName(beanDefinition)), parserContext.getRegistry()); + } + } + } + protected abstract AbstractBeanDefinition parseBinding(String exchangeName, Element binding, ParserContext parserContext); diff --git a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/config/FederatedExchangeParser.java b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/config/FederatedExchangeParser.java new file mode 100644 index 00000000..728ef77a --- /dev/null +++ b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/config/FederatedExchangeParser.java @@ -0,0 +1,115 @@ +/* + * Copyright 2002-2012 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 org.springframework.amqp.core.ExchangeTypes; +import org.springframework.amqp.core.FederatedExchange; +import org.springframework.beans.factory.support.AbstractBeanDefinition; +import org.springframework.beans.factory.support.BeanDefinitionBuilder; +import org.springframework.beans.factory.xml.ParserContext; +import org.springframework.util.StringUtils; +import org.springframework.util.xml.DomUtils; +import org.w3c.dom.Element; + +/** + * @author Gary Russell + * + */ +public class FederatedExchangeParser extends AbstractExchangeParser { + + private final static String BACKING_TYPE_ATTRIBUTE = "backing-type"; + + private final static String UPSTREAM_SET_ATTRIBUTE = "upstream-set"; + + private static final String DIRECT_BINDINGS_ELE = "direct-bindings"; + + private static final String TOPIC_BINDINGS_ELE = "topic-bindings"; + + private static final String TOPIC_FANOUT_ELE = "fanout-bindings"; + + private static final String TOPIC_HEADERS_ELE = "headers-bindings"; + + @Override + protected Class getBeanClass(Element element) { + return FederatedExchange.class; + } + + + @Override + protected void doParse(Element element, ParserContext parserContext, + BeanDefinitionBuilder builder) { + super.doParse(element, parserContext, builder); + NamespaceUtils.setValueIfAttributeDefined(builder, element, BACKING_TYPE_ATTRIBUTE); + NamespaceUtils.setValueIfAttributeDefined(builder, element, UPSTREAM_SET_ATTRIBUTE); + } + + + @Override + protected void parseBindings(Element element, ParserContext parserContext, + BeanDefinitionBuilder builder, String exchangeName) { + String backingType = element.getAttribute(BACKING_TYPE_ATTRIBUTE); + Element bindings; + bindings = DomUtils.getChildElementByTagName(element, DIRECT_BINDINGS_ELE); + if (bindings != null && !ExchangeTypes.DIRECT.equals(backingType)) { + parserContext.getReaderContext().error( + "Cannot have direct-bindings if backing-type not 'direct'", + element); + } + if (bindings == null) { + bindings = DomUtils.getChildElementByTagName(element, TOPIC_BINDINGS_ELE); + if (bindings != null && !ExchangeTypes.TOPIC.equals(backingType)) { + parserContext.getReaderContext().error( + "Cannot have topic-bindings if backing-type not 'topic'", + element); + } + } + if (bindings == null) { + bindings = DomUtils.getChildElementByTagName(element, TOPIC_FANOUT_ELE); + if (bindings != null && !ExchangeTypes.FANOUT.equals(backingType)) { + parserContext.getReaderContext().error( + "Cannot have fanout-bindings if backing-type not 'fanout'", + element); + } + } + if (bindings == null) { + bindings = DomUtils.getChildElementByTagName(element, TOPIC_HEADERS_ELE); + if (bindings != null && !ExchangeTypes.HEADERS.equals(backingType)) { + parserContext.getReaderContext().error( + "Cannot have headers-bindings if backing-type not 'headers'", + element); + } + } + if (StringUtils.hasText(backingType)) { + if (ExchangeTypes.DIRECT.equals(backingType)) { + doParseBindings(parserContext, exchangeName, bindings, new DirectExchangeParser()); + } + else if (ExchangeTypes.TOPIC.equals(backingType)) { + doParseBindings(parserContext, exchangeName, bindings, new TopicExchangeParser()); + } + else if (ExchangeTypes.FANOUT.equals(backingType)) { + doParseBindings(parserContext, exchangeName, bindings, new FanoutExchangeParser()); + } + else if (ExchangeTypes.HEADERS.equals(backingType)) { + doParseBindings(parserContext, exchangeName, bindings, new HeadersExchangeParser()); + } + } + } + + @Override + protected AbstractBeanDefinition parseBinding(String exchangeName, Element binding, + ParserContext parserContext) { + throw new UnsupportedOperationException("Not supported for federated exchange"); + } + +} diff --git a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/config/RabbitNamespaceHandler.java b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/config/RabbitNamespaceHandler.java index 1a660aa0..9f1d54f8 100644 --- a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/config/RabbitNamespaceHandler.java +++ b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/config/RabbitNamespaceHandler.java @@ -34,6 +34,7 @@ public class RabbitNamespaceHandler extends NamespaceHandlerSupport { registerBeanDefinitionParser("topic-exchange", new TopicExchangeParser()); registerBeanDefinitionParser("fanout-exchange", new FanoutExchangeParser()); registerBeanDefinitionParser("headers-exchange", new HeadersExchangeParser()); + registerBeanDefinitionParser("federated-exchange", new FederatedExchangeParser()); registerBeanDefinitionParser("listener-container", new ListenerContainerParser()); registerBeanDefinitionParser("admin", new AdminParser()); registerBeanDefinitionParser("connection-factory", new ConnectionFactoryParser()); diff --git a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/CachingConnectionFactory.java b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/CachingConnectionFactory.java index 062c9493..1b1e465c 100644 --- a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/CachingConnectionFactory.java +++ b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/CachingConnectionFactory.java @@ -408,11 +408,11 @@ public class CachingConnectionFactory extends AbstractConnectionFactory { } public void destroy() { + reset(); if (this.target != null) { getConnectionListener().onClose(target); RabbitUtils.closeConnection(this.target); } - reset(); this.target = null; } diff --git a/spring-rabbit/src/main/resources/org/springframework/amqp/rabbit/config/spring-rabbit-1.0.xsd b/spring-rabbit/src/main/resources/org/springframework/amqp/rabbit/config/spring-rabbit-1.0.xsd index 0b083f55..7451bc89 100644 --- a/spring-rabbit/src/main/resources/org/springframework/amqp/rabbit/config/spring-rabbit-1.0.xsd +++ b/spring-rabbit/src/main/resources/org/springframework/amqp/rabbit/config/spring-rabbit-1.0.xsd @@ -237,6 +237,122 @@ + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + diff --git a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/config/ExchangeParserIntegrationTests.java b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/config/ExchangeParserIntegrationTests.java index a4d2a9af..ea5edde7 100644 --- a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/config/ExchangeParserIntegrationTests.java +++ b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/config/ExchangeParserIntegrationTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2008 the original author or authors. + * Copyright 2002-2012 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 diff --git a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/config/ExchangeParserTests.java b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/config/ExchangeParserTests.java index bc7aded9..52b22c73 100644 --- a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/config/ExchangeParserTests.java +++ b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/config/ExchangeParserTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2011 the original author or authors. + * Copyright 2002-2012 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. @@ -25,11 +25,18 @@ import org.junit.Before; import org.junit.Test; import org.springframework.amqp.core.DirectExchange; import org.springframework.amqp.core.FanoutExchange; +import org.springframework.amqp.core.FederatedExchange; import org.springframework.amqp.core.HeadersExchange; import org.springframework.amqp.core.TopicExchange; import org.springframework.beans.factory.xml.XmlBeanFactory; import org.springframework.core.io.ClassPathResource; - +/** + * @author Dave Syer + * @author Mark Fisher + * @author Gary Russell + * @since 1.0 + * + */ public final class ExchangeParserTests { private XmlBeanFactory beanFactory; @@ -101,4 +108,48 @@ public final class ExchangeParserTests { assertEquals("bar", exchange.getArguments().get("foo")); } + @Test + public void testFederatedDirectExchange() throws Exception { + FederatedExchange exchange = beanFactory.getBean("fedDirect", FederatedExchange.class); + assertNotNull(exchange); + assertEquals("fedDirect", exchange.getName()); + assertTrue(exchange.isDurable()); + assertFalse(exchange.isAutoDelete()); + assertEquals("direct", exchange.getArguments().get("type")); + assertEquals("upstream-set1", exchange.getArguments().get("upstream-set")); + } + + @Test + public void testFederatedTopicExchange() throws Exception { + FederatedExchange exchange = beanFactory.getBean("fedTopic", FederatedExchange.class); + assertNotNull(exchange); + assertEquals("fedTopic", exchange.getName()); + assertTrue(exchange.isDurable()); + assertFalse(exchange.isAutoDelete()); + assertEquals("topic", exchange.getArguments().get("type")); + assertEquals("upstream-set2", exchange.getArguments().get("upstream-set")); + } + + @Test + public void testFederatedFanoutExchange() throws Exception { + FederatedExchange exchange = beanFactory.getBean("fedFanout", FederatedExchange.class); + assertNotNull(exchange); + assertEquals("fedFanout", exchange.getName()); + assertTrue(exchange.isDurable()); + assertFalse(exchange.isAutoDelete()); + assertEquals("fanout", exchange.getArguments().get("type")); + assertEquals("upstream-set3", exchange.getArguments().get("upstream-set")); + } + + @Test + public void testFederatedHeadersExchange() throws Exception { + FederatedExchange exchange = beanFactory.getBean("fedHeaders", FederatedExchange.class); + assertNotNull(exchange); + assertEquals("fedHeaders", exchange.getName()); + assertTrue(exchange.isDurable()); + assertFalse(exchange.isAutoDelete()); + assertEquals("headers", exchange.getArguments().get("type")); + assertEquals("upstream-set4", exchange.getArguments().get("upstream-set")); + } + } diff --git a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/config/FederatedExchangeParserIntegrationTests.java b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/config/FederatedExchangeParserIntegrationTests.java new file mode 100644 index 00000000..dc1419ea --- /dev/null +++ b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/config/FederatedExchangeParserIntegrationTests.java @@ -0,0 +1,67 @@ +/* + * Copyright 2002-2012 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.assertEquals; + +import org.junit.Rule; +import org.junit.Test; +import org.junit.runner.RunWith; +import org.springframework.amqp.core.Exchange; +import org.springframework.amqp.core.Queue; +import org.springframework.amqp.rabbit.connection.ConnectionFactory; +import org.springframework.amqp.rabbit.core.RabbitAdmin; +import org.springframework.amqp.rabbit.core.RabbitTemplate; +import org.springframework.amqp.rabbit.test.BrokerFederated; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.beans.factory.annotation.Qualifier; +import org.springframework.test.context.ContextConfiguration; +import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; + +@ContextConfiguration +@RunWith(SpringJUnit4ClassRunner.class) +public final class FederatedExchangeParserIntegrationTests { + + @Rule + public BrokerFederated brokerFederated = BrokerFederated.isRunning(); + + @Autowired + private ConnectionFactory connectionFactory; + + @Autowired + private Exchange fanoutTest; + + @Autowired + @Qualifier("bucket") + private Queue queue; + + @Autowired + private RabbitAdmin admin; + + @Test + public void testBindingsDeclared() throws Exception { + RabbitTemplate template = new RabbitTemplate(connectionFactory); + template.convertAndSend(fanoutTest.getName(), "", "message"); + Thread.sleep(200); + // The queue is anonymous so it will be deleted at the end of the test, but it should get the message as long as + // we use the same connection + String result = (String) template.receiveAndConvert(queue.getName()); + assertEquals("message", result); + admin.deleteExchange("fedDirectTest"); + admin.deleteExchange("fedTopicTest"); + admin.deleteExchange("fedFanoutTest"); + admin.deleteExchange("fedHeadersTest"); + } + +} diff --git a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/test/BrokerFederated.java b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/test/BrokerFederated.java new file mode 100644 index 00000000..b1b0aee7 --- /dev/null +++ b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/test/BrokerFederated.java @@ -0,0 +1,150 @@ +/* + * Copyright 2002-2012 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.test; + +import java.util.HashMap; +import java.util.Map; + +import org.apache.commons.logging.Log; +import org.apache.commons.logging.LogFactory; +import org.junit.Assume; +import org.junit.internal.AssumptionViolatedException; +import org.junit.rules.TestWatchman; +import org.junit.runners.model.FrameworkMethod; +import org.junit.runners.model.Statement; +import org.springframework.amqp.core.FederatedExchange; +import org.springframework.amqp.rabbit.connection.CachingConnectionFactory; +import org.springframework.amqp.rabbit.core.RabbitAdmin; +import org.springframework.util.StringUtils; + +/** + *

+ * A rule that prevents integration tests from failing if the Rabbit broker application is not running or not + * accessible, with test federation configuration: + * 

+ * [
+ *    {rabbitmq_federation,
+ *     [
+ *       {upstream_sets, [{"upstream-set", [[{connection, "upstream-server"} ]]}
+ *                       ]},
+ *       {connections, [{"upstream-server", [{host, "somehost"}]}
+ *                     ]}
+ *     ]
+ *    }
+ * ]
+ *
+ *

+ * + * @See BrokerRunning + * @see Assume + * @see AssumptionViolatedException + * + * @author Gary Russell + * + */ +public class BrokerFederated extends TestWatchman { + + private static Log logger = LogFactory.getLog(BrokerFederated.class); + + // Static so that we only test once on failure: speeds up test suite + private static Map brokerOnline = new HashMap(); + + // Static so that we only test once on failure + private static Map brokerOffline = new HashMap(); + + private final boolean assumeOnline; + + private int DEFAULT_PORT = BrokerTestUtils.getPort(); + + private int port; + + private String hostName = null; + + /** + * @return a new rule that assumes an existing running broker + */ + public static BrokerFederated isRunning() { + return new BrokerFederated(); + } + + private BrokerFederated() { + this.assumeOnline = true; + setPort(DEFAULT_PORT); + } + + /** + * @param port the port to set + */ + public void setPort(int port) { + this.port = port; + if (!brokerOffline.containsKey(port)) { + brokerOffline.put(port, true); + } + if (!brokerOnline.containsKey(port)) { + brokerOnline.put(port, true); + } + } + + /** + * @param hostName the hostName to set + */ + public void setHostName(String hostName) { + this.hostName = hostName; + } + + @Override + public Statement apply(Statement base, FrameworkMethod method, Object target) { + + // Check at the beginning, so this can be used as a static field + if (assumeOnline) { + Assume.assumeTrue(brokerOnline.get(port)); + } else { + Assume.assumeTrue(brokerOffline.get(port)); + } + + CachingConnectionFactory connectionFactory = new CachingConnectionFactory(); + + try { + + connectionFactory.setPort(port); + if (StringUtils.hasText(hostName)) { + connectionFactory.setHost(hostName); + } + RabbitAdmin admin = new RabbitAdmin(connectionFactory); + FederatedExchange exchange = new FederatedExchange("fedDirectRuleTest"); + exchange.setBackingType("direct"); + exchange.setUpstreamSet("upstream-set"); + admin.declareExchange(exchange); + admin.deleteExchange("fedDirectRuleTest"); + + brokerOffline.put(port, false); + + if (!assumeOnline) { + Assume.assumeTrue(brokerOffline.get(port)); + } + + } catch (Exception e) { + logger.warn("Not executing tests because federated connectivity test failed", e); + brokerOnline.put(port, false); + if (assumeOnline) { + Assume.assumeNoException(e); + } + } finally { + connectionFactory.destroy(); + } + + return super.apply(base, method, target); + + } + +} diff --git a/spring-rabbit/src/test/resources/org/springframework/amqp/rabbit/config/ExchangeParserTests-context.xml b/spring-rabbit/src/test/resources/org/springframework/amqp/rabbit/config/ExchangeParserTests-context.xml index 38c533f7..65c4c613 100644 --- a/spring-rabbit/src/test/resources/org/springframework/amqp/rabbit/config/ExchangeParserTests-context.xml +++ b/spring-rabbit/src/test/resources/org/springframework/amqp/rabbit/config/ExchangeParserTests-context.xml @@ -22,4 +22,32 @@ + + + + + + + + + + + + + + + + + + + + + + + + diff --git a/spring-rabbit/src/test/resources/org/springframework/amqp/rabbit/config/FederatedExchangeParserIntegrationTests-context.xml b/spring-rabbit/src/test/resources/org/springframework/amqp/rabbit/config/FederatedExchangeParserIntegrationTests-context.xml new file mode 100644 index 00000000..cff38383 --- /dev/null +++ b/spring-rabbit/src/test/resources/org/springframework/amqp/rabbit/config/FederatedExchangeParserIntegrationTests-context.xml @@ -0,0 +1,48 @@ + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + diff --git a/src/docbkx/amqp.xml b/src/docbkx/amqp.xml index 4f09cc0e..2b63b69a 100644 --- a/src/docbkx/amqp.xml +++ b/src/docbkx/amqp.xml @@ -83,7 +83,7 @@ String getName(); - ExchangeType getExchangeType(); + String getExchangeType(); boolean isDurable(); @@ -93,19 +93,20 @@ }]]> - As you can see, an Exchange also has a 'type' represented by the - enumeration ExchangeType. The basic types are: - Direct, Topic and - Fanout. In the core package you will find + As you can see, an Exchange also has a 'type' represented by constants + defined in ExchangeTypes. The basic types are: + Direct, Topic, + Fanout, Headers and + Federated. In the core package you will find implementations of the Exchange interface for each of those types. The behavior varies across these Exchange types - in terms of how they handle bindings to Queues. A Direct exchange allows + in terms of how they handle bindings to Queues. For example, a Direct exchange allows for a Queue to be bound by a fixed routing key (often the Queue's name). A Topic exchange supports bindings with routing patterns that may include the '*' and '#' wildcards for 'exactly-one' and 'zero-or-more', respectively. The Fanout exchange publishes to all Queues that are bound to it without taking any routing key into consideration. For much more - information about Exchange types, check out . @@ -890,6 +891,34 @@ public class RabbitClientConfiguration extends AbstractStockAppRabbitConfigurati The client is declaring another queue via the declareQueue() method on the AmqpAdmin, and it binds that queue to the market data exchange with a routing pattern that is externalized in a properties file. +
+ Federated Exchanges + + Rabbit supports federation; federated exchanges are backed by one of the + other exchange types. Therefore, when configuring a federated exchange, it + is important to supply bindings of the appropriate type for the backing + exchange. Examples include... + + + + + + + + + + + + +]]> + + + Notice that the child element, e.g. <direct-bindings/> matches the + backing-type attribute. + +
diff --git a/src/docbkx/further-reading.xml b/src/docbkx/further-reading.xml index 6f1135f3..f5dfba97 100644 --- a/src/docbkx/further-reading.xml +++ b/src/docbkx/further-reading.xml @@ -9,7 +9,7 @@ specification is actually quite readable. It is of course the authoritative source of information, and the Spring AMQP code should be very easy to understand for anyone who is familiar with the spec. Our current - implementation of the RabbitMQ support is based on their 2.5.x version, and it officially supports + implementation of the RabbitMQ support is based on their 2.8.x version, and it officially supports AMQP 0.8 and 0.9.1. We recommend reading the 0.9.1 document.