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;