From 94dea0855df6278146bb7aa2b81798410a5126fc Mon Sep 17 00:00:00 2001 From: Oleg Zhurakousky Date: Thu, 11 Aug 2011 09:53:44 -0400 Subject: [PATCH] INT-2030 finished with initial implementation fo Redis-based MessageGroupStore --- .../redis/store/RedisMessageGroup.java | 4 ++- .../redis/store/RedisMessageStore.java | 4 +-- .../store/RedisMessageGroupStoreTests.java | 30 +++++++++++++++++++ .../redis/store/redis-aggregator-config.xml | 23 ++++++++++++++ 4 files changed, 58 insertions(+), 3 deletions(-) create mode 100644 spring-integration-redis/src/test/java/org/springframework/integration/redis/store/redis-aggregator-config.xml 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 1ab8f514a0..e00baac1bc 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 @@ -45,6 +45,9 @@ class RedisMessageGroup implements MessageGroup { this.groupId = groupId; this.redisTemplate = redisTemplate; this.messageStore = messageStore; + BoundListOperations unmarkedOps = this.redisTemplate.boundListOps(UNMARKED_PREFIX + this.groupId.toString()); + BoundListOperations markedOps = this.redisTemplate.boundListOps(MARKED_PREFIX + this.groupId.toString()); + this.rebuildLocalCache(unmarkedOps, markedOps); } public boolean canAdd(Message message) { @@ -231,7 +234,6 @@ class RedisMessageGroup implements MessageGroup { private void rebuildLocalCache(BoundListOperations unmarkedOps, BoundListOperations markedOps){ - if (unmarkedOps != null){ this.unmarked.clear(); this.unmarked.addAll(this.buildMessageList(unmarkedOps)); 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 09c756b07f..1fb96e05a0 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 @@ -156,9 +156,9 @@ public class RedisMessageStore extends AbstractMessageGroupStore implements Mess 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); } + group = new RedisMessageGroup(this, this.redisTemplate, groupId); + this.messageGroups.put(groupId, group); } } 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 3dcea5a630..480ad2e2ce 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 @@ -16,8 +16,11 @@ package org.springframework.integration.redis.store; import org.junit.Test; +import org.springframework.context.support.ClassPathXmlApplicationContext; import org.springframework.data.redis.connection.jedis.JedisConnectionFactory; import org.springframework.integration.Message; +import org.springframework.integration.MessageChannel; +import org.springframework.integration.channel.QueueChannel; import org.springframework.integration.message.GenericMessage; import org.springframework.integration.redis.rules.RedisAvailable; import org.springframework.integration.redis.rules.RedisAvailableTests; @@ -27,6 +30,7 @@ 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.assertNull; import static org.junit.Assert.assertTrue; /** @@ -218,4 +222,30 @@ public class RedisMessageGroupStoreTests extends RedisAvailableTests { assertNotSame(mg1, mg2); } + @Test + @RedisAvailable + public void testWithAggregatorWithShutdown(){ + this.getConnectionFactoryForTest(); // for this test it only ensures that DB was flushed before test + + ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext("redis-aggregator-config.xml", this.getClass()); + MessageChannel input = context.getBean("inputChannel", MessageChannel.class); + QueueChannel output = context.getBean("outputChannel", QueueChannel.class); + + Message m1 = MessageBuilder.withPayload("1").setSequenceNumber(1).setSequenceSize(3).setCorrelationId(1).build(); + Message m2 = MessageBuilder.withPayload("2").setSequenceNumber(2).setSequenceSize(3).setCorrelationId(1).build(); + input.send(m1); + assertNull(output.receive(1000)); + input.send(m2); + assertNull(output.receive(1000)); + context.close(); + + context = new ClassPathXmlApplicationContext("redis-aggregator-config.xml", this.getClass()); + input = context.getBean("inputChannel", MessageChannel.class); + output = context.getBean("outputChannel", QueueChannel.class); + + Message m3 = MessageBuilder.withPayload("3").setSequenceNumber(3).setSequenceSize(3).setCorrelationId(1).build(); + input.send(m3); + assertNotNull(output.receive(1000)); + } + } diff --git a/spring-integration-redis/src/test/java/org/springframework/integration/redis/store/redis-aggregator-config.xml b/spring-integration-redis/src/test/java/org/springframework/integration/redis/store/redis-aggregator-config.xml new file mode 100644 index 0000000000..6e1e9fc039 --- /dev/null +++ b/spring-integration-redis/src/test/java/org/springframework/integration/redis/store/redis-aggregator-config.xml @@ -0,0 +1,23 @@ + + + + + + + + + + + + + + + + + + +