INT-2030 finished implementing all methods for Redis-based MessageGroupStore

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2010 the original author or authors.
+ * Copyright 2002-2011 the original author or authors.
  * 
  * Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in compliance with
  * the License. You may obtain a copy of the License at
@@ -18,7 +18,6 @@ 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.RedisTemplate;
 import org.springframework.integration.Message;
 import org.springframework.integration.store.MessageGroup;
@@ -33,9 +32,14 @@ class RedisMessageGroup implements MessageGroup {
 	private final String MARKED_PREFIX = "MARKED_";
 	private final String UNMARKED_PREFIX = "UNMARKED_";
 	
+	private final List<Message<?>> unmarked = new LinkedList<Message<?>>();
+	private final List<Message<?>> marked = new LinkedList<Message<?>>();
+	
 	private final Object groupId;
 	private final RedisTemplate<String, Object> redisTemplate;
 	private final MessageStore messageStore;
+	private final Object lock = new Object();
+	private final long timestamp = System.currentTimeMillis();
 	
 	public RedisMessageGroup(MessageStore messageStore, RedisTemplate<String, Object> redisTemplate,  Object groupId){
 		this.groupId = groupId;
@@ -44,75 +48,16 @@ class RedisMessageGroup implements MessageGroup {
 	}
 
 	public boolean canAdd(Message<?> message) {
-		// TODO Auto-generated method stub
+		return !isMember(message);
-		return false;
 	}
-	/**
-	 * 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.
-	 */
-	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());
-	}
-	/**
-	 * 
-	 */
-	public void destroy(){
-		// delete unmarked
-		BoundSetOperations<String, Object> mGroupOps = this.redisTemplate.boundSetOps(UNMARKED_PREFIX + this.groupId.toString());
-		for (Object messageId : mGroupOps.members()) {
-			this.messageStore.removeMessage(UUID.fromString(messageId.toString()));
-			//this.redisTemplate.delete(messageId.toString());
-		}
-		this.redisTemplate.delete(UNMARKED_PREFIX + this.groupId.toString());
-		// delete marked
-		mGroupOps = this.redisTemplate.boundSetOps(MARKED_PREFIX + this.groupId.toString());
-		for (Object messageId : mGroupOps.members()) {
-			this.messageStore.removeMessage(UUID.fromString(messageId.toString()));
-		}
-		this.redisTemplate.delete(MARKED_PREFIX + this.groupId.toString());
-	}
-	
-	public void markAll() {
-		BoundSetOperations<String, Object> unmarkedGroupOps = this.redisTemplate.boundSetOps(UNMARKED_PREFIX + this.groupId.toString());
-		long uSize = unmarkedGroupOps.size();
-		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");
-	}
-	
-	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);
-	}
-
 	
 	public Collection<Message<?>> getUnmarked() {
-		BoundSetOperations<String, Object> mGroupOps = this.redisTemplate.boundSetOps(UNMARKED_PREFIX + this.groupId.toString());
+		return this.unmarked;
-		return this.getMessages(mGroupOps);
 	}
 
 	
 	public Collection<Message<?>> getMarked() {
-		BoundSetOperations<String, Object> mGroupOps = this.redisTemplate.boundSetOps(MARKED_PREFIX + this.groupId.toString());
+		return this.marked;
-		return this.getMessages(mGroupOps);
 	}
 
 	
@@ -120,61 +65,181 @@ class RedisMessageGroup implements MessageGroup {
 		return this.groupId;
 	}
 
-	/* (non-Javadoc)
-	 * @see org.springframework.integration.store.MessageGroup#isComplete()
-	 */
 	public boolean isComplete() {
-		// TODO Auto-generated method stub
+		if (size() == 0) {
-		return false;
+			return true;
+		}
+		int sequenceSize = getSequenceSize();
+		// If there is no sequence then it must be incomplete....
+		return sequenceSize > 0 && sequenceSize == size();
 	}
 
-	/* (non-Javadoc)
-	 * @see org.springframework.integration.store.MessageGroup#getSequenceSize()
-	 */
 	public int getSequenceSize() {
-		// TODO Auto-generated method stub
+		if (size() == 0) {
-		return 0;
+			return 0;
+		}
+		return getOne().getHeaders().getSequenceSize();
 	}
 
-	
 	public int size() {
-		BoundSetOperations<String, Object> mGroupOps = this.redisTemplate.boundSetOps(UNMARKED_PREFIX + this.groupId.toString());
+		BoundListOperations<String, Object> mGroupOps = this.redisTemplate.boundListOps(UNMARKED_PREFIX + this.groupId.toString());
-		return mGroupOps.members().size();
+		long usize = mGroupOps.size();
+		mGroupOps = this.redisTemplate.boundListOps(MARKED_PREFIX + this.groupId.toString());
+		long msize = mGroupOps.size();
+		return (int)(usize + msize);
 	}
 
