From f3d525a5e81e6794789967f77e887cf91315ce32 Mon Sep 17 00:00:00 2001 From: Gary Russell Date: Tue, 16 Jun 2015 09:01:29 -0400 Subject: [PATCH] INT-3642: Improve MessageGroupStore Removal JIRA: https://jira.spring.io/browse/INT-3642 Currently, `removeMessageFromGroup` rebuilds the group on every removal. In every case where this method is used in the framework, the result is not used. Add `removeMessagesFromGroup` that removes a collection of messages and returns no result. INT-3642: Polishing - PR Comments INT-3642: Polishing and Fix Group Metadata Size --- .../aggregator/AggregatingMessageHandler.java | 7 +- .../aggregator/CorrelatingMessageBarrier.java | 5 +- .../ResequencingMessageHandler.java | 7 +- .../integration/handler/DelayHandler.java | 2 +- .../AbstractBatchingMessageGroupStore.java | 45 ++++++++ .../store/AbstractKeyValueMessageStore.java | 100 +++++++++++------- .../store/AbstractMessageGroupStore.java | 10 +- .../store/MessageGroupMetadata.java | 10 +- .../integration/store/MessageGroupStore.java | 25 ++++- .../integration/store/SimpleMessageStore.java | 28 ++++- .../integration/store/MessageStoreTests.java | 21 +++- .../store/SimpleMessageStoreTests.java | 26 ++++- .../store/GemfireMessageStoreTests.java | 22 ++++ .../integration/jdbc/JdbcMessageStore.java | 38 ++++++- .../jdbc/JdbcMessageStoreTests.java | 21 ++++ ...stractConfigurableMongoDbMessageStore.java | 5 +- .../ConfigurableMongoDbMessageStore.java | 33 +++++- .../mongodb/store/MongoDbMessageStore.java | 32 +++++- ...AbstractMongoDbMessageGroupStoreTests.java | 35 +++++- ...igurableMongoDbMessageGroupStoreTests.java | 15 ++- .../store/RedisMessageGroupStoreTests.java | 59 +++++++---- .../redis/store/RedisMessageStoreTests.java | 24 ++++- src/reference/asciidoc/whats-new.adoc | 9 ++ 23 files changed, 472 insertions(+), 107 deletions(-) create mode 100644 spring-integration-core/src/main/java/org/springframework/integration/store/AbstractBatchingMessageGroupStore.java diff --git a/spring-integration-core/src/main/java/org/springframework/integration/aggregator/AggregatingMessageHandler.java b/spring-integration-core/src/main/java/org/springframework/integration/aggregator/AggregatingMessageHandler.java index 130320ec54..cd37dcc119 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/aggregator/AggregatingMessageHandler.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/aggregator/AggregatingMessageHandler.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2014 the original author or authors. + * Copyright 2002-2015 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 @@ -26,6 +26,7 @@ import org.springframework.messaging.Message; * * @author Oleg Zhurakousky * @author Artem Bilan + * @author Gary Russell * @since 2.1 */ public class AggregatingMessageHandler extends AbstractCorrelatingMessageHandler { @@ -64,9 +65,7 @@ public class AggregatingMessageHandler extends AbstractCorrelatingMessageHandler remove(messageGroup); } else { - for (Message message : messageGroup.getMessages()) { - this.messageStore.removeMessageFromGroup(messageGroup.getGroupId(), message); - } + this.messageStore.removeMessagesFromGroup(messageGroup.getGroupId(), messageGroup.getMessages()); } } 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 88b300db31..b305c526ea 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 @@ -1,5 +1,5 @@ /* - * Copyright 2002-2014 the original author or authors. + * Copyright 2002-2015 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 @@ -43,6 +43,7 @@ import org.springframework.messaging.Message; * * @author Iwein Fuld * @author Oleg Zhurakousky + * @author Gary Russell * * @see AbstractCorrelatingMessageHandler */ @@ -118,7 +119,7 @@ public class CorrelatingMessageBarrier extends AbstractMessageHandler implements Iterator> messages = group.getMessages().iterator(); if (messages.hasNext()) { nextMessage = messages.next(); - store.removeMessageFromGroup(key, nextMessage); + this.store.removeMessagesFromGroup(key, nextMessage); if (log.isDebugEnabled()) { log.debug(String.format("Released message for key [%s]: %s.", key, nextMessage)); } 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 ddb6ceda8a..b56ecf2696 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 @@ -1,5 +1,5 @@ /* - * Copyright 2002-2011 the original author or authors. + * Copyright 2002-2015 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 @@ -25,6 +25,7 @@ import org.springframework.messaging.Message; * Will remove {@link MessageGroup}s only if 'sequenceSize' is provided and reached. * * @author Oleg Zhurakousky + * @author Gary Russell * @since 2.1 */ public class ResequencingMessageHandler extends AbstractCorrelatingMessageHandler { @@ -83,9 +84,7 @@ public class ResequencingMessageHandler extends AbstractCorrelatingMessageHandle if (completedMessages != null){ int lastReleasedSequenceNumber = this.findLastReleasedSequenceNumber(messageGroup.getGroupId(), completedMessages); messageStore.setLastReleasedSequenceNumberForGroup(messageGroup.getGroupId(), lastReleasedSequenceNumber); - for (Message msg : completedMessages) { - this.messageStore.removeMessageFromGroup(messageGroup.getGroupId(), msg); - } + this.messageStore.removeMessagesFromGroup(messageGroup.getGroupId(), completedMessages); } if (timeout) { this.messageStore.completeGroup(messageGroup.getGroupId()); diff --git a/spring-integration-core/src/main/java/org/springframework/integration/handler/DelayHandler.java b/spring-integration-core/src/main/java/org/springframework/integration/handler/DelayHandler.java index 923f8b7af7..2cc2cd0848 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/handler/DelayHandler.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/handler/DelayHandler.java @@ -352,7 +352,7 @@ public class DelayHandler extends AbstractReplyProducingMessageHandler implement private void doReleaseMessage(Message message) { if (removeDelayedMessageFromMessageStore(message)) { - this.messageStore.removeMessageFromGroup(this.messageGroupId, message); + this.messageStore.removeMessagesFromGroup(this.messageGroupId, message); this.handleMessageInternal(message); } else { diff --git a/spring-integration-core/src/main/java/org/springframework/integration/store/AbstractBatchingMessageGroupStore.java b/spring-integration-core/src/main/java/org/springframework/integration/store/AbstractBatchingMessageGroupStore.java new file mode 100644 index 0000000000..7a2a75d028 --- /dev/null +++ b/spring-integration-core/src/main/java/org/springframework/integration/store/AbstractBatchingMessageGroupStore.java @@ -0,0 +1,45 @@ +/* + * Copyright 2015 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; + + +/** + * @author Gary Russell + * @since 4.2 + * + */ +public abstract class AbstractBatchingMessageGroupStore implements BasicMessageGroupStore { + + private static final int DEFAULT_REMOVE_BATCH_SIZE = 100; + + private volatile int removeBatchSize = DEFAULT_REMOVE_BATCH_SIZE; + + /** + * Set the batch size when bulk removing messages from groups for message stores + * that support batch removal. + * Default 100. + * @param removeBatchSize the batch size. + * @since 4.2 + */ + public void setRemoveBatchSize(int removeBatchSize) { + this.removeBatchSize = removeBatchSize; + } + + public int getRemoveBatchSize() { + return removeBatchSize; + } + +} 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 0b1d6182dd..91afbda1b9 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 @@ -1,5 +1,5 @@ /* - * Copyright 2002-2013 the original author or authors + * Copyright 2002-2015 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. @@ -34,6 +34,7 @@ import org.springframework.util.Assert; * Base class for implementations of Key/Value style {@link MessageGroupStore} and {@link MessageStore} * * @author Oleg Zhurakousky + * @author Gary Russell * @since 2.1 */ public abstract class AbstractKeyValueMessageStore extends AbstractMessageGroupStore implements MessageStore{ @@ -48,9 +49,9 @@ public abstract class AbstractKeyValueMessageStore extends AbstractMessageGroupS @Override public Message getMessage(UUID id) { - Message message = this.getRawMessage(id); + Message message = getRawMessage(id); if (message != null){ - return this.normalizeMessage(message); + return normalizeMessage(message); } return null; } @@ -60,19 +61,19 @@ public abstract class AbstractKeyValueMessageStore extends AbstractMessageGroupS public Message addMessage(Message message) { Assert.notNull(message, "'message' must not be null"); UUID messageId = message.getHeaders().getId(); - this.doStore(MESSAGE_KEY_PREFIX + messageId, message); - return (Message) this.getRawMessage(messageId); + doStore(MESSAGE_KEY_PREFIX + messageId, message); + return (Message) getRawMessage(messageId); } @Override public Message removeMessage(UUID id) { Assert.notNull(id, "'id' must not be null"); - Object message = this.doRemove(MESSAGE_KEY_PREFIX + id); + Object message = doRemove(MESSAGE_KEY_PREFIX + id); if (message != null) { Assert.isInstanceOf(Message.class, message); } if (message != null){ - return this.normalizeMessage((Message) message); + return normalizeMessage((Message) message); } return null; } @@ -80,7 +81,7 @@ public abstract class AbstractKeyValueMessageStore extends AbstractMessageGroupS @Override @ManagedAttribute public long getMessageCount() { - Collection messageIds = this.doListKeys(MESSAGE_KEY_PREFIX + "*"); + Collection messageIds = doListKeys(MESSAGE_KEY_PREFIX + "*"); return (messageIds != null) ? messageIds.size() : 0; } @@ -92,7 +93,7 @@ public abstract class AbstractKeyValueMessageStore extends AbstractMessageGroupS */ @Override public MessageGroup getMessageGroup(Object groupId) { - return this.buildMessageGroup(groupId, false); + return buildMessageGroup(groupId, false); } @@ -105,25 +106,25 @@ public abstract class AbstractKeyValueMessageStore extends AbstractMessageGroupS Assert.notNull(message, "'message' must not be null"); // add message as is to the MG accessible by the caller - SimpleMessageGroup messageGroup = this.getSimpleMessageGroup(this.getMessageGroup(groupId)); + SimpleMessageGroup messageGroup = getSimpleMessageGroup(getMessageGroup(groupId)); messageGroup.add(message); // enrich Message with additional headers and add it to MS - Message enrichedMessage = this.enrichMessage(message); + Message enrichedMessage = enrichMessage(message); - this.addMessage(enrichedMessage); + addMessage(enrichedMessage); // build raw MessageGroup and add enriched Message to it - SimpleMessageGroup rawGroup = this.buildMessageGroup(groupId, true); + SimpleMessageGroup rawGroup = buildMessageGroup(groupId, true); rawGroup.setLastModified(System.currentTimeMillis()); rawGroup.add(enrichedMessage); // store MessageGroupMetadata built from enriched MG - this.doStore(MESSAGE_GROUP_KEY_PREFIX + groupId, new MessageGroupMetadata(rawGroup)); + doStore(MESSAGE_GROUP_KEY_PREFIX + groupId, new MessageGroupMetadata(rawGroup)); // return clean MG - return this.getMessageGroup(groupId); + return getMessageGroup(groupId); } /** @@ -135,33 +136,52 @@ public abstract class AbstractKeyValueMessageStore extends AbstractMessageGroupS Assert.notNull(messageToRemove, "'messageToRemove' must not be null"); // build raw MG - SimpleMessageGroup rawGroup = this.buildMessageGroup(groupId, true); + SimpleMessageGroup rawGroup = buildMessageGroup(groupId, true); // create a clean instance of - SimpleMessageGroup messageGroup = this.normalizeSimpleMessageGroup(rawGroup); + SimpleMessageGroup messageGroup = normalizeSimpleMessageGroup(rawGroup); for (Message message : rawGroup.getMessages()) { if (message.getHeaders().getId().equals(messageToRemove.getHeaders().getId())){ rawGroup.remove(message); } } - this.removeMessage(messageToRemove.getHeaders().getId()); + removeMessage(messageToRemove.getHeaders().getId()); rawGroup.setLastModified(System.currentTimeMillis()); - this.doStore(MESSAGE_GROUP_KEY_PREFIX + groupId, new MessageGroupMetadata(rawGroup)); - messageGroup = this.getSimpleMessageGroup(this.getMessageGroup(groupId)); + doStore(MESSAGE_GROUP_KEY_PREFIX + groupId, new MessageGroupMetadata(rawGroup)); + messageGroup = getSimpleMessageGroup(getMessageGroup(groupId)); return messageGroup; } + @Override + public void removeMessagesFromGroup(Object groupId, Collection> messages) { + Assert.notNull(groupId, "'groupId' must not be null"); + Assert.notNull(messages, "'messages' must not be null"); + + Object mgm = doRetrieve(MESSAGE_GROUP_KEY_PREFIX + groupId); + if (mgm != null) { + Assert.isInstanceOf(MessageGroupMetadata.class, mgm); + MessageGroupMetadata messageGroupMetadata = (MessageGroupMetadata) mgm; + for (Message messageToRemove : messages) { + UUID messageId = messageToRemove.getHeaders().getId(); + messageGroupMetadata.remove(messageId); + doRemove(MESSAGE_KEY_PREFIX + messageId); + } + messageGroupMetadata.setLastModified(System.currentTimeMillis()); + doStore(MESSAGE_GROUP_KEY_PREFIX + groupId, messageGroupMetadata); + } + } + @Override public void completeGroup(Object groupId) { Assert.notNull(groupId, "'groupId' must not be null"); - SimpleMessageGroup messageGroup = this.buildMessageGroup(groupId, true); + SimpleMessageGroup messageGroup = buildMessageGroup(groupId, true); messageGroup.complete(); messageGroup.setLastModified(System.currentTimeMillis()); - this.doStore(MESSAGE_GROUP_KEY_PREFIX + groupId, new MessageGroupMetadata(messageGroup)); + doStore(MESSAGE_GROUP_KEY_PREFIX + groupId, new MessageGroupMetadata(messageGroup)); } /** @@ -170,14 +190,14 @@ public abstract class AbstractKeyValueMessageStore extends AbstractMessageGroupS @Override public void removeMessageGroup(Object groupId) { Assert.notNull(groupId, "'groupId' must not be null"); - Object mgm = this.doRemove(MESSAGE_GROUP_KEY_PREFIX + groupId); + Object mgm = doRemove(MESSAGE_GROUP_KEY_PREFIX + groupId); if (mgm != null) { Assert.isInstanceOf(MessageGroupMetadata.class, mgm); MessageGroupMetadata messageGroupMetadata = (MessageGroupMetadata) mgm; Iterator messageIds = messageGroupMetadata.messageIdIterator(); while (messageIds.hasNext()){ - this.removeMessage(messageIds.next()); + removeMessage(messageIds.next()); } } } @@ -185,16 +205,16 @@ public abstract class AbstractKeyValueMessageStore extends AbstractMessageGroupS @Override public void setLastReleasedSequenceNumberForGroup(Object groupId, int sequenceNumber) { Assert.notNull(groupId, "'groupId' must not be null"); - SimpleMessageGroup messageGroup = this.buildMessageGroup(groupId, true); + SimpleMessageGroup messageGroup = buildMessageGroup(groupId, true); messageGroup.setLastReleasedMessageSequenceNumber(sequenceNumber); messageGroup.setLastModified(System.currentTimeMillis()); - this.doStore(MESSAGE_GROUP_KEY_PREFIX + groupId, new MessageGroupMetadata(messageGroup)); + doStore(MESSAGE_GROUP_KEY_PREFIX + groupId, new MessageGroupMetadata(messageGroup)); } @Override public Message pollMessageFromGroup(Object groupId) { Assert.notNull(groupId, "'groupId' must not be null"); - Object mgm = this.doRetrieve(MESSAGE_GROUP_KEY_PREFIX + groupId); + Object mgm = doRetrieve(MESSAGE_GROUP_KEY_PREFIX + groupId); if (mgm != null) { Assert.isInstanceOf(MessageGroupMetadata.class, mgm); MessageGroupMetadata messageGroupMetadata = (MessageGroupMetadata) mgm; @@ -203,8 +223,8 @@ public abstract class AbstractKeyValueMessageStore extends AbstractMessageGroupS if (firstId != null){ messageGroupMetadata.remove(firstId); messageGroupMetadata.setLastModified(System.currentTimeMillis()); - this.doStore(MESSAGE_GROUP_KEY_PREFIX + groupId, messageGroupMetadata); - return this.removeMessage(firstId); + doStore(MESSAGE_GROUP_KEY_PREFIX + groupId, messageGroupMetadata); + return removeMessage(firstId); } } return null; @@ -213,8 +233,8 @@ public abstract class AbstractKeyValueMessageStore extends AbstractMessageGroupS @Override @SuppressWarnings("unchecked") public Iterator iterator() { - final Iterator idIterator = this.normalizeKeys( - (Collection) this.doListKeys(MESSAGE_GROUP_KEY_PREFIX + "*")) + final Iterator idIterator = normalizeKeys( + (Collection) doListKeys(MESSAGE_GROUP_KEY_PREFIX + "*")) .iterator(); return new MessageGroupIterator(idIterator); } @@ -236,7 +256,7 @@ public abstract class AbstractKeyValueMessageStore extends AbstractMessageGroupS @Override public int messageGroupSize(Object groupId) { - Object mgm = this.doRetrieve(MESSAGE_GROUP_KEY_PREFIX + groupId); + Object mgm = doRetrieve(MESSAGE_GROUP_KEY_PREFIX + groupId); if (mgm != null) { Assert.isInstanceOf(MessageGroupMetadata.class, mgm); MessageGroupMetadata messageGroupMetadata = (MessageGroupMetadata) mgm; @@ -255,7 +275,7 @@ public abstract class AbstractKeyValueMessageStore extends AbstractMessageGroupS @SuppressWarnings({ "rawtypes", "unchecked" }) private Message normalizeMessage(Message message){ - Message normalizedMessage = this.getMessageBuilderFactory().fromMessage(message) + Message normalizedMessage = getMessageBuilderFactory().fromMessage(message) .removeHeader("CREATED_DATE") .build(); Map innerMap = (Map) new DirectFieldAccessor(normalizedMessage.getHeaders()).getPropertyValue("headers"); @@ -269,7 +289,7 @@ public abstract class AbstractKeyValueMessageStore extends AbstractMessageGroupS */ @SuppressWarnings({ "rawtypes", "unchecked" }) private Message enrichMessage(Message message){ - Message enrichedMessage = this.getMessageBuilderFactory().fromMessage(message) + Message enrichedMessage = getMessageBuilderFactory().fromMessage(message) .setHeader(CREATED_DATE, System.currentTimeMillis()) .build(); Map innerMap = (Map) new DirectFieldAccessor(enrichedMessage.getHeaders()).getPropertyValue("headers"); @@ -280,7 +300,7 @@ public abstract class AbstractKeyValueMessageStore extends AbstractMessageGroupS private SimpleMessageGroup buildMessageGroup(Object groupId, boolean raw){ Assert.notNull(groupId, "'groupId' must not be null"); - Object mgm = this.doRetrieve(MESSAGE_GROUP_KEY_PREFIX + groupId); + Object mgm = doRetrieve(MESSAGE_GROUP_KEY_PREFIX + groupId); if (mgm != null) { Assert.isInstanceOf(MessageGroupMetadata.class, mgm); MessageGroupMetadata messageGroupMetadata = (MessageGroupMetadata) mgm; @@ -289,10 +309,10 @@ public abstract class AbstractKeyValueMessageStore extends AbstractMessageGroupS Iterator messageIds = messageGroupMetadata.messageIdIterator(); while (messageIds.hasNext()){ if (raw){ - messages.add(this.getRawMessage(messageIds.next())); + messages.add(getRawMessage(messageIds.next())); } else { - messages.add(this.getMessage(messageIds.next())); + messages.add(getMessage(messageIds.next())); } } @@ -319,7 +339,7 @@ public abstract class AbstractKeyValueMessageStore extends AbstractMessageGroupS private SimpleMessageGroup normalizeSimpleMessageGroup(SimpleMessageGroup messageGroup){ SimpleMessageGroup normalizedGroup = new SimpleMessageGroup(messageGroup.getGroupId()); for (Message message : messageGroup.getMessages()) { - Message normailizedMessage = this.normalizeMessage(message); + Message normailizedMessage = normalizeMessage(message); normalizedGroup.add(normailizedMessage); } return normalizedGroup; @@ -327,7 +347,7 @@ public abstract class AbstractKeyValueMessageStore extends AbstractMessageGroupS private Message getRawMessage(UUID id) { Assert.notNull(id, "'id' must not be null"); - Object message = this.doRetrieve(MESSAGE_KEY_PREFIX + id); + Object message = doRetrieve(MESSAGE_KEY_PREFIX + id); return (Message) message; } @@ -341,7 +361,7 @@ public abstract class AbstractKeyValueMessageStore extends AbstractMessageGroupS @Override public boolean hasNext() { - return idIterator.hasNext(); + return this.idIterator.hasNext(); } @Override 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 432c38b62b..0dc50d1ed3 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 @@ -13,6 +13,7 @@ package org.springframework.integration.store; +import java.util.Arrays; import java.util.Collection; import java.util.LinkedHashSet; @@ -39,8 +40,8 @@ import org.springframework.messaging.Message; */ @ManagedResource @IntegrationManagedResource -public abstract class AbstractMessageGroupStore implements MessageGroupStore, Iterable, - BeanFactoryAware { +public abstract class AbstractMessageGroupStore extends AbstractBatchingMessageGroupStore + implements MessageGroupStore, Iterable, BeanFactoryAware { protected final Log logger = LogFactory.getLog(getClass()); @@ -165,6 +166,11 @@ public abstract class AbstractMessageGroupStore implements MessageGroupStore, It throw new UnsupportedOperationException("Not yet implemented for this store"); } + @Override + public void removeMessagesFromGroup(Object key, Message... messages) { + removeMessagesFromGroup(key, Arrays.asList(messages)); + } + @Override public Message getOneMessageFromGroup(Object groupId) { throw new UnsupportedOperationException("Not yet implemented for this store"); 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 dd558804f6..015e4ff053 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 @@ -1,5 +1,5 @@ /* - * Copyright 2002-2014 the original author or authors. + * Copyright 2002-2015 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. @@ -50,8 +50,6 @@ public class MessageGroupMetadata implements Serializable { private final boolean hasMessages; - private final int size; - private final UUID first; public MessageGroupMetadata(MessageGroup messageGroup) { @@ -66,10 +64,6 @@ public class MessageGroupMetadata implements Serializable { for (Message message : messageGroup.getMessages()) { this.messageIds.add(message.getHeaders().getId()); } - this.size = this.messageIds.size(); - } - else { - this.size = messageGroup.size(); } this.complete = messageGroup.isComplete(); this.timestamp = messageGroup.getTimestamp(); @@ -102,7 +96,7 @@ public class MessageGroupMetadata implements Serializable { } public int size(){ - return this.size; + return this.messageIds.size(); } public UUID firstId(){ 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 0f5f9fe07e..66e79d23e3 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 @@ -1,5 +1,5 @@ /* - * Copyright 2002-2014 the original author or authors. + * Copyright 2002-2015 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 @@ -12,6 +12,7 @@ */ package org.springframework.integration.store; +import java.util.Collection; import java.util.Iterator; import org.springframework.jmx.export.annotation.ManagedAttribute; @@ -50,7 +51,7 @@ public interface MessageGroupStore extends BasicMessageGroupStore { int getMessageGroupCount(); /** - * Persist a deletion on a single message from the group. The group is modified to reflect that 'messageToRemove' is + * 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. @@ -59,6 +60,26 @@ public interface MessageGroupStore extends BasicMessageGroupStore { */ MessageGroup removeMessageFromGroup(Object key, Message messageToRemove); + /** + * 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)}. * 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 f6a8158e1a..927712b918 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 @@ -1,5 +1,5 @@ /* - * Copyright 2002-2014 the original author or authors. + * Copyright 2002-2015 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 @@ -269,6 +269,30 @@ public class SimpleMessageStore extends AbstractMessageGroupStore } } + @Override + public void removeMessagesFromGroup(Object groupId, Collection> messages) { + Lock lock = this.lockRegistry.obtain(groupId); + try { + lock.lockInterruptibly(); + try { + SimpleMessageGroup group = this.groupIdToMessageGroup.get(groupId); + Assert.notNull(group, "MessageGroup for groupId '" + groupId + "' " + + "can not be located while attempting to remove Message(s) from the MessageGroup"); + for (Message messageToRemove : messages) { + group.remove(messageToRemove); + } + group.setLastModified(System.currentTimeMillis()); + } + finally { + lock.unlock(); + } + } + catch (InterruptedException e) { + Thread.currentThread().interrupt(); + throw new MessagingException("Interrupted while obtaining lock", e); + } + } + @Override public Iterator iterator() { return new HashSet(groupIdToMessageGroup.values()).iterator(); @@ -325,7 +349,7 @@ public class SimpleMessageStore extends AbstractMessageGroupStore if (!CollectionUtils.isEmpty(messageList)){ message = messageList.iterator().next(); if (message != null){ - this.removeMessageFromGroup(groupId, message); + this.removeMessagesFromGroup(groupId, message); } } return message; 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 d401fb4ebb..d476112de6 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 @@ -1,5 +1,5 @@ /* - * Copyright 2002-2013 the original author or authors. + * Copyright 2002-2015 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. @@ -25,9 +25,10 @@ import java.util.Iterator; import java.util.List; import org.junit.Test; + +import org.springframework.integration.store.MessageGroupStore.MessageGroupCallback; import org.springframework.messaging.Message; import org.springframework.messaging.support.GenericMessage; -import org.springframework.integration.store.MessageGroupStore.MessageGroupCallback; import org.springframework.test.util.ReflectionTestUtils; /** @@ -40,6 +41,7 @@ public class MessageStoreTests { public void shouldRegisterCallbacks() throws Exception { TestMessageStore store = new TestMessageStore(); store.setExpiryCallbacks(Arrays. asList(new MessageGroupStore.MessageGroupCallback() { + @Override public void execute(MessageGroupStore messageGroupStore, MessageGroup group) { } })); @@ -52,6 +54,7 @@ public class MessageStoreTests { TestMessageStore store = new TestMessageStore(); final List list = new ArrayList(); store.registerMessageGroupExpiryCallback(new MessageGroupCallback() { + @Override public void execute(MessageGroupStore messageGroupStore, MessageGroup group) { list.add(group.getOne().getPayload().toString()); messageGroupStore.removeMessageGroup(group.getGroupId()); @@ -84,41 +87,55 @@ public class MessageStoreTests { private boolean removed = false; + @Override public Iterator iterator() { return Arrays.asList(testMessages).iterator(); } + @Override public MessageGroup addMessageToGroup(Object correlationKey, Message message) { throw new UnsupportedOperationException(); } + @Override public MessageGroup getMessageGroup(Object correlationKey) { return removed ? new SimpleMessageGroup(correlationKey) : testMessages; } + @Override public MessageGroup removeMessageFromGroup(Object key, Message messageToRemove) { throw new UnsupportedOperationException(); } + @Override + public void removeMessagesFromGroup(Object key, Collection> messages) { + throw new UnsupportedOperationException(); + } + + @Override public void removeMessageGroup(Object correlationKey) { if (correlationKey.equals(testMessages.getGroupId())) { removed = true; } } + @Override public void setLastReleasedSequenceNumberForGroup(Object groupId, int sequenceNumber) { throw new UnsupportedOperationException(); } + @Override public void completeGroup(Object groupId) { throw new UnsupportedOperationException(); } + @Override public Message pollMessageFromGroup(Object groupId) { return null; } + @Override public int messageGroupSize(Object groupId) { return 0; } diff --git a/spring-integration-core/src/test/java/org/springframework/integration/store/SimpleMessageStoreTests.java b/spring-integration-core/src/test/java/org/springframework/integration/store/SimpleMessageStoreTests.java index edeb3ff15a..15eaa3c601 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/store/SimpleMessageStoreTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/store/SimpleMessageStoreTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2013 the original author or authors. + * Copyright 2002-2015 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. @@ -27,10 +27,11 @@ import java.util.Collection; import java.util.List; import org.junit.Test; -import org.springframework.messaging.Message; -import org.springframework.messaging.MessagingException; + import org.springframework.integration.store.MessageGroupStore.MessageGroupCallback; import org.springframework.integration.support.MessageBuilder; +import org.springframework.messaging.Message; +import org.springframework.messaging.MessagingException; import org.springframework.test.util.ReflectionTestUtils; /** @@ -120,6 +121,7 @@ public class SimpleMessageStoreTests { public void shouldRegisterCallbacks() throws Exception { SimpleMessageStore store = new SimpleMessageStore(); store.setExpiryCallbacks(Arrays. asList(new MessageGroupStore.MessageGroupCallback() { + @Override public void execute(MessageGroupStore messageGroupStore, MessageGroup group) { } })); @@ -132,6 +134,7 @@ public class SimpleMessageStoreTests { SimpleMessageStore store = new SimpleMessageStore(); final List list = new ArrayList(); store.registerMessageGroupExpiryCallback(new MessageGroupCallback() { + @Override public void execute(MessageGroupStore messageGroupStore, MessageGroup group) { list.add(group.getOne().getPayload().toString()); messageGroupStore.removeMessageGroup(group.getGroupId()); @@ -148,4 +151,21 @@ public class SimpleMessageStoreTests { } + @Test + public void testAddAndRemoveMessagesFromMessageGroup() throws Exception { + SimpleMessageStore messageStore = new SimpleMessageStore(); + String groupId = "X"; + 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); + } + MessageGroup group = messageStore.getMessageGroup(groupId); + assertEquals(25, group.size()); + messageStore.removeMessagesFromGroup(groupId, messages); + group = messageStore.getMessageGroup(groupId); + assertEquals(0, group.size()); + } + } 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 670dc20efc..90113342fc 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 @@ -20,6 +20,8 @@ import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertNotNull; import static org.junit.Assert.assertSame; +import java.util.ArrayList; +import java.util.List; import java.util.Properties; import org.junit.After; @@ -30,6 +32,7 @@ import org.springframework.data.gemfire.CacheFactoryBean; import org.springframework.data.gemfire.RegionFactoryBean; import org.springframework.integration.channel.DirectChannel; import org.springframework.integration.history.MessageHistory; +import org.springframework.integration.store.MessageGroup; import org.springframework.integration.support.MessageBuilder; import org.springframework.integration.test.util.TestUtils; import org.springframework.messaging.Message; @@ -98,6 +101,25 @@ public class GemfireMessageStoreTests { assertEquals("channel", fooChannelHistory.get("type")); } + @Test + public void testAddAndRemoveMessagesFromMessageGroup() throws Exception { + GemfireMessageStore messageStore = new GemfireMessageStore(this.region); + messageStore.afterPropertiesSet(); + + String groupId = "X"; + 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); + } + MessageGroup group = messageStore.getMessageGroup(groupId); + assertEquals(25, group.size()); + messageStore.removeMessagesFromGroup(groupId, messages); + group = messageStore.getMessageGroup(groupId); + assertEquals(0, group.size()); + } + @Before public void init() throws Exception { CacheFactoryBean cacheFactoryBean = new CacheFactoryBean(); 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 409dcc8c43..05ed21663c 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 @@ -19,6 +19,7 @@ import java.sql.SQLException; import java.sql.Timestamp; import java.sql.Types; import java.util.ArrayList; +import java.util.Collection; import java.util.Date; import java.util.HashMap; import java.util.Iterator; @@ -47,6 +48,7 @@ import org.springframework.integration.store.SimpleMessageGroup; import org.springframework.integration.util.UUIDConverter; import org.springframework.jdbc.core.JdbcOperations; import org.springframework.jdbc.core.JdbcTemplate; +import org.springframework.jdbc.core.ParameterizedPreparedStatementSetter; import org.springframework.jdbc.core.PreparedStatementSetter; import org.springframework.jdbc.core.RowCallbackHandler; import org.springframework.jdbc.core.RowMapper; @@ -483,6 +485,40 @@ public class JdbcMessageStore extends AbstractMessageGroupStore implements Messa return getMessageGroup(groupId); } + @Override + public void removeMessagesFromGroup(Object groupId, Collection> messages) { + Assert.notNull(groupId, "'groupId' must not be null"); + Assert.notNull(messages, "'messages' must not be null"); + + final String groupKey = getKey(groupId); + + if (logger.isDebugEnabled()){ + logger.debug("Removing messages from group with group key=" + groupKey); + } + jdbcTemplate.batchUpdate(getQuery(Query.REMOVE_MESSAGE_FROM_GROUP), + 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, region); + } + }); + 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, region); + } + }); + this.updateMessageGroup(groupKey); + } + @Override public void removeMessageGroup(Object groupId) { @@ -560,7 +596,7 @@ public class JdbcMessageStore extends AbstractMessageGroupStore implements Messa Message polledMessage = this.doPollForMessage(key); if (polledMessage != null){ - this.removeMessageFromGroup(groupId, polledMessage); + this.removeMessagesFromGroup(groupId, polledMessage); } return polledMessage; } 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 2b759503cc..6b8a7943b8 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,6 +30,8 @@ import java.io.IOException; import java.io.InputStream; import java.io.InputStreamReader; import java.io.OutputStream; +import java.util.ArrayList; +import java.util.List; import java.util.Properties; import java.util.UUID; @@ -246,6 +248,25 @@ public class JdbcMessageStoreTests { 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); + } + MessageGroup group = messageStore.getMessageGroup(groupId); + assertEquals(25, group.size()); + messageStore.removeMessagesFromGroup(groupId, messages); + group = messageStore.getMessageGroup(groupId); + assertEquals(0, group.size()); + } + @Test @Transactional public void testRemoveMessageGroup() throws Exception { 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 8f202880dd..dbeb764179 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 @@ -49,6 +49,7 @@ 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.BasicMessageGroupStore; import org.springframework.integration.support.DefaultMessageBuilderFactory; import org.springframework.integration.support.MessageBuilderFactory; @@ -65,8 +66,8 @@ import org.springframework.util.Assert; * @since 4.0 */ -public abstract class AbstractConfigurableMongoDbMessageStore implements BasicMessageGroupStore, InitializingBean, - ApplicationContextAware { +public abstract class AbstractConfigurableMongoDbMessageStore extends AbstractBatchingMessageGroupStore + implements BasicMessageGroupStore, InitializingBean, ApplicationContextAware { public final static String SEQUENCE_NAME = "messagesSequence"; 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 eb46420a50..f40e1a6983 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 @@ -17,6 +17,7 @@ package org.springframework.integration.mongodb.store; import java.util.ArrayList; +import java.util.Arrays; import java.util.Collection; import java.util.Iterator; import java.util.LinkedHashSet; @@ -48,6 +49,7 @@ import org.springframework.util.Assert; * * @author Amol Nayak * @author Artem Bilan + * @author Gary Russell * @since 3.0 */ public class ConfigurableMongoDbMessageStore extends AbstractConfigurableMongoDbMessageStore @@ -205,10 +207,39 @@ public class ConfigurableMongoDbMessageStore extends AbstractConfigurableMongoDb Query query = groupIdQuery(groupId) .addCriteria(Criteria.where(MessageDocumentFields.MESSAGE_ID).is(messageToRemove.getHeaders().getId())); - mongoTemplate.remove(query, collectionName); + this.mongoTemplate.remove(query, this.collectionName); updateGroup(groupId, lastModifiedUpdate()); return getMessageGroup(groupId); + } + @Override + public void removeMessagesFromGroup(Object groupId, Collection> messages) { + Assert.notNull(groupId, "'groupId' must not be null"); + Assert.notNull(messages, "'messageToRemove' must not be null"); + + Collection ids = new ArrayList(); + for (Message messageToRemove : messages) { + ids.add(messageToRemove.getHeaders().getId()); + if (ids.size() >= getRemoveBatchSize()) { + removeMessages(groupId, ids); + ids.clear(); + } + } + if (ids.size() > 0) { + removeMessages(groupId, ids); + } + updateGroup(groupId, lastModifiedUpdate()); + } + + private void removeMessages(Object groupId, Collection ids) { + Query query = groupIdQuery(groupId) + .addCriteria(Criteria.where(MessageDocumentFields.MESSAGE_ID).in(ids.toArray())); + this.mongoTemplate.remove(query, this.collectionName); + } + + @Override + public void removeMessagesFromGroup(Object groupId, Message... messages) { + removeMessagesFromGroup(groupId, Arrays.asList(messages)); } @Override 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 7b78153462..7645f4c458 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 @@ -17,6 +17,7 @@ package org.springframework.integration.mongodb.store; import java.util.ArrayList; +import java.util.Collection; import java.util.HashMap; import java.util.HashSet; import java.util.Iterator; @@ -76,6 +77,7 @@ import org.springframework.util.StringUtils; import com.mongodb.BasicDBList; import com.mongodb.BasicDBObject; +import com.mongodb.BulkWriteOperation; import com.mongodb.DBObject; @@ -313,6 +315,35 @@ public class MongoDbMessageStore extends AbstractMessageGroupStore return getMessageGroup(groupId); } + @Override + public void removeMessagesFromGroup(Object groupId, Collection> messages) { + Assert.notNull(groupId, "'groupId' must not be null"); + Assert.notNull(messages, "'messageToRemove' must not be null"); + + Collection ids = new ArrayList(); + for (Message messageToRemove : messages) { + ids.add(messageToRemove.getHeaders().getId()); + if (ids.size() >= getRemoveBatchSize()) { + bulkRemove(groupId, ids); + ids.clear(); + } + } + if (ids.size() > 0) { + bulkRemove(groupId, ids); + } + updateGroup(groupId, lastModifiedUpdate()); + } + + private void bulkRemove(Object groupId, Collection ids) { + BulkWriteOperation bulkOp = this.template.getCollection(this.collectionName) + .initializeOrderedBulkOperation(); + for (UUID id : ids) { + bulkOp.find(whereMessageIdIsAndGroupIdIs(id, groupId).getQueryObject()) + .remove(); + } + bulkOp.execute(); + } + @Override public void removeMessageGroup(Object groupId) { this.template.remove(whereGroupIdIs(groupId), this.collectionName); @@ -400,7 +431,6 @@ public class MongoDbMessageStore extends AbstractMessageGroupStore return new Query(Criteria.where("headers.id._value").is(id.toString()).and(GROUP_ID_KEY).is(groupId)); } - private static Query whereGroupIdOrder(Object groupId) { return whereGroupIdIs(groupId).with(new Sort(Sort.Direction.DESC, GROUP_UPDATE_TIMESTAMP_KEY, SEQUENCE)); } 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 0bddd71bf7..f20a4670f5 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 @@ -1,5 +1,5 @@ /* - * Copyright 2002-2013 the original author or authors. + * Copyright 2002-2015 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. @@ -15,14 +15,19 @@ */ package org.springframework.integration.mongodb.store; -import static org.junit.Assert.*; +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.assertTrue; +import java.util.ArrayList; import java.util.Collection; import java.util.Iterator; +import java.util.List; import java.util.Properties; import java.util.UUID; -import com.mongodb.MongoClient; import org.junit.Test; import org.springframework.context.support.ClassPathXmlApplicationContext; @@ -32,6 +37,7 @@ import org.springframework.integration.channel.QueueChannel; import org.springframework.integration.history.MessageHistory; import org.springframework.integration.mongodb.rules.MongoDbAvailable; import org.springframework.integration.mongodb.rules.MongoDbAvailableTests; +import org.springframework.integration.store.AbstractBatchingMessageGroupStore; import org.springframework.integration.store.MessageGroup; import org.springframework.integration.store.MessageGroupStore; import org.springframework.integration.store.MessageStore; @@ -41,6 +47,8 @@ import org.springframework.messaging.Message; import org.springframework.messaging.MessageChannel; import org.springframework.messaging.support.GenericMessage; +import com.mongodb.MongoClient; + /** * @author Oleg Zhurakousky * @author Gary Russell @@ -394,6 +402,26 @@ public abstract class AbstractMongoDbMessageGroupStoreTests extends MongoDbAvail assertEquals(2, counter); } + @Test + @MongoDbAvailable + public void testAddAndRemoveMessagesFromMessageGroup() throws Exception { + MessageGroupStore messageStore = (MessageGroupStore) this.getMessageStore(); + String groupId = "X"; + messageStore.removeMessageGroup("X"); + ((AbstractBatchingMessageGroupStore) 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); + } + MessageGroup group = messageStore.getMessageGroup(groupId); + assertEquals(25, group.size()); + messageStore.removeMessagesFromGroup(groupId, messages); + group = messageStore.getMessageGroup(groupId); + assertEquals(0, group.size()); + } + // @Test // @MongoDbAvailable // public void testConcurrentModifications() throws Exception{ @@ -461,6 +489,7 @@ public abstract class AbstractMongoDbMessageGroupStoreTests extends MongoDbAvail Message m3 = MessageBuilder.withPayload("3").setSequenceNumber(3).setSequenceSize(3).setCorrelationId(1).build(); input.send(m3); assertNotNull(output.receive(2000)); + context.close(); } @Test 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 fedef70913..7a1bfecfee 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 @@ -15,12 +15,12 @@ */ package org.springframework.integration.mongodb.store; -import static org.junit.Assert.*; +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertNotNull; +import static org.junit.Assert.assertThat; import java.util.Map; -import com.mongodb.DBObject; -import com.mongodb.MongoClient; import org.hamcrest.Matchers; import org.junit.Test; @@ -38,6 +38,9 @@ import org.springframework.integration.support.MessageBuilder; import org.springframework.integration.test.util.TestUtils; import org.springframework.messaging.Message; +import com.mongodb.DBObject; +import com.mongodb.MongoClient; + /** * @author Amol Nayak * @author Artem Bilan @@ -45,9 +48,6 @@ import org.springframework.messaging.Message; */ public class ConfigurableMongoDbMessageGroupStoreTests extends AbstractMongoDbMessageGroupStoreTests { - /* (non-Javadoc) - * @see org.springframework.integration.mongodb.store.AbstractMongoDbMessageGroupStoreTests#getMessageGroupStore() - */ @Override protected ConfigurableMongoDbMessageStore getMessageGroupStore() throws Exception { MongoDbFactory mongoDbFactory = new SimpleMongoDbFactory(new MongoClient(), "test"); @@ -59,9 +59,6 @@ public class ConfigurableMongoDbMessageGroupStoreTests extends AbstractMongoDbMe return mongoDbMessageStore; } - /* (non-Javadoc) - * @see org.springframework.integration.mongodb.store.AbstractMongoDbMessageGroupStoreTests#getMessageStore() - */ @Override protected MessageStore getMessageStore() throws Exception { return this.getMessageGroupStore(); 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 77916b22a7..a22d0a81c7 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 @@ -1,5 +1,5 @@ /* - * Copyright 2007-2014 the original author or authors + * Copyright 2007-2015 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. @@ -29,8 +29,6 @@ import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; import java.util.concurrent.TimeUnit; -import junit.framework.AssertionFailedError; - import org.junit.After; import org.junit.Before; import org.junit.Ignore; @@ -51,6 +49,8 @@ import org.springframework.messaging.Message; import org.springframework.messaging.MessageChannel; import org.springframework.messaging.support.GenericMessage; +import junit.framework.AssertionFailedError; + /** * @author Oleg Zhurakousky * @author Artem Bilan @@ -62,7 +62,7 @@ public class RedisMessageGroupStoreTests extends RedisAvailableTests { @Before @After public void setUpTearDown() { - StringRedisTemplate template = this.createStringRedisTemplate(this.getConnectionFactoryForTest()); + StringRedisTemplate template = createStringRedisTemplate(getConnectionFactoryForTest()); template.delete("MESSAGE_GROUP_1"); template.delete("MESSAGE_GROUP_2"); template.delete("MESSAGE_GROUP_3"); @@ -71,7 +71,7 @@ public class RedisMessageGroupStoreTests extends RedisAvailableTests { @Test @RedisAvailable public void testNonExistingEmptyMessageGroup() throws Exception{ - RedisConnectionFactory jcf = this.getConnectionFactoryForTest(); + RedisConnectionFactory jcf = getConnectionFactoryForTest(); RedisMessageStore store = new RedisMessageStore(jcf); MessageGroup messageGroup = store.getMessageGroup(1); @@ -83,7 +83,7 @@ public class RedisMessageGroupStoreTests extends RedisAvailableTests { @Test @RedisAvailable public void testMessageGroupUpdatedDateChangesWithEachAddedMessage() throws Exception{ - RedisConnectionFactory jcf = this.getConnectionFactoryForTest(); + RedisConnectionFactory jcf = getConnectionFactoryForTest(); RedisMessageStore store = new RedisMessageStore(jcf); MessageGroup messageGroup = store.getMessageGroup(1); @@ -110,7 +110,7 @@ public class RedisMessageGroupStoreTests extends RedisAvailableTests { @Test @RedisAvailable public void testMessageGroupWithAddedMessage() throws Exception{ - RedisConnectionFactory jcf = this.getConnectionFactoryForTest(); + RedisConnectionFactory jcf = getConnectionFactoryForTest(); RedisMessageStore store = new RedisMessageStore(jcf); MessageGroup messageGroup = store.getMessageGroup(1); @@ -128,7 +128,7 @@ public class RedisMessageGroupStoreTests extends RedisAvailableTests { @Test @RedisAvailable public void testRemoveMessageGroup() throws Exception{ - RedisConnectionFactory jcf = this.getConnectionFactoryForTest(); + RedisConnectionFactory jcf = getConnectionFactoryForTest(); RedisMessageStore store = new RedisMessageStore(jcf); MessageGroup messageGroup = store.getMessageGroup(1); @@ -155,7 +155,7 @@ public class RedisMessageGroupStoreTests extends RedisAvailableTests { @Test @RedisAvailable public void testCompleteMessageGroup() throws Exception{ - RedisConnectionFactory jcf = this.getConnectionFactoryForTest(); + RedisConnectionFactory jcf = getConnectionFactoryForTest(); RedisMessageStore store = new RedisMessageStore(jcf); MessageGroup messageGroup = store.getMessageGroup(1); @@ -169,7 +169,7 @@ public class RedisMessageGroupStoreTests extends RedisAvailableTests { @Test @RedisAvailable public void testLastReleasedSequenceNumber() throws Exception{ - RedisConnectionFactory jcf = this.getConnectionFactoryForTest(); + RedisConnectionFactory jcf = getConnectionFactoryForTest(); RedisMessageStore store = new RedisMessageStore(jcf); MessageGroup messageGroup = store.getMessageGroup(1); @@ -183,7 +183,7 @@ public class RedisMessageGroupStoreTests extends RedisAvailableTests { @Test @RedisAvailable public void testRemoveMessageFromTheGroup() throws Exception{ - RedisConnectionFactory jcf = this.getConnectionFactoryForTest(); + RedisConnectionFactory jcf = getConnectionFactoryForTest(); RedisMessageStore store = new RedisMessageStore(jcf); MessageGroup messageGroup = store.getMessageGroup(1); @@ -206,7 +206,7 @@ public class RedisMessageGroupStoreTests extends RedisAvailableTests { @Test @RedisAvailable public void testWithMessageHistory() throws Exception{ - RedisConnectionFactory jcf = this.getConnectionFactoryForTest(); + RedisConnectionFactory jcf = getConnectionFactoryForTest(); RedisMessageStore store = new RedisMessageStore(jcf); store.getMessageGroup(1); @@ -233,7 +233,7 @@ public class RedisMessageGroupStoreTests extends RedisAvailableTests { @Test @RedisAvailable public void testRemoveNonExistingMessageFromTheGroup() throws Exception{ - RedisConnectionFactory jcf = this.getConnectionFactoryForTest(); + RedisConnectionFactory jcf = getConnectionFactoryForTest(); RedisMessageStore store = new RedisMessageStore(jcf); MessageGroup messageGroup = store.getMessageGroup(1); @@ -244,7 +244,7 @@ public class RedisMessageGroupStoreTests extends RedisAvailableTests { @Test @RedisAvailable public void testRemoveNonExistingMessageFromNonExistingTheGroup() throws Exception{ - RedisConnectionFactory jcf = this.getConnectionFactoryForTest(); + RedisConnectionFactory jcf = getConnectionFactoryForTest(); RedisMessageStore store = new RedisMessageStore(jcf); store.removeMessageFromGroup(1, new GenericMessage("2")); } @@ -254,7 +254,7 @@ public class RedisMessageGroupStoreTests extends RedisAvailableTests { @Test @RedisAvailable public void testMultipleInstancesOfGroupStore() throws Exception{ - RedisConnectionFactory jcf = this.getConnectionFactoryForTest(); + RedisConnectionFactory jcf = getConnectionFactoryForTest(); RedisMessageStore store1 = new RedisMessageStore(jcf); RedisMessageStore store2 = new RedisMessageStore(jcf); @@ -275,7 +275,7 @@ public class RedisMessageGroupStoreTests extends RedisAvailableTests { @Test @RedisAvailable public void testIteratorOfMessageGroups() throws Exception{ - RedisConnectionFactory jcf = this.getConnectionFactoryForTest(); + RedisConnectionFactory jcf = getConnectionFactoryForTest(); RedisMessageStore store1 = new RedisMessageStore(jcf); RedisMessageStore store2 = new RedisMessageStore(jcf); @@ -317,7 +317,7 @@ public class RedisMessageGroupStoreTests extends RedisAvailableTests { @Test @RedisAvailable @Ignore public void testConcurrentModifications() throws Exception{ - RedisConnectionFactory jcf = this.getConnectionFactoryForTest(); + RedisConnectionFactory jcf = getConnectionFactoryForTest(); final RedisMessageStore store1 = new RedisMessageStore(jcf); final RedisMessageStore store2 = new RedisMessageStore(jcf); @@ -361,7 +361,7 @@ public class RedisMessageGroupStoreTests extends RedisAvailableTests { @Test @RedisAvailable public void testWithAggregatorWithShutdown(){ - ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext("redis-aggregator-config.xml", this.getClass()); + ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext("redis-aggregator-config.xml", getClass()); MessageChannel input = context.getBean("inputChannel", MessageChannel.class); QueueChannel output = context.getBean("outputChannel", QueueChannel.class); @@ -373,7 +373,7 @@ public class RedisMessageGroupStoreTests extends RedisAvailableTests { assertNull(output.receive(1000)); context.close(); - context = new ClassPathXmlApplicationContext("redis-aggregator-config.xml", this.getClass()); + context = new ClassPathXmlApplicationContext("redis-aggregator-config.xml", getClass()); input = context.getBean("inputChannel", MessageChannel.class); output = context.getBean("outputChannel", QueueChannel.class); @@ -383,4 +383,25 @@ public class RedisMessageGroupStoreTests extends RedisAvailableTests { context.close(); } + @Test + @RedisAvailable + public void testAddAndRemoveMessagesFromMessageGroup() throws Exception { + RedisConnectionFactory jcf = getConnectionFactoryForTest(); + RedisMessageStore messageStore = new RedisMessageStore(jcf); + String groupId = "X"; + messageStore.removeMessageGroup("X"); + 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); + } + MessageGroup group = messageStore.getMessageGroup(groupId); + assertEquals(25, group.size()); + messageStore.removeMessagesFromGroup(groupId, messages); + group = messageStore.getMessageGroup(groupId); + assertEquals(0, group.size()); + messageStore.removeMessageGroup("X"); + } + } 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 135b97dfb8..0e7a82ed6c 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 @@ -1,5 +1,5 @@ /* - * Copyright 2007-2013 the original author or authors + * Copyright 2007-2015 the original author or authors * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -21,6 +21,8 @@ import static org.junit.Assert.assertNotSame; import static org.junit.Assert.assertNull; import java.io.Serializable; +import java.util.ArrayList; +import java.util.List; import java.util.Properties; import java.util.UUID; @@ -34,11 +36,14 @@ import org.springframework.integration.channel.DirectChannel; import org.springframework.integration.history.MessageHistory; import org.springframework.integration.redis.rules.RedisAvailable; import org.springframework.integration.redis.rules.RedisAvailableTests; +import org.springframework.integration.store.MessageGroup; +import org.springframework.integration.support.MessageBuilder; import org.springframework.messaging.Message; import org.springframework.messaging.support.GenericMessage; /** * @author Oleg Zhurakousky + * @author Gary Russell * */ public class RedisMessageStoreTests extends RedisAvailableTests { @@ -153,6 +158,23 @@ public class RedisMessageStoreTests extends RedisAvailableTests { assertEquals("channel", fooChannelHistory.get("type")); } + @Test + @RedisAvailable + public void testAddAndRemoveMessagesFromMessageGroup() throws Exception { + RedisConnectionFactory jcf = this.getConnectionFactoryForTest(); + RedisMessageStore messageStore = new RedisMessageStore(jcf); + String groupId = "X"; + 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.removeMessagesFromGroup(groupId, messages); + MessageGroup group = messageStore.getMessageGroup(groupId); + assertEquals(0, group.size()); + } + @SuppressWarnings("serial") public static class Person implements Serializable{ private Address address; diff --git a/src/reference/asciidoc/whats-new.adoc b/src/reference/asciidoc/whats-new.adoc index 41507fa99d..5f2a2a6e06 100644 --- a/src/reference/asciidoc/whats-new.adoc +++ b/src/reference/asciidoc/whats-new.adoc @@ -168,3 +168,12 @@ metadata store if it implements `Flushable` (e.g. the `PropertiesPersistenMetada When using Java 8, gateway methods can now return `CompletableFuture`. See <> for more information. + +[[x4.2-aggregator-perf]] +==== Aggregator Performance + +This release includes some performance improvements for aggregating components (aggregator, resequencer, etc), +by more efficiently removing messages from groups when they are released. +New methods (`removeMessagesFromGroup`) have been added to the message store. +Set the `removeBatchSize` property (default `100`) to adjust the number of messages deleted in each operation. +Currently, JDBC, Redis and MongoDB message stores support this property.