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

INT-2030 added support for getOne() and markAll() to Redis MessageGroupStore

Browse Source
This commit is contained in:
Oleg Zhurakousky
2011-08-10 13:14:39 -04:00
parent 81f07f2c5c
commit 5388149b3a
2 changed files with 59 additions and 9 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -12,8 +12,8 @@
*/
package org.springframework.integration.redis.store;
import java.util.ArrayList;
import java.util.Collection;
import java.util.LinkedList;
import java.util.List;
import org.springframework.data.redis.core.BoundSetOperations;
@@ -43,7 +43,10 @@ class RedisMessageGroup implements MessageGroup {
// TODO Auto-generated method stub
return false;
}
/**
*
* @param message
*/
public void add(Message<?> message) {
BoundSetOperations<String, Object> mGroupOps = this.redisTemplate.boundSetOps(UNMARKED_PREFIX + this.groupId.toString());
String key = message.getHeaders().getId().toString();
@@ -51,7 +54,9 @@ class RedisMessageGroup implements MessageGroup {
BoundValueOperations<String, Object> bvOps = this.redisTemplate.boundValueOps(key);
bvOps.set(message);
}
/**
*
*/
public void destroy(){
// delete unmarked
BoundSetOperations<String, Object> mGroupOps = this.redisTemplate.boundSetOps(UNMARKED_PREFIX + this.groupId.toString());
@@ -68,7 +73,16 @@ class RedisMessageGroup implements MessageGroup {
}
public void markAll() {
BoundSetOperations<String, Object> 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<String, Object> markedGroupOps = this.redisTemplate.boundSetOps(UNMARKED_PREFIX + this.groupId.toString());
long mSize = markedGroupOps.size();
Assert.isTrue(mSize == uSize, "Failed to mark ALl messages in the message group");
}
@@ -110,11 +124,18 @@ class RedisMessageGroup implements MessageGroup {
return mGroupOps.members().size();
}
/* (non-Javadoc)
* @see org.springframework.integration.store.MessageGroup#getOne()
*/
public Message<?> getOne() {
// TODO Auto-generated method stub
List<Message<?>> messages = (List<Message<?>>) this.getUnmarked();
if (messages.isEmpty()){
messages = (List<Message<?>>) this.getMarked();
if (!messages.isEmpty()){
return messages.get(0);
}
}
else {
return messages.get(0);
}
return null;
}
@@ -127,7 +148,7 @@ class RedisMessageGroup implements MessageGroup {
}
private Collection<Message<?>> getMessages(BoundSetOperations<String, Object> mGroupOps){
List<Message<?>> messages = new ArrayList<Message<?>>();
List<Message<?>> messages = new LinkedList<Message<?>>();
for (Object messageId : mGroupOps.members()) {
BoundValueOperations<String, Object> bvOps = this.redisTemplate.boundValueOps(messageId.toString());
Object message = bvOps.get();

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

@@ -85,4 +85,33 @@ public class RedisMessageGroupStoreTests extends RedisAvailableTests {
assertEquals(0, messageGroup.size());
}
@Test
@RedisAvailable
public void testMarkAllMessagesInMessageGroup(){
JedisConnectionFactory jcf = this.getConnectionFactoryForTest();
RedisMessageStore store = new RedisMessageStore(jcf);
MessageGroup messageGroup = store.getMessageGroup(1);
((RedisMessageGroup)messageGroup).add(new GenericMessage<String>("1"));
((RedisMessageGroup)messageGroup).add(new GenericMessage<String>("2"));
((RedisMessageGroup)messageGroup).add(new GenericMessage<String>("3"));
((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
}
@Test
@RedisAvailable
public void testGetOneFromMessageGroup(){
JedisConnectionFactory jcf = this.getConnectionFactoryForTest();
RedisMessageStore store = new RedisMessageStore(jcf);
MessageGroup messageGroup = store.getMessageGroup(1);
((RedisMessageGroup)messageGroup).add(new GenericMessage<String>("1"));
((RedisMessageGroup)messageGroup).add(new GenericMessage<String>("2"));
((RedisMessageGroup)messageGroup).add(new GenericMessage<String>("3"));
Message<?> message = messageGroup.getOne();
assertEquals("1", message.getPayload());
}
}