Spring/spring-integration
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

INT-2030 finished with initial implementation fo Redis-based MessageGroupStore

Browse Source
This commit is contained in:
Oleg Zhurakousky
2011-08-11 09:53:44 -04:00
parent 28bcb4bfcf
commit 94dea0855d
4 changed files with 58 additions and 3 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
spring-integration-redis/src/main/java/org/springframework/integration/redis/store/RedisMessageGroup.java
View File

@@ -45,6 +45,9 @@ class RedisMessageGroup implements MessageGroup {
this.groupId = groupId;
this.redisTemplate = redisTemplate;
this.messageStore = messageStore;
BoundListOperations<String, Object> unmarkedOps = this.redisTemplate.boundListOps(UNMARKED_PREFIX + this.groupId.toString());
BoundListOperations<String, Object> 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<String, Object> unmarkedOps, BoundListOperations<String, Object> markedOps){
if (unmarkedOps != null){
this.unmarked.clear();
this.unmarked.addAll(this.buildMessageList(unmarkedOps));

4
spring-integration-redis/src/main/java/org/springframework/integration/redis/store/RedisMessageStore.java
View File

@@ -156,9 +156,9 @@ public class RedisMessageStore extends AbstractMessageGroupStore implements Mess
BoundSetOperations<String, Object> 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);
}
}

30
spring-integration-redis/src/test/java/org/springframework/integration/redis/store/RedisMessageGroupStoreTests.java
View File

@@ -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));
}
}

23
spring-integration-redis/src/test/java/org/springframework/integration/redis/store/redis-aggregator-config.xml Normal file
View File

@@ -0,0 +1,23 @@
<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xmlns:int="http://www.springframework.org/schema/integration"
xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd
http://www.springframework.org/schema/integration http://www.springframework.org/schema/integration/spring-integration-2.0.xsd">
<int:aggregator input-channel="inputChannel" output-channel="outputChannel" message-store="redisStore"/>
<int:channel id="outputChannel">
<int:queue/>
</int:channel>
<bean id="redisStore" class="org.springframework.integration.redis.store.RedisMessageStore">
<constructor-arg ref="redisConnectionFactory"/>
</bean>
<bean id="redisConnectionFactory" class="org.springframework.data.redis.connection.jedis.JedisConnectionFactory">
<property name="port" value="7379" />
</bean>
</beans>