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

INT-2030 added support for marking individual messages in the group, polishing of existing code, added tests

Browse Source
This commit is contained in:
Oleg Zhurakousky
2011-08-10 15:16:36 -04:00
parent f1e2434398
commit c6e90e5a0f
3 changed files with 67 additions and 29 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -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<String, Object> mGroupListOps = this.redisTemplate.boundListOps("L" + UNMARKED_PREFIX + this.groupId.toString());
BoundSetOperations<String, Object> 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<String, Object> 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<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());
String destiinationKey = MARKED_PREFIX + this.groupId.toString();
unmarkedGroupOps.rename(destiinationKey);
BoundSetOperations<String, Object> 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<String, Object> 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<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;
BoundListOperations<String, Object> mGroupListOps = this.redisTemplate.boundListOps("L" + UNMARKED_PREFIX + this.groupId.toString());
Object id = mGroupListOps.leftPop();
return this.messageStore.getMessage(UUID.fromString(id.toString()));
// 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;
}
/* (non-Javadoc)

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

@@ -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) {

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

@@ -95,9 +95,29 @@ public class RedisMessageGroupStoreTests extends RedisAvailableTests {
((RedisMessageGroup)messageGroup).add(new GenericMessage<String>("1"));
((RedisMessageGroup)messageGroup).add(new GenericMessage<String>("2"));
((RedisMessageGroup)messageGroup).add(new GenericMessage<String>("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<String>("1");
((RedisMessageGroup)messageGroup).add(messageToMark);
((RedisMessageGroup)messageGroup).add(new GenericMessage<String>("2"));
((RedisMessageGroup)messageGroup).add(new GenericMessage<String>("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