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

INT-3642: Improve MessageGroupStore Removal

Browse Source 
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
This commit is contained in:
Gary Russell
2015-06-16 09:01:29 -04:00
committed by Artem Bilan
parent 0065ed8c49
commit f3d525a5e8
23 changed files with 472 additions and 107 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -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());
}
}

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

@@ -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<Message<?>> 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));
}

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

@@ -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());

2
spring-integration-core/src/main/java/org/springframework/integration/handler/DelayHandler.java
View File

@@ -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 {

45
spring-integration-core/src/main/java/org/springframework/integration/store/AbstractBatchingMessageGroupStore.java Normal file
View File

@@ -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;
}
}

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

@@ -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 <T> Message<T> addMessage(Message<T> message) {
Assert.notNull(message, "'message' must not be null");
UUID messageId = message.getHeaders().getId();
this.doStore(MESSAGE_KEY_PREFIX + messageId, message);
return (Message<T>) this.getRawMessage(messageId);
doStore(MESSAGE_KEY_PREFIX + messageId, message);
return (Message<T>) 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<Message<?>> 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<UUID> 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<MessageGroup> iterator() {
final Iterator<?> idIterator = this.normalizeKeys(
(Collection<String>) this.doListKeys(MESSAGE_GROUP_KEY_PREFIX + "*"))
final Iterator<?> idIterator = normalizeKeys(
(Collection<String>) 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<UUID> 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

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

@@ -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<MessageGroup>,
BeanFactoryAware {
public abstract class AbstractMessageGroupStore extends AbstractBatchingMessageGroupStore
implements MessageGroupStore, Iterable<MessageGroup>, 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");

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

@@ -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(){

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

@@ -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<Message<?>> 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)}.
*

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

@@ -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<Message<?>> 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<MessageGroup> iterator() {
return new HashSet<MessageGroup>(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;

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

@@ -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.<MessageGroupCallback> 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<String> list = new ArrayList<String>();
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<MessageGroup> 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<Message<?>> 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;
}

26
spring-integration-core/src/test/java/org/springframework/integration/store/SimpleMessageStoreTests.java
View File

@@ -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.<MessageGroupCallback> 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<String> list = new ArrayList<String>();
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<Message<?>> messages = new ArrayList<Message<?>>();
for (int i = 0; i < 25; i++) {
Message<String> 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());
}
}