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

AMQP-215 Federated Exchange Support

Browse Source 
Initial commit - schema, parser, parser tests,
integration tests.

TODO: doc.

Note: Federated exchanges need some hard config
on the server. The test cases use this

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

If not available, the tests will be ignored.

AMQP-215 Reference Documentation

Add information about federated exchanges.

AMQP-215 Polishing - remove test exchanges.
This commit is contained in:
Gary Russell
2012-04-20 19:21:04 -04:00
committed by Oleg Zhurakousky
parent 24accd560e
commit e63fb7896f
16 changed files with 710 additions and 23 deletions
Download Patch File Download Diff File Expand all files Collapse all files

11
spring-amqp-core/src/main/java/org/springframework/amqp/core/AbstractExchange.java
View File

@@ -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<String, Object>();
}
}
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.

5
spring-amqp-core/src/main/java/org/springframework/amqp/core/ExchangeTypes.java
View File

@@ -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";
}

60
spring-amqp-core/src/main/java/org/springframework/amqp/core/FederatedExchange.java Normal file
View File

@@ -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<String,Object> 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;
}
}

28
spring-rabbit/src/main/java/org/springframework/amqp/rabbit/config/AbstractExchangeParser.java
View File

@@ -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);

115
spring-rabbit/src/main/java/org/springframework/amqp/rabbit/config/FederatedExchangeParser.java Normal file
View File

@@ -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");
}
}

1
spring-rabbit/src/main/java/org/springframework/amqp/rabbit/config/RabbitNamespaceHandler.java
View File

@@ -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());

2
spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/CachingConnectionFactory.java
View File

@@ -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;
}

116
spring-rabbit/src/main/resources/org/springframework/amqp/rabbit/config/spring-rabbit-1.0.xsd
View File

@@ -237,6 +237,122 @@
</xsd:complexType>
</xsd:element>
<xsd:element name="federated-exchange">
<xsd:annotation>
<xsd:documentation><![CDATA[
Creates a federated exchange for producers to send messages to. Uses an existing exchange
with the same name if it exists on the broker, or declares a
new one. A federated exchange uses a backing exchange of type defined by
the backing-type attribute. It connects to a set of upstream exchanges as
defined by the upstream-set. The bindings elements must match the bindings
for the backing exchange.
]]></xsd:documentation>
</xsd:annotation>
<xsd:complexType>
<xsd:complexContent>
<xsd:extension base="exchangeType">
<xsd:sequence>
<xsd:choice>
<xsd:element name="direct-bindings" minOccurs="0" maxOccurs="1">
<xsd:annotation>
<xsd:documentation><![CDATA[
Groups bindings of queues to this exchange.
]]></xsd:documentation>
</xsd:annotation>
<xsd:complexType>
<xsd:choice>
<xsd:element name="binding" maxOccurs="unbounded" type="directBindingType">
<xsd:annotation>
<xsd:documentation><![CDATA[
Declares a binding of a queue to this exchange either with
an explicit routing key, or using the queue name implicitly.
]]></xsd:documentation>
</xsd:annotation>
</xsd:element>
</xsd:choice>
</xsd:complexType>
</xsd:element>
<xsd:element name="topic-bindings" minOccurs="0" maxOccurs="1">
<xsd:annotation>
<xsd:documentation><![CDATA[
Groups bindings of queues to this exchange.
]]></xsd:documentation>
</xsd:annotation>
<xsd:complexType>
<xsd:choice>
<xsd:element name="binding" maxOccurs="unbounded" type="topicBindingType">
<xsd:annotation>
<xsd:documentation><![CDATA[
Declares a binding of a queue to this exchange either with
a routing pattern, e.g. "uk.weather.*" or "uk.#".
]]></xsd:documentation>
</xsd:annotation>
</xsd:element>
</xsd:choice>
</xsd:complexType>
</xsd:element>
<xsd:element name="fanout-bindings" minOccurs="0" maxOccurs="1">
<xsd:annotation>
<xsd:documentation><![CDATA[
Groups bindings of queues to this exchange.
]]></xsd:documentation>
</xsd:annotation>
<xsd:complexType>
<xsd:choice>
<xsd:element name="binding" maxOccurs="unbounded" type="bindingType">
<xsd:annotation>
<xsd:documentation><![CDATA[
Binds a queue to this exchange. All messages sent to this exchange will be
placed on this queue by the broker.
]]></xsd:documentation>
</xsd:annotation>
</xsd:element>
</xsd:choice>
</xsd:complexType>
</xsd:element>
<xsd:element name="headers-bindings" minOccurs="0" maxOccurs="1">
<xsd:annotation>
<xsd:documentation><![CDATA[
Groups bindings of queues to this exchange.
]]></xsd:documentation>
</xsd:annotation>
<xsd:complexType>
<xsd:choice>
<xsd:element name="binding" maxOccurs="unbounded" type="headersBindingType">
<xsd:annotation>
<xsd:documentation><![CDATA[
Binds a queue to this exchange. Messages sent to this exchange will be
placed on this queue by the broker if they contain a header that matches
this binding (key-value pair).
]]></xsd:documentation>
</xsd:annotation>
</xsd:element>
</xsd:choice>
</xsd:complexType>
</xsd:element>
</xsd:choice>
</xsd:sequence>
<xsd:attribute name="backing-type" use="optional" type="xsd:string">
<xsd:annotation>
<xsd:documentation><![CDATA[
The type of the exchange used as a backing exchange for the
federated exhange.
]]></xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="upstream-set" use="optional" type="xsd:string">
<xsd:annotation>
<xsd:documentation><![CDATA[
The names of the set of upstream exchanges to which this
federated exchange connects.
]]></xsd:documentation>
</xsd:annotation>
</xsd:attribute>
</xsd:extension>
</xsd:complexContent>
</xsd:complexType>
</xsd:element>
<xsd:complexType name="exchangeType">
<xsd:sequence>
<xsd:element ref="exchange-arguments" minOccurs="0" maxOccurs="1" />

