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 e076db1d52..1ab8f514a0 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 @@ -1,5 +1,5 @@ /* - * Copyright 2002-2010 the original author or authors. + * 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 @@ -18,7 +18,6 @@ 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.RedisTemplate; import org.springframework.integration.Message; import org.springframework.integration.store.MessageGroup; @@ -33,9 +32,14 @@ class RedisMessageGroup implements MessageGroup { private final String MARKED_PREFIX = "MARKED_"; private final String UNMARKED_PREFIX = "UNMARKED_"; + 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; @@ -44,75 +48,16 @@ class RedisMessageGroup implements MessageGroup { } public boolean canAdd(Message message) { - // TODO Auto-generated method stub - return false; + return !isMember(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. - */ - 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()); - } - /** - * - */ - public void destroy(){ - // delete unmarked - BoundSetOperations mGroupOps = this.redisTemplate.boundSetOps(UNMARKED_PREFIX + this.groupId.toString()); - for (Object messageId : mGroupOps.members()) { - this.messageStore.removeMessage(UUID.fromString(messageId.toString())); - //this.redisTemplate.delete(messageId.toString()); - } - this.redisTemplate.delete(UNMARKED_PREFIX + this.groupId.toString()); - // delete marked - mGroupOps = this.redisTemplate.boundSetOps(MARKED_PREFIX + this.groupId.toString()); - for (Object messageId : mGroupOps.members()) { - this.messageStore.removeMessage(UUID.fromString(messageId.toString())); - } - this.redisTemplate.delete(MARKED_PREFIX + this.groupId.toString()); - } - - public void markAll() { - BoundSetOperations unmarkedGroupOps = this.redisTemplate.boundSetOps(UNMARKED_PREFIX + this.groupId.toString()); - long uSize = unmarkedGroupOps.size(); - 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"); - } - - 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); - } - public Collection> getUnmarked() { - BoundSetOperations mGroupOps = this.redisTemplate.boundSetOps(UNMARKED_PREFIX + this.groupId.toString()); - return this.getMessages(mGroupOps); + return this.unmarked; } public Collection> getMarked() { - BoundSetOperations mGroupOps = this.redisTemplate.boundSetOps(MARKED_PREFIX + this.groupId.toString()); - return this.getMessages(mGroupOps); + return this.marked; } @@ -120,61 +65,181 @@ class RedisMessageGroup implements MessageGroup { return this.groupId; } - /* (non-Javadoc) - * @see org.springframework.integration.store.MessageGroup#isComplete() - */ public boolean isComplete() { - // TODO Auto-generated method stub - return false; + if (size() == 0) { + return true; + } + int sequenceSize = getSequenceSize(); + // If there is no sequence then it must be incomplete.... + return sequenceSize > 0 && sequenceSize == size(); } - /* (non-Javadoc) - * @see org.springframework.integration.store.MessageGroup#getSequenceSize() - */ public int getSequenceSize() { - // TODO Auto-generated method stub - return 0; + if (size() == 0) { + return 0; + } + return getOne().getHeaders().getSequenceSize(); } - public int size() { - BoundSetOperations mGroupOps = this.redisTemplate.boundSetOps(UNMARKED_PREFIX + this.groupId.toString()); - return mGroupOps.members().size(); + BoundListOperations mGroupOps = this.redisTemplate.boundListOps(UNMARKED_PREFIX + this.groupId.toString()); + long usize = mGroupOps.size(); + mGroupOps = this.redisTemplate.boundListOps(MARKED_PREFIX + this.groupId.toString()); + long msize = mGroupOps.size(); + return (int)(usize + msize); } - public Message getOne() { - 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; + Message one = unmarked.get(0); + if (one == null) { + one = marked.get(0); + } + return one; } - /* (non-Javadoc) - * @see org.springframework.integration.store.MessageGroup#getTimestamp() - */ public long getTimestamp() { - // TODO Auto-generated method stub - return 0; + return this.timestamp; } - private Collection> getMessages(BoundSetOperations mGroupOps){ + protected void markAll() { + BoundListOperations unmarkedOps = this.redisTemplate.boundListOps(UNMARKED_PREFIX + this.groupId.toString()); + BoundListOperations markedOps = this.redisTemplate.boundListOps(MARKED_PREFIX + this.groupId.toString()); + long uSize = unmarkedOps.size(); + String destiinationKey = MARKED_PREFIX + this.groupId.toString(); + + synchronized (lock) { + unmarkedOps.rename(destiinationKey); + 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(UNMARKED_PREFIX + this.groupId.toString()); + List messageIds = unmarkedOps.range(0, unmarkedOps.size()-1); + + synchronized (lock) { + for (Object id : messageIds) { + if (messageId.equals(id)){ + BoundListOperations markedOps = this.redisTemplate.boundListOps(MARKED_PREFIX + this.groupId.toString()); + markedOps.rightPush(id); + System.out.println(unmarkedOps.size()); + unmarkedOps.remove(0, id); + System.out.println(unmarkedOps.size()); + 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(UNMARKED_PREFIX + this.groupId.toString()); + 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(UNMARKED_PREFIX + this.groupId.toString()); + BoundListOperations markedOps = this.redisTemplate.boundListOps(MARKED_PREFIX + this.groupId.toString()); + + synchronized (lock) { + unmarkedOps.remove(0, messageId.toString()); + markedOps.remove(0, messageId.toString()); + this.messageStore.removeMessage(messageId); + this.rebuildLocalCache(unmarkedOps, markedOps); + } + + } + /** + * + */ + protected void destroy(){ + BoundListOperations mGroupListOps = this.redisTemplate.boundListOps(UNMARKED_PREFIX + this.groupId.toString()); + List messageIds = mGroupListOps.range(0, mGroupListOps.size()-1); + for (Object messageId : messageIds) { + this.messageStore.removeMessage(UUID.fromString(messageId.toString())); + } + this.redisTemplate.delete(UNMARKED_PREFIX + this.groupId.toString()); + + mGroupListOps = this.redisTemplate.boundListOps(MARKED_PREFIX + this.groupId.toString()); + messageIds = mGroupListOps.range(0, mGroupListOps.size()-1); + for (Object messageId : messageIds) { + this.messageStore.removeMessage(UUID.fromString(messageId.toString())); + } + this.redisTemplate.delete(MARKED_PREFIX + this.groupId.toString()); + } + + private Collection> buildMessageList(BoundListOperations mGroupOps){ List> messages = new LinkedList>(); - for (Object messageId : mGroupOps.members()) { + 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(UNMARKED_PREFIX + this.groupId.toString()); + Collection> unmarked = this.buildMessageList(mGroupOps); + mGroupOps = this.redisTemplate.boundListOps(MARKED_PREFIX + this.groupId.toString()); + 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 da75e1644f..09c756b07f 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 @@ -16,8 +16,10 @@ package org.springframework.integration.redis.store; import java.util.ArrayList; +import java.util.HashMap; import java.util.Iterator; import java.util.List; +import java.util.Map; import java.util.UUID; import org.springframework.beans.factory.InitializingBean; @@ -49,8 +51,12 @@ public class RedisMessageStore extends AbstractMessageGroupStore implements Mess 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){ this.redisTemplate = new RedisTemplate(); this.redisTemplate.setConnectionFactory(connectionFactory); @@ -144,12 +150,19 @@ public class RedisMessageStore extends AbstractMessageGroupStore implements Mess */ public MessageGroup getMessageGroup(Object groupId) { Assert.notNull(groupId, "'groupId' must not be null"); - - BoundSetOperations mGroupsOps = this.redisTemplate.boundSetOps(MESSAGE_GROUPS_KEY); - if (!mGroupsOps.isMember(groupId)){ - mGroupsOps.add(groupId); + 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 new RedisMessageGroup(this, this.redisTemplate, groupId); + + return group; } /** * @@ -188,8 +201,13 @@ public class RedisMessageStore extends AbstractMessageGroupStore implements Mess public void removeMessageGroup(Object groupId) { Assert.notNull(groupId, "'groupId' must not be null"); RedisMessageGroup messageGroup = (RedisMessageGroup) this.getMessageGroup(groupId); - messageGroup.destroy(); - this.redisTemplate.delete(groupId.toString()); + synchronized (messageGroup) { + messageGroup.destroy(); + BoundSetOperations mGroupsOps = this.redisTemplate.boundSetOps(MESSAGE_GROUPS_KEY); + mGroupsOps.remove(groupId); + this.redisTemplate.delete(groupId.toString()); + this.messageGroups.remove(groupId); + } } @Override 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 00c7fd24f7..3dcea5a630 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 @@ -22,9 +22,11 @@ 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.support.MessageBuilder; import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertNotNull; +import static org.junit.Assert.assertNotSame; import static org.junit.Assert.assertTrue; /** @@ -85,6 +87,23 @@ public class RedisMessageGroupStoreTests extends RedisAvailableTests { assertEquals(0, messageGroup.size()); } + @Test + @RedisAvailable + public void testRemoveMessageFromTheGroup(){ + 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")); + assertEquals(3, messageGroup.size()); + + store.removeMessageFromGroup(1, message); + assertEquals(2, messageGroup.size()); + } + @Test @RedisAvailable public void testMarkAllMessagesInMessageGroup(){ @@ -134,4 +153,69 @@ public class RedisMessageGroupStoreTests extends RedisAvailableTests { assertEquals("1", message.getPayload()); } + @Test + @RedisAvailable + public void testAddToMessageGroup(){ + JedisConnectionFactory jcf = this.getConnectionFactoryForTest(); + RedisMessageStore store = new RedisMessageStore(jcf); + MessageGroup messageGroup = store.getMessageGroup(1); + + 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()); + } + + @Test + @RedisAvailable + public void testCanAddToMessageGroup(){ + JedisConnectionFactory jcf = this.getConnectionFactoryForTest(); + RedisMessageStore store = 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); + + MessageGroup mg1 = store.getMessageGroup(1); + MessageGroup mg2 = store.getMessageGroup(1); + + assertEquals(mg1, mg2); + + store.removeMessageGroup(1); + + mg2 = store.getMessageGroup(1); + assertNotSame(mg1, mg2); + } + }