From 201f0fc2b16c38ca7c7ac09a1f4edc581746a750 Mon Sep 17 00:00:00 2001 From: Ryan Barker Date: Fri, 2 Oct 2015 14:48:29 -0700 Subject: [PATCH] INT-3846: `SimpleMessageStore` Improvements JIRAs: https://jira.spring.io/browse/INT-3830 https://jira.spring.io/browse/INT-3523 https://jira.spring.io/browse/INT-3846 * Fix the OOM condition, when we `release()` `UpperBound` independently of the previous `remove` result (https://jira.spring.io/browse/INT-3846) * Fix "confuse" around `groupCapacity`, when we really didn't care about individual groups (https://jira.spring.io/browse/INT-3523) * Add `upperBoundTimeout` to have a hook to wait some time for the empty slot in the store (https://jira.spring.io/browse/INT-3830) * Fix some JavaDocs warnings * Fix some typos * Fix inconsistency in the `DelayHandler` around `removeMessageFromGroup` when `MS` is `SimpleMessageStore` * Remove `UpperBound.release()` operation from `SimpleMessageStore.removeGroup()`. The waiting process should worry about the new `UpperBound` instance. Some other polishing `tryAcquire` outside of the `lock` Move `tryAcquire` within the `addMessageToGroup` outside of `lock`. But do that only for groups which already exist. For the new groups we have a fresh `UpperBound`, so no need to worry about dead lock and we can obtain a permit immediately. SimpleMessageGroup: BlockingQueue -> LinkedHashSet `SimpleMessageStore`: use "unsynchonized" `SimpleMessageGroup` Make some synchronization fixes according to the migration to the `LinkedHashSet` Avoid extra `Collection` `ResequencingMessageHandler`: compare `size()` of collections instead of `containsAll()` Fix `ConcurrentModificationException` in the `AbstractKeyValueMessageStore` Add `SimpleMessageStore.clearMessageGroup()` Accept polishing and fix `RedisChannelMessageStoreTests` `@Deprecated` `MessageGroupStore.removeMessageFromGroup()` Fix some typos Introduce `SimpleMessageGroupFactory` Extract `MessageGroupFactory` and address PR comments Polishing after rebase JavaDocs and Reference Manual Fix JavaDocs --- .../aggregator/AggregatingMessageHandler.java | 13 +- .../ResequencingMessageHandler.java | 22 +- .../integration/handler/DelayHandler.java | 21 +- .../AbstractBatchingMessageGroupStore.java | 24 +- .../store/AbstractKeyValueMessageStore.java | 58 ++--- .../store/AbstractMessageGroupStore.java | 6 +- .../integration/store/MessageGroup.java | 24 +- .../store/MessageGroupFactory.java | 39 +++ .../integration/store/MessageGroupStore.java | 5 +- .../integration/store/SimpleMessageGroup.java | 55 ++-- .../store/SimpleMessageGroupFactory.java | 90 +++++++ .../integration/store/SimpleMessageStore.java | 237 +++++++++++++----- .../integration/util/UpperBound.java | 11 +- ...bstractCorrelatingMessageHandlerTests.java | 12 +- .../aggregator/ResequencerTests.java | 40 ++- .../store/MessageGroupQueueTests.java | 20 +- .../integration/store/MessageStoreTests.java | 13 +- .../store/SimpleMessageGroupTests.java | 40 ++- .../store/SimpleMessageStoreTests.java | 156 +++++++++++- .../file/FileWritingMessageHandler.java | 2 +- .../gemfire/store/GemfireGroupStoreTests.java | 22 +- .../integration/jdbc/JdbcMessageStore.java | 46 ++-- .../jdbc/store/JdbcChannelMessageStore.java | 26 +- .../jdbc/JdbcMessageStoreTests.java | 8 +- .../mysql/MySqlJdbcMessageStoreTests.java | 38 +-- .../ConfigurableMongoDbMessageStore.java | 6 +- .../store/MongoDbChannelMessageStore.java | 5 +- .../mongodb/store/MongoDbMessageStore.java | 6 +- ...AbstractMongoDbMessageGroupStoreTests.java | 23 +- .../redis/store/RedisChannelMessageStore.java | 31 ++- .../RedisChannelPriorityMessageStore.java | 9 +- .../store/RedisChannelMessageStoreTests.java | 9 +- .../store/RedisMessageGroupStoreTests.java | 22 +- src/reference/asciidoc/message-store.adoc | 10 + src/reference/asciidoc/whats-new.adoc | 8 + 35 files changed, 891 insertions(+), 266 deletions(-) create mode 100644 spring-integration-core/src/main/java/org/springframework/integration/store/MessageGroupFactory.java create mode 100644 spring-integration-core/src/main/java/org/springframework/integration/store/SimpleMessageGroupFactory.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 cd37dcc119..7194b98d29 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-2015 the original author or authors. + * Copyright 2002-2016 the original author or authors. * * Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in compliance with * the License. You may obtain a copy of the License at @@ -17,6 +17,7 @@ import java.util.Collection; import org.springframework.integration.store.MessageGroup; import org.springframework.integration.store.MessageGroupStore; +import org.springframework.integration.store.SimpleMessageStore; import org.springframework.messaging.Message; /** @@ -59,13 +60,19 @@ public class AggregatingMessageHandler extends AbstractCorrelatingMessageHandler @Override protected void afterRelease(MessageGroup messageGroup, Collection> completedMessages) { - this.messageStore.completeGroup(messageGroup.getGroupId()); + Object groupId = messageGroup.getGroupId(); + this.messageStore.completeGroup(groupId); if (this.expireGroupsUponCompletion) { remove(messageGroup); } else { - this.messageStore.removeMessagesFromGroup(messageGroup.getGroupId(), messageGroup.getMessages()); + if (this.messageStore instanceof SimpleMessageStore) { + ((SimpleMessageStore) this.messageStore).clearMessageGroup(groupId); + } + else { + this.messageStore.removeMessagesFromGroup(groupId, messageGroup.getMessages()); + } } } 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 3e205354f3..542ba389fd 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-2015 the original author or authors. + * Copyright 2002-2016 the original author or authors. * * Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in compliance with * the License. You may obtain a copy of the License at @@ -18,6 +18,7 @@ import java.util.Collection; import org.springframework.integration.IntegrationMessageHeaderAccessor; import org.springframework.integration.store.MessageGroup; import org.springframework.integration.store.MessageGroupStore; +import org.springframework.integration.store.SimpleMessageStore; import org.springframework.messaging.Message; /** @@ -26,6 +27,7 @@ import org.springframework.messaging.Message; * * @author Oleg Zhurakousky * @author Gary Russell + * @author Artem Bilan * @since 2.1 */ public class ResequencingMessageHandler extends AbstractCorrelatingMessageHandler { @@ -69,7 +71,6 @@ public class ResequencingMessageHandler extends AbstractCorrelatingMessageHandle @Override protected void afterRelease(MessageGroup messageGroup, Collection> completedMessages, boolean timeout) { - int size = messageGroup.getMessages().size(); int sequenceSize = 0; Message message = messageGroup.getOne(); @@ -81,13 +82,20 @@ public class ResequencingMessageHandler extends AbstractCorrelatingMessageHandle remove(messageGroup); } else { - if (completedMessages != null){ - int lastReleasedSequenceNumber = this.findLastReleasedSequenceNumber(messageGroup.getGroupId(), completedMessages); - messageStore.setLastReleasedSequenceNumberForGroup(messageGroup.getGroupId(), lastReleasedSequenceNumber); - this.messageStore.removeMessagesFromGroup(messageGroup.getGroupId(), completedMessages); + Object groupId = messageGroup.getGroupId(); + if (completedMessages != null) { + int lastReleasedSequenceNumber = findLastReleasedSequenceNumber(groupId, completedMessages); + this.messageStore.setLastReleasedSequenceNumberForGroup(groupId, lastReleasedSequenceNumber); + if (this.messageStore instanceof SimpleMessageStore + && completedMessages.size() == messageGroup.size()) { + ((SimpleMessageStore) this.messageStore).clearMessageGroup(groupId); + } + else { + this.messageStore.removeMessagesFromGroup(groupId, completedMessages); + } } if (timeout) { - this.messageStore.completeGroup(messageGroup.getGroupId()); + this.messageStore.completeGroup(groupId); } } } 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 002a318220..07341a501e 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 @@ -1,5 +1,5 @@ /* - * Copyright 2002-2015 the original author or authors. + * Copyright 2002-2016 the original author or authors. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -17,6 +17,7 @@ package org.springframework.integration.handler; import java.io.Serializable; +import java.util.Collection; import java.util.Date; import java.util.List; import java.util.concurrent.atomic.AtomicBoolean; @@ -36,7 +37,6 @@ import org.springframework.integration.expression.ExpressionUtils; import org.springframework.integration.store.MessageGroup; import org.springframework.integration.store.MessageGroupStore; import org.springframework.integration.store.MessageStore; -import org.springframework.integration.store.SimpleMessageGroup; import org.springframework.integration.store.SimpleMessageStore; import org.springframework.integration.support.management.IntegrationManagedResource; import org.springframework.jmx.export.annotation.ManagedResource; @@ -331,7 +331,9 @@ public class DelayHandler extends AbstractReplyProducingMessageHandler implement private void doReleaseMessage(Message message) { if (removeDelayedMessageFromMessageStore(message)) { - this.messageStore.removeMessagesFromGroup(this.messageGroupId, message); + if (!(this.messageStore instanceof SimpleMessageStore)) { + this.messageStore.removeMessagesFromGroup(this.messageGroupId, message); + } this.handleMessageInternal(message); } else { @@ -344,9 +346,16 @@ public class DelayHandler extends AbstractReplyProducingMessageHandler implement private boolean removeDelayedMessageFromMessageStore(Message message) { if (this.messageStore instanceof SimpleMessageStore) { - SimpleMessageGroup messageGroup = - (SimpleMessageGroup) this.messageStore.getMessageGroup(this.messageGroupId); - return messageGroup.remove(message); + synchronized (this.messageGroupId) { + Collection> messages = this.messageStore.getMessageGroup(this.messageGroupId).getMessages(); + if (messages.contains(message)) { + this.messageStore.removeMessagesFromGroup(this.messageGroupId, message); + return true; + } + else { + return false; + } + } } else { return ((MessageStore) this.messageStore).removeMessage(message.getHeaders().getId()) != null; 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 index 7a2a75d028..66e834a5fa 100644 --- 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 @@ -1,5 +1,5 @@ /* - * Copyright 2015 the original author or authors. + * Copyright 2015-2016 the original author or authors. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -13,11 +13,15 @@ * See the License for the specific language governing permissions and * limitations under the License. */ + package org.springframework.integration.store; +import org.springframework.util.Assert; + /** * @author Gary Russell + * @author Artem Bilan * @since 4.2 * */ @@ -27,6 +31,8 @@ public abstract class AbstractBatchingMessageGroupStore implements BasicMessageG private volatile int removeBatchSize = DEFAULT_REMOVE_BATCH_SIZE; + private volatile MessageGroupFactory messageGroupFactory = new SimpleMessageGroupFactory(); + /** * Set the batch size when bulk removing messages from groups for message stores * that support batch removal. @@ -42,4 +48,20 @@ public abstract class AbstractBatchingMessageGroupStore implements BasicMessageG return removeBatchSize; } + /** + * Specify the {@link MessageGroupFactory} to create {@link MessageGroup} object where + * it is necessary. + * Defaults to {@link SimpleMessageGroupFactory}. + * @param messageGroupFactory the {@link MessageGroupFactory} to use. + * @since 4.3 + */ + public void setMessageGroupFactory(MessageGroupFactory messageGroupFactory) { + Assert.notNull(messageGroupFactory, "'messageGroupFactory' must not be null"); + this.messageGroupFactory = messageGroupFactory; + } + + protected MessageGroupFactory getMessageGroupFactory() { + return this.messageGroupFactory; + } + } 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 d25c40a678..e42f7077f5 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-2015 the original author or authors + * Copyright 2002-2016 the original author or authors * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -35,6 +35,7 @@ import org.springframework.util.Assert; * * @author Oleg Zhurakousky * @author Gary Russell + * @author Artem Bilan * @since 2.1 */ public abstract class AbstractKeyValueMessageStore extends AbstractMessageGroupStore implements MessageStore { @@ -106,7 +107,7 @@ 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 = getSimpleMessageGroup(getMessageGroup(groupId)); + MessageGroup messageGroup = getMessageGroup(groupId); messageGroup.add(message); @@ -116,7 +117,7 @@ public abstract class AbstractKeyValueMessageStore extends AbstractMessageGroupS addMessage(enrichedMessage); // build raw MessageGroup and add enriched Message to it - SimpleMessageGroup rawGroup = buildMessageGroup(groupId, true); + MessageGroup rawGroup = buildMessageGroup(groupId, true); rawGroup.setLastModified(System.currentTimeMillis()); rawGroup.add(enrichedMessage); @@ -131,26 +132,35 @@ public abstract class AbstractKeyValueMessageStore extends AbstractMessageGroupS * Remove a Message from the group with the provided group ID. */ @Override + @Deprecated public MessageGroup removeMessageFromGroup(Object groupId, Message messageToRemove) { Assert.notNull(groupId, "'groupId' must not be null"); Assert.notNull(messageToRemove, "'messageToRemove' must not be null"); // build raw MG - SimpleMessageGroup rawGroup = buildMessageGroup(groupId, true); + MessageGroup rawGroup = buildMessageGroup(groupId, true); // create a clean instance of - SimpleMessageGroup messageGroup = normalizeSimpleMessageGroup(rawGroup); + MessageGroup messageGroup = normalizeSimpleMessageGroup(rawGroup); + + + Message actualMessageToRemove = null; for (Message message : rawGroup.getMessages()) { if (message.getHeaders().getId().equals(messageToRemove.getHeaders().getId())){ - rawGroup.remove(message); + actualMessageToRemove = message; + break; } } - removeMessage(messageToRemove.getHeaders().getId()); - rawGroup.setLastModified(System.currentTimeMillis()); - doStore(MESSAGE_GROUP_KEY_PREFIX + groupId, new MessageGroupMetadata(rawGroup)); - messageGroup = getSimpleMessageGroup(getMessageGroup(groupId)); + if (actualMessageToRemove != null) { + rawGroup.remove(actualMessageToRemove); + removeMessage(messageToRemove.getHeaders().getId()); + rawGroup.setLastModified(System.currentTimeMillis()); + + doStore(MESSAGE_GROUP_KEY_PREFIX + groupId, new MessageGroupMetadata(rawGroup)); + messageGroup = getMessageGroup(groupId); + } return messageGroup; } @@ -178,7 +188,7 @@ public abstract class AbstractKeyValueMessageStore extends AbstractMessageGroupS @Override public void completeGroup(Object groupId) { Assert.notNull(groupId, "'groupId' must not be null"); - SimpleMessageGroup messageGroup = buildMessageGroup(groupId, true); + MessageGroup messageGroup = buildMessageGroup(groupId, true); messageGroup.complete(); messageGroup.setLastModified(System.currentTimeMillis()); doStore(MESSAGE_GROUP_KEY_PREFIX + groupId, new MessageGroupMetadata(messageGroup)); @@ -205,7 +215,7 @@ public abstract class AbstractKeyValueMessageStore extends AbstractMessageGroupS @Override public void setLastReleasedSequenceNumberForGroup(Object groupId, int sequenceNumber) { Assert.notNull(groupId, "'groupId' must not be null"); - SimpleMessageGroup messageGroup = buildMessageGroup(groupId, true); + MessageGroup messageGroup = buildMessageGroup(groupId, true); messageGroup.setLastReleasedMessageSequenceNumber(sequenceNumber); messageGroup.setLastModified(System.currentTimeMillis()); doStore(MESSAGE_GROUP_KEY_PREFIX + groupId, new MessageGroupMetadata(messageGroup)); @@ -298,7 +308,7 @@ public abstract class AbstractKeyValueMessageStore extends AbstractMessageGroupS return enrichedMessage; } - private SimpleMessageGroup buildMessageGroup(Object groupId, boolean raw){ + private MessageGroup buildMessageGroup(Object groupId, boolean raw){ Assert.notNull(groupId, "'groupId' must not be null"); Object mgm = doRetrieve(MESSAGE_GROUP_KEY_PREFIX + groupId); if (mgm != null) { @@ -319,28 +329,20 @@ public abstract class AbstractKeyValueMessageStore extends AbstractMessageGroupS } } - SimpleMessageGroup messageGroup = new SimpleMessageGroup(messages, - groupId, messageGroupMetadata.getTimestamp(), messageGroupMetadata.isComplete()); + MessageGroup messageGroup = getMessageGroupFactory() + .create(messages, groupId, messageGroupMetadata.getTimestamp(), messageGroupMetadata.isComplete()); messageGroup.setLastModified(messageGroupMetadata.getLastModified()); - messageGroup.setLastReleasedMessageSequenceNumber(messageGroupMetadata.getLastReleasedMessageSequenceNumber()); + messageGroup.setLastReleasedMessageSequenceNumber( + messageGroupMetadata.getLastReleasedMessageSequenceNumber()); return messageGroup; } else { - return new SimpleMessageGroup(groupId); + return getMessageGroupFactory().create(groupId); } } - private SimpleMessageGroup getSimpleMessageGroup(MessageGroup messageGroup){ - if (messageGroup instanceof SimpleMessageGroup){ - return (SimpleMessageGroup) messageGroup; - } - else { - return new SimpleMessageGroup(messageGroup); - } - } - - private SimpleMessageGroup normalizeSimpleMessageGroup(SimpleMessageGroup messageGroup){ - SimpleMessageGroup normalizedGroup = new SimpleMessageGroup(messageGroup.getGroupId()); + private MessageGroup normalizeSimpleMessageGroup(MessageGroup messageGroup){ + MessageGroup normalizedGroup = getMessageGroupFactory().create(messageGroup.getGroupId()); for (Message message : messageGroup.getMessages()) { Message normalizedMessage = normalizeMessage(message); normalizedGroup.add(normalizedMessage); 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 01e62ab088..e69cfd4c05 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 @@ -1,5 +1,5 @@ /* - * Copyright 2002-2015 the original author or authors. + * Copyright 2002-2016 the original author or authors. * * Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in compliance with * the License. You may obtain a copy of the License at @@ -103,11 +103,11 @@ public abstract class AbstractMessageGroupStore extends AbstractBatchingMessageG @Override public void registerMessageGroupExpiryCallback(MessageGroupCallback callback) { - expiryCallbacks.add(callback); + this.expiryCallbacks.add(callback); } @Override - public int expireMessageGroups(long timeout) { + public synchronized int expireMessageGroups(long timeout) { int count = 0; long threshold = System.currentTimeMillis() - timeout; for (MessageGroup group : this) { diff --git a/spring-integration-core/src/main/java/org/springframework/integration/store/MessageGroup.java b/spring-integration-core/src/main/java/org/springframework/integration/store/MessageGroup.java index d476cc73ea..27bf7d95bd 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/store/MessageGroup.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/store/MessageGroup.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2014 the original author or authors. + * Copyright 2002-2016 the original author or authors. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -29,6 +29,7 @@ import org.springframework.messaging.Message; * @author Dave Syer * @author Oleg Zhurakousky * @author Gary Russell + * @author Artem Bilan */ public interface MessageGroup { @@ -40,6 +41,21 @@ public interface MessageGroup { */ boolean canAdd(Message message); + /** + * Add the message to this group. + * @param messageToAdd the message to add. + * @since 4.3 + */ + void add(Message messageToAdd); + + /** + * Remove the message from this group. + * @param messageToRemove the message to remove. + * @return {@code true} if a message was removed. + * @since 4.3 + */ + boolean remove(Message messageToRemove); + /** * Returns all available Messages from the group at the time of invocation * @@ -57,6 +73,8 @@ public interface MessageGroup { */ int getLastReleasedMessageSequenceNumber(); + void setLastReleasedMessageSequenceNumber(int sequenceNumber); + /** * @return true if the group is complete (i.e. no more messages are expected to be added) */ @@ -92,4 +110,8 @@ public interface MessageGroup { */ long getLastModified(); + void setLastModified(long lastModified); + + void clear(); + } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/store/MessageGroupFactory.java b/spring-integration-core/src/main/java/org/springframework/integration/store/MessageGroupFactory.java new file mode 100644 index 0000000000..fe1eee05b8 --- /dev/null +++ b/spring-integration-core/src/main/java/org/springframework/integration/store/MessageGroupFactory.java @@ -0,0 +1,39 @@ +/* + * Copyright 2016 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.integration.store; + +import java.util.Collection; + +import org.springframework.messaging.Message; + +/** + * The {@link MessageGroup} factory strategy. + * This strategy is used from the {@link MessageGroup}-aware components, e.g. {@code MessageGroupStore}. + * + * @author Artem Bilan + * @since 4.3 + */ +public interface MessageGroupFactory { + + MessageGroup create(Object groupId); + + MessageGroup create(Collection> messages, Object groupId); + + MessageGroup create(Collection> messages, Object groupId, long timestamp, + boolean complete); + +} 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 66e79d23e3..edc46d1b1c 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-2015 the original author or authors. + * Copyright 2002-2016 the original author or authors. * * Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in compliance with * the License. You may obtain a copy of the License at @@ -10,6 +10,7 @@ * an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the * specific language governing permissions and limitations under the License. */ + package org.springframework.integration.store; import java.util.Collection; @@ -57,7 +58,9 @@ public interface MessageGroupStore extends BasicMessageGroupStore { * @param key The groupId for the group containing the message. * @param messageToRemove The message to be removed. * @return The message Group. + * @deprecated in favor of {@link #removeMessagesFromGroup} */ + @Deprecated MessageGroup removeMessageFromGroup(Object key, Message messageToRemove); /** diff --git a/spring-integration-core/src/main/java/org/springframework/integration/store/SimpleMessageGroup.java b/spring-integration-core/src/main/java/org/springframework/integration/store/SimpleMessageGroup.java index fd1e8527b1..590c36472e 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/store/SimpleMessageGroup.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/store/SimpleMessageGroup.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2014 the original author or authors. + * Copyright 2002-2016 the original author or authors. * * Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in compliance with * the License. You may obtain a copy of the License at @@ -15,45 +15,63 @@ package org.springframework.integration.store; import java.util.Collection; import java.util.Collections; -import java.util.concurrent.BlockingQueue; -import java.util.concurrent.LinkedBlockingQueue; +import java.util.Iterator; +import java.util.LinkedHashSet; import org.springframework.integration.IntegrationMessageHeaderAccessor; import org.springframework.messaging.Message; +import org.springframework.util.Assert; /** - * Represents a mutable group of correlated messages that is bound to a certain {@link MessageStore} and group id. The - * group will grow during its lifetime, when messages are added to it. This MessageGroup is thread safe. + * Represents a mutable group of correlated messages that is bound to a certain {@link MessageStore} and group id. + * The group will grow during its lifetime, when messages are added to it. + * This MessageGroup is thread safe. * * @author Iwein Fuld * @author Oleg Zhurakousky * @author Dave Syer * @author Gary Russell + * @author Artem Bilan + * * @since 2.0 */ public class SimpleMessageGroup implements MessageGroup { private final Object groupId; - public final BlockingQueue> messages = new LinkedBlockingQueue>(); - - private volatile int lastReleasedMessageSequence; + private final Collection> messages; private final long timestamp; + private volatile int lastReleasedMessageSequence; + private volatile long lastModified; private volatile boolean complete; public SimpleMessageGroup(Object groupId) { - this(Collections.> emptyList(), groupId, System.currentTimeMillis(), false); + this(Collections.> emptyList(), groupId); } public SimpleMessageGroup(Collection> messages, Object groupId) { this(messages, groupId, System.currentTimeMillis(), false); } - public SimpleMessageGroup(Collection> messages, Object groupId, long timestamp, boolean complete) { + public SimpleMessageGroup(MessageGroup messageGroup) { + this(messageGroup.getMessages(), messageGroup.getGroupId(), messageGroup.getTimestamp(), + messageGroup.isComplete()); + } + + public SimpleMessageGroup(Collection> messages, Object groupId, long timestamp, + boolean complete) { + this(new LinkedHashSet>(), messages, groupId, timestamp, complete); + } + + SimpleMessageGroup(Collection> internalStore, Collection> messages, Object groupId, + long timestamp, boolean complete) { + Assert.notNull(internalStore, "'internalStore' must not be null"); + Assert.notNull(messages, "'messages' must not be null"); + this.messages = internalStore; this.groupId = groupId; this.timestamp = timestamp; this.complete = complete; @@ -64,10 +82,6 @@ public class SimpleMessageGroup implements MessageGroup { } } - public SimpleMessageGroup(MessageGroup messageGroup) { - this(messageGroup.getMessages(), messageGroup.getGroupId(), messageGroup.getTimestamp(), messageGroup.isComplete()); - } - @Override public long getTimestamp() { return timestamp; @@ -87,8 +101,8 @@ public class SimpleMessageGroup implements MessageGroup { return true; } - public void add(Message message) { - addMessage(message); + public void add(Message messageToAdd) { + addMessage(messageToAdd); } public boolean remove(Message message) { @@ -101,7 +115,7 @@ public class SimpleMessageGroup implements MessageGroup { } private boolean addMessage(Message message) { - return this.messages.offer(message); + return this.messages.add(message); } @Override @@ -143,8 +157,10 @@ public class SimpleMessageGroup implements MessageGroup { @Override public Message getOne() { - Message one = messages.peek(); - return one; + synchronized (this.messages) { + Iterator> iterator = this.messages.iterator(); + return iterator.hasNext() ? iterator.next() : null; + } } public void clear(){ @@ -160,4 +176,5 @@ public class SimpleMessageGroup implements MessageGroup { ", lastModified=" + lastModified + '}'; } + } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/store/SimpleMessageGroupFactory.java b/spring-integration-core/src/main/java/org/springframework/integration/store/SimpleMessageGroupFactory.java new file mode 100644 index 0000000000..4bb084c8fe --- /dev/null +++ b/spring-integration-core/src/main/java/org/springframework/integration/store/SimpleMessageGroupFactory.java @@ -0,0 +1,90 @@ +/* + * Copyright 2016 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on + * an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the + * specific language governing permissions and limitations under the License. + */ + +package org.springframework.integration.store; + +import java.util.Collection; +import java.util.Collections; +import java.util.LinkedHashSet; +import java.util.concurrent.LinkedBlockingQueue; + +import org.springframework.messaging.Message; + +/** + * The {@link MessageGroupFactory} implementation to produce {@link SimpleMessageGroup} instances. + * The {@link GroupType} modificator specifies the internal collection for the {@link SimpleMessageGroup}. + * The {@link GroupType#HASH_SET} is the default type. + * + * @author Artem Bilan + * @since 4.3 + */ +public class SimpleMessageGroupFactory implements MessageGroupFactory { + + private final GroupType type; + + public SimpleMessageGroupFactory() { + this(GroupType.HASH_SET); + } + + public SimpleMessageGroupFactory(GroupType type) { + this.type = type; + } + + @Override + public MessageGroup create(Object groupId) { + return create(Collections.> emptyList(), groupId); + } + + @Override + public MessageGroup create(Collection> messages, Object groupId) { + return create(messages, groupId, System.currentTimeMillis(), false); + } + + @Override + public MessageGroup create(Collection> messages, Object groupId, long timestamp, + boolean complete) { + return new SimpleMessageGroup(this.type.get(), messages, groupId, timestamp, complete); + } + + public enum GroupType { + + BLOCKING_QUEUE { + + @Override + Collection> get() { + return new LinkedBlockingQueue>(); + } + + }, + HASH_SET { + + @Override + Collection> get() { + return new LinkedHashSet>(); + } + + }, + SYNCHRONISED_SET { + + @Override + Collection> get() { + return Collections.>synchronizedSet(new LinkedHashSet>()); + } + + }; + + abstract Collection> get(); + + } + +} diff --git a/spring-integration-core/src/main/java/org/springframework/integration/store/SimpleMessageStore.java b/spring-integration-core/src/main/java/org/springframework/integration/store/SimpleMessageStore.java index 5278706d26..966916d74d 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-2015 the original author or authors. + * Copyright 2002-2016 the original author or authors. * * Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in compliance with * the License. You may obtain a copy of the License at @@ -31,69 +31,104 @@ import org.springframework.util.Assert; import org.springframework.util.CollectionUtils; /** - * Map-based in-memory implementation of {@link MessageStore} and {@link MessageGroupStore}. Enforces a maximum capacity for the - * store. + * Map-based in-memory implementation of {@link MessageStore} and {@link MessageGroupStore}. + * Enforces a maximum capacity for the store. * * @author Iwein Fuld * @author Mark Fisher * @author Dave Syer * @author Oleg Zhurakousky * @author Gary Russell + * @author Ryan Barker + * @author Artem Bilan * * @since 2.0 */ public class SimpleMessageStore extends AbstractMessageGroupStore implements MessageStore, ChannelMessageStore { - private volatile LockRegistry lockRegistry; + private final ConcurrentMap> idToMessage = new ConcurrentHashMap>(); - private final ConcurrentMap> idToMessage; + private final ConcurrentMap groupIdToMessageGroup = + new ConcurrentHashMap(); - private final ConcurrentMap groupIdToMessageGroup; + private final ConcurrentMap groupToUpperBound = new ConcurrentHashMap(); + + private final int groupCapacity; + + private final int individualCapacity; private final UpperBound individualUpperBound; - private final UpperBound groupUpperBound; + private volatile LockRegistry lockRegistry; private volatile boolean isUsed; private volatile boolean copyOnGet = false; + private final long upperBoundTimeout; + /** * Creates a SimpleMessageStore with a maximum size limited by the given capacity, or unlimited size if the given * capacity is less than 1. The capacities are applied independently to messages stored via * {@link #addMessage(Message)} and to those stored via {@link #addMessageToGroup(Object, Message)}. In both cases * the capacity applies to the number of messages that can be stored, and once that limit is reached attempting to * store another will result in an exception. - * * @param individualCapacity The message capacity. - * @param groupCapacity The capacity of each group. + * @param groupCapacity The capacity of each group. */ public SimpleMessageStore(int individualCapacity, int groupCapacity) { this(individualCapacity, groupCapacity, new DefaultLockRegistry()); } /** - * See {@link #SimpleMessageStore(int, int)}. - * Also allows the provision of a custom {@link LockRegistry} - * rather than using the default. - * + * Creates a SimpleMessageStore with a maximum size limited by the given capacity and the timeout in millisecond + * to wait for the empty slot in the store. * @param individualCapacity The message capacity. - * @param groupCapacity The capacity of each group. - * @param lockRegistry The lock registry. + * @param groupCapacity The capacity of each group. + * @param upperBoundTimeout The time to wait if the store is at max capacity. + * @see #SimpleMessageStore(int, int) + * @since 4.3 + */ + public SimpleMessageStore(int individualCapacity, int groupCapacity, long upperBoundTimeout) { + this(individualCapacity, groupCapacity, upperBoundTimeout, new DefaultLockRegistry()); + } + + /** + * Creates a SimpleMessageStore with a maximum size limited by the given capacity and LockRegistry + * for the message group operations concurrency. + * @param individualCapacity The message capacity. + * @param groupCapacity The capacity of each group. + * @param lockRegistry The lock registry. + * @see #SimpleMessageStore(int, int, long, LockRegistry) */ public SimpleMessageStore(int individualCapacity, int groupCapacity, LockRegistry lockRegistry) { + this(individualCapacity, groupCapacity, 0, lockRegistry); + } + + + /** + * Creates a SimpleMessageStore with a maximum size limited by the given capacity, + * the timeout in millisecond to wait for the empty slot in the store and LockRegistry + * for the message group operations concurrency. + * @param individualCapacity The message capacity. + * @param groupCapacity The capacity of each group. + * @param upperBoundTimeout The time to wait if the store is at max capacity + * @param lockRegistry The lock registry. + * @since 4.3 + */ + public SimpleMessageStore(int individualCapacity, int groupCapacity, long upperBoundTimeout, + LockRegistry lockRegistry) { Assert.notNull(lockRegistry, "The LockRegistry cannot be null"); - this.idToMessage = new ConcurrentHashMap>(); - this.groupIdToMessageGroup = new ConcurrentHashMap(); this.individualUpperBound = new UpperBound(individualCapacity); - this.groupUpperBound = new UpperBound(groupCapacity); + this.individualCapacity = individualCapacity; + this.groupCapacity = groupCapacity; this.lockRegistry = lockRegistry; + this.upperBoundTimeout = upperBoundTimeout; } /** * Creates a SimpleMessageStore with the same capacity for individual and grouped messages. - * * @param capacity The capacity. */ public SimpleMessageStore(int capacity) { @@ -132,9 +167,11 @@ public class SimpleMessageStore extends AbstractMessageGroupStore @Override public Message addMessage(Message message) { this.isUsed = true; - if (!individualUpperBound.tryAcquire(0)) { + if (!individualUpperBound.tryAcquire(this.upperBoundTimeout)) { throw new MessagingException(this.getClass().getSimpleName() - + " was out of capacity at, try constructing it with a larger capacity."); + + " was out of capacity (" + + this.individualCapacity + + "), try constructing it with a larger capacity."); } this.idToMessage.put(message.getHeaders().getId(), message); return message; @@ -148,8 +185,11 @@ public class SimpleMessageStore extends AbstractMessageGroupStore @Override public Message removeMessage(UUID key) { if (key != null) { - individualUpperBound.release(); - return this.idToMessage.remove(key); + Message message = this.idToMessage.remove(key); + if (message != null) { + this.individualUpperBound.release(); + } + return message; } else { return null; @@ -160,9 +200,9 @@ public class SimpleMessageStore extends AbstractMessageGroupStore public MessageGroup getMessageGroup(Object groupId) { Assert.notNull(groupId, "'groupId' must not be null"); - SimpleMessageGroup group = groupIdToMessageGroup.get(groupId); + MessageGroup group = groupIdToMessageGroup.get(groupId); if (group == null) { - return new SimpleMessageGroup(groupId); + return getMessageGroupFactory().create(groupId); } if (this.copyOnGet) { return copy(group); @@ -174,29 +214,16 @@ public class SimpleMessageStore extends AbstractMessageGroupStore @Override protected MessageGroup copy(MessageGroup group) { - SimpleMessageGroup simpleMessageGroup = new SimpleMessageGroup(group); - simpleMessageGroup.setLastModified(group.getLastModified()); - return simpleMessageGroup; - } - - @Override - public MessageGroup addMessageToGroup(Object groupId, Message message) { - if (!groupUpperBound.tryAcquire(0)) { - throw new MessagingException(this.getClass().getSimpleName() - + " was out of capacity at, try constructing it with a larger capacity."); - } + Object groupId = group.getGroupId(); Lock lock = this.lockRegistry.obtain(groupId); try { lock.lockInterruptibly(); try { - SimpleMessageGroup group = this.groupIdToMessageGroup.get(groupId); - if (group == null) { - group = new SimpleMessageGroup(groupId); - this.groupIdToMessageGroup.putIfAbsent(groupId, group); - } - group.add(message); - this.groupIdToMessageGroup.get(groupId).setLastModified(System.currentTimeMillis()); - return group; + MessageGroup simpleMessageGroup = getMessageGroupFactory() + .create(group.getMessages(), groupId, group.getTimestamp(), group.isComplete()); + simpleMessageGroup.setLastModified(group.getLastModified()); + simpleMessageGroup.setLastReleasedMessageSequenceNumber(group.getLastReleasedMessageSequenceNumber()); + return simpleMessageGroup; } finally { lock.unlock(); @@ -208,18 +235,65 @@ public class SimpleMessageStore extends AbstractMessageGroupStore } } + @Override + public MessageGroup addMessageToGroup(Object groupId, Message message) { + Lock lock = this.lockRegistry.obtain(groupId); + try { + lock.lockInterruptibly(); + boolean unlocked = false; + try { + UpperBound upperBound; + MessageGroup group = this.groupIdToMessageGroup.get(groupId); + if (group == null) { + group = getMessageGroupFactory().create(groupId); + this.groupIdToMessageGroup.putIfAbsent(groupId, group); + upperBound = new UpperBound(this.groupCapacity); + upperBound.tryAcquire(-1); + this.groupToUpperBound.putIfAbsent(groupId, upperBound); + } + else { + upperBound = this.groupToUpperBound.get(groupId); + Assert.state(upperBound != null, "'upperBound' must not be null."); + lock.unlock(); + if (!upperBound.tryAcquire(this.upperBoundTimeout)) { + unlocked = true; + throw new MessagingException(this.getClass().getSimpleName() + + " was out of capacity (" + + this.groupCapacity + + ") for group '" + + groupId + + "', try constructing it with a larger capacity."); + } + lock.lockInterruptibly(); + } + group.add(message); + group.setLastModified(System.currentTimeMillis()); + return group; + } + finally { + if (!unlocked) { + lock.unlock(); + } + } + } + catch (InterruptedException e) { + Thread.currentThread().interrupt(); + throw new MessagingException("Interrupted while obtaining lock", e); + } + } + @Override public void removeMessageGroup(Object groupId) { Lock lock = this.lockRegistry.obtain(groupId); try { lock.lockInterruptibly(); try { - if (!groupIdToMessageGroup.containsKey(groupId)) { - return; + MessageGroup messageGroup = this.groupIdToMessageGroup.remove(groupId); + if (messageGroup != null) { + UpperBound upperBound = this.groupToUpperBound.remove(groupId); + Assert.state(upperBound != null, "'upperBound' must not be null."); + upperBound.release(this.groupCapacity); } - - groupUpperBound.release(groupIdToMessageGroup.get(groupId).size()); - groupIdToMessageGroup.remove(groupId); } finally { lock.unlock(); @@ -232,16 +306,21 @@ public class SimpleMessageStore extends AbstractMessageGroupStore } @Override + @Deprecated public MessageGroup removeMessageFromGroup(Object groupId, Message messageToRemove) { Lock lock = this.lockRegistry.obtain(groupId); try { lock.lockInterruptibly(); try { - SimpleMessageGroup group = this.groupIdToMessageGroup.get(groupId); + MessageGroup group = this.groupIdToMessageGroup.get(groupId); Assert.notNull(group, "MessageGroup for groupId '" + groupId + "' " + "can not be located while attempting to remove Message from the MessageGroup"); - group.remove(messageToRemove); - group.setLastModified(System.currentTimeMillis()); + if (group.remove(messageToRemove)) { + UpperBound upperBound = this.groupToUpperBound.get(groupId); + Assert.state(upperBound != null, "'upperBound' must not be null."); + upperBound.release(); + group.setLastModified(System.currentTimeMillis()); + } return group; } finally { @@ -260,13 +339,21 @@ public class SimpleMessageStore extends AbstractMessageGroupStore try { lock.lockInterruptibly(); try { - SimpleMessageGroup group = this.groupIdToMessageGroup.get(groupId); + MessageGroup 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"); + UpperBound upperBound = this.groupToUpperBound.get(groupId); + Assert.state(upperBound != null, "'upperBound' must not be null."); + boolean modified = false; for (Message messageToRemove : messages) { - group.remove(messageToRemove); + if (group.remove(messageToRemove)) { + upperBound.release(); + modified = true; + } + } + if (modified) { + group.setLastModified(System.currentTimeMillis()); } - group.setLastModified(System.currentTimeMillis()); } finally { lock.unlock(); @@ -289,7 +376,7 @@ public class SimpleMessageStore extends AbstractMessageGroupStore try { lock.lockInterruptibly(); try { - SimpleMessageGroup group = this.groupIdToMessageGroup.get(groupId); + MessageGroup group = this.groupIdToMessageGroup.get(groupId); Assert.notNull(group, "MessageGroup for groupId '" + groupId + "' " + "can not be located while attempting to set 'lastReleasedSequenceNumber'"); group.setLastReleasedMessageSequenceNumber(sequenceNumber); @@ -311,7 +398,7 @@ public class SimpleMessageStore extends AbstractMessageGroupStore try { lock.lockInterruptibly(); try { - SimpleMessageGroup group = this.groupIdToMessageGroup.get(groupId); + MessageGroup group = this.groupIdToMessageGroup.get(groupId); Assert.notNull(group, "MessageGroup for groupId '" + groupId + "' " + "can not be located while attempting to complete the MessageGroup"); group.complete(); @@ -329,11 +416,11 @@ public class SimpleMessageStore extends AbstractMessageGroupStore @Override public Message pollMessageFromGroup(Object groupId) { - Collection> messageList = this.getMessageGroup(groupId).getMessages(); + Collection> messageList = getMessageGroup(groupId).getMessages(); Message message = null; - if (!CollectionUtils.isEmpty(messageList)){ + if (!CollectionUtils.isEmpty(messageList)) { message = messageList.iterator().next(); - if (message != null){ + if (message != null) { this.removeMessagesFromGroup(groupId, message); } } @@ -342,17 +429,41 @@ public class SimpleMessageStore extends AbstractMessageGroupStore @Override public int messageGroupSize(Object groupId) { - return this.getMessageGroup(groupId).size(); + return getMessageGroup(groupId).size(); } @Override public MessageGroupMetadata getGroupMetadata(Object groupId) { - return new MessageGroupMetadata(this.getMessageGroup(groupId)); + return new MessageGroupMetadata(getMessageGroup(groupId)); } @Override public Message getOneMessageFromGroup(Object groupId) { - return this.getMessageGroup(groupId).getOne(); + return getMessageGroup(groupId).getOne(); + } + + public void clearMessageGroup(Object groupId) { + Lock lock = this.lockRegistry.obtain(groupId); + try { + lock.lockInterruptibly(); + try { + MessageGroup group = this.groupIdToMessageGroup.get(groupId); + Assert.notNull(group, "MessageGroup for groupId '" + groupId + "' " + + "can not be located while attempting to complete the MessageGroup"); + group.clear(); + group.setLastModified(System.currentTimeMillis()); + UpperBound upperBound = this.groupToUpperBound.get(groupId); + Assert.state(upperBound != null, "'upperBound' must not be null."); + upperBound.release(this.groupCapacity); + } + finally { + lock.unlock(); + } + } + catch (InterruptedException e) { + Thread.currentThread().interrupt(); + throw new MessagingException("Interrupted while obtaining lock", e); + } } } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/util/UpperBound.java b/spring-integration-core/src/main/java/org/springframework/integration/util/UpperBound.java index 8812c2ecf7..baabffa390 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/util/UpperBound.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/util/UpperBound.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2014 the original author or authors. + * Copyright 2002-2016 the original author or authors. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -25,6 +25,8 @@ import java.util.concurrent.TimeUnit; * * @author Mark Fisher * @author Iwein Fuld + * @author Artem Bilan + * * @since 2.0 */ public final class UpperBound { @@ -56,7 +58,7 @@ public final class UpperBound { * indefinitely. * * @param timeoutInMilliseconds The time to wait until a permit is available. - * @return true if a permit is aquired. + * @return true if a permit is acquired. */ public boolean tryAcquire(long timeoutInMilliseconds) { if (this.semaphore != null) { @@ -97,4 +99,9 @@ public final class UpperBound { } } + @Override + public String toString() { + return super.toString() + "[Permits = " + + (this.semaphore != null ? this.semaphore.availablePermits() : "UNLIMITED") + "]"; + } } diff --git a/spring-integration-core/src/test/java/org/springframework/integration/aggregator/AbstractCorrelatingMessageHandlerTests.java b/spring-integration-core/src/test/java/org/springframework/integration/aggregator/AbstractCorrelatingMessageHandlerTests.java index 991939c985..99dd58aa54 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/aggregator/AbstractCorrelatingMessageHandlerTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/aggregator/AbstractCorrelatingMessageHandlerTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2014 the original author or authors. + * Copyright 2002-2016 the original author or authors. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -54,7 +54,7 @@ public class AbstractCorrelatingMessageHandlerTests { @Test // INT-2751 public void testReaperDoesntReapAProcessingGroup() throws Exception { final MessageGroupStore groupStore = new SimpleMessageStore(); - final CountDownLatch waitForSendlatch = new CountDownLatch(1); + final CountDownLatch waitForSendLatch = new CountDownLatch(1); final CountDownLatch waitReapStartLatch = new CountDownLatch(1); final CountDownLatch waitReapCompleteLatch = new CountDownLatch(1); AbstractCorrelatingMessageHandler handler = new AbstractCorrelatingMessageHandler( @@ -85,7 +85,7 @@ public class AbstractCorrelatingMessageHandlerTests { catch (InterruptedException e) { Thread.currentThread().interrupt(); } - waitForSendlatch.countDown(); + waitForSendLatch.countDown(); try { Thread.sleep(100); } @@ -95,6 +95,7 @@ public class AbstractCorrelatingMessageHandlerTests { groupStore.expireMessageGroups(50); waitReapCompleteLatch.countDown(); } + }); final List> outputMessages = new ArrayList>(); @@ -109,7 +110,7 @@ public class AbstractCorrelatingMessageHandlerTests { // wake reaper waitReapStartLatch.countDown(); try { - waitForSendlatch.await(10, TimeUnit.SECONDS); + waitForSendLatch.await(10, TimeUnit.SECONDS); // wait a little longer for reaper to grab groups Thread.sleep(2000); // simulate tx commit @@ -132,6 +133,7 @@ public class AbstractCorrelatingMessageHandlerTests { public boolean canRelease(MessageGroup group) { return group.size() == 2; } + }); QueueChannel discards = new QueueChannel(); @@ -290,7 +292,7 @@ public class AbstractCorrelatingMessageHandlerTests { mgs.addMessageToGroup("foo", secondMessage); MessageGroup group = mgs.getMessageGroup("foo"); // remove a message - mgs.removeMessageFromGroup("foo", secondMessage); + mgs.removeMessagesFromGroup("foo", secondMessage); // force lastModified to be the same MessageGroup groupNow = mgs.getMessageGroup("foo"); new DirectFieldAccessor(group).setPropertyValue("lastModified", groupNow.getLastModified()); diff --git a/spring-integration-core/src/test/java/org/springframework/integration/aggregator/ResequencerTests.java b/spring-integration-core/src/test/java/org/springframework/integration/aggregator/ResequencerTests.java index b462a6c0a7..638c25b92b 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/aggregator/ResequencerTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/aggregator/ResequencerTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2015 the original author or authors. + * Copyright 2002-2016 the original author or authors. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -16,17 +16,22 @@ package org.springframework.integration.aggregator; +import static org.hamcrest.Matchers.containsString; import static org.hamcrest.Matchers.is; import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertNotNull; import static org.junit.Assert.assertNull; import static org.junit.Assert.assertThat; +import static org.junit.Assert.fail; import static org.mockito.Mockito.mock; import java.util.ArrayList; import java.util.Arrays; import java.util.Collections; +import org.junit.Before; +import org.junit.Test; + import org.springframework.beans.factory.BeanFactory; import org.springframework.expression.spel.standard.SpelExpressionParser; import org.springframework.integration.IntegrationMessageHeaderAccessor; @@ -36,11 +41,9 @@ import org.springframework.integration.store.SimpleMessageStore; import org.springframework.integration.support.MessageBuilder; import org.springframework.messaging.Message; import org.springframework.messaging.MessageChannel; +import org.springframework.messaging.MessagingException; import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler; -import org.junit.Before; -import org.junit.Test; - /** * @author Marius Bogoevici * @author Alex Peters @@ -48,6 +51,7 @@ import org.junit.Test; * @author Iwein Fuld * @author Oleg Zhurakousky * @author Gary Russell + * @author Artem Bilan */ public class ResequencerTests { @@ -166,10 +170,12 @@ public class ResequencerTests { @Test public void testResequencingWithIncompleteSequenceRelease() throws InterruptedException { this.resequencer.setReleaseStrategy(new SequenceSizeReleaseStrategy(true)); + // INT-3846 + this.resequencer.setMessageStore(new SimpleMessageStore(3)); QueueChannel replyChannel = new QueueChannel(); - Message message1 = createMessage("123", "ABC", 4, 2, replyChannel); - Message message2 = createMessage("456", "ABC", 4, 1, replyChannel); - Message message3 = createMessage("789", "ABC", 4, 4, replyChannel); + Message message1 = createMessage("123", "ABC", 4, 4, replyChannel); + Message message2 = createMessage("456", "ABC", 4, 2, replyChannel); + Message message3 = createMessage("789", "ABC", 4, 1, replyChannel); // release 2 after this one Message message4 = createMessage("XYZ", "ABC", 4, 3, replyChannel); this.resequencer.handleMessage(message1); this.resequencer.handleMessage(message2); @@ -193,6 +199,26 @@ public class ResequencerTests { assertEquals(new Integer(4), new IntegrationMessageHeaderAccessor(reply4).getSequenceNumber()); } + @Test + public void testResequencingWithCapacity() throws InterruptedException { + this.resequencer.setReleaseStrategy(new SequenceSizeReleaseStrategy(true)); + // INT-3846 + this.resequencer.setMessageStore(new SimpleMessageStore(3, 2)); + QueueChannel replyChannel = new QueueChannel(); + Message message1 = createMessage("123", "ABC", 4, 4, replyChannel); + Message message2 = createMessage("456", "ABC", 4, 2, replyChannel); + Message message3 = createMessage("789", "ABC", 4, 1, replyChannel); + this.resequencer.handleMessage(message1); + this.resequencer.handleMessage(message2); + try { + this.resequencer.handleMessage(message3); + fail("Expected exception"); + } + catch (MessagingException e) { + assertThat(e.getMessage(), containsString("out of capacity (2) for group 'ABC'")); + } + } + @Test public void testResequencingWithPartialSequenceAndComparator() throws InterruptedException { this.resequencer.setReleaseStrategy(new SequenceSizeReleaseStrategy(true)); diff --git a/spring-integration-core/src/test/java/org/springframework/integration/store/MessageGroupQueueTests.java b/spring-integration-core/src/test/java/org/springframework/integration/store/MessageGroupQueueTests.java index 4eff756553..11f2b1a322 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/store/MessageGroupQueueTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/store/MessageGroupQueueTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2011 the original author or authors. + * Copyright 2002-2016 the original author or authors. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -16,6 +16,11 @@ package org.springframework.integration.store; +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertFalse; +import static org.junit.Assert.assertNotNull; +import static org.junit.Assert.assertTrue; + import java.util.HashSet; import java.util.Set; import java.util.concurrent.Callable; @@ -27,16 +32,11 @@ import java.util.concurrent.TimeUnit; import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; -import org.junit.Ignore; import org.junit.Test; + import org.springframework.messaging.Message; import org.springframework.messaging.support.GenericMessage; -import static org.junit.Assert.assertEquals; -import static org.junit.Assert.assertFalse; -import static org.junit.Assert.assertNotNull; -import static org.junit.Assert.assertTrue; - /** * @author Dave Syer * @since 2.0 @@ -92,17 +92,15 @@ public class MessageGroupQueueTests { } @Test - @Ignore public void testConcurrentAccess() throws Exception { doTestConcurrentAccess(50, 20, new HashSet()); } @Test - @Ignore public void testConcurrentAccessUniqueResults() throws Exception { doTestConcurrentAccess(50, 20, null); } - + private void doTestConcurrentAccess(int concurrency, final int maxPerTask, final Set set) throws Exception { SimpleMessageStore messageGroupStore = new SimpleMessageStore(); @@ -163,7 +161,7 @@ public class MessageGroupQueueTests { assertEquals(0, queue.size()); messageGroupStore.expireMessageGroups(-10000); assertEquals(Integer.MAX_VALUE, queue.remainingCapacity()); - + executorService.shutdown(); } 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 d476112de6..b0caf0600a 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-2015 the original author or authors. + * Copyright 2002-2016 the original author or authors. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -19,8 +19,8 @@ package org.springframework.integration.store; import static org.junit.Assert.assertEquals; import java.util.ArrayList; -import java.util.Arrays; import java.util.Collection; +import java.util.Collections; import java.util.Iterator; import java.util.List; @@ -40,7 +40,8 @@ public class MessageStoreTests { @Test public void shouldRegisterCallbacks() throws Exception { TestMessageStore store = new TestMessageStore(); - store.setExpiryCallbacks(Arrays. asList(new MessageGroupStore.MessageGroupCallback() { + store.setExpiryCallbacks(Collections.singletonList(new MessageGroupCallback() { + @Override public void execute(MessageGroupStore messageGroupStore, MessageGroup group) { } @@ -82,14 +83,15 @@ public class MessageStoreTests { private static class TestMessageStore extends AbstractMessageGroupStore { @SuppressWarnings("unchecked") - MessageGroup testMessages = new SimpleMessageGroup(Arrays.asList(new GenericMessage("foo")), "bar"); + MessageGroup testMessages = + new SimpleMessageGroup(Collections.singletonList(new GenericMessage("foo")), "bar"); private boolean removed = false; @Override public Iterator iterator() { - return Arrays.asList(testMessages).iterator(); + return Collections.singletonList(testMessages).iterator(); } @Override @@ -103,6 +105,7 @@ public class MessageStoreTests { } @Override + @Deprecated public MessageGroup removeMessageFromGroup(Object key, Message messageToRemove) { throw new UnsupportedOperationException(); } diff --git a/spring-integration-core/src/test/java/org/springframework/integration/store/SimpleMessageGroupTests.java b/spring-integration-core/src/test/java/org/springframework/integration/store/SimpleMessageGroupTests.java index 69d7a5688c..6f61d1c0b7 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/store/SimpleMessageGroupTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/store/SimpleMessageGroupTests.java @@ -1,24 +1,42 @@ +/* + * Copyright 2009-2016 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on + * an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the + * specific language governing permissions and limitations under the License. + */ + package org.springframework.integration.store; import static org.hamcrest.CoreMatchers.is; import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertThat; +import static org.junit.Assert.assertTrue; import static org.mockito.Mockito.mock; import java.lang.reflect.Constructor; import java.util.ArrayList; +import java.util.Collection; import java.util.Collections; import java.util.List; import org.junit.Test; -import org.springframework.messaging.Message; import org.springframework.integration.support.MessageBuilder; +import org.springframework.messaging.Message; +import org.springframework.messaging.support.GenericMessage; +import org.springframework.util.StopWatch; /** * @author Iwein Fuld * @author Oleg Zhurakousky * @author Dave Syer + * @author Artem Bilan */ public class SimpleMessageGroupTests { @@ -57,7 +75,7 @@ public class SimpleMessageGroupTests { assertThat(group.canAdd(message1), is(true)); } - @Test // shoudl not fail with NPE (see INT-2666) + @Test // should not fail with NPE (see INT-2666) public void shouldIgnoreNullValuesWhenInitializedWithCollectionContainingNulls() throws Exception{ Message m1 = mock(Message.class); Message m2 = mock(Message.class); @@ -68,4 +86,22 @@ public class SimpleMessageGroupTests { SimpleMessageGroup grp = new SimpleMessageGroup(messages, 1); assertEquals(2, grp.getMessages().size()); } + + @Test + // This test used to take 2 min and half to run; now ~200 milliseconds. + public void testPerformance_INT3846() { + Collection> messages = new ArrayList<>(); + for (int i = 0; i < 100000; i++) { + messages.add(new GenericMessage("foo")); + } + SimpleMessageGroup group = new SimpleMessageGroup(messages, this.key); + StopWatch watch = new StopWatch(); + watch.start(); + for (Message message : messages) { + group.getMessages().contains(message); + } + watch.stop(); + assertTrue(watch.getTotalTimeMillis() < 5000); + } + } 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 15eaa3c601..2bf4c3832a 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-2015 the original author or authors. + * Copyright 2002-2016 the original author or authors. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -16,15 +16,22 @@ package org.springframework.integration.store; +import static org.hamcrest.CoreMatchers.containsString; +import static org.hamcrest.CoreMatchers.instanceOf; import static org.hamcrest.CoreMatchers.is; import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertNotSame; import static org.junit.Assert.assertThat; +import static org.junit.Assert.assertTrue; +import static org.junit.Assert.fail; import java.util.ArrayList; import java.util.Arrays; import java.util.Collection; import java.util.List; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.Executors; +import java.util.concurrent.TimeUnit; import org.junit.Test; @@ -32,12 +39,15 @@ import org.springframework.integration.store.MessageGroupStore.MessageGroupCallb import org.springframework.integration.support.MessageBuilder; import org.springframework.messaging.Message; import org.springframework.messaging.MessagingException; +import org.springframework.messaging.support.GenericMessage; import org.springframework.test.util.ReflectionTestUtils; /** * @author Iwein Fuld * @author Dave Syer * @author Gary Russell + * @author Ryan Barker + * @author Artem Bilan */ public class SimpleMessageStoreTests { @@ -47,7 +57,7 @@ public class SimpleMessageStoreTests { SimpleMessageStore store = new SimpleMessageStore(); Message testMessage1 = MessageBuilder.withPayload("foo").build(); store.addMessage(testMessage1); - assertThat((Message) store.getMessage(testMessage1.getHeaders().getId()), is(testMessage1)); + assertThat(store.getMessage(testMessage1.getHeaders().getId()), is(testMessage1)); } @Test(expected = MessagingException.class) @@ -59,6 +69,73 @@ public class SimpleMessageStoreTests { store.addMessage(testMessage2); } + @Test + public void shouldReleaseCapacity() { + SimpleMessageStore store = new SimpleMessageStore(1); + Message testMessage1 = MessageBuilder.withPayload("foo").build(); + Message testMessage2 = MessageBuilder.withPayload("bar").build(); + store.addMessage(testMessage1); + try { + store.addMessage(testMessage2); + fail("Should have thrown"); + } + catch (Exception e) { + assertThat(e, instanceOf(MessagingException.class)); + assertThat(e.getMessage(), containsString("was out of capacity (1)")); + } + store.removeMessage(testMessage2.getHeaders().getId()); + try { + store.addMessage(testMessage2); + fail("Should have thrown"); + } + catch (Exception e) { + assertThat(e, instanceOf(MessagingException.class)); + assertThat(e.getMessage(), containsString("was out of capacity (1)")); + } + store.removeMessage(testMessage1.getHeaders().getId()); + store.addMessage(testMessage2); + + } + + @Test + public void shouldWaitIfCapacity() throws InterruptedException { + final SimpleMessageStore store2 = new SimpleMessageStore(1, 1, 1000); + final Message testMessage1 = MessageBuilder.withPayload("foo").build(); + final Message testMessage2 = MessageBuilder.withPayload("bar").build(); + + store2.addMessage(testMessage1); + + final CountDownLatch message2Latch = new CountDownLatch(1); + + Executors.newSingleThreadExecutor().execute(new Runnable() { + + @Override + public void run() { + store2.addMessage(testMessage2); + message2Latch.countDown(); + } + + }); + // Simulate a blocked consumer + Thread.sleep(10); + Message t1 = store2.removeMessage(testMessage1.getHeaders().getId()); + assertEquals(testMessage1, t1); + + assertTrue(message2Latch.await(10, TimeUnit.SECONDS)); + Message t2 = store2.getMessage(testMessage2.getHeaders().getId()); + assertEquals(testMessage2, t2); + } + + @Test(expected = MessagingException.class) + public void shouldTimeoutAfterWaitIfCapacity() throws InterruptedException { + SimpleMessageStore store2 = new SimpleMessageStore(1, 1, 10); + store2.addMessage(new GenericMessage("foo")); + // This should throw + store2.addMessage(new GenericMessage("foo")); + fail("Should have thrown already"); + } + + @Test(expected = MessagingException.class) public void shouldNotHoldMoreThanGroupCapacity() { SimpleMessageStore store = new SimpleMessageStore(0, 1); @@ -68,6 +145,44 @@ public class SimpleMessageStoreTests { store.addMessageToGroup("foo", testMessage2); } + @Test + public void shouldWaitIfGroupCapacity() throws InterruptedException { + final SimpleMessageStore store2 = new SimpleMessageStore(1, 1, 1000); + final Message testMessage1 = MessageBuilder.withPayload("foo").build(); + final Message testMessage2 = MessageBuilder.withPayload("bar").build(); + + store2.addMessageToGroup("foo", testMessage1); + + final CountDownLatch message2Latch = new CountDownLatch(1); + + Executors.newSingleThreadExecutor().execute(new Runnable() { + + @Override + public void run() { + store2.addMessageToGroup("foo", testMessage2); + message2Latch.countDown(); + } + + }); + // Simulate a blocked consumer + Thread.sleep(10); + store2.removeMessagesFromGroup("foo", testMessage1); + + assertTrue(message2Latch.await(10, TimeUnit.SECONDS)); + MessageGroup messageGroup = store2.getMessageGroup("foo"); + messageGroup.getMessages().contains(testMessage2); + } + + @Test(expected = MessagingException.class) + public void shouldTimeoutAfterWaitIfGroupCapacity() { + SimpleMessageStore store2 = new SimpleMessageStore(1, 1, 1); + store2.addMessageToGroup("foo", MessageBuilder.withPayload("foo").build()); + // This should throw + store2.addMessageToGroup("foo", MessageBuilder.withPayload("bar").build()); + fail("Should have thrown already"); + } + + @Test public void shouldHoldCapacityExactly() { SimpleMessageStore store = new SimpleMessageStore(2); @@ -77,6 +192,34 @@ public class SimpleMessageStoreTests { store.addMessage(testMessage2); } + @Test + public void shouldReleaseGroupCapacity() { + SimpleMessageStore store = new SimpleMessageStore(0, 1); + Message testMessage1 = MessageBuilder.withPayload("foo").build(); + Message testMessage2 = MessageBuilder.withPayload("bar").build(); + store.addMessageToGroup("foo", testMessage1); + try { + store.addMessageToGroup("foo", testMessage2); + fail("Should have thrown"); + } + catch (Exception e) { + assertThat(e, instanceOf(MessagingException.class)); + assertThat(e.getMessage(), containsString("was out of capacity (1) for group 'foo'")); + } + store.removeMessagesFromGroup("foo", testMessage2); + try { + store.addMessageToGroup("foo", testMessage2); + fail("Should have thrown"); + } + catch (Exception e) { + assertThat(e, instanceOf(MessagingException.class)); + assertThat(e.getMessage(), containsString("was out of capacity (1) for group 'foo'")); + } + store.removeMessagesFromGroup("foo", testMessage1); + store.addMessageToGroup("foo", testMessage2); + } + + @Test public void shouldListByCorrelation() throws Exception { SimpleMessageStore store = new SimpleMessageStore(); @@ -91,7 +234,8 @@ public class SimpleMessageStoreTests { Message testMessage1 = MessageBuilder.withPayload("foo").build(); store.addMessageToGroup("bar", testMessage1); Message testMessage2 = store.getMessageGroup("bar").getOne(); - MessageGroup group = store.removeMessageFromGroup("bar", testMessage2); + store.removeMessagesFromGroup("bar", testMessage2); + MessageGroup group = store.getMessageGroup("bar"); assertEquals(0, group.size()); assertEquals(0, store.getMessageGroup("bar").size()); } @@ -120,10 +264,12 @@ public class SimpleMessageStoreTests { @Test public void shouldRegisterCallbacks() throws Exception { SimpleMessageStore store = new SimpleMessageStore(); - store.setExpiryCallbacks(Arrays. asList(new MessageGroupStore.MessageGroupCallback() { + store.setExpiryCallbacks(Arrays.asList(new MessageGroupStore.MessageGroupCallback() { + @Override public void execute(MessageGroupStore messageGroupStore, MessageGroup group) { } + })); assertEquals(1, ((Collection) ReflectionTestUtils.getField(store, "expiryCallbacks")).size()); } @@ -134,11 +280,13 @@ 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()); } + }); Message testMessage1 = MessageBuilder.withPayload("foo").build(); diff --git a/spring-integration-file/src/main/java/org/springframework/integration/file/FileWritingMessageHandler.java b/spring-integration-file/src/main/java/org/springframework/integration/file/FileWritingMessageHandler.java index f1ac5861cf..c7cfb70b1d 100644 --- a/spring-integration-file/src/main/java/org/springframework/integration/file/FileWritingMessageHandler.java +++ b/spring-integration-file/src/main/java/org/springframework/integration/file/FileWritingMessageHandler.java @@ -307,7 +307,7 @@ public class FileWritingMessageHandler extends AbstractReplyProducingMessageHand if (!resultFile.exists() && generatedFileName.replaceAll("/", Matcher.quoteReplacement(File.separator)) .contains(File.separator)) { - resultFile.getParentFile().mkdirs(); + resultFile.getParentFile().mkdirs(); //NOSONAR - will fail on the writing below } if (payload instanceof File) { resultFile = handleFileMessage((File) payload, tempFile, resultFile); diff --git a/spring-integration-gemfire/src/test/java/org/springframework/integration/gemfire/store/GemfireGroupStoreTests.java b/spring-integration-gemfire/src/test/java/org/springframework/integration/gemfire/store/GemfireGroupStoreTests.java index 54001ef40b..0f82385810 100644 --- a/spring-integration-gemfire/src/test/java/org/springframework/integration/gemfire/store/GemfireGroupStoreTests.java +++ b/spring-integration-gemfire/src/test/java/org/springframework/integration/gemfire/store/GemfireGroupStoreTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2007-2015 the original author or authors + * Copyright 2007-2016 the original author or authors * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -13,6 +13,7 @@ * See the License for the specific language governing permissions and * limitations under the License. */ + package org.springframework.integration.gemfire.store; import static org.junit.Assert.assertEquals; @@ -32,7 +33,6 @@ import java.util.concurrent.TimeUnit; import org.junit.After; import org.junit.Before; import org.junit.Ignore; -import org.junit.Rule; import org.junit.Test; import org.springframework.context.support.ClassPathXmlApplicationContext; @@ -52,13 +52,13 @@ import org.springframework.util.Assert; import com.gemstone.gemfire.cache.Cache; import com.gemstone.gemfire.cache.Region; import com.gemstone.gemfire.cache.Scope; - import junit.framework.AssertionFailedError; /** * @author Oleg Zhurakousky * @author David Turanski * @author Gary Russell + * @author Artem Bilan * */ public class GemfireGroupStoreTests { @@ -67,7 +67,7 @@ public class GemfireGroupStoreTests { private Region region; - @Rule +// @Rule public LongRunningIntegrationTest longTests = new LongRunningIntegrationTest(); @Test @@ -119,7 +119,7 @@ public class GemfireGroupStoreTests { messageGroup = store.getMessageGroup(1); assertEquals(3, messageGroup.size()); - messageGroup = store.removeMessageFromGroup(messageGroup.getGroupId(), message); + store.removeMessagesFromGroup(messageGroup.getGroupId(), message); messageGroup = store.getMessageGroup(1); assertEquals(2, messageGroup.size()); @@ -163,14 +163,14 @@ public class GemfireGroupStoreTests { store.afterPropertiesSet(); MessageGroup messageGroup = store.getMessageGroup(1); store.addMessageToGroup(messageGroup.getGroupId(), new GenericMessage("1")); - store.removeMessageFromGroup(1, new GenericMessage("2")); + store.removeMessagesFromGroup(1, new GenericMessage("2")); } @Test public void testRemoveNonExistingMessageFromNonExistingTheGroup() throws Exception { GemfireMessageStore store = new GemfireMessageStore(this.region); store.afterPropertiesSet(); - store.removeMessageFromGroup(1, new GenericMessage("2")); + store.removeMessagesFromGroup(1, new GenericMessage("2")); } @Test @@ -214,7 +214,8 @@ public class GemfireGroupStoreTests { GemfireMessageStore store3 = new GemfireMessageStore(this.region); store3.afterPropertiesSet(); - messageGroup = store3.removeMessageFromGroup(1, message); + store3.removeMessagesFromGroup(1, message); + messageGroup = store3.getMessageGroup(1); assertEquals(1, messageGroup.getMessages().size()); } @@ -308,7 +309,8 @@ public class GemfireGroupStoreTests { executor.execute(new Runnable() { @Override public void run() { - MessageGroup group = store2.removeMessageFromGroup(1, message); + store2.removeMessagesFromGroup(1, message); + MessageGroup group = store2.getMessageGroup(1); if (group.getMessages().size() != 0) { failures.add("REMOVE"); throw new AssertionFailedError("Failed on Remove"); @@ -318,7 +320,7 @@ public class GemfireGroupStoreTests { executor.shutdown(); executor.awaitTermination(10, TimeUnit.SECONDS); - store2.removeMessageFromGroup(1, message); // ensures that if ADD thread executed after REMOVE, the store is empty for the next cycle + store2.removeMessagesFromGroup(1, message); // ensures that if ADD thread executed after REMOVE, the store is empty for the next cycle } assertTrue(failures.size() == 0); } 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 05ed21663c..05da97a323 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 @@ -1,5 +1,5 @@ /* - * Copyright 2002-2015 the original author or authors. + * Copyright 2002-2016 the original author or authors. * * Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in compliance with * the License. You may obtain a copy of the License at @@ -44,7 +44,6 @@ import org.springframework.integration.jdbc.store.JdbcChannelMessageStore; import org.springframework.integration.store.AbstractMessageGroupStore; import org.springframework.integration.store.MessageGroup; import org.springframework.integration.store.MessageStore; -import org.springframework.integration.store.SimpleMessageGroup; import org.springframework.integration.util.UUIDConverter; import org.springframework.jdbc.core.JdbcOperations; import org.springframework.jdbc.core.JdbcTemplate; @@ -79,6 +78,7 @@ import org.springframework.util.StringUtils; * @author Gunnar Hillert * @author Will Schipp * @author Gary Russell + * @author Artem Bilan * * @since 2.0 */ @@ -169,8 +169,6 @@ public class JdbcMessageStore extends AbstractMessageGroupStore implements Messa } } - public static final int DEFAULT_LONG_STRING_LENGTH = 2500; - /** * The name of the message header that stores a flag to indicate that the message has been saved. This is an * optimization for the put method. @@ -427,10 +425,12 @@ public class JdbcMessageStore extends AbstractMessageGroupStore implements Messa final AtomicReference completeFlag = new AtomicReference(); final AtomicReference lastReleasedSequenceRef = new AtomicReference(); - List> messages = jdbcTemplate.query(getQuery(Query.LIST_MESSAGES_BY_GROUP_KEY), new Object[] { key, region }, mapper); + List> messages = jdbcTemplate.query(getQuery(Query.LIST_MESSAGES_BY_GROUP_KEY), + new Object[] { key, region }, mapper); jdbcTemplate.query(getQuery(Query.GET_GROUP_INFO), new Object[] { key, region}, new RowCallbackHandler() { + @Override public void processRow(ResultSet rs) throws SQLException { updateDate.set(rs.getTimestamp("UPDATED_DATE")); @@ -441,6 +441,7 @@ public class JdbcMessageStore extends AbstractMessageGroupStore implements Messa lastReleasedSequenceRef.set(rs.getInt("LAST_RELEASED_SEQUENCE")); } + }); if (createDate.get() == null && updateDate.get() == null) { @@ -449,22 +450,23 @@ public class JdbcMessageStore extends AbstractMessageGroupStore implements Messa logger.warn("Missing group row for message id: " + message.getHeaders().getId()); } } - return new SimpleMessageGroup(groupId); + return getMessageGroupFactory().create(groupId); } long timestamp = createDate.get().getTime(); boolean complete = completeFlag.get(); - - SimpleMessageGroup messageGroup = new SimpleMessageGroup(messages, groupId, timestamp, complete); - messageGroup.setLastModified(updateDate.get().getTime()); - + long lastModified = updateDate.get().getTime(); int lastReleasedSequenceNumber = lastReleasedSequenceRef.get(); - messageGroup.setLastReleasedMessageSequenceNumber(lastReleasedSequenceNumber); + MessageGroup messageGroup = getMessageGroupFactory() + .create(messages, groupId, timestamp, complete); + messageGroup.setLastModified(lastModified); + messageGroup.setLastReleasedMessageSequenceNumber(lastReleasedSequenceNumber); return messageGroup; } @Override + @Deprecated public MessageGroup removeMessageFromGroup(Object groupId, Message messageToRemove) { final String groupKey = getKey(groupId); final String messageId = getKey(messageToRemove.getHeaders().getId()); @@ -499,22 +501,22 @@ public class JdbcMessageStore extends AbstractMessageGroupStore implements Messa messages, getRemoveBatchSize(), new ParameterizedPreparedStatementSetter>() { - @Override - public void setValues(PreparedStatement ps, Message messageToRemove) throws SQLException { - ps.setString(1, groupKey); - ps.setString(2, getKey(messageToRemove.getHeaders().getId())); - ps.setString(3, region); - } + @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); - } + @Override + public void setValues(PreparedStatement ps, Message messageToRemove) throws SQLException { + ps.setString(1, getKey(messageToRemove.getHeaders().getId())); + ps.setString(2, region); + } }); this.updateMessageGroup(groupKey); } diff --git a/spring-integration-jdbc/src/main/java/org/springframework/integration/jdbc/store/JdbcChannelMessageStore.java b/spring-integration-jdbc/src/main/java/org/springframework/integration/jdbc/store/JdbcChannelMessageStore.java index a4b090d6c8..90f0393c17 100644 --- a/spring-integration-jdbc/src/main/java/org/springframework/integration/jdbc/store/JdbcChannelMessageStore.java +++ b/spring-integration-jdbc/src/main/java/org/springframework/integration/jdbc/store/JdbcChannelMessageStore.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2015 the original author or authors. + * Copyright 2002-2016 the original author or authors. * * Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in compliance with * the License. You may obtain a copy of the License at @@ -25,7 +25,6 @@ import java.util.UUID; import java.util.concurrent.locks.Lock; import java.util.concurrent.locks.ReadWriteLock; import java.util.concurrent.locks.ReentrantReadWriteLock; - import javax.sql.DataSource; import org.apache.commons.logging.Log; @@ -49,10 +48,11 @@ import org.springframework.integration.jdbc.store.channel.MySqlChannelMessageSto import org.springframework.integration.jdbc.store.channel.OracleChannelMessageStoreQueryProvider; import org.springframework.integration.jdbc.store.channel.PostgresChannelMessageStoreQueryProvider; import org.springframework.integration.store.MessageGroup; +import org.springframework.integration.store.MessageGroupFactory; import org.springframework.integration.store.MessageGroupStore; import org.springframework.integration.store.MessageStore; import org.springframework.integration.store.PriorityCapableChannelMessageStore; -import org.springframework.integration.store.SimpleMessageGroup; +import org.springframework.integration.store.SimpleMessageGroupFactory; import org.springframework.integration.support.DefaultMessageBuilderFactory; import org.springframework.integration.support.MessageBuilderFactory; import org.springframework.integration.support.utils.IntegrationUtils; @@ -151,6 +151,8 @@ public class JdbcChannelMessageStore implements PriorityCapableChannelMessageSto private volatile Map queryCache = new HashMap(); + private volatile MessageGroupFactory messageGroupFactory = new SimpleMessageGroupFactory(); + private boolean usingIdCache = false; private boolean priorityEnabled; @@ -360,6 +362,22 @@ public class JdbcChannelMessageStore implements PriorityCapableChannelMessageSto return this.priorityEnabled; } + /** + * Specify the {@link MessageGroupFactory} to create {@link MessageGroup} object where + * it is necessary. + * Defaults to {@link SimpleMessageGroupFactory}. + * @param messageGroupFactory the {@link MessageGroupFactory} to use. + * @since 4.3 + */ + public void setMessageGroupFactory(MessageGroupFactory messageGroupFactory) { + Assert.notNull(messageGroupFactory, "'messageGroupFactory' must not be null"); + this.messageGroupFactory = messageGroupFactory; + } + + protected MessageGroupFactory getMessageGroupFactory() { + return this.messageGroupFactory; + } + @Override public void setBeanFactory(BeanFactory beanFactory) throws BeansException { this.beanFactory = beanFactory; @@ -468,7 +486,7 @@ public class JdbcChannelMessageStore implements PriorityCapableChannelMessageSto */ @Override public MessageGroup getMessageGroup(Object groupId) { - return new SimpleMessageGroup(groupId); + return getMessageGroupFactory().create(groupId); } /** 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 6b8a7943b8..bea58883b9 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 @@ -1,5 +1,5 @@ /* - * Copyright 2002-2014 the original author or authors. + * Copyright 2002-2016 the original author or authors. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -243,7 +243,7 @@ public class JdbcMessageStoreTests { String groupId = "X"; Message message = MessageBuilder.withPayload("foo").setCorrelationId(groupId).build(); messageStore.addMessageToGroup(groupId, message); - messageStore.removeMessageFromGroup(groupId, message); + messageStore.removeMessagesFromGroup(groupId, message); MessageGroup group = messageStore.getMessageGroup(groupId); assertEquals(0, group.size()); } @@ -532,7 +532,7 @@ public class JdbcMessageStoreTests { messageStore.completeGroup(messageGroup.getGroupId()); //now clear the messages for (Message message : messageGroup.getMessages()) { - messageStore.removeMessageFromGroup(groupId, message); + messageStore.removeMessagesFromGroup(groupId, message); }//end for //'add' the other message --> emulated by getting the messageGroup messageGroup = messageStore.getMessageGroup(groupId); @@ -540,6 +540,4 @@ public class JdbcMessageStoreTests { assertTrue(messageGroup.isComplete()); } - - } diff --git a/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/mysql/MySqlJdbcMessageStoreTests.java b/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/mysql/MySqlJdbcMessageStoreTests.java index 02285d1722..0f2267ff12 100644 --- a/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/mysql/MySqlJdbcMessageStoreTests.java +++ b/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/mysql/MySqlJdbcMessageStoreTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2013 the original author or authors. + * Copyright 2002-2016 the original author or authors. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -42,6 +42,7 @@ import org.junit.Before; import org.junit.Ignore; import org.junit.Test; import org.junit.runner.RunWith; + import org.springframework.beans.factory.annotation.Autowired; import org.springframework.core.serializer.Deserializer; import org.springframework.core.serializer.Serializer; @@ -84,10 +85,11 @@ import org.springframework.transaction.support.TransactionTemplate; * schema-mysql-5_6_4.sql * * @author Gunnar Hillert + * @author Artem Bilan */ @ContextConfiguration @RunWith(SpringJUnit4ClassRunner.class) -@DirtiesContext(classMode=ClassMode.AFTER_EACH_TEST_METHOD) +@DirtiesContext(classMode = ClassMode.AFTER_EACH_TEST_METHOD) @Ignore public class MySqlJdbcMessageStoreTests { @@ -111,16 +113,17 @@ public class MySqlJdbcMessageStoreTests { public void afterTest() { final JdbcTemplate jdbcTemplate = new JdbcTemplate(dataSource); new TransactionTemplate(this.transactionManager).execute(new TransactionCallback() { - public Void doInTransaction(TransactionStatus status) { - final int deletedGroupToMessageRows = jdbcTemplate.update("delete from INT_GROUP_TO_MESSAGE"); - final int deletedMessages = jdbcTemplate.update("delete from INT_MESSAGE"); - final int deletedMessageGroups = jdbcTemplate.update("delete from INT_MESSAGE_GROUP"); - LOG.info(String.format("Cleaning Database - Deleted Messages: %s, " + - "Deleted GroupToMessage Rows: %s, Deleted Message Groups: %s", - deletedMessages, deletedGroupToMessageRows, deletedMessageGroups)); - return null; - } + public Void doInTransaction(TransactionStatus status) { + final int deletedGroupToMessageRows = jdbcTemplate.update("delete from INT_GROUP_TO_MESSAGE"); + final int deletedMessages = jdbcTemplate.update("delete from INT_MESSAGE"); + final int deletedMessageGroups = jdbcTemplate.update("delete from INT_MESSAGE_GROUP"); + + LOG.info(String.format("Cleaning Database - Deleted Messages: %s, " + + "Deleted GroupToMessage Rows: %s, Deleted Message Groups: %s", + deletedMessages, deletedGroupToMessageRows, deletedMessageGroups)); + return null; + } }); } @@ -146,7 +149,7 @@ public class MySqlJdbcMessageStoreTests { @Test @Transactional - public void testWithMessageHistory() throws Exception{ + public void testWithMessageHistory() throws Exception { Message message = new GenericMessage("Hello"); DirectChannel fooChannel = new DirectChannel(); @@ -179,12 +182,14 @@ public class MySqlJdbcMessageStoreTests { public void testSerializer() throws Exception { // N.B. these serializers are not realistic (just for test purposes) messageStore.setSerializer(new Serializer>() { + public void serialize(Message object, OutputStream outputStream) throws IOException { outputStream.write(((Message) object).getPayload().toString().getBytes()); outputStream.flush(); } }); messageStore.setDeserializer(new Deserializer>() { + public GenericMessage deserialize(InputStream inputStream) throws IOException { BufferedReader reader = new BufferedReader(new InputStreamReader(inputStream)); return new GenericMessage(reader.readLine()); @@ -277,7 +282,7 @@ public class MySqlJdbcMessageStoreTests { String groupId = "X"; Message message = MessageBuilder.withPayload("foo").setCorrelationId(groupId).build(); messageStore.addMessageToGroup(groupId, message); - messageStore.removeMessageFromGroup(groupId, message); + messageStore.removeMessagesFromGroup(groupId, message); MessageGroup group = messageStore.getMessageGroup(groupId); assertEquals(0, group.size()); } @@ -297,7 +302,7 @@ public class MySqlJdbcMessageStoreTests { String uuidGroupId = UUIDConverter.getUUID(groupId).toString(); assertTrue(template.queryForList( - "SELECT * from INT_GROUP_TO_MESSAGE where GROUP_KEY = '" + uuidGroupId + "'").size() == 0); + "SELECT * from INT_GROUP_TO_MESSAGE where GROUP_KEY = '" + uuidGroupId + "'").size() == 0); } @Test @@ -362,6 +367,7 @@ public class MySqlJdbcMessageStoreTests { Message message = MessageBuilder.withPayload("foo").setCorrelationId(groupId).build(); messageStore.addMessageToGroup(groupId, message); messageStore.registerMessageGroupExpiryCallback(new MessageGroupCallback() { + public void execute(MessageGroupStore messageGroupStore, MessageGroup group) { messageGroupStore.removeMessageGroup(group.getGroupId()); } @@ -385,6 +391,7 @@ public class MySqlJdbcMessageStoreTests { messageStore.setTimeoutOnIdle(true); messageStore.addMessageToGroup(groupId, message); messageStore.registerMessageGroupExpiryCallback(new MessageGroupCallback() { + public void execute(MessageGroupStore messageGroupStore, MessageGroup group) { messageGroupStore.removeMessageGroup(group.getGroupId()); } @@ -511,10 +518,11 @@ public class MySqlJdbcMessageStoreTests { assertNotNull(messageFromRegion2); LOG.info("messageFromRegion1: " + messageFromRegion1.getHeaders().getId() + "; Sequence #: " + new IntegrationMessageHeaderAccessor(messageFromRegion1).getSequenceNumber()); - LOG.info("messageFromRegion2: " + messageFromRegion2.getHeaders().getId() + "; Sequence #: " +new IntegrationMessageHeaderAccessor(messageFromRegion2).getSequenceNumber()); + LOG.info("messageFromRegion2: " + messageFromRegion2.getHeaders().getId() + "; Sequence #: " + new IntegrationMessageHeaderAccessor(messageFromRegion2).getSequenceNumber()); assertEquals(Integer.valueOf(1), (Integer) messageFromRegion1.getHeaders().get(IntegrationMessageHeaderAccessor.SEQUENCE_NUMBER)); assertEquals(Integer.valueOf(2), (Integer) messageFromRegion2.getHeaders().get(IntegrationMessageHeaderAccessor.SEQUENCE_NUMBER)); } + } 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 f40e1a6983..9ad99466a0 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 @@ -1,5 +1,5 @@ /* - * Copyright 2013-2015 the original author or authors. + * Copyright 2013-2016 the original author or authors. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -35,7 +35,6 @@ import org.springframework.integration.store.MessageGroup; import org.springframework.integration.store.MessageGroupMetadata; import org.springframework.integration.store.MessageGroupStore; import org.springframework.integration.store.MessageStore; -import org.springframework.integration.store.SimpleMessageGroup; import org.springframework.jmx.export.annotation.ManagedAttribute; import org.springframework.messaging.Message; import org.springframework.util.Assert; @@ -161,7 +160,7 @@ public class ConfigurableMongoDbMessageStore extends AbstractConfigurableMongoDb for (MessageDocument document : messageDocuments) { messages.add(document.getMessage()); } - SimpleMessageGroup group = new SimpleMessageGroup(messages, groupId, createdTime, complete); + MessageGroup group = getMessageGroupFactory().create(messages, groupId, createdTime, complete); group.setLastReleasedMessageSequenceNumber(lastReleasedSequence); group.setLastModified(lastModifiedTime); @@ -201,6 +200,7 @@ public class ConfigurableMongoDbMessageStore extends AbstractConfigurableMongoDb } @Override + @Deprecated public MessageGroup removeMessageFromGroup(final Object groupId, final Message messageToRemove) { Assert.notNull(groupId, "'groupId' must not be null"); Assert.notNull(messageToRemove, "'messageToRemove' must not be null"); diff --git a/spring-integration-mongodb/src/main/java/org/springframework/integration/mongodb/store/MongoDbChannelMessageStore.java b/spring-integration-mongodb/src/main/java/org/springframework/integration/mongodb/store/MongoDbChannelMessageStore.java index 45590d0120..64815010a1 100644 --- a/spring-integration-mongodb/src/main/java/org/springframework/integration/mongodb/store/MongoDbChannelMessageStore.java +++ b/spring-integration-mongodb/src/main/java/org/springframework/integration/mongodb/store/MongoDbChannelMessageStore.java @@ -1,5 +1,5 @@ /* - * Copyright 2014 the original author or authors. + * Copyright 2014-2016 the original author or authors. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -25,7 +25,6 @@ import org.springframework.data.mongodb.core.query.Query; import org.springframework.integration.IntegrationMessageHeaderAccessor; import org.springframework.integration.store.MessageGroup; import org.springframework.integration.store.PriorityCapableChannelMessageStore; -import org.springframework.integration.store.SimpleMessageGroup; import org.springframework.messaging.Message; import org.springframework.util.Assert; @@ -119,7 +118,7 @@ public class MongoDbChannelMessageStore extends AbstractConfigurableMongoDbMessa */ @Override public MessageGroup getMessageGroup(Object groupId) { - return new SimpleMessageGroup(groupId); + return getMessageGroupFactory().create(groupId); } @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 b73bb42b6f..d04fcab033 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 @@ -1,5 +1,5 @@ /* - * Copyright 2002-2015 the original author or authors. + * Copyright 2002-2016 the original author or authors. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -64,7 +64,6 @@ import org.springframework.integration.store.AbstractMessageGroupStore; import org.springframework.integration.store.MessageGroup; import org.springframework.integration.store.MessageGroupStore; import org.springframework.integration.store.MessageStore; -import org.springframework.integration.store.SimpleMessageGroup; import org.springframework.integration.support.MutableMessageBuilder; import org.springframework.jmx.export.annotation.ManagedAttribute; import org.springframework.messaging.Message; @@ -264,7 +263,7 @@ public class MongoDbMessageStore extends AbstractMessageGroupStore messages.add(messageWrapper.getMessage()); } - SimpleMessageGroup messageGroup = new SimpleMessageGroup(messages, groupId, timestamp, completeGroup); + MessageGroup messageGroup = getMessageGroupFactory().create(messages, groupId, timestamp, completeGroup); messageGroup.setLastModified(lastModified); if (lastReleasedSequenceNumber > 0){ messageGroup.setLastReleasedMessageSequenceNumber(lastReleasedSequenceNumber); @@ -305,6 +304,7 @@ public class MongoDbMessageStore extends AbstractMessageGroupStore } @Override + @Deprecated public MessageGroup removeMessageFromGroup(final Object groupId, final Message messageToRemove) { Assert.notNull(groupId, "'groupId' must not be null"); Assert.notNull(messageToRemove, "'messageToRemove' must not be null"); 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 f20a4670f5..15847aaa0e 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-2015 the original author or authors. + * Copyright 2002-2016 the original author or authors. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -13,6 +13,7 @@ * See the License for the specific language governing permissions and * limitations under the License. */ + package org.springframework.integration.mongodb.store; import static org.junit.Assert.assertEquals; @@ -53,7 +54,7 @@ import com.mongodb.MongoClient; * @author Oleg Zhurakousky * @author Gary Russell * @author Amol Nayak - * + * @author Artem Bilan */ public abstract class AbstractMongoDbMessageGroupStoreTests extends MongoDbAvailableTests { @@ -197,22 +198,22 @@ public abstract class AbstractMongoDbMessageGroupStoreTests extends MongoDbAvail assertEquals(1, store.messageGroupSize(2)); assertEquals(1, store.messageGroupSize(3)); assertEquals(1, store.messageGroupSize(4)); - store.removeMessageFromGroup(3, messageA); + store.removeMessagesFromGroup(3, messageA); assertEquals(1, store.messageGroupSize(1)); assertEquals(1, store.messageGroupSize(2)); assertEquals(0, store.messageGroupSize(3)); assertEquals(1, store.messageGroupSize(4)); - store.removeMessageFromGroup(4, messageA); + store.removeMessagesFromGroup(4, messageA); assertEquals(1, store.messageGroupSize(1)); assertEquals(1, store.messageGroupSize(2)); assertEquals(0, store.messageGroupSize(3)); assertEquals(0, store.messageGroupSize(4)); - store.removeMessageFromGroup(2, messageA); + store.removeMessagesFromGroup(2, messageA); assertEquals(1, store.messageGroupSize(1)); assertEquals(0, store.messageGroupSize(2)); assertEquals(0, store.messageGroupSize(3)); assertEquals(0, store.messageGroupSize(4)); - store.removeMessageFromGroup(1, messageA); + store.removeMessagesFromGroup(1, messageA); assertEquals(0, store.messageGroupSize(1)); assertEquals(0, store.messageGroupSize(2)); assertEquals(0, store.messageGroupSize(3)); @@ -262,7 +263,8 @@ public abstract class AbstractMongoDbMessageGroupStoreTests extends MongoDbAvail assertNotNull(messageGroup); assertEquals(2, messageGroup.size()); - messageGroup = store.removeMessageFromGroup(1, messageA); + store.removeMessagesFromGroup(1, messageA); + messageGroup = store.getMessageGroup(1); assertEquals(1, messageGroup.size()); // validate that the updates were propagated to Mongo as well @@ -339,7 +341,8 @@ public abstract class AbstractMongoDbMessageGroupStoreTests extends MongoDbAvail assertNotNull(messageGroup); assertEquals(3, messageGroup.size()); - messageGroup = store.removeMessageFromGroup(1, message); + store.removeMessagesFromGroup(1, message); + messageGroup = store.getMessageGroup(1); assertEquals(2, messageGroup.size()); } @@ -363,7 +366,7 @@ public abstract class AbstractMongoDbMessageGroupStoreTests extends MongoDbAvail assertNotNull(messageGroup); assertEquals(3, messageGroup.size()); - store3.removeMessageFromGroup(1, message); + store3.removeMessagesFromGroup(1, message); messageGroup = store2.getMessageGroup(1); assertEquals(2, messageGroup.size()); @@ -391,7 +394,7 @@ public abstract class AbstractMongoDbMessageGroupStoreTests extends MongoDbAvail } assertEquals(3, counter); - store2.removeMessageFromGroup(1, message); + store2.removeMessagesFromGroup(1, message); iterator = store3.iterator(); counter = 0; diff --git a/spring-integration-redis/src/main/java/org/springframework/integration/redis/store/RedisChannelMessageStore.java b/spring-integration-redis/src/main/java/org/springframework/integration/redis/store/RedisChannelMessageStore.java index 84f7039891..54ad88d8be 100644 --- a/spring-integration-redis/src/main/java/org/springframework/integration/redis/store/RedisChannelMessageStore.java +++ b/spring-integration-redis/src/main/java/org/springframework/integration/redis/store/RedisChannelMessageStore.java @@ -1,5 +1,5 @@ /* - * Copyright 2014 the original author or authors. + * Copyright 2014-2016 the original author or authors. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -13,6 +13,7 @@ * See the License for the specific language governing permissions and * limitations under the License. */ + package org.springframework.integration.redis.store; import java.util.List; @@ -27,7 +28,8 @@ import org.springframework.data.redis.serializer.RedisSerializer; import org.springframework.data.redis.serializer.StringRedisSerializer; import org.springframework.integration.store.ChannelMessageStore; import org.springframework.integration.store.MessageGroup; -import org.springframework.integration.store.SimpleMessageGroup; +import org.springframework.integration.store.MessageGroupFactory; +import org.springframework.integration.store.SimpleMessageGroupFactory; import org.springframework.jmx.export.annotation.ManagedAttribute; import org.springframework.messaging.Message; import org.springframework.util.Assert; @@ -38,6 +40,7 @@ import org.springframework.util.Assert; * Requires {@link #setBeanName(String)} which is used as part of the key. * * @author Gary Russell + * @author Artem Bilan * @since 4.0 * */ @@ -45,6 +48,8 @@ public class RedisChannelMessageStore implements ChannelMessageStore, BeanNameAw private final RedisTemplate> redisTemplate; + private volatile MessageGroupFactory messageGroupFactory = new SimpleMessageGroupFactory(); + private String beanName; /** @@ -71,6 +76,22 @@ public class RedisChannelMessageStore implements ChannelMessageStore, BeanNameAw this.redisTemplate.setValueSerializer(valueSerializer); } + /** + * Specify the {@link MessageGroupFactory} to create {@link MessageGroup} object where + * it is necessary. + * Defaults to {@link SimpleMessageGroupFactory}. + * @param messageGroupFactory the {@link MessageGroupFactory} to use. + * @since 4.3 + */ + public void setMessageGroupFactory(MessageGroupFactory messageGroupFactory) { + Assert.notNull(messageGroupFactory, "'messageGroupFactory' must not be null"); + this.messageGroupFactory = messageGroupFactory; + } + + protected MessageGroupFactory getMessageGroupFactory() { + return this.messageGroupFactory; + } + @Override public void setBeanName(String name) { Assert.notNull(name, "'beanName' must not be null"); @@ -78,11 +99,11 @@ public class RedisChannelMessageStore implements ChannelMessageStore, BeanNameAw } protected String getBeanName() { - return beanName; + return this.beanName; } protected RedisTemplate> getRedisTemplate() { - return redisTemplate; + return this.redisTemplate; } @Override @@ -99,7 +120,7 @@ public class RedisChannelMessageStore implements ChannelMessageStore, BeanNameAw @Override public MessageGroup getMessageGroup(Object groupId) { List> messages = this.redisTemplate.boundListOps(groupId).range(0, -1); - return new SimpleMessageGroup(messages, groupId); + return getMessageGroupFactory().create(messages, groupId); } @Override diff --git a/spring-integration-redis/src/main/java/org/springframework/integration/redis/store/RedisChannelPriorityMessageStore.java b/spring-integration-redis/src/main/java/org/springframework/integration/redis/store/RedisChannelPriorityMessageStore.java index aae4e74c55..0a36b2b7a5 100644 --- a/spring-integration-redis/src/main/java/org/springframework/integration/redis/store/RedisChannelPriorityMessageStore.java +++ b/spring-integration-redis/src/main/java/org/springframework/integration/redis/store/RedisChannelPriorityMessageStore.java @@ -1,5 +1,5 @@ /* - * Copyright 2014 the original author or authors. + * Copyright 2014-2016 the original author or authors. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -27,7 +27,6 @@ import org.springframework.data.redis.connection.RedisConnectionFactory; import org.springframework.integration.IntegrationMessageHeaderAccessor; import org.springframework.integration.store.MessageGroup; import org.springframework.integration.store.PriorityCapableChannelMessageStore; -import org.springframework.integration.store.SimpleMessageGroup; import org.springframework.jmx.export.annotation.ManagedAttribute; import org.springframework.messaging.Message; import org.springframework.util.Assert; @@ -41,10 +40,12 @@ import org.springframework.util.Assert; * Requires that groupId is a String. * * @author Gary Russell + * @author Artem Bilan * @since 4.0 * */ -public class RedisChannelPriorityMessageStore extends RedisChannelMessageStore implements PriorityCapableChannelMessageStore { +public class RedisChannelPriorityMessageStore extends RedisChannelMessageStore + implements PriorityCapableChannelMessageStore { private final Comparator keysComparator = new Comparator() { @@ -85,7 +86,7 @@ public class RedisChannelPriorityMessageStore extends RedisChannelMessageStore i List> messages = this.getRedisTemplate().boundListOps(key).range(0, -1); allMessages.addAll(messages); } - return new SimpleMessageGroup(allMessages, groupId); + return getMessageGroupFactory().create(allMessages, groupId); } diff --git a/spring-integration-redis/src/test/java/org/springframework/integration/redis/store/RedisChannelMessageStoreTests.java b/spring-integration-redis/src/test/java/org/springframework/integration/redis/store/RedisChannelMessageStoreTests.java index 4646ea5458..c2f3e4fa90 100644 --- a/spring-integration-redis/src/test/java/org/springframework/integration/redis/store/RedisChannelMessageStoreTests.java +++ b/spring-integration-redis/src/test/java/org/springframework/integration/redis/store/RedisChannelMessageStoreTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2014 the original author or authors. + * Copyright 2014-2016 the original author or authors. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -13,6 +13,7 @@ * See the License for the specific language governing permissions and * limitations under the License. */ + package org.springframework.integration.redis.store; import static org.junit.Assert.assertEquals; @@ -118,9 +119,9 @@ public class RedisChannelMessageStoreTests extends RedisAvailableTests { @RedisAvailable public void testPriority() { for (int i = 0; i < 10; i++) { - Message message = MessageBuilder.withPayload(i).setPriority(i).build(); - this.testChannel3.send(message); - this.testChannel3.send(message); + this.testChannel3.send(MessageBuilder.withPayload(i).setPriority(i).build()); + //We need unique messages + this.testChannel3.send(MessageBuilder.withPayload(i).setPriority(i).build()); } this.testChannel3.send(MessageBuilder.withPayload(99).setPriority(199).build()); this.testChannel3.send(MessageBuilder.withPayload(98).build()); 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 b8c4e20bfa..4a6534253c 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-2015 the original author or authors + * Copyright 2007-2016 the original author or authors * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -13,6 +13,7 @@ * See the License for the specific language governing permissions and * limitations under the License. */ + package org.springframework.integration.redis.store; import static org.junit.Assert.assertEquals; @@ -29,7 +30,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; @@ -50,11 +50,12 @@ 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 * @author Gary Russell - * */ public class RedisMessageGroupStoreTests extends RedisAvailableTests { @@ -190,7 +191,8 @@ public class RedisMessageGroupStoreTests extends RedisAvailableTests { messageGroup = store.addMessageToGroup(messageGroup.getGroupId(), new GenericMessage("3")); assertEquals(3, messageGroup.size()); - messageGroup = store.removeMessageFromGroup(1, message); + store.removeMessagesFromGroup(1, message); + messageGroup = store.getMessageGroup(1); assertEquals(2, messageGroup.size()); // make sure the store is properly rebuild from Redis @@ -235,7 +237,7 @@ public class RedisMessageGroupStoreTests extends RedisAvailableTests { MessageGroup messageGroup = store.getMessageGroup(1); store.addMessageToGroup(messageGroup.getGroupId(), new GenericMessage("1")); - store.removeMessageFromGroup(1, new GenericMessage("2")); + store.removeMessagesFromGroup(1, new GenericMessage("2")); } @Test @@ -243,7 +245,7 @@ public class RedisMessageGroupStoreTests extends RedisAvailableTests { public void testRemoveNonExistingMessageFromNonExistingTheGroup() throws Exception{ RedisConnectionFactory jcf = getConnectionFactoryForTest(); RedisMessageStore store = new RedisMessageStore(jcf); - store.removeMessageFromGroup(1, new GenericMessage("2")); + store.removeMessagesFromGroup(1, new GenericMessage("2")); } @@ -264,7 +266,8 @@ public class RedisMessageGroupStoreTests extends RedisAvailableTests { RedisMessageStore store3 = new RedisMessageStore(jcf); - messageGroup = store3.removeMessageFromGroup(1, message); + store3.removeMessagesFromGroup(1, message); + messageGroup = store3.getMessageGroup(1); assertEquals(1, messageGroup.getMessages().size()); } @@ -340,7 +343,8 @@ public class RedisMessageGroupStoreTests extends RedisAvailableTests { executor.execute(new Runnable() { @Override public void run() { - MessageGroup group = store2.removeMessageFromGroup(1, message); + store2.removeMessagesFromGroup(1, message); + MessageGroup group = store2.getMessageGroup(1); if (group.getMessages().size() != 0){ failures.add("REMOVE"); throw new AssertionFailedError("Failed on Remove"); @@ -350,7 +354,7 @@ public class RedisMessageGroupStoreTests extends RedisAvailableTests { executor.shutdown(); executor.awaitTermination(10, TimeUnit.SECONDS); - store2.removeMessageFromGroup(1, message); // ensures that if ADD thread executed after REMOVE, the store is empty for the next cycle + store2.removeMessagesFromGroup(1, message); // ensures that if ADD thread executed after REMOVE, the store is empty for the next cycle } assertTrue(failures.size() == 0); } diff --git a/src/reference/asciidoc/message-store.adoc b/src/reference/asciidoc/message-store.adoc index 5e83eccf83..5045aa4e6d 100644 --- a/src/reference/asciidoc/message-store.adoc +++ b/src/reference/asciidoc/message-store.adoc @@ -84,3 +84,13 @@ Manipulation of the group outside of the aggregator may cause unpredictable resu For this reason, users should not perform such manipulation, or set the `copyOnGet` property to `true`. ===== + +[[message-group-factory]] +===== MessageGroupFactory + +Starting with _version 4.3_, some `MessageGroupStore` implementations can be injected with a custom +`MessageGroupFactory` strategy to create/customize the `MessageGroup` instances used by the `MessageGroupStore`. +This defaults to a `SimpleMessageGroupFactory` which produces `SimpleMessageGroup` s based on the `GroupType.HASH_SET` +(`LinkedHashSet`) internal collection. +Other possible options are `SYNCHRONISED_SET` and `BLOCKING_QUEUE`, where the last one can be used to reinstate the +previous `SimpleMessageGroup` behavior. diff --git a/src/reference/asciidoc/whats-new.adoc b/src/reference/asciidoc/whats-new.adoc index e6bd77b48e..6410c5ffc9 100644 --- a/src/reference/asciidoc/whats-new.adoc +++ b/src/reference/asciidoc/whats-new.adoc @@ -9,6 +9,14 @@ development process. [[x4.3-new-components]] === New Components +==== MessageGroupFactory + +The new `MessageGroupFactory` strategy has been introduced to allow a control over `MessageGroup` instances +in `MessageGroupStore` logic. +The `SimpleMessageGroupFactory` is provided for the `SimpleMessageGroup` with the `GroupType.HASH_SET` as the default +factory for the standard `MessageGroupStore` implementations. +See <> for more information. + [[x4.3-general]] === General Changes