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

1. simplified size() method to simply return the sum of sizes from marked and unmarked
2. removed System,out.println
3. added more assertions to tests
4. cosmetic changes
This commit is contained in:
Oleg Zhurakousky
2011-08-11 17:36:01 -04:00
committed by Mark Fisher
parent a598238c5e
commit ab47ca704b
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(){