From ab18ddd199a8dddb630c138c8abf56a0a59e97f5 Mon Sep 17 00:00:00 2001 From: Mark Fisher Date: Thu, 11 Mar 2010 17:05:28 +0000 Subject: [PATCH] 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 element. --- .../config/xml/AbstractRouterParser.java | 4 +- .../router/AbstractMessageRouter.java | 44 +++++++++- .../router/RecipientListRouter.java | 82 +++---------------- .../config/xml/spring-integration-2.0.xsd | 32 +++++++- 4 files changed, 87 insertions(+), 75 deletions(-) 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. + + +