diff --git a/spring-integration-redis/src/main/java/org/springframework/integration/redis/store/RedisMessageStore.java b/spring-integration-redis/src/main/java/org/springframework/integration/redis/store/RedisMessageStore.java index 19fc9e51ad..ce1d052c66 100644 --- a/spring-integration-redis/src/main/java/org/springframework/integration/redis/store/RedisMessageStore.java +++ b/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> redisTemplate; + private volatile RedisSerializer valueSerializer = new JdkSerializationRedisSerializer(); + public RedisMessageStore(RedisConnectionFactory connectionFactory){ this.redisTemplate = new RedisTemplate>(); this.redisTemplate.setConnectionFactory(connectionFactory); - this.redisTemplate.setKeySerializer(new JacksonJsonRedisSerializer(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> ops = redisTemplate.boundValueOps(id); + return ops.get(); + } + return null; + } + + + @SuppressWarnings("unchecked") + public Message addMessage(Message message) { + Assert.notNull(message, "'message' must not be null"); + BoundValueOperations> ops = redisTemplate.boundValueOps(message.getHeaders().getId()); + ops.set(message); + return (Message) 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() { + 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() { - + public void setValueSerializer(RedisSerializer valueSerializer) { + this.valueSerializer = valueSerializer; + } + + private boolean messageExists(final UUID id){ + return redisTemplate.execute(new RedisCallback() { public Boolean doInRedis(RedisConnection connection) throws DataAccessException { return connection.exists(id.toString().getBytes()); } }); - if (!exists){ - return null; + } + + private static class UuidSerializer implements RedisSerializer { + + public byte[] serialize(UUID t) throws SerializationException { + return t.toString().getBytes(); } - else { - BoundValueOperations> ops = redisTemplate.boundValueOps(id); - return ops.get(); - } + + public UUID deserialize(byte[] bytes) throws SerializationException { + return UUID.fromString(new String(bytes)); + } + } - - public Message addMessage(Message 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; - } - } diff --git a/spring-integration-redis/src/test/java/org/springframework/integration/redis/rules/RedisAvailableTests.java b/spring-integration-redis/src/test/java/org/springframework/integration/redis/rules/RedisAvailableTests.java index ec4758a274..ab24d9bafb 100644 --- a/spring-integration-redis/src/test/java/org/springframework/integration/redis/rules/RedisAvailableTests.java +++ b/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>(); + rt.setConnectionFactory(jcf); + rt.execute(new RedisCallback() { + + public Object doInRedis(RedisConnection connection) + throws DataAccessException { + connection.flushDb(); + return null; + } + }); + return jcf; + } } diff --git a/spring-integration-redis/src/test/java/org/springframework/integration/redis/store/RedisMessageStoreTests.java b/spring-integration-redis/src/test/java/org/springframework/integration/redis/store/RedisMessageStoreTests.java index 21e9f2b75a..30b642b133 100644 --- a/spring-integration-redis/src/test/java/org/springframework/integration/redis/store/RedisMessageStoreTests.java +++ b/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 stringMessage = new GenericMessage("Hello Redis"); + Message 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 stringMessage = new GenericMessage("Hello Redis"); + store.addMessage(stringMessage); + Message retrievedMessage = (Message) 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 stringMessage = new GenericMessage("Hello Redis"); + store.addMessage(stringMessage); + Message retrievedMessage = (Message) 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()); + } }