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

Merge pull request #65 from markfisher/INT-2118

Browse Source 
added 'exchange-name-expression' attributes to the schema
  added support for 'exchange-name-expression' on AMQP outbound adapters
  changed default connection-factory value to 'rabbitConnectionFactory' to be consistent with the Spring AMQP listener-container
This commit is contained in:
Mark Fisher
2011-09-13 11:51:38 -04:00
parent 797b4f39f0 84375120ad
commit a28e554749
5 changed files with 49 additions and 14 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
spring-integration-amqp/src/main/java/org/springframework/integration/amqp/config/AbstractAmqpInboundAdapterParser.java
View File

@@ -111,7 +111,7 @@ abstract class AbstractAmqpInboundAdapterParser extends AbstractSingleBeanDefini
"org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer");
String connectionFactoryRef = element.getAttribute("connection-factory");
if (!StringUtils.hasText(connectionFactoryRef)) {
connectionFactoryRef = "connectionFactory";
connectionFactoryRef = "rabbitConnectionFactory";
}
builder.addConstructorArgReference(connectionFactoryRef);
for (String attributeName : CONTAINER_VALUE_ATTRIBUTES) {

1
spring-integration-amqp/src/main/java/org/springframework/integration/amqp/config/AmqpOutboundChannelAdapterParser.java
View File

@@ -42,6 +42,7 @@ public class AmqpOutboundChannelAdapterParser extends AbstractOutboundChannelAda
}
builder.addConstructorArgReference(amqpTemplateRef);
IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "exchange-name");
IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "exchange-name-expression");
IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "routing-key");
IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "routing-key-expression");
return builder.getBeanDefinition();

3
spring-integration-amqp/src/main/java/org/springframework/integration/amqp/config/AmqpOutboundGatewayParser.java
View File

@@ -45,9 +45,10 @@ public class AmqpOutboundGatewayParser extends AbstractConsumerEndpointParser {
builder.addConstructorArgReference(amqpTemplateRef);
builder.addPropertyValue("expectReply", true);
IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "exchange-name");
IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "exchange-name-expression");
IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "routing-key");
IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "reply-channel", "outputChannel");
IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "routing-key-expression");
IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "reply-channel", "outputChannel");
return builder;
}

41
spring-integration-amqp/src/main/java/org/springframework/integration/amqp/outbound/AmqpOutboundEndpoint.java
View File