-	
 	public Message<?> getOne() {
-		BoundListOperations<String, Object> mGroupListOps = this.redisTemplate.boundListOps("L" + UNMARKED_PREFIX + this.groupId.toString());
+		Message<?> one = unmarked.get(0);
-		Object id = mGroupListOps.leftPop();
+		if (one == null) {
-		return this.messageStore.getMessage(UUID.fromString(id.toString()));
+			one = marked.get(0);
-//		List<Message<?>> messages = (List<Message<?>>) this.getUnmarked();
+		}
-//		if (messages.isEmpty()){
+		return one;
-//			messages = (List<Message<?>>) this.getMarked();
-//			if (!messages.isEmpty()){
-//				return messages.get(0);
-//			}
-//		}
-//		else {
-//			return messages.get(0);
-//		}
-		//return null;
 	}
 
-	/* (non-Javadoc)
-	 * @see org.springframework.integration.store.MessageGroup#getTimestamp()
-	 */
 	public long getTimestamp() {
-		// TODO Auto-generated method stub
+		return this.timestamp;
-		return 0;
 	}
 	
-	private Collection<Message<?>> getMessages(BoundSetOperations<String, Object> mGroupOps){
+	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());
+		long uSize = unmarkedOps.size();
+		String destiinationKey = MARKED_PREFIX + this.groupId.toString();
+		
+		synchronized (lock) {
+			unmarkedOps.rename(destiinationKey);
+			this.unmarked.clear();
+			this.rebuildLocalCache(null, markedOps);
+		}
+		
+		long mSize = markedOps.size();
+		Assert.isTrue(mSize == uSize, "Failed to mark All messages in the message group");	
+	}
+	
+	protected void markMessage(String messageId) {
+		BoundListOperations<String, Object> unmarkedOps = this.redisTemplate.boundListOps(UNMARKED_PREFIX + this.groupId.toString());
+		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());
+					markedOps.rightPush(id);
+					System.out.println(unmarkedOps.size());
+					unmarkedOps.remove(0, id);
+					System.out.println(unmarkedOps.size());
+					this.rebuildLocalCache(unmarkedOps, markedOps);
+					return;
+				}
+			}
+		}
+	}
+	
+	/**
+	 * 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.
+	 */
+	protected void add(Message<?> message) {
+		String messageId = message.getHeaders().getId().toString();
+		BoundListOperations<String, Object> unmarkedOps = this.redisTemplate.boundListOps(UNMARKED_PREFIX + this.groupId.toString());
+		synchronized (lock) {
+			unmarkedOps.rightPush(messageId);
+			this.messageStore.addMessage(message);
+			this.rebuildLocalCache(unmarkedOps, null);
+		}	
+	}
+	
+	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());
+		
+		synchronized (lock) {
+			unmarkedOps.remove(0, messageId.toString());		
+			markedOps.remove(0, messageId.toString());
+			this.messageStore.removeMessage(messageId);
+			this.rebuildLocalCache(unmarkedOps, markedOps);
+		}
+		
+	}
+	/**
+	 * 
+	 */
+	protected void destroy(){
+		BoundListOperations<String, Object> mGroupListOps = this.redisTemplate.boundListOps(UNMARKED_PREFIX + this.groupId.toString());
+		List<Object> messageIds =  mGroupListOps.range(0, mGroupListOps.size()-1);
+		for (Object messageId : messageIds) {
+			this.messageStore.removeMessage(UUID.fromString(messageId.toString()));
+		}
+		this.redisTemplate.delete(UNMARKED_PREFIX + this.groupId.toString());
+		
+		mGroupListOps = this.redisTemplate.boundListOps(MARKED_PREFIX + this.groupId.toString());
+		messageIds =  mGroupListOps.range(0, mGroupListOps.size()-1);
+		for (Object messageId : messageIds) {
+			this.messageStore.removeMessage(UUID.fromString(messageId.toString()));
+		}
+		this.redisTemplate.delete(MARKED_PREFIX + this.groupId.toString());
+	}
+	
+	private Collection<Message<?>> buildMessageList(BoundListOperations<String, Object> mGroupOps){
 		List<Message<?>> messages = new LinkedList<Message<?>>();
-		for (Object messageId : mGroupOps.members()) {
+		List<Object> messageIds = mGroupOps.range(0, mGroupOps.size()-1);
+		for (Object messageId : messageIds) {
 			Message<?> message = this.messageStore.getMessage(UUID.fromString(messageId.toString()));
 			messages.add((Message<?>) message);
 		}
 		return messages;
 	}
 	
+	private boolean isMember(Message<?> message){
+		if (size() == 0) {
+			return false;
+		}
+		Integer messageSequenceNumber = message.getHeaders().getSequenceNumber();
+		if (messageSequenceNumber != null && messageSequenceNumber > 0) {
+			Integer messageSequenceSize = message.getHeaders().getSequenceSize();
+			if (!messageSequenceSize.equals(getSequenceSize())) {
+				return true;
+			}
+			else {
+				synchronized (lock) {
+					
+					BoundListOperations<String, Object> mGroupOps = this.redisTemplate.boundListOps(UNMARKED_PREFIX + this.groupId.toString());
+					Collection<Message<?>> unmarked = this.buildMessageList(mGroupOps);
+					mGroupOps = this.redisTemplate.boundListOps(MARKED_PREFIX + this.groupId.toString());
+					Collection<Message<?>> marked = this.buildMessageList(mGroupOps);
+					if (containsSequenceNumber(unmarked, messageSequenceNumber)
+							|| containsSequenceNumber(marked, messageSequenceNumber)) {
+						return true;
+					}
+				}
+			}
+		}
+		return false;
+	}
+	
+	private boolean containsSequenceNumber(Collection<Message<?>> messages, Integer messageSequenceNumber) {
+		for (Message<?> member : messages) {
+			Integer memberSequenceNumber = member.getHeaders().getSequenceNumber();
+			if (messageSequenceNumber.equals(memberSequenceNumber)) {
+				return true;
+			}
+		}
+		return false;
+	}
+	
+	private void rebuildLocalCache(BoundListOperations<String, Object> unmarkedOps, BoundListOperations<String, Object> markedOps){
+		
+		
+		if (unmarkedOps != null){
+			this.unmarked.clear();
+			this.unmarked.addAll(this.buildMessageList(unmarkedOps));
+		}
+		if (markedOps != null){
+			this.marked.clear();
+			this.marked.addAll(this.buildMessageList(markedOps));
+		}
+	}
+
 }

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

