diff --git a/spring-integration-core/src/main/java/org/springframework/integration/aggregator/CorrelatingMessageBarrier.java b/spring-integration-core/src/main/java/org/springframework/integration/aggregator/CorrelatingMessageBarrier.java index 156cc38ccb..21dbb1cb07 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/aggregator/CorrelatingMessageBarrier.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/aggregator/CorrelatingMessageBarrier.java @@ -95,7 +95,7 @@ public class CorrelatingMessageBarrier extends AbstractMessageHandler implements Object correlationKey = this.correlationStrategy.getCorrelationKey(message); Object lock = getLock(correlationKey); synchronized (lock) { - this.store.addMessageToGroup(correlationKey, message); + this.store.addMessagesToGroup(correlationKey, message); } if (log.isDebugEnabled()) { log.debug(String.format("Handled message for key [%s]: %s.", correlationKey, message)); diff --git a/spring-integration-core/src/main/java/org/springframework/integration/aggregator/ResequencingMessageHandler.java b/spring-integration-core/src/main/java/org/springframework/integration/aggregator/ResequencingMessageHandler.java index 028acef6e4..8987eea6b0 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/aggregator/ResequencingMessageHandler.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/aggregator/ResequencingMessageHandler.java @@ -18,7 +18,6 @@ package org.springframework.integration.aggregator; import java.util.Collection; -import org.springframework.integration.IntegrationMessageHeaderAccessor; import org.springframework.integration.store.MessageGroup; import org.springframework.integration.store.MessageGroupStore; import org.springframework.integration.store.SimpleMessageStore; @@ -74,12 +73,9 @@ public class ResequencingMessageHandler extends AbstractCorrelatingMessageHandle @Override protected void afterRelease(MessageGroup messageGroup, Collection> completedMessages, boolean timeout) { - int size = messageGroup.getMessages().size(); - int sequenceSize = 0; - Message message = messageGroup.getOne(); - if (message != null) { - sequenceSize = new IntegrationMessageHeaderAccessor(message).getSequenceSize(); - } + int size = messageGroup.size(); + int sequenceSize = messageGroup.getSequenceSize(); + // If there is no sequence then it must be incomplete or unbounded if (sequenceSize > 0 && sequenceSize == size) { remove(messageGroup); diff --git a/spring-integration-core/src/main/java/org/springframework/integration/aggregator/SequenceSizeReleaseStrategy.java b/spring-integration-core/src/main/java/org/springframework/integration/aggregator/SequenceSizeReleaseStrategy.java index 687d134501..8f7668ef01 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/aggregator/SequenceSizeReleaseStrategy.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/aggregator/SequenceSizeReleaseStrategy.java @@ -36,6 +36,7 @@ import org.springframework.messaging.Message; * @author Dave Syer * @author Iwein Fuld * @author Oleg Zhurakousky + * @author Artem Bilan * @author Enrique Rodríguez */ public class SequenceSizeReleaseStrategy implements ReleaseStrategy { @@ -55,9 +56,9 @@ public class SequenceSizeReleaseStrategy implements ReleaseStrategy { } /** - * Flag that determines if partial sequences are allowed. If true then as soon as enough messages arrive that can be - * ordered they will be released, provided they all have sequence numbers greater than those already released. - * + * Flag that determines if partial sequences are allowed. If true then as soon as + * enough messages arrive that can be ordered they will be released, provided they + * all have sequence numbers greater than those already released. * @param releasePartialSequences true when partial sequences should be released. */ public void setReleasePartialSequences(boolean releasePartialSequences) { @@ -69,13 +70,12 @@ public class SequenceSizeReleaseStrategy implements ReleaseStrategy { boolean canRelease = false; - Collection> messages = messageGroup.getMessages(); - - if (this.releasePartialSequences && !messages.isEmpty()) { - + int size = messageGroup.size(); + if (this.releasePartialSequences && size > 0) { if (logger.isTraceEnabled()) { logger.trace("Considering partial release of group [" + messageGroup + "]"); } + Collection> messages = messageGroup.getMessages(); Message minMessage = Collections.min(messages, this.comparator); int nextSequenceNumber = new IntegrationMessageHeaderAccessor(minMessage).getSequenceNumber(); @@ -86,13 +86,11 @@ public class SequenceSizeReleaseStrategy implements ReleaseStrategy { } } else { - int size = messages.size(); - if (size == 0) { canRelease = true; } else { - int sequenceSize = new IntegrationMessageHeaderAccessor(messageGroup.getOne()).getSequenceSize(); + int sequenceSize = messageGroup.getSequenceSize(); // If there is no sequence then it must be incomplete.... if (sequenceSize == size) { canRelease = true; diff --git a/spring-integration-core/src/main/java/org/springframework/integration/store/AbstractKeyValueMessageStore.java b/spring-integration-core/src/main/java/org/springframework/integration/store/AbstractKeyValueMessageStore.java index c51af5714b..fcca5ddb1c 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/store/AbstractKeyValueMessageStore.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/store/AbstractKeyValueMessageStore.java @@ -94,38 +94,62 @@ public abstract class AbstractKeyValueMessageStore extends AbstractMessageGroupS */ @Override public MessageGroup getMessageGroup(Object groupId) { - return buildMessageGroup(groupId, false); + MessageGroupMetadata metadata = getGroupMetadata(groupId); + if (metadata != null) { + + MessageGroup messageGroup = getMessageGroupFactory() + .create(this, groupId, metadata.getTimestamp(), metadata.isComplete()); + messageGroup.setLastModified(metadata.getLastModified()); + messageGroup.setLastReleasedMessageSequenceNumber(metadata.getLastReleasedMessageSequenceNumber()); + return messageGroup; + } + else { + return new SimpleMessageGroup(groupId); + } } - - /** - * Add a Message to the group with the provided group ID. - */ @Override - public MessageGroup addMessageToGroup(Object groupId, Message message) { + public MessageGroupMetadata getGroupMetadata(Object groupId) { Assert.notNull(groupId, "'groupId' must not be null"); - Assert.notNull(message, "'message' must not be null"); + Object mgm = this.doRetrieve(MESSAGE_GROUP_KEY_PREFIX + groupId); + if (mgm != null) { + Assert.isInstanceOf(MessageGroupMetadata.class, mgm); + return (MessageGroupMetadata) mgm; + } + return null; + } - // add message as is to the MG accessible by the caller - MessageGroup messageGroup = getMessageGroup(groupId); + @Override + public void addMessagesToGroup(Object groupId, Message... messages) { + Assert.notNull(groupId, "'groupId' must not be null"); + Assert.notNull(messages, "'messages' must not be null"); - messageGroup.add(message); + MessageGroupMetadata metadata = getGroupMetadata(groupId); + SimpleMessageGroup group = null; + if (metadata == null) { + group = new SimpleMessageGroup(groupId); + } - // enrich Message with additional headers and add it to MS - Message enrichedMessage = enrichMessage(message); + for (Message message : messages) { + // enrich Message with additional headers and add it to MS + Message enrichedMessage = enrichMessage(message); + addMessage(enrichedMessage); + if (metadata != null) { + metadata.add(enrichedMessage.getHeaders().getId()); + } + else { + group.add(enrichedMessage); + } + } - addMessage(enrichedMessage); + if (group != null) { + metadata = new MessageGroupMetadata(group); + } - // build raw MessageGroup and add enriched Message to it - MessageGroup rawGroup = buildMessageGroup(groupId, true); - rawGroup.setLastModified(System.currentTimeMillis()); - rawGroup.add(enrichedMessage); + metadata.setLastModified(System.currentTimeMillis()); // store MessageGroupMetadata built from enriched MG - doStore(MESSAGE_GROUP_KEY_PREFIX + groupId, new MessageGroupMetadata(rawGroup)); - - // return clean MG - return getMessageGroup(groupId); + doStore(MESSAGE_GROUP_KEY_PREFIX + groupId, metadata); } /** @@ -137,32 +161,17 @@ public abstract class AbstractKeyValueMessageStore extends AbstractMessageGroupS Assert.notNull(groupId, "'groupId' must not be null"); Assert.notNull(messageToRemove, "'messageToRemove' must not be null"); - // build raw MG - MessageGroup rawGroup = buildMessageGroup(groupId, true); + UUID id = messageToRemove.getHeaders().getId(); + removeMessage(id); - // create a clean instance of - MessageGroup messageGroup = normalizeSimpleMessageGroup(rawGroup); - - - Message actualMessageToRemove = null; - - for (Message message : rawGroup.getMessages()) { - if (message.getHeaders().getId().equals(messageToRemove.getHeaders().getId())) { - actualMessageToRemove = message; - break; - } + MessageGroupMetadata metadata = getGroupMetadata(groupId); + if (metadata != null) { + metadata.remove(id); + metadata.setLastModified(System.currentTimeMillis()); + doStore(MESSAGE_GROUP_KEY_PREFIX + groupId, metadata); } - if (actualMessageToRemove != null) { - rawGroup.remove(actualMessageToRemove); - removeMessage(messageToRemove.getHeaders().getId()); - rawGroup.setLastModified(System.currentTimeMillis()); - - doStore(MESSAGE_GROUP_KEY_PREFIX + groupId, new MessageGroupMetadata(rawGroup)); - messageGroup = getMessageGroup(groupId); - } - - return messageGroup; + return getMessageGroup(groupId); } @@ -188,10 +197,12 @@ public abstract class AbstractKeyValueMessageStore extends AbstractMessageGroupS @Override public void completeGroup(Object groupId) { Assert.notNull(groupId, "'groupId' must not be null"); - MessageGroup messageGroup = buildMessageGroup(groupId, true); - messageGroup.complete(); - messageGroup.setLastModified(System.currentTimeMillis()); - doStore(MESSAGE_GROUP_KEY_PREFIX + groupId, new MessageGroupMetadata(messageGroup)); + MessageGroupMetadata metadata = getGroupMetadata(groupId); + if (metadata != null) { + metadata.complete(); + metadata.setLastModified(System.currentTimeMillis()); + doStore(MESSAGE_GROUP_KEY_PREFIX + groupId, metadata); + } } /** @@ -215,37 +226,62 @@ public abstract class AbstractKeyValueMessageStore extends AbstractMessageGroupS @Override public void setLastReleasedSequenceNumberForGroup(Object groupId, int sequenceNumber) { Assert.notNull(groupId, "'groupId' must not be null"); - MessageGroup messageGroup = buildMessageGroup(groupId, true); - messageGroup.setLastReleasedMessageSequenceNumber(sequenceNumber); - messageGroup.setLastModified(System.currentTimeMillis()); - doStore(MESSAGE_GROUP_KEY_PREFIX + groupId, new MessageGroupMetadata(messageGroup)); + MessageGroupMetadata metadata = getGroupMetadata(groupId); + if (metadata == null) { + SimpleMessageGroup messageGroup = new SimpleMessageGroup(groupId); + metadata = new MessageGroupMetadata(messageGroup); + } + metadata.setLastReleasedMessageSequenceNumber(sequenceNumber); + metadata.setLastModified(System.currentTimeMillis()); + doStore(MESSAGE_GROUP_KEY_PREFIX + groupId, metadata); } @Override public Message pollMessageFromGroup(Object groupId) { - Assert.notNull(groupId, "'groupId' must not be null"); - Object mgm = doRetrieve(MESSAGE_GROUP_KEY_PREFIX + groupId); - if (mgm != null) { - Assert.isInstanceOf(MessageGroupMetadata.class, mgm); - MessageGroupMetadata messageGroupMetadata = (MessageGroupMetadata) mgm; - - UUID firstId = messageGroupMetadata.firstId(); + MessageGroupMetadata groupMetadata = getGroupMetadata(groupId); + if (groupMetadata != null) { + UUID firstId = groupMetadata.firstId(); if (firstId != null) { - messageGroupMetadata.remove(firstId); - messageGroupMetadata.setLastModified(System.currentTimeMillis()); - doStore(MESSAGE_GROUP_KEY_PREFIX + groupId, messageGroupMetadata); + groupMetadata.remove(firstId); + groupMetadata.setLastModified(System.currentTimeMillis()); + doStore(MESSAGE_GROUP_KEY_PREFIX + groupId, groupMetadata); return removeMessage(firstId); } } return null; } + @Override + public Message getOneMessageFromGroup(Object groupId) { + MessageGroupMetadata groupMetadata = getGroupMetadata(groupId); + if (groupMetadata != null) { + UUID messageId = groupMetadata.firstId(); + if (messageId != null) { + return getMessage(messageId); + } + } + return null; + } + + @Override + public Collection> getMessagesForGroup(Object groupId) { + MessageGroupMetadata groupMetadata = getGroupMetadata(groupId); + ArrayList> messages = new ArrayList>(); + if (groupMetadata != null) { + Iterator messageIds = groupMetadata.messageIdIterator(); + while (messageIds.hasNext()) { + messages.add(getMessage(messageIds.next())); + } + } + return messages; + } + @Override @SuppressWarnings("unchecked") public Iterator iterator() { final Iterator idIterator = normalizeKeys( (Collection) doListKeys(MESSAGE_GROUP_KEY_PREFIX + "*")) - .iterator(); + .iterator(); return new MessageGroupIterator(idIterator); } @@ -266,13 +302,13 @@ public abstract class AbstractKeyValueMessageStore extends AbstractMessageGroupS @Override public int messageGroupSize(Object groupId) { - Object mgm = doRetrieve(MESSAGE_GROUP_KEY_PREFIX + groupId); + MessageGroupMetadata mgm = getGroupMetadata(groupId); if (mgm != null) { - Assert.isInstanceOf(MessageGroupMetadata.class, mgm); - MessageGroupMetadata messageGroupMetadata = (MessageGroupMetadata) mgm; - return messageGroupMetadata.size(); + return mgm.size(); + } + else { + return 0; } - return 0; } protected abstract Object doRetrieve(Object id); @@ -308,48 +344,6 @@ public abstract class AbstractKeyValueMessageStore extends AbstractMessageGroupS return enrichedMessage; } - private MessageGroup buildMessageGroup(Object groupId, boolean raw) { - Assert.notNull(groupId, "'groupId' must not be null"); - Object mgm = doRetrieve(MESSAGE_GROUP_KEY_PREFIX + groupId); - if (mgm != null) { - Assert.isInstanceOf(MessageGroupMetadata.class, mgm); - MessageGroupMetadata messageGroupMetadata = (MessageGroupMetadata) mgm; - ArrayList> messages = new ArrayList>(); - - Iterator messageIds = messageGroupMetadata.messageIdIterator(); - while (messageIds.hasNext()) { - UUID next = messageIds.next(); - if (next != null) { - if (raw) { - messages.add(getRawMessage(next)); - } - else { - messages.add(getMessage(next)); - } - } - } - - MessageGroup messageGroup = getMessageGroupFactory() - .create(messages, groupId, messageGroupMetadata.getTimestamp(), messageGroupMetadata.isComplete()); - messageGroup.setLastModified(messageGroupMetadata.getLastModified()); - messageGroup.setLastReleasedMessageSequenceNumber( - messageGroupMetadata.getLastReleasedMessageSequenceNumber()); - return messageGroup; - } - else { - return getMessageGroupFactory().create(groupId); - } - } - - private MessageGroup normalizeSimpleMessageGroup(MessageGroup messageGroup) { - MessageGroup normalizedGroup = getMessageGroupFactory().create(messageGroup.getGroupId()); - for (Message message : messageGroup.getMessages()) { - Message normalizedMessage = normalizeMessage(message); - normalizedGroup.add(normalizedMessage); - } - return normalizedGroup; - } - private Message getRawMessage(UUID id) { Assert.notNull(id, "'id' must not be null"); Object message = doRetrieve(MESSAGE_KEY_PREFIX + id); @@ -380,4 +374,5 @@ public abstract class AbstractKeyValueMessageStore extends AbstractMessageGroupS throw new UnsupportedOperationException(); } } + } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/store/AbstractMessageGroupStore.java b/spring-integration-core/src/main/java/org/springframework/integration/store/AbstractMessageGroupStore.java index 23effc035c..7074f3463b 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/store/AbstractMessageGroupStore.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/store/AbstractMessageGroupStore.java @@ -49,6 +49,9 @@ public abstract class AbstractMessageGroupStore extends AbstractBatchingMessageG private final Collection expiryCallbacks = new LinkedHashSet(); + private final MessageGroupFactory persistentMessageGroupFactory = + new SimpleMessageGroupFactory(SimpleMessageGroupFactory.GroupType.PERSISTENT); + private volatile boolean timeoutOnIdle; private volatile BeanFactory beanFactory; @@ -57,6 +60,8 @@ public abstract class AbstractMessageGroupStore extends AbstractBatchingMessageG private volatile boolean messageBuilderFactorySet; + private boolean lazyLoadMessageGroups = true; + public AbstractMessageGroupStore() { super(); } @@ -77,6 +82,16 @@ public abstract class AbstractMessageGroupStore extends AbstractBatchingMessageG return this.messageBuilderFactory; } + @Override + protected MessageGroupFactory getMessageGroupFactory() { + if (this.lazyLoadMessageGroups) { + return this.persistentMessageGroupFactory; + } + else { + return super.getMessageGroupFactory(); + } + } + /** * Convenient injection point for expiry callbacks in the message store. Each of the callbacks provided will simply * be registered with the store using {@link #registerMessageGroupExpiryCallback(MessageGroupCallback)}. @@ -98,13 +113,24 @@ public abstract class AbstractMessageGroupStore extends AbstractBatchingMessageG * the {@link MessageGroup} was created. If you want the timeout to be based on the time * the {@link MessageGroup} was idling (e.g., inactive from the last update) invoke this method with 'true'. * Default is 'false'. - * * @param timeoutOnIdle The boolean. */ public void setTimeoutOnIdle(boolean timeoutOnIdle) { this.timeoutOnIdle = timeoutOnIdle; } + /** + * Specify if the result of the {@link #getMessageGroup(Object)} should be wrapped + * to the {@link PersistentMessageGroup} - a lazy-load proxy for messages in group + * Defaults to {@code true}. + *

The target logic is based on the {@link SimpleMessageGroupFactory.GroupType#PERSISTENT}. + * @param lazyLoadMessageGroups the {@code boolean} flag to use. + * @since 4.3 + */ + public void setLazyLoadMessageGroups(boolean lazyLoadMessageGroups) { + this.lazyLoadMessageGroups = lazyLoadMessageGroups; + } + @Override public void registerMessageGroupExpiryCallback(MessageGroupCallback callback) { this.expiryCallbacks.add(callback); @@ -119,7 +145,7 @@ public abstract class AbstractMessageGroupStore extends AbstractBatchingMessageG long timestamp = group.getTimestamp(); if (this.isTimeoutOnIdle() && group.getLastModified() > 0) { - timestamp = group.getLastModified(); + timestamp = group.getLastModified(); } if (timestamp <= threshold) { @@ -175,8 +201,9 @@ public abstract class AbstractMessageGroupStore extends AbstractBatchingMessageG } @Override - public Message getOneMessageFromGroup(Object groupId) { - throw new UnsupportedOperationException("Not yet implemented for this store"); + public MessageGroup addMessageToGroup(Object groupId, Message message) { + addMessagesToGroup(groupId, message); + return getMessageGroup(groupId); } private void expire(MessageGroup group) { diff --git a/spring-integration-core/src/main/java/org/springframework/integration/store/MessageGroupFactory.java b/spring-integration-core/src/main/java/org/springframework/integration/store/MessageGroupFactory.java index 5be6bf5dd5..322f1892a1 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/store/MessageGroupFactory.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/store/MessageGroupFactory.java @@ -29,11 +29,57 @@ import org.springframework.messaging.Message; */ public interface MessageGroupFactory { + /** + * Create a {@link MessageGroup} instance based on the provided {@code groupId}. + * @param groupId the group id to use. + * @return the {@link MessageGroup} instance. + */ MessageGroup create(Object groupId); + /** + * Create a {@link MessageGroup} instance based on the provided {@code groupId} + * and with the {@code messages} for the group. + * @param messages the messages for the group. + * @param groupId the group id to use. + * @return the {@link MessageGroup} instance. + */ MessageGroup create(Collection> messages, Object groupId); + /** + * Create a {@link MessageGroup} instance based on the provided {@code groupId} + * and with the {@code messages} for the group. + * In addition the creating {@code timestamp} and {@code complete} flag may be used to customize + * the target {@link MessageGroup} object. + * @param messages the messages for the group. + * @param groupId the group id to use. + * @param timestamp the creation time. + * @param complete the {@code boolean} flag to indicate that group is completed. + * @return the {@link MessageGroup} instance. + */ MessageGroup create(Collection> messages, Object groupId, long timestamp, boolean complete); + /** + * Create a {@link MessageGroup} instance based on the provided {@code groupId}. + * The {@link MessageGroupStore} may be consulted for the messages and metadata for the {@link MessageGroup}. + * @param messageGroupStore the {@link MessageGroupStore} for additional {@link MessageGroup} information. + * @param groupId the group id to use. + * @return the {@link MessageGroup} instance. + */ + MessageGroup create(MessageGroupStore messageGroupStore, Object groupId); + + /** + * Create a {@link MessageGroup} instance based on the provided {@code groupId}. + * The {@link MessageGroupStore} may be consulted for the messages and metadata for the {@link MessageGroup}. + * In addition the creating {@code timestamp} and {@code complete} flag may be used to customize + * the target {@link MessageGroup} object. + * @param messageGroupStore the {@link MessageGroupStore} for additional {@link MessageGroup} information. + * @param groupId the group id to use. + * @param timestamp the creation time. + * @param complete the {@code boolean} flag to indicate that group is completed. + * @return the {@link MessageGroup} instance. + */ + MessageGroup create(MessageGroupStore messageGroupStore, Object groupId, long timestamp, + boolean complete); + } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/store/MessageGroupMetadata.java b/spring-integration-core/src/main/java/org/springframework/integration/store/MessageGroupMetadata.java index c4510d0f8f..42e7932701 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/store/MessageGroupMetadata.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/store/MessageGroupMetadata.java @@ -26,10 +26,11 @@ import org.springframework.messaging.Message; import org.springframework.util.Assert; /** - * Immutable Value Object holding metadata about a MessageGroup. + * Value Object holding metadata about a MessageGroup in the MessageGroupStore. * * @author Oleg Zhurakousky * @author Gary Russell + * @author Artem Bilan * @since 2.1 */ public class MessageGroupMetadata implements Serializable { @@ -40,47 +41,35 @@ public class MessageGroupMetadata implements Serializable { private final List messageIds = new LinkedList(); - private final boolean complete; - private final long timestamp; + private volatile boolean complete; + private volatile long lastModified; - private final int lastReleasedMessageSequenceNumber; - - private final boolean hasMessages; - - private final UUID first; + private volatile int lastReleasedMessageSequenceNumber; public MessageGroupMetadata(MessageGroup messageGroup) { - this(messageGroup, true, null); - } - - public MessageGroupMetadata(MessageGroup messageGroup, boolean hasMessages, UUID first) { - Assert.notNull(messageGroup, "'messageGroup' must not be null"); this.groupId = messageGroup.getGroupId(); - if (hasMessages) { - for (Message message : messageGroup.getMessages()) { - this.messageIds.add(message.getHeaders().getId()); - } + for (Message message : messageGroup.getMessages()) { + this.messageIds.add(message.getHeaders().getId()); } this.complete = messageGroup.isComplete(); this.timestamp = messageGroup.getTimestamp(); this.lastReleasedMessageSequenceNumber = messageGroup.getLastReleasedMessageSequenceNumber(); this.lastModified = messageGroup.getLastModified(); - this.hasMessages = hasMessages; - this.first = first; } public void remove(UUID messageId) { - if (!this.hasMessages) { - throw new IllegalStateException("Messages are not available, fetch the entire group"); - } this.messageIds.remove(messageId); } - public void setLastModified(long lastModified) { + boolean add(UUID messageId) { + return !this.messageIds.contains(messageId) && this.messageIds.add(messageId); + } + + void setLastModified(long lastModified) { this.lastModified = lastModified; } @@ -89,9 +78,6 @@ public class MessageGroupMetadata implements Serializable { } public Iterator messageIdIterator() { - if (!this.hasMessages) { - throw new IllegalStateException("Messages are not available, fetch the entire group"); - } return this.messageIds.iterator(); } @@ -100,18 +86,16 @@ public class MessageGroupMetadata implements Serializable { } public UUID firstId() { - if (this.first != null) { - return this.first; - } - if (!this.hasMessages) { - throw new IllegalStateException("Messages are not available, fetch the entire group"); - } if (this.messageIds.size() > 0) { - return this.messageIds.iterator().next(); + return this.messageIds.get(0); } return null; } + void complete() { + this.complete = true; + } + public boolean isComplete() { return this.complete; } @@ -127,4 +111,9 @@ public class MessageGroupMetadata implements Serializable { public int getLastReleasedMessageSequenceNumber() { return this.lastReleasedMessageSequenceNumber; } + + void setLastReleasedMessageSequenceNumber(int lastReleasedMessageSequenceNumber) { + this.lastReleasedMessageSequenceNumber = lastReleasedMessageSequenceNumber; + } + } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/store/MessageGroupStore.java b/spring-integration-core/src/main/java/org/springframework/integration/store/MessageGroupStore.java index 17aed5091a..b434f8f46c 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/store/MessageGroupStore.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/store/MessageGroupStore.java @@ -29,6 +29,7 @@ import org.springframework.messaging.Message; * @author Dave Syer * @author Oleg Zhurakousky * @author Gary Russell + * @author Artem Bilan * * @since 2.0 * @@ -36,9 +37,8 @@ import org.springframework.messaging.Message; public interface MessageGroupStore extends BasicMessageGroupStore { /** - * Optional attribute giving the number of messages in the store over all groups. Implementations may decline to - * respond by throwing an exception. - * + * Optional attribute giving the number of messages in the store over all groups. + * Implementations may decline to respond by throwing an exception. * @return the number of messages * @throws UnsupportedOperationException if not implemented */ @@ -46,9 +46,8 @@ public interface MessageGroupStore extends BasicMessageGroupStore { int getMessageCountForAllMessageGroups(); /** - * Optional attribute giving the number of message groups. Implementations may decline - * to respond by throwing an exception. - * + * Optional attribute giving the number of message groups. + * Implementations may decline to respond by throwing an exception. * @return the number message groups * @throws UnsupportedOperationException if not implemented */ @@ -56,9 +55,8 @@ public interface MessageGroupStore extends BasicMessageGroupStore { int getMessageGroupCount(); /** - * Persist the deletion of a single message from the group. The group is modified to reflect that 'messageToRemove' is - * no longer present in the group. - * + * Persist the deletion of a single message from the group. + * The group is modified to reflect that 'messageToRemove' is no longer present in the group. * @param key The groupId for the group containing the message. * @param messageToRemove The message to be removed. * @return The message Group. @@ -69,27 +67,22 @@ public interface MessageGroupStore extends BasicMessageGroupStore { /** * Persist the deletion of messages from the group. - * * @param key The groupId for the group containing the message(s). * @param messages The messages to be removed. - * * @since 4.2 */ void removeMessagesFromGroup(Object key, Collection> messages); /** * Persist the deletion of messages from the group. - * * @param key The groupId for the group containing the message(s). * @param messages The messages to be removed. - * * @since 4.2 */ void removeMessagesFromGroup(Object key, Message... messages); /** * Register a callback for when a message group is expired through {@link #expireMessageGroups(long)}. - * * @param callback A callback to execute when a message group is cleaned up. */ void registerMessageGroupExpiryCallback(MessageGroupCallback callback); @@ -99,10 +92,8 @@ public interface MessageGroupStore extends BasicMessageGroupStore { * each of the registered callbacks on them in turn. For example: call with a timeout of 100 to expire all groups * that were created more than 100 milliseconds ago, and are not yet complete. Use a timeout of 0 (or negative to be * on the safe side) to expire all message groups. - * * @param timeout the timeout threshold to use * @return the number of message groups expired - * * @see #registerMessageGroupExpiryCallback(MessageGroupCallback) */ @ManagedOperation @@ -110,7 +101,6 @@ public interface MessageGroupStore extends BasicMessageGroupStore { /** * Allows you to set the sequence number of the last released Message. Used for Resequencing use cases - * * @param groupId The group identifier. * @param sequenceNumber The sequence number. */ @@ -125,7 +115,6 @@ public interface MessageGroupStore extends BasicMessageGroupStore { * 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. * For example any attempt to add/remove new Message form the group should not be allowed. - * * @param groupId The group identifier. */ void completeGroup(Object groupId); @@ -140,13 +129,30 @@ public interface MessageGroupStore extends BasicMessageGroupStore { MessageGroupMetadata getGroupMetadata(Object groupId); /** - * Return the one {@link org.springframework.messaging.Message} from {@link org.springframework.integration.store.MessageGroup}. + * Return the one {@link Message} from {@link MessageGroup}. * @param groupId The group identifier. - * @return the {@link org.springframework.messaging.Message}. + * @return the {@link Message}. * @since 4.0 */ Message getOneMessageFromGroup(Object groupId); + /** + * Store messages with an association to a group id. + * This can be used to group messages together. + * @param groupId The group id to store messages under. + * @param messages The messages to add. + * @since 4.3 + */ + void addMessagesToGroup(Object groupId, Message... messages); + + /** + * Retrieve messages for the provided group id. + * @param groupId The group id to retrieve messages for. + * @return the messages for group. + * @since 4.3 + */ + Collection> getMessagesForGroup(Object groupId); + /** * Invoked when a MessageGroupStore expires a group. */ diff --git a/spring-integration-core/src/main/java/org/springframework/integration/store/PersistentMessageGroup.java b/spring-integration-core/src/main/java/org/springframework/integration/store/PersistentMessageGroup.java new file mode 100644 index 0000000000..f1e4645f7b --- /dev/null +++ b/spring-integration-core/src/main/java/org/springframework/integration/store/PersistentMessageGroup.java @@ -0,0 +1,226 @@ +/* + * Copyright 2016 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.store; + +import java.util.AbstractCollection; +import java.util.Collection; +import java.util.Collections; +import java.util.Iterator; + +import org.apache.commons.logging.Log; +import org.apache.commons.logging.LogFactory; + +import org.springframework.integration.IntegrationMessageHeaderAccessor; +import org.springframework.messaging.Message; + +/** + * @author Artem Bilan + * @since 4.3 + */ +class PersistentMessageGroup implements MessageGroup { + + private static final Log logger = LogFactory.getLog(PersistentMessageGroup.class); + + private MessageGroupStore messageGroupStore; + + private final Collection> messages = new PersistentCollection(); + + private final MessageGroup original; + + private volatile Message oneMessage; + + private volatile int size; + + PersistentMessageGroup(MessageGroupStore messageGroupStore, MessageGroup original) { + this.messageGroupStore = messageGroupStore; + this.original = original; + } + + public void setSize(int size) { + this.size = size; + } + + @Override + public Collection> getMessages() { + return Collections.unmodifiableCollection(this.messages); + } + + @Override + public Message getOne() { + if (this.oneMessage == null) { + synchronized (this) { + if (this.oneMessage == null) { + if (logger.isDebugEnabled()) { + logger.debug("Lazy loading of one message for messageGroup: " + this.original.getGroupId()); + } + this.oneMessage = this.messageGroupStore.getOneMessageFromGroup(this.original.getGroupId()); + } + } + } + return this.oneMessage; + } + + @Override + public int getSequenceSize() { + if (size() == 0) { + return 0; + } + else { + Message message = getOne(); + if (message != null) { + Integer sequenceSize = message.getHeaders() + .get(IntegrationMessageHeaderAccessor.SEQUENCE_SIZE, Integer.class); + return (sequenceSize != null ? sequenceSize : 0); + } + else { + return 0; + } + } + } + + @Override + public int size() { + if (this.size == 0) { + synchronized (this) { + if (this.size == 0) { + if (logger.isDebugEnabled()) { + logger.debug("Lazy loading of group size for messageGroup: " + this.original.getGroupId()); + } + this.size = this.messageGroupStore.messageGroupSize(this.original.getGroupId()); + } + } + } + return this.size; + } + + @Override + public Object getGroupId() { + return this.original.getGroupId(); + } + + @Override + public boolean canAdd(Message message) { + return this.original.canAdd(message); + } + + @Override + public int getLastReleasedMessageSequenceNumber() { + return this.original.getLastReleasedMessageSequenceNumber(); + } + + @Override + public boolean isComplete() { + return this.original.isComplete(); + } + + @Override + public void complete() { + this.original.complete(); + } + + @Override + public long getTimestamp() { + return this.original.getTimestamp(); + } + + @Override + public long getLastModified() { + return this.original.getLastModified(); + } + + @Override + public void setLastModified(long lastModified) { + this.original.setLastModified(lastModified); + } + + @Override + public void add(Message messageToAdd) { + this.original.add(messageToAdd); + } + + @Override + public boolean remove(Message messageToRemove) { + return this.original.remove(messageToRemove); + } + + @Override + public void setLastReleasedMessageSequenceNumber(int sequenceNumber) { + this.original.setLastReleasedMessageSequenceNumber(sequenceNumber); + } + + @Override + public void clear() { + this.original.clear(); + } + + + private class PersistentCollection extends AbstractCollection> { + + private volatile Collection> collection; + + private void load() { + if (this.collection == null) { + synchronized (this) { + if (this.collection == null) { + Object groupId = PersistentMessageGroup.this.original.getGroupId(); + if (logger.isDebugEnabled()) { + logger.debug("Lazy loading of messages for messageGroup: " + groupId); + } + this.collection = PersistentMessageGroup.this.messageGroupStore.getMessagesForGroup(groupId); + } + } + } + } + + @Override + public boolean contains(Object o) { + load(); + return this.collection.contains(o); + } + + @Override + public Object[] toArray() { + load(); + return this.collection.toArray(); + } + + @Override + public T[] toArray(T[] a) { + load(); + return this.collection.toArray(a); + } + + @Override + public boolean containsAll(Collection c) { + load(); + return this.collection.containsAll(c); + } + + @Override + public Iterator> iterator() { + load(); + return this.collection.iterator(); + } + + @Override + public int size() { + return PersistentMessageGroup.this.size(); + } + + } + +} diff --git a/spring-integration-core/src/main/java/org/springframework/integration/store/SimpleMessageGroupFactory.java b/spring-integration-core/src/main/java/org/springframework/integration/store/SimpleMessageGroupFactory.java index a14fd2ff0f..c805c58dd7 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/store/SimpleMessageGroupFactory.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/store/SimpleMessageGroupFactory.java @@ -59,6 +59,28 @@ public class SimpleMessageGroupFactory implements MessageGroupFactory { return new SimpleMessageGroup(this.type.get(), messages, groupId, timestamp, complete); } + @Override + public MessageGroup create(MessageGroupStore messageGroupStore, Object groupId) { + if (GroupType.PERSISTENT.equals(this.type)) { + return new PersistentMessageGroup(messageGroupStore, new SimpleMessageGroup(groupId)); + } + else { + return create(messageGroupStore.getMessagesForGroup(groupId), groupId); + } + } + + @Override + public MessageGroup create(MessageGroupStore messageGroupStore, Object groupId, long timestamp, boolean complete) { + if (GroupType.PERSISTENT.equals(this.type)) { + SimpleMessageGroup original = new SimpleMessageGroup(Collections.>emptyList(), groupId, + timestamp, complete); + return new PersistentMessageGroup(messageGroupStore, original); + } + else { + return create(messageGroupStore.getMessagesForGroup(groupId), groupId, timestamp, complete); + } + } + public enum GroupType { BLOCKING_QUEUE { @@ -84,6 +106,14 @@ public class SimpleMessageGroupFactory implements MessageGroupFactory { return Collections.>synchronizedSet(new LinkedHashSet>()); } + }, + PERSISTENT { + + @Override + Collection> get() { + return HASH_SET.get(); + } + }; abstract Collection> get(); diff --git a/spring-integration-core/src/main/java/org/springframework/integration/store/SimpleMessageStore.java b/spring-integration-core/src/main/java/org/springframework/integration/store/SimpleMessageStore.java index d03007c764..c736440a68 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/store/SimpleMessageStore.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/store/SimpleMessageStore.java @@ -239,7 +239,10 @@ public class SimpleMessageStore extends AbstractMessageGroupStore } @Override - public MessageGroup addMessageToGroup(Object groupId, Message message) { + public void addMessagesToGroup(Object groupId, Message... messages) { + Assert.notNull(groupId, "'groupId' must not be null"); + Assert.notNull(messages, "'messages' must not be null"); + Lock lock = this.lockRegistry.obtain(groupId); try { lock.lockInterruptibly(); @@ -247,31 +250,38 @@ public class SimpleMessageStore extends AbstractMessageGroupStore try { UpperBound upperBound; MessageGroup group = this.groupIdToMessageGroup.get(groupId); + MessagingException outOfCapacityException = + new MessagingException(getClass().getSimpleName() + + " was out of capacity (" + this.groupCapacity + ") for group '" + groupId + + "', try constructing it with a larger capacity."); if (group == null) { + if (this.groupCapacity > 0 && messages.length > this.groupCapacity) { + throw outOfCapacityException; + } group = getMessageGroupFactory().create(groupId); this.groupIdToMessageGroup.putIfAbsent(groupId, group); upperBound = new UpperBound(this.groupCapacity); - upperBound.tryAcquire(-1); + for (Message message : messages) { + upperBound.tryAcquire(-1); + group.add(message); + } this.groupToUpperBound.putIfAbsent(groupId, upperBound); } else { upperBound = this.groupToUpperBound.get(groupId); Assert.state(upperBound != null, "'upperBound' must not be null."); - lock.unlock(); - if (!upperBound.tryAcquire(this.upperBoundTimeout)) { - unlocked = true; - throw new MessagingException(this.getClass().getSimpleName() - + " was out of capacity (" - + this.groupCapacity - + ") for group '" - + groupId - + "', try constructing it with a larger capacity."); + for (Message message : messages) { + lock.unlock(); + if (!upperBound.tryAcquire(this.upperBoundTimeout)) { + unlocked = true; + throw outOfCapacityException; + } + lock.lockInterruptibly(); + group.add(message); } - lock.lockInterruptibly(); } - group.add(message); + group.setLastModified(System.currentTimeMillis()); - return group; } finally { if (!unlocked) { @@ -445,6 +455,11 @@ public class SimpleMessageStore extends AbstractMessageGroupStore return getMessageGroup(groupId).getOne(); } + @Override + public Collection> getMessagesForGroup(Object groupId) { + return getMessageGroup(groupId).getMessages(); + } + public void clearMessageGroup(Object groupId) { Lock lock = this.lockRegistry.obtain(groupId); try { diff --git a/spring-integration-core/src/test/java/org/springframework/integration/aggregator/AbstractCorrelatingMessageHandlerTests.java b/spring-integration-core/src/test/java/org/springframework/integration/aggregator/AbstractCorrelatingMessageHandlerTests.java index afbbe69849..e0021f3afb 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/aggregator/AbstractCorrelatingMessageHandlerTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/aggregator/AbstractCorrelatingMessageHandlerTests.java @@ -293,9 +293,8 @@ public class AbstractCorrelatingMessageHandlerTests { Method forceComplete = AbstractCorrelatingMessageHandler.class.getDeclaredMethod("forceComplete", MessageGroup.class); forceComplete.setAccessible(true); - mgs.addMessageToGroup("foo", new GenericMessage("foo")); GenericMessage secondMessage = new GenericMessage("bar"); - mgs.addMessageToGroup("foo", secondMessage); + mgs.addMessagesToGroup("foo", new GenericMessage("foo"), secondMessage); MessageGroup group = mgs.getMessageGroup("foo"); // remove a message mgs.removeMessagesFromGroup("foo", secondMessage); @@ -324,7 +323,7 @@ public class AbstractCorrelatingMessageHandlerTests { QueueChannel outputChannel = new QueueChannel(); handler.setOutputChannel(outputChannel); MessageGroupStore mgs = TestUtils.getPropertyValue(handler, "messageStore", MessageGroupStore.class); - mgs.addMessageToGroup("foo", new GenericMessage("foo")); + mgs.addMessagesToGroup("foo", new GenericMessage("foo")); mgs.completeGroup("foo"); mgs = spy(mgs); new DirectFieldAccessor(handler).setPropertyValue("messageStore", mgs); @@ -357,7 +356,7 @@ public class AbstractCorrelatingMessageHandlerTests { QueueChannel outputChannel = new QueueChannel(); handler.setOutputChannel(outputChannel); MessageGroupStore mgs = TestUtils.getPropertyValue(handler, "messageStore", MessageGroupStore.class); - mgs.addMessageToGroup("foo", new GenericMessage("foo")); + mgs.addMessagesToGroup("foo", new GenericMessage("foo")); MessageGroup group = new SimpleMessageGroup(mgs.getMessageGroup("foo")); mgs.completeGroup("foo"); mgs = spy(mgs); @@ -393,7 +392,7 @@ public class AbstractCorrelatingMessageHandlerTests { QueueChannel outputChannel = new QueueChannel(); handler.setOutputChannel(outputChannel); MessageGroupStore mgs = TestUtils.getPropertyValue(handler, "messageStore", MessageGroupStore.class); - mgs.addMessageToGroup("foo", new GenericMessage("foo")); + mgs.addMessagesToGroup("foo", new GenericMessage("foo")); MessageGroup group = new SimpleMessageGroup(mgs.getMessageGroup("foo")); mgs = spy(mgs); new DirectFieldAccessor(handler).setPropertyValue("messageStore", mgs); diff --git a/spring-integration-core/src/test/java/org/springframework/integration/config/xml/GatewayParserTests.java b/spring-integration-core/src/test/java/org/springframework/integration/config/xml/GatewayParserTests.java index d4c25d5b73..2d2f507510 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/config/xml/GatewayParserTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/config/xml/GatewayParserTests.java @@ -36,7 +36,6 @@ import java.util.concurrent.atomic.AtomicReference; import org.apache.commons.logging.Log; import org.junit.Test; import org.junit.runner.RunWith; -import reactor.rx.Promise; import org.springframework.beans.DirectFieldAccessor; import org.springframework.beans.factory.BeanNameAware; @@ -65,6 +64,8 @@ import org.springframework.test.annotation.DirtiesContext; import org.springframework.test.context.ContextConfiguration; import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; +import reactor.rx.Promise; + /** * @author Mark Fisher * @author Artem Bilan @@ -173,7 +174,7 @@ public class GatewayParserTests { this.startResponder(requestChannel, replyChannel); TestService service = context.getBean("promise", TestService.class); Promise> result = service.promise("foo"); - Message reply = result.await(1, TimeUnit.SECONDS); + Message reply = result.await(10, TimeUnit.SECONDS); assertEquals("foo", reply.getPayload()); assertNotNull(TestUtils.getPropertyValue(context.getBean("&promise"), "asyncExecutor")); } diff --git a/spring-integration-core/src/test/java/org/springframework/integration/store/MessageStoreTests.java b/spring-integration-core/src/test/java/org/springframework/integration/store/MessageStoreTests.java index b0caf0600a..c00e624614 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/store/MessageStoreTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/store/MessageStoreTests.java @@ -34,6 +34,7 @@ import org.springframework.test.util.ReflectionTestUtils; /** * @author Dave Syer * @author Gary Russell + * @author Artem Bilan */ public class MessageStoreTests { @@ -80,7 +81,7 @@ public class MessageStoreTests { assertEquals(1, store.getMessageCountForAllMessageGroups()); } - private static class TestMessageStore extends AbstractMessageGroupStore { + private static class TestMessageStore extends SimpleMessageStore { @SuppressWarnings("unchecked") MessageGroup testMessages = @@ -95,7 +96,7 @@ public class MessageStoreTests { } @Override - public MessageGroup addMessageToGroup(Object correlationKey, Message message) { + public void addMessagesToGroup(Object groupId, Message... messages) { throw new UnsupportedOperationException(); } diff --git a/spring-integration-file/src/main/java/org/springframework/integration/file/WatchServiceDirectoryScanner.java b/spring-integration-file/src/main/java/org/springframework/integration/file/WatchServiceDirectoryScanner.java index da05fdd46c..18bb01e7a6 100644 --- a/spring-integration-file/src/main/java/org/springframework/integration/file/WatchServiceDirectoryScanner.java +++ b/spring-integration-file/src/main/java/org/springframework/integration/file/WatchServiceDirectoryScanner.java @@ -246,7 +246,7 @@ public class WatchServiceDirectoryScanner extends DefaultDirectoryScanner implem if (logger.isDebugEnabled()) { logger.debug("registering: " + dir + " for file creation events"); } - dir.register(this.watcher, StandardWatchEventKinds.ENTRY_CREATE); + dir.register(this.watcher, StandardWatchEventKinds.ENTRY_CREATE, StandardWatchEventKinds.ENTRY_MODIFY); } } diff --git a/spring-integration-gemfire/src/test/java/org/springframework/integration/gemfire/store/GemfireGroupStoreTests.java b/spring-integration-gemfire/src/test/java/org/springframework/integration/gemfire/store/GemfireGroupStoreTests.java index 93f2255021..a48d6ba0bb 100644 --- a/spring-integration-gemfire/src/test/java/org/springframework/integration/gemfire/store/GemfireGroupStoreTests.java +++ b/spring-integration-gemfire/src/test/java/org/springframework/integration/gemfire/store/GemfireGroupStoreTests.java @@ -163,7 +163,7 @@ public class GemfireGroupStoreTests { GemfireMessageStore store = new GemfireMessageStore(this.region); store.afterPropertiesSet(); MessageGroup messageGroup = store.getMessageGroup(1); - store.addMessageToGroup(messageGroup.getGroupId(), new GenericMessage("1")); + store.addMessagesToGroup(messageGroup.getGroupId(), new GenericMessage("1")); store.removeMessagesFromGroup(1, new GenericMessage("2")); } @@ -180,7 +180,7 @@ public class GemfireGroupStoreTests { store.afterPropertiesSet(); MessageGroup messageGroup = store.getMessageGroup(1); Message messageToMark = new GenericMessage("1"); - store.addMessageToGroup(messageGroup.getGroupId(), messageToMark); + store.addMessagesToGroup(messageGroup.getGroupId(), messageToMark); store.completeGroup(messageGroup.getGroupId()); messageGroup = store.getMessageGroup(1); assertTrue(messageGroup.isComplete()); @@ -192,7 +192,7 @@ public class GemfireGroupStoreTests { store.afterPropertiesSet(); MessageGroup messageGroup = store.getMessageGroup(1); Message messageToMark = new GenericMessage("1"); - store.addMessageToGroup(messageGroup.getGroupId(), messageToMark); + store.addMessagesToGroup(messageGroup.getGroupId(), messageToMark); store.setLastReleasedSequenceNumberForGroup(messageGroup.getGroupId(), 5); messageGroup = store.getMessageGroup(1); assertEquals(5, messageGroup.getLastReleasedMessageSequenceNumber()); @@ -207,7 +207,7 @@ public class GemfireGroupStoreTests { store2.afterPropertiesSet(); Message message = new GenericMessage("1"); - store1.addMessageToGroup(1, message); + store1.addMessagesToGroup(1, message); MessageGroup messageGroup = store2.addMessageToGroup(1, new GenericMessage("2")); assertEquals(2, messageGroup.getMessages().size()); @@ -236,7 +236,7 @@ public class GemfireGroupStoreTests { message = MessageHistory.write(message, fooChannel); message = MessageHistory.write(message, barChannel); - store.addMessageToGroup(1, message); + store.addMessagesToGroup(1, message); message = store.getMessageGroup(1).getMessages().iterator().next(); @@ -255,10 +255,9 @@ public class GemfireGroupStoreTests { GemfireMessageStore store2 = new GemfireMessageStore(this.region); store2.afterPropertiesSet(); - store1.addMessageToGroup(1, new GenericMessage("1")); - store2.addMessageToGroup(2, new GenericMessage("2")); - store1.addMessageToGroup(3, new GenericMessage("3")); - store2.addMessageToGroup(3, new GenericMessage("3A")); + store1.addMessagesToGroup(1, new GenericMessage("1")); + store2.addMessagesToGroup(2, new GenericMessage("2")); + store1.addMessagesToGroup(3, new GenericMessage("3"), new GenericMessage("3A")); Iterator messageGroups = store1.iterator(); int counter = 0; diff --git a/spring-integration-gemfire/src/test/java/org/springframework/integration/gemfire/store/GemfireMessageStoreTests.java b/spring-integration-gemfire/src/test/java/org/springframework/integration/gemfire/store/GemfireMessageStoreTests.java index b5051356bd..81709e6287 100644 --- a/spring-integration-gemfire/src/test/java/org/springframework/integration/gemfire/store/GemfireMessageStoreTests.java +++ b/spring-integration-gemfire/src/test/java/org/springframework/integration/gemfire/store/GemfireMessageStoreTests.java @@ -110,7 +110,7 @@ public class GemfireMessageStoreTests { List> messages = new ArrayList>(); for (int i = 0; i < 25; i++) { Message message = MessageBuilder.withPayload("foo").setCorrelationId(groupId).build(); - messageStore.addMessageToGroup(groupId, message); + messageStore.addMessagesToGroup(groupId, message); messages.add(message); } MessageGroup group = messageStore.getMessageGroup(groupId); diff --git a/spring-integration-jdbc/src/main/java/org/springframework/integration/jdbc/JdbcMessageStore.java b/spring-integration-jdbc/src/main/java/org/springframework/integration/jdbc/JdbcMessageStore.java index c3e088ca78..8fdb67f4a7 100644 --- a/spring-integration-jdbc/src/main/java/org/springframework/integration/jdbc/JdbcMessageStore.java +++ b/spring-integration-jdbc/src/main/java/org/springframework/integration/jdbc/JdbcMessageStore.java @@ -20,8 +20,8 @@ import java.sql.PreparedStatement; import java.sql.ResultSet; import java.sql.SQLException; import java.sql.Timestamp; -import java.sql.Types; import java.util.ArrayList; +import java.util.Arrays; import java.util.Collection; import java.util.Date; import java.util.HashMap; @@ -47,6 +47,7 @@ import org.springframework.integration.jdbc.store.JdbcChannelMessageStore; import org.springframework.integration.store.AbstractMessageGroupStore; import org.springframework.integration.store.MessageGroup; import org.springframework.integration.store.MessageStore; +import org.springframework.integration.store.SimpleMessageGroup; import org.springframework.integration.util.UUIDConverter; import org.springframework.jdbc.core.JdbcOperations; import org.springframework.jdbc.core.JdbcTemplate; @@ -98,8 +99,8 @@ public class JdbcMessageStore extends AbstractMessageGroupStore implements Messa GROUP_EXISTS("SELECT COUNT(GROUP_KEY) FROM %PREFIX%MESSAGE_GROUP where GROUP_KEY=? and REGION=?"), CREATE_MESSAGE_GROUP("INSERT into %PREFIX%MESSAGE_GROUP" + - "(GROUP_KEY, REGION, MARKED, COMPLETE, LAST_RELEASED_SEQUENCE, CREATED_DATE, UPDATED_DATE)" - + " values (?, ?, 0, 0, 0, ?, ?)"), + "(GROUP_KEY, REGION, MARKED, COMPLETE, LAST_RELEASED_SEQUENCE, CREATED_DATE, UPDATED_DATE)" + + " values (?, ?, 0, 0, 0, ?, ?)"), UPDATE_MESSAGE_GROUP("UPDATE %PREFIX%MESSAGE_GROUP set UPDATED_DATE=? where GROUP_KEY=? and REGION=?"), @@ -123,13 +124,13 @@ public class JdbcMessageStore extends AbstractMessageGroupStore implements Messa "where %PREFIX%MESSAGE.MESSAGE_ID = " + "(SELECT min(m.MESSAGE_ID) from %PREFIX%MESSAGE m " + "join %PREFIX%GROUP_TO_MESSAGE on m.MESSAGE_ID = %PREFIX%GROUP_TO_MESSAGE.MESSAGE_ID " + - "where CREATED_DATE = " + - "(SELECT min(CREATED_DATE) from %PREFIX%MESSAGE, %PREFIX%GROUP_TO_MESSAGE " + - "where %PREFIX%MESSAGE.MESSAGE_ID = %PREFIX%GROUP_TO_MESSAGE.MESSAGE_ID " + - "and %PREFIX%GROUP_TO_MESSAGE.GROUP_KEY = ? " + - "and %PREFIX%MESSAGE.REGION = ?) " + - "and %PREFIX%GROUP_TO_MESSAGE.GROUP_KEY = ? " + - "and m.REGION = ?)"), + "where CREATED_DATE = " + + "(SELECT min(CREATED_DATE) from %PREFIX%MESSAGE, %PREFIX%GROUP_TO_MESSAGE " + + "where %PREFIX%MESSAGE.MESSAGE_ID = %PREFIX%GROUP_TO_MESSAGE.MESSAGE_ID " + + "and %PREFIX%GROUP_TO_MESSAGE.GROUP_KEY = ? " + + "and %PREFIX%MESSAGE.REGION = ?) " + + "and %PREFIX%GROUP_TO_MESSAGE.GROUP_KEY = ? " + + "and m.REGION = ?)"), GET_GROUP_INFO("SELECT COMPLETE, LAST_RELEASED_SEQUENCE, CREATED_DATE, UPDATED_DATE" + " from %PREFIX%MESSAGE_GROUP where GROUP_KEY = ? and REGION=?"), @@ -183,6 +184,8 @@ public class JdbcMessageStore extends AbstractMessageGroupStore implements Messa */ public static final String CREATED_DATE_KEY = JdbcMessageStore.class.getSimpleName() + ".CREATED_DATE"; + private final MessageMapper mapper = new MessageMapper(); + private volatile String region = "DEFAULT"; private volatile String tablePrefix = DEFAULT_TABLE_PREFIX; @@ -195,8 +198,6 @@ public class JdbcMessageStore extends AbstractMessageGroupStore implements Messa private volatile LobHandler lobHandler = new DefaultLobHandler(); - private volatile MessageMapper mapper = new MessageMapper(); - private volatile Map queryCache = new HashMap(); /** @@ -299,8 +300,7 @@ public class JdbcMessageStore extends AbstractMessageGroupStore implements Messa if (message == null) { return null; } - int updated = this.jdbcTemplate.update(getQuery(Query.DELETE_MESSAGE), new Object[] { getKey(id), this.region }, new int[] { - Types.VARCHAR, Types.VARCHAR }); + int updated = this.jdbcTemplate.update(getQuery(Query.DELETE_MESSAGE), getKey(id), this.region); if (updated != 0) { return message; } @@ -315,7 +315,8 @@ public class JdbcMessageStore extends AbstractMessageGroupStore implements Messa @Override public Message getMessage(UUID id) { - List> list = this.jdbcTemplate.query(getQuery(Query.GET_MESSAGE), new Object[] { getKey(id), this.region }, this.mapper); + List> list = + this.jdbcTemplate.query(getQuery(Query.GET_MESSAGE), this.mapper, getKey(id), this.region); if (list.isEmpty()) { return null; } @@ -346,6 +347,7 @@ public class JdbcMessageStore extends AbstractMessageGroupStore implements Messa final byte[] messageBytes = this.serializer.convert(result); this.jdbcTemplate.update(getQuery(Query.CREATE_MESSAGE), new PreparedStatementSetter() { + @Override public void setValues(PreparedStatement ps) throws SQLException { if (logger.isDebugEnabled()) { @@ -361,45 +363,52 @@ public class JdbcMessageStore extends AbstractMessageGroupStore implements Messa } @Override - public MessageGroup addMessageToGroup(Object groupId, Message message) { + public void addMessagesToGroup(Object groupId, Message... messages) { final String groupKey = getKey(groupId); - final String messageId = getKey(message.getHeaders().getId()); - boolean groupNotExist = this.jdbcTemplate.queryForObject(this.getQuery(Query.GROUP_EXISTS), Integer.class, groupKey, this.region) < 1; + boolean groupNotExist = this.jdbcTemplate.queryForObject(this.getQuery(Query.GROUP_EXISTS), + Integer.class, groupKey, this.region) < 1; final Timestamp updatedDate = new Timestamp(System.currentTimeMillis()); final Timestamp createdDate = groupNotExist ? updatedDate : - this.jdbcTemplate.queryForObject(getQuery(Query.GET_GROUP_CREATED_DATE), new Object[] { groupKey, this.region}, Timestamp.class); + this.jdbcTemplate.queryForObject(getQuery(Query.GET_GROUP_CREATED_DATE), Timestamp.class, groupKey, + this.region); if (groupNotExist) { try { - this.doCreateMessageGroup(groupKey, createdDate); + doCreateMessageGroup(groupKey, createdDate); } catch (DuplicateKeyException e) { logger.warn("Lost race to create group; attempting update instead", e); - this.doUpdateMessageGroup(groupKey, updatedDate); + doUpdateMessageGroup(groupKey, updatedDate); } } else { - this.doUpdateMessageGroup(groupKey, updatedDate); + doUpdateMessageGroup(groupKey, updatedDate); } - this.addMessage(message); + for (Message message : messages) { + addMessage(message); + } + this.jdbcTemplate.batchUpdate(getQuery(Query.CREATE_GROUP_TO_MESSAGE), + Arrays.asList(messages), + 100, + new ParameterizedPreparedStatementSetter>() { - this.jdbcTemplate.update(getQuery(Query.CREATE_GROUP_TO_MESSAGE), new PreparedStatementSetter() { - @Override - public void setValues(PreparedStatement ps) throws SQLException { - if (logger.isDebugEnabled()) { - logger.debug("Inserting message with id key=" + messageId + " and created date=" + createdDate); - } - ps.setString(1, groupKey); - ps.setString(2, messageId); - ps.setString(3, JdbcMessageStore.this.region); - } - }); - return getMessageGroup(groupId); + @Override + public void setValues(PreparedStatement ps, Message messageToAdd) throws SQLException { + String messageId = getKey(messageToAdd.getHeaders().getId()); + if (logger.isDebugEnabled()) { + logger.debug("Inserting message with id key=" + messageId + + " and created date=" + createdDate); + } + ps.setString(1, groupKey); + ps.setString(2, messageId); + ps.setString(3, JdbcMessageStore.this.region); + } + }); } @Override @@ -411,14 +420,16 @@ public class JdbcMessageStore extends AbstractMessageGroupStore implements Messa @Override @ManagedAttribute public int getMessageCountForAllMessageGroups() { - return this.jdbcTemplate.queryForObject(getQuery(Query.COUNT_ALL_MESSAGES_IN_GROUPS), Integer.class, this.region); + return this.jdbcTemplate.queryForObject(getQuery(Query.COUNT_ALL_MESSAGES_IN_GROUPS), + Integer.class, this.region); } @Override @ManagedAttribute public int messageGroupSize(Object groupId) { String key = getKey(groupId); - return this.jdbcTemplate.queryForObject(getQuery(Query.COUNT_ALL_MESSAGES_IN_GROUP), Integer.class, key, this.region); + return this.jdbcTemplate.queryForObject(getQuery(Query.COUNT_ALL_MESSAGES_IN_GROUP), + Integer.class, key, this.region); } @Override @@ -429,43 +440,29 @@ public class JdbcMessageStore extends AbstractMessageGroupStore implements Messa final AtomicReference completeFlag = new AtomicReference(); final AtomicReference lastReleasedSequenceRef = new AtomicReference(); - List> messages = this.jdbcTemplate.query(getQuery(Query.LIST_MESSAGES_BY_GROUP_KEY), - new Object[] { key, this.region }, this.mapper); + this.jdbcTemplate.query(getQuery(Query.GET_GROUP_INFO), new RowCallbackHandler() { - this.jdbcTemplate.query(getQuery(Query.GET_GROUP_INFO), new Object[] { key, this.region}, - new RowCallbackHandler() { + @Override + public void processRow(ResultSet rs) throws SQLException { + updateDate.set(rs.getTimestamp("UPDATED_DATE")); - @Override - public void processRow(ResultSet rs) throws SQLException { - updateDate.set(rs.getTimestamp("UPDATED_DATE")); + createDate.set(rs.getTimestamp("CREATED_DATE")); - createDate.set(rs.getTimestamp("CREATED_DATE")); + completeFlag.set(rs.getInt("COMPLETE") > 0); - completeFlag.set(rs.getInt("COMPLETE") > 0); + lastReleasedSequenceRef.set(rs.getInt("LAST_RELEASED_SEQUENCE")); + } - lastReleasedSequenceRef.set(rs.getInt("LAST_RELEASED_SEQUENCE")); - } - - }); + }, key, this.region); if (createDate.get() == null && updateDate.get() == null) { - if (logger.isWarnEnabled()) { - for (Message message : messages) { - logger.warn("Missing group row for message id: " + message.getHeaders().getId()); - } - } - return getMessageGroupFactory().create(groupId); + return new SimpleMessageGroup(groupId); } - long timestamp = createDate.get().getTime(); - boolean complete = completeFlag.get(); - long lastModified = updateDate.get().getTime(); - int lastReleasedSequenceNumber = lastReleasedSequenceRef.get(); - MessageGroup messageGroup = getMessageGroupFactory() - .create(messages, groupId, timestamp, complete); - messageGroup.setLastModified(lastModified); - messageGroup.setLastReleasedMessageSequenceNumber(lastReleasedSequenceNumber); + .create(this, groupId, createDate.get().getTime(), completeFlag.get()); + messageGroup.setLastModified(updateDate.get().getTime()); + messageGroup.setLastReleasedMessageSequenceNumber(lastReleasedSequenceRef.get()); return messageGroup; } @@ -476,6 +473,7 @@ public class JdbcMessageStore extends AbstractMessageGroupStore implements Messa final String messageId = getKey(messageToRemove.getHeaders().getId()); this.jdbcTemplate.update(getQuery(Query.REMOVE_MESSAGE_FROM_GROUP), new PreparedStatementSetter() { + @Override public void setValues(PreparedStatement ps) throws SQLException { if (logger.isDebugEnabled()) { @@ -486,8 +484,8 @@ public class JdbcMessageStore extends AbstractMessageGroupStore implements Messa ps.setString(3, JdbcMessageStore.this.region); } }); - this.removeMessage(messageToRemove.getHeaders().getId()); - this.updateMessageGroup(groupKey); + removeMessage(messageToRemove.getHeaders().getId()); + updateMessageGroup(groupKey); return getMessageGroup(groupId); } @@ -505,23 +503,27 @@ public class JdbcMessageStore extends AbstractMessageGroupStore implements Messa messages, getRemoveBatchSize(), new ParameterizedPreparedStatementSetter>() { + @Override public void setValues(PreparedStatement ps, Message messageToRemove) throws SQLException { ps.setString(1, groupKey); ps.setString(2, getKey(messageToRemove.getHeaders().getId())); ps.setString(3, JdbcMessageStore.this.region); } - }); + + }); this.jdbcTemplate.batchUpdate(getQuery(Query.DELETE_MESSAGE), messages, getRemoveBatchSize(), new ParameterizedPreparedStatementSetter>() { + @Override public void setValues(PreparedStatement ps, Message messageToRemove) throws SQLException { ps.setString(1, getKey(messageToRemove.getHeaders().getId())); ps.setString(2, JdbcMessageStore.this.region); } - }); + + }); this.updateMessageGroup(groupKey); } @@ -535,6 +537,7 @@ public class JdbcMessageStore extends AbstractMessageGroupStore implements Messa } this.jdbcTemplate.update(getQuery(Query.REMOVE_GROUP_TO_MESSAGE_JOIN), new PreparedStatementSetter() { + @Override public void setValues(PreparedStatement ps) throws SQLException { if (logger.isDebugEnabled()) { @@ -546,6 +549,7 @@ public class JdbcMessageStore extends AbstractMessageGroupStore implements Messa }); this.jdbcTemplate.update(getQuery(Query.DELETE_MESSAGE_GROUP), new PreparedStatementSetter() { + @Override public void setValues(PreparedStatement ps) throws SQLException { if (logger.isDebugEnabled()) { @@ -563,6 +567,7 @@ public class JdbcMessageStore extends AbstractMessageGroupStore implements Messa final String groupKey = getKey(groupId); this.jdbcTemplate.update(getQuery(Query.COMPLETE_GROUP), new PreparedStatementSetter() { + @Override public void setValues(PreparedStatement ps) throws SQLException { if (logger.isDebugEnabled()) { @@ -582,10 +587,12 @@ public class JdbcMessageStore extends AbstractMessageGroupStore implements Messa final String groupKey = getKey(groupId); this.jdbcTemplate.update(getQuery(Query.UPDATE_LAST_RELEASED_SEQUENCE), new PreparedStatementSetter() { + @Override public void setValues(PreparedStatement ps) throws SQLException { if (logger.isDebugEnabled()) { - logger.debug("Updating the sequence number of the last released Message in the MessageGroup: " + groupKey); + logger.debug("Updating the sequence number of the last released Message in the MessageGroup: " + + groupKey); } ps.setTimestamp(1, new Timestamp(updatedDate)); ps.setInt(2, sequenceNumber); @@ -607,11 +614,23 @@ public class JdbcMessageStore extends AbstractMessageGroupStore implements Messa return polledMessage; } + @Override + public Message getOneMessageFromGroup(Object groupId) { + return doPollForMessage(getKey(groupId)); + } + + @Override + public Collection> getMessagesForGroup(Object groupId) { + return this.jdbcTemplate.query(getQuery(Query.LIST_MESSAGES_BY_GROUP_KEY), this.mapper, getKey(groupId), + this.region); + } + @Override public Iterator iterator() { - final Iterator iterator = this.jdbcTemplate.query(getQuery(Query.LIST_GROUP_KEYS), new Object[] { this.region }, - new SingleColumnRowMapper()).iterator(); + final Iterator iterator = this.jdbcTemplate.query(getQuery(Query.LIST_GROUP_KEYS), + new SingleColumnRowMapper(), this.region) + .iterator(); return new Iterator() { @@ -663,14 +682,13 @@ public class JdbcMessageStore extends AbstractMessageGroupStore implements Messa /** * This method executes a call to the DB to get the oldest Message in the MessageGroup - * Override this method if need to. For example if you DB supports advanced function such as FIRST etc. - * + * Override this method if need to. For example if your DB supports advanced function such as FIRST etc. * @param groupIdKey String representation of message group ID * @return a message; could be null if query produced no Messages */ protected Message doPollForMessage(String groupIdKey) { - List> messages = this.jdbcTemplate.query(getQuery(Query.POLL_FROM_GROUP), - new Object[] { groupIdKey, this.region, groupIdKey, this.region }, this.mapper); + List> messages = this.jdbcTemplate.query(getQuery(Query.POLL_FROM_GROUP), this.mapper, + groupIdKey, this.region, groupIdKey, this.region); Assert.isTrue(messages.size() == 0 || messages.size() == 1); if (messages.size() > 0) { return messages.get(0); @@ -680,6 +698,7 @@ public class JdbcMessageStore extends AbstractMessageGroupStore implements Messa private void doCreateMessageGroup(final String groupKey, final Timestamp createdDate) { this.jdbcTemplate.update(getQuery(Query.CREATE_MESSAGE_GROUP), new PreparedStatementSetter() { + @Override public void setValues(PreparedStatement ps) throws SQLException { if (logger.isDebugEnabled()) { @@ -695,6 +714,7 @@ public class JdbcMessageStore extends AbstractMessageGroupStore implements Messa private void doUpdateMessageGroup(final String groupKey, final Timestamp updatedDate) { this.jdbcTemplate.update(getQuery(Query.UPDATE_MESSAGE_GROUP), new PreparedStatementSetter() { + @Override public void setValues(PreparedStatement ps) throws SQLException { if (logger.isDebugEnabled()) { @@ -709,6 +729,7 @@ public class JdbcMessageStore extends AbstractMessageGroupStore implements Messa private void updateMessageGroup(final String groupId) { this.jdbcTemplate.update(getQuery(Query.UPDATE_GROUP), new PreparedStatementSetter() { + @Override public void setValues(PreparedStatement ps) throws SQLException { if (logger.isDebugEnabled()) { @@ -726,15 +747,14 @@ public class JdbcMessageStore extends AbstractMessageGroupStore implements Messa final List messageIds = new ArrayList(); - this.jdbcTemplate.query(getQuery(Query.LIST_MESSAGEIDS_BY_GROUP_KEY), new Object[] { key, this.region }, + this.jdbcTemplate.query(getQuery(Query.LIST_MESSAGEIDS_BY_GROUP_KEY), new RowCallbackHandler() { @Override public void processRow(ResultSet rs) throws SQLException { messageIds.add(UUID.fromString(rs.getString(1))); } - } - ); + }, key, this.region); return messageIds; } diff --git a/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/DelayerHandlerRescheduleIntegrationTests-context.xml b/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/DelayerHandlerRescheduleIntegrationTests-context.xml index f72f70736c..9fef2edb78 100644 --- a/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/DelayerHandlerRescheduleIntegrationTests-context.xml +++ b/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/DelayerHandlerRescheduleIntegrationTests-context.xml @@ -7,7 +7,9 @@ http://www.springframework.org/schema/integration http://www.springframework.org/schema/integration/spring-integration.xsd"> + class="org.springframework.integration.jdbc.DelayerHandlerRescheduleIntegrationTests$TestJdbcMessageStore"> + + messageInStore = messageGroup.getMessages().iterator().next(); Object payload = messageInStore.getPayload(); @@ -140,7 +147,8 @@ public class DelayerHandlerRescheduleIntegrationTests { @Test //INT-2649 public void testRollbackOnDelayerHandlerReleaseTask() throws Exception { - AbstractApplicationContext context = new ClassPathXmlApplicationContext("DelayerHandlerRescheduleIntegrationTests-context.xml", this.getClass()); + AbstractApplicationContext context = + new ClassPathXmlApplicationContext("DelayerHandlerRescheduleIntegrationTests-context.xml", getClass()); MessageChannel input = context.getBean("transactionalDelayerInput", MessageChannel.class); MessageGroupStore messageStore = context.getBean("messageStore", MessageGroupStore.class); diff --git a/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/JdbcMessageStoreTests.java b/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/JdbcMessageStoreTests.java index 4d7629fc5c..ed11cc36e7 100644 --- a/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/JdbcMessageStoreTests.java +++ b/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/JdbcMessageStoreTests.java @@ -30,15 +30,17 @@ import java.io.IOException; import java.io.InputStream; import java.io.InputStreamReader; import java.io.OutputStream; +import java.sql.PreparedStatement; +import java.sql.SQLException; +import java.sql.Timestamp; import java.util.ArrayList; import java.util.List; import java.util.Properties; import java.util.UUID; +import java.util.concurrent.CountDownLatch; import javax.sql.DataSource; -import org.apache.commons.logging.Log; -import org.apache.commons.logging.LogFactory; import org.junit.Before; import org.junit.Test; import org.junit.runner.RunWith; @@ -55,6 +57,7 @@ import org.springframework.integration.store.MessageGroupStore.MessageGroupCallb import org.springframework.integration.support.MessageBuilder; import org.springframework.integration.util.UUIDConverter; import org.springframework.jdbc.core.JdbcTemplate; +import org.springframework.jdbc.core.PreparedStatementSetter; import org.springframework.messaging.Message; import org.springframework.messaging.support.GenericMessage; import org.springframework.test.annotation.DirtiesContext; @@ -74,10 +77,9 @@ import org.springframework.transaction.annotation.Transactional; @ContextConfiguration @RunWith(SpringJUnit4ClassRunner.class) @DirtiesContext // close at the end after class +@Transactional public class JdbcMessageStoreTests { - private static final Log LOG = LogFactory.getLog(JdbcMessageStoreTests.class); - @Autowired private DataSource dataSource; @@ -89,14 +91,12 @@ public class JdbcMessageStoreTests { } @Test - @Transactional public void testGetNonExistent() throws Exception { Message result = messageStore.getMessage(UUID.randomUUID()); assertNull(result); } @Test - @Transactional public void testAddAndGet() throws Exception { Message message = MessageBuilder.withPayload("foo").build(); Message saved = messageStore.addMessage(message); @@ -109,7 +109,6 @@ public class JdbcMessageStoreTests { } @Test - @Transactional public void testWithMessageHistory() throws Exception { Message message = new GenericMessage("Hello"); @@ -131,7 +130,6 @@ public class JdbcMessageStoreTests { } @Test - @Transactional public void testSize() throws Exception { Message message = MessageBuilder.withPayload("foo").build(); messageStore.addMessage(message); @@ -139,7 +137,6 @@ public class JdbcMessageStoreTests { } @Test - @Transactional public void testSerializer() throws Exception { // N.B. these serializers are not realistic (just for test purposes) messageStore.setSerializer(new Serializer>() { @@ -165,7 +162,6 @@ public class JdbcMessageStoreTests { } @Test - @Transactional public void testAddAndGetWithDifferentRegion() throws Exception { Message message = MessageBuilder.withPayload("foo").build(); Message saved = messageStore.addMessage(message); @@ -175,17 +171,16 @@ public class JdbcMessageStoreTests { } @Test - @Transactional public void testAddAndUpdate() throws Exception { - Message message = MessageBuilder.withPayload("foo").setCorrelationId("X").build(); + Message message = MessageBuilder.withPayload("foo").setCorrelationId("X").build(); message = messageStore.addMessage(message); message = MessageBuilder.fromMessage(message).setCorrelationId("Y").build(); message = messageStore.addMessage(message); - assertEquals("Y", new IntegrationMessageHeaderAccessor(messageStore.getMessage(message.getHeaders().getId())).getCorrelationId()); + message = messageStore.getMessage(message.getHeaders().getId()); + assertEquals("Y", new IntegrationMessageHeaderAccessor(message).getCorrelationId()); } @Test - @Transactional public void testAddAndUpdateAlreadySaved() throws Exception { Message message = MessageBuilder.withPayload("foo").build(); message = messageStore.addMessage(message); @@ -194,7 +189,6 @@ public class JdbcMessageStoreTests { } @Test - @Transactional public void testAddAndUpdateAlreadySavedAndCopied() throws Exception { Message message = MessageBuilder.withPayload("foo").build(); Message saved = messageStore.addMessage(message); @@ -206,7 +200,6 @@ public class JdbcMessageStoreTests { } @Test - @Transactional public void testAddAndUpdateWithChange() throws Exception { Message message = MessageBuilder.withPayload("foo").build(); Message saved = messageStore.addMessage(message); @@ -218,7 +211,6 @@ public class JdbcMessageStoreTests { } @Test - @Transactional public void testAddAndRemoveMessageGroup() throws Exception { Message message = MessageBuilder.withPayload("foo").build(); message = messageStore.addMessage(message); @@ -226,40 +218,36 @@ public class JdbcMessageStoreTests { } @Test - @Transactional public void testAddAndGetMessageGroup() throws Exception { String groupId = "X"; Message message = MessageBuilder.withPayload("foo").setCorrelationId(groupId).build(); long now = System.currentTimeMillis(); - messageStore.addMessageToGroup(groupId, message); + messageStore.addMessagesToGroup(groupId, message); MessageGroup group = messageStore.getMessageGroup(groupId); assertEquals(1, group.size()); assertTrue("Timestamp too early: " + group.getTimestamp() + "<" + now, group.getTimestamp() >= now); } @Test - @Transactional public void testAddAndRemoveMessageFromMessageGroup() throws Exception { String groupId = "X"; Message message = MessageBuilder.withPayload("foo").setCorrelationId(groupId).build(); - messageStore.addMessageToGroup(groupId, message); + messageStore.addMessagesToGroup(groupId, message); messageStore.removeMessagesFromGroup(groupId, message); MessageGroup group = messageStore.getMessageGroup(groupId); assertEquals(0, group.size()); } @Test - @Transactional - @DirtiesContext public void testAddAndRemoveMessagesFromMessageGroup() throws Exception { String groupId = "X"; messageStore.setRemoveBatchSize(10); List> messages = new ArrayList>(); for (int i = 0; i < 25; i++) { Message message = MessageBuilder.withPayload("foo").setCorrelationId(groupId).build(); - messageStore.addMessageToGroup(groupId, message); messages.add(message); } + messageStore.addMessagesToGroup(groupId, messages.toArray(new Message[messages.size()])); MessageGroup group = messageStore.getMessageGroup(groupId); assertEquals(25, group.size()); messageStore.removeMessagesFromGroup(groupId, messages); @@ -268,29 +256,27 @@ public class JdbcMessageStoreTests { } @Test - @Transactional public void testRemoveMessageGroup() throws Exception { - JdbcTemplate template = new JdbcTemplate(dataSource); + JdbcTemplate template = new JdbcTemplate(this.dataSource); template.afterPropertiesSet(); String groupId = "X"; Message message = MessageBuilder.withPayload("foo").setCorrelationId(groupId).build(); - messageStore.addMessageToGroup(groupId, message); + messageStore.addMessagesToGroup(groupId, message); messageStore.removeMessageGroup(groupId); MessageGroup group = messageStore.getMessageGroup(groupId); assertEquals(0, group.size()); String uuidGroupId = UUIDConverter.getUUID(groupId).toString(); assertTrue(template.queryForList( - "SELECT * from INT_GROUP_TO_MESSAGE where GROUP_KEY = '" + uuidGroupId + "'").size() == 0); + "SELECT * from INT_GROUP_TO_MESSAGE where GROUP_KEY = ?", uuidGroupId).size() == 0); } @Test - @Transactional public void testCompleteMessageGroup() throws Exception { String groupId = "X"; Message message = MessageBuilder.withPayload("foo").setCorrelationId(groupId).build(); - messageStore.addMessageToGroup(groupId, message); + messageStore.addMessagesToGroup(groupId, message); messageStore.completeGroup(groupId); MessageGroup group = messageStore.getMessageGroup(groupId); assertTrue(group.isComplete()); @@ -298,112 +284,150 @@ public class JdbcMessageStoreTests { } @Test - @Transactional public void testUpdateLastReleasedSequence() throws Exception { String groupId = "X"; Message message = MessageBuilder.withPayload("foo").setCorrelationId(groupId).build(); - messageStore.addMessageToGroup(groupId, message); + messageStore.addMessagesToGroup(groupId, message); messageStore.setLastReleasedSequenceNumberForGroup(groupId, 5); MessageGroup group = messageStore.getMessageGroup(groupId); assertEquals(5, group.getLastReleasedMessageSequenceNumber()); } @Test - @Transactional public void testMessageGroupCount() throws Exception { String groupId = "X"; Message message = MessageBuilder.withPayload("foo").build(); - messageStore.addMessageToGroup(groupId, message); + messageStore.addMessagesToGroup(groupId, message); assertEquals(1, messageStore.getMessageGroupCount()); } @Test - @Transactional public void testMessageGroupSizes() throws Exception { String groupId = "X"; Message message = MessageBuilder.withPayload("foo").build(); - messageStore.addMessageToGroup(groupId, message); + messageStore.addMessagesToGroup(groupId, message); assertEquals(1, messageStore.getMessageCountForAllMessageGroups()); } @Test - @Transactional public void testOrderInMessageGroup() throws Exception { String groupId = "X"; - messageStore.addMessageToGroup(groupId, MessageBuilder.withPayload("foo").setCorrelationId(groupId).build()); + this.messageStore.addMessagesToGroup(groupId, + MessageBuilder.withPayload("foo").setCorrelationId(groupId).build()); Thread.sleep(1); - messageStore.addMessageToGroup(groupId, MessageBuilder.withPayload("bar").setCorrelationId(groupId).build()); - MessageGroup group = messageStore.getMessageGroup(groupId); + this.messageStore.addMessagesToGroup(groupId, + MessageBuilder.withPayload("bar").setCorrelationId(groupId).build()); + MessageGroup group = this.messageStore.getMessageGroup(groupId); assertEquals(2, group.size()); - assertEquals("foo", messageStore.pollMessageFromGroup(groupId).getPayload()); - assertEquals("bar", messageStore.pollMessageFromGroup(groupId).getPayload()); + assertEquals("foo", this.messageStore.pollMessageFromGroup(groupId).getPayload()); + assertEquals("bar", this.messageStore.pollMessageFromGroup(groupId).getPayload()); } @Test - @Transactional public void testExpireMessageGroupOnCreateOnly() throws Exception { - String groupId = "X"; + final String groupId = "X"; Message message = MessageBuilder.withPayload("foo").setCorrelationId(groupId).build(); - messageStore.addMessageToGroup(groupId, message); + messageStore.addMessagesToGroup(groupId, message); + final CountDownLatch groupRemovalLatch = new CountDownLatch(1); messageStore.registerMessageGroupExpiryCallback(new MessageGroupCallback() { + @Override public void execute(MessageGroupStore messageGroupStore, MessageGroup group) { messageGroupStore.removeMessageGroup(group.getGroupId()); } + }); - Thread.sleep(1000); + messageStore.expireMessageGroups(2000); MessageGroup group = messageStore.getMessageGroup(groupId); assertEquals(1, group.size()); - messageStore.addMessageToGroup(groupId, MessageBuilder.withPayload("bar").setCorrelationId(groupId).build()); - Thread.sleep(2001); + + messageStore.addMessagesToGroup(groupId, MessageBuilder.withPayload("bar").setCorrelationId(groupId).build()); + + JdbcTemplate template = new JdbcTemplate(this.dataSource); + template.afterPropertiesSet(); + + template.update("UPDATE INT_MESSAGE_GROUP set CREATED_DATE=? where GROUP_KEY=? and REGION=?", + new PreparedStatementSetter() { + + @Override + public void setValues(PreparedStatement ps) throws SQLException { + ps.setTimestamp(1, new Timestamp(System.currentTimeMillis() - 10000)); + ps.setString(2, UUIDConverter.getUUID(groupId).toString()); + ps.setString(3, "DEFAULT"); + } + + }); + messageStore.expireMessageGroups(2000); + group = messageStore.getMessageGroup(groupId); assertEquals(0, group.size()); } @Test - @Transactional public void testExpireMessageGroupOnIdleOnly() throws Exception { String groupId = "X"; Message message = MessageBuilder.withPayload("foo").setCorrelationId(groupId).build(); messageStore.setTimeoutOnIdle(true); - messageStore.addMessageToGroup(groupId, message); + messageStore.addMessagesToGroup(groupId, message); messageStore.registerMessageGroupExpiryCallback(new MessageGroupCallback() { + @Override public void execute(MessageGroupStore messageGroupStore, MessageGroup group) { messageGroupStore.removeMessageGroup(group.getGroupId()); } + }); - Thread.sleep(1000); + + JdbcTemplate template = new JdbcTemplate(this.dataSource); + template.afterPropertiesSet(); + + updateMessageGroup(template, groupId, 1000); + messageStore.expireMessageGroups(2000); MessageGroup group = messageStore.getMessageGroup(groupId); assertEquals(1, group.size()); - Thread.sleep(2000); - messageStore.addMessageToGroup(groupId, MessageBuilder.withPayload("bar").setCorrelationId(groupId).build()); + + updateMessageGroup(template, groupId, 2000); + + messageStore.addMessagesToGroup(groupId, MessageBuilder.withPayload("bar").setCorrelationId(groupId).build()); group = messageStore.getMessageGroup(groupId); assertEquals(2, group.size()); - Thread.sleep(2000); + + updateMessageGroup(template, groupId, 2000); + messageStore.expireMessageGroups(2000); group = messageStore.getMessageGroup(groupId); assertEquals(0, group.size()); } + private void updateMessageGroup(JdbcTemplate template, final String groupId, final long timeout) { + template.update("UPDATE INT_MESSAGE_GROUP set UPDATED_DATE=? where GROUP_KEY=? and REGION=?", + new PreparedStatementSetter() { + + @Override + public void setValues(PreparedStatement ps) throws SQLException { + ps.setTimestamp(1, new Timestamp(System.currentTimeMillis() - timeout)); + ps.setString(2, UUIDConverter.getUUID(groupId).toString()); + ps.setString(3, "DEFAULT"); + } + + }); + + } + @Test - @Transactional public void testMessagePollingFromTheGroup() throws Exception { String groupId = "X"; - messageStore.addMessageToGroup(groupId, MessageBuilder.withPayload("foo").setCorrelationId(groupId).build()); - Thread.sleep(1); - messageStore.addMessageToGroup(groupId, MessageBuilder.withPayload("bar").setCorrelationId(groupId).build()); - Thread.sleep(1); - messageStore.addMessageToGroup(groupId, MessageBuilder.withPayload("baz").setCorrelationId(groupId).build()); + messageStore.addMessagesToGroup(groupId, MessageBuilder.withPayload("foo").setCorrelationId(groupId).build()); + messageStore.addMessagesToGroup(groupId, MessageBuilder.withPayload("bar").setCorrelationId(groupId).build()); + messageStore.addMessagesToGroup(groupId, MessageBuilder.withPayload("baz").setCorrelationId(groupId).build()); - messageStore.addMessageToGroup("Y", MessageBuilder.withPayload("barA").setCorrelationId(groupId).build()); - Thread.sleep(1); - messageStore.addMessageToGroup("Y", MessageBuilder.withPayload("bazA").setCorrelationId(groupId).build()); + messageStore.addMessagesToGroup("Y", MessageBuilder.withPayload("barA").setCorrelationId(groupId).build(), + MessageBuilder.withPayload("bazA").setCorrelationId(groupId).build()); MessageGroup group = messageStore.getMessageGroup("X"); assertEquals(3, group.size()); @@ -424,7 +448,6 @@ public class JdbcMessageStoreTests { } @Test - @Transactional public void testSameMessageToMultipleGroups() throws Exception { final String group1Id = "group1"; @@ -450,16 +473,12 @@ public class JdbcMessageStoreTests { assertNotNull(messageFromGroup1); assertNotNull(messageFromGroup2); - LOG.info("messageFromGroup1: " + messageFromGroup1.getHeaders().getId() + "; Sequence #: " + new IntegrationMessageHeaderAccessor(messageFromGroup1).getSequenceNumber()); - LOG.info("messageFromGroup2: " + messageFromGroup2.getHeaders().getId() + "; Sequence #: " + new IntegrationMessageHeaderAccessor(messageFromGroup1).getSequenceNumber()); - - assertEquals(Integer.valueOf(1), messageFromGroup1.getHeaders().get(IntegrationMessageHeaderAccessor.SEQUENCE_NUMBER)); - assertEquals(Integer.valueOf(2), messageFromGroup2.getHeaders().get(IntegrationMessageHeaderAccessor.SEQUENCE_NUMBER)); + assertEquals(1, messageFromGroup1.getHeaders().get(IntegrationMessageHeaderAccessor.SEQUENCE_NUMBER)); + assertEquals(2, messageFromGroup2.getHeaders().get(IntegrationMessageHeaderAccessor.SEQUENCE_NUMBER)); } @Test - @Transactional public void testSameMessageAndGroupToMultipleRegions() throws Exception { final String groupId = "myGroup"; @@ -492,16 +511,12 @@ public class JdbcMessageStoreTests { assertNotNull(messageFromRegion1); assertNotNull(messageFromRegion2); - LOG.info("messageFromRegion1: " + messageFromRegion1.getHeaders().getId() + "; Sequence #: " + new IntegrationMessageHeaderAccessor(messageFromRegion1).getSequenceNumber()); - LOG.info("messageFromRegion2: " + messageFromRegion2.getHeaders().getId() + "; Sequence #: " + new IntegrationMessageHeaderAccessor(messageFromRegion1).getSequenceNumber()); - - assertEquals(Integer.valueOf(1), messageFromRegion1.getHeaders().get(IntegrationMessageHeaderAccessor.SEQUENCE_NUMBER)); - assertEquals(Integer.valueOf(2), messageFromRegion2.getHeaders().get(IntegrationMessageHeaderAccessor.SEQUENCE_NUMBER)); + assertEquals(1, messageFromRegion1.getHeaders().get(IntegrationMessageHeaderAccessor.SEQUENCE_NUMBER)); + assertEquals(2, messageFromRegion2.getHeaders().get(IntegrationMessageHeaderAccessor.SEQUENCE_NUMBER)); } @Test - @Transactional public void testCompletedNotExpiredGroupINT3037() throws Exception { /* * based on the aggregator scenario as follows; @@ -518,11 +533,18 @@ public class JdbcMessageStoreTests { //init String groupId = "group"; //build the messages - Message oneOfTwo = MessageBuilder.withPayload("hello").setSequenceNumber(1).setSequenceSize(2).setCorrelationId(groupId).build(); - Message twoOfTwo = MessageBuilder.withPayload("world").setSequenceNumber(2).setSequenceSize(2).setCorrelationId(groupId).build(); + Message oneOfTwo = MessageBuilder.withPayload("hello") + .setSequenceNumber(1) + .setSequenceSize(2) + .setCorrelationId(groupId) + .build(); + Message twoOfTwo = MessageBuilder.withPayload("world") + .setSequenceNumber(2) + .setSequenceSize(2) + .setCorrelationId(groupId) + .build(); //add to the messageStore - messageStore.addMessageToGroup(groupId, oneOfTwo); - messageStore.addMessageToGroup(groupId, twoOfTwo); + messageStore.addMessagesToGroup(groupId, oneOfTwo, twoOfTwo); //check that 2 messages are there assertTrue(messageStore.getMessageGroupCount() == 1); assertTrue(messageStore.getMessageCount() == 2); diff --git a/spring-integration-mongodb/src/main/java/org/springframework/integration/mongodb/store/AbstractConfigurableMongoDbMessageStore.java b/spring-integration-mongodb/src/main/java/org/springframework/integration/mongodb/store/AbstractConfigurableMongoDbMessageStore.java index 470ee4b692..95a2c72d40 100644 --- a/spring-integration-mongodb/src/main/java/org/springframework/integration/mongodb/store/AbstractConfigurableMongoDbMessageStore.java +++ b/spring-integration-mongodb/src/main/java/org/springframework/integration/mongodb/store/AbstractConfigurableMongoDbMessageStore.java @@ -17,7 +17,9 @@ package org.springframework.integration.mongodb.store; import java.util.ArrayList; +import java.util.Collection; import java.util.HashSet; +import java.util.Iterator; import java.util.List; import java.util.Map; import java.util.Set; @@ -49,8 +51,9 @@ import org.springframework.data.mongodb.core.mapping.MongoMappingContext; import org.springframework.data.mongodb.core.query.Criteria; import org.springframework.data.mongodb.core.query.Query; import org.springframework.data.mongodb.core.query.Update; -import org.springframework.integration.store.AbstractBatchingMessageGroupStore; +import org.springframework.integration.store.AbstractMessageGroupStore; import org.springframework.integration.store.BasicMessageGroupStore; +import org.springframework.integration.store.MessageGroup; import org.springframework.integration.support.DefaultMessageBuilderFactory; import org.springframework.integration.support.MessageBuilderFactory; import org.springframework.integration.support.utils.IntegrationUtils; @@ -66,7 +69,7 @@ import org.springframework.util.Assert; * @since 4.0 */ -public abstract class AbstractConfigurableMongoDbMessageStore extends AbstractBatchingMessageGroupStore +public abstract class AbstractConfigurableMongoDbMessageStore extends AbstractMessageGroupStore implements BasicMessageGroupStore, InitializingBean, ApplicationContextAware { public final static String SEQUENCE_NAME = "messagesSequence"; @@ -219,6 +222,47 @@ public abstract class AbstractConfigurableMongoDbMessageStore extends AbstractBa return Query.query(Criteria.where(MessageDocumentFields.GROUP_ID).is(groupId)); } + @Override + @Deprecated + public MessageGroup removeMessageFromGroup(Object key, Message messageToRemove) { + throw new UnsupportedOperationException("The operation isn't implemented for this class."); + } + + @Override + public void removeMessagesFromGroup(Object key, Collection> messages) { + throw new UnsupportedOperationException("The operation isn't implemented for this class."); + } + + @Override + public void setLastReleasedSequenceNumberForGroup(Object groupId, int sequenceNumber) { + throw new UnsupportedOperationException("The operation isn't implemented for this class."); + } + + @Override + public Iterator iterator() { + throw new UnsupportedOperationException("The operation isn't implemented for this class."); + } + + @Override + public void completeGroup(Object groupId) { + throw new UnsupportedOperationException("The operation isn't implemented for this class."); + } + + @Override + public Message getOneMessageFromGroup(Object groupId) { + throw new UnsupportedOperationException("The operation isn't implemented for this class."); + } + + @Override + public void addMessagesToGroup(Object groupId, Message... messages) { + throw new UnsupportedOperationException("The operation isn't implemented for this class."); + } + + @Override + public Collection> getMessagesForGroup(Object groupId) { + throw new UnsupportedOperationException("The operation isn't implemented for this class."); + } + /** * A {@link GenericConverter} implementation to convert {@link Message} to * serialized {@link byte[]} to store {@link Message} to the MongoDB. diff --git a/spring-integration-mongodb/src/main/java/org/springframework/integration/mongodb/store/ConfigurableMongoDbMessageStore.java b/spring-integration-mongodb/src/main/java/org/springframework/integration/mongodb/store/ConfigurableMongoDbMessageStore.java index df2e3f1def..9fbdca011b 100644 --- a/spring-integration-mongodb/src/main/java/org/springframework/integration/mongodb/store/ConfigurableMongoDbMessageStore.java +++ b/spring-integration-mongodb/src/main/java/org/springframework/integration/mongodb/store/ConfigurableMongoDbMessageStore.java @@ -35,6 +35,7 @@ import org.springframework.integration.store.MessageGroup; import org.springframework.integration.store.MessageGroupMetadata; import org.springframework.integration.store.MessageGroupStore; import org.springframework.integration.store.MessageStore; +import org.springframework.integration.store.SimpleMessageGroup; import org.springframework.jmx.export.annotation.ManagedAttribute; import org.springframework.jmx.export.annotation.ManagedOperation; import org.springframework.messaging.Message; @@ -53,7 +54,7 @@ import org.springframework.util.Assert; * @since 3.0 */ public class ConfigurableMongoDbMessageStore extends AbstractConfigurableMongoDbMessageStore - implements MessageStore, MessageGroupStore, Iterable { + implements MessageStore { public final static String DEFAULT_COLLECTION_NAME = "configurableStoreMessages"; @@ -141,40 +142,39 @@ public class ConfigurableMongoDbMessageStore extends AbstractConfigurableMongoDb Assert.notNull(groupId, "'groupId' must not be null"); Query query = groupOrderQuery(groupId); - List messageDocuments = this.mongoTemplate.find(query, MessageDocument.class, - this.collectionName); + MessageDocument messageDocument = this.mongoTemplate.findOne(query, MessageDocument.class, this.collectionName); - long createdTime = 0; - long lastModifiedTime = 0; - int lastReleasedSequence = 0; - boolean complete = false; + if (messageDocument != null) { + long createdTime = messageDocument.getCreatedTime(); + long lastModifiedTime = messageDocument.getLastModifiedTime(); + boolean complete = messageDocument.isComplete(); + int lastReleasedSequence = messageDocument.getLastReleasedSequence(); + + MessageGroup messageGroup = getMessageGroupFactory() + .create(this, groupId, createdTime, complete); + messageGroup.setLastModified(lastModifiedTime); + messageGroup.setLastReleasedMessageSequenceNumber(lastReleasedSequence); + return messageGroup; - if (messageDocuments.size() > 0) { - MessageDocument document = messageDocuments.get(0); - createdTime = document.getCreatedTime(); - lastModifiedTime = document.getLastModifiedTime(); - complete = document.isComplete(); - lastReleasedSequence = document.getLastReleasedSequence(); } - - List> messages = new ArrayList>(); - for (MessageDocument document : messageDocuments) { - messages.add(document.getMessage()); + else { + return new SimpleMessageGroup(groupId); } - MessageGroup group = getMessageGroupFactory().create(messages, groupId, createdTime, complete); - group.setLastReleasedMessageSequenceNumber(lastReleasedSequence); - group.setLastModified(lastModifiedTime); - - return group; } @Override - public MessageGroup addMessageToGroup(final Object groupId, final Message message) { + public MessageGroup addMessageToGroup(Object groupId, Message message) { + addMessagesToGroup(groupId, message); + return getMessageGroup(groupId); + } + + @Override + public void addMessagesToGroup(Object groupId, Message... messages) { Assert.notNull(groupId, "'groupId' must not be null"); - Assert.notNull(message, "'message' must not be null"); + Assert.notNull(messages, "'message' must not be null"); Query query = groupOrderQuery(groupId); - MessageDocument messageDocument = mongoTemplate.findOne(query, MessageDocument.class, collectionName); + MessageDocument messageDocument = this.mongoTemplate.findOne(query, MessageDocument.class, this.collectionName); long createdTime = 0; int lastReleasedSequence = 0; @@ -186,18 +186,17 @@ public class ConfigurableMongoDbMessageStore extends AbstractConfigurableMongoDb complete = messageDocument.isComplete(); } - MessageDocument document = new MessageDocument(message); - document.setGroupId(groupId); - document.setComplete(complete); - document.setLastReleasedSequence(lastReleasedSequence); - document.setCreatedTime(createdTime == 0 ? System.currentTimeMillis() : createdTime); - document.setLastModifiedTime(System.currentTimeMillis()); - document.setSequence(getNextId()); - - addMessageDocument(document); - - return getMessageGroup(groupId); + for (Message message : messages) { + MessageDocument document = new MessageDocument(message); + document.setGroupId(groupId); + document.setComplete(complete); + document.setLastReleasedSequence(lastReleasedSequence); + document.setCreatedTime(createdTime == 0 ? System.currentTimeMillis() : createdTime); + document.setLastModifiedTime(System.currentTimeMillis()); + document.setSequence(getNextId()); + addMessageDocument(document); + } } @Override @@ -335,7 +334,28 @@ public class ConfigurableMongoDbMessageStore extends AbstractConfigurableMongoDb @Override public Message getOneMessageFromGroup(Object groupId) { - throw new UnsupportedOperationException("Not yet implemented for this store"); + Assert.notNull(groupId, "'groupId' must not be null"); + Query query = groupOrderQuery(groupId); + MessageDocument messageDocument = this.mongoTemplate.findOne(query, MessageDocument.class, this.collectionName); + if (messageDocument != null) { + return messageDocument.getMessage(); + } + else { + return null; + } + } + + @Override + public Collection> getMessagesForGroup(Object groupId) { + Assert.notNull(groupId, "'groupId' must not be null"); + Query query = groupOrderQuery(groupId); + List documents = this.mongoTemplate.find(query, MessageDocument.class, this.collectionName); + List> messages = new ArrayList>(); + + for (MessageDocument document : documents) { + messages.add(document.getMessage()); + } + return messages; } private void expire(MessageGroup group) { diff --git a/spring-integration-mongodb/src/main/java/org/springframework/integration/mongodb/store/MongoDbMessageStore.java b/spring-integration-mongodb/src/main/java/org/springframework/integration/mongodb/store/MongoDbMessageStore.java index 0f93670838..eecbc38058 100644 --- a/spring-integration-mongodb/src/main/java/org/springframework/integration/mongodb/store/MongoDbMessageStore.java +++ b/spring-integration-mongodb/src/main/java/org/springframework/integration/mongodb/store/MongoDbMessageStore.java @@ -60,6 +60,7 @@ import org.springframework.integration.store.AbstractMessageGroupStore; import org.springframework.integration.store.MessageGroup; import org.springframework.integration.store.MessageGroupStore; import org.springframework.integration.store.MessageStore; +import org.springframework.integration.store.SimpleMessageGroup; import org.springframework.integration.support.MutableMessage; import org.springframework.integration.support.MutableMessageBuilder; import org.springframework.jmx.export.annotation.ManagedAttribute; @@ -242,37 +243,30 @@ public class MongoDbMessageStore extends AbstractMessageGroupStore public MessageGroup getMessageGroup(Object groupId) { Assert.notNull(groupId, "'groupId' must not be null"); Query query = whereGroupIdOrder(groupId); - List messageWrappers = this.template.find(query, MessageWrapper.class, this.collectionName); - List> messages = new ArrayList>(); - long timestamp = 0; - long lastModified = 0; - int lastReleasedSequenceNumber = 0; - boolean completeGroup = false; - if (messageWrappers.size() > 0) { - MessageWrapper messageWrapper = messageWrappers.get(0); - timestamp = messageWrapper.get_Group_timestamp(); - lastModified = messageWrapper.get_Group_update_timestamp(); - completeGroup = messageWrapper.get_Group_complete(); - lastReleasedSequenceNumber = messageWrapper.get_LastReleasedSequenceNumber(); - } + MessageWrapper messageWrapper = this.template.findOne(query, MessageWrapper.class, this.collectionName); - for (MessageWrapper messageWrapper : messageWrappers) { - messages.add(messageWrapper.getMessage()); - } + if (messageWrapper != null) { + long createdTime = messageWrapper.get_Group_timestamp(); + long lastModifiedTime = messageWrapper.get_Group_update_timestamp(); + boolean complete = messageWrapper.get_Group_complete(); + int lastReleasedSequence = messageWrapper.get_LastReleasedSequenceNumber(); - MessageGroup messageGroup = getMessageGroupFactory().create(messages, groupId, timestamp, completeGroup); - messageGroup.setLastModified(lastModified); - if (lastReleasedSequenceNumber > 0) { - messageGroup.setLastReleasedMessageSequenceNumber(lastReleasedSequenceNumber); - } + MessageGroup messageGroup = getMessageGroupFactory() + .create(this, groupId, createdTime, complete); + messageGroup.setLastModified(lastModifiedTime); + messageGroup.setLastReleasedMessageSequenceNumber(lastReleasedSequence); + return messageGroup; - return messageGroup; + } + else { + return new SimpleMessageGroup(groupId); + } } @Override - public MessageGroup addMessageToGroup(final Object groupId, final Message message) { + public void addMessagesToGroup(Object groupId, Message... messages) { Assert.notNull(groupId, "'groupId' must not be null"); - Assert.notNull(message, "'message' must not be null"); + Assert.notNull(messages, "'message' must not be null"); Query query = whereGroupIdOrder(groupId); MessageWrapper messageDocument = this.template.findOne(query, MessageWrapper.class, this.collectionName); @@ -286,18 +280,17 @@ public class MongoDbMessageStore extends AbstractMessageGroupStore complete = messageDocument.get_Group_complete(); } + for (Message message : messages) { + MessageWrapper wrapper = new MessageWrapper(message); + wrapper.set_GroupId(groupId); + wrapper.set_Group_timestamp(createdTime == 0 ? System.currentTimeMillis() : createdTime); + wrapper.set_Group_update_timestamp(System.currentTimeMillis()); + wrapper.set_Group_complete(complete); + wrapper.set_LastReleasedSequenceNumber(lastReleasedSequence); + wrapper.setSequence(getNextId()); - MessageWrapper wrapper = new MessageWrapper(message); - wrapper.set_GroupId(groupId); - wrapper.set_Group_timestamp(createdTime == 0 ? System.currentTimeMillis() : createdTime); - wrapper.set_Group_update_timestamp(System.currentTimeMillis()); - wrapper.set_Group_complete(complete); - wrapper.set_LastReleasedSequenceNumber(lastReleasedSequence); - wrapper.setSequence(getNextId()); - - addMessageDocument(wrapper); - return getMessageGroup(groupId); - + addMessageDocument(wrapper); + } } @Override @@ -393,6 +386,32 @@ public class MongoDbMessageStore extends AbstractMessageGroupStore this.updateGroup(groupId, lastModifiedUpdate().set(GROUP_COMPLETE_KEY, true)); } + @Override + public Message getOneMessageFromGroup(Object groupId) { + Assert.notNull(groupId, "'groupId' must not be null"); + Query query = whereGroupIdOrder(groupId); + MessageWrapper messageWrapper = this.template.findOne(query, MessageWrapper.class, this.collectionName); + if (messageWrapper != null) { + return messageWrapper.getMessage(); + } + else { + return null; + } + } + + @Override + public Collection> getMessagesForGroup(Object groupId) { + Assert.notNull(groupId, "'groupId' must not be null"); + Query query = whereGroupIdOrder(groupId); + List messageWrappers = this.template.find(query, MessageWrapper.class, this.collectionName); + List> messages = new ArrayList>(); + + for (MessageWrapper messageWrapper : messageWrappers) { + messages.add(messageWrapper.getMessage()); + } + return messages; + } + @Override @ManagedAttribute public int getMessageCountForAllMessageGroups() { diff --git a/spring-integration-mongodb/src/test/java/org/springframework/integration/mongodb/store/AbstractMongoDbMessageGroupStoreTests.java b/spring-integration-mongodb/src/test/java/org/springframework/integration/mongodb/store/AbstractMongoDbMessageGroupStoreTests.java index 20f7126b0b..f9b093dadb 100644 --- a/spring-integration-mongodb/src/test/java/org/springframework/integration/mongodb/store/AbstractMongoDbMessageGroupStoreTests.java +++ b/spring-integration-mongodb/src/test/java/org/springframework/integration/mongodb/store/AbstractMongoDbMessageGroupStoreTests.java @@ -16,10 +16,12 @@ package org.springframework.integration.mongodb.store; +import static org.hamcrest.Matchers.containsString; import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertFalse; import static org.junit.Assert.assertNotNull; import static org.junit.Assert.assertNull; +import static org.junit.Assert.assertThat; import static org.junit.Assert.assertTrue; import java.util.ArrayList; @@ -42,7 +44,6 @@ import org.springframework.integration.store.AbstractBatchingMessageGroupStore; import org.springframework.integration.store.MessageGroup; import org.springframework.integration.store.MessageGroupStore; import org.springframework.integration.store.MessageStore; -import org.springframework.integration.store.SimpleMessageGroup; import org.springframework.integration.support.MessageBuilder; import org.springframework.messaging.Message; import org.springframework.messaging.MessageChannel; @@ -63,10 +64,11 @@ public abstract class AbstractMongoDbMessageGroupStoreTests extends MongoDbAvail public void testNonExistingEmptyMessageGroup() throws Exception { this.cleanupCollections(new SimpleMongoDbFactory(new MongoClient(), "test")); MessageGroupStore store = getMessageGroupStore(); + store.addMessagesToGroup(1, new GenericMessage("foo")); MessageGroup messageGroup = store.getMessageGroup(1); assertNotNull(messageGroup); - assertTrue(messageGroup instanceof SimpleMessageGroup); - assertEquals(0, messageGroup.size()); + assertThat(messageGroup.getClass().getName(), containsString("PersistentMessageGroup")); + assertEquals(1, messageGroup.size()); } @Test @@ -78,7 +80,7 @@ public abstract class AbstractMongoDbMessageGroupStoreTests extends MongoDbAvail MessageGroup messageGroup = store.getMessageGroup(1); Message messageA = new GenericMessage("A"); Message messageB = new GenericMessage("B"); - store.addMessageToGroup(1, messageA); + store.addMessagesToGroup(1, messageA); messageGroup = store.addMessageToGroup(1, messageB); assertNotNull(messageGroup); assertEquals(2, messageGroup.size()); @@ -101,7 +103,7 @@ public abstract class AbstractMongoDbMessageGroupStoreTests extends MongoDbAvail Message messageA = MessageBuilder.withPayload("A").setHeader("foo", uuidA).build(); UUID uuidB = UUID.randomUUID(); Message messageB = MessageBuilder.withPayload("B").setHeader("foo", uuidB).build(); - store.addMessageToGroup(id, messageA); + store.addMessagesToGroup(id, messageA); messageGroup = store.addMessageToGroup(id, messageB); assertNotNull(messageGroup); assertEquals(2, messageGroup.size()); @@ -122,8 +124,7 @@ public abstract class AbstractMongoDbMessageGroupStoreTests extends MongoDbAvail MessageGroupStore store = this.getMessageGroupStore(); Message messageA = new GenericMessage("A"); Message messageB = new GenericMessage("B"); - store.addMessageToGroup(1, messageA); - store.addMessageToGroup(1, messageB); + store.addMessagesToGroup(1, messageA, messageB); assertEquals(2, store.messageGroupSize(1)); } @@ -134,9 +135,9 @@ public abstract class AbstractMongoDbMessageGroupStoreTests extends MongoDbAvail MessageGroupStore store = this.getMessageGroupStore(); Message messageA = new GenericMessage("A"); Message messageB = new GenericMessage("B"); - store.addMessageToGroup(1, messageA); + store.addMessagesToGroup(1, messageA); Thread.sleep(10); - store.addMessageToGroup(1, messageB); + store.addMessagesToGroup(1, messageB); assertEquals(2, store.messageGroupSize(1)); Message out = store.pollMessageFromGroup(1); assertNotNull(out); @@ -153,10 +154,10 @@ public abstract class AbstractMongoDbMessageGroupStoreTests extends MongoDbAvail this.cleanupCollections(new SimpleMongoDbFactory(new MongoClient(), "test")); MessageGroupStore store = this.getMessageGroupStore(); Message messageA = new GenericMessage("A"); - store.addMessageToGroup(1, messageA); - store.addMessageToGroup(2, messageA); - store.addMessageToGroup(3, messageA); - store.addMessageToGroup(4, messageA); + store.addMessagesToGroup(1, messageA); + store.addMessagesToGroup(2, messageA); + store.addMessagesToGroup(3, messageA); + store.addMessagesToGroup(4, messageA); assertEquals(1, store.messageGroupSize(1)); assertEquals(1, store.messageGroupSize(2)); assertEquals(1, store.messageGroupSize(3)); @@ -190,10 +191,10 @@ public abstract class AbstractMongoDbMessageGroupStoreTests extends MongoDbAvail MessageGroupStore store = this.getMessageGroupStore(); Message messageA = new GenericMessage("A"); - store.addMessageToGroup(1, messageA); - store.addMessageToGroup(2, messageA); - store.addMessageToGroup(3, messageA); - store.addMessageToGroup(4, messageA); + store.addMessagesToGroup(1, messageA); + store.addMessagesToGroup(2, messageA); + store.addMessagesToGroup(3, messageA); + store.addMessagesToGroup(4, messageA); assertEquals(1, store.messageGroupSize(1)); assertEquals(1, store.messageGroupSize(2)); assertEquals(1, store.messageGroupSize(3)); @@ -258,7 +259,7 @@ public abstract class AbstractMongoDbMessageGroupStoreTests extends MongoDbAvail MessageGroup messageGroup = store.getMessageGroup(1); Message messageA = new GenericMessage("A"); Message messageB = new GenericMessage("B"); - store.addMessageToGroup(1, messageA); + store.addMessagesToGroup(1, messageA); messageGroup = store.addMessageToGroup(1, messageB); assertNotNull(messageGroup); assertEquals(2, messageGroup.size()); @@ -306,7 +307,7 @@ public abstract class AbstractMongoDbMessageGroupStoreTests extends MongoDbAvail MessageGroup messageGroup = store.getMessageGroup(1); assertNotNull(messageGroup); Message message = new GenericMessage("Hello"); - store.addMessageToGroup(messageGroup.getGroupId(), message); + store.addMessagesToGroup(messageGroup.getGroupId(), message); store.completeGroup(messageGroup.getGroupId()); messageGroup = store.getMessageGroup(1); assertTrue(messageGroup.isComplete()); @@ -321,7 +322,7 @@ public abstract class AbstractMongoDbMessageGroupStoreTests extends MongoDbAvail MessageGroup messageGroup = store.getMessageGroup(1); assertNotNull(messageGroup); Message message = new GenericMessage("Hello"); - store.addMessageToGroup(messageGroup.getGroupId(), message); + store.addMessagesToGroup(messageGroup.getGroupId(), message); store.setLastReleasedSequenceNumberForGroup(messageGroup.getGroupId(), 5); messageGroup = store.getMessageGroup(1); assertEquals(5, messageGroup.getLastReleasedMessageSequenceNumber()); @@ -335,8 +336,7 @@ public abstract class AbstractMongoDbMessageGroupStoreTests extends MongoDbAvail MessageGroup messageGroup = store.getMessageGroup(1); Message message = new GenericMessage("2"); - store.addMessageToGroup(1, new GenericMessage("1")); - store.addMessageToGroup(1, message); + store.addMessagesToGroup(1, new GenericMessage("1"), message); messageGroup = store.addMessageToGroup(1, new GenericMessage("3")); assertNotNull(messageGroup); assertEquals(3, messageGroup.size()); @@ -355,9 +355,7 @@ public abstract class AbstractMongoDbMessageGroupStoreTests extends MongoDbAvail MessageGroupStore store2 = this.getMessageGroupStore(); Message message = new GenericMessage("1"); - store1.addMessageToGroup(1, message); - store2.addMessageToGroup(1, new GenericMessage("2")); - store1.addMessageToGroup(1, new GenericMessage("3")); + store1.addMessagesToGroup(1, message, new GenericMessage("2"), new GenericMessage("3")); MessageGroupStore store3 = this.getMessageGroupStore(); @@ -380,9 +378,9 @@ public abstract class AbstractMongoDbMessageGroupStoreTests extends MongoDbAvail MessageGroupStore store2 = this.getMessageGroupStore(); Message message = new GenericMessage("1"); - store2.addMessageToGroup(1, message); - store1.addMessageToGroup(2, new GenericMessage("2")); - store2.addMessageToGroup(3, new GenericMessage("3")); + store2.addMessagesToGroup(1, message); + store1.addMessagesToGroup(2, new GenericMessage("2")); + store2.addMessagesToGroup(3, new GenericMessage("3")); MessageGroupStore store3 = this.getMessageGroupStore(); Iterator iterator = store3.iterator(); @@ -415,7 +413,7 @@ public abstract class AbstractMongoDbMessageGroupStoreTests extends MongoDbAvail List> messages = new ArrayList>(); for (int i = 0; i < 25; i++) { Message message = MessageBuilder.withPayload("foo").setCorrelationId(groupId).build(); - messageStore.addMessageToGroup(groupId, message); + messageStore.addMessagesToGroup(groupId, message); messages.add(message); } MessageGroup group = messageStore.getMessageGroup(groupId); @@ -477,20 +475,41 @@ public abstract class AbstractMongoDbMessageGroupStoreTests extends MongoDbAvail MessageChannel input = context.getBean("inputChannel", MessageChannel.class); QueueChannel output = context.getBean("outputChannel", QueueChannel.class); - Message m1 = MessageBuilder.withPayload("1").setSequenceNumber(1).setSequenceSize(3).setCorrelationId(1).build(); - Message m2 = MessageBuilder.withPayload("2").setSequenceNumber(2).setSequenceSize(3).setCorrelationId(1).build(); + Message m1 = MessageBuilder.withPayload("1") + .setSequenceNumber(1) + .setSequenceSize(10) + .setCorrelationId(1) + .build(); + Message m2 = MessageBuilder.withPayload("2") + .setSequenceNumber(2) + .setSequenceSize(10) + .setCorrelationId(1) + .build(); input.send(m1); assertNull(output.receive(1000)); input.send(m2); assertNull(output.receive(1000)); + + for (int i = 3; i < 10; i++) { + input.send(MessageBuilder.withPayload("" + i) + .setSequenceNumber(i) + .setSequenceSize(10) + .setCorrelationId(1) + .build()); + } + context.close(); context = new ClassPathXmlApplicationContext(config, this.getClass()); input = context.getBean("inputChannel", MessageChannel.class); output = context.getBean("outputChannel", QueueChannel.class); - Message m3 = MessageBuilder.withPayload("3").setSequenceNumber(3).setSequenceSize(3).setCorrelationId(1).build(); - input.send(m3); + Message m10 = MessageBuilder.withPayload("10") + .setSequenceNumber(10) + .setSequenceSize(10) + .setCorrelationId(1) + .build(); + input.send(m10); assertNotNull(output.receive(2000)); context.close(); } @@ -511,7 +530,7 @@ public abstract class AbstractMongoDbMessageGroupStoreTests extends MongoDbAvail message = MessageHistory.write(message, fooChannel); message = MessageHistory.write(message, barChannel); - store.addMessageToGroup(1, message); + store.addMessagesToGroup(1, message); MessageGroup group = store.getMessageGroup(1); assertNotNull(group); Collection> messages = group.getMessages(); diff --git a/spring-integration-mongodb/src/test/java/org/springframework/integration/mongodb/store/ConfigurableMongoDbMessageGroupStoreTests.java b/spring-integration-mongodb/src/test/java/org/springframework/integration/mongodb/store/ConfigurableMongoDbMessageGroupStoreTests.java index d07705077e..67e1d6d585 100644 --- a/spring-integration-mongodb/src/test/java/org/springframework/integration/mongodb/store/ConfigurableMongoDbMessageGroupStoreTests.java +++ b/spring-integration-mongodb/src/test/java/org/springframework/integration/mongodb/store/ConfigurableMongoDbMessageGroupStoreTests.java @@ -23,6 +23,7 @@ import static org.junit.Assert.assertThat; import java.util.Map; import org.hamcrest.Matchers; +import org.junit.Ignore; import org.junit.Test; import org.springframework.context.support.ClassPathXmlApplicationContext; @@ -34,10 +35,13 @@ import org.springframework.integration.IntegrationMessageHeaderAccessor; import org.springframework.integration.channel.PriorityChannel; import org.springframework.integration.channel.QueueChannel; import org.springframework.integration.mongodb.rules.MongoDbAvailable; +import org.springframework.integration.store.AbstractMessageGroupStore; import org.springframework.integration.store.MessageStore; import org.springframework.integration.support.MessageBuilder; import org.springframework.integration.test.util.TestUtils; import org.springframework.messaging.Message; +import org.springframework.messaging.MessageChannel; +import org.springframework.util.StopWatch; import com.mongodb.DBObject; import com.mongodb.MongoClient; @@ -68,7 +72,49 @@ public class ConfigurableMongoDbMessageGroupStoreTests extends AbstractMongoDbMe @Test @MongoDbAvailable public void testWithAggregatorWithShutdown() throws Exception { - super.testWithAggregatorWithShutdown("mongo-aggregator-confugurable-config.xml"); + super.testWithAggregatorWithShutdown("mongo-aggregator-configurable-config.xml"); + } + + @Test + @Ignore("The performance test. Enough slow. Also needs the release strategy changed to size() == 1000") + @MongoDbAvailable + public void messageGroupStoreLazyLoadPerformance() throws Exception { + cleanupCollections(new SimpleMongoDbFactory(new MongoClient(), "test")); + + StopWatch watch = new StopWatch("Lazy-Load Performance"); + + int sequenceSize = 1000; + + performLazyLoadEagerTest(watch, sequenceSize, true); + + performLazyLoadEagerTest(watch, sequenceSize, false); + + System. out .println(watch.prettyPrint()); // checkstyle + } + + private void performLazyLoadEagerTest(StopWatch watch, int sequenceSize, boolean lazyLoad) { + ClassPathXmlApplicationContext context = + new ClassPathXmlApplicationContext("mongo-aggregator-configurable-config.xml", getClass()); + context.refresh(); + + AbstractMessageGroupStore messageGroupStore = context.getBean("mongoStore", AbstractMessageGroupStore.class); + messageGroupStore.setLazyLoadMessageGroups(lazyLoad); + MessageChannel input = context.getBean("inputChannel", MessageChannel.class); + QueueChannel output = context.getBean("outputChannel", QueueChannel.class); + + watch.start(lazyLoad ? "Lazy-Load" : "Eager"); + + for (int i = 0; i < sequenceSize; i++) { + input.send(MessageBuilder.withPayload("" + i) + .setCorrelationId(1) + .build()); + } + + assertNotNull(output.receive(20000)); + + watch.stop(); + + context.close(); } @Test diff --git a/spring-integration-mongodb/src/test/java/org/springframework/integration/mongodb/store/MongoDbMessageGroupStoreTests.java b/spring-integration-mongodb/src/test/java/org/springframework/integration/mongodb/store/MongoDbMessageGroupStoreTests.java index be808bc39f..b400367c95 100644 --- a/spring-integration-mongodb/src/test/java/org/springframework/integration/mongodb/store/MongoDbMessageGroupStoreTests.java +++ b/spring-integration-mongodb/src/test/java/org/springframework/integration/mongodb/store/MongoDbMessageGroupStoreTests.java @@ -16,13 +16,14 @@ package org.springframework.integration.mongodb.store; -import com.mongodb.MongoClient; import org.junit.Test; import org.springframework.data.mongodb.core.SimpleMongoDbFactory; import org.springframework.integration.mongodb.rules.MongoDbAvailable; import org.springframework.integration.store.MessageStore; +import com.mongodb.MongoClient; + /** * @author Oleg Zhurakousky * @author Gary Russell @@ -33,7 +34,8 @@ public class MongoDbMessageGroupStoreTests extends AbstractMongoDbMessageGroupSt @Override protected MongoDbMessageStore getMessageGroupStore() throws Exception { - MongoDbMessageStore mongoDbMessageStore = new MongoDbMessageStore(new SimpleMongoDbFactory(new MongoClient(), "test")); + MongoDbMessageStore mongoDbMessageStore = + new MongoDbMessageStore(new SimpleMongoDbFactory(new MongoClient(), "test")); mongoDbMessageStore.afterPropertiesSet(); return mongoDbMessageStore; } diff --git a/spring-integration-mongodb/src/test/java/org/springframework/integration/mongodb/store/mongo-aggregator-config.xml b/spring-integration-mongodb/src/test/java/org/springframework/integration/mongodb/store/mongo-aggregator-config.xml index dece0dbcbf..40044438bd 100644 --- a/spring-integration-mongodb/src/test/java/org/springframework/integration/mongodb/store/mongo-aggregator-config.xml +++ b/spring-integration-mongodb/src/test/java/org/springframework/integration/mongodb/store/mongo-aggregator-config.xml @@ -5,7 +5,8 @@ xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd http://www.springframework.org/schema/integration http://www.springframework.org/schema/integration/spring-integration.xsd"> - + diff --git a/spring-integration-mongodb/src/test/java/org/springframework/integration/mongodb/store/mongo-aggregator-confugurable-config.xml b/spring-integration-mongodb/src/test/java/org/springframework/integration/mongodb/store/mongo-aggregator-configurable-config.xml similarity index 91% rename from spring-integration-mongodb/src/test/java/org/springframework/integration/mongodb/store/mongo-aggregator-confugurable-config.xml rename to spring-integration-mongodb/src/test/java/org/springframework/integration/mongodb/store/mongo-aggregator-configurable-config.xml index c127dc0c36..1f4e4f1bac 100644 --- a/spring-integration-mongodb/src/test/java/org/springframework/integration/mongodb/store/mongo-aggregator-confugurable-config.xml +++ b/spring-integration-mongodb/src/test/java/org/springframework/integration/mongodb/store/mongo-aggregator-configurable-config.xml @@ -5,7 +5,8 @@ xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd http://www.springframework.org/schema/integration http://www.springframework.org/schema/integration/spring-integration.xsd"> - + diff --git a/spring-integration-mongodb/src/test/resources/log4j.properties b/spring-integration-mongodb/src/test/resources/log4j.properties new file mode 100644 index 0000000000..91e7970a40 --- /dev/null +++ b/spring-integration-mongodb/src/test/resources/log4j.properties @@ -0,0 +1,8 @@ +log4j.rootCategory=WARN, stdout + +log4j.appender.stdout=org.apache.log4j.ConsoleAppender +log4j.appender.stdout.layout=org.apache.log4j.PatternLayout +log4j.appender.stdout.layout.ConversionPattern=%d{HH:mm:ss.SSS} %-5p [%t][%c] %m%n + +log4j.category.org.springframework.integration=WARN +log4j.category.org.springframework.integration.mongodb=INFO diff --git a/spring-integration-redis/src/test/java/org/springframework/integration/redis/store/RedisMessageGroupStoreTests.java b/spring-integration-redis/src/test/java/org/springframework/integration/redis/store/RedisMessageGroupStoreTests.java index 6aae11810d..29bcf6fba8 100644 --- a/spring-integration-redis/src/test/java/org/springframework/integration/redis/store/RedisMessageGroupStoreTests.java +++ b/spring-integration-redis/src/test/java/org/springframework/integration/redis/store/RedisMessageGroupStoreTests.java @@ -186,8 +186,7 @@ public class RedisMessageGroupStoreTests extends RedisAvailableTests { MessageGroup messageGroup = store.getMessageGroup(1); Message message = new GenericMessage("2"); - store.addMessageToGroup(messageGroup.getGroupId(), new GenericMessage("1")); - store.addMessageToGroup(messageGroup.getGroupId(), message); + store.addMessagesToGroup(messageGroup.getGroupId(), new GenericMessage("1"), message); messageGroup = store.addMessageToGroup(messageGroup.getGroupId(), new GenericMessage("3")); assertEquals(3, messageGroup.size()); @@ -218,7 +217,7 @@ public class RedisMessageGroupStoreTests extends RedisAvailableTests { message = MessageHistory.write(message, fooChannel); message = MessageHistory.write(message, barChannel); - store.addMessageToGroup(1, message); + store.addMessagesToGroup(1, message); message = store.getMessageGroup(1).getMessages().iterator().next(); @@ -236,7 +235,7 @@ public class RedisMessageGroupStoreTests extends RedisAvailableTests { RedisMessageStore store = new RedisMessageStore(jcf); MessageGroup messageGroup = store.getMessageGroup(1); - store.addMessageToGroup(messageGroup.getGroupId(), new GenericMessage("1")); + store.addMessagesToGroup(messageGroup.getGroupId(), new GenericMessage("1")); store.removeMessagesFromGroup(1, new GenericMessage("2")); } @@ -259,7 +258,7 @@ public class RedisMessageGroupStoreTests extends RedisAvailableTests { RedisMessageStore store2 = new RedisMessageStore(jcf); Message message = new GenericMessage("1"); - store1.addMessageToGroup(1, message); + store1.addMessagesToGroup(1, message); MessageGroup messageGroup = store2.addMessageToGroup(1, new GenericMessage("2")); assertEquals(2, messageGroup.getMessages().size()); @@ -280,10 +279,9 @@ public class RedisMessageGroupStoreTests extends RedisAvailableTests { RedisMessageStore store2 = new RedisMessageStore(jcf); - store1.addMessageToGroup(1, new GenericMessage("1")); - store2.addMessageToGroup(2, new GenericMessage("2")); - store1.addMessageToGroup(3, new GenericMessage("3")); - store2.addMessageToGroup(3, new GenericMessage("3A")); + store1.addMessagesToGroup(1, new GenericMessage("1")); + store2.addMessagesToGroup(2, new GenericMessage("2")); + store1.addMessagesToGroup(3, new GenericMessage("3"), new GenericMessage("3A")); Iterator messageGroups = store1.iterator(); int counter = 0; @@ -394,7 +392,7 @@ public class RedisMessageGroupStoreTests extends RedisAvailableTests { List> messages = new ArrayList>(); for (int i = 0; i < 25; i++) { Message message = MessageBuilder.withPayload("foo").setCorrelationId(groupId).build(); - messageStore.addMessageToGroup(groupId, message); + messageStore.addMessagesToGroup(groupId, message); messages.add(message); } MessageGroup group = messageStore.getMessageGroup(groupId); diff --git a/spring-integration-redis/src/test/java/org/springframework/integration/redis/store/RedisMessageStoreTests.java b/spring-integration-redis/src/test/java/org/springframework/integration/redis/store/RedisMessageStoreTests.java index 465827429a..5d501cace2 100644 --- a/spring-integration-redis/src/test/java/org/springframework/integration/redis/store/RedisMessageStoreTests.java +++ b/spring-integration-redis/src/test/java/org/springframework/integration/redis/store/RedisMessageStoreTests.java @@ -169,7 +169,7 @@ public class RedisMessageStoreTests extends RedisAvailableTests { List> messages = new ArrayList>(); for (int i = 0; i < 25; i++) { Message message = MessageBuilder.withPayload("foo").setCorrelationId(groupId).build(); - messageStore.addMessageToGroup(groupId, message); + messageStore.addMessagesToGroup(groupId, message); messages.add(message); } messageStore.removeMessagesFromGroup(groupId, messages); diff --git a/src/reference/asciidoc/message-store.adoc b/src/reference/asciidoc/message-store.adoc index 8c64f102fe..e02acc71d8 100644 --- a/src/reference/asciidoc/message-store.adoc +++ b/src/reference/asciidoc/message-store.adoc @@ -94,3 +94,40 @@ This defaults to a `SimpleMessageGroupFactory` which produces `SimpleMessageGrou (`LinkedHashSet`) internal collection. Other possible options are `SYNCHRONISED_SET` and `BLOCKING_QUEUE`, where the last one can be used to reinstate the previous `SimpleMessageGroup` behavior. +Also the `PERSISTENT` option is available. See the next section for more information. + +[[lazy-load-message-group]] +==== Persistence MessageGroupStore and Lazy-Load + +Starting with _version 4.3_, all persistence `MessageGroupStore` s retrieve `MessageGroup` s and their `messages` +from the store with the _Lazy-Load_ manner. +In most cases it is useful for the Correlation `MessageHandler` s (<> and <>), +when it would be an overhead to load entire `MessageGroup` from the store on each correlation operation. + +To switch off the lazy-load behavior the `AbstractMessageGroupStore.setLazyLoadMessageGroups(false)` option +can be used from the configuration. + +Our performance tests for _lazy-load_ on MongoDB `MessageStore` (<>) and +`` (<>) +with custom `release-strategy` like: + +[source,xml] +---- + + +---- + +demonstrate this results for 1000 simple messages: + +.... +StopWatch 'Lazy-Load Performance': running time (millis) = 38918 +----------------------------------------- +ms % Task name +----------------------------------------- +02652 007% Lazy-Load +36266 093% Eager +.... + diff --git a/src/reference/asciidoc/whats-new.adoc b/src/reference/asciidoc/whats-new.adoc index a0c786b251..0a34ad6848 100644 --- a/src/reference/asciidoc/whats-new.adoc +++ b/src/reference/asciidoc/whats-new.adoc @@ -21,6 +21,11 @@ The `SimpleMessageGroupFactory` is provided for the `SimpleMessageGroup` with th factory for the standard `MessageGroupStore` implementations. See <> for more information. +==== PersistentMessageGroup + +The `PersistentMessageGroup`, - lazy-load proxy, - implementation is provided for persistent `MessageGroupStore` s, +which return this instance for the `getMessageGroup()` when their `lazyLoadMessageGroups` is `true` (defaults). +See <> for more information. [[x4.3-general]] === General Changes