@@ -41,19 +41,24 @@ import org.springframework.util.Assert;
*/
public class AmqpOutboundEndpoint extends AbstractReplyProducingMessageHandler {
private static final ExpressionParser expressionParser = new SpelExpressionParser(new SpelParserConfiguration(true, true));
private final AmqpTemplate amqpTemplate;
private volatile boolean expectReply;
private volatile String exchangeName = "";
private volatile String routingKey = "";
private volatile boolean expectReply;
private volatile String exchangeNameExpression;
private static final ExpressionParser expressionParser = new SpelExpressionParser(new SpelParserConfiguration(true, true));
private volatile String routingKeyExpression;
private volatile ExpressionEvaluatingMessageProcessor<String> routingKeyGenerator;
private volatile String routingKeyExpression;
private volatile ExpressionEvaluatingMessageProcessor<String> exchangeNameGenerator;
private volatile AmqpHeaderMapper headerMapper = new DefaultAmqpHeaderMapper();
@@ -61,9 +66,15 @@ public class AmqpOutboundEndpoint extends AbstractReplyProducingMessageHandler {
@Override
protected void onInit() {
super.onInit();
Assert.state(exchangeNameExpression == null || "".equals(exchangeName),
"Either an exchangeName or an exchangeNameExpression can be provided, but not both");
if (exchangeNameExpression != null) {
Expression expression = expressionParser.parseExpression(this.exchangeNameExpression);
this.exchangeNameGenerator = new ExpressionEvaluatingMessageProcessor<String>(expression, String.class);
}
Assert.state(routingKeyExpression == null || "".equals(routingKey),
"Either a routingKey or a routingKeyExpression can be provided, but not both");
if (routingKeyExpression!=null) {
if (routingKeyExpression != null) {
Expression expression = expressionParser.parseExpression(this.routingKeyExpression);
this.routingKeyGenerator = new ExpressionEvaluatingMessageProcessor<String>(expression, String.class);
}
@@ -78,6 +89,10 @@ public class AmqpOutboundEndpoint extends AbstractReplyProducingMessageHandler {
this.exchangeName = exchangeName;
}
public void setExchangeNameExpression(String exchangeNameExpression) {
this.exchangeNameExpression = exchangeNameExpression;
}
public void setRoutingKey(String routingKey) {
this.routingKey = routingKey;
}
@@ -97,21 +112,25 @@ public class AmqpOutboundEndpoint extends AbstractReplyProducingMessageHandler {
@Override
protected Object handleRequestMessage(Message<?> requestMessage) {
String exchangeName = this.exchangeName;
String routingKey = this.routingKey;
if (this.routingKeyGenerator!=null) {
if (this.exchangeNameGenerator != null) {
exchangeName = this.exchangeNameGenerator.processMessage(requestMessage);
}
if (this.routingKeyGenerator != null) {
routingKey = this.routingKeyGenerator.processMessage(requestMessage);
}
if (this.expectReply) {
return this.sendAndReceive(requestMessage, routingKey);
return this.sendAndReceive(exchangeName, routingKey, requestMessage);
}
else {
this.send(requestMessage, routingKey);
this.send(exchangeName, routingKey, requestMessage);
return null;
}
}
private void send(final Message<?> requestMessage, String routingKey) {
this.amqpTemplate.convertAndSend(this.exchangeName, routingKey, requestMessage.getPayload(),
private void send(String exchangeName, String routingKey, final Message<?> requestMessage) {
this.amqpTemplate.convertAndSend(exchangeName, routingKey, requestMessage.getPayload(),
new MessagePostProcessor() {
public org.springframework.amqp.core.Message postProcessMessage(
org.springframework.amqp.core.Message message) throws AmqpException {
@@ -121,14 +140,14 @@ public class AmqpOutboundEndpoint extends AbstractReplyProducingMessageHandler {
});
}
private Message<?> sendAndReceive(Message<?> requestMessage, String routingKey) {
private Message<?> sendAndReceive(String exchangeName, String routingKey, Message<?> requestMessage) {
// TODO: add a convertSendAndReceive method that accepts a MessagePostProcessor so we can map headers?
Assert.isTrue(amqpTemplate instanceof RabbitTemplate, "RabbitTemplate implementation is required for send and receive");
MessageConverter converter = ((RabbitTemplate) this.amqpTemplate).getMessageConverter();
MessageProperties amqpMessageProperties = new MessageProperties();
this.headerMapper.fromHeaders(requestMessage.getHeaders(), amqpMessageProperties);
org.springframework.amqp.core.Message amqpMessage = converter.toMessage(requestMessage.getPayload(), amqpMessageProperties);
org.springframework.amqp.core.Message amqpReplyMessage = this.amqpTemplate.sendAndReceive(this.exchangeName, routingKey, amqpMessage);
org.springframework.amqp.core.Message amqpReplyMessage = this.amqpTemplate.sendAndReceive(exchangeName, routingKey, amqpMessage);
if (amqpReplyMessage == null) {
return null;
}

16
spring-integration-amqp/src/main/resources/org/springframework/integration/amqp/config/spring-integration-amqp-2.1.xsd
View File

@@ -43,7 +43,14 @@
<xsd:attribute name="exchange-name" type="xsd:string">
<xsd:annotation>
<xsd:documentation>
The name of the AMQP Exchange to which Messages should be sent. If not provided, Messages will be sent to the default, no-name Exchange.
The fixed name of the AMQP Exchange to which Messages should be sent. If not provided, Messages will be sent to the default, no-name Exchange.
</xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="exchange-name-expression" type="xsd:string">
<xsd:annotation>
<xsd:documentation>
The exchange name to use when sending Messages evaluated as an expression on the message (e.g. 'headers.exchange'). By default, this will be an emtpy String.
</xsd:documentation>
</xsd:annotation>
</xsd:attribute>
@@ -156,6 +163,13 @@
</xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="exchange-name-expression" type="xsd:string">
<xsd:annotation>
<xsd:documentation>
The exchange name to use when sending Messages evaluated as an expression on the message (e.g. 'headers.exchange'). By default, this will be an emtpy String.
</xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="routing-key" type="xsd:string">
<xsd:annotation>
<xsd:documentation>