Partial fix for INT-24 (add resequencer). A namespace element needs also to be created.
This commit is contained in:
Marius Bogoevici
@@ -0,0 +1,104 @@
|
||||
/*
|
||||
* Copyright 2002-2008 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.
|
||||
* You may obtain a copy of the License at
|
||||
*
|
||||
* http://www.apache.org/licenses/LICENSE-2.0
|
||||
*
|
||||
* Unless required by applicable law or agreed to in writing, software
|
||||
* distributed under the License is distributed on an "AS IS" BASIS,
|
||||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
* See the License for the specific language governing permissions and
|
||||
* limitations under the License.
|
||||
*/
|
||||
package org.springframework.integration.router;
|
||||
|
||||
import java.util.ArrayList;
|
||||
import java.util.Collections;
|
||||
import java.util.List;
|
||||
import java.util.concurrent.locks.ReentrantLock;
|
||||
|
||||
import org.apache.commons.logging.Log;
|
||||
import org.apache.commons.logging.LogFactory;
|
||||
import org.springframework.integration.message.Message;
|
||||
|
||||
/**
|
||||
* Default implementation for a {@link MessageBarrier}.
|
||||
*
|
||||
* @author Marius Bogoevici
|
||||
*/
|
||||
public abstract class AbstractMessageBarrier implements MessageBarrier {
|
||||
|
||||
private final Log logger = LogFactory.getLog(this.getClass());
|
||||
|
||||
protected final List<Message<?>> messages = new ArrayList<Message<?>>();
|
||||
|
||||
private volatile boolean complete = false;
|
||||
|
||||
private final ReentrantLock lock = new ReentrantLock();
|
||||
|
||||
private final long timestamp = System.currentTimeMillis();
|
||||
|
||||
|
||||
/**
|
||||
* Returns the creation time of this barrier as the number of milliseconds
|
||||
* since January 1, 1970.
|
||||
* @see System#currentTimeMillis()
|
||||
*/
|
||||
public long getTimestamp() {
|
||||
return this.timestamp;
|
||||
}
|
||||
|
||||
protected boolean isComplete(){
|
||||
return this.complete;
|
||||
}
|
||||
|
||||
/**
|
||||
* Adds a message to the aggregation group and releases <em>if available</em>.
|
||||
* Otherwise, the return value will be <code>null</code>.
|
||||
*/
|
||||
public List<Message<?>> addAndRelease(Message<?> message) {
|
||||
try {
|
||||
this.lock.lock();
|
||||
if (this.complete) {
|
||||
if (logger.isDebugEnabled()) {
|
||||
logger.debug("Message received after aggregation has already completed: " + message);
|
||||
}
|
||||
return null;
|
||||
}
|
||||
addMessage(message);
|
||||
this.complete = hasReceivedAllMessages();
|
||||
return releaseAvailableMessages();
|
||||
}
|
||||
finally {
|
||||
this.lock.unlock();
|
||||
}
|
||||
}
|
||||
|
||||
protected void addMessage(Message<?> message) {
|
||||
this.messages.add(message);
|
||||
}
|
||||
|
||||
public List<Message<?>> getMessages() {
|
||||
return Collections.unmodifiableList(this.messages);
|
||||
}
|
||||
|
||||
/**
|
||||
* Subclasses will implement this method to indicate if all possible messages that could be received by
|
||||
* a given barrier have already been received (e.g. all messages from a given sequence)
|
||||
*/
|
||||
protected abstract boolean hasReceivedAllMessages();
|
||||
|
||||
|
||||
/**
|
||||
* Subclasses will implement this method to return the messages that can be released by this barrier, after
|
||||
* the receipt of a given message. It might be possible that a number of messages are released before the barrier
|
||||
* has ended its work (partial release) and this depends completely on the implementation of the barrier.
|
||||
* However, once hasReceivedAllMessages() is deemed true, only one call to releaseAvailableMessages() shall
|
||||
* yield results.
|
||||
*/
|
||||
protected abstract List<Message<?>> releaseAvailableMessages();
|
||||
|
||||
}
|
||||
@@ -0,0 +1,293 @@
|
||||
/*
|
||||
* Copyright 2002-2008 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.
|
||||
* You may obtain a copy of the License at
|
||||
*
|
||||
* http://www.apache.org/licenses/LICENSE-2.0
|
||||
*
|
||||
* Unless required by applicable law or agreed to in writing, software
|
||||
* distributed under the License is distributed on an "AS IS" BASIS,
|
||||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
* See the License for the specific language governing permissions and
|
||||
* limitations under the License.
|
||||
*/
|
||||
package org.springframework.integration.router;
|
||||
|
||||
import java.util.List;
|
||||
import java.util.Map;
|
||||
import java.util.concurrent.ArrayBlockingQueue;
|
||||
import java.util.concurrent.BlockingQueue;
|
||||
import java.util.concurrent.ConcurrentHashMap;
|
||||
import java.util.concurrent.ConcurrentMap;
|
||||
import java.util.concurrent.Executors;
|
||||
import java.util.concurrent.ScheduledExecutorService;
|
||||
import java.util.concurrent.TimeUnit;
|
||||
|
||||
import org.apache.commons.logging.Log;
|
||||
import org.apache.commons.logging.LogFactory;
|
||||
import org.springframework.beans.factory.InitializingBean;
|
||||
import org.springframework.integration.channel.MessageChannel;
|
||||
import org.springframework.integration.handler.MessageHandler;
|
||||
import org.springframework.integration.message.Message;
|
||||
import org.springframework.integration.message.MessageHandlingException;
|
||||
import org.springframework.util.Assert;
|
||||
import org.springframework.util.CollectionUtils;
|
||||
|
||||
/**
|
||||
* Base class for {@link MessageBarrier}-based MessageHandlers.
|
||||
* A {@link MessageHandler} implementation that waits for a group of
|
||||
* {@link Message Messages} to arrive and process them together.
|
||||
* Uses a {@link MessageBarrier} to store messages and to decide on how
|
||||
* the messages can be released.
|
||||
* <p>
|
||||
* Each {@link Message} that is received by this handler will be associated with
|
||||
* a group based upon the '<code>correlationId</code>' property of its
|
||||
* header. If no such property is available, a {@link MessageHandlingException}
|
||||
* will be thrown.
|
||||
* <p>
|
||||
* The '<code>timeout</code>' value determines how long to wait for the
|
||||
* complete group after the arrival of the first {@link Message} of the group.
|
||||
* The default value is 1 minute. If the timeout elapses prior to completion,
|
||||
* then Messages with that timed-out 'correlationId' will be sent to the
|
||||
* 'discardChannel' if provided.
|
||||
*
|
||||
* @author Mark Fisher
|
||||
* @author Marius Bogoevici
|
||||
*/
|
||||
public abstract class AbstractMessageBarrierHandler implements MessageHandler, InitializingBean {
|
||||
|
||||
public final static long DEFAULT_SEND_TIMEOUT = 1000;
|
||||
|
||||
public final static long DEFAULT_TIMEOUT = 60000;
|
||||
|
||||
public final static long DEFAULT_REAPER_INTERVAL = 1000;
|
||||
|
||||
public final static int DEFAULT_TRACKED_CORRRELATION_ID_CAPACITY = 1000;
|
||||
|
||||
protected final Log logger = LogFactory.getLog(this.getClass());
|
||||
|
||||
protected volatile MessageChannel defaultReplyChannel;
|
||||
|
||||
private volatile MessageChannel discardChannel;
|
||||
|
||||
protected volatile long sendTimeout = DEFAULT_SEND_TIMEOUT;
|
||||
|
||||
protected final ConcurrentMap<Object, MessageBarrier> barriers
|
||||
= new ConcurrentHashMap<Object, MessageBarrier>();
|
||||
|
||||
private volatile long timeout = DEFAULT_TIMEOUT;
|
||||
|
||||
private volatile boolean sendPartialResultOnTimeout = false;
|
||||
|
||||
private volatile long reaperInterval = DEFAULT_REAPER_INTERVAL;
|
||||
|
||||
private volatile int trackedCorrelationIdCapacity = DEFAULT_TRACKED_CORRRELATION_ID_CAPACITY;
|
||||
|
||||
protected volatile BlockingQueue<Object> trackedCorrelationIds;
|
||||
|
||||
protected final ScheduledExecutorService executor;
|
||||
|
||||
private volatile boolean initialized;
|
||||
|
||||
|
||||
public AbstractMessageBarrierHandler(ScheduledExecutorService executor) {
|
||||
this.executor = (executor != null) ? executor : Executors.newSingleThreadScheduledExecutor();
|
||||
}
|
||||
|
||||
|
||||
/**
|
||||
* Set the default channel for sending aggregated Messages. Note that
|
||||
* precedence will be given to the 'returnAddress' of the aggregated
|
||||
* message itself, then to the 'returnAddress' of the original message.
|
||||
*/
|
||||
public void setDefaultReplyChannel(MessageChannel defaultReplyChannel) {
|
||||
this.defaultReplyChannel = defaultReplyChannel;
|
||||
}
|
||||
|
||||
/**
|
||||
* Specify a channel for sending Messages that arrive after their aggregation
|
||||
* group has either completed or timed-out.
|
||||
*/
|
||||
public void setDiscardChannel(MessageChannel discardChannel) {
|
||||
this.discardChannel = discardChannel;
|
||||
}
|
||||
|
||||
/**
|
||||
* Set the timeout for sending aggregation results and discarded Messages.
|
||||
*/
|
||||
public void setSendTimeout(long sendTimeout) {
|
||||
this.sendTimeout = sendTimeout;
|
||||
}
|
||||
|
||||
/**
|
||||
* Specify whether to aggregate and send the resulting Message when the
|
||||
* timeout elapses prior to the CompletionStrategy.
|
||||
*/
|
||||
public void setSendPartialResultOnTimeout(boolean sendPartialResultOnTimeout) {
|
||||
this.sendPartialResultOnTimeout = sendPartialResultOnTimeout;
|
||||
}
|
||||
|
||||
/**
|
||||
* Set the interval in milliseconds for the reaper thread. Default is 1000.
|
||||
*/
|
||||
public void setReaperInterval(long reaperInterval) {
|
||||
Assert.isTrue(reaperInterval > 0, "'reaperInterval' must be a positive value");
|
||||
this.reaperInterval = reaperInterval;
|
||||
}
|
||||
|
||||
/**
|
||||
* Set the number of completed correlationIds to track. Default is 1000.
|
||||
*/
|
||||
public void setTrackedCorrelationIdCapacity(int trackedCorrelationIdCapacity) {
|
||||
Assert.isTrue(trackedCorrelationIdCapacity > 0, "'trackedCorrelationIdCapacity' must be a positive value");
|
||||
this.trackedCorrelationIdCapacity = trackedCorrelationIdCapacity;
|
||||
}
|
||||
|
||||
/**
|
||||
* Initialize this handler.
|
||||
*/
|
||||
public void afterPropertiesSet() {
|
||||
this.trackedCorrelationIds = new ArrayBlockingQueue<Object>(this.trackedCorrelationIdCapacity);
|
||||
this.executor.scheduleWithFixedDelay(new ReaperTask(),
|
||||
this.reaperInterval, this.reaperInterval, TimeUnit.MILLISECONDS);
|
||||
this.initialized = true;
|
||||
}
|
||||
|
||||
/**
|
||||
* Maximum time to wait (in milliseconds) for the completion strategy to
|
||||
* become true. The default is 60000 (1 minute).
|
||||
*/
|
||||
public void setTimeout(long timeout) {
|
||||
Assert.isTrue(timeout >= 0, "'timeout' must not be negative");
|
||||
this.timeout = timeout;
|
||||
}
|
||||
|
||||
public Message<?> handle(Message<?> message) {
|
||||
if (!this.initialized) {
|
||||
this.afterPropertiesSet();
|
||||
}
|
||||
Object correlationId = message.getHeader().getCorrelationId();
|
||||
if (correlationId == null) {
|
||||
throw new MessageHandlingException(message,
|
||||
this.getClass().getSimpleName() + " requires the 'correlationId' property");
|
||||
}
|
||||
if (this.trackedCorrelationIds.contains(correlationId)) {
|
||||
if (logger.isDebugEnabled()) {
|
||||
logger.debug("Handling for correlationId '" + correlationId +
|
||||
"' has already completed or timed out.");
|
||||
}
|
||||
this.sendToDiscardChannelIfAvailable(message);
|
||||
return null;
|
||||
}
|
||||
MessageBarrier barrier = barriers.putIfAbsent(correlationId, createMessageBarrier());
|
||||
if (barrier == null) {
|
||||
barrier = barriers.get(correlationId);
|
||||
}
|
||||
List<Message<?>> releasedMessages = barrier.addAndRelease(message);
|
||||
if (CollectionUtils.isEmpty(releasedMessages)) {
|
||||
return null;
|
||||
}
|
||||
if (isBarrierRemovable(correlationId, releasedMessages)) {
|
||||
this.removeBarrier(correlationId);
|
||||
}
|
||||
afterRelease(correlationId, releasedMessages);
|
||||
return null;
|
||||
}
|
||||
|
||||
private void afterRelease(Object correlationId, List<Message<?>> releasedMessages) {
|
||||
Message<?>[] processedMessages = this.processReleasedMessages(correlationId, releasedMessages);
|
||||
for (Message<?> result : processedMessages) {
|
||||
MessageChannel replyChannel = this.resolveReplyChannelFromMessage(result);
|
||||
if (replyChannel == null) {
|
||||
replyChannel = this.resolveReplyChannelFromMessage(releasedMessages.get(0));
|
||||
if (replyChannel == null) {
|
||||
replyChannel = this.defaultReplyChannel;
|
||||
}
|
||||
}
|
||||
if (replyChannel != null) {
|
||||
replyChannel.send(result, this.sendTimeout);
|
||||
}
|
||||
else if (logger.isWarnEnabled()) {
|
||||
logger.warn("unable to determine reply channel for aggregation result: " + result);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
private void sendToDiscardChannelIfAvailable(Message<?> message) {
|
||||
if (this.discardChannel != null) {
|
||||
if (!this.discardChannel.send(message, this.sendTimeout)) {
|
||||
if (logger.isWarnEnabled()) {
|
||||
logger.warn("unable to send to 'discardChannel', message: " + message);
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
protected MessageChannel resolveReplyChannelFromMessage(Message<?> message) {
|
||||
Object returnAddress = message.getHeader().getReturnAddress();
|
||||
if (returnAddress != null) {
|
||||
if (returnAddress instanceof MessageChannel) {
|
||||
return (MessageChannel) returnAddress;
|
||||
}
|
||||
if (logger.isWarnEnabled()) {
|
||||
logger.warn("Aggregator can only reply to a 'returnAddress' of type MessageChannel.");
|
||||
}
|
||||
}
|
||||
return null;
|
||||
}
|
||||
|
||||
private void removeBarrier(Object correlationId) {
|
||||
if (this.barriers.remove(correlationId) != null) {
|
||||
synchronized (this.trackedCorrelationIds) {
|
||||
boolean added = this.trackedCorrelationIds.offer(correlationId);
|
||||
if (!added) {
|
||||
this.trackedCorrelationIds.poll();
|
||||
this.trackedCorrelationIds.offer(correlationId);
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
private class ReaperTask implements Runnable {
|
||||
|
||||
public void run() {
|
||||
long currentTime = System.currentTimeMillis();
|
||||
for (Map.Entry<Object, MessageBarrier> entry : barriers.entrySet()) {
|
||||
if (currentTime - entry.getValue().getTimestamp() >= timeout) {
|
||||
Object correlationId = entry.getKey();
|
||||
List<Message<?>> messages = entry.getValue().getMessages();
|
||||
removeBarrier(correlationId);
|
||||
if (sendPartialResultOnTimeout) {
|
||||
afterRelease(correlationId, messages);
|
||||
}
|
||||
else {
|
||||
for (Message<?> message : messages) {
|
||||
sendToDiscardChannelIfAvailable(message);
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Factory method for creating a suitable MessageBarrier
|
||||
* implementation.
|
||||
*/
|
||||
protected abstract MessageBarrier createMessageBarrier();
|
||||
|
||||
/**
|
||||
* Implements the logic for deciding whether, based on what the
|
||||
* MessageBarrier has released so far, work for the correlationId
|
||||
* can be considered done and the barrier can be released.
|
||||
*/
|
||||
protected abstract boolean isBarrierRemovable(Object correlationId, List<Message<?>> releasedMessages);
|
||||
|
||||
/**
|
||||
* Implements the logic for transforming the released Messages.
|
||||
*/
|
||||
protected abstract Message<?>[] processReleasedMessages(Object correlationId, List<Message<?>> messages);
|
||||
|
||||
}
|
||||
@@ -17,88 +17,34 @@
|
||||
package org.springframework.integration.router;
|
||||
|
||||
import java.util.List;
|
||||
import java.util.Map;
|
||||
import java.util.concurrent.ArrayBlockingQueue;
|
||||
import java.util.concurrent.BlockingQueue;
|
||||
import java.util.concurrent.ConcurrentHashMap;
|
||||
import java.util.concurrent.ConcurrentMap;
|
||||
import java.util.concurrent.Executors;
|
||||
import java.util.concurrent.ScheduledExecutorService;
|
||||
import java.util.concurrent.TimeUnit;
|
||||
|
||||
import org.apache.commons.logging.Log;
|
||||
import org.apache.commons.logging.LogFactory;
|
||||
|
||||
import org.springframework.beans.factory.InitializingBean;
|
||||
import org.springframework.integration.channel.MessageChannel;
|
||||
import org.springframework.integration.handler.MessageHandler;
|
||||
import org.springframework.integration.message.Message;
|
||||
import org.springframework.integration.message.MessageHandlingException;
|
||||
import org.springframework.util.Assert;
|
||||
import org.springframework.util.CollectionUtils;
|
||||
|
||||
/**
|
||||
* A {@link MessageHandler} implementation that waits for a <em>complete</em>
|
||||
* An {@link AbstractMessageBarrierHandler} that waits for a <em>complete</em>
|
||||
* group of {@link Message Messages} to arrive and then delegates to an
|
||||
* {@link Aggregator} to combine them into a single {@link Message}.
|
||||
* <p>
|
||||
* Each {@link Message} that is received by this handler will be associated with
|
||||
* a group based upon the '<code>correlationId</code>' property of its
|
||||
* header. If no such property is available, a {@link MessageHandlingException}
|
||||
* will be thrown.
|
||||
* <p>
|
||||
* The default strategy for determining whether a group is complete is based on
|
||||
* the '<code>sequenceSize</code>' property of the header. Alternatively, a
|
||||
* custom implementation of the {@link CompletionStrategy} may be provided.
|
||||
* <p>
|
||||
* The '<code>timeout</code>' value determines how long to wait for the
|
||||
* complete group after the arrival of the first {@link Message} of the group.
|
||||
* The default value is 1 minute. If the timeout elapses prior to completion,
|
||||
* then Messages with that timed-out 'correlationId' will be sent to the
|
||||
* 'discardChannel' if provided.
|
||||
* All considerations regarding <code>timeout</code> and grouping by
|
||||
* '<code>correlationId</code>' from {@link AbstractMessageBarrierHandler} apply
|
||||
* here as well.
|
||||
*
|
||||
* @author Mark Fisher
|
||||
* @author Marius Bogoevici
|
||||
*/
|
||||
public class AggregatingMessageHandler implements MessageHandler, InitializingBean {
|
||||
|
||||
public final static long DEFAULT_SEND_TIMEOUT = 1000;
|
||||
|
||||
public final static long DEFAULT_TIMEOUT = 60000;
|
||||
|
||||
public final static long DEFAULT_REAPER_INTERVAL = 1000;
|
||||
|
||||
public final static int DEFAULT_TRACKED_CORRRELATION_ID_CAPACITY = 1000;
|
||||
|
||||
|
||||
private final Log logger = LogFactory.getLog(this.getClass());
|
||||
public class AggregatingMessageHandler extends AbstractMessageBarrierHandler {
|
||||
|
||||
private final Aggregator aggregator;
|
||||
|
||||
private volatile MessageChannel defaultReplyChannel;
|
||||
|
||||
private volatile MessageChannel discardChannel;
|
||||
|
||||
private volatile long sendTimeout = DEFAULT_SEND_TIMEOUT;
|
||||
|
||||
private volatile CompletionStrategy completionStrategy = new SequenceSizeCompletionStrategy();
|
||||
|
||||
private final ConcurrentMap<Object, AggregationBarrier> barriers = new ConcurrentHashMap<Object, AggregationBarrier>();
|
||||
|
||||
private volatile long timeout = DEFAULT_TIMEOUT;
|
||||
|
||||
private volatile boolean sendPartialResultOnTimeout = false;
|
||||
|
||||
private volatile long reaperInterval = DEFAULT_REAPER_INTERVAL;
|
||||
|
||||
private volatile int trackedCorrelationIdCapacity = DEFAULT_TRACKED_CORRRELATION_ID_CAPACITY;
|
||||
|
||||
private volatile BlockingQueue<Object> trackedCorrelationIds;
|
||||
|
||||
private final ScheduledExecutorService executor;
|
||||
|
||||
private volatile boolean initialized;
|
||||
|
||||
|
||||
/**
|
||||
* Create a handler that delegates to the provided aggregator to combine a
|
||||
@@ -107,74 +53,16 @@ public class AggregatingMessageHandler implements MessageHandler, InitializingBe
|
||||
* single-threaded executor will be created.
|
||||
*/
|
||||
public AggregatingMessageHandler(Aggregator aggregator, ScheduledExecutorService executor) {
|
||||
super(executor);
|
||||
Assert.notNull(aggregator, "'aggregator' must not be null");
|
||||
this.aggregator = aggregator;
|
||||
this.executor = (executor != null) ? executor : Executors.newSingleThreadScheduledExecutor();
|
||||
}
|
||||
|
||||
public AggregatingMessageHandler(Aggregator aggregator) {
|
||||
this(aggregator, null);
|
||||
}
|
||||
|
||||
|
||||
/**
|
||||
* Set the default channel for sending aggregated Messages. Note that
|
||||
* precedence will be given to the 'returnAddress' of the aggregated
|
||||
* message itself, then to the 'returnAddress' of the original message.
|
||||
*/
|
||||
public void setDefaultReplyChannel(MessageChannel defaultReplyChannel) {
|
||||
this.defaultReplyChannel = defaultReplyChannel;
|
||||
}
|
||||
|
||||
/**
|
||||
* Specify a channel for sending Messages that arrive after their aggregation
|
||||
* group has either completed or timed-out.
|
||||
*/
|
||||
public void setDiscardChannel(MessageChannel discardChannel) {
|
||||
this.discardChannel = discardChannel;
|
||||
}
|
||||
|
||||
/**
|
||||
* Set the timeout for sending aggregation results and discarded Messages.
|
||||
*/
|
||||
public void setSendTimeout(long sendTimeout) {
|
||||
this.sendTimeout = sendTimeout;
|
||||
}
|
||||
|
||||
/**
|
||||
* Specify whether to aggregate and send the resulting Message when the
|
||||
* timeout elapses prior to the CompletionStrategy.
|
||||
*/
|
||||
public void setSendPartialResultOnTimeout(boolean sendPartialResultOnTimeout) {
|
||||
this.sendPartialResultOnTimeout = sendPartialResultOnTimeout;
|
||||
}
|
||||
|
||||
/**
|
||||
* Set the interval in milliseconds for the reaper thread. Default is 1000.
|
||||
*/
|
||||
public void setReaperInterval(long reaperInterval) {
|
||||
Assert.isTrue(reaperInterval > 0, "'reaperInterval' must be a positive value");
|
||||
this.reaperInterval = reaperInterval;
|
||||
}
|
||||
|
||||
/**
|
||||
* Set the number of completed correlationIds to track. Default is 1000.
|
||||
*/
|
||||
public void setTrackedCorrelationIdCapacity(int trackedCorrelationIdCapacity) {
|
||||
Assert.isTrue(trackedCorrelationIdCapacity > 0, "'trackedCorrelationIdCapacity' must be a positive value");
|
||||
this.trackedCorrelationIdCapacity = trackedCorrelationIdCapacity;
|
||||
}
|
||||
|
||||
/**
|
||||
* Initialize this handler.
|
||||
*/
|
||||
public void afterPropertiesSet() {
|
||||
this.trackedCorrelationIds = new ArrayBlockingQueue<Object>(this.trackedCorrelationIdCapacity);
|
||||
this.executor.scheduleWithFixedDelay(new ReaperTask(),
|
||||
this.reaperInterval, this.reaperInterval, TimeUnit.MILLISECONDS);
|
||||
this.initialized = true;
|
||||
}
|
||||
|
||||
|
||||
/**
|
||||
* Strategy to determine whether the group of messages is complete.
|
||||
*/
|
||||
@@ -183,128 +71,24 @@ public class AggregatingMessageHandler implements MessageHandler, InitializingBe
|
||||
this.completionStrategy = completionStrategy;
|
||||
}
|
||||
|
||||
/**
|
||||
* Maximum time to wait (in milliseconds) for the completion strategy to
|
||||
* become true. The default is 60000 (1 minute).
|
||||
*/
|
||||
public void setTimeout(long timeout) {
|
||||
Assert.isTrue(timeout >= 0, "'timeout' must not be negative");
|
||||
this.timeout = timeout;
|
||||
protected MessageBarrier createMessageBarrier() {
|
||||
return new AggregationBarrier(this.completionStrategy);
|
||||
}
|
||||
|
||||
public Message<?> handle(Message<?> message) {
|
||||
if (!this.initialized) {
|
||||
this.afterPropertiesSet();
|
||||
}
|
||||
Object correlationId = message.getHeader().getCorrelationId();
|
||||
if (correlationId == null) {
|
||||
throw new MessageHandlingException(message,
|
||||
this.getClass().getSimpleName() + " requires the 'correlationId' property");
|
||||
}
|
||||
if (this.trackedCorrelationIds.contains(correlationId)) {
|
||||
if (logger.isDebugEnabled()) {
|
||||
logger.debug("Aggregation for correlationId '" + correlationId +
|
||||
"' has already completed or timed out.");
|
||||
}
|
||||
this.sendToDiscardChannelIfAvailable(message);
|
||||
return null;
|
||||
}
|
||||
AggregationBarrier barrier = barriers.putIfAbsent(correlationId,
|
||||
new AggregationBarrier(this.completionStrategy));
|
||||
if (barrier == null) {
|
||||
barrier = barriers.get(correlationId);
|
||||
}
|
||||
List<Message<?>> releasedMessages = barrier.addAndRelease(message);
|
||||
if (CollectionUtils.isEmpty(releasedMessages)) {
|
||||
return null;
|
||||
}
|
||||
this.removeBarrier(correlationId);
|
||||
this.aggregationCompleted(correlationId, releasedMessages);
|
||||
return null;
|
||||
protected boolean isBarrierRemovable(Object correlationId, List<Message<?>> releasedMessages) {
|
||||
return releasedMessages != null && releasedMessages.size() > 0;
|
||||
}
|
||||
|
||||
private void sendToDiscardChannelIfAvailable(Message<?> message) {
|
||||
if (this.discardChannel != null) {
|
||||
if (!this.discardChannel.send(message, this.sendTimeout)) {
|
||||
if (logger.isWarnEnabled()) {
|
||||
logger.warn("unable to send to 'discardChannel', message: " + message);
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
private void aggregationCompleted(Object correlationId, List<Message<?>> messages) {
|
||||
|
||||
protected Message<?>[] processReleasedMessages(Object correlationId, List<Message<?>> messages) {
|
||||
if (CollectionUtils.isEmpty(messages)) {
|
||||
if (logger.isDebugEnabled()) {
|
||||
logger.debug("no messages to aggregate");
|
||||
}
|
||||
return;
|
||||
return new Message<?>[0];
|
||||
}
|
||||
Message<?> result = aggregator.aggregate(messages);
|
||||
if (result.getHeader().getCorrelationId() == null) {
|
||||
result.getHeader().setCorrelationId(correlationId);
|
||||
}
|
||||
MessageChannel replyChannel = this.resolveReplyChannelFromMessage(result);
|
||||
if (replyChannel == null) {
|
||||
replyChannel = this.resolveReplyChannelFromMessage(messages.get(0));
|
||||
if (replyChannel == null) {
|
||||
replyChannel = this.defaultReplyChannel;
|
||||
}
|
||||
}
|
||||
if (replyChannel != null) {
|
||||
replyChannel.send(result, this.sendTimeout);
|
||||
}
|
||||
else if (logger.isWarnEnabled()) {
|
||||
logger.warn("unable to determine reply channel for aggregation result: " + result);
|
||||
}
|
||||
}
|
||||
|
||||
private void removeBarrier(Object correlationId) {
|
||||
if (this.barriers.remove(correlationId) != null) {
|
||||
synchronized (this.trackedCorrelationIds) {
|
||||
boolean added = this.trackedCorrelationIds.offer(correlationId);
|
||||
if (!added) {
|
||||
this.trackedCorrelationIds.poll();
|
||||
this.trackedCorrelationIds.offer(correlationId);
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
private MessageChannel resolveReplyChannelFromMessage(Message<?> message) {
|
||||
Object returnAddress = message.getHeader().getReturnAddress();
|
||||
if (returnAddress != null) {
|
||||
if (returnAddress instanceof MessageChannel) {
|
||||
return (MessageChannel) returnAddress;
|
||||
}
|
||||
if (logger.isWarnEnabled()) {
|
||||
logger.warn("Aggregator can only reply to a 'returnAddress' of type MessageChannel.");
|
||||
}
|
||||
}
|
||||
return null;
|
||||
}
|
||||
|
||||
|
||||
private class ReaperTask implements Runnable {
|
||||
|
||||
public void run() {
|
||||
long currentTime = System.currentTimeMillis();
|
||||
for (Map.Entry<Object, AggregationBarrier> entry : barriers.entrySet()) {
|
||||
if (currentTime - entry.getValue().getTimestamp() >= timeout) {
|
||||
Object correlationId = entry.getKey();
|
||||
List<Message<?>> messages = entry.getValue().getMessages();
|
||||
removeBarrier(correlationId);
|
||||
if (sendPartialResultOnTimeout) {
|
||||
aggregationCompleted(correlationId, messages);
|
||||
}
|
||||
else {
|
||||
for (Message<?> message : messages) {
|
||||
sendToDiscardChannelIfAvailable(message);
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
return new Message<?>[] { result };
|
||||
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
@@ -17,14 +17,8 @@
|
||||
package org.springframework.integration.router;
|
||||
|
||||
import java.util.List;
|
||||
import java.util.concurrent.CopyOnWriteArrayList;
|
||||
import java.util.concurrent.locks.ReentrantLock;
|
||||
|
||||
import org.apache.commons.logging.Log;
|
||||
import org.apache.commons.logging.LogFactory;
|
||||
|
||||
import org.springframework.integration.message.Message;
|
||||
import org.springframework.util.Assert;
|
||||
|
||||
/**
|
||||
* MessageBarrier implementation for message aggregation. Delegates to a
|
||||
@@ -34,63 +28,21 @@ import org.springframework.util.Assert;
|
||||
* @author Marius Bogoevici
|
||||
* @author Mark Fisher
|
||||
*/
|
||||
public class AggregationBarrier implements MessageBarrier {
|
||||
public class AggregationBarrier extends AbstractMessageBarrier {
|
||||
|
||||
private final Log logger = LogFactory.getLog(this.getClass());
|
||||
|
||||
private final List<Message<?>> messages = new CopyOnWriteArrayList<Message<?>>();
|
||||
|
||||
private final CompletionStrategy completionStrategy;
|
||||
|
||||
private volatile boolean complete = false;
|
||||
|
||||
private final ReentrantLock lock = new ReentrantLock();
|
||||
|
||||
private final long timestamp = System.currentTimeMillis();
|
||||
protected final CompletionStrategy completionStrategy;
|
||||
|
||||
|
||||
/**
|
||||
* Create an AggregationBarrier with the given {@link CompletionStrategy}.
|
||||
*/
|
||||
public AggregationBarrier(CompletionStrategy completionStrategy) {
|
||||
Assert.notNull(completionStrategy, "'completionStrategy' must not be null");
|
||||
this.completionStrategy = completionStrategy;
|
||||
}
|
||||
|
||||
|
||||
/**
|
||||
* Returns the creation time of this barrier as the number of milliseconds
|
||||
* since January 1, 1970.
|
||||
* @see java.lang.System#currentTimeMillis()
|
||||
*/
|
||||
public long getTimestamp() {
|
||||
return this.timestamp;
|
||||
protected List<Message<?>> releaseAvailableMessages() {
|
||||
return (this.isComplete()) ? this.getMessages() : null;
|
||||
}
|
||||
|
||||
/**
|
||||
* Adds a message to the aggregation group and releases <em>if complete</em>.
|
||||
* Otherwise, the return value will be <code>null</code>.
|
||||
*/
|
||||
public List<Message<?>> addAndRelease(Message<?> message) {
|
||||
try {
|
||||
this.lock.lock();
|
||||
if (this.complete) {
|
||||
if (logger.isDebugEnabled()) {
|
||||
logger.debug("Message received after aggregation has already completed: " + message);
|
||||
}
|
||||
return null;
|
||||
}
|
||||
this.messages.add(message);
|
||||
this.complete = completionStrategy.isComplete(this.messages);
|
||||
return (this.complete) ? this.messages : null;
|
||||
}
|
||||
finally {
|
||||
this.lock.unlock();
|
||||
}
|
||||
protected boolean hasReceivedAllMessages() {
|
||||
return completionStrategy.isComplete(this.messages);
|
||||
}
|
||||
|
||||
public List<Message<?>> getMessages() {
|
||||
return this.messages;
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
@@ -26,9 +26,14 @@ import org.springframework.integration.message.Message;
|
||||
* {@link Message} arrives.
|
||||
*
|
||||
* @author Mark Fisher
|
||||
* @author Marius Bogoevici
|
||||
*/
|
||||
public interface MessageBarrier {
|
||||
|
||||
List<Message<?>> addAndRelease(Message<?> message);
|
||||
|
||||
long getTimestamp();
|
||||
|
||||
List<Message<?>> getMessages();
|
||||
|
||||
}
|
||||
|
||||
@@ -0,0 +1,97 @@
|
||||
/*
|
||||
* Copyright 2002-2008 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.
|
||||
* You may obtain a copy of the License at
|
||||
*
|
||||
* http://www.apache.org/licenses/LICENSE-2.0
|
||||
*
|
||||
* Unless required by applicable law or agreed to in writing, software
|
||||
* distributed under the License is distributed on an "AS IS" BASIS,
|
||||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
* See the License for the specific language governing permissions and
|
||||
* limitations under the License.
|
||||
*/
|
||||
package org.springframework.integration.router;
|
||||
|
||||
import java.util.ArrayList;
|
||||
import java.util.Collections;
|
||||
import java.util.Comparator;
|
||||
import java.util.Iterator;
|
||||
import java.util.List;
|
||||
|
||||
import org.springframework.integration.message.Message;
|
||||
|
||||
/**
|
||||
* MessageBarrier implementation for resequencing. It can either
|
||||
* release partial sequences as messages arrive, or the full sequence.
|
||||
* @author Marius Bogoevici
|
||||
*/
|
||||
public class ResequencingMessageBarrier extends AbstractMessageBarrier {
|
||||
|
||||
private int lastReleasedSequenceNumber;
|
||||
|
||||
private final Comparator<Message<?>> resequencingComparator;
|
||||
|
||||
private final boolean releasePartialSequences;
|
||||
|
||||
|
||||
/**
|
||||
* @param releasePartialSequences specifies whether partial sequences should
|
||||
* be released as they arrive, or the resequencer
|
||||
*/
|
||||
public ResequencingMessageBarrier(boolean releasePartialSequences) {
|
||||
this.resequencingComparator = new Comparator<Message<?>>() {
|
||||
public int compare(Message<?> m1, Message<?> m2) {
|
||||
return m1.getHeader().getSequenceNumber() - m2.getHeader().getSequenceNumber();
|
||||
}
|
||||
};
|
||||
this.releasePartialSequences = releasePartialSequences;
|
||||
this.lastReleasedSequenceNumber = 0;
|
||||
}
|
||||
|
||||
|
||||
protected void addMessage(Message<?> message) {
|
||||
int insertionPoint = Collections.binarySearch(messages, message, resequencingComparator);
|
||||
if (insertionPoint < 0) {
|
||||
insertionPoint = -(insertionPoint + 1);
|
||||
this.messages.add(insertionPoint, message);
|
||||
}
|
||||
}
|
||||
|
||||
protected boolean hasReceivedAllMessages() {
|
||||
//verify that there is a contiguous sequence of messages from the first to the last
|
||||
//and the last message is last in sequence, and the first message is the next one to be delivered
|
||||
//(aggregated, this means that the last possibile partial sequence of messages has been received
|
||||
Message<?> firstMessage = this.messages.get(0);
|
||||
Message<?> lastMessage = this.messages.get(messages.size() - 1);
|
||||
return (lastMessage.getHeader().getSequenceNumber() == lastMessage.getHeader().getSequenceSize()
|
||||
&& (lastMessage.getHeader().getSequenceNumber() - firstMessage.getHeader().getSequenceNumber()
|
||||
== this.messages.size() - 1
|
||||
&& this.lastReleasedSequenceNumber == firstMessage.getHeader().getSequenceNumber() - 1));
|
||||
}
|
||||
|
||||
protected List<Message<?>> releaseAvailableMessages() {
|
||||
if (this.releasePartialSequences || hasReceivedAllMessages()) {
|
||||
ArrayList<Message<?>> releasedMessages = new ArrayList<Message<?>>();
|
||||
Iterator<Message<?>> it = this.messages.iterator();
|
||||
while (it.hasNext()) {
|
||||
Message<?> currentMessage = it.next();
|
||||
if (this.lastReleasedSequenceNumber == currentMessage.getHeader().getSequenceNumber() - 1) {
|
||||
releasedMessages.add(currentMessage);
|
||||
this.lastReleasedSequenceNumber = currentMessage.getHeader().getSequenceNumber();
|
||||
it.remove();
|
||||
}
|
||||
else {
|
||||
break;
|
||||
}
|
||||
}
|
||||
return releasedMessages;
|
||||
}
|
||||
else {
|
||||
return new ArrayList<Message<?>>();
|
||||
}
|
||||
}
|
||||
|
||||
}
|
||||
@@ -0,0 +1,64 @@
|
||||
/*
|
||||
* Copyright 2002-2008 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.
|
||||
* You may obtain a copy of the License at
|
||||
*
|
||||
* http://www.apache.org/licenses/LICENSE-2.0
|
||||
*
|
||||
* Unless required by applicable law or agreed to in writing, software
|
||||
* distributed under the License is distributed on an "AS IS" BASIS,
|
||||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
* See the License for the specific language governing permissions and
|
||||
* limitations under the License.
|
||||
*/
|
||||
|
||||
package org.springframework.integration.router;
|
||||
|
||||
import java.util.List;
|
||||
import java.util.concurrent.ScheduledExecutorService;
|
||||
|
||||
import org.springframework.integration.message.Message;
|
||||
|
||||
|
||||
/**
|
||||
* An {@link AbstractMessageBarrierHandler} that waits for a group of
|
||||
* {@link Message Messages} to arrive and re-sends them in order, sorted
|
||||
* by their <code>sequenceNumber</code>.
|
||||
* <p>
|
||||
* This handler can either release partial sequences of messages or can
|
||||
* wait for the whole sequence to arrive before re-sending them.
|
||||
* <p>
|
||||
* All considerations regarding <code>timeout</code> and grouping by
|
||||
* '<code>correlationId</code>' from {@link AbstractMessageBarrierHandler} apply
|
||||
* here as well.
|
||||
*
|
||||
* @author Marius Bogoevici
|
||||
*/
|
||||
public class ResequencingMessageHandler extends AbstractMessageBarrierHandler{
|
||||
|
||||
private boolean releasePartialSequences;
|
||||
|
||||
public ResequencingMessageHandler(boolean releasePartialSequences) {
|
||||
this(null, releasePartialSequences);
|
||||
}
|
||||
|
||||
public ResequencingMessageHandler(ScheduledExecutorService executor, boolean releasePartialSequences) {
|
||||
super(executor);
|
||||
this.releasePartialSequences = releasePartialSequences;
|
||||
}
|
||||
|
||||
protected MessageBarrier createMessageBarrier() {
|
||||
return new ResequencingMessageBarrier(releasePartialSequences);
|
||||
}
|
||||
|
||||
protected Message<?>[] processReleasedMessages(Object correlationId, List<Message<?>> messages) {
|
||||
return messages.toArray(new Message<?>[messages.size()]);
|
||||
}
|
||||
|
||||
protected boolean isBarrierRemovable(Object correlationId, List<Message<?>> releasedMessages) {
|
||||
return (releasedMessages.get(releasedMessages.size() - 1).getHeader().getSequenceNumber() ==
|
||||
releasedMessages.get(releasedMessages.size() - 1).getHeader().getSequenceSize());
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,143 @@
|
||||
/*
|
||||
* Copyright 2002-2008 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.
|
||||
* You may obtain a copy of the License at
|
||||
*
|
||||
* http://www.apache.org/licenses/LICENSE-2.0
|
||||
*
|
||||
* Unless required by applicable law or agreed to in writing, software
|
||||
* distributed under the License is distributed on an "AS IS" BASIS,
|
||||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
* See the License for the specific language governing permissions and
|
||||
* limitations under the License.
|
||||
*/
|
||||
package org.springframework.integration.router;
|
||||
|
||||
import java.util.concurrent.CountDownLatch;
|
||||
import java.util.concurrent.TimeUnit;
|
||||
|
||||
import static org.junit.Assert.assertEquals;
|
||||
import static org.junit.Assert.assertNotNull;
|
||||
import static org.junit.Assert.assertNull;
|
||||
|
||||
import org.junit.Test;
|
||||
|
||||
import org.springframework.integration.channel.MessageChannel;
|
||||
import org.springframework.integration.channel.QueueChannel;
|
||||
import org.springframework.integration.message.Message;
|
||||
import org.springframework.integration.message.StringMessage;
|
||||
|
||||
/**
|
||||
* @author Marius Bogoevici
|
||||
*/
|
||||
public class ResequencerMessageHandlerTests {
|
||||
|
||||
@Test
|
||||
public void testBasicResequencing() throws InterruptedException {
|
||||
ResequencingMessageHandler aggregator = new ResequencingMessageHandler(false);
|
||||
QueueChannel replyChannel = new QueueChannel();
|
||||
Message<?> message1 = createMessage("123", "ABC", 3, 3, replyChannel);
|
||||
Message<?> message2 = createMessage("456", "ABC", 3, 1, replyChannel);
|
||||
Message<?> message3 = createMessage("789", "ABC", 3, 2, replyChannel);
|
||||
CountDownLatch latch = new CountDownLatch(3);
|
||||
aggregator.handle(message1);
|
||||
aggregator.handle(message3);
|
||||
aggregator.handle(message2);
|
||||
latch.await(1000, TimeUnit.MILLISECONDS);
|
||||
Message<?> reply1 = replyChannel.receive(500);
|
||||
Message<?> reply2 = replyChannel.receive(500);
|
||||
Message<?> reply3 = replyChannel.receive(500);
|
||||
assertNotNull(reply1);
|
||||
assertEquals(1, reply1.getHeader().getSequenceNumber());
|
||||
assertNotNull(reply2);
|
||||
assertEquals(2, reply2.getHeader().getSequenceNumber());
|
||||
assertNotNull(reply3);
|
||||
assertEquals(3, reply3.getHeader().getSequenceNumber());
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testResequencingWithIncompleteSequenceRelease() throws InterruptedException {
|
||||
ResequencingMessageHandler aggregator = new ResequencingMessageHandler(true);
|
||||
QueueChannel replyChannel = new QueueChannel();
|
||||
Message<?> message1 = createMessage("123", "ABC", 4, 2, replyChannel);
|
||||
Message<?> message2 = createMessage("456", "ABC", 4, 1, replyChannel);
|
||||
Message<?> message3 = createMessage("789", "ABC", 4, 4, replyChannel);
|
||||
Message<?> message4 = createMessage("XYZ", "ABC", 4, 3, replyChannel);
|
||||
CountDownLatch latch = new CountDownLatch(3);
|
||||
aggregator.handle(message1);
|
||||
aggregator.handle(message2);
|
||||
aggregator.handle(message3);
|
||||
latch.await(1000, TimeUnit.MILLISECONDS);
|
||||
Message<?> reply1 = replyChannel.receive(500);
|
||||
Message<?> reply2 = replyChannel.receive(500);
|
||||
Message<?> reply3 = replyChannel.receive(500);
|
||||
// only messages 1 and 2 must have been received by now
|
||||
assertNotNull(reply1);
|
||||
assertEquals(1, reply1.getHeader().getSequenceNumber());
|
||||
assertNotNull(reply2);
|
||||
assertEquals(2, reply2.getHeader().getSequenceNumber());
|
||||
assertNull(reply3);
|
||||
// when sending the last message, the whole sequence must have been sent
|
||||
latch = new CountDownLatch(1);
|
||||
aggregator.handle(message4);
|
||||
latch.await(1000, TimeUnit.MILLISECONDS);
|
||||
reply3 = replyChannel.receive(500);
|
||||
Message<?> reply4 = replyChannel.receive(500);
|
||||
assertNotNull(reply3);
|
||||
assertEquals(3, reply3.getHeader().getSequenceNumber());
|
||||
assertNotNull(reply4);
|
||||
assertEquals(4, reply4.getHeader().getSequenceNumber());
|
||||
}
|
||||
|
||||
|
||||
@Test
|
||||
public void testResequencingWithCompleteSequenceRelease() throws InterruptedException {
|
||||
ResequencingMessageHandler aggregator = new ResequencingMessageHandler(false);
|
||||
QueueChannel replyChannel = new QueueChannel();
|
||||
Message<?> message1 = createMessage("123", "ABC", 4, 2, replyChannel);
|
||||
Message<?> message2 = createMessage("456", "ABC", 4, 1, replyChannel);
|
||||
Message<?> message3 = createMessage("789", "ABC", 4, 4, replyChannel);
|
||||
Message<?> message4 = createMessage("XYZ", "ABC", 4, 3, replyChannel);
|
||||
CountDownLatch latch = new CountDownLatch(3);
|
||||
aggregator.handle(message1);
|
||||
aggregator.handle(message2);
|
||||
aggregator.handle(message3);
|
||||
latch.await(1000, TimeUnit.MILLISECONDS);
|
||||
Message<?> reply1 = replyChannel.receive(500);
|
||||
Message<?> reply2 = replyChannel.receive(500);
|
||||
Message<?> reply3 = replyChannel.receive(500);
|
||||
// no message must have been received by now
|
||||
assertNull(reply1);
|
||||
assertNull(reply2);
|
||||
assertNull(reply3);
|
||||
// when sending the last message, the whole sequence must have been sent
|
||||
latch = new CountDownLatch(1);
|
||||
aggregator.handle(message4);
|
||||
latch.await(1000, TimeUnit.MILLISECONDS);
|
||||
reply1 = replyChannel.receive(500);
|
||||
reply2 = replyChannel.receive(500);
|
||||
reply3 = replyChannel.receive(500);
|
||||
Message<?> reply4 = replyChannel.receive(500);
|
||||
assertNotNull(reply1);
|
||||
assertEquals(1, reply1.getHeader().getSequenceNumber());
|
||||
assertNotNull(reply2);
|
||||
assertEquals(2, reply2.getHeader().getSequenceNumber());
|
||||
assertNotNull(reply3);
|
||||
assertEquals(3, reply3.getHeader().getSequenceNumber());
|
||||
assertNotNull(reply4);
|
||||
assertEquals(4, reply4.getHeader().getSequenceNumber());
|
||||
}
|
||||
|
||||
private static Message<?> createMessage(String payload, Object correlationId,
|
||||
int sequenceSize, int sequenceNumber, MessageChannel replyChannel) {
|
||||
StringMessage message = new StringMessage(payload);
|
||||
message.getHeader().setCorrelationId(correlationId);
|
||||
message.getHeader().setSequenceSize(sequenceSize);
|
||||
message.getHeader().setSequenceNumber(sequenceNumber);
|
||||
message.getHeader().setReturnAddress(replyChannel);
|
||||
return message;
|
||||
}
|
||||
|
||||
}
|
||||
Reference in New Issue
Block a user