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

INT-1437: push sequence details into message builder

Browse Source 
INT-1420: add javadocs
This commit is contained in:
Dave Syer
2010-09-14 07:43:51 +01:00
parent 965b51d2a6 9d22d8abce
commit 6233647a54
33 changed files with 588 additions and 354 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
spring-integration-core/src/main/java/org/springframework/integration/MessageHeaders.java
View File

@@ -70,6 +70,8 @@ public final class MessageHeaders implements Map<String, Object>, Serializable {
public static final String SEQUENCE_SIZE = PREFIX + "sequenceSize";
public static final String SEQUENCE_DETAILS = PREFIX + "sequenceDetails";
private final Map<String, Object> headers;

33
spring-integration-core/src/main/java/org/springframework/integration/aggregator/AbstractAggregatingMessageGroupProcessor.java
View File

@@ -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<String, Object> 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<Object[]> incomingSequenceDetails = new ArrayList<Object[]>(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);

53
spring-integration-core/src/main/java/org/springframework/integration/dispatcher/BroadcastingDispatcher.java
View File

@@ -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 <code>false</code> (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 <code>false</code>.
* A broadcasting dispatcher implementation. If the 'ignoreFailures' property is set to <code>false</code> (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 <code>false</code>.
* <p>
* If the 'ignoreFailures' flag is set to <code>true</code> 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 <code>true</code>, 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 <code>true</code> 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 <code>true</code>, 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 <code>false</code> meaning that an
* Exception will be thrown when a handler fails. To override this and
* suppress Exceptions, set the value to <code>true</code>.
* Specify whether failures for one or more of the handlers should be ignored. By default this is <code>false</code>
* meaning that an Exception will be thrown when a handler fails. To override this and suppress Exceptions, set the
* value to <code>true</code>.
* <p>
* 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 <em>not</em> be applied
*/
public void setApplySequence(boolean applySequence) {
@@ -93,13 +82,8 @@ public class BroadcastingDispatcher extends AbstractDispatcher {
List<MessageHandler> 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;

4
spring-integration-core/src/main/java/org/springframework/integration/handler/AbstractReplyProducingMessageHandler.java
View File

@@ -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;
}

48
spring-integration-core/src/main/java/org/springframework/integration/router/AbstractMessageRouter.java
View File

@@ -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 <code>false</code> 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
* <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) {
@@ -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 <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>.
* 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;
@@ -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<MessageChannel> determineTargetChannels(Message<?> message);

101
spring-integration-core/src/main/java/org/springframework/integration/splitter/AbstractMessageSplitter.java
View File

@@ -16,6 +16,10 @@
package org.springframework.integration.splitter;
import java.util.ArrayList;
import java.util.Collection;
import java.util.List;
import org.springframework.integration.Message;
import org.springframework.integration.MessageHeaders;
import org.springframework.integration.handler.AbstractReplyProducingMessageHandler;
@@ -23,93 +27,60 @@ import org.springframework.integration.support.MessageBuilder;
import org.springframework.util.CollectionUtils;
import org.springframework.util.ObjectUtils;
import java.util.*;
/**
* Base class for Message-splitting handlers.
*
*
* @author Mark Fisher
* @author Dave Syer
* @author Iwein Fuld
*/
public abstract class AbstractMessageSplitter extends AbstractReplyProducingMessageHandler {
public static final String SEQUENCE_DETAILS = MessageHeaders.PREFIX + "sequenceDetails";
@Override
@SuppressWarnings("unchecked")
protected final Object handleRequestMessage(Message<?> message) {
Object result = this.splitMessage(message);
// return null if 'null', empty Collection or empty Array
if ( result == null ||
(result instanceof Collection && CollectionUtils.isEmpty((Collection<?>)result)) ||
(result.getClass().isArray() && ObjectUtils.isEmpty((Object[]) result)) ) {
if (result == null || (result instanceof Collection && CollectionUtils.isEmpty((Collection) result))
|| (result.getClass().isArray() && ObjectUtils.isEmpty((Object[]) result))) {
return null;
}
MessageHeaders headers = message.getHeaders();
Object incomingCorrelationId = headers.getCorrelationId();
List<Object[]> incomingSequenceDetails = extractSequenceDetails(headers, incomingCorrelationId);
Object correlationId = headers.getId();
List<MessageBuilder> messageBuilders;
List<MessageBuilder<?>> messageBuilders = new ArrayList<MessageBuilder<?>>();
if (result instanceof Collection) {
messageBuilders = messageBuildersForCollection(result, incomingSequenceDetails, correlationId);
} else if (result.getClass().isArray()) {
messageBuilders = messageBuildersForArray(result, incomingSequenceDetails, correlationId);
} else {
messageBuilders = Collections.singletonList(this.createBuilder(result, incomingSequenceDetails, correlationId, 1, 1));
}
return messageBuilders;
}
private List<MessageBuilder> messageBuildersForArray(Object result, List<Object[]> incomingSequenceDetails, Object correlationId) {
List<MessageBuilder> messageBuilders = new ArrayList<MessageBuilder>();
Object[] items = (Object[]) result;
int sequenceNumber = 0;
int sequenceSize = items.length;
for (Object item : items) {
messageBuilders.add(this.createBuilder(
item, incomingSequenceDetails, correlationId, ++sequenceNumber, sequenceSize));
}
return messageBuilders;
}
private List<MessageBuilder> messageBuildersForCollection(Object result, List<Object[]> incomingSequenceDetails, Object correlationId) {
List<MessageBuilder> messageBuilders = new ArrayList<MessageBuilder>();
Collection<?> items = (Collection<?>) result;
int sequenceNumber = 0;
int sequenceSize = items.size();
for (Object item : items) {
messageBuilders.add(this.createBuilder(
item, incomingSequenceDetails, correlationId, ++sequenceNumber, sequenceSize));
}
return messageBuilders;
}
private List<Object[]> extractSequenceDetails(MessageHeaders headers, Object incomingCorrelationId) {
List<Object[]> incomingSequenceDetails = headers.get(SEQUENCE_DETAILS, List.class);
if (incomingCorrelationId != null) {
if (incomingSequenceDetails == null) {
incomingSequenceDetails = new ArrayList<Object[]>();
} else {
incomingSequenceDetails = new ArrayList<Object[]>(incomingSequenceDetails);
Collection<?> items = (Collection<?>) result;
int sequenceNumber = 0;
int sequenceSize = items.size();
for (Object item : items) {
messageBuilders.add(this.createBuilder(item, headers, correlationId, ++sequenceNumber, sequenceSize));
}
incomingSequenceDetails.add(new Object[]{
incomingCorrelationId, headers.getSequenceNumber(), headers.getSequenceSize()});
incomingSequenceDetails = Collections.unmodifiableList(incomingSequenceDetails);
}
return incomingSequenceDetails;
else if (result.getClass().isArray()) {
Object[] items = (Object[]) result;
int sequenceNumber = 0;
int sequenceSize = items.length;
for (Object item : items) {
messageBuilders.add(this.createBuilder(item, headers, correlationId, ++sequenceNumber, sequenceSize));
}
}
else {
messageBuilders.add(this.createBuilder(result, headers, correlationId, 1, 1));
}
return messageBuilders;
}
@SuppressWarnings({"unchecked", "rawtypes"})
private MessageBuilder createBuilder(Object item, List<Object[]> 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;
}

109
spring-integration-core/src/main/java/org/springframework/integration/support/MessageBuilder.java
View File

@@ -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<T> {
@@ -45,7 +50,6 @@ public final class MessageBuilder<T> {
private volatile boolean modified;
/**
* Private constructor to be invoked from the static factory methods only.
*/
@@ -58,14 +62,11 @@ public final class MessageBuilder<T> {
}
}
/**
* 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 <T> MessageBuilder<T> fromMessage(Message<T> message) {
Assert.notNull(message, "message must not be null");
@@ -83,13 +84,12 @@ public final class MessageBuilder<T> {
return builder;
}
/**
* Set the value for the given header name. If the provided value is
* <code>null</code>, the header will be removed.
* Set the value for the given header name. If the provided value is <code>null</code>, the header will be removed.
*/
public MessageBuilder<T> 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<T> {
}
/**
* 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<T> setHeaderIfAbsent(String headerName, Object headerValue) {
if (this.headers.get(headerName) == null) {
@@ -117,7 +116,8 @@ public final class MessageBuilder<T> {
* Remove the value for the given header name.
*/
public MessageBuilder<T> 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<T> {
}
/**
* 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<T> {
}
/**
* Copy the name-value pairs from the provided Map. This operation will
* <em>not</em> overwrite any existing values.
* Copy the name-value pairs from the provided Map. This operation will <em>not</em> overwrite any existing values.
*/
public MessageBuilder<T> copyHeadersIfAbsent(Map<String, Object> headersToCopy) {
Set<String> keys = headersToCopy.keySet();
@@ -170,6 +168,50 @@ public final class MessageBuilder<T> {
return this.setHeader(MessageHeaders.CORRELATION_ID, correlationId);
}
public MessageBuilder<T> pushSequenceDetails(Object correlationId, int sequenceNumber, int sequenceSize) {
Object incomingCorrelationId = headers.get(MessageHeaders.CORRELATION_ID);
@SuppressWarnings("unchecked")
List<List<Object>> incomingSequenceDetails = (List<List<Object>>) headers.get(MessageHeaders.SEQUENCE_DETAILS);
if (incomingCorrelationId != null) {
if (incomingSequenceDetails == null) {
incomingSequenceDetails = new ArrayList<List<Object>>();
}
else {
incomingSequenceDetails = new ArrayList<List<Object>>(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<T> popSequenceDetails() {
String key = MessageHeaders.SEQUENCE_DETAILS;
if (!headers.containsKey(key)) {
return this;
}
@SuppressWarnings("unchecked")
List<List<Object>> incomingSequenceDetails = new ArrayList<List<Object>>((List<List<Object>>) headers.get(key));
List<Object> 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<T> setReplyChannel(MessageChannel replyChannel) {
return this.setHeader(MessageHeaders.REPLY_CHANNEL, replyChannel);
}
@@ -212,27 +254,24 @@ public final class MessageBuilder<T> {
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.");
}
}
}

35
spring-integration-core/src/test/java/org/springframework/integration/aggregator/scenarios/NestedAggregationTests-context.xml
View File

@@ -1,27 +1,30 @@
<?xml version="1.0" encoding="UTF-8"?>
<beans:beans xmlns:beans="http://www.springframework.org/schema/beans"
xmlns="http://www.springframework.org/schema/integration"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://www.springframework.org/schema/beans
<beans:beans xmlns:beans="http://www.springframework.org/schema/beans" xmlns="http://www.springframework.org/schema/integration"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://www.springframework.org/schema/beans
http://www.springframework.org/schema/beans/spring-beans-2.5.xsd
http://www.springframework.org/schema/integration
http://www.springframework.org/schema/integration/spring-integration-1.0.xsd">
<channel id="input"/>
<channel id="splitter" />
<splitter id="upstream-splitter"
input-channel="input"
output-channel="upstream-splits"/>
<splitter id="upstream-splitter" input-channel="splitter" output-channel="upstream-splits" />
<splitter id="downstream-splitter"
input-channel="upstream-splits"
output-channel="downstream-splits"/>
<splitter id="downstream-splitter" input-channel="upstream-splits" output-channel="downstream-splits" />
<aggregator id="first-aggregator"
timeout="1000"
input-channel="downstream-splits" output-channel="pre-output"/>
<aggregator id="first-aggregator" timeout="1000" input-channel="downstream-splits" output-channel="pre-output" />
<aggregator id="second-aggregator"
input-channel="pre-output"/>
<aggregator id="second-aggregator" input-channel="pre-output" />
<channel id="router" />
<recipient-list-router id="upstream-router" input-channel="router" apply-sequence="true">
<recipient channel="upstream-splits" />
<recipient channel="upstream-another" />
</recipient-list-router>
<splitter id="downstream-another" input-channel="upstream-another" output-channel="downstream-splits" />
</beans:beans>

24
spring-integration-core/src/test/java/org/springframework/integration/aggregator/scenarios/NestedAggregationTests.java
View File

@@ -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<String> result = sendAndReceiveMessage(input, 2000);
@SuppressWarnings("unchecked")
Message<?> input = new GenericMessage<List<List<String>>>(Arrays.asList(Arrays.asList("foo", "bar", "spam"),
Arrays.asList("bar", "foo")));
List<String> result = sendAndReceiveMessage(splitter, 2000, input);
assertNotNull("Expected result and got null", result);
assertEquals("[[foo, bar, spam], [bar, foo]]", result.toString());
}
private List<String> sendAndReceiveMessage(DirectChannel channel, int timeout) {
@Test
public void testAggregatorWithNestedRouter() throws Exception {
Message<?> input = new GenericMessage<List<String>>(Arrays.asList("bar", "foo"));
List<String> result = sendAndReceiveMessage(router, 2000, input);
assertNotNull("Expected result and got null", result);
assertEquals("[[bar, foo], [bar, foo]]", result.toString());
}
private List<String> sendAndReceiveMessage(DirectChannel channel, int timeout, Message<?> input) {
MessagingTemplate messagingTemplate = new MessagingTemplate();
messagingTemplate.setReceiveTimeout(timeout);
@SuppressWarnings("unchecked")
Message<List<String>> message = (Message<List<String>>) messagingTemplate.sendAndReceive(channel,
new GenericMessage<List<List<String>>>(Arrays.asList(Arrays.asList("foo", "bar", "spam"), Arrays.asList("bar",
"foo"))));
Message<List<String>> message = (Message<List<String>>) messagingTemplate.sendAndReceive(channel, input);
return message == null ? null : message.getPayload();

7
spring-integration-core/src/test/java/org/springframework/integration/endpoint/CorrelationIdTests.java
View File

@@ -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")