2
spring-rabbit/src/test/java/org/springframework/amqp/rabbit/config/ExchangeParserIntegrationTests.java
View File

@@ -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

55
spring-rabbit/src/test/java/org/springframework/amqp/rabbit/config/ExchangeParserTests.java
View File

@@ -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"));
}
}

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

@@ -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");
}
}

150
spring-rabbit/src/test/java/org/springframework/amqp/rabbit/test/BrokerFederated.java Normal file
View File

@@ -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;
/**
* <p>
* A rule that prevents integration tests from failing if the Rabbit broker application is not running or not
* accessible, with test federation configuration:
* <pre>
* [
* {rabbitmq_federation,
* [
* {upstream_sets, [{"upstream-set", [[{connection, "upstream-server"} ]]}
* ]},
* {connections, [{"upstream-server", [{host, "somehost"}]}
* ]}
* ]
* }
* ]
* </pre>
* </p>
*
* @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<Integer,Boolean> brokerOnline = new HashMap<Integer, Boolean>();
// Static so that we only test once on failure
private static Map<Integer,Boolean> brokerOffline = new HashMap<Integer, Boolean>();
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);
}
}

28
spring-rabbit/src/test/resources/org/springframework/amqp/rabbit/config/ExchangeParserTests-context.xml
View File

@@ -22,4 +22,32 @@
<rabbit:headers-exchange name="headers" xmlns="http://www.springframework.org/schema/rabbit"/>
<rabbit:federated-exchange name="fedDirect" backing-type="direct"
upstream-set="upstream-set1">
<rabbit:direct-bindings>
<rabbit:binding key="x" queue="dirQ" />
</rabbit:direct-bindings>
</rabbit:federated-exchange>
<rabbit:federated-exchange name="fedTopic" backing-type="topic"
upstream-set="upstream-set2">
<rabbit:topic-bindings>
<rabbit:binding pattern="xx.*" queue="topic" />
</rabbit:topic-bindings>
</rabbit:federated-exchange>
<rabbit:federated-exchange name="fedFanout" backing-type="fanout"
upstream-set="upstream-set3">
<rabbit:fanout-bindings>
<rabbit:binding queue="fan" />
</rabbit:fanout-bindings>
</rabbit:federated-exchange>
<rabbit:federated-exchange name="fedHeaders" backing-type="headers"
upstream-set="upstream-set4">
<rabbit:headers-bindings>
<rabbit:binding key="head" value="head" queue="head" />
</rabbit:headers-bindings>
</rabbit:federated-exchange>
</beans>

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

@@ -0,0 +1,48 @@
<?xml version="1.0" encoding="UTF-8"?>
<beans:beans xmlns="http://www.springframework.org/schema/rabbit"
xmlns:beans="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-1.0.xsd
http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd">
<federated-exchange name="fedDirectTest" backing-type="direct"
upstream-set="upstream-set">
<direct-bindings>
<binding queue="bucket" />
</direct-bindings>
</federated-exchange>
<federated-exchange name="fedTopicTest" backing-type="topic"
upstream-set="upstream-set">
<topic-bindings>
<binding queue="bucket" pattern="bucket.#"/>
</topic-bindings>
</federated-exchange>
<federated-exchange name="fedFanoutTest" backing-type="fanout"
upstream-set="upstream-set">
<fanout-bindings>
<binding queue="bucket" />
</fanout-bindings>
</federated-exchange>
<federated-exchange name="fedHeadersTest" backing-type="headers"
upstream-set="upstream-set">
<headers-bindings>
<binding queue="bucket" key="type" value="bucket" />
</headers-bindings>
</federated-exchange>
<fanout-exchange name="fanoutTest">
<bindings>
<binding queue="bucket" />
</bindings>
</fanout-exchange>
<queue name="bucket" />
<admin id="admin-test" connection-factory="connectionFactory" auto-startup="true"/>
<connection-factory id="connectionFactory"/>
</beans:beans>

43
src/docbkx/amqp.xml
View File

@@ -83,7 +83,7 @@
String getName();
ExchangeType getExchangeType();
String getExchangeType();
boolean isDurable();
@@ -93,19 +93,20 @@
}]]></programlisting>
<para>As you can see, an Exchange also has a 'type' represented by the
enumeration <classname>ExchangeType</classname>. The basic types are:
<literal>Direct</literal>, <literal>Topic</literal> and
<literal>Fanout</literal>. In the core package you will find
<para>As you can see, an Exchange also has a 'type' represented by constants
defined in <classname>ExchangeTypes</classname>. The basic types are:
<literal>Direct</literal>, <literal>Topic</literal>,
<literal>Fanout</literal>, <literal>Headers</literal> and
<literal>Federated</literal>. In the core package you will find
implementations of the <interfacename>Exchange</interfacename> 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 <xref
information about these and the other Exchange types, check out <xref
linkend="resources" />.</para>
<note>
@@ -890,6 +891,34 @@ public class RabbitClientConfiguration extends AbstractStockAppRabbitConfigurati
<para>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.</para>
<section>
<title>Federated Exchanges</title>
<para>
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...
</para>
<para>
<programlisting><![CDATA[<federated-exchange name="fedDirect" backing-type="direct"
upstream-set="upstream-set">
<direct-bindings>
<binding queue="bucket" />
</direct-bindings>
</federated-exchange>
<federated-exchange name="fedTopic" backing-type="topic"
upstream-set="upstream-set">
<topic-bindings>
<binding queue="bucket" pattern="bucket.#"/>
</topic-bindings>
</federated-exchange>]]></programlisting>
</para>
<para>
Notice that the child element, e.g. &lt;direct-bindings/&gt; matches the
backing-type attribute.
</para>
</section>
</section>
<section>

2
src/docbkx/further-reading.xml
View File

@@ -9,7 +9,7 @@
<ulink url="http://www.amqp.org/confluence/display/AMQP/AMQP+Specification">specification</ulink>
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.
</para>