From 9d22d8abce115be627b0611535d1bcaef28b9d37 Mon Sep 17 00:00:00 2001 From: Dave Syer Date: Fri, 10 Sep 2010 10:44:25 +0100 Subject: [PATCH] INT-1347: move sequence details to MessageBuilder so they can be shared by router and splitter --- .../integration/MessageHeaders.java | 2 + ...tractAggregatingMessageGroupProcessor.java | 33 +----- .../dispatcher/BroadcastingDispatcher.java | 53 +++------ .../AbstractReplyProducingMessageHandler.java | 4 +- .../router/AbstractMessageRouter.java | 48 +++----- .../splitter/AbstractMessageSplitter.java | 45 +++----- .../integration/support/MessageBuilder.java | 109 ++++++++++++------ .../NestedAggregationTests-context.xml | 35 +++--- .../scenarios/NestedAggregationTests.java | 24 +++- .../endpoint/CorrelationIdTests.java | 7 +- .../src/test/java/log4j.properties | 4 +- 11 files changed, 172 insertions(+), 192 deletions(-) diff --git a/spring-integration-core/src/main/java/org/springframework/integration/MessageHeaders.java b/spring-integration-core/src/main/java/org/springframework/integration/MessageHeaders.java index 80b26e0bde..85f3c8f2a4 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/MessageHeaders.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/MessageHeaders.java @@ -70,6 +70,8 @@ public final class MessageHeaders implements Map, Serializable { public static final String SEQUENCE_SIZE = PREFIX + "sequenceSize"; + public static final String SEQUENCE_DETAILS = PREFIX + "sequenceDetails"; + private final Map headers; diff --git a/spring-integration-core/src/main/java/org/springframework/integration/aggregator/AbstractAggregatingMessageGroupProcessor.java b/spring-integration-core/src/main/java/org/springframework/integration/aggregator/AbstractAggregatingMessageGroupProcessor.java index 0253aa7901..2937f065c5 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/aggregator/AbstractAggregatingMessageGroupProcessor.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/aggregator/AbstractAggregatingMessageGroupProcessor.java @@ -13,20 +13,15 @@ package org.springframework.integration.aggregator; -import java.util.ArrayList; -import java.util.Arrays; import java.util.HashMap; import java.util.HashSet; -import java.util.List; import java.util.Map; import java.util.Set; import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; - import org.springframework.integration.Message; import org.springframework.integration.MessageHeaders; -import org.springframework.integration.splitter.AbstractMessageSplitter; import org.springframework.integration.store.MessageGroup; import org.springframework.integration.support.MessageBuilder; import org.springframework.util.Assert; @@ -44,19 +39,18 @@ public abstract class AbstractAggregatingMessageGroupProcessor implements Messag private final Log logger = LogFactory.getLog(this.getClass()); - public final Object processMessageGroup(MessageGroup group) { Assert.notNull(group, "MessageGroup must not be null"); Map headers = this.aggregateHeaders(group); Object payload = this.aggregatePayloads(group, headers); MessageBuilder builder; if (payload instanceof Message) { - builder = MessageBuilder.fromMessage((Message) payload); + builder = MessageBuilder.fromMessage((Message) payload).copyHeadersIfAbsent(headers); } else { builder = MessageBuilder.withPayload(payload).copyHeadersIfAbsent(headers); } - return builder.build(); + return builder.popSequenceDetails().build(); } /** @@ -71,28 +65,7 @@ public abstract class AbstractAggregatingMessageGroupProcessor implements Messag MessageHeaders currentHeaders = message.getHeaders(); for (String key : currentHeaders.keySet()) { if (MessageHeaders.ID.equals(key) || MessageHeaders.TIMESTAMP.equals(key) - || MessageHeaders.SEQUENCE_SIZE.equals(key) || MessageHeaders.SEQUENCE_NUMBER.equals(key) - || MessageHeaders.CORRELATION_ID.equals(key)) { - continue; - } - if (AbstractMessageSplitter.SEQUENCE_DETAILS.equals(key) - && !aggregatedHeaders.containsKey(MessageHeaders.CORRELATION_ID)) { - @SuppressWarnings("unchecked") - List incomingSequenceDetails = new ArrayList(currentHeaders - .get(key, List.class)); - Object[] sequenceDetails = incomingSequenceDetails.remove(incomingSequenceDetails.size() - 1); - Assert.state(sequenceDetails.length == 3, "Wrong sequence details (not created by splitter?): " - + Arrays.asList(sequenceDetails)); - aggregatedHeaders.put(MessageHeaders.CORRELATION_ID, sequenceDetails[0]); - Integer sequenceNumber = (Integer) sequenceDetails[1]; - Integer sequenceSize = (Integer) sequenceDetails[2]; - if (sequenceSize > 0) { - aggregatedHeaders.put(MessageHeaders.SEQUENCE_NUMBER, sequenceNumber); - aggregatedHeaders.put(MessageHeaders.SEQUENCE_SIZE, sequenceSize); - } - if (!incomingSequenceDetails.isEmpty()) { - aggregatedHeaders.put(AbstractMessageSplitter.SEQUENCE_DETAILS, incomingSequenceDetails); - } + || MessageHeaders.SEQUENCE_SIZE.equals(key) || MessageHeaders.SEQUENCE_NUMBER.equals(key)) { continue; } Object value = currentHeaders.get(key); diff --git a/spring-integration-core/src/main/java/org/springframework/integration/dispatcher/BroadcastingDispatcher.java b/spring-integration-core/src/main/java/org/springframework/integration/dispatcher/BroadcastingDispatcher.java index 551b0f4e16..2cac5f6bf4 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/dispatcher/BroadcastingDispatcher.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/dispatcher/BroadcastingDispatcher.java @@ -17,28 +17,22 @@ package org.springframework.integration.dispatcher; import java.util.List; -import java.util.UUID; import java.util.concurrent.Executor; import org.springframework.integration.Message; -import org.springframework.integration.MessageHeaders; import org.springframework.integration.MessagingException; import org.springframework.integration.core.MessageHandler; import org.springframework.integration.support.MessageBuilder; /** - * A broadcasting dispatcher implementation. If the 'ignoreFailures' property - * is set to false (the default), it will fail fast such that any - * Exception thrown by a MessageHandler may prevent subsequent handlers from - * receiving the Message. However, when an Executor is provided, the Messages - * may be dispatched in separate Threads so that other handlers are invoked even - * when the 'ignoreFailures' flag is false. + * A broadcasting dispatcher implementation. If the 'ignoreFailures' property is set to false (the + * default), it will fail fast such that any Exception thrown by a MessageHandler may prevent subsequent handlers from + * receiving the Message. However, when an Executor is provided, the Messages may be dispatched in separate Threads so + * that other handlers are invoked even when the 'ignoreFailures' flag is false. *

- * If the 'ignoreFailures' flag is set to true on the other hand, - * it will make a best effort to send the message to each of its handlers. In - * other words, when 'ignoreFailures' is true, if it fails to send - * to any one handler, it will simply log a warn-level message but continue to - * send the Message to any other handlers. + * If the 'ignoreFailures' flag is set to true on the other hand, it will make a best effort to send the + * message to each of its handlers. In other words, when 'ignoreFailures' is true, if it fails to send to + * any one handler, it will simply log a warn-level message but continue to send the Message to any other handlers. * * @author Mark Fisher * @author Iwein Fuld @@ -52,7 +46,6 @@ public class BroadcastingDispatcher extends AbstractDispatcher { private final Executor executor; - public BroadcastingDispatcher() { this.executor = null; } @@ -61,26 +54,22 @@ public class BroadcastingDispatcher extends AbstractDispatcher { this.executor = executor; } - /** - * Specify whether failures for one or more of the handlers should be - * ignored. By default this is false meaning that an - * Exception will be thrown when a handler fails. To override this and - * suppress Exceptions, set the value to true. + * Specify whether failures for one or more of the handlers should be ignored. By default this is false + * meaning that an Exception will be thrown when a handler fails. To override this and suppress Exceptions, set the + * value to true. *

- * Keep in mind that when using an Executor, even without ignoring the - * failures, other handlers may be invoked after one throws an Exception. - * Since the Executor is most likely using a different thread, this flag would - * only affect whether an error Message is sent to the error channel or not in - * the case that such an Executor has been configured. + * Keep in mind that when using an Executor, even without ignoring the failures, other handlers may be invoked after + * one throws an Exception. Since the Executor is most likely using a different thread, this flag would only affect + * whether an error Message is sent to the error channel or not in the case that such an Executor has been + * configured. */ public void setIgnoreFailures(boolean ignoreFailures) { this.ignoreFailures = ignoreFailures; } /** - * Specify whether to apply sequence numbers to the messages - * prior to sending to the handlers. By default, sequence + * Specify whether to apply sequence numbers to the messages prior to sending to the handlers. By default, sequence * numbers will not be applied */ public void setApplySequence(boolean applySequence) { @@ -93,13 +82,8 @@ public class BroadcastingDispatcher extends AbstractDispatcher { List handlers = this.getHandlers(); int sequenceSize = handlers.size(); for (final MessageHandler handler : handlers) { - final Message messageToSend = (!this.applySequence) ? message - : MessageBuilder.fromMessage(message) - .setSequenceNumber(sequenceNumber++) - .setSequenceSize(sequenceSize) - .setCorrelationId(message.getHeaders().getId()) - .setHeader(MessageHeaders.ID, UUID.randomUUID()) - .build(); + final Message messageToSend = (!this.applySequence) ? message : MessageBuilder.fromMessage(message) + .pushSequenceDetails(message.getHeaders().getId(), sequenceNumber++, sequenceSize).build(); if (this.executor != null) { this.executor.execute(new Runnable() { public void run() { @@ -123,8 +107,7 @@ public class BroadcastingDispatcher extends AbstractDispatcher { } catch (RuntimeException e) { if (!this.ignoreFailures) { - if (e instanceof MessagingException && - ((MessagingException) e).getFailedMessage() == null) { + if (e instanceof MessagingException && ((MessagingException) e).getFailedMessage() == null) { ((MessagingException) e).setFailedMessage(message); } throw e; diff --git a/spring-integration-core/src/main/java/org/springframework/integration/handler/AbstractReplyProducingMessageHandler.java b/spring-integration-core/src/main/java/org/springframework/integration/handler/AbstractReplyProducingMessageHandler.java index 3b6daf494e..897fb27099 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/handler/AbstractReplyProducingMessageHandler.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/handler/AbstractReplyProducingMessageHandler.java @@ -114,7 +114,7 @@ public abstract class AbstractReplyProducingMessageHandler extends AbstractMessa } private void handleResult(Object result, MessageHeaders requestHeaders) { - if (result instanceof Iterable && this.shouldSplitReply((Iterable) result)) { + if (result instanceof Iterable && this.shouldSplitReply((Iterable) result)) { for (Object o : (Iterable) result) { this.produceReply(o, requestHeaders); } @@ -131,7 +131,7 @@ public abstract class AbstractReplyProducingMessageHandler extends AbstractMessa private Message createReplyMessage(Object reply, MessageHeaders requestHeaders) { MessageBuilder builder = null; - if (reply instanceof Message) { + if (reply instanceof Message) { if (!this.shouldCopyRequestHeaders()) { return (Message) reply; } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/router/AbstractMessageRouter.java b/spring-integration-core/src/main/java/org/springframework/integration/router/AbstractMessageRouter.java index 100d6c0e12..c314d4c149 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/router/AbstractMessageRouter.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/router/AbstractMessageRouter.java @@ -17,12 +17,10 @@ package org.springframework.integration.router; import java.util.Collection; -import java.util.UUID; import org.springframework.integration.Message; import org.springframework.integration.MessageChannel; import org.springframework.integration.MessageDeliveryException; -import org.springframework.integration.MessageHeaders; import org.springframework.integration.MessagingException; import org.springframework.integration.core.MessagingTemplate; import org.springframework.integration.handler.AbstractMessageHandler; @@ -45,39 +43,35 @@ public abstract class AbstractMessageRouter extends AbstractMessageHandler { private final MessagingTemplate messagingTemplate = new MessagingTemplate(); - /** - * Set the default channel where Messages should be sent if channel - * resolution fails to return any channels. If no default channel is - * provided, the router will either drop the Message or throw an Exception - * depending on the value of {@link #resolutionRequired}. + * Set the default channel where Messages should be sent if channel resolution fails to return any channels. If no + * default channel is provided, the router will either drop the Message or throw an Exception depending on the value + * of {@link #resolutionRequired}. */ public void setDefaultOutputChannel(MessageChannel defaultOutputChannel) { this.defaultOutputChannel = defaultOutputChannel; } /** - * Set the timeout for sending a message to the resolved channel. By - * default, there is no timeout, meaning the send will block indefinitely. + * Set the timeout for sending a message to the resolved channel. By default, there is no timeout, meaning the send + * will block indefinitely. */ public void setTimeout(long timeout) { this.messagingTemplate.setSendTimeout(timeout); } /** - * Set whether this router should always be required to resolve at least one - * channel. The default is 'false'. To trigger an exception whenever the - * resolver returns null or an empty channel list, and this endpoint has - * no 'defaultOutputChannel' configured, set this value to 'true'. + * Set whether this router should always be required to resolve at least one channel. The default is 'false'. To + * trigger an exception whenever the resolver returns null or an empty channel list, and this endpoint has no + * 'defaultOutputChannel' configured, set this value to 'true'. */ public void setResolutionRequired(boolean resolutionRequired) { 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 + * 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) { @@ -85,12 +79,10 @@ public abstract class AbstractMessageRouter extends AbstractMessageHandler { } /** - * 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. + * 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; @@ -116,13 +108,8 @@ public abstract class AbstractMessageRouter extends AbstractMessageHandler { 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(); + final Message messageToSend = (!this.applySequence) ? message : MessageBuilder.fromMessage(message) + .pushSequenceDetails(message.getHeaders().getId(), sequenceNumber++, sequenceSize).build(); if (channel != null) { try { this.messagingTemplate.send(channel, messageToSend); @@ -151,8 +138,7 @@ public abstract class AbstractMessageRouter extends AbstractMessageHandler { } /** - * Subclasses must implement this method to return the target channels for - * a given Message. + * Subclasses must implement this method to return the target channels for a given Message. */ protected abstract Collection determineTargetChannels(Message message); diff --git a/spring-integration-core/src/main/java/org/springframework/integration/splitter/AbstractMessageSplitter.java b/spring-integration-core/src/main/java/org/springframework/integration/splitter/AbstractMessageSplitter.java index 75e3551551..2b7b8ff806 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/splitter/AbstractMessageSplitter.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/splitter/AbstractMessageSplitter.java @@ -18,9 +18,7 @@ package org.springframework.integration.splitter; import java.util.ArrayList; import java.util.Collection; -import java.util.Collections; import java.util.List; -import java.util.UUID; import org.springframework.integration.Message; import org.springframework.integration.MessageHeaders; @@ -35,8 +33,6 @@ import org.springframework.integration.support.MessageBuilder; */ public abstract class AbstractMessageSplitter extends AbstractReplyProducingMessageHandler { - public static final String SEQUENCE_DETAILS = MessageHeaders.PREFIX + "sequenceDetails"; - @Override @SuppressWarnings("unchecked") protected final Object handleRequestMessage(Message message) { @@ -45,19 +41,6 @@ public abstract class AbstractMessageSplitter extends AbstractReplyProducingMess return null; } MessageHeaders headers = message.getHeaders(); - Object incomingCorrelationId = headers.getCorrelationId(); - List incomingSequenceDetails = headers.get(SEQUENCE_DETAILS, List.class); - if (incomingCorrelationId != null) { - if (incomingSequenceDetails == null) { - incomingSequenceDetails = new ArrayList(); - } - else { - incomingSequenceDetails = new ArrayList(incomingSequenceDetails); - } - incomingSequenceDetails.add(new Object[] { - incomingCorrelationId, headers.getSequenceNumber(), headers.getSequenceSize() }); - incomingSequenceDetails = Collections.unmodifiableList(incomingSequenceDetails); - } Object correlationId = headers.getId(); List> messageBuilders = new ArrayList>(); if (result instanceof Collection) { @@ -65,8 +48,7 @@ public abstract class AbstractMessageSplitter extends AbstractReplyProducingMess int sequenceNumber = 0; int sequenceSize = items.size(); for (Object item : items) { - messageBuilders.add(this.createBuilder( - item, incomingSequenceDetails, correlationId, ++sequenceNumber, sequenceSize)); + messageBuilders.add(this.createBuilder(item, headers, correlationId, ++sequenceNumber, sequenceSize)); } } else if (result.getClass().isArray()) { @@ -74,26 +56,27 @@ public abstract class AbstractMessageSplitter extends AbstractReplyProducingMess int sequenceNumber = 0; int sequenceSize = items.length; for (Object item : items) { - messageBuilders.add(this.createBuilder( - item, incomingSequenceDetails, correlationId, ++sequenceNumber, sequenceSize)); + messageBuilders.add(this.createBuilder(item, headers, correlationId, ++sequenceNumber, sequenceSize)); } } else { - messageBuilders.add(this.createBuilder(result, incomingSequenceDetails, correlationId, 1, 1)); + messageBuilders.add(this.createBuilder(result, headers, correlationId, 1, 1)); } return messageBuilders; } - @SuppressWarnings({"unchecked", "rawtypes"}) - private MessageBuilder createBuilder(Object item, List incomingSequenceDetails, Object correlationId, - int sequenceNumber, int sequenceSize) { - MessageBuilder builder = (item instanceof Message) ? MessageBuilder.fromMessage((Message) item) - : MessageBuilder.withPayload(item); - builder.setCorrelationId(correlationId).setSequenceNumber(sequenceNumber).setSequenceSize(sequenceSize) - .setHeader(MessageHeaders.ID, UUID.randomUUID()); - if (incomingSequenceDetails != null) { - builder.setHeader(SEQUENCE_DETAILS, incomingSequenceDetails); + @SuppressWarnings( { "unchecked" }) + private MessageBuilder createBuilder(Object item, MessageHeaders headers, Object correlationId, int sequenceNumber, + int sequenceSize) { + MessageBuilder builder; + if (item instanceof Message) { + builder = MessageBuilder.fromMessage((Message) item); } + else { + builder = MessageBuilder.withPayload(item); + builder.copyHeaders(headers); + } + builder.pushSequenceDetails(correlationId, sequenceNumber, sequenceSize); return builder; } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/support/MessageBuilder.java b/spring-integration-core/src/main/java/org/springframework/integration/support/MessageBuilder.java index 0494371163..d63d69c7b7 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/support/MessageBuilder.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/support/MessageBuilder.java @@ -16,8 +16,12 @@ package org.springframework.integration.support; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.Collections; import java.util.Date; import java.util.HashMap; +import java.util.List; import java.util.Map; import java.util.Set; import java.util.UUID; @@ -34,6 +38,7 @@ import org.springframework.util.StringUtils; * @author Arjen Poutsma * @author Mark Fisher * @author Oleg Zhurakousky + * @author Dave Syer */ public final class MessageBuilder { @@ -45,7 +50,6 @@ public final class MessageBuilder { private volatile boolean modified; - /** * Private constructor to be invoked from the static factory methods only. */ @@ -58,14 +62,11 @@ public final class MessageBuilder { } } - /** - * Create a builder for a new {@link Message} instance pre-populated with - * all of the headers copied from the provided message. The payload of the - * provided Message will also be used as the payload for the new message. + * Create a builder for a new {@link Message} instance pre-populated with all of the headers copied from the + * provided message. The payload of the provided Message will also be used as the payload for the new message. * - * @param message the Message from which the payload and all headers - * will be copied + * @param message the Message from which the payload and all headers will be copied */ public static MessageBuilder fromMessage(Message message) { Assert.notNull(message, "message must not be null"); @@ -83,13 +84,12 @@ public final class MessageBuilder { return builder; } - /** - * Set the value for the given header name. If the provided value is - * null, the header will be removed. + * Set the value for the given header name. If the provided value is null, the header will be removed. */ public MessageBuilder setHeader(String headerName, Object headerValue) { - if (StringUtils.hasLength(headerName) && !headerName.equals(MessageHeaders.ID) && !headerName.equals(MessageHeaders.TIMESTAMP)) { + if (StringUtils.hasLength(headerName) && !headerName.equals(MessageHeaders.ID) + && !headerName.equals(MessageHeaders.TIMESTAMP)) { this.verifyType(headerName, headerValue); this.modified = true; if (headerValue == null) { @@ -103,8 +103,7 @@ public final class MessageBuilder { } /** - * Set the value for the given header name only if the header name - * is not already associated with a value. + * Set the value for the given header name only if the header name is not already associated with a value. */ public MessageBuilder setHeaderIfAbsent(String headerName, Object headerValue) { if (this.headers.get(headerName) == null) { @@ -117,7 +116,8 @@ public final class MessageBuilder { * Remove the value for the given header name. */ public MessageBuilder removeHeader(String headerName) { - if (StringUtils.hasLength(headerName) && !headerName.equals(MessageHeaders.ID) && !headerName.equals(MessageHeaders.TIMESTAMP)) { + if (StringUtils.hasLength(headerName) && !headerName.equals(MessageHeaders.ID) + && !headerName.equals(MessageHeaders.TIMESTAMP)) { this.modified = true; this.headers.remove(headerName); } @@ -125,10 +125,9 @@ public final class MessageBuilder { } /** - * Copy the name-value pairs from the provided Map. This operation will - * overwrite any existing values. Use {{@link #copyHeadersIfAbsent(Map)} - * to avoid overwriting values. Note that the 'id' and 'timestamp' header - * values will never be overwritten. + * Copy the name-value pairs from the provided Map. This operation will overwrite any existing values. Use { + * {@link #copyHeadersIfAbsent(Map)} to avoid overwriting values. Note that the 'id' and 'timestamp' header values + * will never be overwritten. * * @see MessageHeaders#ID * @see MessageHeaders#TIMESTAMP @@ -142,8 +141,7 @@ public final class MessageBuilder { } /** - * Copy the name-value pairs from the provided Map. This operation will - * not overwrite any existing values. + * Copy the name-value pairs from the provided Map. This operation will not overwrite any existing values. */ public MessageBuilder copyHeadersIfAbsent(Map headersToCopy) { Set keys = headersToCopy.keySet(); @@ -170,6 +168,50 @@ public final class MessageBuilder { return this.setHeader(MessageHeaders.CORRELATION_ID, correlationId); } + public MessageBuilder pushSequenceDetails(Object correlationId, int sequenceNumber, int sequenceSize) { + Object incomingCorrelationId = headers.get(MessageHeaders.CORRELATION_ID); + @SuppressWarnings("unchecked") + List> incomingSequenceDetails = (List>) headers.get(MessageHeaders.SEQUENCE_DETAILS); + if (incomingCorrelationId != null) { + if (incomingSequenceDetails == null) { + incomingSequenceDetails = new ArrayList>(); + } + else { + incomingSequenceDetails = new ArrayList>(incomingSequenceDetails); + } + incomingSequenceDetails.add(Arrays.asList(incomingCorrelationId, headers + .get(MessageHeaders.SEQUENCE_NUMBER), headers.get(MessageHeaders.SEQUENCE_SIZE))); + incomingSequenceDetails = Collections.unmodifiableList(incomingSequenceDetails); + } + if (incomingSequenceDetails != null) { + setHeader(MessageHeaders.SEQUENCE_DETAILS, incomingSequenceDetails); + } + return setCorrelationId(correlationId).setSequenceNumber(sequenceNumber).setSequenceSize(sequenceSize); + } + + public MessageBuilder popSequenceDetails() { + String key = MessageHeaders.SEQUENCE_DETAILS; + if (!headers.containsKey(key)) { + return this; + } + @SuppressWarnings("unchecked") + List> incomingSequenceDetails = new ArrayList>((List>) headers.get(key)); + List sequenceDetails = incomingSequenceDetails.remove(incomingSequenceDetails.size() - 1); + Assert.state(sequenceDetails.size() == 3, "Wrong sequence details (not created by MessageBuilder?): " + + sequenceDetails); + setCorrelationId(sequenceDetails.get(0)); + Integer sequenceNumber = (Integer) sequenceDetails.get(1); + Integer sequenceSize = (Integer) sequenceDetails.get(2); + if (sequenceSize > 0) { + setSequenceNumber(sequenceNumber); + setSequenceSize(sequenceSize); + } + if (!incomingSequenceDetails.isEmpty()) { + headers.put(MessageHeaders.SEQUENCE_DETAILS, incomingSequenceDetails); + } + return this; + } + public MessageBuilder setReplyChannel(MessageChannel replyChannel) { return this.setHeader(MessageHeaders.REPLY_CHANNEL, replyChannel); } @@ -212,27 +254,24 @@ public final class MessageBuilder { private void verifyType(String headerName, Object headerValue) { if (headerName != null && headerValue != null) { if (MessageHeaders.ID.equals(headerName)) { - Assert.isTrue(headerValue instanceof UUID, - "The '" + headerName + "' header value must be a UUID."); + Assert.isTrue(headerValue instanceof UUID, "The '" + headerName + "' header value must be a UUID."); } else if (MessageHeaders.TIMESTAMP.equals(headerName)) { - Assert.isTrue(headerValue instanceof Long, - "The '" + headerName + "' header value must be a Long."); + Assert.isTrue(headerValue instanceof Long, "The '" + headerName + "' header value must be a Long."); } else if (MessageHeaders.EXPIRATION_DATE.equals(headerName)) { - Assert.isTrue(headerValue instanceof Date || headerValue instanceof Long, - "The '" + headerName + "' header value must be a Date or Long."); + Assert.isTrue(headerValue instanceof Date || headerValue instanceof Long, "The '" + headerName + + "' header value must be a Date or Long."); } - else if (MessageHeaders.ERROR_CHANNEL.equals(headerName) || - MessageHeaders.REPLY_CHANNEL.endsWith(headerName)) { - Assert.isTrue(headerValue instanceof MessageChannel || - headerValue instanceof String, - "The '" + headerName + "' header value must be a MessageChannel or String."); + else if (MessageHeaders.ERROR_CHANNEL.equals(headerName) + || MessageHeaders.REPLY_CHANNEL.endsWith(headerName)) { + Assert.isTrue(headerValue instanceof MessageChannel || headerValue instanceof String, "The '" + + headerName + "' header value must be a MessageChannel or String."); } - else if (MessageHeaders.SEQUENCE_NUMBER.equals(headerName) || - MessageHeaders.SEQUENCE_SIZE.equals(headerName)) { - Assert.isTrue(Integer.class.isAssignableFrom(headerValue.getClass()), - "The '" + headerName + "' header value must be an Integer."); + else if (MessageHeaders.SEQUENCE_NUMBER.equals(headerName) + || MessageHeaders.SEQUENCE_SIZE.equals(headerName)) { + Assert.isTrue(Integer.class.isAssignableFrom(headerValue.getClass()), "The '" + headerName + + "' header value must be an Integer."); } } } diff --git a/spring-integration-core/src/test/java/org/springframework/integration/aggregator/scenarios/NestedAggregationTests-context.xml b/spring-integration-core/src/test/java/org/springframework/integration/aggregator/scenarios/NestedAggregationTests-context.xml index 3c5e7f1a56..815176bd4c 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/aggregator/scenarios/NestedAggregationTests-context.xml +++ b/spring-integration-core/src/test/java/org/springframework/integration/aggregator/scenarios/NestedAggregationTests-context.xml @@ -1,27 +1,30 @@ - - + - + - + - + - + + + + + + + + + + + + \ No newline at end of file diff --git a/spring-integration-core/src/test/java/org/springframework/integration/aggregator/scenarios/NestedAggregationTests.java b/spring-integration-core/src/test/java/org/springframework/integration/aggregator/scenarios/NestedAggregationTests.java index 56cd2ac0d7..c957e99ab5 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/aggregator/scenarios/NestedAggregationTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/aggregator/scenarios/NestedAggregationTests.java @@ -41,24 +41,36 @@ import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; public class NestedAggregationTests { @Autowired - DirectChannel input; + DirectChannel splitter; + + @Autowired + DirectChannel router; @Test public void testAggregatorWithNestedSplitter() throws Exception { - List result = sendAndReceiveMessage(input, 2000); + @SuppressWarnings("unchecked") + Message input = new GenericMessage>>(Arrays.asList(Arrays.asList("foo", "bar", "spam"), + Arrays.asList("bar", "foo"))); + List result = sendAndReceiveMessage(splitter, 2000, input); assertNotNull("Expected result and got null", result); assertEquals("[[foo, bar, spam], [bar, foo]]", result.toString()); } - private List sendAndReceiveMessage(DirectChannel channel, int timeout) { + @Test + public void testAggregatorWithNestedRouter() throws Exception { + Message input = new GenericMessage>(Arrays.asList("bar", "foo")); + List result = sendAndReceiveMessage(router, 2000, input); + assertNotNull("Expected result and got null", result); + assertEquals("[[bar, foo], [bar, foo]]", result.toString()); + } + + private List sendAndReceiveMessage(DirectChannel channel, int timeout, Message input) { MessagingTemplate messagingTemplate = new MessagingTemplate(); messagingTemplate.setReceiveTimeout(timeout); @SuppressWarnings("unchecked") - Message> message = (Message>) messagingTemplate.sendAndReceive(channel, - new GenericMessage>>(Arrays.asList(Arrays.asList("foo", "bar", "spam"), Arrays.asList("bar", - "foo")))); + Message> message = (Message>) messagingTemplate.sendAndReceive(channel, input); return message == null ? null : message.getPayload(); diff --git a/spring-integration-core/src/test/java/org/springframework/integration/endpoint/CorrelationIdTests.java b/spring-integration-core/src/test/java/org/springframework/integration/endpoint/CorrelationIdTests.java index dfdb46f4f7..f629e21ec8 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/endpoint/CorrelationIdTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/endpoint/CorrelationIdTests.java @@ -20,13 +20,12 @@ import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertTrue; import org.junit.Test; - import org.springframework.integration.Message; +import org.springframework.integration.MessageHeaders; import org.springframework.integration.channel.DirectChannel; import org.springframework.integration.channel.QueueChannel; import org.springframework.integration.handler.ServiceActivatingHandler; import org.springframework.integration.message.GenericMessage; -import org.springframework.integration.splitter.AbstractMessageSplitter; import org.springframework.integration.splitter.MethodInvokingSplitter; import org.springframework.integration.support.MessageBuilder; @@ -126,8 +125,8 @@ public class CorrelationIdTests { Message reply2 = testChannel.receive(100); assertEquals(message.getHeaders().getId(), reply1.getHeaders().getCorrelationId()); assertEquals(message.getHeaders().getId(), reply2.getHeaders().getCorrelationId()); - assertTrue("Sequence details missing", reply1.getHeaders().containsKey(AbstractMessageSplitter.SEQUENCE_DETAILS)); - assertTrue("Sequence details missing", reply2.getHeaders().containsKey(AbstractMessageSplitter.SEQUENCE_DETAILS)); + assertTrue("Sequence details missing", reply1.getHeaders().containsKey(MessageHeaders.SEQUENCE_DETAILS)); + assertTrue("Sequence details missing", reply2.getHeaders().containsKey(MessageHeaders.SEQUENCE_DETAILS)); } @SuppressWarnings("unused") diff --git a/spring-integration-jdbc/src/test/java/log4j.properties b/spring-integration-jdbc/src/test/java/log4j.properties index 1dcc129804..93d810c941 100644 --- a/spring-integration-jdbc/src/test/java/log4j.properties +++ b/spring-integration-jdbc/src/test/java/log4j.properties @@ -6,6 +6,6 @@ log4j.appender.stdout.layout.ConversionPattern=%d{ABSOLUTE} %5p %t %c{2}:%L - %m log4j.category.org.springframework=WARN -log4j.category.org.springframework.integration=DEBUG -log4j.category.org.springframework.integration.jdbc=DEBUG +# log4j.category.org.springframework.integration=DEBUG +# log4j.category.org.springframework.integration.jdbc=DEBUG log4j.category.org.springframework.jdbc=DEBUG