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

INT-1105, INT-1063: Big merge. Removing old aggregation and correlation stuff, and reaping the reaper.

Browse Source
This commit is contained in:
David Syer
2010-05-02 04:31:05 +00:00
parent ec76ea497a
commit d04c7008b6
58 changed files with 1658 additions and 3797 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
org.springframework.integration/src/main/java/org/springframework/integration/aggregator/AbstractAggregatingMessageGroupProcessor.java
View File

@@ -64,7 +64,7 @@ public abstract class AbstractAggregatingMessageGroupProcessor implements Messag
protected Map<String, Object> aggregateHeaders(MessageGroup group) {
Map<String, Object> aggregatedHeaders = new HashMap<String, Object>();
Set<String> conflictKeys = new HashSet<String>();
for (Message<?> message : group.getMessages()) {
for (Message<?> message : group.getUnmarked()) {
MessageHeaders currentHeaders = message.getHeaders();
for (String key : currentHeaders.keySet()) {
if (MessageHeaders.ID.equals(key) || MessageHeaders.TIMESTAMP.equals(key)

95
org.springframework.integration/src/main/java/org/springframework/integration/aggregator/AbstractMessageAggregator.java
View File

@@ -1,95 +0,0 @@
/*
* Copyright 2002-2010 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* 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.aggregator;
import java.util.ArrayList;
import java.util.List;
import org.springframework.integration.core.Message;
import org.springframework.integration.message.MessageBuilder;
import org.springframework.util.Assert;
import org.springframework.util.CollectionUtils;
/**
* A base class for aggregating a group of Messages into a single Message.
* Extends {@link AbstractMessageBarrierHandler} and waits for a
* <em>complete</em> group of {@link Message Messages} to arrive. Subclasses
* must provide the implementation of the {@link #aggregateMessages(List)}
* method to combine the group of Messages into a single {@link Message}.
*
* <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>
* 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 abstract class AbstractMessageAggregator extends
AbstractMessageBarrierHandler<List<Message<?>>> {
private volatile CompletionStrategy completionStrategy = new SequenceSizeCompletionStrategy();
/**
* Strategy to determine whether the group of messages is complete.
*/
public void setCompletionStrategy(CompletionStrategy completionStrategy) {
Assert.notNull(completionStrategy,
"'completionStrategy' must not be null");
this.completionStrategy = completionStrategy;
}
@Override
public String getComponentType() {
return "aggregator";
}
@Override
protected MessageBarrier<List<Message<?>>> createMessageBarrier(Object correlationKey) {
return new MessageBarrier<List<Message<?>>>(new ArrayList<Message<?>>(), correlationKey);
}
@Override
protected void processBarrier(MessageBarrier<List<Message<?>>> barrier) {
if (!barrier.isComplete() && !CollectionUtils.isEmpty(barrier.getMessages())) {
if (this.completionStrategy.isComplete(barrier.getMessages())) {
barrier.setComplete();
}
}
if (barrier.isComplete()) {
this.removeBarrier(barrier.getCorrelationKey());
Message<?> result = this.aggregateMessages(barrier.getMessages());
if (result != null) {
if (result.getHeaders().getCorrelationId() == null) {
result = MessageBuilder.fromMessage(result)
.setCorrelationId(barrier.getCorrelationKey())
.build();
}
this.sendReply(result, this.resolveReplyChannelFromMessage(barrier.getMessages().get(0)));
}
}
}
protected abstract Message<?> aggregateMessages(List<Message<?>> messages);
}

397
org.springframework.integration/src/main/java/org/springframework/integration/aggregator/AbstractMessageBarrierHandler.java
View File

@@ -1,397 +0,0 @@
/*
* Copyright 2002-2010 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* 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.aggregator;
import java.util.Collection;
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.ScheduledFuture;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.beans.factory.BeanFactoryAware;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.integration.channel.MessageChannelTemplate;
import org.springframework.integration.core.Message;
import org.springframework.integration.core.MessageChannel;
import org.springframework.integration.core.MessageHeaders;
import org.springframework.integration.core.MessageProducer;
import org.springframework.integration.handler.AbstractMessageHandler;
import org.springframework.integration.message.MessageBuilder;
import org.springframework.integration.message.MessageDeliveryException;
import org.springframework.integration.message.MessageHandler;
import org.springframework.integration.message.MessageHandlingException;
import org.springframework.scheduling.TaskScheduler;
import org.springframework.util.Assert;
/**
* Base class for {@link MessageBarrier}-based Message Handlers. A
* {@link MessageHandler} implementation that waits for a group of
* {@link Message Messages} to arrive and processes them together. Uses a
* {@link MessageBarrier} to store messages and to decide how the messages
* should 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 unless 'sendPartialResultsOnTimeout' is set to
* true in which case the incomplete group will be sent to the output channel.
* <p>
* Subclasses must decide what kind of a Collection they want to use. The semantics
* of adding a Message to the MessageBarrier will be decided by the Collection type.
* <p>
* Note: this class is not part of the Spring Integration API, but
* an internal class, used for implementing components that need to keep
* a list of messages until they are ready to be released or processed
* (e.g. Resequencer or Aggregator). As such it is subject to change in future
* versions.
*
* @author Mark Fisher
* @author Marius Bogoevici
*/
public abstract class AbstractMessageBarrierHandler<T extends Collection<? extends Message<?>>>
extends AbstractMessageHandler implements MessageProducer, BeanFactoryAware, 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());
private volatile MessageChannel outputChannel;
private final MessageChannelTemplate channelTemplate = new MessageChannelTemplate();
private volatile MessageChannel discardChannel;
protected final ConcurrentMap<Object, MessageBarrier<T>> barriers = new ConcurrentHashMap<Object, MessageBarrier<T>>();
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;
private volatile boolean autoStartup = true;
private volatile ScheduledFuture<?> reaperFutureTask;
private volatile boolean initialized;
private final Object lifecycleMonitor = new Object();
private volatile CorrelationStrategy correlationStrategy = new HeaderAttributeCorrelationStrategy(MessageHeaders.CORRELATION_ID);
public AbstractMessageBarrierHandler() {
this.channelTemplate.setSendTimeout(DEFAULT_SEND_TIMEOUT);
}
public void setOutputChannel(MessageChannel outputChannel) {
this.outputChannel = outputChannel;
}
/**
* 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;
}
/**
* Specify whether to aggregate and send the resulting Message when the
* timeout elapses prior to the CompletionStrategy returning true.
*/
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) {
this.trackedCorrelationIdCapacity = trackedCorrelationIdCapacity;
}
/**
* 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 be a positive value");
this.timeout = timeout;
}
public void setSendTimeout(long sendTimeout) {
this.channelTemplate.setSendTimeout(sendTimeout);
}
public void setTaskScheduler(TaskScheduler taskScheduler) {
super.setTaskScheduler(taskScheduler);
}
public void setAutoStartup(boolean autoStartup) {
this.autoStartup = autoStartup;
}
public void setCorrelationStrategy(CorrelationStrategy correlationStrategy) {
this.correlationStrategy = correlationStrategy;
}
@Override
public final void onInit() {
synchronized (this.lifecycleMonitor) {
if (!this.initialized) {
if (this.trackedCorrelationIdCapacity > 0) {
this.trackedCorrelationIds = new ArrayBlockingQueue<Object>(this.trackedCorrelationIdCapacity);
}
if (this.autoStartup) {
this.start();
}
this.initialized = true;
}
}
}
public boolean isRunning() {
synchronized (this.lifecycleMonitor) {
return this.reaperFutureTask != null;
}
}
public void start() {
synchronized (this.lifecycleMonitor) {
if (this.isRunning()) {
return;
}
Assert.state(this.getTaskScheduler() != null, "TaskScheduler must not be null");
this.reaperFutureTask = this.getTaskScheduler().scheduleWithFixedDelay(
new PrunerTask(), this.reaperInterval);
}
}
public void stop() {
synchronized (this.lifecycleMonitor) {
if (this.isRunning()) {
this.reaperFutureTask.cancel(true);
}
}
}
@Override
protected final void handleMessageInternal(Message<?> message) {
if (!this.initialized) {
this.afterPropertiesSet();
}
Object correlationKey = this.correlationStrategy.getCorrelationKey(message);
if (correlationKey == null) {
throw new MessageHandlingException(message, this.getClass().getSimpleName()
+ " requires the 'correlationKey' property");
}
if (this.trackedCorrelationIds != null && this.trackedCorrelationIds.contains(correlationKey)) {
if (logger.isDebugEnabled()) {
logger.debug("Handling of Message group with correlationKey '" + correlationKey
+ "' has already completed or timed out.");
}
this.discardMessage(message);
}
else {
this.processMessage(message, correlationKey);
}
}
private void discardMessage(Message<?> message) {
if (this.discardChannel != null) {
boolean sent = this.channelTemplate.send(message, this.discardChannel);
if (!sent && logger.isWarnEnabled()) {
logger.warn("unable to send to 'discardChannel', message: " + message);
}
}
}
@SuppressWarnings("unchecked")
private void processMessage(Message<?> message, Object correlationKey) {
MessageBarrier<T> barrier = barriers.putIfAbsent(correlationKey, createMessageBarrier(correlationKey));
if (barrier == null) {
barrier = barriers.get(correlationKey);
}
synchronized (barrier) {
if (canAddMessage(message, barrier)) {
((MessageBarrier)barrier).getMessages().add(message);
}
processBarrier(barrier);
}
}
protected final void sendReplies(Collection<Message<?>> messages, MessageChannel defaultReplyChannel) {
if (messages.isEmpty()) {
return;
}
for (Message<?> result : messages) {
sendReply(result, defaultReplyChannel);
}
}
protected final void sendReply(Message<?> message, MessageChannel defaultReplyChannel) {
MessageChannel replyChannel = this.outputChannel;
if (replyChannel == null) {
replyChannel = this.resolveReplyChannelFromMessage(message);
if (replyChannel == null) {
replyChannel = defaultReplyChannel;
}
}
if (replyChannel != null) {
if (defaultReplyChannel != null && !defaultReplyChannel.equals(replyChannel)) {
message = MessageBuilder.fromMessage(message)
.setHeaderIfAbsent(MessageHeaders.REPLY_CHANNEL, defaultReplyChannel)
.build();
}
if (!this.channelTemplate.send(message, replyChannel)) {
throw new MessageDeliveryException(message, "failed to send reply Message");
}
}
else if (logger.isWarnEnabled()) {
logger.warn("unable to determine reply target for aggregation result: " + message);
}
}
protected final MessageChannel resolveReplyChannelFromMessage(Message<?> message) {
Object replyChannel = message.getHeaders().getReplyChannel();
if (replyChannel != null) {
if (replyChannel instanceof MessageChannel) {
return (MessageChannel) replyChannel;
}
if (logger.isWarnEnabled()) {
logger.warn("Aggregator can only reply to a 'replyChannel' of type MessageChannel.");
}
}
return null;
}
protected final void removeBarrier(Object correlationId) {
if (this.barriers.remove(correlationId) != null
&& this.trackedCorrelationIds != null) {
synchronized (this.trackedCorrelationIds) {
boolean added = this.trackedCorrelationIds.offer(correlationId);
if (!added) {
this.trackedCorrelationIds.poll();
this.trackedCorrelationIds.offer(correlationId);
}
}
}
}
/**
* Verifies that a message can be added to the barrier. To be overridden by subclasses, which may add
* their own verifications. Subclasses overriding this method must call the method from the superclass.
*/
protected boolean canAddMessage(Message<?> message, MessageBarrier<T> barrier) {
if (barrier.isComplete()) {
if (logger.isDebugEnabled()) {
logger.debug("Message received after aggregation has already completed: " + message);
}
return false;
}
return true;
}
/**
* Factory method for creating a MessageBarrier implementation.
*/
protected abstract MessageBarrier<T> createMessageBarrier(Object correlationKey);
/**
* A method for processing the information in the message barrier after a message has been added or on pruning.
* The decision as to whether the messages from the {@link MessageBarrier}
* can be released normally belongs here, although calling code may forcibly set the MessageBarrier's 'complete'
* flag to true before invoking the method.
* @param barrier the {@link MessageBarrier} to be processed
*/
protected abstract void processBarrier(MessageBarrier<T> barrier);
/**
* A method for discarding the content of the message barrier.
* Can be overridden by subclasses.
* @param entry
* @param barrier
*/
protected void discardBarrier(MessageBarrier<T> barrier) {
for (Message<?> message : barrier.getMessages()) {
if (logger.isDebugEnabled()) {
logger.debug("Handling of Message group with correlation key '" + barrier.getCorrelationKey()+ "' has timed out.");
}
discardMessage(message);
}
}
/**
* A task that runs periodically, pruning the timed-out message barriers.
*/
private class PrunerTask implements Runnable {
public void run() {
long currentTime = System.currentTimeMillis();
for (Map.Entry<Object, MessageBarrier<T>> entry : barriers.entrySet()) {
if (currentTime - entry.getValue().getTimestamp() >= timeout) {
MessageBarrier<T> barrier = entry.getValue();
synchronized (barrier) {
removeBarrier(entry.getKey());
if (sendPartialResultOnTimeout) {
barrier.setComplete();
processBarrier(barrier);
}
else {
discardBarrier(barrier);
}
}
}
}
}
}
}

381
org.springframework.integration/src/main/java/org/springframework/integration/aggregator/CorrelatingMessageHandler.java
View File

@@ -1,36 +1,24 @@
/*
* Copyright 2002-2010 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* 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.
*
* 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.aggregator;
import java.util.Collection;
import java.util.Queue;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.DelayQueue;
import java.util.concurrent.Delayed;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.ReentrantLock;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.context.Lifecycle;
import org.springframework.integration.channel.ChannelResolver;
import org.springframework.integration.channel.MessageChannelTemplate;
import org.springframework.integration.channel.NullChannel;
@@ -46,33 +34,29 @@ import org.springframework.scheduling.TaskScheduler;
import org.springframework.util.Assert;
/**
* MessageHandler that holds a buffer of correlated messages in a MessageStore.
* This class takes care of correlated groups of messages that can be completed
* in batches. It is useful for aggregating, resequencing, or custom
* MessageHandler that holds a buffer of correlated messages in a MessageStore. This class takes care of correlated
* groups of messages that can be completed in batches. It is useful for aggregating, resequencing, or custom
* implementations requiring correlation.
* <p/>
* To customize this handler inject
* {@link org.springframework.integration.aggregator.CorrelationStrategy},
* {@link org.springframework.integration.aggregator.CompletionStrategy}, and
* {@link org.springframework.integration.aggregator.MessageGroupProcessor}
* implementations as you require.
* To customize this handler inject {@link CorrelationStrategy}, {@link ReleaseStrategy}, and
* {@link MessageGroupProcessor} implementations as you require.
* <p/>
* By default the CorrelationStrategy will be a
* HeaderAttributeCorrelationStrategy and the CompletionStrategy will be a
* SequenceSizeCompletionStrategy.
* By default the CorrelationStrategy will be a HeaderAttributeCorrelationStrategy and the ReleaseStrategy will be a
* SequenceSizeReleaseStrategy.
*
* @author Iwein Fuld
* @author Dave Syer
* @since 2.0
*/
public class CorrelatingMessageHandler extends AbstractMessageHandler implements MessageProducer, Lifecycle {
public class CorrelatingMessageHandler extends AbstractMessageHandler implements MessageProducer {
private static final Log logger = LogFactory.getLog(CorrelatingMessageHandler.class);
private static final long DEFAULT_SEND_TIMEOUT = 1000L;
public static final long DEFAULT_SEND_TIMEOUT = 1000L;
private static final long DEFAULT_REAPER_INTERVAL = 1000L;
public static final long DEFAULT_REAPER_INTERVAL = 1000L;
private static final long DEFAULT_TIMEOUT = 60000L;
public static final long DEFAULT_TIMEOUT = 60000L;
private final MessageStore store;
@@ -81,7 +65,7 @@ public class CorrelatingMessageHandler extends AbstractMessageHandler implements
private volatile CorrelationStrategy correlationStrategy = new HeaderAttributeCorrelationStrategy(
MessageHeaders.CORRELATION_ID);
private volatile CompletionStrategy completionStrategy = new SequenceSizeCompletionStrategy();
private volatile ReleaseStrategy ReleaseStrategy = new SequenceSizeReleaseStrategy();
private MessageChannel outputChannel;
@@ -89,41 +73,29 @@ public class CorrelatingMessageHandler extends AbstractMessageHandler implements
private volatile MessageChannel discardChannel = new NullChannel();
private final IdTracker tracker = new IdTracker();
private boolean sendPartialResultOnTimeout = false;
private final BlockingQueue<DelayedKey> keysInBuffer = new DelayQueue<DelayedKey>();
private volatile ScheduledFuture<?> reaperFutureTask;
private volatile long reaperInterval = DEFAULT_REAPER_INTERVAL;
private volatile long timeout = DEFAULT_TIMEOUT;
private volatile boolean sendPartialResultOnTimeout;
private final Object lifecycleMonitor = new Object();
private final ConcurrentMap<Object, Object> locks = new ConcurrentHashMap<Object, Object>();
public CorrelatingMessageHandler(MessageStore store, CorrelationStrategy correlationStrategy,
CompletionStrategy completionStrategy, MessageGroupProcessor processor) {
ReleaseStrategy ReleaseStrategy, MessageGroupProcessor processor) {
Assert.notNull(store);
Assert.notNull(processor);
Assert.notNull(correlationStrategy);
Assert.notNull(completionStrategy);
this.store = store;
this.outputProcessor = processor;
this.correlationStrategy = correlationStrategy;
this.completionStrategy = completionStrategy;
this.correlationStrategy = correlationStrategy == null ? new HeaderAttributeCorrelationStrategy(
MessageHeaders.CORRELATION_ID) : correlationStrategy;
this.ReleaseStrategy = ReleaseStrategy == null ? new SequenceSizeReleaseStrategy() : ReleaseStrategy;
this.channelTemplate.setSendTimeout(DEFAULT_SEND_TIMEOUT);
}
public CorrelatingMessageHandler(MessageStore store, MessageGroupProcessor processor) {
this(store, new HeaderAttributeCorrelationStrategy(MessageHeaders.CORRELATION_ID),
new SequenceSizeCompletionStrategy(), processor);
this(store, null, null, processor);
}
public CorrelatingMessageHandler(MessageGroupProcessor processor) {
this(new SimpleMessageStore(0), new HeaderAttributeCorrelationStrategy(MessageHeaders.CORRELATION_ID),
new SequenceSizeCompletionStrategy(), processor);
new SequenceSizeReleaseStrategy(), processor);
}
public void setCorrelationStrategy(CorrelationStrategy correlationStrategy) {
@@ -131,9 +103,9 @@ public class CorrelatingMessageHandler extends AbstractMessageHandler implements
this.correlationStrategy = correlationStrategy;
}
public void setCompletionStrategy(CompletionStrategy completionStrategy) {
Assert.notNull(completionStrategy);
this.completionStrategy = completionStrategy;
public void setReleaseStrategy(ReleaseStrategy ReleaseStrategy) {
Assert.notNull(ReleaseStrategy);
this.ReleaseStrategy = ReleaseStrategy;
}
public void setTaskScheduler(TaskScheduler taskScheduler) {
@@ -141,11 +113,9 @@ public class CorrelatingMessageHandler extends AbstractMessageHandler implements
}
public void setTimeout(long timeout) {
this.timeout = timeout;
}
public void setReaperInterval(long reaperInterval) {
this.reaperInterval = reaperInterval;
}
public void setOutputChannel(MessageChannel outputChannel) {
@@ -176,232 +146,115 @@ public class CorrelatingMessageHandler extends AbstractMessageHandler implements
@Override
protected void handleMessageInternal(Message<?> message) throws Exception {
Object correlationKey = correlationStrategy.getCorrelationKey(message);
if (logger.isDebugEnabled()) {
logger.debug("Handling message with correlationKey [" + correlationKey + "]: " + message);
}
try {
if (tracker.waitForLockIfNotTracked(correlationKey)) {
Collection<Message<?>> messages = store.list(correlationKey);
MessageGroup group = new MessageGroup(messages, correlationKey);
if (group.hasNoMessageSuperseding(message)) {
store(message, correlationKey);
group.add(message);
if (completionStrategy.isComplete(group.getMessages())) {
if (logger.isDebugEnabled()) {
logger.debug("Completing group with correlationKey [" + correlationKey + "]");
}
outputProcessor.processAndSend(group, channelTemplate, this.resolveReplyChannel(message,
this.outputChannel));
if (!correlationKey.equals(message.getHeaders().getCorrelationId())) {
// TODO: strategise the treatment of overwritten correlation
message = MessageBuilder.fromMessage(message).setCorrelationId(correlationKey).build();
}
Object lock = getLock(correlationKey);
synchronized (lock) {
Collection<Message<?>> messages = store.list(correlationKey);
MessageGroup group = new MessageGroup(messages, correlationKey);
if (group.add(message)) {
// TODO: use try/catch to detect problem in group.add() and use
// that to decide on discard?
store(message, correlationKey);
if (ReleaseStrategy.canRelease(group)) {
if (logger.isDebugEnabled()) {
logger.debug("Completing group with correlationKey [" + correlationKey + "]");
}
outputProcessor.processAndSend(group, channelTemplate, this.resolveReplyChannel(message,
this.outputChannel));
if (group.isComplete()) {
complete(group);
}
}
else {
discardChannel.send(message);
else {
partialComplete(group);
}
} // If not releasing any messages the group might still be complete
else if (group.isComplete()) {
for (Message<?> discard : group.getUnmarked()) {
discardChannel.send(discard);
}
complete(group);
}
}
else {
discardChannel.send(message);
}
}
finally {
tracker.unlock(correlationKey);
}
// TODO: arrange for this to be called if user desires, e.g. periodically
public final boolean forceComplete(Object correlationKey) {
Object lock = getLock(correlationKey);
synchronized (lock) {
Collection<Message<?>> all = store.list(correlationKey);
MessageGroup group = new MessageGroup(all, correlationKey);
if (all.size() > 0) {
// last chance for normal completion
if (ReleaseStrategy.canRelease(group)) {
outputProcessor.processAndSend(group, channelTemplate, resolveReplyChannel(all.iterator().next(),
this.outputChannel));
complete(group);
}
else {
if (sendPartialResultOnTimeout) {
if (logger.isInfoEnabled()) {
logger.info("Processing partially complete messages for key [" + correlationKey + "] to: "
+ outputChannel);
}
outputProcessor.processAndSend(group, channelTemplate, resolveReplyChannel(all.iterator()
.next(), this.outputChannel));
}
else {
if (logger.isInfoEnabled()) {
logger.info("Discarding partially complete messages for key [" + correlationKey + "] to: "
+ discardChannel);
}
for (Message<?> message : all) {
discardChannel.send(message);
}
}
complete(group);
}
return true;
}
return false;
}
}
private Object getLock(Object correlationKey) {
locks.putIfAbsent(correlationKey, correlationKey);
return locks.get(correlationKey);
}
private void partialComplete(MessageGroup group) {
for (Message<?> message : group.getMessages()) {
store.delete(group.getCorrelationKey(), message.getHeaders().getId());
for (Message<?> message : group.getUnmarked()) {
store.mark(group.getCorrelationKey(), message.getHeaders().getId());
}
}
private void complete(MessageGroup group) {
tracker.pushCorrelationId(group.getCorrelationKey());
store.deleteAll(group.getCorrelationKey());
Object correlationKey = group.getCorrelationKey();
store.deleteAll(correlationKey);
locks.remove(correlationKey);
}
@SuppressWarnings("unchecked")
private void store(Message<?> message, Object correlationKey) {
Message toStore = message;
if (!correlationKey.equals(message.getHeaders().getCorrelationId())) {
toStore = MessageBuilder.fromMessage(message).setCorrelationId(correlationKey).build();
}
store.put(correlationKey, toStore);
if (!keysInBuffer.contains(correlationKey)) {
keysInBuffer.add(new DelayedKey(correlationKey, timeout));
}
}
public boolean isRunning() {
synchronized (this.lifecycleMonitor) {
return this.reaperFutureTask != null;
}
}
public void start() {
synchronized (this.lifecycleMonitor) {
if (this.isRunning()) {
return;
}
Assert.state(this.getTaskScheduler() != null, "'taskScheduler' must not be null");
this.reaperFutureTask = this.getTaskScheduler().scheduleWithFixedDelay(new PrunerTask(),
this.reaperInterval);
}
}
public void stop() {
synchronized (this.lifecycleMonitor) {
if (this.isRunning()) {
this.reaperFutureTask.cancel(true);
}
}
}
private class PrunerTask implements Runnable {
public void run() {
if (logger.isTraceEnabled()) {
logger.trace("PrunerTask is running");
}
DelayedKey delayedKey;
try {
while ((delayedKey = keysInBuffer.poll(reaperInterval, TimeUnit.MILLISECONDS)) != null) {
Object key = delayedKey.getKey();
if (logger.isDebugEnabled()) {
logger.debug(this + "'s PrunerTask is processing " + key);
}
if (!forceComplete(key)) {
keysInBuffer.offer(delayedKey);
}
}
}
catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
}
}
protected final boolean forceComplete(Object key) {
try {
if (tracker.tryLockFor(key)) {
Collection<Message<?>> all = store.list(key);
MessageGroup group = new MessageGroup(all, key);
if (all.size() > 0) {
// last chance for normal completion
MessageChannel outputChannel = resolveReplyChannel(all.iterator().next(), this.outputChannel);
if (completionStrategy.isComplete(all)) {
outputProcessor.processAndSend(group, channelTemplate, outputChannel);
complete(group);
} else {
if (sendPartialResultOnTimeout) {
if (logger.isInfoEnabled()) {
logger.info("Processing partially complete messages for key [" + key + "] to: "
+ outputChannel);
}
outputProcessor.processAndSend(group, channelTemplate, outputChannel);
}
else {
if (logger.isInfoEnabled()) {
logger.info("Discarding partially complete messages for key [" + key + "] to: "
+ discardChannel);
}
for (Message<?> message : all) {
discardChannel.send(message);
}
}
partialComplete(group);
}
}
return true;
}
else {
return false;
}
}
finally {
tracker.unlock(key);
}
}
private final class DelayedKey implements Delayed {
private final Object key;
private final Long releaseTime;
private final TimeUnit unit = TimeUnit.MILLISECONDS;
public DelayedKey(Object correlationKey, long delay) {
Assert.notNull(correlationKey, "'correlationKey' must not be null");
this.key = correlationKey;
this.releaseTime = System.currentTimeMillis() + delay;
}
public long getDelay(TimeUnit unit) {
return unit.convert(this.releaseTime - System.currentTimeMillis(), this.unit);
}
public int compareTo(Delayed o) {
return ((Long) this.getDelay(this.unit)).compareTo(o.getDelay(this.unit));
}
public Object getKey() {
return key;
}
}
private final class IdTracker {
private final ConcurrentMap<Object, ReentrantLock> trackerLocks = new ConcurrentHashMap<Object, ReentrantLock>();
private final Queue<Object> trackedCorrelationIds = new LinkedBlockingQueue<Object>();
private void pushCorrelationId(Object correlationKey) {
while (!trackedCorrelationIds.offer(correlationKey)) {
// make room in the queue
trackedCorrelationIds.poll();
}
trackerLocks.remove(correlationKey);
}
/**
* Call this method to check if an id is tracked and obtain a lock for
* it. Don't forget to finally unlock afterwards.
* @return false if the key was tracked, true after obtaining the lock
* otherwise
*/
private boolean waitForLockIfNotTracked(Object correlationKey) {
ReentrantLock lock = trackerLocks.get(correlationKey);
if (lock == null) {
if (trackedCorrelationIds.contains(correlationKey)) {
// this correlation key is already processed
// in the near past: disallow processing
return false;
}
lock = new ReentrantLock();
ReentrantLock original = trackerLocks.putIfAbsent(correlationKey, lock);
lock = original == null ? lock : original;
}
lock.lock();
return true;
}
private boolean tryLockFor(Object correlationKey) {
ReentrantLock lock = trackerLocks.get(correlationKey);
if (lock == null) {
lock = new ReentrantLock();
ReentrantLock original = trackerLocks.putIfAbsent(correlationKey, lock);
lock = original == null ? lock : original;
}
return lock.tryLock();
}
private void unlock(Object correlationKey) {
ReentrantLock lock = trackerLocks.get(correlationKey);
if (lock != null && lock.isHeldByCurrentThread()) {
lock.unlock();
}
}
store.put(correlationKey, message);
}
}

2
org.springframework.integration/src/main/java/org/springframework/integration/aggregator/DefaultAggregatingMessageGroupProcessor.java
View File

@@ -36,7 +36,7 @@ public class DefaultAggregatingMessageGroupProcessor extends AbstractAggregating
@Override
protected final Object aggregatePayloads(MessageGroup group) {
Collection<Message<?>> messages = group.getMessages();
Collection<Message<?>> messages = group.getUnmarked();
Assert.notEmpty(messages, this.getClass().getSimpleName() + " cannot process empty message groups");
List<Object> payloads = new ArrayList<Object>(messages.size());
for (Message<?> message : messages) {

60
org.springframework.integration/src/main/java/org/springframework/integration/aggregator/DefaultMessageAggregator.java
View File

@@ -1,60 +0,0 @@
/*
* Copyright 2002-2009 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.aggregator;
import java.util.ArrayList;
import java.util.List;
import org.springframework.integration.core.Message;
import org.springframework.integration.message.MessageBuilder;
/**
* The Default Message Aggregator implementation that combines a group of
* messages into a single message containing a {@link List} of all payloads. The
* elements of the List are in order of their receiving. Any MessageHeader value
* is ignored except the <code>correlationId</code>.
*
* <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>
*
* <p>
* All considerations regarding <code>timeout</code> and grouping by
* <code>correlationId</code> from {@link AbstractMessageBarrierHandler} apply
* here as well.
* </p>
*
* @author Alex Peters
* @since 1.0.3
*/
public class DefaultMessageAggregator extends AbstractMessageAggregator {
/**
* {@inheritDoc}
*/
@Override
protected Message<?> aggregateMessages(List<Message<?>> messages) {
List<Object> payloads = new ArrayList<Object>(messages.size());
for (Message<?> message : messages) {
payloads.add(message.getPayload());
}
return MessageBuilder.withPayload(payloads).build();
}
}

91
org.springframework.integration/src/main/java/org/springframework/integration/aggregator/DefaultResequencerStrategies.java
View File

@@ -1,91 +0,0 @@
/*
* Copyright 2002-2010 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* 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.aggregator;
import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
import java.util.Comparator;
import java.util.List;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.atomic.AtomicInteger;
import org.springframework.integration.channel.MessageChannelTemplate;
import org.springframework.integration.core.Message;
import org.springframework.integration.core.MessageChannel;
import org.springframework.integration.core.MessageHeaders;
/**
* This class implements all the strategy interfaces needed for a default
* resequencer.
*
* @author Iwein Fuld
* @since 2.0
*/
public class DefaultResequencerStrategies implements CorrelationStrategy, CompletionStrategy, MessageGroupProcessor {
private final ConcurrentMap<Object, AtomicInteger> nextMessagesToPass = new ConcurrentHashMap<Object, AtomicInteger>();
private final ConcurrentMap<Object, AtomicInteger> lastMessagesToPass = new ConcurrentHashMap<Object, AtomicInteger>();
private volatile SequenceNumberComparator sequenceNumberComparator = new SequenceNumberComparator();
private volatile boolean releasePartialSequences;
public Object getCorrelationKey(Message<?> message) {
Object key = message.getHeaders().getCorrelationId();
nextMessagesToPass.putIfAbsent(key, new AtomicInteger(1));
return key;
}
public boolean isComplete(Collection<? extends Message<?>> messages) {
return releasePartialSequences
|| (!messages.isEmpty() && messages.iterator().next().getHeaders().getSequenceSize() == messages.size());
}
public void processAndSend(MessageGroup group, MessageChannelTemplate channelTemplate, MessageChannel outputChannel) {
Collection<Message<?>> all = group.getMessages();
Object correlationKey = group.getCorrelationKey();
if (all.size() > 0) {
List<Message<?>> sorted = new ArrayList<Message<?>>(all);
Collections.sort(sorted, sequenceNumberComparator);
AtomicInteger nextSequence = nextMessagesToPass.get(correlationKey);
for (Message<?> message : sorted) {
final int sequenceNumber = message.getHeaders().getSequenceNumber();
if (sequenceNumber <= nextSequence.get()) {
channelTemplate.send(message, outputChannel);
nextSequence.compareAndSet(sequenceNumber, sequenceNumber + 1);
}
}
MessageHeaders headers = sorted.get(0).getHeaders();
if (all.size() == headers.getSequenceSize()) {
// TODO: it's only complete if this is true...
}
}
}
public void setReleasePartialSequences(boolean releasePartialSequences) {
this.releasePartialSequences = releasePartialSequences;
}
private static class SequenceNumberComparator implements Comparator<Message<?>> {
public int compare(Message<?> o1, Message<?> o2) {
return o1.getHeaders().getSequenceNumber().compareTo(o2.getHeaders().getSequenceNumber());
}
}
}

107
org.springframework.integration/src/main/java/org/springframework/integration/aggregator/MessageBarrier.java
View File

@@ -1,107 +0,0 @@
/*
* 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.aggregator;
import java.util.HashMap;
import java.util.Map;
import java.util.Collection;
import org.springframework.integration.core.Message;
/**
* Utility class for AbstractMessageBarrierHandler and its subclasses for
* storing objects while in transit. It is a wrapper around a {@link java.util.Collection},
* providing special properties for recording the complete status, the creation
* time (for determining if a group of messages has timed out), and the
* correlation id for a group of messages (available after the first message has
* been added to it). This is a parameterized type, allowing different different
* client classes to use different types of Collections and their respective features.
*
* Can store/retrieve attributes through its setAttribute() and getAttribute() methods.
*
* This class is not thread-safe and will be synchronized by the calling code.
*
* @author Marius Bogoevici
*/
public class MessageBarrier<T extends Collection<? extends Message<?>>> {
protected final T messages;
private volatile boolean complete = false;
private Object correlationKey;
private final long timestamp = System.currentTimeMillis();
private final Map<String, Object> attributes = new HashMap<String, Object>();
public MessageBarrier(T messages, Object correlationKey) {
this.messages = messages;
this.correlationKey = correlationKey;
}
public Object getCorrelationKey() {
return this.correlationKey;
}
/**
* 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;
}
/**
* Marks the barrier as complete.
*/
public void setComplete() {
this.complete = true;
}
/**
* True if the barrier has received all the messages and can proceed to
* release them.
*/
public boolean isComplete() {
return this.complete;
}
public T getMessages() {
return this.messages;
}
/**
* Sets a the value of a given attribute on the MessageBarrier.
* @param attributeName
* @param value
*/
public void setAttribute(String attributeName, Object value) {
this.attributes.put(attributeName, value);
}
/**
* Gets the value of a given attribute from the MessageBarrier.
* @param attributeName
*/
public <V> V getAttribute(String attributeName) {
return (V)this.attributes.get(attributeName);
}
}

50
org.springframework.integration/src/main/java/org/springframework/integration/aggregator/MessageCountReleaseStrategy.java Executable file
View File

@@ -0,0 +1,50 @@
/*
* Copyright 2002-2010 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in compliance with
* the License. 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.aggregator;
/**
* A {@link ReleaseStrategy} that releases only the first <code>n</code> messages, where <code>n</code> is a threshold.
*
* @author Dave Syer
*
*/
public class MessageCountReleaseStrategy implements ReleaseStrategy {
private final int threshold;
/**
* @param threshold the number of messages to accept before releasing
*/
public MessageCountReleaseStrategy(int threshold) {
super();
this.threshold = threshold;
}
/**
* Convenient constructor is only one message is required (threshold=1).
*/
public MessageCountReleaseStrategy() {
this(1);
}
/**
* Release the group if it has more messages than the threshold and has not previously been released. Previous
* releases leave an imprint on the group in the form of marked messages. It is possible that more messages than the
* threshold could be released, but only if multiple consumers receive messages from the same group concurrently.
*/
public boolean canRelease(MessageGroup group) {
return group.size() >= threshold && group.getMarked().size() == 0;
}
}

150
org.springframework.integration/src/main/java/org/springframework/integration/aggregator/MessageGroup.java
View File

@@ -1,97 +1,145 @@
/*
* Copyright 2002-2010 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* 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.
*
* 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.aggregator;
import java.util.ArrayList;
import java.util.Collection;
import java.util.HashSet;
import org.springframework.integration.core.Message;
import org.springframework.integration.store.MessageStore;
/**
* Represents a mutable group of correlated messages that is bound to a certain
* {@link org.springframework.integration.store.MessageStore} and correlation
* key. The group will grow during its lifetime, when messages are
* <code>add</code>ed to it. <strong>This is not thread safe and should not be
* used for long running aggregations</strong>.
* <p/>
* According to its
* {@link org.springframework.integration.aggregator.CompletionStrategy} it can
* be <i>complete</i> depending on the messages in the group.
* <p/>
* Optionally MessageGroupListeners can be added to get callbacks when (parts
* of) the group are processed or the whole group is completed.
* Represents a mutable group of correlated messages that is bound to a certain {@link MessageStore} and correlation
* key. The group will grow during its lifetime, when messages are <code>add</code>ed to it. <strong>This is not thread
* safe and should not be used for long running aggregations</strong>.
*
* @author Iwein Fuld
* @author Oleg Zhurakousky
* @author Dave Syer
*
* @since 2.0
*/
public class MessageGroup {
private final Object correlationKey;
private final Collection<Message<?>> messages = new ArrayList<Message<?>>();
private final Collection<Message<?>> marked = new HashSet<Message<?>>();
private final Collection<Message<?>> unmarked = new HashSet<Message<?>>();
public MessageGroup(Object correlationKey) {
this.correlationKey = correlationKey;
}
public MessageGroup(Collection<? extends Message<?>> originalMessages, Object correlationKey) {
this.correlationKey = correlationKey;
this.messages.addAll(originalMessages);
this(correlationKey);
for (Message<?> message : originalMessages) {
add(message);
}
}
/**
* This method determines whether messages have been added to this group
* that supersede the given message based on its sequence id. This can be
* helpful to avoid ending up with sequences larger than their required
* sequence size or sequences that are missing certain sequence numbers.
* Add a message to the internal list. This is needed to avoid hitting the underlying store or copying the internal
* list. Use with care.
*/
public boolean hasNoMessageSuperseding(Message<?> message) {
Integer messageSequenceNumber = message.getHeaders().getSequenceNumber();
if (messageSequenceNumber != null && messageSequenceNumber > 0) {
for (Message<?> member : messages) {
Integer memberSequenceNumber = member.getHeaders().getSequenceNumber();
if (messageSequenceNumber.equals(memberSequenceNumber)) {
return false;
}
}
public boolean add(Message<?> message) {
if (isMember(message)) {
return false;
}
if (message.getHeaders().containsKey(MessageStore.PROCESSED)
&& message.getHeaders().get(MessageStore.PROCESSED, Boolean.class)) {
this.marked.add(message);
} else {
this.unmarked.add(message);
}
return true;
}
/**
* Add a message to the internal list. This is needed to avoid hitting the
* underlying store or copying the internal list. Use with care.
* @return internal message list, modification is allowed, but not recommended
*/
protected void add(Message<?> message) {
messages.add(message);
public Collection<Message<?>> getUnmarked() {
return unmarked;
}
/**
* @return internal message list, modification is allowed, but not
* recommended
* @return internal message list, modification is allowed, but not recommended
*/
public Collection<Message<?>> getMessages() {
return messages;
public Collection<Message<?>> getMarked() {
return marked;
}
/**
* @return the correlation key that links these messages together according
* to a particular CorrelationStrategy
* @return the correlation key that links these messages together according to a particular CorrelationStrategy
*/
public Object getCorrelationKey() {
return correlationKey;
}
public boolean isComplete() {
if (size() == 0) {
return true;
}
int sequenceSize = getSequenceSize();
// If there is no sequence then it must be complete....
return sequenceSize == 0 || sequenceSize == size();
}
public int getSequenceSize() {
if (size() == 0) {
return 0;
}
return getOne().getHeaders().getSequenceSize();
}
private Message<?> getOne() {
return unmarked.isEmpty() ? (marked.isEmpty() ? null : marked.iterator().next()) : unmarked.iterator().next();
}
public int size() {
return marked.size() + unmarked.size();
}
/**
* This method determines whether messages have been added to this group that supersede the given message based on
* its sequence id. This can be helpful to avoid ending up with sequences larger than their required sequence size
* or sequences that are missing certain sequence numbers.
*/
private boolean isMember(Message<?> message) {
if (size() == 0) {
return false;
}
Integer messageSequenceNumber = message.getHeaders().getSequenceNumber();
if (messageSequenceNumber != null && messageSequenceNumber > 0) {
Integer messageSequenceSize = message.getHeaders().getSequenceSize();
if (!messageSequenceSize.equals(getSequenceSize())
|| containsSequenceNumber(unmarked, messageSequenceNumber)
|| containsSequenceNumber(marked, messageSequenceNumber)) {
return true;
}
}
return false;
}
private boolean containsSequenceNumber(Collection<Message<?>> messages, Integer messageSequenceNumber) {
for (Message<?> member : messages) {
Integer memberSequenceNumber = member.getHeaders().getSequenceNumber();
if (messageSequenceNumber.equals(memberSequenceNumber)) {
return true;
}
}
return false;
}
}

119
org.springframework.integration/src/main/java/org/springframework/integration/aggregator/MethodInvokingAggregator.java
View File

@@ -1,119 +0,0 @@
/*
* 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.aggregator;
import java.lang.annotation.Annotation;
import java.lang.reflect.Method;
import java.util.List;
import java.util.concurrent.atomic.AtomicReference;
import org.springframework.aop.support.AopUtils;
import org.springframework.core.annotation.AnnotationUtils;
import org.springframework.integration.annotation.Aggregator;
import org.springframework.integration.core.Message;
import org.springframework.integration.message.GenericMessage;
import org.springframework.util.Assert;
import org.springframework.util.CollectionUtils;
import org.springframework.util.ReflectionUtils;
/**
* {@link AbstractMessageAggregator} adapter for methods annotated with
* {@link Aggregator @Aggregator} annotation and for <code>aggregator</code>
* elements (e.g. &lt;aggregator ref="beanReference" method="methodName"/&gt;).
*
* @author Marius Bogoevici
* @author Mark Fisher
*/
public class MethodInvokingAggregator extends AbstractMessageAggregator {
private final MessageListMethodAdapter methodInvoker;
public MethodInvokingAggregator(Object object, Method method) {
this.methodInvoker = new MessageListMethodAdapter(object, method);
}
public MethodInvokingAggregator(Object object, String methodName) {
this.methodInvoker = new MessageListMethodAdapter(object, methodName);
}
public MethodInvokingAggregator(Object object) {
Assert.notNull(object, "object must not be null");
Method method = this.findAggregatorMethod(object);
Assert.notNull(method, "unable to resolve Aggregator method on target class ["
+ object.getClass() + "]");
this.methodInvoker = new MessageListMethodAdapter(object, method);
}
public Message<?> aggregateMessages(List<Message<?>> messages) {
if (CollectionUtils.isEmpty(messages)) {
return null;
}
Object returnedValue = this.methodInvoker.executeMethod(messages);
if (returnedValue == null) {
return null;
}
if (returnedValue instanceof Message) {
return (Message<?>) returnedValue;
}
return new GenericMessage<Object>(returnedValue);
}
private Method findAggregatorMethod(Object candidate) {
Class<?> targetClass = AopUtils.getTargetClass(candidate);
if (targetClass == null) {
targetClass = candidate.getClass();
}
Method method = this.findAnnotatedMethod(targetClass);
if (method == null) {
method = this.findSinglePublicMethod(targetClass);
}
return method;
}
private Method findAnnotatedMethod(final Class<?> targetClass) {
final AtomicReference<Method> annotatedMethod = new AtomicReference<Method>();
ReflectionUtils.doWithMethods(targetClass, new ReflectionUtils.MethodCallback() {
public void doWith(Method method) throws IllegalArgumentException, IllegalAccessException {
Annotation annotation = AnnotationUtils.findAnnotation(method, Aggregator.class);
if (annotation != null) {
Assert.isNull(annotatedMethod.get(), "found more than one method on target class ["
+ targetClass + "] with the annotation type [" + Aggregator.class.getName() + "]");
annotatedMethod.set(method);
}
}
});
return annotatedMethod.get();
}
private Method findSinglePublicMethod(Class<?> targetClass) {
Method result = null;
for (Method method : targetClass.getMethods()) {
if (!method.getDeclaringClass().equals(Object.class)) {
if (result != null) {
throw new IllegalArgumentException(
"Class [" + targetClass + "] contains more than one public Method.");
}
result = method;
}
}
return result;
}
}

47
org.springframework.integration/src/main/java/org/springframework/integration/aggregator/MethodInvokingAggregatorFactoryBean.java
View File

@@ -1,47 +0,0 @@
package org.springframework.integration.aggregator;
import org.springframework.beans.factory.FactoryBean;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.integration.core.MessageHeaders;
import org.springframework.integration.store.MessageStore;
import org.springframework.integration.store.SimpleMessageStore;
public class MethodInvokingAggregatorFactoryBean implements
FactoryBean<CorrelatingMessageHandler> , InitializingBean{
private static final int DEFAULT_CAPACITY = Integer.MAX_VALUE;
private MessageStore store = new SimpleMessageStore(DEFAULT_CAPACITY);
private CorrelationStrategy correlationStrategy = new HeaderAttributeCorrelationStrategy(
MessageHeaders.CORRELATION_ID);
private CompletionStrategy completionStrategy = new SequenceSizeCompletionStrategy();
private MessageGroupProcessor processor;
private Object target;
public void setTarget(Object target) {
this.target = target;
}
public void afterPropertiesSet() throws Exception {
// build correllation strategy
// build completion strategy
// build processor
}
public CorrelatingMessageHandler getObject() throws Exception {
return new CorrelatingMessageHandler(store, correlationStrategy,
completionStrategy, processor);
}
public Class<? extends CorrelatingMessageHandler> getObjectType() {
return CorrelatingMessageHandler.class;
}
public boolean isSingleton() {
return false;
}
}

100
org.springframework.integration/src/main/java/org/springframework/integration/aggregator/MethodInvokingMessageGroupProcessor.java
View File

@@ -18,17 +18,19 @@ package org.springframework.integration.aggregator;
import java.lang.annotation.Annotation;
import java.lang.reflect.Method;
import java.util.Arrays;
import java.util.Collection;
import java.util.HashSet;
import java.util.Iterator;
import java.util.List;
import java.util.Set;
import java.util.concurrent.atomic.AtomicReference;
import org.springframework.aop.support.AopUtils;
import org.springframework.core.annotation.AnnotationUtils;
import org.springframework.integration.annotation.Aggregator;
import org.springframework.integration.annotation.Header;
import org.springframework.integration.core.Message;
import org.springframework.util.Assert;
import org.springframework.util.ReflectionUtils;
/**
* MessageGroupProcessor that serves as an adapter for the invocation of a POJO method.
@@ -41,23 +43,20 @@ public class MethodInvokingMessageGroupProcessor extends AbstractAggregatingMess
private final MessageListMethodAdapter adapter;
/**
* Creates a wrapper around the target passed in. This constructor will
* choose the best fitting method and throw an exception when methods are
* ambiguous or no fitting methods can be found.
* Creates a wrapper around the target passed in. This constructor will choose the best fitting method and throw an
* exception when methods are ambiguous or no fitting methods can be found.
*
* @param target the object to wrap
* @throws IllegalStateException when no single method can be found unambiguously
*/
public MethodInvokingMessageGroupProcessor(Object target) {
this.adapter = new MessageListMethodAdapter(target, this.selectMethodFrom(target));
this.adapter = new MessageListMethodAdapter(target, this.findAggregatorMethod(target));
}
/**
* Creates a wrapper around the object passed in. This constructor will look
* for a named method specifically and fail when it cannot find a method
* with the given name.
* Creates a wrapper around the object passed in. This constructor will look for a named method specifically and
* fail when it cannot find a method with the given name.
*
* @param target the object to wrap
* @param method the name of the method to look for
@@ -66,34 +65,54 @@ public class MethodInvokingMessageGroupProcessor extends AbstractAggregatingMess
this.adapter = new MessageListMethodAdapter(target, method);
}
@Override
protected final Object aggregatePayloads(MessageGroup group) {
final Collection<Message<?>> messagesUpForProcessing = group.getMessages();
final Collection<Message<?>> messagesUpForProcessing = group.getUnmarked();
Object result = this.adapter.executeMethod(messagesUpForProcessing);
return result;
}
private Method selectMethodFrom(Object target) {
Method[] methods = target.getClass().getMethods();
Set<Method> candidates = new HashSet<Method>(Arrays.asList(methods));
removeObjectMethodsFrom(candidates);
removeVoidMethodsFrom(candidates);
removeListIncompatibleMethodsFrom(candidates);
Set<Method> notAnnotatedCandidates = new HashSet<Method>();
if (candidates.size() > 1) {
notAnnotatedCandidates.addAll(removeNotAnnotatedFrom(candidates));
private Method findAggregatorMethod(Object candidate) {
Class<?> targetClass = AopUtils.getTargetClass(candidate);
if (targetClass == null) {
targetClass = candidate.getClass();
}
// if no methods are annotated we need to look in more detail in the unannotated methods
if (candidates.size() < 1) {
candidates = notAnnotatedCandidates;
removeUnfittingFrom(candidates);
Method method = this.findAnnotatedMethod(targetClass);
if (method == null) {
method = this.findSinglePublicMethod(targetClass);
}
Assert.state(candidates.size() == 1,
"Method selection failed, there should be exactly one candidate, found [" + candidates + "]");
return candidates.iterator().next();
return method;
}
private Method findAnnotatedMethod(final Class<?> targetClass) {
final AtomicReference<Method> annotatedMethod = new AtomicReference<Method>();
ReflectionUtils.doWithMethods(targetClass, new ReflectionUtils.MethodCallback() {
public void doWith(Method method) throws IllegalArgumentException, IllegalAccessException {
Annotation annotation = AnnotationUtils.findAnnotation(method, Aggregator.class);
if (annotation != null) {
Assert.isNull(annotatedMethod.get(), "found more than one method on target class [" + targetClass
+ "] with the annotation type [" + Aggregator.class.getName() + "]");
annotatedMethod.set(method);
}
}
});
return annotatedMethod.get();
}
private Method findSinglePublicMethod(Class<?> targetClass) {
Set<Method> methods = new HashSet<Method>();
for (Method method : targetClass.getMethods()) {
if (!method.getDeclaringClass().equals(Object.class)) {
methods.add(method);
}
}
removeListIncompatibleMethodsFrom(methods);
removeVoidMethodsFrom(methods);
removeUnfittingFrom(methods);
if (methods.size() > 1) {
throw new IllegalArgumentException("Class [" + targetClass + "] contains more than one public Method.");
}
return methods.isEmpty() ? null : methods.iterator().next();
}
private void removeListIncompatibleMethodsFrom(Set<Method> candidates) {
@@ -101,7 +120,7 @@ public class MethodInvokingMessageGroupProcessor extends AbstractAggregatingMess
public boolean select(Method method) {
int found = 0;
for (Class<?> parameterClass : method.getParameterTypes()) {
if (parameterClass.isAssignableFrom(List.class)) {
if (Collection.class.isAssignableFrom(parameterClass)) {
found++;
}
}
@@ -118,15 +137,6 @@ public class MethodInvokingMessageGroupProcessor extends AbstractAggregatingMess
});
}
private Set<Method> removeNotAnnotatedFrom(Set<Method> candidates) {
return removeMethodsMatchingSelector(candidates, new MethodSelector() {
public boolean select(Method method) {
Aggregator annotation = method.getAnnotation(Aggregator.class);
return (annotation == null);
}
});
}
private Set<Method> removeUnfittingFrom(Set<Method> candidates) {
return removeMethodsMatchingSelector(candidates, new MethodSelector() {
public boolean select(Method method) {
@@ -141,7 +151,7 @@ public class MethodInvokingMessageGroupProcessor extends AbstractAggregatingMess
int candidateParametersFound = 0;
for (int i = 0; i < parameterTypes.length; i++) {
Class<?> parameterType = parameterTypes[i];
if (parameterType.isAssignableFrom(List.class)) {
if (Collection.class.isAssignableFrom(parameterType)) {
boolean headerAnnotationFound = false;
for (Annotation annotation : parameterAnnotations[i]) {
if (annotation instanceof Header) {
@@ -156,14 +166,6 @@ public class MethodInvokingMessageGroupProcessor extends AbstractAggregatingMess
return candidateParametersFound == 1;
}
private void removeObjectMethodsFrom(Set<Method> candidates) {
removeMethodsMatchingSelector(candidates, new MethodSelector() {
public boolean select(Method method) {
return method.getDeclaringClass().equals(Object.class);
}
});
}
private Set<Method> removeMethodsMatchingSelector(Set<Method> candidates, MethodSelector selector) {
Set<Method> removed = new HashSet<Method>();
Iterator<Method> iterator = candidates.iterator();

4
org.springframework.integration/src/main/java/org/springframework/integration/aggregator/PassThroughMessageGroupProcessor.java
View File

@@ -7,7 +7,7 @@ import org.springframework.integration.core.MessageChannel;
/**
* This implementation of MessageGroupProcessor will forward all messages inside the group to the given output channel.
* This is useful if there is no requirement to process the messages, but they should just be blocked as a group until
* their CompletionStrategy lets them pass through.
* their ReleaseStrategy lets them pass through.
*
* @author Iwein Fuld
* @since 2.0.0
@@ -15,7 +15,7 @@ import org.springframework.integration.core.MessageChannel;
public class PassThroughMessageGroupProcessor implements MessageGroupProcessor {
public void processAndSend(MessageGroup group, MessageChannelTemplate channelTemplate, MessageChannel outputChannel) {
for (Message<?> message : group.getMessages()) {
for (Message<?> message : group.getUnmarked()) {
channelTemplate.send(message, outputChannel);
}
}

8
org.springframework.integration/src/main/java/org/springframework/integration/aggregator/CompletionStrategy.java → org.springframework.integration/src/main/java/org/springframework/integration/aggregator/ReleaseStrategy.java
View File

@@ -16,19 +16,17 @@
package org.springframework.integration.aggregator;
import java.util.Collection;
import org.springframework.integration.core.Message;
/**
* Strategy for determining when a group of messages reaches a state of
* completion (i.e. can trip a barrier).
*
* @author Mark Fisher
* @see AbstractMessageAggregator
* @author Dave Syer
*/
public interface CompletionStrategy {
public interface ReleaseStrategy {
boolean isComplete(Collection<? extends Message<?>> messages);
boolean canRelease(MessageGroup group);
}

20
org.springframework.integration/src/main/java/org/springframework/integration/aggregator/CompletionStrategyAdapter.java → org.springframework.integration/src/main/java/org/springframework/integration/aggregator/ReleaseStrategyAdapter.java
View File

@@ -17,36 +17,34 @@
package org.springframework.integration.aggregator;
import java.lang.reflect.Method;
import java.util.Collection;
import org.springframework.integration.core.Message;
import org.springframework.util.Assert;
/**
* Adapter for methods annotated with
* {@link org.springframework.integration.annotation.CompletionStrategy @CompletionStrategy}
* and for '<code>completion-strategy</code>' elements that include a '<code>method</code>'
* attribute (e.g. &lt;completion-strategy ref="beanReference" method="methodName"/&gt;).
* {@link org.springframework.integration.annotation.ReleaseStrategy @ReleaseStrategy}
* and for '<code>release-strategy</code>' elements that include a '<code>method</code>'
* attribute (e.g. &lt;release-strategy ref="beanReference" method="methodName"/&gt;).
*
* @author Marius Bogoevici
*/
public class CompletionStrategyAdapter extends MessageListMethodAdapter implements CompletionStrategy {
public class ReleaseStrategyAdapter extends MessageListMethodAdapter implements ReleaseStrategy {
public CompletionStrategyAdapter(Object object, Method method) {
public ReleaseStrategyAdapter(Object object, Method method) {
super(object, method);
this.assertMethodReturnsBoolean();
}
public CompletionStrategyAdapter(Object object, String methodName) {
public ReleaseStrategyAdapter(Object object, String methodName) {
super(object, methodName);
this.assertMethodReturnsBoolean();
}
public boolean isComplete(Collection<? extends Message<?>> messages) {
return ((Boolean) executeMethod(messages)).booleanValue();
public boolean canRelease(MessageGroup messages) {
return ((Boolean) executeMethod(messages.getUnmarked())).booleanValue() && messages.getMarked().isEmpty();
}
private void assertMethodReturnsBoolean() {
Assert.isTrue(Boolean.class.equals(this.getMethod().getReturnType())
|| boolean.class.equals(this.getMethod().getReturnType()),

160
org.springframework.integration/src/main/java/org/springframework/integration/aggregator/Resequencer.java
View File

@@ -17,135 +17,65 @@
package org.springframework.integration.aggregator;
import java.util.ArrayList;
import java.util.Iterator;
import java.util.Collection;
import java.util.Collections;
import java.util.Comparator;
import java.util.List;
import java.util.SortedSet;
import java.util.TreeSet;
import org.springframework.integration.channel.MessageChannelTemplate;
import org.springframework.integration.core.Message;
import org.springframework.util.CollectionUtils;
import org.springframework.integration.core.MessageChannel;
/**
* 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.
* This class implements all the strategy interfaces needed for a default
* resequencer.
*
* It is assumed that all messages have the same <code>sequence_size</code> header attribute
* and that the sequence numbers of the messages are successive, starting with
* 1 up to <code>sequenceSize</code>. Messages that do not satisfy this condition are
* considered out-of-sequence and thus rejected.
*
*
* Note: messages with the same sequence number will be treated as equivalent
* by this class (i.e. after a message with a given sequence number is received,
* further messages from within the same group, that have the same sequence number,
* will be ignored.
*
* @author Marius Bogoevici
* @author Alex Peters
* @author Iwein Fuld
* @author Dave Syer
* @since 2.0
*/
public class Resequencer extends AbstractMessageBarrierHandler<SortedSet<Message<?>>> {
public class Resequencer implements CorrelationStrategy, ReleaseStrategy, MessageGroupProcessor {
private volatile boolean releasePartialSequences = true;
private static final String LAST_RELEASED_SEQUENCE_NUMBER = "last.released.sequence.number";
private volatile SequenceNumberComparator sequenceNumberComparator = new SequenceNumberComparator();
private volatile boolean releasePartialSequences;
public Object getCorrelationKey(Message<?> message) {
// TODO: remove this (as its duplicating the default)
Object correlationKey = message.getHeaders().getCorrelationId();
return correlationKey;
}
public boolean canRelease(MessageGroup messages) {
if (releasePartialSequences) {
List<Message<?>> sorted = new ArrayList<Message<?>>(messages.getUnmarked());
Collections.sort(sorted, sequenceNumberComparator);
int head = sorted.get(sorted.size() - 1).getHeaders().getSequenceNumber();
int tail = sorted.get(0).getHeaders().getSequenceNumber() - 1;
return tail == messages.getMarked().size() && head - tail == sorted.size();
}
return messages.isComplete();
}
public void processAndSend(MessageGroup group, MessageChannelTemplate channelTemplate, MessageChannel outputChannel) {
Collection<Message<?>> messages = group.getUnmarked();
if (messages.size() > 0) {
List<Message<?>> sorted = new ArrayList<Message<?>>(messages);
Collections.sort(sorted, sequenceNumberComparator);
for (Message<?> message : sorted) {
channelTemplate.send(message, outputChannel);
}
}
}
public void setReleasePartialSequences(boolean releasePartialSequences) {
this.releasePartialSequences = releasePartialSequences;
}
@Override
public String getComponentType() {
return "resequencer";
}
@Override
protected MessageBarrier<SortedSet<Message<?>>> createMessageBarrier(Object correlationKey) {
MessageBarrier<SortedSet<Message<?>>> messageBarrier
= new MessageBarrier<SortedSet<Message<?>>>(new TreeSet<Message<?>>(new MessageSequenceComparator()), correlationKey);
messageBarrier.setAttribute(LAST_RELEASED_SEQUENCE_NUMBER, 0);
return messageBarrier;
private static class SequenceNumberComparator implements Comparator<Message<?>> {
public int compare(Message<?> o1, Message<?> o2) {
return o1.getHeaders().getSequenceNumber().compareTo(o2.getHeaders().getSequenceNumber());
}
}
@Override
protected void processBarrier(MessageBarrier<SortedSet<Message<?>>> barrier) {
if (hasReceivedAllMessages(barrier)) {
barrier.setComplete();
}
List<Message<?>> releasedMessages = releaseAvailableMessages(barrier);
if (!CollectionUtils.isEmpty(releasedMessages)) {
Message<?> lastMessage = releasedMessages.get(releasedMessages.size()-1);
if (lastMessage.getHeaders().getSequenceNumber().equals(lastMessage.getHeaders().getSequenceSize())) {
this.removeBarrier(barrier.getCorrelationKey());
}
this.sendReplies(releasedMessages, this.resolveReplyChannelFromMessage(releasedMessages.get(0)));
}
}
private boolean hasReceivedAllMessages(MessageBarrier<SortedSet<Message<?>>> barrier) {
if(barrier.getMessages().isEmpty()) {
return false;
}
int sequenceSize = barrier.getMessages().first().getHeaders().getSequenceSize();
int messagesCurrentlyInBarrier = barrier.getMessages().size();
Integer lastReleasedSequenceNumber = barrier.getAttribute(LAST_RELEASED_SEQUENCE_NUMBER);
return (lastReleasedSequenceNumber + messagesCurrentlyInBarrier == sequenceSize);
}
private List<Message<?>> releaseAvailableMessages(MessageBarrier<SortedSet<Message<?>>> barrier) {
if (this.releasePartialSequences || barrier.isComplete()) {
ArrayList<Message<?>> releasedMessages = new ArrayList<Message<?>>();
Iterator<Message<?>> it = barrier.getMessages().iterator();
Integer lastReleasedSequenceNumber = barrier.getAttribute(LAST_RELEASED_SEQUENCE_NUMBER);
while (it.hasNext()) {
Message<?> currentMessage = it.next();
if (lastReleasedSequenceNumber == currentMessage.getHeaders().getSequenceNumber() - 1) {
releasedMessages.add(currentMessage);
lastReleasedSequenceNumber = currentMessage.getHeaders().getSequenceNumber();
it.remove();
}
else {
break;
}
}
barrier.setAttribute(LAST_RELEASED_SEQUENCE_NUMBER, lastReleasedSequenceNumber);
return releasedMessages;
}
else {
return new ArrayList<Message<?>>();
}
}
@Override
protected boolean canAddMessage(Message<?> message, MessageBarrier<SortedSet<Message<?>>> barrier) {
if (!super.canAddMessage(message, barrier)) {
return false;
}
Integer lastReleasedSequenceNumber = barrier.getAttribute(LAST_RELEASED_SEQUENCE_NUMBER);
if (barrier.messages.contains(message)
|| lastReleasedSequenceNumber >= message.getHeaders().getSequenceNumber()) {
logger.debug("A message with the same sequence number has been already received: " + message);
return false;
}
if (message.getHeaders().getSequenceSize() < message.getHeaders().getSequenceNumber()) {
logger.debug("The message has a sequence number which is larger than the sequence size: "+ message);
return false;
}
if (!barrier.getMessages().isEmpty() &&
! message.getHeaders().getSequenceSize().equals(barrier.getMessages().first().getHeaders().getSequenceSize())) {
logger.debug("The message has a sequence size which is different from other messages handled so far: " + message
+ ", expected value is " + barrier.getMessages().first().getHeaders().getSequenceNumber());
return false;
}
return true;
}
}

41
org.springframework.integration/src/main/java/org/springframework/integration/aggregator/SequenceSizeCompletionStrategy.java
View File

@@ -1,41 +0,0 @@
/*
* 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.aggregator;
import java.util.Collection;
import org.springframework.integration.core.Message;
import org.springframework.util.CollectionUtils;
/**
* An implementation of {@link CompletionStrategy} that simply compares the
* current size of the message list to the expected 'sequenceSize' according to
* the first {@link Message} in the list.
*
* @author Mark Fisher
* @author Marius Bogoevici
*/
public class SequenceSizeCompletionStrategy implements CompletionStrategy {
public boolean isComplete(Collection<? extends Message<?>> messages) {
if (CollectionUtils.isEmpty(messages)) {
return false;
}
return messages.size() != 0 && (messages.size() >= messages.iterator().next().getHeaders().getSequenceSize());
}
}

22
org.springframework.integration/src/main/java/org/springframework/integration/aggregator/MessageGroupListener.java → org.springframework.integration/src/main/java/org/springframework/integration/aggregator/SequenceSizeReleaseStrategy.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2009 the original author or authors.
* 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.
@@ -13,19 +13,21 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.integration.aggregator;
import org.springframework.integration.core.Message;
/**
* Listener that can be configured with a MessageGroup to receive notifications on processing of (parts of) the group
* and exactly one notification when the whole group completes.
*
* @author Iwein Fuld
* An implementation of {@link ReleaseStrategy} that simply compares the
* current size of the message list to the expected 'sequenceSize'.
*
* @author Mark Fisher
* @author Marius Bogoevici
* @author Dave Syer
*/
public interface MessageGroupListener {
public class SequenceSizeReleaseStrategy implements ReleaseStrategy {
void onProcessingOf(Message<?>... processedMessages);
public boolean canRelease(MessageGroup messages) {
return messages.isComplete();
}
void onCompletionOf(Object correlationKey);
}

11
org.springframework.integration/src/main/java/org/springframework/integration/annotation/Aggregator.java
View File

@@ -22,7 +22,7 @@ import java.lang.annotation.Retention;
import java.lang.annotation.RetentionPolicy;
import java.lang.annotation.Target;
import org.springframework.integration.aggregator.AbstractMessageAggregator;
import org.springframework.integration.aggregator.CorrelatingMessageHandler;
/**
* Indicates that a method is capable of aggregating messages.
@@ -56,12 +56,12 @@ public @interface Aggregator {
/**
* timeout for sending results to the reply target (in milliseconds)
*/
long sendTimeout() default AbstractMessageAggregator.DEFAULT_SEND_TIMEOUT;
long sendTimeout() default CorrelatingMessageHandler.DEFAULT_SEND_TIMEOUT;
/**
* maximum time to wait for completion (in milliseconds)
*/
long timeout() default AbstractMessageAggregator.DEFAULT_TIMEOUT;
long timeout() default CorrelatingMessageHandler.DEFAULT_TIMEOUT;
/**
* indicates whether to send an incomplete aggregate on timeout
@@ -71,13 +71,14 @@ public @interface Aggregator {
/**
* interval for the task that checks for timed-out aggregates
*/
long reaperInterval() default AbstractMessageAggregator.DEFAULT_REAPER_INTERVAL;
long reaperInterval() default CorrelatingMessageHandler.DEFAULT_REAPER_INTERVAL;
/**
* maximum number of correlation IDs to maintain so that received messages
* may be recognized as belonging to an aggregate that has already completed
* or timed out
*/
int trackedCorrelationIdCapacity() default AbstractMessageAggregator.DEFAULT_TRACKED_CORRRELATION_ID_CAPACITY;
// TODO: remove / deal with tracked id capacity
int trackedCorrelationIdCapacity() default 42;
}

2
org.springframework.integration/src/main/java/org/springframework/integration/annotation/CompletionStrategy.java → org.springframework.integration/src/main/java/org/springframework/integration/annotation/ReleaseStrategy.java
View File

@@ -31,6 +31,6 @@ import java.lang.annotation.Target;
@Retention(RetentionPolicy.RUNTIME)
@Target(ElementType.METHOD)
@Documented
public @interface CompletionStrategy {
public @interface ReleaseStrategy {
}

51
org.springframework.integration/src/main/java/org/springframework/integration/config/annotation/AggregatorAnnotationPostProcessor.java
View File

@@ -18,18 +18,20 @@ package org.springframework.integration.config.annotation;
import java.lang.annotation.Annotation;
import java.lang.reflect.Method;
import java.util.concurrent.atomic.AtomicReference;
import org.springframework.beans.factory.ListableBeanFactory;
import org.springframework.core.annotation.AnnotationUtils;
import org.springframework.integration.aggregator.AbstractMessageAggregator;
import org.springframework.integration.aggregator.CompletionStrategyAdapter;
import org.springframework.integration.aggregator.MethodInvokingAggregator;
import org.springframework.integration.aggregator.ReleaseStrategyAdapter;
import org.springframework.integration.aggregator.CorrelatingMessageHandler;
import org.springframework.integration.aggregator.CorrelationStrategyAdapter;
import org.springframework.integration.aggregator.MethodInvokingMessageGroupProcessor;
import org.springframework.integration.annotation.Aggregator;
import org.springframework.integration.annotation.CompletionStrategy;
import org.springframework.integration.annotation.ReleaseStrategy;
import org.springframework.integration.annotation.CorrelationStrategy;
import org.springframework.integration.core.MessageChannel;
import org.springframework.integration.message.MessageHandler;
import org.springframework.integration.store.SimpleMessageStore;
import org.springframework.util.Assert;
import org.springframework.util.ReflectionUtils;
import org.springframework.util.StringUtils;
@@ -48,49 +50,54 @@ public class AggregatorAnnotationPostProcessor extends AbstractMethodAnnotationP
@Override
protected MessageHandler createHandler(Object bean, Method method, Aggregator annotation) {
MethodInvokingAggregator aggregator = new MethodInvokingAggregator(bean, method);
this.configureCompletionStrategy(bean, aggregator);
this.configureCorrelationStrategy(bean, aggregator);
MethodInvokingMessageGroupProcessor processor = new MethodInvokingMessageGroupProcessor(bean, method.getName());
ReleaseStrategyAdapter ReleaseStrategy = getReleaseStrategy(bean);
CorrelationStrategyAdapter correlationStrategy = getCorrelationStrategy(bean);
CorrelatingMessageHandler handler = new CorrelatingMessageHandler(new SimpleMessageStore(), correlationStrategy, ReleaseStrategy, processor);
String discardChannelName = annotation.discardChannel();
if (StringUtils.hasText(discardChannelName)) {
MessageChannel discardChannel = this.channelResolver.resolveChannelName(discardChannelName);
Assert.notNull(discardChannel, "failed to resolve discardChannel '" + discardChannelName + "'");
aggregator.setDiscardChannel(discardChannel);
handler.setDiscardChannel(discardChannel);
}
String outputChannelName = annotation.outputChannel();
if (StringUtils.hasText(outputChannelName)) {
aggregator.setOutputChannel(this.channelResolver.resolveChannelName(outputChannelName));
handler.setOutputChannel(this.channelResolver.resolveChannelName(outputChannelName));
}
aggregator.setSendTimeout(annotation.sendTimeout());
aggregator.setSendPartialResultOnTimeout(annotation.sendPartialResultsOnTimeout());
aggregator.setReaperInterval(annotation.reaperInterval());
aggregator.setTimeout(annotation.timeout());
aggregator.setTrackedCorrelationIdCapacity(annotation.trackedCorrelationIdCapacity());
aggregator.setBeanFactory(this.beanFactory);
aggregator.afterPropertiesSet();
return aggregator;
handler.setSendTimeout(annotation.sendTimeout());
handler.setSendPartialResultOnTimeout(annotation.sendPartialResultsOnTimeout());
handler.setReaperInterval(annotation.reaperInterval());
handler.setTimeout(annotation.timeout());
// handler.setTrackedCorrelationIdCapacity(annotation.trackedCorrelationIdCapacity());
handler.setBeanFactory(this.beanFactory);
handler.afterPropertiesSet();
return handler;
}
private void configureCompletionStrategy(final Object bean, final AbstractMessageAggregator aggregator) {
private ReleaseStrategyAdapter getReleaseStrategy(final Object bean) {
final AtomicReference<ReleaseStrategyAdapter> reference = new AtomicReference<ReleaseStrategyAdapter>();
ReflectionUtils.doWithMethods(bean.getClass(), new ReflectionUtils.MethodCallback() {
public void doWith(Method method) throws IllegalArgumentException, IllegalAccessException {
Annotation annotation = AnnotationUtils.getAnnotation(method, CompletionStrategy.class);
Annotation annotation = AnnotationUtils.getAnnotation(method, ReleaseStrategy.class);
if (annotation != null) {
aggregator.setCompletionStrategy(new CompletionStrategyAdapter(bean, method));
reference.set(new ReleaseStrategyAdapter(bean, method));
}
}
});
return reference.get();
}
private void configureCorrelationStrategy(final Object bean, final AbstractMessageAggregator aggregator) {
private CorrelationStrategyAdapter getCorrelationStrategy(final Object bean) {
final AtomicReference<CorrelationStrategyAdapter> reference = new AtomicReference<CorrelationStrategyAdapter>();
ReflectionUtils.doWithMethods(bean.getClass(), new ReflectionUtils.MethodCallback() {
public void doWith(Method method) throws IllegalArgumentException, IllegalAccessException {
Annotation annotation = AnnotationUtils.getAnnotation(method, CorrelationStrategy.class);
if (annotation != null) {
aggregator.setCorrelationStrategy(new CorrelationStrategyAdapter(bean, method));
reference.set(new CorrelationStrategyAdapter(bean, method));
}
}
});
return reference.get();
}
}

176
org.springframework.integration/src/main/java/org/springframework/integration/config/xml/AggregatorParser.java
View File

@@ -18,6 +18,7 @@ package org.springframework.integration.config.xml;
import org.w3c.dom.Element;
import org.springframework.beans.factory.config.BeanDefinition;
import org.springframework.beans.factory.parsing.BeanComponentDefinition;
import org.springframework.beans.factory.support.BeanDefinitionBuilder;
import org.springframework.beans.factory.support.BeanDefinitionReaderUtils;
@@ -25,116 +26,117 @@ import org.springframework.beans.factory.xml.ParserContext;
import org.springframework.util.StringUtils;
/**
* Parser for the <em>aggregator</em> element of the integration namespace.
* Registers the annotation-driven post-processors.
*
* Parser for the <em>aggregator</em> element of the integration namespace. Registers the annotation-driven
* post-processors.
*
* @author Marius Bogoevici
* @author Mark Fisher
* @author Oleg Zhurakousky
*/
public class AggregatorParser extends AbstractConsumerEndpointParser {
private static final String COMPLETION_STRATEGY_REF_ATTRIBUTE = "completion-strategy";
private static final String RELEASE_STRATEGY_REF_ATTRIBUTE = "release-strategy";
private static final String COMPLETION_STRATEGY_METHOD_ATTRIBUTE = "completion-strategy-method";
private static final String RELEASE_STRATEGY_METHOD_ATTRIBUTE = "release-strategy-method";
private static final String CORRELATION_STRATEGY_REF_ATTRIBUTE = "correlation-strategy";
private static final String CORRELATION_STRATEGY_REF_ATTRIBUTE = "correlation-strategy";
private static final String CORRELATION_STRATEGY_METHOD_ATTRIBUTE = "correlation-strategy-method";
private static final String CORRELATION_STRATEGY_METHOD_ATTRIBUTE = "correlation-strategy-method";
private static final String OUTPUT_CHANNEL_ATTRIBUTE = "output-channel";
private static final String OUTPUT_CHANNEL_ATTRIBUTE = "output-channel";
private static final String DISCARD_CHANNEL_ATTRIBUTE = "discard-channel";
private static final String DISCARD_CHANNEL_ATTRIBUTE = "discard-channel";
private static final String SEND_TIMEOUT_ATTRIBUTE = "send-timeout";
private static final String SEND_TIMEOUT_ATTRIBUTE = "send-timeout";
private static final String SEND_PARTIAL_RESULT_ON_TIMEOUT_ATTRIBUTE = "send-partial-result-on-timeout";
private static final String SEND_PARTIAL_RESULT_ON_TIMEOUT_ATTRIBUTE = "send-partial-result-on-timeout";
private static final String REAPER_INTERVAL_ATTRIBUTE = "reaper-interval";
private static final String REAPER_INTERVAL_ATTRIBUTE = "reaper-interval";
//private static final String TRACKED_CORRELATION_ID_CAPACITY_ATTRIBUTE = "tracked-correlation-id-capacity";
private static final String TIMEOUT_ATTRIBUTE = "timeout";
private static final String TIMEOUT_ATTRIBUTE = "timeout";
private static final String RELEASE_STRATEGY_PROPERTY = "releaseStrategy";
private static final String AUTO_STARTUP_ATTRIBUTE = "auto-startup";
private static final String COMPLETION_STRATEGY_PROPERTY = "completionStrategy";
private static final String CORRELATION_STRATEGY_PROPERTY = "correlationStrategy";
private static final String CORRELATION_STRATEGY_PROPERTY = "correlationStrategy";
@Override
protected BeanDefinitionBuilder parseHandler(Element element, ParserContext parserContext) {
BeanComponentDefinition innerHandlerDefinition = IntegrationNamespaceUtils.parseInnerHandlerDefinition(
element, parserContext);
String ref = element.getAttribute(REF_ATTRIBUTE);
BeanDefinitionBuilder builder = BeanDefinitionBuilder.genericBeanDefinition(
IntegrationNamespaceUtils.BASE_PACKAGE + ".aggregator.CorrelatingMessageHandler");
@Override
protected BeanDefinitionBuilder parseHandler(Element element, ParserContext parserContext) {
BeanComponentDefinition innerHandlerDefinition = IntegrationNamespaceUtils.parseInnerHandlerDefinition(element, parserContext);
String ref = element.getAttribute(REF_ATTRIBUTE);
BeanDefinitionBuilder builder;
BeanDefinitionBuilder processorBuilder = null;
if (innerHandlerDefinition != null || StringUtils.hasText(ref)) {
processorBuilder = BeanDefinitionBuilder.genericBeanDefinition(
IntegrationNamespaceUtils.BASE_PACKAGE + ".aggregator.MethodInvokingMessageGroupProcessor");
if (innerHandlerDefinition != null) {
processorBuilder.addConstructorArgValue(innerHandlerDefinition);
}
else {
processorBuilder.addConstructorArgReference(ref);
}
builder.addConstructorArgValue(processorBuilder.getBeanDefinition());
}
else {
builder.addConstructorArgValue(BeanDefinitionBuilder.genericBeanDefinition(
IntegrationNamespaceUtils.BASE_PACKAGE + ".aggregator.DefaultAggregatingMessageGroupProcessor")
.getBeanDefinition());
}
builder = BeanDefinitionBuilder.genericBeanDefinition(
IntegrationNamespaceUtils.BASE_PACKAGE + ".aggregator.CorrelatingMessageHandler");
BeanDefinitionBuilder processorBuilder = null;
if (StringUtils.hasText(element.getAttribute(METHOD_ATTRIBUTE))) {
String method = element.getAttribute(METHOD_ATTRIBUTE);
processorBuilder.getRawBeanDefinition().getConstructorArgumentValues()
.addGenericArgumentValue(method, "java.lang.String");
}
if (innerHandlerDefinition != null || StringUtils.hasText(ref)) {
processorBuilder = BeanDefinitionBuilder.genericBeanDefinition(
IntegrationNamespaceUtils.BASE_PACKAGE + ".aggregator.MethodInvokingMessageGroupProcessor");
builder.addConstructorArgValue(processorBuilder.getBeanDefinition());
} else {
builder.addConstructorArgValue(BeanDefinitionBuilder.genericBeanDefinition(
IntegrationNamespaceUtils.BASE_PACKAGE + ".aggregator.DefaultAggregatingMessageGroupProcessor").getBeanDefinition());
}
IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, DISCARD_CHANNEL_ATTRIBUTE);
IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, OUTPUT_CHANNEL_ATTRIBUTE);
IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, SEND_TIMEOUT_ATTRIBUTE);
IntegrationNamespaceUtils.setValueIfAttributeDefined(
builder, element, SEND_PARTIAL_RESULT_ON_TIMEOUT_ATTRIBUTE);
IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, REAPER_INTERVAL_ATTRIBUTE);
// IntegrationNamespaceUtils.setValueIfAttributeDefined(builder,
// element, TRACKED_CORRELATION_ID_CAPACITY_ATTRIBUTE);
IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, AUTO_STARTUP_ATTRIBUTE);
IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, TIMEOUT_ATTRIBUTE);
this.injectPropertyWithBean(COMPLETION_STRATEGY_REF_ATTRIBUTE, COMPLETION_STRATEGY_METHOD_ATTRIBUTE,
COMPLETION_STRATEGY_PROPERTY, "CompletionStrategyAdapter", element, builder, parserContext);
this.injectPropertyWithBean(CORRELATION_STRATEGY_REF_ATTRIBUTE, CORRELATION_STRATEGY_METHOD_ATTRIBUTE,
CORRELATION_STRATEGY_PROPERTY, "CorrelationStrategyAdapter", element, builder, parserContext);
return builder;
}
if (innerHandlerDefinition != null) {
processorBuilder.addConstructorArgValue(innerHandlerDefinition);
} else {
if (StringUtils.hasText(ref)) {
processorBuilder.addConstructorArgReference(ref);
}
}
if (StringUtils.hasText(element.getAttribute(METHOD_ATTRIBUTE))) {
String method = element.getAttribute(METHOD_ATTRIBUTE);
processorBuilder.getRawBeanDefinition().getConstructorArgumentValues().addGenericArgumentValue(method, "java.lang.String");
}
private void injectPropertyWithBean(String beanRefAttribute, String methodRefAttribute, String beanProperty,
String adapterClass, Element element, BeanDefinitionBuilder builder, ParserContext parserContext) {
final String beanRef = element.getAttribute(beanRefAttribute);
final String beanMethod = element.getAttribute(methodRefAttribute);
if (StringUtils.hasText(beanRef)) {
if (StringUtils.hasText(beanMethod)) {
String adapterBeanName = this.createAdapter(beanRef, beanMethod, adapterClass, parserContext);
builder.addPropertyReference(beanProperty, adapterBeanName);
}
else {
builder.addPropertyReference(beanProperty, beanRef);
}
}
}
IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element,
DISCARD_CHANNEL_ATTRIBUTE);
IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element,
OUTPUT_CHANNEL_ATTRIBUTE);
IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element,
SEND_TIMEOUT_ATTRIBUTE);
IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element,
SEND_PARTIAL_RESULT_ON_TIMEOUT_ATTRIBUTE);
IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element,
REAPER_INTERVAL_ATTRIBUTE);
// IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, TRACKED_CORRELATION_ID_CAPACITY_ATTRIBUTE);
IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "auto-startup");
IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, TIMEOUT_ATTRIBUTE);
this.injectPropertyWithBean(RELEASE_STRATEGY_REF_ATTRIBUTE,
RELEASE_STRATEGY_METHOD_ATTRIBUTE, RELEASE_STRATEGY_PROPERTY,
"ReleaseStrategyAdapter", element, builder, parserContext);
this.injectPropertyWithBean(CORRELATION_STRATEGY_REF_ATTRIBUTE,
CORRELATION_STRATEGY_METHOD_ATTRIBUTE, CORRELATION_STRATEGY_PROPERTY,
"CorrelationStrategyAdapter", element, builder, parserContext);
return builder;
}
private String createAdapter(String ref, String method, String unqualifiedClassName, ParserContext parserContext) {
BeanDefinitionBuilder builder = BeanDefinitionBuilder.genericBeanDefinition(
IntegrationNamespaceUtils.BASE_PACKAGE + ".aggregator." + unqualifiedClassName);
builder.addConstructorArgReference(ref);
builder.getRawBeanDefinition().getConstructorArgumentValues()
.addGenericArgumentValue(method, "java.lang.String");
return BeanDefinitionReaderUtils.registerWithGeneratedName(
builder.getBeanDefinition(), parserContext.getRegistry());
}
private void injectPropertyWithBean(String beanRefAttribute, String methodRefAttribute,
String beanProperty, String adapterClass, Element element,
BeanDefinitionBuilder builder, ParserContext parserContext) {
final String beanRef = element.getAttribute(beanRefAttribute);
final String beanMethod = element.getAttribute(methodRefAttribute);
if (StringUtils.hasText(beanRef)) {
if (StringUtils.hasText(beanMethod)) {
String adapterBeanName = this.createAdapter(beanRef, beanMethod, adapterClass,
parserContext);
builder.addPropertyReference(beanProperty, adapterBeanName);
} else {
builder.addPropertyReference(beanProperty, beanRef);
}
}
}
private String createAdapter(String ref, String method, String unqualifiedClassName,
ParserContext parserContext) {
BeanDefinitionBuilder builder = BeanDefinitionBuilder.genericBeanDefinition(
IntegrationNamespaceUtils.BASE_PACKAGE + ".aggregator." + unqualifiedClassName);
builder.addConstructorArgReference(ref);
builder.getRawBeanDefinition().getConstructorArgumentValues().addGenericArgumentValue(method, "java.lang.String");
return BeanDefinitionReaderUtils.registerWithGeneratedName(builder.getBeanDefinition(),
parserContext.getRegistry());
}
}

48
org.springframework.integration/src/main/java/org/springframework/integration/config/xml/ResequencerParser.java
View File

@@ -16,12 +16,11 @@
package org.springframework.integration.config.xml;
import org.w3c.dom.Element;
import org.springframework.beans.factory.support.BeanDefinitionBuilder;
import org.springframework.beans.factory.support.BeanDefinitionReaderUtils;
import org.springframework.beans.factory.xml.ParserContext;
import org.springframework.util.StringUtils;
import org.w3c.dom.Element;
/**
* Parser for the &lt;resequencer&gt; element.
@@ -32,38 +31,57 @@ public class ResequencerParser extends AbstractConsumerEndpointParser {
@Override
protected BeanDefinitionBuilder parseHandler(Element element, ParserContext parserContext) {
BeanDefinitionBuilder builder = BeanDefinitionBuilder.genericBeanDefinition(
BeanDefinitionBuilder builder = BeanDefinitionBuilder
.genericBeanDefinition(IntegrationNamespaceUtils.BASE_PACKAGE + ".aggregator.CorrelatingMessageHandler");
BeanDefinitionBuilder processorBuilder = BeanDefinitionBuilder.genericBeanDefinition(
IntegrationNamespaceUtils.BASE_PACKAGE + ".aggregator.Resequencer");
IntegrationNamespaceUtils.setValueIfAttributeDefined(processorBuilder, element, "release-partial-sequences");
// TODO: expose message store as an XML attribute
builder.addConstructorArgValue(BeanDefinitionBuilder.genericBeanDefinition(
IntegrationNamespaceUtils.BASE_PACKAGE + ".store.SimpleMessageStore").getBeanDefinition());
String correlationStrategyRef = getCorrelationStrategyRef(element, parserContext);
String processorRef = BeanDefinitionReaderUtils.registerWithGeneratedName(processorBuilder
.getBeanDefinition(), parserContext.getRegistry());
if (correlationStrategyRef != null) {
builder.addConstructorArgReference(correlationStrategyRef);
}
else {
builder.addConstructorArgReference(processorRef);
}
// Completion strategy
builder.addConstructorArgReference(processorRef);
// Message group processor
builder.addConstructorArgReference(processorRef);
IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "discard-channel");
this.configureCorrelationStrategy(builder, element, parserContext);
IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "send-timeout");
IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "release-partial-sequences");
IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "send-partial-result-on-timeout");
IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "reaper-interval");
IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "tracked-correlation-id-capacity");
// IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "tracked-correlation-id-capacity");
IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "timeout");
IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "auto-startup");
return builder;
}
private void configureCorrelationStrategy(BeanDefinitionBuilder builder, Element element, ParserContext parserContext) {
private String getCorrelationStrategyRef(Element element, ParserContext parserContext) {
String ref = element.getAttribute("correlation-strategy");
String method = element.getAttribute("correlation-strategy-method");
String correlationStrategyProperty = "correlationStrategy";
if (StringUtils.hasText(ref)) {
if (StringUtils.hasText(method)) {
BeanDefinitionBuilder adapterBuilder = BeanDefinitionBuilder.genericBeanDefinition(
IntegrationNamespaceUtils.BASE_PACKAGE + ".aggregator.CorrelationStrategyAdapter");
BeanDefinitionBuilder adapterBuilder = BeanDefinitionBuilder
.genericBeanDefinition(IntegrationNamespaceUtils.BASE_PACKAGE
+ ".aggregator.CorrelationStrategyAdapter");
adapterBuilder.addConstructorArgReference(ref);
adapterBuilder.getRawBeanDefinition().getConstructorArgumentValues().addGenericArgumentValue(method, "java.lang.String");
String adapterBeanName = BeanDefinitionReaderUtils.registerWithGeneratedName(
adapterBuilder.getBeanDefinition(), parserContext.getRegistry());
builder.addPropertyReference(correlationStrategyProperty, adapterBeanName);
adapterBuilder.getRawBeanDefinition().getConstructorArgumentValues().addGenericArgumentValue(method,
"java.lang.String");
String adapterBeanName = BeanDefinitionReaderUtils.registerWithGeneratedName(adapterBuilder
.getBeanDefinition(), parserContext.getRegistry());
return adapterBeanName;
}
else {
builder.addPropertyReference(correlationStrategyProperty, ref);
return ref;
}
}
return null;
}
}

60
org.springframework.integration/src/main/java/org/springframework/integration/store/MessageStore.java
View File

@@ -20,11 +20,11 @@ import java.util.Collection;
import java.util.UUID;
import org.springframework.integration.core.Message;
import org.springframework.integration.core.MessageHeaders;
/**
* Strategy interface for storing and retrieving messages. The interface mimics
* the semantics for REST for the methods named after REST operations. This is
* helpful when mapping to a RESTful API.
* Strategy interface for storing and retrieving messages. The interface mimics the semantics for REST for the methods
* named after REST operations. This is helpful when mapping to a RESTful API.
*
* @author Mark Fisher
* @author Iwein Fuld
@@ -33,43 +33,40 @@ import org.springframework.integration.core.Message;
*/
public interface MessageStore {
String PROCESSED = MessageHeaders.PREFIX + "processed";
/**
* Return the Message with the given id, or <i>null</i> if no Message with
* that id exists in the MessageStore.
* Return the Message with the given id, or <i>null</i> if no Message with that id exists in the MessageStore.
*/
Message<?> get(UUID id);
/**
* Put the provided Message into the MessageStore. The store may need to
* mutate the message internally, and if it does then the return value can
* be different than the input. The id of the return value will be used as
* an index so that the {@link #get(UUID)} and {@link #delete(Object)}
* behave properly. Since messages are immutable, putting the same message
* more than once is a no-op.
* Put the provided Message into the MessageStore. The store may need to mutate the message internally, and if it
* does then the return value can be different than the input. The id of the return value will be used as an index
* so that the {@link #get(UUID)} and {@link #delete(Object)} behave properly. Since messages are immutable, putting
* the same message more than once is a no-op.
*
* @return the message that was stored
*/
<T> Message<T> put(Message<T> message);
/**
* Remove the Message with the given id from the MessageStore, if present,
* and return it. If no Message with that id is present in the store, this
* will return <i>null</i>.
* Remove the Message with the given id from the MessageStore, if present, and return it. If no Message with that id
* is present in the store, this will return <i>null</i>.
*/
Message<?> delete(UUID id);
/**
* Return all Messages currently in the MessageStore that were stored using
* {@link #put(Object, Message)} or {@link #put(Object, Collection)} with
* this correlation id.
* Return all Messages currently in the MessageStore that were stored using {@link #put(Object, Message)} or
* {@link #put(Object, Collection)} with this correlation id.
*
* @see org.springframework.integration.core.MessageHeaders#getCorrelationId()
*/
Collection<Message<?>> list(Object correlationId);
/**
* Store a message with an association to a correlation id. This can be used
* to group messages together instead of storing them just under their id.
* Store a message with an association to a correlation id. This can be used to group messages together instead of
* storing them just under their id.
*
* @param correlationId the correlation id to store the message under
* @param message a message
@@ -77,28 +74,17 @@ public interface MessageStore {
void put(Object correlationId, Message<?> message);
/**
* Store a group of message with an association to a correlation id.
* Mark a message from the association with this correlation id. If the message was previously added using
* {@link #put(Object, Message)} then it will be removed and re-inserted with the {@value #PROCESSED} flag set in
* the headers.
*
* @param correlationId the correlation id to store the message under
* @param messages a collection of messages
*
* @see MessageStore#put(UUID, Message)
* @param correlationId the correlation id to mark the message under
*/
void put(Object correlationId, Collection<Message<?>> messages);
Message<?> mark(Object correlationId, UUID messageId);
/**
* Delete a message from the association with this correlation id. If the
* message was stored under through {@link #put(Message)} as well, then it
* is still accessible via {@link #get(UUID)}.
*
* @param correlationId the correlation id to delete all messages under
*/
Message<?> delete(Object correlationId, UUID messageId);
/**
* Delete all the messages from the association with this correlation id. If
* the messages were stored under their id through {@link #put(Message)}
* they are still accessible via {@link #get(UUID)}.
* Delete all the messages from the association with this correlation id. If the messages were stored under their id
* through {@link #put(Message)} they are still accessible via {@link #get(UUID)}.
*
* @param correlationId the correlation id to delete all messages under
*/

10
org.springframework.integration/src/main/java/org/springframework/integration/store/SimpleMessageStore.java
View File

@@ -26,6 +26,7 @@ import java.util.concurrent.ConcurrentMap;
import org.springframework.integration.core.Message;
import org.springframework.integration.core.MessagingException;
import org.springframework.integration.message.MessageBuilder;
import org.springframework.integration.util.UpperBound;
import org.springframework.util.Assert;
@@ -101,15 +102,11 @@ public class SimpleMessageStore implements MessageStore {
return new HashSet<Message<?>>(collection);
}
public void put(Object correlationId, Collection<Message<?>> messages) {
getMessagesInternal(correlationId).addAll(messages);
}
public void put(Object correlationId, Message<?> message) {
getMessagesInternal(correlationId).add(message);
}
public Message<?> delete(Object correlationId, UUID messageId) {
public Message<?> mark(Object correlationId, UUID messageId) {
if (!correlationToMessage.containsKey(correlationId)) {
return null;
}
@@ -119,9 +116,10 @@ public class SimpleMessageStore implements MessageStore {
Message<?> message = (Message<?>) iterator.next();
if (message.getHeaders().getId().equals(messageId)) {
iterator.remove();
result = message;
result = MessageBuilder.fromMessage(message).setHeader(PROCESSED, true).build();
}
}
messages.add(result);
return result;
}

6
org.springframework.integration/src/main/resources/org/springframework/integration/config/xml/spring-integration-2.0.xsd
View File

@@ -1422,7 +1422,7 @@
</xsd:annotation>
<xsd:complexContent>
<xsd:extension base="innerEndpointDefinitionAware">
<xsd:attribute name="completion-strategy" type="xsd:string">
<xsd:attribute name="release-strategy" type="xsd:string">
<xsd:annotation>
<xsd:appinfo>
<tool:annotation kind="ref">
@@ -1431,11 +1431,11 @@
</xsd:appinfo>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="completion-strategy-method" type="xsd:string">
<xsd:attribute name="release-strategy-method" type="xsd:string">
<xsd:annotation>
<xsd:appinfo>
<tool:annotation>
<tool:expected-method type-ref="@completion-strategy"/>
<tool:expected-method type-ref="@release-strategy"/>
</tool:annotation>
</xsd:appinfo>
</xsd:annotation>