From bffd40eaf442f4c7f486db1b5aa5387d5b0e5bec Mon Sep 17 00:00:00 2001 From: Oleg Zhurakousky Date: Mon, 22 Aug 2011 15:50:46 -0400 Subject: [PATCH] INT-2030: RedisMessageStore refactored RedisMessageStore to use SimpleMessageGroup. Deleted RedisMessageGroup polished RedisMessageStore, made it more inline with JdbcMessageStore, added additional tests more polishing on RedisMessageStore, added more tests removed test that was no longer valid. Downgraded spring-data-redis dependency to M4 changed UUID serializer to be StringRedisSerializer instead of custom class which essentially did the same thing polished code based on the review comments. Rename onX methods to doX, removed this.hasKey in favor of this.redisTemplate.hasKey polishing removed afterPropertiesSet() calls in test --- build.gradle | 2 +- .../redis/store/RedisMessageGroup.java | 250 ---------------- .../redis/store/RedisMessageStore.java | 280 +++++++++++------- .../redis/config/RedisChannelParserTests.java | 1 - .../store/RedisMessageGroupStoreTests.java | 241 +++++++++------ .../redis/store/RedisMessageStoreTests.java | 8 - 6 files changed, 324 insertions(+), 458 deletions(-) delete mode 100644 spring-integration-redis/src/main/java/org/springframework/integration/redis/store/RedisMessageGroup.java diff --git a/build.gradle b/build.gradle index 631d752b78..daa9c4f426 100644 --- a/build.gradle +++ b/build.gradle @@ -126,7 +126,7 @@ configure(javaprojects) { springAmqpVersion = '1.0.0.RELEASE' springDataMongoVersion = '1.0.0.BUILD-SNAPSHOT' springDataCommonsVersion = '1.2.0.BUILD-SNAPSHOT' - springDataRedisVersion = '1.0.0.BUILD-SNAPSHOT' + springDataRedisVersion = '1.0.0.M4' springGemfireVersion = '1.1.0.BUILD-SNAPSHOT' springSecurityVersion = '3.0.5.RELEASE' springWsVersion = '2.0.2.RELEASE' diff --git a/spring-integration-redis/src/main/java/org/springframework/integration/redis/store/RedisMessageGroup.java b/spring-integration-redis/src/main/java/org/springframework/integration/redis/store/RedisMessageGroup.java deleted file mode 100644 index 5d69139ea4..0000000000 --- a/spring-integration-redis/src/main/java/org/springframework/integration/redis/store/RedisMessageGroup.java +++ /dev/null @@ -1,250 +0,0 @@ -/* - * Copyright 2002-2011 the original author or authors. - * - * Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in compliance with - * the License. You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on - * an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the - * specific language governing permissions and limitations under the License. - */ -package org.springframework.integration.redis.store; - -import java.util.Collection; -import java.util.LinkedList; -import java.util.List; -import java.util.UUID; - -import org.springframework.data.redis.core.BoundListOperations; -import org.springframework.data.redis.core.RedisTemplate; -import org.springframework.integration.Message; -import org.springframework.integration.store.MessageGroup; -import org.springframework.integration.store.MessageStore; -import org.springframework.util.Assert; - -/** - * @author Oleg Zhurakousky - * @since 2.1 - */ -class RedisMessageGroup implements MessageGroup { - private final String MARKED_PREFIX = "MARKED_"; - private final String UNMARKED_PREFIX = "UNMARKED_"; - - private final String unmarkedId; - private final String markedId; - - private final List> unmarked = new LinkedList>(); - private final List> marked = new LinkedList>(); - - private final Object groupId; - private final RedisTemplate redisTemplate; - private final MessageStore messageStore; - private final Object lock = new Object(); - private final long timestamp = System.currentTimeMillis(); - - public RedisMessageGroup(MessageStore messageStore, RedisTemplate redisTemplate, Object groupId){ - this.groupId = groupId; - this.redisTemplate = redisTemplate; - this.messageStore = messageStore; - this.unmarkedId = UNMARKED_PREFIX + groupId; - this.markedId = MARKED_PREFIX + groupId; - BoundListOperations unmarkedOps = this.redisTemplate.boundListOps(unmarkedId); - BoundListOperations markedOps = this.redisTemplate.boundListOps(markedId); - this.rebuildLocalCache(unmarkedOps, markedOps); - } - - public boolean canAdd(Message message) { - return !isMember(message); - } - - public Collection> getUnmarked() { - return this.unmarked; - } - - - public Collection> getMarked() { - return this.marked; - } - - - public Object getGroupId() { - return this.groupId; - } - - public boolean isComplete() { - if (size() == 0) { - return true; - } - int sequenceSize = getSequenceSize(); - // If there is no sequence then it must be incomplete.... - return sequenceSize > 0 && sequenceSize == size(); - } - - public int getSequenceSize() { - if (size() == 0) { - return 0; - } - return getOne().getHeaders().getSequenceSize(); - } - - public int size() { - synchronized (lock) { - return marked.size() + unmarked.size(); - } - } - - public Message getOne() { - synchronized (lock) { - Message one = unmarked.get(0); - if (one == null) { - one = marked.get(0); - } - return one; - } - } - - public long getTimestamp() { - return this.timestamp; - } - - protected void markAll() { - BoundListOperations unmarkedOps = this.redisTemplate.boundListOps(unmarkedId); - BoundListOperations markedOps = this.redisTemplate.boundListOps(markedId); - long uSize = unmarkedOps.size(); - - synchronized (lock) { - unmarkedOps.rename(markedId); - this.unmarked.clear(); - this.rebuildLocalCache(null, markedOps); - } - - long mSize = markedOps.size(); - Assert.isTrue(mSize == uSize, "Failed to mark All messages in the message group"); - } - - protected void markMessage(String messageId) { - BoundListOperations unmarkedOps = this.redisTemplate.boundListOps(unmarkedId); - List messageIds = unmarkedOps.range(0, unmarkedOps.size()-1); - - synchronized (lock) { - for (Object id : messageIds) { - if (messageId.equals(id)){ - BoundListOperations markedOps = this.redisTemplate.boundListOps(markedId); - markedOps.rightPush(id); - unmarkedOps.remove(0, id); - this.rebuildLocalCache(unmarkedOps, markedOps); - return; - } - } - } - } - - /** - * Will add message to this MessageGroup. This particular implementation will first add the Message ID to the Redis set of - * Unmarked IDs and than will use the underlying MessageStore to add the actual Message. - */ - protected void add(Message message) { - String messageId = message.getHeaders().getId().toString(); - BoundListOperations unmarkedOps = this.redisTemplate.boundListOps(unmarkedId); - synchronized (lock) { - unmarkedOps.rightPush(messageId); - this.messageStore.addMessage(message); - this.rebuildLocalCache(unmarkedOps, null); - } - } - - protected void remove(Message message) { - UUID messageId = message.getHeaders().getId(); - BoundListOperations unmarkedOps = this.redisTemplate.boundListOps(unmarkedId); - BoundListOperations markedOps = this.redisTemplate.boundListOps(markedId); - - synchronized (lock) { - unmarkedOps.remove(0, messageId.toString()); - markedOps.remove(0, messageId.toString()); - this.messageStore.removeMessage(messageId); - this.rebuildLocalCache(unmarkedOps, markedOps); - } - - } - /** - * - */ - protected void destroy(){ - BoundListOperations unmarkedOps = this.redisTemplate.boundListOps(unmarkedId); - List messageIds = unmarkedOps.range(0, unmarkedOps.size()-1); - for (Object messageId : messageIds) { - this.messageStore.removeMessage(UUID.fromString(messageId.toString())); - } - this.redisTemplate.delete(unmarkedId); - - BoundListOperations markedOps = this.redisTemplate.boundListOps(markedId); - messageIds = markedOps.range(0, markedOps.size()-1); - for (Object messageId : messageIds) { - this.messageStore.removeMessage(UUID.fromString(messageId.toString())); - } - this.redisTemplate.delete(markedId); - this.rebuildLocalCache(unmarkedOps, markedOps); - } - - private Collection> buildMessageList(BoundListOperations mGroupOps){ - List> messages = new LinkedList>(); - List messageIds = mGroupOps.range(0, mGroupOps.size()-1); - for (Object messageId : messageIds) { - Message message = this.messageStore.getMessage(UUID.fromString(messageId.toString())); - messages.add((Message) message); - } - return messages; - } - - private boolean isMember(Message message){ - if (size() == 0) { - return false; - } - Integer messageSequenceNumber = message.getHeaders().getSequenceNumber(); - if (messageSequenceNumber != null && messageSequenceNumber > 0) { - Integer messageSequenceSize = message.getHeaders().getSequenceSize(); - if (!messageSequenceSize.equals(getSequenceSize())) { - return true; - } - else { - synchronized (lock) { - - BoundListOperations mGroupOps = this.redisTemplate.boundListOps(unmarkedId); - Collection> unmarked = this.buildMessageList(mGroupOps); - mGroupOps = this.redisTemplate.boundListOps(markedId); - Collection> marked = this.buildMessageList(mGroupOps); - if (containsSequenceNumber(unmarked, messageSequenceNumber) - || containsSequenceNumber(marked, messageSequenceNumber)) { - return true; - } - } - } - } - return false; - } - - private boolean containsSequenceNumber(Collection> messages, Integer messageSequenceNumber) { - for (Message member : messages) { - Integer memberSequenceNumber = member.getHeaders().getSequenceNumber(); - if (messageSequenceNumber.equals(memberSequenceNumber)) { - return true; - } - } - return false; - } - - private void rebuildLocalCache(BoundListOperations unmarkedOps, BoundListOperations markedOps){ - - if (unmarkedOps != null){ - this.unmarked.clear(); - this.unmarked.addAll(this.buildMessageList(unmarkedOps)); - } - if (markedOps != null){ - this.marked.clear(); - this.marked.addAll(this.buildMessageList(markedOps)); - } - } - -} diff --git a/spring-integration-redis/src/main/java/org/springframework/integration/redis/store/RedisMessageStore.java b/spring-integration-redis/src/main/java/org/springframework/integration/redis/store/RedisMessageStore.java index a8de46be65..a91b9d5866 100644 --- a/spring-integration-redis/src/main/java/org/springframework/integration/redis/store/RedisMessageStore.java +++ b/spring-integration-redis/src/main/java/org/springframework/integration/redis/store/RedisMessageStore.java @@ -13,19 +13,22 @@ * See the License for the specific language governing permissions and * limitations under the License. */ + package org.springframework.integration.redis.store; import java.util.ArrayList; -import java.util.HashMap; +import java.util.Collection; +import java.util.Collections; import java.util.Iterator; +import java.util.LinkedList; import java.util.List; -import java.util.Map; +import java.util.Set; import java.util.UUID; -import org.springframework.beans.factory.InitializingBean; import org.springframework.dao.DataAccessException; import org.springframework.data.redis.connection.RedisConnection; import org.springframework.data.redis.connection.RedisConnectionFactory; +import org.springframework.data.redis.core.BoundListOperations; import org.springframework.data.redis.core.BoundSetOperations; import org.springframework.data.redis.core.BoundValueOperations; import org.springframework.data.redis.core.RedisCallback; @@ -33,59 +36,70 @@ import org.springframework.data.redis.core.RedisTemplate; import org.springframework.data.redis.serializer.JdkSerializationRedisSerializer; import org.springframework.data.redis.serializer.RedisSerializer; import org.springframework.data.redis.serializer.SerializationException; +import org.springframework.data.redis.serializer.StringRedisSerializer; import org.springframework.integration.Message; 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.MessageStoreException; +import org.springframework.integration.store.SimpleMessageGroup; import org.springframework.jmx.export.annotation.ManagedAttribute; import org.springframework.util.Assert; /** + * An implementation of both the {@link MessageStore} and {@link MessageGroupStore} + * strategies that relies upon Redis for persistence. + * * @author Oleg Zhurakousky * @since 2.1 - * */ -public class RedisMessageStore extends AbstractMessageGroupStore implements MessageStore, InitializingBean{ - - private final String MESSAGE_GROUPS_KEY = "MESSAGE_GROUPS"; +public class RedisMessageStore extends AbstractMessageGroupStore implements MessageStore { + + private static final String MESSAGE_GROUPS_KEY = "MESSAGE_GROUPS"; + + private static final String MARKED_PREFIX = "MARKED_"; + + private static final String UNMARKED_PREFIX = "UNMARKED_"; + private final RedisTemplate redisTemplate; - private final Map messageGroups = new HashMap(); - - private volatile RedisSerializer valueSerializer = new JdkSerializationRedisSerializer(); - - private final Object lock = new Object(); - - public RedisMessageStore(RedisConnectionFactory connectionFactory){ + + public RedisMessageStore(RedisConnectionFactory connectionFactory) { this.redisTemplate = new RedisTemplate(); this.redisTemplate.setConnectionFactory(connectionFactory); - this.redisTemplate.setKeySerializer(new KeySerializer()); - this.redisTemplate.setValueSerializer(this.valueSerializer); + this.redisTemplate.setKeySerializer(new StringRedisSerializer()); + this.redisTemplate.setValueSerializer(new JdkSerializationRedisSerializer()); + } + + + public void setValueSerializer(RedisSerializer valueSerializer) { + Assert.notNull(valueSerializer, "'valueSerializer' must not be null"); + this.redisTemplate.setValueSerializer(valueSerializer); } public Message getMessage(final UUID id) { Assert.notNull(id, "'id' must not be null"); - if (this.keyExists(id)){ + if (this.redisTemplate.hasKey(id.toString())) { BoundValueOperations ops = redisTemplate.boundValueOps(id.toString()); Object result = ops.get(); Assert.isInstanceOf(Message.class, result, "Return value is not an instace of Message"); return (Message) result; } - return null; + return null; } - @SuppressWarnings("unchecked") public Message addMessage(Message message) { Assert.notNull(message, "'message' must not be null"); - BoundValueOperations ops = redisTemplate.boundValueOps(message.getHeaders().getId().toString()); + BoundValueOperations ops = redisTemplate.boundValueOps(message.getHeaders().getId().toString()); try { ops.set(message); - } catch (SerializationException e) { - throw new MessageStoreException(message, "It seems like while relying on the default RedisSerializer (JdkSerializationRedisSerializer) " + - "the Message contains data that is not Serializable. Either make it Serializable or provide your own implementation of " + + } + catch (SerializationException e) { + throw new MessageStoreException(message, "If relying on the default RedisSerializer (JdkSerializationRedisSerializer) " + + "the Message must be Serializable. Either make it Serializable or provide your own implementation of " + "RedisSerializer via 'setValueSerializer(..)'", e); } Object result = ops.get(); @@ -93,131 +107,187 @@ public class RedisMessageStore extends AbstractMessageGroupStore implements Mess return (Message) result; } - public Message removeMessage(UUID id) { Assert.notNull(id, "'id' must not be null"); Message message = this.getMessage(id); - if (message != null){ + if (message != null) { this.redisTemplate.delete(id.toString()); - return message; - } - else { - throw new IllegalArgumentException("Message with id '" + id + "' can not be removed since it does NOT exist"); } + return message; } @ManagedAttribute public long getMessageCount() { - return redisTemplate.execute(new RedisCallback() { - public Integer doInRedis(RedisConnection connection) - throws DataAccessException { - long l = connection.dbSize(); - Assert.isTrue(l <= Integer.MAX_VALUE, "Message count is out of range"); - return (int)l; + return redisTemplate.execute(new RedisCallback() { + public Long doInRedis(RedisConnection connection) throws DataAccessException { + return connection.dbSize(); } }); } - - public void setValueSerializer(RedisSerializer valueSerializer) { - this.valueSerializer = valueSerializer; - } - - private boolean keyExists(final UUID id){ - return redisTemplate.execute(new RedisCallback() { - public Boolean doInRedis(RedisConnection connection) - throws DataAccessException { - return connection.exists(id.toString().getBytes()); - } - }); - } - - private static class KeySerializer implements RedisSerializer { - public byte[] serialize(String value) throws SerializationException { - return value.getBytes(); - } - public String deserialize(byte[] bytes) throws SerializationException { - return new String(bytes); - } - - } + // MESSAGE GROUP methods - public void afterPropertiesSet() throws Exception { - Assert.notNull(this.valueSerializer, "'valueSerializer' must not be null"); - } /** - * + * Will create a new instance of SimpleMessageGroup initializing it with + * data collected from the Redis Message Store. */ public MessageGroup getMessageGroup(Object groupId) { Assert.notNull(groupId, "'groupId' must not be null"); - RedisMessageGroup group = this.messageGroups.get(groupId); - if (group == null){ - synchronized (lock) { - BoundSetOperations mGroupsOps = this.redisTemplate.boundSetOps(MESSAGE_GROUPS_KEY); - if (!mGroupsOps.isMember(groupId)){ - mGroupsOps.add(groupId); - } - group = new RedisMessageGroup(this, this.redisTemplate, groupId); - this.messageGroups.put(groupId, group); - } - } - - return group; + long timestamp = System.currentTimeMillis(); + Collection> unmarkedMessages = this.buildMessageList(this.redisTemplate.boundListOps(UNMARKED_PREFIX + groupId)); + Collection> markedMessages = this.buildMessageList(this.redisTemplate.boundListOps(MARKED_PREFIX + groupId)); + this.doCreateMessageGroupIfNecessary(groupId); + return new SimpleMessageGroup(unmarkedMessages, markedMessages, groupId, timestamp); } + /** - * + * Add a Message to the group with the provided group ID. */ public MessageGroup addMessageToGroup(Object groupId, Message message) { - Assert.notNull(groupId, "'groupId' must not be null"); + Assert.notNull(groupId, "'groupId' must not be null"); Assert.notNull(message, "'message' must not be null"); - - MessageGroup mg = this.getMessageGroup(groupId); - - Assert.isInstanceOf(RedisMessageGroup.class, mg, "MessageGroup is not an instance of RedisMessageGroup"); - ((RedisMessageGroup)mg).add(message); - return mg; + synchronized (groupId) { + this.doAddMessageToGroup(message, groupId); + this.addMessage(message); + return this.getMessageGroup(groupId); + } } + /** + * Mark all messages in the provided group. + */ public MessageGroup markMessageGroup(MessageGroup group) { - Assert.isInstanceOf(RedisMessageGroup.class, group, "MessageGroup is not an instance of RedisMessageGroup"); - ((RedisMessageGroup)group).markAll(); - return group; + Assert.notNull(group, "'group' must not be null"); + Object groupId = group.getGroupId(); + synchronized (groupId) { + this.doMarkMessageGroup(groupId); + return this.getMessageGroup(groupId); + } } - public MessageGroup removeMessageFromGroup(Object key, - Message messageToRemove) { - RedisMessageGroup messageGroup = (RedisMessageGroup) this.getMessageGroup(key); - messageGroup.remove(messageToRemove); - return messageGroup; + /** + * Remove a Message from the group with the provided group ID. + */ + public MessageGroup removeMessageFromGroup(Object groupId, Message messageToRemove) { + Assert.notNull(groupId, "'groupId' must not be null"); + Assert.notNull(messageToRemove, "'messageToRemove' must not be null"); + UUID messageId = messageToRemove.getHeaders().getId(); + synchronized (groupId) { + this.doRemoveMessageFromGroup(groupId, messageId); + this.removeMessage(messageId); + return this.getMessageGroup(groupId); + } } - public MessageGroup markMessageFromGroup(Object key, - Message messageToMark) { - RedisMessageGroup messageGroup = (RedisMessageGroup) this.getMessageGroup(key); - messageGroup.markMessage(messageToMark.getHeaders().getId().toString()); - return messageGroup; + /** + * Mark the given Message within the group corresponding to the provided group ID. + */ + public MessageGroup markMessageFromGroup(Object groupId, Message messageToMark) { + Assert.notNull(groupId, "'groupId' must not be null"); + Assert.notNull(messageToMark, "'messageToMark' must not be null"); + String messageIdAsString = messageToMark.getHeaders().getId().toString(); + synchronized (groupId) { + this.doMarkMessageFromGroup(messageIdAsString, groupId); + return this.getMessageGroup(groupId); + } } + /** + * Remove the MessageGroup with the provided group ID. + */ public void removeMessageGroup(Object groupId) { Assert.notNull(groupId, "'groupId' must not be null"); - RedisMessageGroup messageGroup = (RedisMessageGroup) this.getMessageGroup(groupId); - synchronized (messageGroup) { - messageGroup.destroy(); - BoundSetOperations mGroupsOps = this.redisTemplate.boundSetOps(MESSAGE_GROUPS_KEY); - mGroupsOps.remove(groupId); - this.redisTemplate.delete(groupId.toString()); - this.messageGroups.remove(groupId); + synchronized (groupId) { + this.doRemoveMessageGroup(groupId); } } @Override public Iterator iterator() { BoundSetOperations mGroupsOps = this.redisTemplate.boundSetOps(MESSAGE_GROUPS_KEY); + Set messageGroupIds = mGroupsOps.members(); List messageGroups = new ArrayList(); - for (Object msgGroupId : mGroupsOps.members()) { - messageGroups.add(new RedisMessageGroup(this, this.redisTemplate, msgGroupId)); + for (Object messageGroupId : messageGroupIds) { + messageGroups.add(this.getMessageGroup(messageGroupId)); } return messageGroups.iterator(); } + + private Collection> buildMessageList(BoundListOperations messageGroupOps) { + List> messages = new LinkedList>(); + if (messageGroupOps.size() == 0) { + return Collections.emptyList(); + } + List messageIds = messageGroupOps.range(0, messageGroupOps.size() - 1); + for (Object messageId : messageIds) { + Message message = this.getMessage(UUID.fromString(messageId.toString())); + if (message != null) { + messages.add((Message) message); + } + } + return messages; + } + + /* candidates for future abstract methods */ + + private void doCreateMessageGroupIfNecessary(Object groupId) { + BoundSetOperations messageGroupsOps = this.redisTemplate.boundSetOps(MESSAGE_GROUPS_KEY); + if (!messageGroupsOps.members().contains(groupId)) { + messageGroupsOps.add(groupId); + } + } + + private void doAddMessageToGroup(Message message, Object groupId) { + String messageId = message.getHeaders().getId().toString(); + BoundListOperations unmarkedOps = this.redisTemplate.boundListOps(UNMARKED_PREFIX + groupId); + unmarkedOps.rightPush(messageId); + } + + private void doMarkMessageGroup(Object groupId) { + BoundListOperations unmarkedOps = this.redisTemplate.boundListOps(UNMARKED_PREFIX + groupId); + unmarkedOps.rename(MARKED_PREFIX + groupId); + } + + private void doRemoveMessageFromGroup(Object groupId, UUID messageId) { + BoundListOperations unmarkedOps = this.redisTemplate.boundListOps(UNMARKED_PREFIX + groupId); + BoundListOperations markedOps = this.redisTemplate.boundListOps(MARKED_PREFIX + groupId); + unmarkedOps.remove(0, messageId.toString()); + markedOps.remove(0, messageId.toString()); + } + + private void doMarkMessageFromGroup(String messageIdAsString, Object groupId) { + BoundListOperations unmarkedOps = this.redisTemplate.boundListOps(UNMARKED_PREFIX + groupId); + if (unmarkedOps.size() > 0) { + List messageIds = unmarkedOps.range(0, unmarkedOps.size() - 1); + int objectIndex = messageIds.indexOf(messageIdAsString); + if (objectIndex > -1) { + BoundListOperations markedOps = this.redisTemplate.boundListOps(MARKED_PREFIX + groupId); + markedOps.rightPush(messageIdAsString); + unmarkedOps.remove(0, messageIdAsString); + } + } + } + + private void doRemoveMessageGroup(Object groupId) { + BoundListOperations unmarkedOps = this.redisTemplate.boundListOps(UNMARKED_PREFIX + groupId); + if (unmarkedOps.size() > 0) { + List messageIds = unmarkedOps.range(0, unmarkedOps.size() - 1); + for (Object messageId : messageIds) { + this.removeMessage(UUID.fromString(messageId.toString())); + } + this.redisTemplate.delete(UNMARKED_PREFIX + groupId); + } + BoundListOperations markedOps = this.redisTemplate.boundListOps(MARKED_PREFIX + groupId); + if (markedOps.size() > 0) { + List messageIds = markedOps.range(0, markedOps.size() - 1); + for (Object messageId : messageIds) { + this.removeMessage(UUID.fromString(messageId.toString())); + } + this.redisTemplate.delete(MARKED_PREFIX + groupId); + } + BoundSetOperations messageGroupsOps = this.redisTemplate.boundSetOps(MESSAGE_GROUPS_KEY); + messageGroupsOps.remove(groupId); + } + } diff --git a/spring-integration-redis/src/test/java/org/springframework/integration/redis/config/RedisChannelParserTests.java b/spring-integration-redis/src/test/java/org/springframework/integration/redis/config/RedisChannelParserTests.java index 440bbc61fb..52118f77cc 100644 --- a/spring-integration-redis/src/test/java/org/springframework/integration/redis/config/RedisChannelParserTests.java +++ b/spring-integration-redis/src/test/java/org/springframework/integration/redis/config/RedisChannelParserTests.java @@ -65,7 +65,6 @@ public class RedisChannelParserTests extends RedisAvailableTests{ redisChannel.send(m); Thread.sleep(1000); Mockito.verify(marker, Mockito.times(1)).mark(); - System.out.println("done"); } interface Marker { 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 cd13613228..2faaf2850a 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 @@ -15,6 +15,15 @@ */ package org.springframework.integration.redis.store; +import java.util.ArrayList; +import java.util.Iterator; +import java.util.List; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.TimeUnit; + +import junit.framework.AssertionFailedError; + import org.junit.Test; import org.springframework.context.support.ClassPathXmlApplicationContext; import org.springframework.data.redis.connection.jedis.JedisConnectionFactory; @@ -25,6 +34,7 @@ import org.springframework.integration.message.GenericMessage; import org.springframework.integration.redis.rules.RedisAvailable; import org.springframework.integration.redis.rules.RedisAvailableTests; import org.springframework.integration.store.MessageGroup; +import org.springframework.integration.store.SimpleMessageGroup; import org.springframework.integration.support.MessageBuilder; import static org.junit.Assert.assertEquals; @@ -41,53 +51,57 @@ public class RedisMessageGroupStoreTests extends RedisAvailableTests { @Test @RedisAvailable - public void testNonExistingEmptyMessageGroup(){ + public void testNonExistingEmptyMessageGroup() throws Exception{ JedisConnectionFactory jcf = this.getConnectionFactoryForTest(); RedisMessageStore store = new RedisMessageStore(jcf); MessageGroup messageGroup = store.getMessageGroup(1); assertNotNull(messageGroup); - assertTrue(messageGroup instanceof RedisMessageGroup); + assertTrue(messageGroup instanceof SimpleMessageGroup); assertEquals(0, messageGroup.size()); } @Test @RedisAvailable - public void testMessageGroupWithAddedMessageViaMessageGroup(){ + public void testMessageGroupWithAddedMessage() throws Exception{ JedisConnectionFactory jcf = this.getConnectionFactoryForTest(); RedisMessageStore store = new RedisMessageStore(jcf); MessageGroup messageGroup = store.getMessageGroup(1); Message message = new GenericMessage("Hello"); - ((RedisMessageGroup)messageGroup).add(message); + messageGroup = store.addMessageToGroup(1, message); + assertEquals(1, messageGroup.size()); + + // make sure the store is properly rebuild from Redis + store = new RedisMessageStore(jcf); + + messageGroup = store.getMessageGroup(1); assertEquals(1, messageGroup.size()); } @Test @RedisAvailable - public void testMessageGroupWithAddedMessageViaMessageGroupStore(){ + public void testRemoveMessageGroup() throws Exception{ JedisConnectionFactory jcf = this.getConnectionFactoryForTest(); RedisMessageStore store = new RedisMessageStore(jcf); MessageGroup messageGroup = store.getMessageGroup(1); Message message = new GenericMessage("Hello"); - store.addMessageToGroup(1, message); - assertEquals(1, messageGroup.size()); - } - - @Test - @RedisAvailable - public void testRemoveMessageGroup(){ - JedisConnectionFactory jcf = this.getConnectionFactoryForTest(); - RedisMessageStore store = new RedisMessageStore(jcf); - - MessageGroup messageGroup = store.getMessageGroup(1); - Message message = new GenericMessage("Hello"); - ((RedisMessageGroup)messageGroup).add(message); + messageGroup = store.addMessageToGroup(messageGroup.getGroupId(), message); assertEquals(1, messageGroup.size()); store.removeMessageGroup(1); + MessageGroup messageGroupA = store.getMessageGroup(1); + assertNotSame(messageGroup, messageGroupA); + assertEquals(0, messageGroupA.getMarked().size()); + assertEquals(0, messageGroupA.getUnmarked().size()); + assertEquals(0, messageGroupA.size()); + + // make sure the store is properly rebuild from Redis + store = new RedisMessageStore(jcf); + messageGroup = store.getMessageGroup(1); + assertEquals(0, messageGroup.getMarked().size()); assertEquals(0, messageGroup.getUnmarked().size()); assertEquals(0, messageGroup.size()); @@ -95,136 +109,177 @@ public class RedisMessageGroupStoreTests extends RedisAvailableTests { @Test @RedisAvailable - public void testRemoveMessageFromTheGroup(){ + public void testRemoveMessageFromTheGroup() throws Exception{ JedisConnectionFactory jcf = this.getConnectionFactoryForTest(); RedisMessageStore store = new RedisMessageStore(jcf); MessageGroup messageGroup = store.getMessageGroup(1); Message message = new GenericMessage("2"); - ((RedisMessageGroup)messageGroup).add(new GenericMessage("1")); - ((RedisMessageGroup)messageGroup).add(message); - ((RedisMessageGroup)messageGroup).add(new GenericMessage("3")); + store.addMessageToGroup(messageGroup.getGroupId(), new GenericMessage("1")); + store.addMessageToGroup(messageGroup.getGroupId(), message); + messageGroup = store.addMessageToGroup(messageGroup.getGroupId(), new GenericMessage("3")); assertEquals(3, messageGroup.size()); - store.removeMessageFromGroup(1, message); + messageGroup = store.removeMessageFromGroup(1, message); assertEquals(2, messageGroup.size()); + + // make sure the store is properly rebuild from Redis + store = new RedisMessageStore(jcf); + + messageGroup = store.getMessageGroup(1); + assertEquals(2, messageGroup.size()); + } @Test @RedisAvailable - public void testMarkAllMessagesInMessageGroup(){ + public void testMarkAllMessagesInMessageGroup() throws Exception{ JedisConnectionFactory jcf = this.getConnectionFactoryForTest(); RedisMessageStore store = new RedisMessageStore(jcf); MessageGroup messageGroup = store.getMessageGroup(1); - ((RedisMessageGroup)messageGroup).add(new GenericMessage("1")); - ((RedisMessageGroup)messageGroup).add(new GenericMessage("2")); - ((RedisMessageGroup)messageGroup).add(new GenericMessage("3")); + store.addMessageToGroup(messageGroup.getGroupId(), new GenericMessage("1")); + store.addMessageToGroup(messageGroup.getGroupId(), new GenericMessage("2")); + messageGroup = store.addMessageToGroup(messageGroup.getGroupId(), new GenericMessage("3")); + assertEquals(3, messageGroup.getUnmarked().size()); assertEquals(0, messageGroup.getMarked().size()); - ((RedisMessageGroup)messageGroup).markAll(); + messageGroup = store.markMessageGroup(messageGroup); + + assertEquals(0, messageGroup.getUnmarked().size()); + assertEquals(3, messageGroup.getMarked().size()); + + // make sure the store is properly rebuild from Redis + store = new RedisMessageStore(jcf); + + messageGroup = store.getMessageGroup(1); assertEquals(0, messageGroup.getUnmarked().size()); assertEquals(3, messageGroup.getMarked().size()); } @Test @RedisAvailable - public void testMarkMessageInMessageGroup(){ + public void testMarkMessageInMessageGroup() throws Exception{ JedisConnectionFactory jcf = this.getConnectionFactoryForTest(); RedisMessageStore store = new RedisMessageStore(jcf); MessageGroup messageGroup = store.getMessageGroup(1); Message messageToMark = new GenericMessage("1"); - ((RedisMessageGroup)messageGroup).add(messageToMark); - ((RedisMessageGroup)messageGroup).add(new GenericMessage("2")); - ((RedisMessageGroup)messageGroup).add(new GenericMessage("3")); + store.addMessageToGroup(messageGroup.getGroupId(), messageToMark); + store.addMessageToGroup(messageGroup.getGroupId(), new GenericMessage("2")); + messageGroup = store.addMessageToGroup(messageGroup.getGroupId(), new GenericMessage("3")); + assertEquals(3, messageGroup.getUnmarked().size()); assertEquals(0, messageGroup.getMarked().size()); - store.markMessageFromGroup(1, messageToMark); + messageGroup = store.markMessageFromGroup(1, messageToMark); + assertEquals(2, messageGroup.getUnmarked().size()); + assertEquals(1, messageGroup.getMarked().size()); + + // make sure the store is properly rebuild from Redis + store = new RedisMessageStore(jcf); + + messageGroup = store.getMessageGroup(1); assertEquals(2, messageGroup.getUnmarked().size()); assertEquals(1, messageGroup.getMarked().size()); } @Test @RedisAvailable - public void testGetOneFromMessageGroup(){ + public void testMultipleInstancesOfGroupStore() throws Exception{ JedisConnectionFactory jcf = this.getConnectionFactoryForTest(); - RedisMessageStore store = new RedisMessageStore(jcf); - - MessageGroup messageGroup = store.getMessageGroup(1); - ((RedisMessageGroup)messageGroup).add(new GenericMessage("1")); - ((RedisMessageGroup)messageGroup).add(new GenericMessage("2")); - ((RedisMessageGroup)messageGroup).add(new GenericMessage("3")); - Message message = messageGroup.getOne(); - assertEquals("1", message.getPayload()); + RedisMessageStore store1 = new RedisMessageStore(jcf); + + RedisMessageStore store2 = new RedisMessageStore(jcf); + + Message message = new GenericMessage("1"); + store1.addMessageToGroup(1, message); + MessageGroup messageGroup = store2.addMessageToGroup(1, new GenericMessage("2")); + + assertEquals(2, messageGroup.getUnmarked().size()); + assertEquals(0, messageGroup.getMarked().size()); + + RedisMessageStore store3 = new RedisMessageStore(jcf); + + messageGroup = store3.markMessageFromGroup(1, message); + + assertEquals(1, messageGroup.getUnmarked().size()); + assertEquals(1, messageGroup.getMarked().size()); } @Test @RedisAvailable - public void testAddToMessageGroup(){ + public void testIteratorOfMessageGroups() throws Exception{ JedisConnectionFactory jcf = this.getConnectionFactoryForTest(); - RedisMessageStore store = new RedisMessageStore(jcf); - MessageGroup messageGroup = store.getMessageGroup(1); + RedisMessageStore store1 = new RedisMessageStore(jcf); + RedisMessageStore store2 = new RedisMessageStore(jcf); - store.addMessageToGroup(1, new GenericMessage("1")); - assertEquals(1, messageGroup.size()); - store.addMessageToGroup(1, new GenericMessage("2")); - assertEquals(2, messageGroup.size()); - store.addMessageToGroup(1, new GenericMessage("3")); - assertEquals(3, messageGroup.size()); + + store1.addMessageToGroup(1, new GenericMessage("1")); + store2.addMessageToGroup(2, new GenericMessage("2")); + store1.addMessageToGroup(3, new GenericMessage("3")); + store2.addMessageToGroup(3, new GenericMessage("3A")); + + Iterator messageGroups = store1.iterator(); + int counter = 0; + while (messageGroups.hasNext()) { + messageGroups.next(); + counter++; + } + assertEquals(3, counter); + + store2.removeMessageGroup(3); + + messageGroups = store1.iterator(); + counter = 0; + while (messageGroups.hasNext()) { + messageGroups.next(); + counter++; + } + assertEquals(2, counter); } @Test @RedisAvailable - public void testCanAddToMessageGroup(){ + public void testConcurrentModifications() throws Exception{ JedisConnectionFactory jcf = this.getConnectionFactoryForTest(); - RedisMessageStore store = new RedisMessageStore(jcf); + final RedisMessageStore store1 = new RedisMessageStore(jcf); + final RedisMessageStore store2 = new RedisMessageStore(jcf); - - MessageGroup messageGroup = store.getMessageGroup(1); - Message m1 = MessageBuilder.withPayload("1").setSequenceNumber(1).setSequenceSize(3).setCorrelationId(1).build(); - Message m2 = MessageBuilder.withPayload("2").setSequenceNumber(2).setSequenceSize(3).setCorrelationId(1).build(); - Message m3 = MessageBuilder.withPayload("3").setSequenceNumber(3).setSequenceSize(3).setCorrelationId(1).build(); - - if (messageGroup.canAdd(m1)){ - store.addMessageToGroup(1, m1); - } - assertEquals(1, messageGroup.size()); - - if (messageGroup.canAdd(m2)){ - store.addMessageToGroup(1, m2); - } - assertEquals(2, messageGroup.size()); - - if (messageGroup.canAdd(m3)){ - store.addMessageToGroup(1, m3); - } - assertEquals(3, messageGroup.size()); - - if (messageGroup.canAdd(m3)){ - store.addMessageToGroup(1, m3); - } - assertEquals(3, messageGroup.size()); - } - - @Test - @RedisAvailable - public void testSameInstance(){ - JedisConnectionFactory jcf = this.getConnectionFactoryForTest(); - RedisMessageStore store = new RedisMessageStore(jcf); + final Message message = new GenericMessage("1"); - MessageGroup mg1 = store.getMessageGroup(1); - MessageGroup mg2 = store.getMessageGroup(1); + ExecutorService executor = null; - assertEquals(mg1, mg2); + final List failures = new ArrayList(); - store.removeMessageGroup(1); - - mg2 = store.getMessageGroup(1); - assertNotSame(mg1, mg2); + for (int i = 0; i < 100; i++) { + executor = Executors.newCachedThreadPool(); + + executor.execute(new Runnable() { + public void run() { + MessageGroup group = store1.addMessageToGroup(1, message); + if (group.getUnmarked().size() != 1){ + failures.add("ADD"); + throw new AssertionFailedError("Failed on ADD"); + } + } + }); + executor.execute(new Runnable() { + public void run() { + MessageGroup group = store2.removeMessageFromGroup(1, message); + if (group.getUnmarked().size() != 0){ + failures.add("REMOVE"); + throw new AssertionFailedError("Failed on Remove"); + } + } + }); + + 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 + } + assertTrue(failures.size() == 0); } - @Test @RedisAvailable public void testWithAggregatorWithShutdown(){ diff --git a/spring-integration-redis/src/test/java/org/springframework/integration/redis/store/RedisMessageStoreTests.java b/spring-integration-redis/src/test/java/org/springframework/integration/redis/store/RedisMessageStoreTests.java index 196c53f4b6..d74ab0e569 100644 --- a/spring-integration-redis/src/test/java/org/springframework/integration/redis/store/RedisMessageStoreTests.java +++ b/spring-integration-redis/src/test/java/org/springframework/integration/redis/store/RedisMessageStoreTests.java @@ -116,14 +116,6 @@ public class RedisMessageStoreTests extends RedisAvailableTests { assertNull(store.getMessage(stringMessage.getHeaders().getId())); } - @Test(expected=IllegalArgumentException.class) - @RedisAvailable - public void testRemoveNonExistingMessage(){ - JedisConnectionFactory jcf = this.getConnectionFactoryForTest(); - RedisMessageStore store = new RedisMessageStore(jcf); - store.removeMessage(UUID.randomUUID()); - } - @SuppressWarnings("serial") public static class Person implements Serializable{ private Address address;