diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/config/xml/AbstractRouterParser.java b/org.springframework.integration/src/main/java/org/springframework/integration/config/xml/AbstractRouterParser.java
index 7f1ab08077..4a9407028e 100644
--- a/org.springframework.integration/src/main/java/org/springframework/integration/config/xml/AbstractRouterParser.java
+++ b/org.springframework.integration/src/main/java/org/springframework/integration/config/xml/AbstractRouterParser.java
@@ -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;
}
diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/router/AbstractMessageRouter.java b/org.springframework.integration/src/main/java/org/springframework/integration/router/AbstractMessageRouter.java
index e1bccaca83..6743ddec3f 100644
--- a/org.springframework.integration/src/main/java/org/springframework/integration/router/AbstractMessageRouter.java
+++ b/org.springframework.integration/src/main/java/org/springframework/integration/router/AbstractMessageRouter.java
@@ -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 false meaning that an Exception
+ * will be thrown whenever a send fails. To override this and suppress
+ * Exceptions, set the value to true.
+ */
+ 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 false meaning that sequence headers will
+ * not be applied. If planning to use an Aggregator downstream with
+ * the default correlation and completion strategies, you should set this
+ * flag to true.
+ */
+ public void setApplySequence(boolean applySequence) {
+ this.applySequence = applySequence;
+ }
+
@Override
protected void handleMessageInternal(Message message) {
boolean sent = false;
Collection 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);
+ }
}
}
}
diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/router/RecipientListRouter.java b/org.springframework.integration/src/main/java/org/springframework/integration/router/RecipientListRouter.java
index df9d4928d3..483ffe56d6 100644
--- a/org.springframework.integration/src/main/java/org/springframework/integration/router/RecipientListRouter.java
+++ b/org.springframework.integration/src/main/java/org/springframework/integration/router/RecipientListRouter.java
@@ -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> 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 false meaning
- * that an Exception will be thrown whenever a send fails. To override
- * this and suppress Exceptions, set the value to true.
- */
- 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 false meaning that sequence headers will
- * not be applied. If planning to use an Aggregator downstream
- * with the default correlation and completion strategies, you should set
- * this flag to true.
- */
- 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 recipients = new ArrayList();
- Map> map =
- new HashMap>(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 determineTargetChannels(Message message) {
+ List recipients = new ArrayList();
+ Map> map = new HashMap>(this.channelMap);
+ for (MessageSelector selector : map.keySet()) {
+ if (selector.accept(message)) {
+ recipients.addAll(map.get(selector));
+ }
+ }
+ return recipients;
+ }
}
diff --git a/org.springframework.integration/src/main/resources/org/springframework/integration/config/xml/spring-integration-2.0.xsd b/org.springframework.integration/src/main/resources/org/springframework/integration/config/xml/spring-integration-2.0.xsd
index ec636c43fb..864679397a 100644
--- a/org.springframework.integration/src/main/resources/org/springframework/integration/config/xml/spring-integration-2.0.xsd
+++ b/org.springframework.integration/src/main/resources/org/springframework/integration/config/xml/spring-integration-2.0.xsd
@@ -1318,9 +1318,35 @@
-
-
+
+
+
+ Specify whether this router should always be required to return at least one channel or name.
+
+
+
+
+
+
+ Specify whether a failure to resolve a channel name returned by this router should be ignored.
+
+
+
+
+
+
+ Specify whether a failure to send to a single channel should be ignored.
+ Otherwise MessageDeliveryExceptions will be thrown.
+
+
+
+
+
+
+ Specify whether sequence number and size headers should be added to each Message.
+
+
+