INT-2029 finished with initial implementation of RedisMessageStore.

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

@@ -17,64 +17,107 @@ package org.springframework.integration.redis.store;
 
 import java.util.UUID;
 
+import org.springframework.beans.factory.InitializingBean;
 import org.springframework.dao.DataAccessException;
 import org.springframework.data.redis.connection.RedisConnection;
 import org.springframework.data.redis.connection.RedisConnectionFactory;
 import org.springframework.data.redis.core.BoundValueOperations;
 import org.springframework.data.redis.core.RedisCallback;
 import org.springframework.data.redis.core.RedisTemplate;
-import org.springframework.data.redis.serializer.JacksonJsonRedisSerializer;
+import org.springframework.data.redis.serializer.JdkSerializationRedisSerializer;
+import org.springframework.data.redis.serializer.RedisSerializer;
+import org.springframework.data.redis.serializer.SerializationException;
 import org.springframework.integration.Message;
 import org.springframework.integration.store.MessageStore;
+import org.springframework.util.Assert;
 
 /**
  * @author Oleg Zhurakousky
  * @since 2.1
  *
  */
-public class RedisMessageStore implements MessageStore {
+public class RedisMessageStore implements MessageStore, InitializingBean{
 
 	private final RedisTemplate<UUID, Message<?>> redisTemplate;
 	
+	private volatile RedisSerializer<?> valueSerializer = new JdkSerializationRedisSerializer();
+	
 	public RedisMessageStore(RedisConnectionFactory connectionFactory){
 		this.redisTemplate = new RedisTemplate<UUID, Message<?>>();
 		this.redisTemplate.setConnectionFactory(connectionFactory);
-		this.redisTemplate.setKeySerializer(new JacksonJsonRedisSerializer<UUID>(UUID.class));	
+		this.redisTemplate.setKeySerializer(new UuidSerializer());	
+		this.redisTemplate.setValueSerializer(this.valueSerializer);	
+	}
+
+	public Message<?> getMessage(final UUID id) {
+		Assert.notNull(id, "'id' must not be null");
+		if (this.messageExists(id)){
+			BoundValueOperations<UUID, Message<?>> ops = redisTemplate.boundValueOps(id);		
+			return ops.get();
+		}
+		return null;	
+	}
+
+	
+	@SuppressWarnings("unchecked")
+	public <T> Message<T> addMessage(Message<T> message) {
+		Assert.notNull(message, "'message' must not be null");
+		BoundValueOperations<UUID, Message<?>> ops = redisTemplate.boundValueOps(message.getHeaders().getId());	
+		ops.set(message);
+		return (Message<T>) ops.get();
+	}
+
+	
+	public Message<?> removeMessage(UUID id) {
+		Assert.notNull(id, "'id' must not be null");
+		Message<?> message = this.getMessage(id);
+		if (message != null){
+			this.redisTemplate.delete(id);
+			return message;
+		}
+		else {
+			throw new IllegalArgumentException("Message with id '" + id + "' can not be removed since it does NOT exist");
+		}
+	}
+
+	
+	public int getMessageCount() {
+		return redisTemplate.execute(new RedisCallback<Integer>() {
+			public Integer doInRedis(RedisConnection connection)
+					throws DataAccessException {
+				long l = connection.dbSize();
+				Assert.isTrue(l <= Integer.MAX_VALUE, "Message count is out of range");
+				return (int)l;
+			}
+		});
 	}
 	
-	public Message<?> getMessage(final UUID id) {
-		boolean exists = redisTemplate.execute(new RedisCallback<Boolean>() {
-
+	public void setValueSerializer(RedisSerializer<?> valueSerializer) {
+		this.valueSerializer = valueSerializer;
+	}
+	
+	private boolean messageExists(final UUID id){
+		return redisTemplate.execute(new RedisCallback<Boolean>() {
 			public Boolean doInRedis(RedisConnection connection)
 					throws DataAccessException {
 				return connection.exists(id.toString().getBytes());
 			}
 		});
-		if (!exists){
-			return null;
+	}
+	
+	private static class UuidSerializer implements RedisSerializer<UUID> {
+
+		public byte[] serialize(UUID t) throws SerializationException {
+			return t.toString().getBytes();
 		}
-		else {
-			BoundValueOperations<UUID, Message<?>> ops = redisTemplate.boundValueOps(id);		
-			return ops.get();
-		}	
+
+		public UUID deserialize(byte[] bytes) throws SerializationException {
+			return UUID.fromString(new String(bytes));
+		}
+		
 	}
 
-	
-	public <T> Message<T> addMessage(Message<T> message) {
-		// TODO Auto-generated method stub
-		return null;
+	public void afterPropertiesSet() throws Exception {
+		Assert.notNull(this.valueSerializer, "'valueSerializer' must not be null");
 	}
-
-	
-	public Message<?> removeMessage(UUID id) {
-		// TODO Auto-generated method stub
-		return null;
-	}
-
-	
-	public int getMessageCount() {
-		// TODO Auto-generated method stub
-		return 0;
-	}
-
 }

spring-integration-redis/src/test/java/org/springframework/integration/redis/rules/RedisAvailableTests.java

@@ -15,7 +15,15 @@
  */
 package org.springframework.integration.redis.rules;
 
+import java.util.UUID;
+
 import org.junit.Rule;
+import org.springframework.dao.DataAccessException;
+import org.springframework.data.redis.connection.RedisConnection;
+import org.springframework.data.redis.connection.jedis.JedisConnectionFactory;
+import org.springframework.data.redis.core.RedisCallback;
+import org.springframework.data.redis.core.RedisTemplate;
+import org.springframework.integration.Message;
 
 /**
  * @author Oleg Zhurakousky
@@ -24,4 +32,22 @@ import org.junit.Rule;
 public class RedisAvailableTests {
 	@Rule
 	public RedisAvailableRule redisAvailableRule = new RedisAvailableRule();
+	
+	@SuppressWarnings({ "rawtypes", "unchecked" })
+	public JedisConnectionFactory getConnectionFactoryForTest(){
+		JedisConnectionFactory jcf = new JedisConnectionFactory();
+		jcf.setPort(7379);
+		jcf.afterPropertiesSet();
+		RedisTemplate rt = new RedisTemplate<UUID, Message<?>>();
+		rt.setConnectionFactory(jcf);
+		rt.execute(new RedisCallback() {
+
+			public Object doInRedis(RedisConnection connection)
+					throws DataAccessException {
+				connection.flushDb();
+				return null;
+			}
+		});
+		return jcf;
+	}
 }

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

@@ -17,28 +17,82 @@ package org.springframework.integration.redis.store;
 
 import java.util.UUID;
 
-import org.junit.Ignore;
 import org.junit.Test;
 import org.springframework.data.redis.connection.jedis.JedisConnectionFactory;
 import org.springframework.integration.Message;
+import org.springframework.integration.message.GenericMessage;
+import org.springframework.integration.redis.rules.RedisAvailable;
+import org.springframework.integration.redis.rules.RedisAvailableTests;
 
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertNotSame;
 import static org.junit.Assert.assertNull;
 
 /**
  * @author Oleg Zhurakousky
  *
  */
-public class RedisMessageStoreTests {
+public class RedisMessageStoreTests extends RedisAvailableTests {
 
 	@Test
-	@Ignore
-	public void testGetNonExistingMessage(){
-		
-		JedisConnectionFactory jcf = new JedisConnectionFactory();
-		jcf.setPort(6379);
-		jcf.afterPropertiesSet();
+	@RedisAvailable
+	public void testGetNonExistingMessage(){	
+		JedisConnectionFactory jcf = this.getConnectionFactoryForTest();
 		RedisMessageStore store = new RedisMessageStore(jcf);
 		Message<?> message = store.getMessage(UUID.randomUUID());
 		assertNull(message);
 	}
+	
+	@Test
+	@RedisAvailable
+	public void testGetMessageCountWhenEmpty(){	
+		JedisConnectionFactory jcf = this.getConnectionFactoryForTest();
+		RedisMessageStore store = new RedisMessageStore(jcf);
+		assertEquals(0, store.getMessageCount());
+	}
+	
+	@Test
+	@RedisAvailable
+	public void testAddStringMessage(){	
+		JedisConnectionFactory jcf = this.getConnectionFactoryForTest();
+		RedisMessageStore store = new RedisMessageStore(jcf);
+		Message<String> stringMessage = new GenericMessage<String>("Hello Redis");
+		Message<String> storedMessage =  store.addMessage(stringMessage);
+		assertNotSame(stringMessage, storedMessage);
+		assertEquals("Hello Redis", storedMessage.getPayload());
+	}
+	
+	@SuppressWarnings("unchecked")
+	@Test
+	@RedisAvailable
+	public void testAddAndGetStringMessage(){	
+		JedisConnectionFactory jcf = this.getConnectionFactoryForTest();
+		RedisMessageStore store = new RedisMessageStore(jcf);
+		Message<String> stringMessage = new GenericMessage<String>("Hello Redis");
+		store.addMessage(stringMessage);
+		Message<String> retrievedMessage = (Message<String>) store.getMessage(stringMessage.getHeaders().getId());
+		assertNotNull(retrievedMessage);
+		assertEquals("Hello Redis", retrievedMessage.getPayload());
+	}
+	@SuppressWarnings("unchecked")
+	@Test
+	@RedisAvailable
+	public void testAddAndRemoveStringMessage(){	
+		JedisConnectionFactory jcf = this.getConnectionFactoryForTest();
+		RedisMessageStore store = new RedisMessageStore(jcf);
+		Message<String> stringMessage = new GenericMessage<String>("Hello Redis");
+		store.addMessage(stringMessage);
+		Message<String> retrievedMessage = (Message<String>) store.removeMessage(stringMessage.getHeaders().getId());
+		assertNotNull(retrievedMessage);
+		assertEquals("Hello Redis", retrievedMessage.getPayload());
+		assertNull(store.getMessage(stringMessage.getHeaders().getId()));
+	}
+	@Test(expected=IllegalArgumentException.class)
+	@RedisAvailable
+	public void testRemoveNonExistingMessage(){	
+		JedisConnectionFactory jcf = this.getConnectionFactoryForTest();
+		RedisMessageStore store = new RedisMessageStore(jcf);
+		store.removeMessage(UUID.randomUUID());
+	}
 }