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

INT-1019, INT-745 Moved apply-sequence logic into AbstractMessageRouter and made RecipientListRouter a subclass. Also added support for the 'apply-sequence' attribute on a <router/> element.

Browse Source
This commit is contained in:
Mark Fisher
2010-03-11 17:05:28 +00:00
parent a417786782
commit ab18ddd199
4 changed files with 87 additions and 75 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
org.springframework.integration/src/main/java/org/springframework/integration/config/xml/AbstractRouterParser.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2009 the original author or authors.
* 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.
@@ -36,6 +36,8 @@ public abstract class AbstractRouterParser extends AbstractConsumerEndpointParse
IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "timeout");
IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "resolution-required");
IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "ignore-channel-name-resolution-failures");
IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "apply-sequence");
IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "ignore-send-failures");
this.parseRouter(element, builder, parserContext);
return builder;
}

44
org.springframework.integration/src/main/java/org/springframework/integration/router/AbstractMessageRouter.java
View File

@@ -17,11 +17,14 @@
package org.springframework.integration.router;
import java.util.Collection;
import java.util.UUID;
import org.springframework.integration.channel.MessageChannelTemplate;
import org.springframework.integration.core.Message;
import org.springframework.integration.core.MessageChannel;
import org.springframework.integration.core.MessageHeaders;
import org.springframework.integration.handler.AbstractMessageHandler;
import org.springframework.integration.message.MessageBuilder;
import org.springframework.integration.message.MessageDeliveryException;
/**
@@ -38,6 +41,10 @@ public abstract class AbstractMessageRouter extends AbstractMessageHandler {
private volatile boolean resolutionRequired;
private volatile boolean ignoreSendFailures;
private volatile boolean applySequence;
private final MessageChannelTemplate channelTemplate = new MessageChannelTemplate();
@@ -69,16 +76,51 @@ public abstract class AbstractMessageRouter extends AbstractMessageHandler {
this.resolutionRequired = resolutionRequired;
}
/**
* Specify whether send failures for one or more of the recipients should be
* ignored. By default this is <code>false</code> meaning that an Exception
* will be thrown whenever a send fails. To override this and suppress
* Exceptions, set the value to <code>true</code>.
*/
public void setIgnoreSendFailures(boolean ignoreSendFailures) {
this.ignoreSendFailures = ignoreSendFailures;
}
/**
* Specify whether to apply the sequence number and size headers to the
* messages prior to sending to the recipient channels. By default, this
* value is <code>false</code> meaning that sequence headers will
* <em>not</em> be applied. If planning to use an Aggregator downstream with
* the default correlation and completion strategies, you should set this
* flag to <code>true</code>.
*/
public void setApplySequence(boolean applySequence) {
this.applySequence = applySequence;
}
@Override
protected void handleMessageInternal(Message<?> message) {
boolean sent = false;
Collection<MessageChannel> results = this.determineTargetChannels(message);
if (results != null) {
int sequenceSize = results.size();
int sequenceNumber = 1;
for (MessageChannel channel : results) {
final Message<?> messageToSend = (!this.applySequence) ? message
: MessageBuilder.fromMessage(message)
.setSequenceNumber(sequenceNumber++)
.setSequenceSize(sequenceSize)
.setCorrelationId(message.getHeaders().getId())
.setHeader(MessageHeaders.ID, UUID.randomUUID())
.build();
if (channel != null) {
if (this.channelTemplate.send(message, channel)) {
if (this.channelTemplate.send(messageToSend, channel)) {
sent = true;
}
else if (!this.ignoreSendFailures) {
throw new MessageDeliveryException(message,
"Router failed to send to channel: " + channel);
}
}
}
}

82
org.springframework.integration/src/main/java/org/springframework/integration/router/RecipientListRouter.java
View File

@@ -22,16 +22,10 @@ import java.util.Collections;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.UUID;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.integration.channel.MessageChannelTemplate;
import org.springframework.integration.core.Message;
import org.springframework.integration.core.MessageChannel;
import org.springframework.integration.core.MessageHeaders;
import org.springframework.integration.handler.AbstractMessageHandler;
import org.springframework.integration.message.MessageBuilder;
import org.springframework.integration.message.MessageDeliveryException;
import org.springframework.integration.selector.MessageSelector;
import org.springframework.util.Assert;
@@ -63,19 +57,13 @@ import org.springframework.util.Assert;
*
* @author Mark Fisher
*/
public class RecipientListRouter extends AbstractMessageHandler implements InitializingBean {
public class RecipientListRouter extends AbstractMessageRouter implements InitializingBean {
public static final String COMPONENT_TYPE_LABEL = "recipient-list-router";
private volatile boolean ignoreSendFailures;
private volatile boolean applySequence;
private volatile Map<MessageSelector, ? extends Collection<MessageChannel>> channelMap;
private final MessageChannelTemplate channelTemplate = new MessageChannelTemplate();
/**
* Set the output channels of this router.
@@ -102,66 +90,20 @@ public class RecipientListRouter extends AbstractMessageHandler implements Initi
this.channelMap = channelMap;
}
/**
* Set the timeout for sending a message to the resolved channel(s). By
* default, there is no timeout, meaning the send will block indefinitely.
*/
public void setTimeout(long timeout) {
this.channelTemplate.setSendTimeout(timeout);
}
/**
* Specify whether send failures for one or more of the recipients
* should be ignored. By default this is <code>false</code> meaning
* that an Exception will be thrown whenever a send fails. To override
* this and suppress Exceptions, set the value to <code>true</code>.
*/
public void setIgnoreSendFailures(boolean ignoreSendFailures) {
this.ignoreSendFailures = ignoreSendFailures;
}
/**
* Specify whether to apply the sequence number and size headers to the
* messages prior to sending to the recipient channels. By default, this
* value is <code>false</code> meaning that sequence headers will
* <em>not</em> be applied. If planning to use an Aggregator downstream
* with the default correlation and completion strategies, you should set
* this flag to <code>true</code>.
*/
public void setApplySequence(boolean applySequence) {
this.applySequence = applySequence;
}
public void afterPropertiesSet() {
Assert.notEmpty(this.channelMap, "a non-empty channel map is required");
}
@Override
protected void handleMessageInternal(Message<?> message) throws Exception {
List<MessageChannel> recipients = new ArrayList<MessageChannel>();
Map<MessageSelector, Collection<MessageChannel>> map =
new HashMap<MessageSelector, Collection<MessageChannel>>(this.channelMap);
for (MessageSelector selector : map.keySet()) {
if (selector.accept(message)) {
recipients.addAll(map.get(selector));
}
}
int sequenceSize = recipients.size();
int sequenceNumber = 1;
for (MessageChannel channel : recipients) {
final Message<?> messageToSend = (!this.applySequence) ? message
: MessageBuilder.fromMessage(message)
.setSequenceNumber(sequenceNumber++)
.setSequenceSize(sequenceSize)
.setCorrelationId(message.getHeaders().getId())
.setHeader(MessageHeaders.ID, UUID.randomUUID())
.build();
boolean sent = this.channelTemplate.send(messageToSend, channel);
if (!sent && !this.ignoreSendFailures) {
throw new MessageDeliveryException(message,
"RecipientListRouter failed to send to channel: " + channel);
}
}
}
@Override
protected Collection<MessageChannel> determineTargetChannels(Message<?> message) {
List<MessageChannel> recipients = new ArrayList<MessageChannel>();
Map<MessageSelector, Collection<MessageChannel>> map = new HashMap<MessageSelector, Collection<MessageChannel>>(this.channelMap);
for (MessageSelector selector : map.keySet()) {
if (selector.accept(message)) {
recipients.addAll(map.get(selector));
}
}
return recipients;
}
}

32
org.springframework.integration/src/main/resources/org/springframework/integration/config/xml/spring-integration-2.0.xsd
View File

@@ -1318,9 +1318,35 @@
</xsd:appinfo>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="resolution-required" type="xsd:string" />
<xsd:attribute name="ignore-channel-name-resolution-failures"
type="xsd:string" />
<xsd:attribute name="resolution-required" type="xsd:string">
<xsd:annotation>
<xsd:documentation>
Specify whether this router should always be required to return at least one channel or name.
</xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="ignore-channel-name-resolution-failures" type="xsd:string">
<xsd:annotation>
<xsd:documentation>
Specify whether a failure to resolve a channel name returned by this router should be ignored.
</xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="ignore-send-failures" type="xsd:string">
<xsd:annotation>
<xsd:documentation>
Specify whether a failure to send to a single channel should be ignored.
Otherwise MessageDeliveryExceptions will be thrown.
</xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="apply-sequence" type="xsd:string" >
<xsd:annotation>
<xsd:documentation>
Specify whether sequence number and size headers should be added to each Message.
</xsd:documentation>
</xsd:annotation>
</xsd:attribute>
</xsd:extension>
</xsd:complexContent>
</xsd:complexType>