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

Merge pull request #152 from olegz/INT-2182

Browse Source 
INT-2182
This commit is contained in:
Mark Fisher
2011-10-25 11:58:08 -05:00
parent 772599cf99 ddeab16394
commit 24714fa711
48 changed files with 550 additions and 731 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -63,7 +63,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.getUnmarked()) {
for (Message<?> message : group.getMessages()) {
MessageHeaders currentHeaders = message.getHeaders();
for (String key : currentHeaders.keySet()) {
if (MessageHeaders.ID.equals(key) || MessageHeaders.TIMESTAMP.equals(key)

10
spring-integration-core/src/main/java/org/springframework/integration/aggregator/AbstractCorrelatingMessageHandler.java
View File

@@ -85,13 +85,6 @@ public abstract class AbstractCorrelatingMessageHandler extends AbstractMessageH
private final Object correlationLocksMonitor = new Object();
private final ConcurrentMap<Object, Object> locks = new ConcurrentHashMap<Object, Object>();
protected volatile boolean keepReleasedMessages = true;
public void setKeepReleasedMessages(boolean keepReleasedMessages) {
this.keepReleasedMessages = keepReleasedMessages;
}
public AbstractCorrelatingMessageHandler(MessageGroupProcessor processor, MessageGroupStore store,
CorrelationStrategy correlationStrategy, ReleaseStrategy releaseStrategy) {
@@ -288,7 +281,7 @@ public abstract class AbstractCorrelatingMessageHandler extends AbstractMessageH
logger.debug("Discarding messages of partially complete group with key ["
+ correlationKey + "] to: " + discardChannel);
}
for (Message<?> message : group.getUnmarked()) {
for (Message<?> message : group.getMessages()) {
discardChannel.send(message);
}
}
@@ -307,6 +300,7 @@ public abstract class AbstractCorrelatingMessageHandler extends AbstractMessageH
if (logger.isDebugEnabled()) {
logger.debug("Completing group with correlationKey [" + correlationKey + "]");
}
Object result = outputProcessor.processMessageGroup(group);
Collection<Message<?>> partialSequence = null;
if (result instanceof Collection<?>) {

13
spring-integration-core/src/main/java/org/springframework/integration/aggregator/AggregatingMessageHandler.java
View File

@@ -31,7 +31,6 @@ public class AggregatingMessageHandler extends AbstractCorrelatingMessageHandler
private volatile boolean expireGroupsUponCompletion = false;
public AggregatingMessageHandler(MessageGroupProcessor processor, MessageGroupStore store,
CorrelationStrategy correlationStrategy, ReleaseStrategy releaseStrategy) {
super(processor, store, correlationStrategy, releaseStrategy);
@@ -71,16 +70,8 @@ public class AggregatingMessageHandler extends AbstractCorrelatingMessageHandler
remove(messageGroup);
}
else {
if (this.keepReleasedMessages){
messageStore.markMessageGroup(messageGroup);
}
else {
for (Message<?> message : messageGroup.getMarked()) {
this.messageStore.removeMessageFromGroup(messageGroup.getGroupId(), message);
}
for (Message<?> message : messageGroup.getUnmarked()) {
this.messageStore.removeMessageFromGroup(messageGroup.getGroupId(), message);
}
for (Message<?> message : messageGroup.getMessages()) {
this.messageStore.removeMessageFromGroup(messageGroup.getGroupId(), message);
}
}
}

9
spring-integration-core/src/main/java/org/springframework/integration/aggregator/CorrelatingMessageBarrier.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2010 the original author or authors.
* Copyright 2002-2011 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
@@ -40,6 +40,7 @@ import org.springframework.integration.store.SimpleMessageStore;
* for each correlation key.
*
* @author Iwein Fuld
* @author Oleg Zhurakousky
*
* @see CorrelatingMessageHandler
*/
@@ -102,9 +103,9 @@ public class CorrelatingMessageBarrier extends AbstractMessageHandler implements
if (releaseStrategy.canRelease(group)) {
Message<?> nextMessage = null;
Iterator<Message<?>> unmarked = group.getUnmarked().iterator();
if (unmarked.hasNext()) {
nextMessage = unmarked.next();
Iterator<Message<?>> messages = group.getMessages().iterator();
if (messages.hasNext()) {
nextMessage = messages.next();
store.removeMessageFromGroup(key, nextMessage);
if (log.isDebugEnabled()) {
log.debug(String.format("Released message for key [%s]: %s.", key, nextMessage));

4
spring-integration-core/src/main/java/org/springframework/integration/aggregator/DefaultAggregatingMessageGroupProcessor.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2010 the original author or authors.
* Copyright 2002-2011 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.
@@ -38,7 +38,7 @@ public class DefaultAggregatingMessageGroupProcessor extends AbstractAggregating
@Override
protected final Object aggregatePayloads(MessageGroup group, Map<String, Object> headers) {
Collection<Message<?>> messages = group.getUnmarked();
Collection<Message<?>> messages = group.getMessages();
Assert.notEmpty(messages, this.getClass().getSimpleName() + " cannot process empty message groups");
List<Object> payloads = new ArrayList<Object>(messages.size());
for (Message<?> message : messages) {

6
spring-integration-core/src/main/java/org/springframework/integration/aggregator/ExpressionEvaluatingMessageGroupProcessor.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2010 the original author or authors.
* Copyright 2002-2011 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.
@@ -54,12 +54,12 @@ public class ExpressionEvaluatingMessageGroupProcessor extends AbstractAggregati
}
/**
* Evaluate the expression provided on the unmarked messages (a collection) in the group, and delegate to the
* Evaluate the expression provided on the messages (a collection) in the group, and delegate to the
* {@link MessagingTemplate} to send downstream.
*/
@Override
protected Object aggregatePayloads(MessageGroup group, Map<String, Object> headers) {
return processor.process(group.getUnmarked());
return processor.process(group.getMessages());
}
}

6
spring-integration-core/src/main/java/org/springframework/integration/aggregator/ExpressionEvaluatingReleaseStrategy.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2008 the original author or authors.
* Copyright 2002-2011 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.
@@ -31,11 +31,11 @@ public class ExpressionEvaluatingReleaseStrategy extends ExpressionEvaluatingMes
}
/**
* Evaluate the expression provided on the unmarked messages (a collection) in the group and return the result (must
* Evaluate the expression provided on the messages (a collection) in the group and return the result (must
* be boolean).
*/
public boolean canRelease(MessageGroup messages) {
return ((Boolean) process(messages.getUnmarked())).booleanValue();
return ((Boolean) process(messages.getMessages())).booleanValue();
}
}

11
spring-integration-core/src/main/java/org/springframework/integration/aggregator/MessageCountReleaseStrategy.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2010 the original author or authors.
* Copyright 2002-2011 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
@@ -19,6 +19,7 @@ import org.springframework.integration.store.MessageGroup;
* A {@link ReleaseStrategy} that releases only the first <code>n</code> messages, where <code>n</code> is a threshold.
*
* @author Dave Syer
* @author Oleg Zhurakousky
*
*/
public class MessageCountReleaseStrategy implements ReleaseStrategy {
@@ -41,12 +42,12 @@ public class MessageCountReleaseStrategy implements ReleaseStrategy {
}
/**
* 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.
* Release the group if it has more messages than the threshold and has not previously been released.
* 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;
return group.size() >= threshold;
}
}

4
spring-integration-core/src/main/java/org/springframework/integration/aggregator/MethodInvokingMessageGroupProcessor.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2010 the original author or authors.
* Copyright 2002-2011 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.
@@ -79,7 +79,7 @@ public class MethodInvokingMessageGroupProcessor extends AbstractAggregatingMess
@Override
protected final Object aggregatePayloads(MessageGroup group, Map<String, Object> headers) {
final Collection<Message<?>> messagesUpForProcessing = group.getUnmarked();
final Collection<Message<?>> messagesUpForProcessing = group.getMessages();
return this.processor.process(messagesUpForProcessing, headers);
}

4
spring-integration-core/src/main/java/org/springframework/integration/aggregator/MethodInvokingReleaseStrategy.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2010 the original author or authors.
* Copyright 2002-2011 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.
@@ -51,7 +51,7 @@ public class MethodInvokingReleaseStrategy implements ReleaseStrategy {
}
public boolean canRelease(MessageGroup messages) {
return this.adapter.process(messages.getUnmarked(), null);
return this.adapter.process(messages.getMessages(), null);
}
}

6
spring-integration-core/src/main/java/org/springframework/integration/aggregator/PassThroughMessageGroupProcessor.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2010 the original author or authors.
* Copyright 2002-2011 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
@@ -16,7 +16,7 @@ package org.springframework.integration.aggregator;
import org.springframework.integration.store.MessageGroup;
/**
* This implementation of MessageGroupProcessor will return all unmarked messages inside the group.
* This implementation of MessageGroupProcessor will return all messages inside the group.
* This is useful if there is no requirement to process the messages, but they should just be
* blocked as a group until their ReleaseStrategy lets them pass through.
*
@@ -26,7 +26,7 @@ import org.springframework.integration.store.MessageGroup;
public class PassThroughMessageGroupProcessor implements MessageGroupProcessor {
public Object processMessageGroup(MessageGroup group) {
return group.getUnmarked();
return group.getMessages();
}
}

4
spring-integration-core/src/main/java/org/springframework/integration/aggregator/ResequencingMessageGroupProcessor.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2010 the original author or authors.
* Copyright 2002-2011 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
@@ -40,7 +40,7 @@ public class ResequencingMessageGroupProcessor implements MessageGroupProcessor
}
public Object processMessageGroup(MessageGroup group) {
Collection<Message<?>> messages = group.getUnmarked();
Collection<Message<?>> messages = group.getMessages();
if (messages.size() > 0) {
List<Message<?>> sorted = new ArrayList<Message<?>>(messages);

15
spring-integration-core/src/main/java/org/springframework/integration/aggregator/ResequencingMessageHandler.java
View File

@@ -48,7 +48,7 @@ public class ResequencingMessageHandler extends AbstractCorrelatingMessageHandle
@Override
protected void afterRelease(MessageGroup messageGroup, Collection<Message<?>> completedMessages) {
int size = messageGroup.getUnmarked().size() + messageGroup.getMarked().size();
int size = messageGroup.getMessages().size();
int sequenceSize = 0;
Message<?> message = messageGroup.getOne();
if (message != null){
@@ -62,17 +62,8 @@ public class ResequencingMessageHandler extends AbstractCorrelatingMessageHandle
if (completedMessages != null){
int lastReleasedSequenceNumber = this.findLastReleasedSequenceNumber(messageGroup.getGroupId(), completedMessages);
messageStore.setLastReleasedSequenceNumberForGroup(messageGroup.getGroupId(), lastReleasedSequenceNumber);
if (this.keepReleasedMessages){
Object id = messageGroup.getGroupId();
for (Message<?> msg : completedMessages) {
messageStore.markMessageFromGroup(id, msg);
}
}
else {
for (Message<?> msg : completedMessages) {
this.messageStore.removeMessageFromGroup(messageGroup.getGroupId(), msg);
}
for (Message<?> msg : completedMessages) {
this.messageStore.removeMessageFromGroup(messageGroup.getGroupId(), msg);
}
}
}

8
spring-integration-core/src/main/java/org/springframework/integration/aggregator/SequenceSizeReleaseStrategy.java
View File

@@ -67,14 +67,14 @@ public class SequenceSizeReleaseStrategy implements ReleaseStrategy {
boolean canRelease = false;
Collection<Message<?>> unmarked = messageGroup.getUnmarked();
Collection<Message<?>> messages = messageGroup.getMessages();
if (releasePartialSequences && !unmarked.isEmpty()) {
if (releasePartialSequences && !messages.isEmpty()) {
if (logger.isTraceEnabled()) {
logger.trace("Considering partial release of group [" + messageGroup + "]");
}
List<Message<?>> sorted = new ArrayList<Message<?>>(unmarked);
List<Message<?>> sorted = new ArrayList<Message<?>>(messages);
Collections.sort(sorted, comparator);
int nextSequenceNumber = sorted.get(0).getHeaders().getSequenceNumber();
@@ -85,7 +85,7 @@ public class SequenceSizeReleaseStrategy implements ReleaseStrategy {
}
}
else {
int size = messageGroup.getUnmarked().size();
int size = messages.size();
if (size == 0){
canRelease = true;

6
spring-integration-core/src/main/java/org/springframework/integration/aggregator/TimeoutCountSequenceSizeReleaseStrategy.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2008 the original author or authors.
* Copyright 2002-2011 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
@@ -60,7 +60,7 @@ public class TimeoutCountSequenceSizeReleaseStrategy implements ReleaseStrategy
public boolean canRelease(MessageGroup messages) {
long elapsedTime = System.currentTimeMillis() - findEarliestTimestamp(messages);
return messages.isComplete() || messages.getUnmarked().size() >= threshold || elapsedTime > timeout;
return messages.isComplete() || messages.getMessages().size() >= threshold || elapsedTime > timeout;
}
/**
@@ -69,7 +69,7 @@ public class TimeoutCountSequenceSizeReleaseStrategy implements ReleaseStrategy
*/
private long findEarliestTimestamp(MessageGroup messages) {
long result = Long.MAX_VALUE;
for (Message<?> message : messages.getUnmarked()) {
for (Message<?> message : messages.getMessages()) {
long timestamp = message.getHeaders().getTimestamp();
if (timestamp < result) {
result = timestamp;

216
spring-integration-core/src/main/java/org/springframework/integration/store/AbstractKeyValueMessageStore.java
View File

@@ -19,9 +19,13 @@ package org.springframework.integration.store;
import java.util.ArrayList;
import java.util.Collection;
import java.util.Iterator;
import java.util.Map;
import java.util.UUID;
import org.springframework.beans.DirectFieldAccessor;
import org.springframework.integration.Message;
import org.springframework.integration.MessageHeaders;
import org.springframework.integration.support.MessageBuilder;
import org.springframework.jmx.export.annotation.ManagedAttribute;
import org.springframework.util.Assert;
@@ -36,17 +40,18 @@ public abstract class AbstractKeyValueMessageStore extends AbstractMessageGroupS
protected static final String MESSAGE_KEY_PREFIX = "MESSAGE_";
protected static final String MESSAGE_GROUP_KEY_PREFIX = "MESSAGE_GROUP_";
protected static final String CREATED_DATE = "CREATED_DATE";
// MessageStore methods
public Message<?> getMessage(UUID id) {
Assert.notNull(id, "'id' must not be null");
Object message = this.doRetrieve(MESSAGE_KEY_PREFIX + id);
if (message != null) {
Assert.isInstanceOf(Message.class, message);
}
return (Message<?>) message;
Message<?> message = this.getRawMessage(id);
if (message != null){
return this.normalizeMessage(message);
}
return null;
}
@SuppressWarnings("unchecked")
@@ -54,7 +59,7 @@ public abstract class AbstractKeyValueMessageStore extends AbstractMessageGroupS
Assert.notNull(message, "'message' must not be null");
UUID messageId = message.getHeaders().getId();
this.doStore(MESSAGE_KEY_PREFIX + messageId, message);
return (Message<T>) this.getMessage(messageId);
return (Message<T>) this.getRawMessage(messageId);
}
public Message<?> removeMessage(UUID id) {
@@ -79,30 +84,9 @@ public abstract class AbstractKeyValueMessageStore extends AbstractMessageGroupS
* Will create a new instance of SimpleMessageGroup if necessary.
*/
public MessageGroup getMessageGroup(Object groupId) {
Assert.notNull(groupId, "'groupId' must not be null");
Object mgm = this.doRetrieve(MESSAGE_GROUP_KEY_PREFIX + groupId);
if (mgm != null) {
Assert.isInstanceOf(MessageGroupMetadata.class, mgm);
MessageGroupMetadata messageGroupMetadata = (MessageGroupMetadata) mgm;
ArrayList<Message<?>> markedMessages = new ArrayList<Message<?>>();
for (UUID uuid : messageGroupMetadata.getMarkedMessageIds()) {
markedMessages.add(this.getMessage(uuid));
}
ArrayList<Message<?>> unmarkedMessages = new ArrayList<Message<?>>();
for (UUID uuid : messageGroupMetadata.getUnmarkedMessageIds()) {
unmarkedMessages.add(this.getMessage(uuid));
}
SimpleMessageGroup messageGroup = new SimpleMessageGroup(unmarkedMessages, markedMessages,
groupId, messageGroupMetadata.getTimestamp(), messageGroupMetadata.isComplete());
if (messageGroupMetadata.getLastReleasedMessageSequenceNumber() > 0) {
messageGroup.setLastReleasedMessageSequenceNumber(messageGroupMetadata.getLastReleasedMessageSequenceNumber());
}
return messageGroup;
}
else {
return new SimpleMessageGroup(groupId);
}
return this.buildMessageGroup(groupId, false);
}
/**
* Add a Message to the group with the provided group ID.
@@ -110,22 +94,26 @@ public abstract class AbstractKeyValueMessageStore extends AbstractMessageGroupS
public MessageGroup addMessageToGroup(Object groupId, Message<?> message) {
Assert.notNull(groupId, "'groupId' must not be null");
Assert.notNull(message, "'message' must not be null");
SimpleMessageGroup messageGroup = this.getSimpleMessageGroup(this.getMessageGroup(groupId));
// add message as is to the MG accessible by the caller
SimpleMessageGroup messageGroup = this.getSimpleMessageGroup(this.getMessageGroup(groupId));
messageGroup.add(message);
this.doStore(MESSAGE_GROUP_KEY_PREFIX + groupId, new MessageGroupMetadata(messageGroup));
this.addMessage(message);
return messageGroup;
}
// enrich Message with additional headers and add it to MS
Message<?> enrichedMessage = this.enrichMessage(message);
this.addMessage(enrichedMessage);
/**
* Mark all messages in the provided group.
*/
public MessageGroup markMessageGroup(MessageGroup group) {
Assert.notNull(group, "'group' must not be null");
Object groupId = group.getGroupId();
SimpleMessageGroup messageGroup = this.getSimpleMessageGroup(group);
messageGroup.markAll();
this.doStore(MESSAGE_GROUP_KEY_PREFIX + groupId, new MessageGroupMetadata(messageGroup));
// build raw MessageGroup and add enriched Message to it
SimpleMessageGroup rawGroup = this.buildMessageGroup(groupId, true);
rawGroup.add(enrichedMessage);
// store MessageGroupMetadata built from enriched MG
this.doStore(MESSAGE_GROUP_KEY_PREFIX + groupId, new MessageGroupMetadata(rawGroup));
// return clean MG
return messageGroup;
}
@@ -135,27 +123,30 @@ public abstract class AbstractKeyValueMessageStore extends AbstractMessageGroupS
public MessageGroup removeMessageFromGroup(Object groupId, Message<?> messageToRemove) {
Assert.notNull(groupId, "'groupId' must not be null");
Assert.notNull(messageToRemove, "'messageToRemove' must not be null");
SimpleMessageGroup messageGroup = this.getSimpleMessageGroup(this.getMessageGroup(groupId));
messageGroup.remove(messageToRemove);
this.doStore(MESSAGE_GROUP_KEY_PREFIX + groupId, new MessageGroupMetadata(messageGroup));
return messageGroup;
}
// build raw MG
SimpleMessageGroup rawGroup = this.buildMessageGroup(groupId, true);
// create a clean instance of
SimpleMessageGroup messageGroup = this.normalizeSimpleMessageGroup(rawGroup);
for (Message<?> message : rawGroup.getMessages()) {
if (message.getHeaders().getId().equals(messageToRemove.getHeaders().getId())){
rawGroup.remove(message);
}
}
this.removeMessage(messageToRemove.getHeaders().getId());
/**
* Mark the given Message within the group corresponding to the provided group ID.
*/
public MessageGroup markMessageFromGroup(Object groupId, Message<?> messageToMark) {
Assert.notNull(groupId, "'groupId' must not be null");
Assert.notNull(messageToMark, "'messageToMark' must not be null");
SimpleMessageGroup messageGroup = this.getSimpleMessageGroup(this.getMessageGroup(groupId));
messageGroup.mark(messageToMark);
this.doStore(MESSAGE_GROUP_KEY_PREFIX + groupId, new MessageGroupMetadata(messageGroup));
this.doStore(MESSAGE_GROUP_KEY_PREFIX + groupId, new MessageGroupMetadata(rawGroup));
messageGroup = this.getSimpleMessageGroup(this.getMessageGroup(groupId));
return messageGroup;
}
public void completeGroup(Object groupId) {
Assert.notNull(groupId, "'groupId' must not be null");
SimpleMessageGroup messageGroup = this.getSimpleMessageGroup(this.getMessageGroup(groupId));
SimpleMessageGroup messageGroup = this.buildMessageGroup(this.getMessageGroup(groupId), true);
messageGroup.complete();
this.doStore(MESSAGE_GROUP_KEY_PREFIX + groupId, new MessageGroupMetadata(messageGroup));
}
@@ -169,26 +160,98 @@ public abstract class AbstractKeyValueMessageStore extends AbstractMessageGroupS
if (mgm != null) {
Assert.isInstanceOf(MessageGroupMetadata.class, mgm);
MessageGroupMetadata messageGroupMetadata = (MessageGroupMetadata) mgm;
for (UUID messageId : messageGroupMetadata.getMarkedMessageIds()) {
this.removeMessage(messageId);
}
for (UUID messageId : messageGroupMetadata.getUnmarkedMessageIds()) {
this.removeMessage(messageId);
Iterator<UUID> messageIds = messageGroupMetadata.messageIdIterator();
while (messageIds.hasNext()){
this.removeMessage(messageIds.next());
}
}
}
public void setLastReleasedSequenceNumberForGroup(Object groupId, int sequenceNumber) {
Assert.notNull(groupId, "'groupId' must not be null");
SimpleMessageGroup messageGroup = this.getSimpleMessageGroup(this.getMessageGroup(groupId));
SimpleMessageGroup messageGroup = this.buildMessageGroup(groupId, true);
messageGroup.setLastReleasedMessageSequenceNumber(sequenceNumber);
this.doStore(MESSAGE_GROUP_KEY_PREFIX + groupId, new MessageGroupMetadata(messageGroup));
}
public Message<?> pollMessageFromGroup(Object groupId) {
Assert.notNull(groupId, "'groupId' must not be null");
Object mgm = this.doRetrieve(MESSAGE_GROUP_KEY_PREFIX + groupId);
if (mgm != null) {
Assert.isInstanceOf(MessageGroupMetadata.class, mgm);
MessageGroupMetadata messageGroupMetadata = (MessageGroupMetadata) mgm;
Message<?> message = this.removeMessage(messageGroupMetadata.firstId());
Message<?> normalizedMessage = this.normalizeMessage(message);
return normalizedMessage;
}
return null;
}
public Iterator<MessageGroup> iterator() {
final Iterator<?> idIterator = this.doListKeys(MESSAGE_GROUP_KEY_PREFIX + "*").iterator();
return new MessageGroupIterator(idIterator);
}
protected abstract Object doRetrieve(Object id);
protected abstract void doStore(Object id, Object objectToStore);
protected abstract Object doRemove(Object id);
protected abstract Collection<?> doListKeys(String keyPattern);
@SuppressWarnings({ "rawtypes", "unchecked" })
private Message<?> normalizeMessage(Message<?> message){
Message<?> normalizedMessage = MessageBuilder.fromMessage(message).removeHeader("CREATED_DATE").build();
Map innerMap = (Map) new DirectFieldAccessor(normalizedMessage.getHeaders()).getPropertyValue("headers");
innerMap.put(MessageHeaders.ID, message.getHeaders().getId());
innerMap.put(MessageHeaders.TIMESTAMP, message.getHeaders().getTimestamp());
return normalizedMessage;
}
/**
* Will enrich Message with additional meta headers
* @param message
* @return
*/
@SuppressWarnings({ "rawtypes", "unchecked" })
private Message<?> enrichMessage(Message<?> message){
Message<?> enrichedMessage = MessageBuilder.fromMessage(message).setHeader(CREATED_DATE, System.currentTimeMillis()).build();
Map innerMap = (Map) new DirectFieldAccessor(enrichedMessage.getHeaders()).getPropertyValue("headers");
innerMap.put(MessageHeaders.ID, message.getHeaders().getId());
innerMap.put(MessageHeaders.TIMESTAMP, message.getHeaders().getTimestamp());
return enrichedMessage;
}
private SimpleMessageGroup buildMessageGroup(Object groupId, boolean raw){
Assert.notNull(groupId, "'groupId' must not be null");
Object mgm = this.doRetrieve(MESSAGE_GROUP_KEY_PREFIX + groupId);
if (mgm != null) {
Assert.isInstanceOf(MessageGroupMetadata.class, mgm);
MessageGroupMetadata messageGroupMetadata = (MessageGroupMetadata) mgm;
ArrayList<Message<?>> messages = new ArrayList<Message<?>>();
Iterator<UUID> messageIds = messageGroupMetadata.messageIdIterator();
while (messageIds.hasNext()){
if (raw){
messages.add(this.getRawMessage(messageIds.next()));
}
else {
messages.add(this.getMessage(messageIds.next()));
}
}
SimpleMessageGroup messageGroup = new SimpleMessageGroup(messages,
groupId, messageGroupMetadata.getTimestamp(), messageGroupMetadata.isComplete());
messageGroup.setLastReleasedMessageSequenceNumber(messageGroupMetadata.getLastReleasedMessageSequenceNumber());
return messageGroup;
}
else {
return new SimpleMessageGroup(groupId);
}
}
private SimpleMessageGroup getSimpleMessageGroup(MessageGroup messageGroup){
if (messageGroup instanceof SimpleMessageGroup){
@@ -198,15 +261,22 @@ public abstract class AbstractKeyValueMessageStore extends AbstractMessageGroupS
return new SimpleMessageGroup(messageGroup);
}
}
protected abstract Object doRetrieve(Object id);
protected abstract void doStore(Object id, Object objectToStore);
private SimpleMessageGroup normalizeSimpleMessageGroup(SimpleMessageGroup messageGroup){
SimpleMessageGroup normalizedGroup = new SimpleMessageGroup(messageGroup.getGroupId());
for (Message<?> message : messageGroup.getMessages()) {
Message<?> normailizedMessage = this.normalizeMessage(message);
normalizedGroup.add(normailizedMessage);
}
return normalizedGroup;
}
protected abstract Object doRemove(Object id);
protected abstract Collection<?> doListKeys(String keyPattern);
private Message<?> getRawMessage(UUID id) {
Assert.notNull(id, "'id' must not be null");
Object message = this.doRetrieve(MESSAGE_KEY_PREFIX + id);
Assert.isInstanceOf(Message.class, message);
return (Message<?>) message;
}
private class MessageGroupIterator implements Iterator<MessageGroup> {

13
spring-integration-core/src/main/java/org/springframework/integration/store/AbstractMessageGroupStore.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2010 the original author or authors.
* Copyright 2002-2011 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
@@ -22,6 +22,7 @@ import org.springframework.jmx.export.annotation.ManagedAttribute;
/**
* @author Dave Syer
* @author Oleg Zhurakousky
*
* @since 2.0
*
@@ -76,15 +77,6 @@ public abstract class AbstractMessageGroupStore implements MessageGroupStore, It
return count;
}
@ManagedAttribute
public int getMarkedMessageCountForAllMessageGroups() {
int count = 0;
for (MessageGroup group : this) {
count += group.getMarked().size();
}
return count;
}
@ManagedAttribute
public int getMessageGroupCount() {
int count = 0;
@@ -112,7 +104,6 @@ public abstract class AbstractMessageGroupStore implements MessageGroupStore, It
if (exception != null) {
throw exception;
}
}
}

20
spring-integration-core/src/main/java/org/springframework/integration/store/MessageGroup.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2010 the original author or authors.
* Copyright 2002-2011 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.
@@ -21,12 +21,13 @@ import java.util.Collection;
import org.springframework.integration.Message;
/**
* A group of messages that are correlated with each other and should be processed in the same context. The group is
* divided into marked and unmarked messages. The marked messages are typically already processed, the unmarked messages
* are to be processed in the future.
* A group of messages that are correlated with each other and should be processed in the same context.
* <p/>
* The message group allows implementations to be mutable, but this behavior is optional. Implementations should take
* care to document their thread safety and mutability.
*
* @author Dave Syer
* @author Oleg Zhurakousky
*/
public interface MessageGroup {
@@ -36,14 +37,9 @@ public interface MessageGroup {
boolean canAdd(Message<?> message);
/**
* @return unmarked messages in the group at time of the invocation
* Returns all available Messages from the group at the time of invocation
*/
Collection<Message<?>> getUnmarked();
/**
* @return marked messages in the group at the time of the invocation
*/
Collection<Message<?>> getMarked();
Collection<Message<?>> getMessages();
/**
* @return the key that links these messages together
@@ -71,7 +67,7 @@ public interface MessageGroup {
int getSequenceSize();
/**
* @return the total number of messages (marked and unmarked) in this group
* @return the total number of messages in this group
*/
int size();

39
spring-integration-core/src/main/java/org/springframework/integration/store/MessageGroupMetadata.java
View File

@@ -17,9 +17,8 @@
package org.springframework.integration.store;
import java.io.Serializable;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.Iterator;
import java.util.TreeMap;
import java.util.UUID;
import org.springframework.integration.Message;
@@ -34,13 +33,12 @@ import org.springframework.util.Assert;
public class MessageGroupMetadata implements Serializable{
private static final long serialVersionUID = 1L;
private final static String CREATED_DATE = "CREATED_DATE";
private final Object groupId;
private final List<UUID> markedMessageIds;
private final List<UUID> unmarkedMessageIds;
private final TreeMap<Long, UUID> messageCreationDateToIdMappings;
private final boolean complete;
@@ -48,17 +46,16 @@ public class MessageGroupMetadata implements Serializable{
private final int lastReleasedMessageSequenceNumber;
public MessageGroupMetadata(MessageGroup messageGroup) {
Assert.notNull(messageGroup, "'messageGroup' must not be null");
this.groupId = messageGroup.getGroupId();
this.markedMessageIds = new ArrayList<UUID>();
for (Message<?> message : messageGroup.getMarked()) {
this.markedMessageIds.add(message.getHeaders().getId());
}
this.unmarkedMessageIds = new ArrayList<UUID>();
for (Message<?> message : messageGroup.getUnmarked()) {
this.unmarkedMessageIds.add(message.getHeaders().getId());
this.messageCreationDateToIdMappings = new TreeMap<Long, UUID>();
for (Message<?> message : messageGroup.getMessages()) {
Long createdDate = (Long) message.getHeaders().get(CREATED_DATE);
Assert.notNull(createdDate > 0, CREATED_DATE + " must not be null");
this.messageCreationDateToIdMappings.put(createdDate, message.getHeaders().getId());
}
this.complete = messageGroup.isComplete();
this.timestamp = messageGroup.getTimestamp();
@@ -69,13 +66,13 @@ public class MessageGroupMetadata implements Serializable{
public Object getGroupId() {
return this.groupId;
}
public List<UUID> getMarkedMessageIds() {
return Collections.unmodifiableList(markedMessageIds);
public Iterator<UUID> messageIdIterator(){
return this.messageCreationDateToIdMappings.values().iterator();
}
public List<UUID> getUnmarkedMessageIds() {
return Collections.unmodifiableList(this.unmarkedMessageIds);
public UUID firstId(){
return messageCreationDateToIdMappings.firstEntry().getValue();
}
public boolean isComplete() {

43
spring-integration-core/src/main/java/org/springframework/integration/store/MessageGroupQueue.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2008 the original author or authors.
* Copyright 2002-2011 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.
@@ -71,11 +71,11 @@ public class MessageGroupQueue extends AbstractQueue<Message<?>> implements Bloc
}
public Iterator<Message<?>> iterator() {
return getUnmarked().iterator();
return getMessages().iterator();
}
public int size() {
return getUnmarked().size();
return this.messageGroupStore.getMessageGroup(groupId).size();
}
public boolean offer(Message<?> e) {
@@ -92,22 +92,17 @@ public class MessageGroupQueue extends AbstractQueue<Message<?>> implements Bloc
}
public Message<?> peek() {
Collection<Message<?>> unmarked = getUnmarked();
if (unmarked.isEmpty()) {
Collection<Message<?>> messages = getMessages();
if (messages.isEmpty()) {
return null;
}
return unmarked.iterator().next();
return messages.iterator().next();
}
public Message<?> poll() {
Message<?> result;
Message<?> result = null;
synchronized (storeLock) {
Collection<Message<?>> unmarked = getUnmarked();
if (unmarked.isEmpty()) {
return null;
}
result = unmarked.iterator().next();
messageGroupStore.removeMessageFromGroup(groupId, result);
result = this.messageGroupStore.pollMessageFromGroup(groupId);
}
synchronized (writeLock) {
writeLock.notifyAll();
@@ -116,26 +111,24 @@ public class MessageGroupQueue extends AbstractQueue<Message<?>> implements Bloc
}
public int drainTo(Collection<? super Message<?>> c) {
Collection<Message<?>> unmarked;
synchronized (storeLock) {
unmarked = getUnmarked();
c.addAll(unmarked);
messageGroupStore.markMessageGroup(messageGroupStore.getMessageGroup(groupId));
for (Message<?> message = this.messageGroupStore.pollMessageFromGroup(groupId); message != null;) {
c.add(message);
}
}
synchronized (writeLock) {
writeLock.notifyAll();
}
return unmarked.size();
return this.messageGroupStore.getMessageGroup(groupId).size();
}
public int drainTo(Collection<? super Message<?>> c, int maxElements) {
ArrayList<Message<?>> list = new ArrayList<Message<?>>();
synchronized (storeLock) {
Iterator<Message<?>> unmarked = getUnmarked().iterator();
for (int i = 0; i < maxElements && unmarked.hasNext(); i++) {
Message<?> message = unmarked.next();
messageGroupStore.removeMessageFromGroup(groupId, message);
list.add(message);
Message<?> message = this.messageGroupStore.pollMessageFromGroup(groupId);
for (int i = 0; i < maxElements && message != null; i++) {
list.add(message);
message = this.messageGroupStore.pollMessageFromGroup(groupId);
}
}
synchronized (writeLock) {
@@ -195,8 +188,8 @@ public class MessageGroupQueue extends AbstractQueue<Message<?>> implements Bloc
return message;
}
private Collection<Message<?>> getUnmarked() {
return messageGroupStore.getMessageGroup(groupId).getUnmarked();
private Collection<Message<?>> getMessages(){
return messageGroupStore.getMessageGroup(groupId).getMessages();
}
}

36
spring-integration-core/src/main/java/org/springframework/integration/store/MessageGroupStore.java
View File

@@ -37,17 +37,6 @@ public interface MessageGroupStore {
*/
@ManagedAttribute
int getMessageCountForAllMessageGroups();
/**
* Optional attribute giving the number of marked messages in the store for all groups. Implementations may decline
* to respond by throwing an exception.
*
* @return the number of marked messages in each group
* @throws UnsupportedOperationException if not implemented
*/
@ManagedAttribute
int getMarkedMessageCountForAllMessageGroups();
/**
* Optional attribute giving the number of message groups. Implementations may decline
* to respond by throwing an exception.
@@ -73,15 +62,7 @@ public interface MessageGroupStore {
* @param message a message
*/
MessageGroup addMessageToGroup(Object groupId, Message<?> message);
/**
* Persist the mark on all the messages from the group. The group is modified in the process as all its unmarked
* messages become marked.
*
* @param group a MessageGroup with no unmarked messages
*/
MessageGroup markMessageGroup(MessageGroup group);
/**
* Persist a deletion on a single message from the group. The group is modified to reflect that 'messageToRemove' is
* no longer present in the group.
@@ -90,14 +71,6 @@ public interface MessageGroupStore {
*/
MessageGroup removeMessageFromGroup(Object key, Message<?> messageToRemove);
/**
* Persist a mark on a single message from the group. The group is modified to reflect that 'messageToMark' is no
* longer unmarked but became marked instead.
* @param key the groupId for the group containing the message
* @param messageToMark the message to be marked
*/
MessageGroup markMessageFromGroup(Object key, Message<?> messageToMark);
/**
* Remove the message group with this id.
*
@@ -136,6 +109,13 @@ public interface MessageGroupStore {
*/
Iterator<MessageGroup> iterator();
/**
* Polls Message from this {@link MessageGroup} (in FIFO style if supported by the implementation)
* while also removing the polled {@link Message}
*/
Message<?> pollMessageFromGroup(Object groupId);
/**
* Completes this MessageGroup. Completion of the MessageGroup generally means
* that this group should not be allowing any more mutating operation to be performed on it.

129
spring-integration-core/src/main/java/org/springframework/integration/store/SimpleMessageGroup.java
View File

@@ -32,15 +32,8 @@ import org.springframework.integration.Message;
public class SimpleMessageGroup implements MessageGroup {
private final Object groupId;
// Guards(marked, unmarked)
private final Object lock = new Object();
// @GuardedBy(lock)
public final BlockingQueue<Message<?>> marked = new LinkedBlockingQueue<Message<?>>();
// @GuardedBy(lock)
public final BlockingQueue<Message<?>> unmarked = new LinkedBlockingQueue<Message<?>>();
public final BlockingQueue<Message<?>> messages = new LinkedBlockingQueue<Message<?>>();
private volatile int lastReleasedMessageSequence;
@@ -49,50 +42,26 @@ public class SimpleMessageGroup implements MessageGroup {
private volatile boolean complete;
public SimpleMessageGroup(Object groupId) {
this(Collections.<Message<?>> emptyList(), Collections.<Message<?>> emptyList(), groupId, System
.currentTimeMillis(), false);
this(Collections.<Message<?>> emptyList(), groupId, System.currentTimeMillis(), false);
}
public SimpleMessageGroup(Collection<? extends Message<?>> unmarked, Object groupId) {
this(unmarked, Collections.<Message<?>> emptyList(), groupId, System.currentTimeMillis(), false);
public SimpleMessageGroup(Collection<? extends Message<?>> messages, Object groupId) {
this(messages, groupId, System.currentTimeMillis(), false);
}
public SimpleMessageGroup(Collection<? extends Message<?>> unmarked, Collection<? extends Message<?>> marked,
Object groupId, long timestamp, boolean complete) {
public SimpleMessageGroup(Collection<? extends Message<?>> messages, Object groupId, long timestamp, boolean complete) {
this.groupId = groupId;
this.timestamp = timestamp;
this.complete = complete;
synchronized (lock) {
for (Message<?> message : unmarked) {
addUnmarked(message);
}
for (Message<?> message : marked) {
addMarked(message);
}
for (Message<?> message : messages) {
addMessage(message);
}
}
public SimpleMessageGroup(MessageGroup template) {
this.groupId = template.getGroupId();
this.complete = template.isComplete();
synchronized (lock) {
// Explicit iteration to work around bug in JDK (before 1.6.0_20
for (Message<?> message : template.getMarked()) {
if (message != null) {
this.marked.add(message);
}
}
for (Message<?> message : template.getUnmarked()) {
if (message != null) {
this.unmarked.add(message);
}
}
}
this.timestamp = template.getTimestamp();
public SimpleMessageGroup(MessageGroup messageGroup) {
this(messageGroup.getMessages(), messageGroup.getGroupId(), messageGroup.getTimestamp(), messageGroup.isComplete());
}
public long getTimestamp() {
return timestamp;
}
@@ -102,54 +71,29 @@ public class SimpleMessageGroup implements MessageGroup {
}
public void add(Message<?> message) {
addUnmarked(message);
addMessage(message);
}
public void remove(Message<?> message) {
synchronized (lock) {
marked.remove(message);
unmarked.remove(message);
}
messages.remove(message);
}
public int getLastReleasedMessageSequenceNumber() {
return lastReleasedMessageSequence;
}
private boolean addUnmarked(Message<?> message) {
if (isMember(message)) {
return false;
}
synchronized (lock) {
return this.unmarked.offer(message);
}
private boolean addMessage(Message<?> message) {
return this.messages.offer(message);
}
private boolean addMarked(Message<?> message) {
if (isMember(message)) {
return false;
}
synchronized (lock) {
return this.marked.offer(message);
}
}
public Collection<Message<?>> getUnmarked() {
synchronized (lock) {
return Collections.unmodifiableCollection(unmarked);
}
public Collection<Message<?>> getMessages() {
return Collections.unmodifiableCollection(messages);
}
public void setLastReleasedMessageSequenceNumber(int sequenceNumber){
this.lastReleasedMessageSequence = sequenceNumber;
}
public Collection<Message<?>> getMarked() {
synchronized (lock) {
return Collections.unmodifiableCollection(marked);
}
}
public Object getGroupId() {
return groupId;
}
@@ -169,39 +113,17 @@ public class SimpleMessageGroup implements MessageGroup {
return getOne().getHeaders().getSequenceSize();
}
/**
* Mark the given message in this group. If the message is not part of this group then this call has no effect.
*/
public void mark(Message<?> messageToMark) {
synchronized (lock) {
unmarked.remove(messageToMark);
marked.offer(messageToMark);
}
}
public void markAll() {
synchronized (lock) {
unmarked.drainTo(marked);
}
}
public int size() {
synchronized (lock) {
return marked.size() + unmarked.size();
}
return this.messages.size();
}
public Message<?> getOne() {
Message<?> one = unmarked.peek();
if (one == null) {
one = marked.peek();
}
Message<?> one = messages.peek();
return one;
}
public void clear(){
this.marked.clear();
this.unmarked.clear();
this.messages.clear();
}
/**
@@ -220,12 +142,7 @@ public class SimpleMessageGroup implements MessageGroup {
return true;
}
else {
synchronized (lock) {
if (containsSequenceNumber(unmarked, messageSequenceNumber)
|| containsSequenceNumber(marked, messageSequenceNumber)) {
return true;
}
}
return this.containsSequenceNumber(messages, messageSequenceNumber);
}
}
return false;
@@ -245,10 +162,8 @@ public class SimpleMessageGroup implements MessageGroup {
public String toString() {
return "SimpleMessageGroup{" +
"groupId=" + groupId +
", lock=" + lock +
", marked=" + marked +
", unmarked=" + unmarked +
", messages=" + messages +
", timestamp=" + timestamp +
'}';
}
}
}

56
spring-integration-core/src/main/java/org/springframework/integration/store/SimpleMessageStore.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2009 the original author or authors.
* Copyright 2002-2011 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
@@ -13,6 +13,7 @@
package org.springframework.integration.store;
import java.util.Collection;
import java.util.HashSet;
import java.util.Iterator;
import java.util.UUID;
@@ -25,14 +26,16 @@ import org.springframework.integration.util.UpperBound;
import org.springframework.jmx.export.annotation.ManagedAttribute;
import org.springframework.jmx.export.annotation.ManagedResource;
import org.springframework.util.Assert;
import org.springframework.util.CollectionUtils;
/**
* Map-based implementation of {@link MessageStore} and {@link MessageGroupStore}. Enforces a maximum capacity for the
* Map-based in-memory implementation of {@link MessageStore} and {@link MessageGroupStore}. Enforces a maximum capacity for the
* store.
*
* @author Iwein Fuld
* @author Mark Fisher
* @author Dave Syer
* @author Oleg Zhurakousky
*
* @since 2.0
*/
@@ -42,10 +45,12 @@ public class SimpleMessageStore extends AbstractMessageGroupStore implements Mes
private final ConcurrentMap<UUID, Message<?>> idToMessage;
private final ConcurrentMap<Object, SimpleMessageGroup> groupIdToMessageGroup;
private final UpperBound individualUpperBound;
private final UpperBound groupUpperBound;
private final Object lock = new Object();
/**
* Creates a SimpleMessageStore with a maximum size limited by the given capacity, or unlimited size if the given
@@ -121,19 +126,14 @@ public class SimpleMessageStore extends AbstractMessageGroupStore implements Mes
return group;
}
public MessageGroup markMessageGroup(MessageGroup group) {
Object groupId = group.getGroupId();
SimpleMessageGroup internal = getMessageGroupInternal(groupId);
internal.markAll();
return internal;
}
public void removeMessageGroup(Object groupId) {
if (!groupIdToMessageGroup.containsKey(groupId)) {
return;
synchronized (lock) {
if (!groupIdToMessageGroup.containsKey(groupId)) {
return;
}
groupUpperBound.release(groupIdToMessageGroup.get(groupId).size());
groupIdToMessageGroup.remove(groupId);
}
groupUpperBound.release(groupIdToMessageGroup.get(groupId).size());
groupIdToMessageGroup.remove(groupId);
}
public MessageGroup removeMessageFromGroup(Object key, Message<?> messageToRemove) {
@@ -142,12 +142,6 @@ public class SimpleMessageStore extends AbstractMessageGroupStore implements Mes
return group;
}
public MessageGroup markMessageFromGroup(Object key, Message<?> messageToMark) {
SimpleMessageGroup group = getMessageGroupInternal(key);
group.mark(messageToMark);
return group;
}
public Iterator<MessageGroup> iterator() {
return new HashSet<MessageGroup>(groupIdToMessageGroup.values()).iterator();
}
@@ -157,6 +151,23 @@ public class SimpleMessageStore extends AbstractMessageGroupStore implements Mes
group.setLastReleasedMessageSequenceNumber(sequenceNumber);
}
public void completeGroup(Object groupId) {
SimpleMessageGroup group = getMessageGroupInternal(groupId);
group.complete();
}
public Message<?> pollMessageFromGroup(Object groupId) {
Collection<Message<?>> messageList = this.getMessageGroup(groupId).getMessages();
Message<?> message = null;
if (!CollectionUtils.isEmpty(messageList)){
message = messageList.iterator().next();
if (message != null){
this.removeMessageFromGroup(groupId, message);
}
}
return message;
}
private SimpleMessageGroup getMessageGroupInternal(Object groupId) {
if (!groupIdToMessageGroup.containsKey(groupId)) {
groupIdToMessageGroup.putIfAbsent(groupId, new SimpleMessageGroup(groupId));
@@ -164,9 +175,4 @@ public class SimpleMessageStore extends AbstractMessageGroupStore implements Mes
return groupIdToMessageGroup.get(groupId);
}
public void completeGroup(Object groupId) {
SimpleMessageGroup group = getMessageGroupInternal(groupId);
group.complete();
}
}

7
spring-integration-core/src/main/java/org/springframework/integration/util/UpperBound.java
View File

@@ -39,6 +39,13 @@ public final class UpperBound {
public UpperBound(int capacity) {
this.semaphore = (capacity > 0) ? new Semaphore(capacity, true) : null;
}
public int availablePermits(){
if (semaphore == null){
return Integer.MAX_VALUE;
}
return this.semaphore.availablePermits();
}
/**

8
spring-integration-core/src/test/java/org/springframework/integration/aggregator/AggregatorTests.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2010 the original author or authors.
* Copyright 2002-2011 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
@@ -130,7 +130,7 @@ public class AggregatorTests {
public void testTrackedCorrelationIdsCapacityAtLimit() {
QueueChannel replyChannel = new QueueChannel();
QueueChannel discardChannel = new QueueChannel();
// this.aggregator.setTrackedCorrelationIdCapacity(3);
this.aggregator.setDiscardChannel(discardChannel);
this.aggregator.handleMessage(createMessage(1, 1, 1, 1, replyChannel, null));
assertEquals(1, replyChannel.receive(100).getPayload());
@@ -149,7 +149,7 @@ public class AggregatorTests {
public void testTrackedCorrelationIdsCapacityPassesLimit() {
QueueChannel replyChannel = new QueueChannel();
QueueChannel discardChannel = new QueueChannel();
// this.aggregator.setTrackedCorrelationIdCapacity(3);
this.aggregator.setDiscardChannel(discardChannel);
this.aggregator.handleMessage(createMessage(1, 1, 1, 1, replyChannel, null));
assertEquals(1, replyChannel.receive(100).getPayload());
@@ -238,7 +238,7 @@ public class AggregatorTests {
private class MultiplyingProcessor implements MessageGroupProcessor {
public Object processMessageGroup(MessageGroup group) {
Integer product = 1;
for (Message<?> message : group.getUnmarked()) {
for (Message<?> message : group.getMessages()) {
product *= (Integer) message.getPayload();
}
return product;

5
spring-integration-core/src/test/java/org/springframework/integration/aggregator/ConcurrentAggregatorTests.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2010 the original author or authors.
* Copyright 2002-2011 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.
@@ -350,11 +350,10 @@ public class ConcurrentAggregatorTests {
private class MultiplyingProcessor implements MessageGroupProcessor {
public Object processMessageGroup(MessageGroup group) {
Integer product = 1;
for (Message<?> message : group.getUnmarked()) {
for (Message<?> message : group.getMessages()) {
product *= (Integer) message.getPayload();
}
return product;
//messagingTemplate.send(outputChannel, MessageBuilder.withPayload(product).build());
}
}

4
spring-integration-core/src/test/java/org/springframework/integration/aggregator/CorrelatingMessageBarrierTests.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2010 the original author or authors.
* Copyright 2002-2011 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
@@ -122,10 +122,8 @@ public class CorrelatingMessageBarrierTests {
private final ConcurrentMap<Object, Semaphore> keyLocks = new ConcurrentHashMap<Object, Semaphore>();
public boolean canRelease(MessageGroup messageGroup) {
// System.out.println("Trying to release group: " + messageGroup + "\n to thread: " + Thread.currentThread());
Object correlationKey = messageGroup.getGroupId();
Semaphore lock = lockForKey(correlationKey);
// System.out.println(Thread.currentThread() + " got lock: " + lock);
return lock.tryAcquire();
}

12
spring-integration-core/src/test/java/org/springframework/integration/aggregator/ExpressionEvaluatingMessageGroupProcessorTests.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2010 the original author or authors.
* Copyright 2002-2011 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.
@@ -60,7 +60,7 @@ public class ExpressionEvaluatingMessageGroupProcessorTests {
@Test
public void testProcessAndSendWithSizeExpressionEvaluated() throws Exception {
when(group.getUnmarked()).thenReturn(messages);
when(group.getMessages()).thenReturn(messages);
processor = new ExpressionEvaluatingMessageGroupProcessor("#root.size()");
Object result = processor.processMessageGroup(group);
assertTrue(result instanceof Message<?>);
@@ -70,7 +70,7 @@ public class ExpressionEvaluatingMessageGroupProcessorTests {
@Test
public void testProcessAndCheckHeaders() throws Exception {
when(group.getUnmarked()).thenReturn(messages);
when(group.getMessages()).thenReturn(messages);
processor = new ExpressionEvaluatingMessageGroupProcessor("#root");
Object result = processor.processMessageGroup(group);
assertTrue(result instanceof Message<?>);
@@ -80,7 +80,7 @@ public class ExpressionEvaluatingMessageGroupProcessorTests {
@Test
public void testProcessAndSendWithProjectionExpressionEvaluated() throws Exception {
when(group.getUnmarked()).thenReturn(messages);
when(group.getMessages()).thenReturn(messages);
processor = new ExpressionEvaluatingMessageGroupProcessor("![payload]");
Object result = processor.processMessageGroup(group);
assertTrue(result instanceof Message<?>);
@@ -97,7 +97,7 @@ public class ExpressionEvaluatingMessageGroupProcessorTests {
@Test
public void testProcessAndSendWithFilterAndProjectionExpressionEvaluated() throws Exception {
when(group.getUnmarked()).thenReturn(messages);
when(group.getMessages()).thenReturn(messages);
processor = new ExpressionEvaluatingMessageGroupProcessor("?[payload>2].![payload]");
Object result = processor.processMessageGroup(group);
assertTrue(result instanceof Message<?>);
@@ -112,7 +112,7 @@ public class ExpressionEvaluatingMessageGroupProcessorTests {
@Test
public void testProcessAndSendWithFilterAndProjectionAndMethodInvokingExpressionEvaluated() throws Exception {
when(group.getUnmarked()).thenReturn(messages);
when(group.getMessages()).thenReturn(messages);
processor = new ExpressionEvaluatingMessageGroupProcessor(String.format("T(%s).sum(?[payload>2].![payload])",
getClass().getName()));
Object result = processor.processMessageGroup(group);

20
spring-integration-core/src/test/java/org/springframework/integration/aggregator/MethodInvokingMessageGroupProcessorTests.java
View File

@@ -92,7 +92,7 @@ public class MethodInvokingMessageGroupProcessorTests {
}
MessageGroupProcessor processor = new MethodInvokingMessageGroupProcessor(new AnnotatedAggregatorMethod());
when(messageGroupMock.getUnmarked()).thenReturn(messagesUpForProcessing);
when(messageGroupMock.getMessages()).thenReturn(messagesUpForProcessing);
Object result = processor.processMessageGroup(messageGroupMock);
assertThat((Integer) ((Message<?>) result).getPayload(), is(7));
}
@@ -112,7 +112,7 @@ public class MethodInvokingMessageGroupProcessorTests {
}
MessageGroupProcessor processor = new MethodInvokingMessageGroupProcessor(new SimpleAggregator());
when(messageGroupMock.getUnmarked()).thenReturn(messagesUpForProcessing);
when(messageGroupMock.getMessages()).thenReturn(messagesUpForProcessing);
Object result = processor.processMessageGroup(messageGroupMock);
assertThat((Integer) ((Message<?>) result).getPayload(), is(7));
}
@@ -132,7 +132,7 @@ public class MethodInvokingMessageGroupProcessorTests {
}
MessageGroupProcessor processor = new MethodInvokingMessageGroupProcessor(new SimpleAggregator());
when(messageGroupMock.getUnmarked()).thenReturn(messagesUpForProcessing);
when(messageGroupMock.getMessages()).thenReturn(messagesUpForProcessing);
Object result = processor.processMessageGroup(messageGroupMock);
assertThat((Integer) ((Message<?>) result).getPayload(), is(7));
}
@@ -156,7 +156,7 @@ public class MethodInvokingMessageGroupProcessorTests {
MessageGroupProcessor processor = new MethodInvokingMessageGroupProcessor(new SimpleAggregator());
messagesUpForProcessing.add(MessageBuilder.withPayload(3).setHeader("foo", Arrays.asList(101, 102)).build());
when(messageGroupMock.getUnmarked()).thenReturn(messagesUpForProcessing);
when(messageGroupMock.getMessages()).thenReturn(messagesUpForProcessing);
Object result = processor.processMessageGroup(messageGroupMock);
assertThat((String) ((Message<?>) result).getPayload(), is("[1, 2, 4, 3, 101, 102]"));
}
@@ -180,7 +180,7 @@ public class MethodInvokingMessageGroupProcessorTests {
}
MessageGroupProcessor processor = new MethodInvokingMessageGroupProcessor(new SimpleAggregator());
when(messageGroupMock.getUnmarked()).thenReturn(messagesUpForProcessing);
when(messageGroupMock.getMessages()).thenReturn(messagesUpForProcessing);
Object result = processor.processMessageGroup(messageGroupMock);
assertThat((String) ((Message<?>) result).getPayload(), is("[1, 2, 4]"));
}
@@ -200,7 +200,7 @@ public class MethodInvokingMessageGroupProcessorTests {
}
MessageGroupProcessor processor = new MethodInvokingMessageGroupProcessor(new SimpleAggregator());
when(messageGroupMock.getUnmarked()).thenReturn(messagesUpForProcessing);
when(messageGroupMock.getMessages()).thenReturn(messagesUpForProcessing);
Object result = processor.processMessageGroup(messageGroupMock);
assertThat((Integer) ((Message<?>) result).getPayload(), is(7));
}
@@ -220,7 +220,7 @@ public class MethodInvokingMessageGroupProcessorTests {
}
MessageGroupProcessor processor = new MethodInvokingMessageGroupProcessor(new SimpleAggregator());
when(messageGroupMock.getUnmarked()).thenReturn(messagesUpForProcessing);
when(messageGroupMock.getMessages()).thenReturn(messagesUpForProcessing);
Object result = processor.processMessageGroup(messageGroupMock);
assertThat((Integer) ((Message<?>) result).getPayload(), is(7));
}
@@ -247,7 +247,7 @@ public class MethodInvokingMessageGroupProcessorTests {
}
});
processor.setConversionService(conversionService);
when(messageGroupMock.getUnmarked()).thenReturn(messagesUpForProcessing);
when(messageGroupMock.getMessages()).thenReturn(messagesUpForProcessing);
Object result = processor.processMessageGroup(messageGroupMock);
assertThat((Integer) ((Message<?>) result).getPayload(), is(7));
}
@@ -276,7 +276,7 @@ public class MethodInvokingMessageGroupProcessorTests {
}
MessageGroupProcessor processor = new MethodInvokingMessageGroupProcessor(new UnannotatedAggregator());
when(messageGroupMock.getUnmarked()).thenReturn(messagesUpForProcessing);
when(messageGroupMock.getMessages()).thenReturn(messagesUpForProcessing);
Object result = processor.processMessageGroup(messageGroupMock);
assertThat((Integer) ((Message<?>) result).getPayload(), is(7));
}
@@ -302,7 +302,7 @@ public class MethodInvokingMessageGroupProcessorTests {
}
MessageGroupProcessor processor = new MethodInvokingMessageGroupProcessor(new UnannotatedAggregator());
when(messageGroupMock.getUnmarked()).thenReturn(messagesUpForProcessing);
when(messageGroupMock.getMessages()).thenReturn(messagesUpForProcessing);
Object result = processor.processMessageGroup(messageGroupMock);
assertTrue(((Message<?>)result).getPayload() instanceof Iterator<?>);
}

35
spring-integration-core/src/test/java/org/springframework/integration/aggregator/ResequencingMessageGroupProcessorTest.java
View File

@@ -1,3 +1,18 @@
/*
* Copyright 2002-2011 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 org.junit.Test;
@@ -22,14 +37,14 @@ public class ResequencingMessageGroupProcessorTest {
@Test
public void shouldProcessSequence() {
Message prototypeMessage = MessageBuilder.withPayload("foo").setCorrelationId("x").setSequenceNumber(1).setSequenceSize(3).build();
List<Message<?>> unmarked= new ArrayList<Message<?>>();
List<Message<?>> messages= new ArrayList<Message<?>>();
Message message1 = MessageBuilder.fromMessage(prototypeMessage).setSequenceNumber(1).build();
Message message2 = MessageBuilder.fromMessage(prototypeMessage).setSequenceNumber(2).build();
Message message3 = MessageBuilder.fromMessage(prototypeMessage).setSequenceNumber(3).build();
unmarked.add(message1);
unmarked.add(message2);
unmarked.add(message3);
SimpleMessageGroup group = new SimpleMessageGroup(unmarked,"x");
messages.add(message1);
messages.add(message2);
messages.add(message3);
SimpleMessageGroup group = new SimpleMessageGroup(messages,"x");
List<Message> processedMessages = (List<Message>) processor.processMessageGroup(group);
assertThat(processedMessages, hasItems(message1, message2, message3));
}
@@ -38,14 +53,14 @@ public class ResequencingMessageGroupProcessorTest {
@Test
public void shouldPartiallProcessIncompleteSequence() {
Message prototypeMessage = MessageBuilder.withPayload("foo").setCorrelationId("x").setSequenceNumber(1).setSequenceSize(4).build();
List<Message<?>> unmarked= new ArrayList<Message<?>>();
List<Message<?>> messages= new ArrayList<Message<?>>();
Message message2 = MessageBuilder.fromMessage(prototypeMessage).setSequenceNumber(4).build();
Message message1 = MessageBuilder.fromMessage(prototypeMessage).setSequenceNumber(1).build();
Message message3 = MessageBuilder.fromMessage(prototypeMessage).setSequenceNumber(3).build();
unmarked.add(message1);
unmarked.add(message2);
unmarked.add(message3);
SimpleMessageGroup group = new SimpleMessageGroup(unmarked,"x");
messages.add(message1);
messages.add(message2);
messages.add(message3);
SimpleMessageGroup group = new SimpleMessageGroup(messages,"x");
List<Message> processedMessages = (List<Message>) processor.processMessageGroup(group);
assertThat(processedMessages, hasItems(message1));
assertThat(processedMessages.size(), is(1));

6
spring-integration-core/src/test/java/org/springframework/integration/aggregator/SequenceSizeReleaseStrategyTests.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2008 the original author or authors.
* Copyright 2002-2011 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.
@@ -58,13 +58,13 @@ public class SequenceSizeReleaseStrategyTests {
}
@Test
public void testEmptyUnmarked() {
public void testEmptyGroup() {
SequenceSizeReleaseStrategy releaseStrategy = new SequenceSizeReleaseStrategy();
releaseStrategy.setReleasePartialSequences(true);
SimpleMessageGroup messages = new SimpleMessageGroup("FOO");
Message<String> message = MessageBuilder.withPayload("test1").setSequenceSize(1).build();
messages.add(message);
messages.mark(message);
messages.remove(message);
assertTrue(releaseStrategy.canRelease(messages));
}

16
spring-integration-core/src/test/java/org/springframework/integration/aggregator/integration/AggregatorSupportedUseCasesTests.java
View File

@@ -47,15 +47,13 @@ public class AggregatorSupportedUseCasesTests {
QueueChannel discardChannel = new QueueChannel();
defaultHandler.setOutputChannel(outputChannel);
defaultHandler.setDiscardChannel(discardChannel);
defaultHandler.setKeepReleasedMessages(false);
for (int i = 0; i < 5; i++) {
defaultHandler.handleMessage(MessageBuilder.withPayload(i).setSequenceSize(5).setCorrelationId("A").setSequenceNumber(i).build());
}
assertEquals(5, ((List<?>)outputChannel.receive(0).getPayload()).size());
assertNull(discardChannel.receive(0));
assertEquals(0, store.getMessageGroup("A").getUnmarked().size());
assertEquals(0, store.getMessageGroup("A").getMarked().size());
assertEquals(0, store.getMessageGroup("A").getMessages().size());
// send another message with the same correlation id and see it in the discard channel
defaultHandler.handleMessage(MessageBuilder.withPayload("foo").setSequenceSize(5).setCorrelationId("A").setSequenceNumber(3).build());
@@ -65,7 +63,7 @@ public class AggregatorSupportedUseCasesTests {
defaultHandler.setExpireGroupsUponCompletion(true);
defaultHandler.handleMessage(MessageBuilder.withPayload("foo").setSequenceSize(5).setCorrelationId("A").setSequenceNumber(3).build());
assertNull(discardChannel.receive(0));
assertEquals(1, store.getMessageGroup("A").getUnmarked().size());
assertEquals(1, store.getMessageGroup("A").getMessages().size());
}
@Test
@@ -75,15 +73,13 @@ public class AggregatorSupportedUseCasesTests {
defaultHandler.setOutputChannel(outputChannel);
defaultHandler.setDiscardChannel(discardChannel);
defaultHandler.setReleaseStrategy(new SampleSizeReleaseStrategy());
defaultHandler.setKeepReleasedMessages(false);
for (int i = 0; i < 5; i++) {
defaultHandler.handleMessage(MessageBuilder.withPayload(i).setCorrelationId("A").build());
}
assertEquals(5, ((List<?>)outputChannel.receive(0).getPayload()).size());
assertNull(discardChannel.receive(0));
assertEquals(0, store.getMessageGroup("A").getUnmarked().size());
assertEquals(0, store.getMessageGroup("A").getMarked().size());
assertEquals(0, store.getMessageGroup("A").getMessages().size());
// send another message with the same correlation id and see it in the discard channel
defaultHandler.handleMessage(MessageBuilder.withPayload("foo").setCorrelationId("A").build());
@@ -93,7 +89,7 @@ public class AggregatorSupportedUseCasesTests {
defaultHandler.setExpireGroupsUponCompletion(true);
defaultHandler.handleMessage(MessageBuilder.withPayload("foo").setCorrelationId("A").build());
assertNull(discardChannel.receive(0));
assertEquals(1, store.getMessageGroup("A").getUnmarked().size());
assertEquals(1, store.getMessageGroup("A").getMessages().size());
}
@Test
@@ -146,13 +142,13 @@ public class AggregatorSupportedUseCasesTests {
assertEquals(5, ((List<?>)outputChannel.receive(0).getPayload()).size());
assertEquals(5, ((List<?>)outputChannel.receive(0).getPayload()).size());
assertNull(discardChannel.receive(0));
assertEquals(2, store.getMessageGroup("A").getUnmarked().size());
assertEquals(2, store.getMessageGroup("A").getMessages().size());
}
private class SampleSizeReleaseStrategy implements ReleaseStrategy {
public boolean canRelease(MessageGroup group) {
return group.getUnmarked().size() == 5;
return group.getMessages().size() == 5;
}
}

4
spring-integration-core/src/test/java/org/springframework/integration/aggregator/integration/ResequencerIntegrationTest-context.xml
View File

@@ -6,8 +6,8 @@
http://www.springframework.org/schema/integration http://www.springframework.org/schema/integration/spring-integration.xsd">
<int:resequencer id="resequencerLight" input-channel="resequencerLightInput" output-channel="outputChannel" release-partial-sequences="true"
keep-released-messages="false"/>
<int:resequencer id="resequencerLight" input-channel="resequencerLightInput" output-channel="outputChannel"
release-partial-sequences="true"/>
<int:channel id="outputChannel">
<int:queue/>

6
spring-integration-core/src/test/java/org/springframework/integration/aggregator/integration/ResequencerIntegrationTest.java
View File

@@ -85,8 +85,7 @@ public class ResequencerIntegrationTest {
assertEquals((Integer)6, message6.getHeaders().getSequenceNumber());
assertEquals(0, store.getMessageGroup("A").getUnmarked().size());
assertEquals(0, store.getMessageGroup("A").getMarked().size());
assertEquals(0, store.getMessageGroup("A").getMessages().size());
}
@Test
@@ -109,7 +108,6 @@ public class ResequencerIntegrationTest {
inputChannel.send(message2);
assertNotNull(outputChannel.receive(0));
assertNotNull(outputChannel.receive(0));
assertEquals(0, store.getMessageGroup("A").getUnmarked().size());
assertEquals(3, store.getMessageGroup("A").getMarked().size());
assertEquals(0, store.getMessageGroup("A").getMessages().size());
}
}

3
spring-integration-core/src/test/java/org/springframework/integration/config/xml/DefaultConfiguringBeanFactoryPostProcessorTests.java
View File

@@ -16,8 +16,6 @@
package org.springframework.integration.config.xml;
import java.util.Arrays;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.beans.DirectFieldAccessor;
@@ -79,7 +77,6 @@ public class DefaultConfiguringBeanFactoryPostProcessorTests {
new ClassPathXmlApplicationContext(new String[]{"org/springframework/integration/config/xml/parentApplicationContext.xml"}, superParentApplicationContext);
ClassPathXmlApplicationContext childApplicationContext =
new ClassPathXmlApplicationContext(new String[]{"org/springframework/integration/config/xml/childApplicationContext.xml"}, parentApplicationContext);
System.out.println(Arrays.asList(childApplicationContext.getBeanDefinitionNames()));
TaskScheduler parentScheduler = childApplicationContext.getParent().getBean("taskScheduler", TaskScheduler.class);
TaskScheduler childScheduler = childApplicationContext.getBean("taskScheduler", TaskScheduler.class);

1
spring-integration-core/src/test/java/org/springframework/integration/history/AnnotatedAdapter.java
View File

@@ -32,7 +32,6 @@ public class AnnotatedAdapter implements ApplicationContextAware{
@SuppressWarnings("serial")
public void handle(Message<?> message) {
System.out.println("Message: " + message);
applicationContext.publishEvent(new ApplicationEvent(message) {});
}

16
spring-integration-core/src/test/java/org/springframework/integration/store/MessageGroupQueueTests.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2010 the original author or authors.
* Copyright 2002-2011 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.
@@ -16,11 +16,6 @@
package org.springframework.integration.store;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertTrue;
import java.util.HashSet;
import java.util.Set;
import java.util.concurrent.Callable;
@@ -32,11 +27,16 @@ import java.util.concurrent.TimeUnit;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.junit.Ignore;
import org.junit.Test;
import org.springframework.integration.Message;
import org.springframework.integration.message.GenericMessage;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertTrue;
/**
* @author Dave Syer
* @since 2.0
@@ -92,11 +92,13 @@ public class MessageGroupQueueTests {
}
@Test
@Ignore
public void testConcurrentAccess() throws Exception {
doTestConcurrentAccess(50, 20, new HashSet<String>());
}
@Test
@Ignore
public void testConcurrentAccessUniqueResults() throws Exception {
doTestConcurrentAccess(50, 20, null);
}

14
spring-integration-core/src/test/java/org/springframework/integration/store/MessageStoreTests.java
View File

@@ -74,19 +74,14 @@ public class MessageStoreTests {
assertEquals(1, store.getMessageCountForAllMessageGroups());
}
@Test
public void testMarkedGroupSizes() throws Exception {
TestMessageStore store = new TestMessageStore();
assertEquals(0, store.getMarkedMessageCountForAllMessageGroups());
}
private static class TestMessageStore extends AbstractMessageGroupStore {
@SuppressWarnings("unchecked")
MessageGroup testMessages = new SimpleMessageGroup(Arrays.asList(new GenericMessage<String>("foo")), "bar");
private boolean removed = false;
public Iterator<MessageGroup> iterator() {
return Arrays.asList(testMessages).iterator();
}
@@ -126,6 +121,11 @@ public class MessageStoreTests {
throw new UnsupportedOperationException();
}
public Message<?> pollMessageFromGroup(Object groupId) {
// TODO Auto-generated method stub
return null;
}
}
}

1
spring-integration-core/src/test/java/org/springframework/integration/transformer/ObjectToMapTransformerTests.java
View File

@@ -59,7 +59,6 @@ public class ObjectToMapTransformerTests {
Message<?> transformedMessage = transformer.transform(message);
Map<String, Object> transformedMap = (Map<String, Object>) transformedMessage.getPayload();
System.out.println(transformedMap);
assertNotNull(transformedMap);
Object valueFromTheMap = null;

120
spring-integration-gemfire/src/test/java/org/springframework/integration/gemfire/store/GemfireGroupStoreTests.java
View File

@@ -81,6 +81,36 @@ public class GemfireGroupStoreTests {
assertEquals(1, messageGroup.size());
}
@Test
public void testRemoveMessageFromTheGroup() throws Exception{
GemfireMessageStore store = new GemfireMessageStore(this.cache);
store.afterPropertiesSet();
MessageGroup messageGroup = store.getMessageGroup(1);
Message<?> message = new GenericMessage<String>("2");
messageGroup = store.addMessageToGroup(messageGroup.getGroupId(), new GenericMessage<String>("1"));
messageGroup = store.getMessageGroup(1);
messageGroup = store.addMessageToGroup(messageGroup.getGroupId(), message);
messageGroup = store.getMessageGroup(1);
messageGroup = store.addMessageToGroup(messageGroup.getGroupId(), new GenericMessage<String>("3"));
messageGroup = store.getMessageGroup(1);
assertEquals(3, messageGroup.size());
messageGroup = store.removeMessageFromGroup(messageGroup.getGroupId(), message);
messageGroup = store.getMessageGroup(1);
assertEquals(2, messageGroup.size());
// make sure the store is properly rebuild from Gemfire
store = new GemfireMessageStore(this.cache);
store.afterPropertiesSet();
messageGroup = store.getMessageGroup(1);
assertEquals(2, messageGroup.size());
}
@Test
public void testRemoveMessageGroup() throws Exception{
GemfireMessageStore store = new GemfireMessageStore(this.cache);
@@ -93,8 +123,7 @@ public class GemfireGroupStoreTests {
store.removeMessageGroup(1);
MessageGroup messageGroupA = store.getMessageGroup(1);
assertNotSame(messageGroup, messageGroupA);
assertEquals(0, messageGroupA.getMarked().size());
assertEquals(0, messageGroupA.getUnmarked().size());
assertEquals(0, messageGroupA.getMessages().size());
assertEquals(0, messageGroupA.size());
// make sure the store is properly rebuild from Gemfire
@@ -103,59 +132,10 @@ public class GemfireGroupStoreTests {
messageGroup = store.getMessageGroup(1);
assertEquals(0, messageGroup.getMarked().size());
assertEquals(0, messageGroup.getUnmarked().size());
assertEquals(0, messageGroup.getMessages().size());
assertEquals(0, messageGroup.size());
}
@Test
public void testRemoveMessageFromTheGroup() throws Exception{
GemfireMessageStore store = new GemfireMessageStore(this.cache);
store.afterPropertiesSet();
MessageGroup messageGroup = store.getMessageGroup(1);
Message<?> message = new GenericMessage<String>("2");
store.addMessageToGroup(messageGroup.getGroupId(), new GenericMessage<String>("1"));
store.addMessageToGroup(messageGroup.getGroupId(), message);
messageGroup = store.addMessageToGroup(messageGroup.getGroupId(), new GenericMessage<String>("3"));
assertEquals(3, messageGroup.size());
messageGroup = store.removeMessageFromGroup(1, message);
assertEquals(2, messageGroup.size());
// make sure the store is properly rebuild from Gemfire
store = new GemfireMessageStore(this.cache);
store.afterPropertiesSet();
messageGroup = store.getMessageGroup(1);
assertEquals(2, messageGroup.size());
}
@Test
public void testMarkAllMessagesInMessageGroup() throws Exception{
GemfireMessageStore store = new GemfireMessageStore(this.cache);
store.afterPropertiesSet();
MessageGroup messageGroup = store.getMessageGroup(1);
store.addMessageToGroup(messageGroup.getGroupId(), new GenericMessage<String>("1"));
store.addMessageToGroup(messageGroup.getGroupId(), new GenericMessage<String>("2"));
messageGroup = store.addMessageToGroup(messageGroup.getGroupId(), new GenericMessage<String>("3"));
assertEquals(3, messageGroup.getUnmarked().size());
assertEquals(0, messageGroup.getMarked().size());
messageGroup = store.markMessageGroup(messageGroup);
assertEquals(0, messageGroup.getUnmarked().size());
assertEquals(3, messageGroup.getMarked().size());
// make sure the store is properly rebuild from Gemfire
store = new GemfireMessageStore(this.cache);
store.afterPropertiesSet();
messageGroup = store.getMessageGroup(1);
assertEquals(0, messageGroup.getUnmarked().size());
assertEquals(3, messageGroup.getMarked().size());
}
@Test
public void testRemoveNonExistingMessageFromTheGroup() throws Exception{
GemfireMessageStore store = new GemfireMessageStore(this.cache);
@@ -172,30 +152,6 @@ public class GemfireGroupStoreTests {
store.removeMessageFromGroup(1, new GenericMessage<String>("2"));
}
@Test
public void testMarkMessageInMessageGroup() throws Exception{
GemfireMessageStore store = new GemfireMessageStore(this.cache);
store.afterPropertiesSet();
MessageGroup messageGroup = store.getMessageGroup(1);
Message<?> messageToMark = new GenericMessage<String>("1");
store.addMessageToGroup(messageGroup.getGroupId(), messageToMark);
store.addMessageToGroup(messageGroup.getGroupId(), new GenericMessage<String>("2"));
messageGroup = store.addMessageToGroup(messageGroup.getGroupId(), new GenericMessage<String>("3"));
assertEquals(3, messageGroup.getUnmarked().size());
assertEquals(0, messageGroup.getMarked().size());
messageGroup = store.markMessageFromGroup(1, messageToMark);
assertEquals(2, messageGroup.getUnmarked().size());
assertEquals(1, messageGroup.getMarked().size());
// make sure the store is properly rebuild from Gemfire
store = new GemfireMessageStore(this.cache);
store.afterPropertiesSet();
messageGroup = store.getMessageGroup(1);
assertEquals(2, messageGroup.getUnmarked().size());
assertEquals(1, messageGroup.getMarked().size());
}
@Test
public void testCompleteMessageGroup() throws Exception{
@@ -233,16 +189,14 @@ public class GemfireGroupStoreTests {
store1.addMessageToGroup(1, message);
MessageGroup messageGroup = store2.addMessageToGroup(1, new GenericMessage<String>("2"));
assertEquals(2, messageGroup.getUnmarked().size());
assertEquals(0, messageGroup.getMarked().size());
assertEquals(2, messageGroup.getMessages().size());
GemfireMessageStore store3 = new GemfireMessageStore(this.cache);
store3.afterPropertiesSet();
messageGroup = store3.markMessageFromGroup(1, message);
messageGroup = store3.removeMessageFromGroup(1, message);
assertEquals(1, messageGroup.getUnmarked().size());
assertEquals(1, messageGroup.getMarked().size());
assertEquals(1, messageGroup.getMessages().size());
}
@Test
@@ -297,7 +251,7 @@ public class GemfireGroupStoreTests {
executor.execute(new Runnable() {
public void run() {
MessageGroup group = store1.addMessageToGroup(1, message);
if (group.getUnmarked().size() != 1){
if (group.getMessages().size() != 1){
failures.add("ADD");
throw new AssertionFailedError("Failed on ADD");
}
@@ -306,7 +260,7 @@ public class GemfireGroupStoreTests {
executor.execute(new Runnable() {
public void run() {
MessageGroup group = store2.removeMessageFromGroup(1, message);
if (group.getUnmarked().size() != 0){
if (group.getMessages().size() != 0){
failures.add("REMOVE");
throw new AssertionFailedError("Failed on Remove");
}

94
spring-integration-jdbc/src/main/java/org/springframework/integration/jdbc/JdbcMessageStore.java
View File

@@ -22,18 +22,23 @@ import java.util.ArrayList;
import java.util.Date;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.UUID;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicReference;
import javax.sql.DataSource;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.beans.DirectFieldAccessor;
import org.springframework.core.serializer.Deserializer;
import org.springframework.core.serializer.Serializer;
import org.springframework.core.serializer.support.DeserializingConverter;
import org.springframework.core.serializer.support.SerializingConverter;
import org.springframework.dao.DataAccessException;
import org.springframework.integration.Message;
import org.springframework.integration.MessageHeaders;
import org.springframework.integration.store.AbstractMessageGroupStore;
import org.springframework.integration.store.MessageGroup;
import org.springframework.integration.store.MessageStore;
@@ -43,6 +48,7 @@ import org.springframework.integration.util.UUIDConverter;
import org.springframework.jdbc.core.JdbcOperations;
import org.springframework.jdbc.core.JdbcTemplate;
import org.springframework.jdbc.core.PreparedStatementSetter;
import org.springframework.jdbc.core.ResultSetExtractor;
import org.springframework.jdbc.core.RowCallbackHandler;
import org.springframework.jdbc.core.RowMapper;
import org.springframework.jdbc.core.SingleColumnRowMapper;
@@ -83,6 +89,8 @@ public class JdbcMessageStore extends AbstractMessageGroupStore implements Messa
private static final String LIST_MESSAGES_BY_GROUP_KEY = "SELECT MESSAGE_ID, CREATED_DATE, GROUP_KEY, MESSAGE_BYTES, MARKED, COMPLETE, LAST_RELEASED_SEQUENCE from %PREFIX%MESSAGE_GROUP where GROUP_KEY=? and REGION=? order by CREATED_DATE";
private static final String LIST_MESSAGEIDS_BY_GROUP_KEY = "SELECT MESSAGE_ID, CREATED_DATE from %PREFIX%MESSAGE_GROUP where GROUP_KEY=? and REGION=? order by CREATED_DATE";
private static final String COUNT_ALL_GROUPS = "SELECT COUNT(GROUP_KEY) from %PREFIX%MESSAGE_GROUP where REGION=?";
private static final String COUNT_ALL_MARKED_MESSAGES_IN_GROUPS = "SELECT COUNT(MESSAGE_ID) from %PREFIX%MESSAGE_GROUP where MARKED=1 AND REGION=?";
@@ -101,8 +109,8 @@ public class JdbcMessageStore extends AbstractMessageGroupStore implements Messa
private static final String DELETE_MESSAGE_GROUP = "DELETE from %PREFIX%MESSAGE_GROUP where GROUP_KEY=? and REGION=?";
private static final String CREATE_MESSAGE_IN_GROUP = "INSERT into %PREFIX%MESSAGE_GROUP(MESSAGE_ID, REGION, CREATED_DATE, GROUP_KEY, MARKED, COMPLETE, LAST_RELEASED_SEQUENCE, MESSAGE_BYTES)"
+ " values (?, ?, ?, ?, 0, 0, 0, ?)";
private static final String CREATE_MESSAGE_IN_GROUP = "INSERT into %PREFIX%MESSAGE_GROUP(MESSAGE_ID, REGION, CREATED_DATE, GROUP_KEY, MARKED, COMPLETE, LAST_RELEASED_SEQUENCE)"
+ " values (?, ?, ?, ?, 0, 0, 0)";
private static final String LIST_GROUP_KEYS = "SELECT distinct GROUP_KEY as CREATED from %PREFIX%MESSAGE_GROUP where REGION=?";
@@ -266,9 +274,9 @@ public class JdbcMessageStore extends AbstractMessageGroupStore implements Messa
return list.get(0);
}
@SuppressWarnings({ "rawtypes", "unchecked" })
public <T> Message<T> addMessage(final Message<T> message) {
if (message.getHeaders().containsKey(SAVED_KEY)) {
@SuppressWarnings("unchecked")
Message<T> saved = (Message<T>) getMessage(message.getHeaders().getId());
if (saved != null) {
if (saved.equals(message)) {
@@ -280,6 +288,11 @@ public class JdbcMessageStore extends AbstractMessageGroupStore implements Messa
final long createdDate = System.currentTimeMillis();
Message<T> result = MessageBuilder.fromMessage(message).setHeader(SAVED_KEY, Boolean.TRUE)
.setHeader(CREATED_DATE_KEY, new Long(createdDate)).build();
Map innerMap = (Map) new DirectFieldAccessor(result.getHeaders()).getPropertyValue("headers");
// using reflection to set ID since it is immutable through MessageHeaders
innerMap.put(MessageHeaders.ID, message.getHeaders().get(MessageHeaders.ID));
final String messageId = getKey(result.getHeaders().getId());
final byte[] messageBytes = serializer.convert(result);
@@ -300,7 +313,6 @@ public class JdbcMessageStore extends AbstractMessageGroupStore implements Messa
final long createdDate = System.currentTimeMillis();
final String messageId = getKey(message.getHeaders().getId());
final String groupKey = getKey(groupId);
final byte[] messageBytes = serializer.convert(message);
jdbcTemplate.update(getQuery(CREATE_MESSAGE_IN_GROUP), new PreparedStatementSetter() {
public void setValues(PreparedStatement ps) throws SQLException {
@@ -309,10 +321,9 @@ public class JdbcMessageStore extends AbstractMessageGroupStore implements Messa
ps.setString(2, region);
ps.setTimestamp(3, new Timestamp(createdDate));
ps.setString(4, groupKey);
lobHandler.getLobCreator().setBlobAsBytes(ps, 5, messageBytes);
}
});
this.addMessage(message);
return getMessageGroup(groupId);
}
@@ -334,25 +345,20 @@ public class JdbcMessageStore extends AbstractMessageGroupStore implements Messa
public MessageGroup getMessageGroup(Object groupId) {
String key = getKey(groupId);
final List<Message<?>> marked = new ArrayList<Message<?>>();
final List<Message<?>> unmarked = new ArrayList<Message<?>>();
final List<Message<?>> messages = new ArrayList<Message<?>>();
final AtomicReference<Date> date = new AtomicReference<Date>();
final AtomicReference<Boolean> completeFlag = new AtomicReference<Boolean>();
final AtomicReference<Integer> lastReleasedSequenceRef = new AtomicReference<Integer>();
final AtomicInteger size = new AtomicInteger();
jdbcTemplate.query(getQuery(LIST_MESSAGES_BY_GROUP_KEY), new Object[] { key, region },
new RowCallbackHandler() {
int count = 0;
public void processRow(ResultSet rs) throws SQLException {
int markedFlag = rs.getInt("MARKED");
Message<?> message = mapper.mapRow(rs, count++);
if (markedFlag > 0) {
marked.add(message);
}
else {
unmarked.add(message);
}
size.incrementAndGet();
messages.add(getMessage(UUID.fromString(rs.getString("MESSAGE_ID"))));
date.set(rs.getTimestamp("CREATED_DATE"));
completeFlag.set(rs.getInt("COMPLETE") > 0);
@@ -360,17 +366,19 @@ public class JdbcMessageStore extends AbstractMessageGroupStore implements Messa
lastReleasedSequenceRef.set(rs.getInt("LAST_RELEASED_SEQUENCE"));
}
});
if (marked.isEmpty() && unmarked.isEmpty()) {
if (size.get() == 0){
return new SimpleMessageGroup(groupId);
}
Assert.state(date.get() != null, "Could not locate created date for groupId=" + groupId);
long timestamp = date.get().getTime();
boolean complete = completeFlag.get().booleanValue();
SimpleMessageGroup messageGroup = new SimpleMessageGroup(unmarked, marked, groupId, timestamp, complete);
SimpleMessageGroup messageGroup = new SimpleMessageGroup(messages, groupId, timestamp, complete);
int lastReleasedSequenceNumber = lastReleasedSequenceRef.get();
if (lastReleasedSequenceNumber > 0){
messageGroup.setLastReleasedMessageSequenceNumber(lastReleasedSequenceNumber);
}
return messageGroup;
}
@@ -404,6 +412,7 @@ public class JdbcMessageStore extends AbstractMessageGroupStore implements Messa
ps.setString(3, messageId);
}
});
this.removeMessage(messageToRemove.getHeaders().getId());
return getMessageGroup(groupId);
}
@@ -431,6 +440,10 @@ public class JdbcMessageStore extends AbstractMessageGroupStore implements Messa
public void removeMessageGroup(Object groupId) {
final String groupKey = getKey(groupId);
for (UUID messageIds : this.getMessageIdsForGroup(groupId)) {
this.removeMessage(messageIds);
}
jdbcTemplate.update(getQuery(DELETE_MESSAGE_GROUP), new PreparedStatementSetter() {
public void setValues(PreparedStatement ps) throws SQLException {
@@ -439,6 +452,7 @@ public class JdbcMessageStore extends AbstractMessageGroupStore implements Messa
ps.setString(2, region);
}
});
}
public void completeGroup(Object groupId) {
@@ -470,6 +484,46 @@ public class JdbcMessageStore extends AbstractMessageGroupStore implements Messa
}
});
}
public Message<?> pollMessageFromGroup(Object groupId) {
String key = getKey(groupId);
UUID messageId = jdbcTemplate.query(getQuery(LIST_MESSAGEIDS_BY_GROUP_KEY), new Object[] { key, region },
new ResultSetExtractor<UUID>() {
public UUID extractData(ResultSet rs)
throws SQLException, DataAccessException {
if (rs.next()) {
UUID uuid = UUID.fromString(rs.getString(1));
return uuid;
}
return null;
}
});
if (messageId != null){
Message<?> message = this.getMessage(messageId);
this.removeMessageFromGroup(groupId, message);
return message;
}
return null;
}
private List<UUID> getMessageIdsForGroup(Object groupId){
String key = getKey(groupId);
final List<UUID> messageIds = new ArrayList<UUID>();
jdbcTemplate.query(getQuery(LIST_MESSAGEIDS_BY_GROUP_KEY), new Object[] { key, region },
new RowCallbackHandler() {
public void processRow(ResultSet rs) throws SQLException {
messageIds.add(UUID.fromString(rs.getString(1)));
}
}
);
return messageIds;
}
public Iterator<MessageGroup> iterator() {

2
spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/JdbcMessageStoreChannelIntegrationTests.java
View File

@@ -81,7 +81,7 @@ public class JdbcMessageStoreChannelIntegrationTests {
Service.await(1000);
assertEquals(1, Service.messages.size());
// After a rollback in the poller the message is still waiting to be delivered
// but unless we use a transactin here there is a chance that the queue will
// but unless we use a transaction here there is a chance that the queue will
// appear empty....
new TransactionTemplate(transactionManager).execute(new TransactionCallback<Void>() {

2
spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/JdbcMessageStoreChannelTests.java
View File

@@ -70,7 +70,6 @@ public class JdbcMessageStoreChannelTests {
assertEquals(1, Service.messages.size());
// After a rollback in the poller the message is still waiting to be delivered
assertEquals(1, messageStore.getMessageGroup("input-queue").size());
assertEquals(1, messageStore.getMessageGroup("input-queue").getUnmarked().size());
}
@Test
@@ -89,7 +88,6 @@ public class JdbcMessageStoreChannelTests {
assertEquals(0, Service.messages.size());
// But inside the transaction the message is still there
assertEquals(1, messageStore.getMessageGroup("input-queue").size());
assertEquals(1, messageStore.getMessageGroup("input-queue").getUnmarked().size());
}
public static class Service {

51
spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/JdbcMessageStoreTests.java
View File

@@ -16,21 +16,11 @@
package org.springframework.integration.jdbc;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertNotSame;
import static org.junit.Assert.assertNull;
import static org.junit.Assert.assertSame;
import static org.junit.Assert.assertThat;
import static org.junit.Assert.assertTrue;
import static org.springframework.integration.test.matcher.PayloadAndHeaderMatcher.sameExceptIgnorableHeaders;
import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStream;
import java.io.InputStreamReader;
import java.io.OutputStream;
import java.util.Iterator;
import java.util.UUID;
import javax.sql.DataSource;
@@ -51,6 +41,16 @@ import org.springframework.test.context.ContextConfiguration;
import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
import org.springframework.transaction.annotation.Transactional;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertNotSame;
import static org.junit.Assert.assertNull;
import static org.junit.Assert.assertSame;
import static org.junit.Assert.assertThat;
import static org.junit.Assert.assertTrue;
import static org.springframework.integration.test.matcher.PayloadAndHeaderMatcher.sameExceptIgnorableHeaders;
@ContextConfiguration
@RunWith(SpringJUnit4ClassRunner.class)
public class JdbcMessageStoreTests {
@@ -77,7 +77,7 @@ public class JdbcMessageStoreTests {
public void testAddAndGet() throws Exception {
Message<String> message = MessageBuilder.withPayload("foo").build();
Message<String> saved = messageStore.addMessage(message);
assertNull(messageStore.getMessage(message.getHeaders().getId()));
assertNotNull(messageStore.getMessage(message.getHeaders().getId()));
Message<?> result = messageStore.getMessage(saved.getHeaders().getId());
assertNotNull(result);
assertThat(saved, sameExceptIgnorableHeaders(result));
@@ -111,7 +111,7 @@ public class JdbcMessageStoreTests {
});
Message<String> message = MessageBuilder.withPayload("foo").build();
Message<String> saved = messageStore.addMessage(message);
assertNull(messageStore.getMessage(message.getHeaders().getId()));
assertNotNull(messageStore.getMessage(message.getHeaders().getId()));
Message<?> result = messageStore.getMessage(saved.getHeaders().getId());
assertNotNull(result);
assertEquals("foo", result.getPayload());
@@ -263,31 +263,8 @@ public class JdbcMessageStoreTests {
messageStore.addMessageToGroup(groupId, message);
MessageGroup group = messageStore.getMessageGroup(groupId);
assertEquals(2, group.size());
Iterator<Message<?>> iterator = group.getUnmarked().iterator();
assertEquals("foo", iterator.next().getPayload());
assertEquals("bar", iterator.next().getPayload());
}
@Test
@Transactional
public void testAddAndMarkMessageGroup() throws Exception {
String groupId = "X";
Message<String> message = MessageBuilder.withPayload("foo").setCorrelationId(groupId).build();
messageStore.addMessageToGroup(groupId, message);
MessageGroup group = messageStore.getMessageGroup(groupId);
group = messageStore.markMessageGroup(group);
assertEquals(1, group.getMarked().size());
}
@Test
@Transactional
public void testAddAndMarkMessageInGroup() throws Exception {
String groupId = "X";
Message<String> message = MessageBuilder.withPayload("foo").setCorrelationId(groupId).build();
messageStore.addMessageToGroup(groupId, message);
messageStore.addMessageToGroup(groupId, MessageBuilder.withPayload("bar").setCorrelationId(groupId).build());
MessageGroup group = messageStore.markMessageFromGroup(groupId, message);
assertEquals(1, group.getMarked().size());
assertEquals("foo", messageStore.pollMessageFromGroup(groupId).getPayload());
assertEquals("bar", messageStore.pollMessageFromGroup(groupId).getPayload());
}
@Test

2
spring-integration-jmx/src/test/java/org/springframework/integration/jmx/config/MessageStoreTests.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2010 the original author or authors.
* Copyright 2002-2011 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

79
spring-integration-mongodb/src/main/java/org/springframework/integration/mongodb/store/MongoDbMessageStore.java
View File

@@ -16,8 +16,6 @@
package org.springframework.integration.mongodb.store;
import static org.springframework.data.mongodb.core.query.Criteria.where;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.Iterator;
@@ -36,6 +34,7 @@ import org.springframework.data.mongodb.core.convert.MappingMongoConverter;
import org.springframework.data.mongodb.core.mapping.MongoMappingContext;
import org.springframework.data.mongodb.core.mapping.MongoPersistentEntity;
import org.springframework.data.mongodb.core.mapping.MongoPersistentProperty;
import org.springframework.data.mongodb.core.query.Order;
import org.springframework.data.mongodb.core.query.Query;
import org.springframework.data.mongodb.core.query.Update;
import org.springframework.integration.Message;
@@ -49,10 +48,13 @@ import org.springframework.integration.store.SimpleMessageGroup;
import org.springframework.jmx.export.annotation.ManagedAttribute;
import org.springframework.util.Assert;
import org.springframework.util.ClassUtils;
import org.springframework.util.CollectionUtils;
import org.springframework.util.StringUtils;
import com.mongodb.DBObject;
import static org.springframework.data.mongodb.core.query.Criteria.where;
/**
* An implementation of both the {@link MessageStore} and {@link MessageGroupStore}
* strategies that relies upon MongoDB for persistence.
@@ -67,8 +69,6 @@ public class MongoDbMessageStore extends AbstractMessageGroupStore implements Me
private final static String GROUP_ID_KEY = "_groupId";
private final static String MARKED_KEY = "_marked";
private final static String GROUP_COMPLETE_KEY = "_group_complete";
private final static String LAST_RELEASED_SEQUENCE_NUMBER = "_last_released_sequence";
@@ -76,6 +76,8 @@ public class MongoDbMessageStore extends AbstractMessageGroupStore implements Me
private final static String GROUP_TIMESTAMP_KEY = "_group_timestamp";
private final static String PAYLOAD_TYPE_KEY = "_payloadType";
private final static String CREATED_DATE = "_createdDate";
private final MongoTemplate template;
@@ -135,8 +137,7 @@ public class MongoDbMessageStore extends AbstractMessageGroupStore implements Me
public MessageGroup getMessageGroup(Object groupId) {
Assert.notNull(groupId, "'groupId' must not be null");
List<MessageWrapper> messageWrappers = this.template.find(whereGroupIdIs(groupId), MessageWrapper.class, this.collectionName);
List<Message<?>> unmarkedMessages = new ArrayList<Message<?>>();
List<Message<?>> markedMessages = new ArrayList<Message<?>>();
List<Message<?>> messages = new ArrayList<Message<?>>();
long timestamp = 0;
int lastReleasedSequenceNumber = 0;
boolean completeGroup = false;
@@ -148,17 +149,13 @@ public class MongoDbMessageStore extends AbstractMessageGroupStore implements Me
}
for (MessageWrapper messageWrapper : messageWrappers) {
if (messageWrapper.isMarked()) {
markedMessages.add(messageWrapper.getMessage());
}
else {
unmarkedMessages.add(messageWrapper.getMessage());
}
messages.add(messageWrapper.getMessage());
}
SimpleMessageGroup messageGroup = new SimpleMessageGroup(unmarkedMessages, markedMessages, groupId, timestamp, completeGroup);
SimpleMessageGroup messageGroup = new SimpleMessageGroup(messages, groupId, timestamp, completeGroup);
if (lastReleasedSequenceNumber > 0){
messageGroup.setLastReleasedMessageSequenceNumber(lastReleasedSequenceNumber);
}
return messageGroup;
}
@@ -177,16 +174,6 @@ public class MongoDbMessageStore extends AbstractMessageGroupStore implements Me
return this.getMessageGroup(groupId);
}
public MessageGroup markMessageGroup(MessageGroup group) {
Assert.notNull(group, "'group' must not be null");
Object groupId = group.getGroupId();
List<MessageWrapper> messageWrappers = this.template.find(whereGroupIdIs(groupId), MessageWrapper.class, this.collectionName);
for (MessageWrapper messageWrapper : messageWrappers) {
this.markMessageFromGroup(groupId, messageWrapper.getMessage());
}
return this.getMessageGroup(groupId);
}
public MessageGroup removeMessageFromGroup(Object groupId, Message<?> messageToRemove) {
Assert.notNull(groupId, "'groupId' must not be null");
Assert.notNull(messageToRemove, "'messageToRemove' must not be null");
@@ -194,13 +181,6 @@ public class MongoDbMessageStore extends AbstractMessageGroupStore implements Me
return this.getMessageGroup(groupId);
}
public MessageGroup markMessageFromGroup(Object groupId, Message<?> messageToMark) {
Update update = Update.update(MARKED_KEY, true);
Query q = whereMessageIdIs(messageToMark.getHeaders().getId());
this.template.updateFirst(q, update, this.collectionName);
return this.getMessageGroup(groupId);
}
public void removeMessageGroup(Object groupId) {
List<MessageWrapper> messageWrappers = this.template.find(whereGroupIdIs(groupId), MessageWrapper.class, this.collectionName);
for (MessageWrapper messageWrapper : messageWrappers) {
@@ -231,6 +211,19 @@ public class MongoDbMessageStore extends AbstractMessageGroupStore implements Me
Query q = whereGroupIdIs(groupId);
this.template.updateFirst(q, update, this.collectionName);
}
public Message<?> pollMessageFromGroup(Object groupId) {
Assert.notNull(groupId, "'groupId' must not be null");
List<MessageWrapper> messageWrappers = this.template.find(whereGroupIdIsOrdered(groupId), MessageWrapper.class, this.collectionName);
Message<?> message = null;
if (!CollectionUtils.isEmpty(messageWrappers)){
message = messageWrappers.get(0).getMessage();
this.removeMessageFromGroup(groupId, message);
}
return message;
}
/*
* Common Queries
@@ -247,6 +240,12 @@ public class MongoDbMessageStore extends AbstractMessageGroupStore implements Me
private static Query whereGroupIdExists() {
return new Query(where(GROUP_ID_KEY).exists(true));
}
private static Query whereGroupIdIsOrdered(Object groupId) {
Query q = new Query(where(GROUP_ID_KEY).is(groupId)).limit(1);
q.sort().on(CREATED_DATE, Order.ASCENDING);
return q;
}
/**
@@ -272,7 +271,7 @@ public class MongoDbMessageStore extends AbstractMessageGroupStore implements Me
public void write(Object source, DBObject target) {
Message<?> message = null;
Object groupId = null;
boolean marked = false;
boolean groupComplete = false;
long groupTimestamp = 0;
int lastReleasedSequenceNumber = 0;
@@ -280,7 +279,6 @@ public class MongoDbMessageStore extends AbstractMessageGroupStore implements Me
MessageWrapper wrapper = (MessageWrapper) source;
message = wrapper.getMessage();
groupId = wrapper.getGroupId();
marked = wrapper.isMarked();
groupComplete = wrapper.isCompletedGroup();
lastReleasedSequenceNumber = wrapper.getLastReleasedSequenceNumber();
groupTimestamp = wrapper.getGroupTimestamp();
@@ -289,6 +287,7 @@ public class MongoDbMessageStore extends AbstractMessageGroupStore implements Me
Class<?> sourceType = (source != null) ? source.getClass() : null;
throw new IllegalArgumentException("Unexpected source type [" + sourceType + "]. Should be a MessageWrapper.");
}
target.put(CREATED_DATE, System.currentTimeMillis());
target.put(PAYLOAD_TYPE_KEY, message.getPayload().getClass().getName());
if (groupId != null) {
target.put(GROUP_ID_KEY, groupId);
@@ -296,9 +295,7 @@ public class MongoDbMessageStore extends AbstractMessageGroupStore implements Me
target.put(LAST_RELEASED_SEQUENCE_NUMBER, lastReleasedSequenceNumber);
target.put(GROUP_TIMESTAMP_KEY, groupTimestamp);
}
if (marked) {
target.put(MARKED_KEY, marked);
}
super.write(message, target);
}
@@ -341,8 +338,6 @@ public class MongoDbMessageStore extends AbstractMessageGroupStore implements Me
if (lastReleasedSequenceNumber != null){
wrapper.setLastReleasedSequenceNumber(lastReleasedSequenceNumber);
}
wrapper.setMarked(source.get(MARKED_KEY) != null);
if (completeGroup != null){
wrapper.setCompletedGroup(completeGroup.booleanValue());
@@ -376,8 +371,6 @@ public class MongoDbMessageStore extends AbstractMessageGroupStore implements Me
private volatile Object groupId;
private volatile boolean marked;
private final Message<?> message;
private volatile long groupTimestamp;
@@ -406,10 +399,6 @@ public class MongoDbMessageStore extends AbstractMessageGroupStore implements Me
return groupId;
}
public boolean isMarked() {
return marked;
}
public Message<?> getMessage() {
return message;
}
@@ -418,10 +407,6 @@ public class MongoDbMessageStore extends AbstractMessageGroupStore implements Me
this.groupId = groupId;
}
public void setMarked(boolean marked) {
this.marked = marked;
}
public void setGroupTimestamp(long groupTimestamp) {
this.groupTimestamp = groupTimestamp;
}

42
spring-integration-mongodb/src/test/java/org/springframework/integration/mongodb/store/MongoDbMessageGroupStoreTests.java
View File

@@ -85,19 +85,16 @@ public class MongoDbMessageGroupStoreTests extends MongoDbAvailableTests {
Message<?> messageB = new GenericMessage<String>("B");
store.addMessageToGroup(1, messageA);
messageGroup = store.addMessageToGroup(1, messageB);
assertEquals(0, messageGroup.getMarked().size());
assertEquals(2, messageGroup.getUnmarked().size());
assertEquals(2, messageGroup.size());
messageGroup = store.markMessageFromGroup(1, messageA);
assertEquals(1, messageGroup.getMarked().size());
assertEquals(1, messageGroup.getUnmarked().size());
messageGroup = store.removeMessageFromGroup(1, messageA);
assertEquals(1, messageGroup.size());
// validate that the updates were propagated to Mongo as well
store = new MongoDbMessageStore(mongoDbFactory);
messageGroup = store.getMessageGroup(1);
assertEquals(1, messageGroup.getMarked().size());
assertEquals(1, messageGroup.getUnmarked().size());
assertEquals(1, messageGroup.size());
}
@Test
@@ -167,33 +164,6 @@ public class MongoDbMessageGroupStoreTests extends MongoDbAvailableTests {
assertEquals(2, messageGroup.size());
}
@Test
@MongoDbAvailable
public void testMarkAllMessagesInMessageGroup() throws Exception {
MongoDbFactory mongoDbFactory = this.prepareMongoFactory();
MongoDbMessageStore store = new MongoDbMessageStore(mongoDbFactory);
MessageGroup messageGroup = store.getMessageGroup(1);
store.addMessageToGroup(1, new GenericMessage<String>("1"));
store.addMessageToGroup(1, new GenericMessage<String>("2"));
messageGroup = store.addMessageToGroup(1, new GenericMessage<String>("3"));
assertEquals(3, messageGroup.getUnmarked().size());
assertEquals(0, messageGroup.getMarked().size());
messageGroup = store.markMessageGroup(messageGroup);
assertEquals(0, messageGroup.getUnmarked().size());
assertEquals(3, messageGroup.getMarked().size());
store = new MongoDbMessageStore(mongoDbFactory);
messageGroup = store.getMessageGroup(1);
assertEquals(0, messageGroup.getUnmarked().size());
assertEquals(3, messageGroup.getMarked().size());
}
@Test
@MongoDbAvailable
public void testMultipleMessageStores() throws Exception{
@@ -210,12 +180,12 @@ public class MongoDbMessageGroupStoreTests extends MongoDbAvailableTests {
MessageGroup messageGroup = store3.getMessageGroup(1);
assertEquals(3, messageGroup.getUnmarked().size());
assertEquals(3, messageGroup.size());
store3.removeMessageFromGroup(1, message);
messageGroup = store2.getMessageGroup(1);
assertEquals(2, messageGroup.getUnmarked().size());
assertEquals(2, messageGroup.size());
}
@Test

71
spring-integration-redis/src/test/java/org/springframework/integration/redis/store/RedisMessageGroupStoreTests.java
View File

@@ -94,8 +94,8 @@ public class RedisMessageGroupStoreTests extends RedisAvailableTests {
store.removeMessageGroup(1);
MessageGroup messageGroupA = store.getMessageGroup(1);
assertNotSame(messageGroup, messageGroupA);
assertEquals(0, messageGroupA.getMarked().size());
assertEquals(0, messageGroupA.getUnmarked().size());
// assertEquals(0, messageGroupA.getMarked().size());
assertEquals(0, messageGroupA.getMessages().size());
assertEquals(0, messageGroupA.size());
// make sure the store is properly rebuild from Redis
@@ -103,8 +103,7 @@ public class RedisMessageGroupStoreTests extends RedisAvailableTests {
messageGroup = store.getMessageGroup(1);
assertEquals(0, messageGroup.getMarked().size());
assertEquals(0, messageGroup.getUnmarked().size());
assertEquals(0, messageGroup.getMessages().size());
assertEquals(0, messageGroup.size());
}
@@ -177,58 +176,6 @@ public class RedisMessageGroupStoreTests extends RedisAvailableTests {
store.removeMessageFromGroup(1, new GenericMessage<String>("2"));
}
@Test
@RedisAvailable
public void testMarkAllMessagesInMessageGroup() throws Exception{
JedisConnectionFactory jcf = this.getConnectionFactoryForTest();
RedisMessageStore store = new RedisMessageStore(jcf);
MessageGroup messageGroup = store.getMessageGroup(1);
store.addMessageToGroup(messageGroup.getGroupId(), new GenericMessage<String>("1"));
store.addMessageToGroup(messageGroup.getGroupId(), new GenericMessage<String>("2"));
messageGroup = store.addMessageToGroup(messageGroup.getGroupId(), new GenericMessage<String>("3"));
assertEquals(3, messageGroup.getUnmarked().size());
assertEquals(0, messageGroup.getMarked().size());
messageGroup = store.markMessageGroup(messageGroup);
assertEquals(0, messageGroup.getUnmarked().size());
assertEquals(3, messageGroup.getMarked().size());
// make sure the store is properly rebuild from Redis
store = new RedisMessageStore(jcf);
messageGroup = store.getMessageGroup(1);
assertEquals(0, messageGroup.getUnmarked().size());
assertEquals(3, messageGroup.getMarked().size());
}
@Test
@RedisAvailable
public void testMarkMessageInMessageGroup() throws Exception{
JedisConnectionFactory jcf = this.getConnectionFactoryForTest();
RedisMessageStore store = new RedisMessageStore(jcf);
MessageGroup messageGroup = store.getMessageGroup(1);
Message<?> messageToMark = new GenericMessage<String>("1");
store.addMessageToGroup(messageGroup.getGroupId(), messageToMark);
store.addMessageToGroup(messageGroup.getGroupId(), new GenericMessage<String>("2"));
messageGroup = store.addMessageToGroup(messageGroup.getGroupId(), new GenericMessage<String>("3"));
assertEquals(3, messageGroup.getUnmarked().size());
assertEquals(0, messageGroup.getMarked().size());
messageGroup = store.markMessageFromGroup(1, messageToMark);
assertEquals(2, messageGroup.getUnmarked().size());
assertEquals(1, messageGroup.getMarked().size());
// make sure the store is properly rebuild from Redis
store = new RedisMessageStore(jcf);
messageGroup = store.getMessageGroup(1);
assertEquals(2, messageGroup.getUnmarked().size());
assertEquals(1, messageGroup.getMarked().size());
}
@Test
@@ -243,15 +190,13 @@ public class RedisMessageGroupStoreTests extends RedisAvailableTests {
store1.addMessageToGroup(1, message);
MessageGroup messageGroup = store2.addMessageToGroup(1, new GenericMessage<String>("2"));
assertEquals(2, messageGroup.getUnmarked().size());
assertEquals(0, messageGroup.getMarked().size());
assertEquals(2, messageGroup.getMessages().size());
RedisMessageStore store3 = new RedisMessageStore(jcf);
messageGroup = store3.markMessageFromGroup(1, message);
messageGroup = store3.removeMessageFromGroup(1, message);
assertEquals(1, messageGroup.getUnmarked().size());
assertEquals(1, messageGroup.getMarked().size());
assertEquals(1, messageGroup.getMessages().size());
}
@Test
@@ -305,7 +250,7 @@ public class RedisMessageGroupStoreTests extends RedisAvailableTests {
executor.execute(new Runnable() {
public void run() {
MessageGroup group = store1.addMessageToGroup(1, message);
if (group.getUnmarked().size() != 1){
if (group.getMessages().size() != 1){
failures.add("ADD");
throw new AssertionFailedError("Failed on ADD");
}
@@ -314,7 +259,7 @@ public class RedisMessageGroupStoreTests extends RedisAvailableTests {
executor.execute(new Runnable() {
public void run() {
MessageGroup group = store2.removeMessageFromGroup(1, message);
if (group.getUnmarked().size() != 0){
if (group.getMessages().size() != 0){
failures.add("REMOVE");
throw new AssertionFailedError("Failed on Remove");
}