@@ -16,8 +16,10 @@
 package org.springframework.integration.redis.store;
 
 import java.util.ArrayList;
+import java.util.HashMap;
 import java.util.Iterator;
 import java.util.List;
+import java.util.Map;
 import java.util.UUID;
 
 import org.springframework.beans.factory.InitializingBean;
@@ -49,8 +51,12 @@ public class RedisMessageStore extends AbstractMessageGroupStore implements Mess
 
 	private final RedisTemplate<String, Object> redisTemplate;
 	
+	private final Map<Object, RedisMessageGroup> messageGroups = new HashMap<Object, RedisMessageGroup>();
+	
 	private volatile RedisSerializer<?> valueSerializer = new JdkSerializationRedisSerializer();
 	
+	private final Object lock = new Object();
+	
 	public RedisMessageStore(RedisConnectionFactory connectionFactory){
 		this.redisTemplate = new RedisTemplate<String, Object>();
 		this.redisTemplate.setConnectionFactory(connectionFactory);
@@ -144,12 +150,19 @@ public class RedisMessageStore extends AbstractMessageGroupStore implements Mess
 	 */
 	public MessageGroup getMessageGroup(Object groupId) {
 		Assert.notNull(groupId, "'groupId' must not be null");
-		
+		RedisMessageGroup group = this.messageGroups.get(groupId);
-		BoundSetOperations<String, Object> mGroupsOps = this.redisTemplate.boundSetOps(MESSAGE_GROUPS_KEY);
+		if (group == null){
-		if (!mGroupsOps.isMember(groupId)){
+			synchronized (lock) {
-			mGroupsOps.add(groupId);
+				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);
+				}
+			}
 		}
-		return new RedisMessageGroup(this, this.redisTemplate, groupId);
+		
+		return group;
 	}
 	/**
 	 * 
@@ -188,8 +201,13 @@ public class RedisMessageStore extends AbstractMessageGroupStore implements Mess
 	public void removeMessageGroup(Object groupId) {
 		Assert.notNull(groupId, "'groupId' must not be null");
 		RedisMessageGroup messageGroup = (RedisMessageGroup) this.getMessageGroup(groupId);
-		messageGroup.destroy();
+		synchronized (messageGroup) {
-		this.redisTemplate.delete(groupId.toString());
+			messageGroup.destroy();
+			BoundSetOperations<String, Object> mGroupsOps = this.redisTemplate.boundSetOps(MESSAGE_GROUPS_KEY);
+			mGroupsOps.remove(groupId);
+			this.redisTemplate.delete(groupId.toString());
+			this.messageGroups.remove(groupId);
+		}
 	}
 
 	@Override

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

@@ -22,9 +22,11 @@ import org.springframework.integration.message.GenericMessage;
 import org.springframework.integration.redis.rules.RedisAvailable;
 import org.springframework.integration.redis.rules.RedisAvailableTests;
 import org.springframework.integration.store.MessageGroup;
+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.assertTrue;
 
 /**
@@ -85,6 +87,23 @@ public class RedisMessageGroupStoreTests extends RedisAvailableTests {
 		assertEquals(0, messageGroup.size());
 	}
 	
+	@Test
+	@RedisAvailable
+	public void testRemoveMessageFromTheGroup(){	
+		JedisConnectionFactory jcf = this.getConnectionFactoryForTest();
+		RedisMessageStore store = new RedisMessageStore(jcf);
+
+		MessageGroup messageGroup = store.getMessageGroup(1);
+		Message<?> message = new GenericMessage<String>("2");
+		((RedisMessageGroup)messageGroup).add(new GenericMessage<String>("1"));
+		((RedisMessageGroup)messageGroup).add(message);
+		((RedisMessageGroup)messageGroup).add(new GenericMessage<String>("3"));
+		assertEquals(3, messageGroup.size());
+		
+		store.removeMessageFromGroup(1, message);
+		assertEquals(2, messageGroup.size());
+	}
+	
 	@Test
 	@RedisAvailable
 	public void testMarkAllMessagesInMessageGroup(){	
@@ -134,4 +153,69 @@ public class RedisMessageGroupStoreTests extends RedisAvailableTests {
 		assertEquals("1", message.getPayload());
 	}
 	
+	@Test
+	@RedisAvailable
+	public void testAddToMessageGroup(){	
+		JedisConnectionFactory jcf = this.getConnectionFactoryForTest();
+		RedisMessageStore store = new RedisMessageStore(jcf);
+		MessageGroup messageGroup = store.getMessageGroup(1);
+		
+		store.addMessageToGroup(1, new GenericMessage<String>("1"));
+		assertEquals(1, messageGroup.size());
+		store.addMessageToGroup(1, new GenericMessage<String>("2"));
+		assertEquals(2, messageGroup.size());
+		store.addMessageToGroup(1, new GenericMessage<String>("3"));
+		assertEquals(3, messageGroup.size());
+	}
+	
+	@Test
+	@RedisAvailable
+	public void testCanAddToMessageGroup(){	
+		JedisConnectionFactory jcf = this.getConnectionFactoryForTest();
+		RedisMessageStore store = new RedisMessageStore(jcf);
+
+		
+		MessageGroup messageGroup = store.getMessageGroup(1);
+		Message<?> m1 = MessageBuilder.withPayload("1").setSequenceNumber(1).setSequenceSize(3).setCorrelationId(1).build();
+		Message<?> m2 = MessageBuilder.withPayload("2").setSequenceNumber(2).setSequenceSize(3).setCorrelationId(1).build();
+		Message<?> m3 = MessageBuilder.withPayload("3").setSequenceNumber(3).setSequenceSize(3).setCorrelationId(1).build();
+		
+		if (messageGroup.canAdd(m1)){
+			store.addMessageToGroup(1, m1);
+		}
+		assertEquals(1, messageGroup.size());
+		
+		if (messageGroup.canAdd(m2)){
+			store.addMessageToGroup(1, m2);
+		}
+		assertEquals(2, messageGroup.size());
+		
+		if (messageGroup.canAdd(m3)){
+			store.addMessageToGroup(1, m3);
+		}
+		assertEquals(3, messageGroup.size());
+		
+		if (messageGroup.canAdd(m3)){
+			store.addMessageToGroup(1, m3);
+		}
+		assertEquals(3, messageGroup.size());
+	}
+	
+	@Test
+	@RedisAvailable
+	public void testSameInstance(){	
+		JedisConnectionFactory jcf = this.getConnectionFactoryForTest();
+		RedisMessageStore store = new RedisMessageStore(jcf);
+
+		MessageGroup mg1 = store.getMessageGroup(1);
+		MessageGroup mg2 = store.getMessageGroup(1);
+		
+		assertEquals(mg1, mg2);
+		
+		store.removeMessageGroup(1);
+		
+		mg2 = store.getMessageGroup(1);
+		assertNotSame(mg1, mg2);
+	}
+	
 }