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

Merge pull request #6 from olegz/INT-2030

Browse Source 
INT-2030 polished RedisMessageGroup
This commit is contained in:
Mark Fisher
2011-08-12 13:08:06 -04:00
parent a352f19514 699afbe7a1
commit 6d3eac8c49
2 changed files with 36 additions and 30 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -32,6 +32,9 @@ class RedisMessageGroup implements MessageGroup {
private final String MARKED_PREFIX = "MARKED_";
private final String UNMARKED_PREFIX = "UNMARKED_";
private final String unmarkedId;
private final String markedId;
private final List<Message<?>> unmarked = new LinkedList<Message<?>>();
private final List<Message<?>> marked = new LinkedList<Message<?>>();
@@ -45,8 +48,10 @@ 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.unmarkedId = UNMARKED_PREFIX + groupId;
this.markedId = MARKED_PREFIX + groupId;
BoundListOperations<String, Object> unmarkedOps = this.redisTemplate.boundListOps(unmarkedId);
BoundListOperations<String, Object> markedOps = this.redisTemplate.boundListOps(markedId);
this.rebuildLocalCache(unmarkedOps, markedOps);
}
@@ -85,19 +90,19 @@ class RedisMessageGroup implements MessageGroup {
}
public int size() {
BoundListOperations<String, Object> 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);
synchronized (lock) {
return marked.size() + unmarked.size();
}
}
public Message<?> getOne() {
Message<?> one = unmarked.get(0);
if (one == null) {
one = marked.get(0);
synchronized (lock) {
Message<?> one = unmarked.get(0);
if (one == null) {
one = marked.get(0);
}
return one;
}
return one;
}
public long getTimestamp() {
@@ -105,13 +110,12 @@ class RedisMessageGroup implements MessageGroup {
}
protected void markAll() {
BoundListOperations<String, Object> unmarkedOps = this.redisTemplate.boundListOps(UNMARKED_PREFIX + this.groupId.toString());
BoundListOperations<String, Object> markedOps = this.redisTemplate.boundListOps(MARKED_PREFIX + this.groupId.toString());
BoundListOperations<String, Object> unmarkedOps = this.redisTemplate.boundListOps(unmarkedId);
BoundListOperations<String, Object> markedOps = this.redisTemplate.boundListOps(markedId);
long uSize = unmarkedOps.size();
String destiinationKey = MARKED_PREFIX + this.groupId.toString();
synchronized (lock) {
unmarkedOps.rename(destiinationKey);
unmarkedOps.rename(markedId);
this.unmarked.clear();
this.rebuildLocalCache(null, markedOps);
}
@@ -121,17 +125,15 @@ class RedisMessageGroup implements MessageGroup {
}
protected void markMessage(String messageId) {
BoundListOperations<String, Object> unmarkedOps = this.redisTemplate.boundListOps(UNMARKED_PREFIX + this.groupId.toString());
BoundListOperations<String, Object> unmarkedOps = this.redisTemplate.boundListOps(unmarkedId);
List<Object> messageIds = unmarkedOps.range(0, unmarkedOps.size()-1);
synchronized (lock) {
for (Object id : messageIds) {
if (messageId.equals(id)){
BoundListOperations<String, Object> markedOps = this.redisTemplate.boundListOps(MARKED_PREFIX + this.groupId.toString());
BoundListOperations<String, Object> markedOps = this.redisTemplate.boundListOps(markedId);
markedOps.rightPush(id);
System.out.println(unmarkedOps.size());
unmarkedOps.remove(0, id);
System.out.println(unmarkedOps.size());
this.rebuildLocalCache(unmarkedOps, markedOps);
return;
}
@@ -145,7 +147,7 @@ class RedisMessageGroup implements MessageGroup {
*/
protected void add(Message<?> message) {
String messageId = message.getHeaders().getId().toString();
BoundListOperations<String, Object> unmarkedOps = this.redisTemplate.boundListOps(UNMARKED_PREFIX + this.groupId.toString());
BoundListOperations<String, Object> unmarkedOps = this.redisTemplate.boundListOps(unmarkedId);
synchronized (lock) {
unmarkedOps.rightPush(messageId);
this.messageStore.addMessage(message);
@@ -155,8 +157,8 @@ class RedisMessageGroup implements MessageGroup {
protected void remove(Message<?> message) {
UUID messageId = message.getHeaders().getId();
BoundListOperations<String, Object> unmarkedOps = this.redisTemplate.boundListOps(UNMARKED_PREFIX + this.groupId.toString());
BoundListOperations<String, Object> markedOps = this.redisTemplate.boundListOps(MARKED_PREFIX + this.groupId.toString());
BoundListOperations<String, Object> unmarkedOps = this.redisTemplate.boundListOps(unmarkedId);
BoundListOperations<String, Object> markedOps = this.redisTemplate.boundListOps(markedId);
synchronized (lock) {
unmarkedOps.remove(0, messageId.toString());
@@ -170,19 +172,20 @@ class RedisMessageGroup implements MessageGroup {
*
*/
protected void destroy(){
BoundListOperations<String, Object> mGroupListOps = this.redisTemplate.boundListOps(UNMARKED_PREFIX + this.groupId.toString());
List<Object> messageIds = mGroupListOps.range(0, mGroupListOps.size()-1);
BoundListOperations<String, Object> unmarkedOps = this.redisTemplate.boundListOps(unmarkedId);
List<Object> messageIds = unmarkedOps.range(0, unmarkedOps.size()-1);
for (Object messageId : messageIds) {
this.messageStore.removeMessage(UUID.fromString(messageId.toString()));
}
this.redisTemplate.delete(UNMARKED_PREFIX + this.groupId.toString());
this.redisTemplate.delete(unmarkedId);
mGroupListOps = this.redisTemplate.boundListOps(MARKED_PREFIX + this.groupId.toString());
messageIds = mGroupListOps.range(0, mGroupListOps.size()-1);
BoundListOperations<String, Object> markedOps = this.redisTemplate.boundListOps(markedId);
messageIds = markedOps.range(0, markedOps.size()-1);
for (Object messageId : messageIds) {
this.messageStore.removeMessage(UUID.fromString(messageId.toString()));
}
this.redisTemplate.delete(MARKED_PREFIX + this.groupId.toString());
this.redisTemplate.delete(markedId);
this.rebuildLocalCache(unmarkedOps, markedOps);
}
private Collection<Message<?>> buildMessageList(BoundListOperations<String, Object> mGroupOps){
@@ -208,9 +211,9 @@ class RedisMessageGroup implements MessageGroup {
else {
synchronized (lock) {
BoundListOperations<String, Object> mGroupOps = this.redisTemplate.boundListOps(UNMARKED_PREFIX + this.groupId.toString());
BoundListOperations<String, Object> mGroupOps = this.redisTemplate.boundListOps(unmarkedId);
Collection<Message<?>> unmarked = this.buildMessageList(mGroupOps);
mGroupOps = this.redisTemplate.boundListOps(MARKED_PREFIX + this.groupId.toString());
mGroupOps = this.redisTemplate.boundListOps(markedId);
Collection<Message<?>> marked = this.buildMessageList(mGroupOps);
if (containsSequenceNumber(unmarked, messageSequenceNumber)
|| containsSequenceNumber(marked, messageSequenceNumber)) {

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

@@ -88,6 +88,8 @@ public class RedisMessageGroupStoreTests extends RedisAvailableTests {
store.removeMessageGroup(1);
messageGroup = store.getMessageGroup(1);
assertEquals(0, messageGroup.getMarked().size());
assertEquals(0, messageGroup.getUnmarked().size());
assertEquals(0, messageGroup.size());
}
@@ -222,6 +224,7 @@ public class RedisMessageGroupStoreTests extends RedisAvailableTests {
assertNotSame(mg1, mg2);
}
@Test
@RedisAvailable
public void testWithAggregatorWithShutdown(){