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 index 8f05d01026..e076db1d52 100644 --- 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 @@ -17,8 +17,8 @@ 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.BoundSetOperations; -import org.springframework.data.redis.core.BoundValueOperations; import org.springframework.data.redis.core.RedisTemplate; import org.springframework.integration.Message; import org.springframework.integration.store.MessageGroup; @@ -48,15 +48,25 @@ class RedisMessageGroup implements MessageGroup { return false; } /** - * Will add message to this MessageGroup, b - * @param message + * 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. */ - public void add(Message message) { + protected void add(Message message) { + BoundListOperations mGroupListOps = this.redisTemplate.boundListOps("L" + UNMARKED_PREFIX + this.groupId.toString()); BoundSetOperations mGroupOps = this.redisTemplate.boundSetOps(UNMARKED_PREFIX + this.groupId.toString()); String key = message.getHeaders().getId().toString(); mGroupOps.add(key); + mGroupListOps.rightPush(key); + System.out.println("Members: " + mGroupOps.members()); this.messageStore.addMessage(message); } + + protected void remove(Message message) { + BoundSetOperations mGroupOps = this.redisTemplate.boundSetOps(UNMARKED_PREFIX + this.groupId.toString()); + String key = message.getHeaders().getId().toString(); + mGroupOps.remove(key); + this.messageStore.removeMessage(message.getHeaders().getId()); + } /** * */ @@ -71,7 +81,6 @@ class RedisMessageGroup implements MessageGroup { // delete marked mGroupOps = this.redisTemplate.boundSetOps(MARKED_PREFIX + this.groupId.toString()); for (Object messageId : mGroupOps.members()) { - //this.redisTemplate.delete(messageId.toString()); this.messageStore.removeMessage(UUID.fromString(messageId.toString())); } this.redisTemplate.delete(MARKED_PREFIX + this.groupId.toString()); @@ -80,14 +89,18 @@ class RedisMessageGroup implements MessageGroup { public void markAll() { BoundSetOperations unmarkedGroupOps = this.redisTemplate.boundSetOps(UNMARKED_PREFIX + this.groupId.toString()); long uSize = unmarkedGroupOps.size(); - String destinationKey = UNMARKED_PREFIX + this.groupId.toString(); - - for (Object messageId : unmarkedGroupOps.members()) { - unmarkedGroupOps.move(destinationKey, messageId); - } - BoundSetOperations markedGroupOps = this.redisTemplate.boundSetOps(UNMARKED_PREFIX + this.groupId.toString()); + String destiinationKey = MARKED_PREFIX + this.groupId.toString(); + unmarkedGroupOps.rename(destiinationKey); + + BoundSetOperations markedGroupOps = this.redisTemplate.boundSetOps(destiinationKey); long mSize = markedGroupOps.size(); - Assert.isTrue(mSize == uSize, "Failed to mark ALl messages in the message group"); + Assert.isTrue(mSize == uSize, "Failed to mark All messages in the message group"); + } + + protected void markMessage(String messageId) { + BoundSetOperations unmarkedGroupOps = this.redisTemplate.boundSetOps(UNMARKED_PREFIX + this.groupId.toString()); + String destinationKey = MARKED_PREFIX + this.groupId.toString(); + unmarkedGroupOps.move(destinationKey, messageId); } @@ -131,17 +144,20 @@ class RedisMessageGroup implements MessageGroup { public Message getOne() { - List> messages = (List>) this.getUnmarked(); - if (messages.isEmpty()){ - messages = (List>) this.getMarked(); - if (!messages.isEmpty()){ - return messages.get(0); - } - } - else { - return messages.get(0); - } - return null; + BoundListOperations mGroupListOps = this.redisTemplate.boundListOps("L" + UNMARKED_PREFIX + this.groupId.toString()); + Object id = mGroupListOps.leftPop(); + return this.messageStore.getMessage(UUID.fromString(id.toString())); +// List> messages = (List>) this.getUnmarked(); +// if (messages.isEmpty()){ +// messages = (List>) this.getMarked(); +// if (!messages.isEmpty()){ +// return messages.get(0); +// } +// } +// else { +// return messages.get(0); +// } + //return null; } /* (non-Javadoc) 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 ca84b025bf..da75e1644f 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 @@ -173,14 +173,16 @@ public class RedisMessageStore extends AbstractMessageGroupStore implements Mess public MessageGroup removeMessageFromGroup(Object key, Message messageToRemove) { - // TODO Auto-generated method stub - return null; + RedisMessageGroup messageGroup = (RedisMessageGroup) this.getMessageGroup(key); + messageGroup.remove(messageToRemove); + return messageGroup; } public MessageGroup markMessageFromGroup(Object key, Message messageToMark) { - // TODO Auto-generated method stub - return null; + RedisMessageGroup messageGroup = (RedisMessageGroup) this.getMessageGroup(key); + messageGroup.markMessage(messageToMark.getHeaders().getId().toString()); + return messageGroup; } public void removeMessageGroup(Object groupId) { 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 45e53951ea..00c7fd24f7 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 @@ -95,9 +95,29 @@ public class RedisMessageGroupStoreTests extends RedisAvailableTests { ((RedisMessageGroup)messageGroup).add(new GenericMessage("1")); ((RedisMessageGroup)messageGroup).add(new GenericMessage("2")); ((RedisMessageGroup)messageGroup).add(new GenericMessage("3")); + assertEquals(3, messageGroup.getUnmarked().size()); + assertEquals(0, messageGroup.getMarked().size()); ((RedisMessageGroup)messageGroup).markAll(); - // No need to assert for anything here. - // Assertion is done within the method itself and the exception would be thrown if move did not succeed + assertEquals(0, messageGroup.getUnmarked().size()); + assertEquals(3, messageGroup.getMarked().size()); + } + + @Test + @RedisAvailable + public void testMarkMessageInMessageGroup(){ + 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")); + assertEquals(3, messageGroup.getUnmarked().size()); + assertEquals(0, messageGroup.getMarked().size()); + store.markMessageFromGroup(1, messageToMark); + assertEquals(2, messageGroup.getUnmarked().size()); + assertEquals(1, messageGroup.getMarked().size()); } @Test