diff --git a/spring-integration-redis/src/main/java/org/springframework/integration/redis/store/RedisMessageGroup.java b/spring-integration-redis/src/main/java/org/springframework/integration/redis/store/RedisMessageGroup.java new file mode 100644 index 0000000000..2b23d81c99 --- /dev/null +++ b/spring-integration-redis/src/main/java/org/springframework/integration/redis/store/RedisMessageGroup.java @@ -0,0 +1,140 @@ +/* + * Copyright 2002-2010 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 + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on + * an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the + * specific language governing permissions and limitations under the License. + */ +package org.springframework.integration.redis.store; + +import java.util.ArrayList; +import java.util.Collection; +import java.util.List; + +import org.springframework.data.redis.core.BoundSetOperations; +import org.springframework.data.redis.core.BoundValueOperations; +import org.springframework.data.redis.core.RedisTemplate; +import org.springframework.integration.Message; +import org.springframework.integration.store.MessageGroup; +import org.springframework.util.Assert; + +/** + * @author Oleg Zhurakousky + * @since 2.1 + */ +class RedisMessageGroup implements MessageGroup { + private final String MARKED_PREFIX = "MARKED_"; + private final String UNMARKED_PREFIX = "UNMARKED_"; + + private final Object groupId; + private final RedisTemplate redisTemplate; + + public RedisMessageGroup(RedisTemplate redisTemplate, Object groupId){ + this.groupId = groupId; + this.redisTemplate = redisTemplate; + } + + public boolean canAdd(Message message) { + // TODO Auto-generated method stub + return false; + } + + public void add(Message message) { + BoundSetOperations mGroupOps = this.redisTemplate.boundSetOps(UNMARKED_PREFIX + this.groupId.toString()); + String key = message.getHeaders().getId().toString(); + mGroupOps.add(key); + BoundValueOperations bvOps = this.redisTemplate.boundValueOps(key); + bvOps.set(message); + } + + public void destroy(){ + // delete unmarked + BoundSetOperations mGroupOps = this.redisTemplate.boundSetOps(UNMARKED_PREFIX + this.groupId.toString()); + for (Object messageId : mGroupOps.members()) { + 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.redisTemplate.delete(messageId.toString()); + } + this.redisTemplate.delete(MARKED_PREFIX + this.groupId.toString()); + } + + public void markAll() { + + } + + + public Collection> getUnmarked() { + BoundSetOperations mGroupOps = this.redisTemplate.boundSetOps(UNMARKED_PREFIX + this.groupId.toString()); + return this.getMessages(mGroupOps); + } + + + public Collection> getMarked() { + BoundSetOperations mGroupOps = this.redisTemplate.boundSetOps(MARKED_PREFIX + this.groupId.toString()); + return this.getMessages(mGroupOps); + } + + + public Object getGroupId() { + return this.groupId; + } + + /* (non-Javadoc) + * @see org.springframework.integration.store.MessageGroup#isComplete() + */ + public boolean isComplete() { + // TODO Auto-generated method stub + return false; + } + + /* (non-Javadoc) + * @see org.springframework.integration.store.MessageGroup#getSequenceSize() + */ + public int getSequenceSize() { + // TODO Auto-generated method stub + return 0; + } + + + public int size() { + BoundSetOperations mGroupOps = this.redisTemplate.boundSetOps(UNMARKED_PREFIX + this.groupId.toString()); + return mGroupOps.members().size(); + } + + /* (non-Javadoc) + * @see org.springframework.integration.store.MessageGroup#getOne() + */ + public Message getOne() { + // TODO Auto-generated method stub + return null; + } + + /* (non-Javadoc) + * @see org.springframework.integration.store.MessageGroup#getTimestamp() + */ + public long getTimestamp() { + // TODO Auto-generated method stub + return 0; + } + + private Collection> getMessages(BoundSetOperations mGroupOps){ + List> messages = new ArrayList>(); + for (Object messageId : mGroupOps.members()) { + BoundValueOperations bvOps = this.redisTemplate.boundValueOps(messageId.toString()); + Object message = bvOps.get(); + Assert.isInstanceOf(Message.class, message, "Retrieved object is not an instance of Message"); + messages.add((Message) message); + } + return messages; + } + +} 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 d6a283e870..54151f5d22 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 @@ -15,12 +15,16 @@ */ package org.springframework.integration.redis.store; +import java.util.ArrayList; +import java.util.Iterator; +import java.util.List; 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.BoundSetOperations; import org.springframework.data.redis.core.BoundValueOperations; import org.springframework.data.redis.core.RedisCallback; import org.springframework.data.redis.core.RedisTemplate; @@ -28,6 +32,8 @@ 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.AbstractMessageGroupStore; +import org.springframework.integration.store.MessageGroup; import org.springframework.integration.store.MessageStore; import org.springframework.integration.store.MessageStoreException; import org.springframework.util.Assert; @@ -37,24 +43,28 @@ import org.springframework.util.Assert; * @since 2.1 * */ -public class RedisMessageStore implements MessageStore, InitializingBean{ +public class RedisMessageStore extends AbstractMessageGroupStore implements MessageStore, InitializingBean{ + + private final String MESSAGE_GROUPS_KEY = "MESSAGE_GROUPS"; - private final RedisTemplate> redisTemplate; + private final RedisTemplate redisTemplate; private volatile RedisSerializer valueSerializer = new JdkSerializationRedisSerializer(); public RedisMessageStore(RedisConnectionFactory connectionFactory){ - this.redisTemplate = new RedisTemplate>(); + this.redisTemplate = new RedisTemplate(); this.redisTemplate.setConnectionFactory(connectionFactory); - this.redisTemplate.setKeySerializer(new UuidSerializer()); + this.redisTemplate.setKeySerializer(new KeySerializer()); 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(); + if (this.keyExists(id)){ + BoundValueOperations ops = redisTemplate.boundValueOps(id.toString()); + Object result = ops.get(); + Assert.isInstanceOf(Message.class, result, "Return value is not an instace of Message"); + return (Message) result; } return null; } @@ -63,7 +73,7 @@ public class RedisMessageStore implements MessageStore, InitializingBean{ @SuppressWarnings("unchecked") public Message addMessage(Message message) { Assert.notNull(message, "'message' must not be null"); - BoundValueOperations> ops = redisTemplate.boundValueOps(message.getHeaders().getId()); + BoundValueOperations ops = redisTemplate.boundValueOps(message.getHeaders().getId().toString()); try { ops.set(message); } catch (SerializationException e) { @@ -71,8 +81,9 @@ public class RedisMessageStore implements MessageStore, InitializingBean{ "the Message contains data that is not Serializable. Either make it Serializable or provide your own implementation of " + "RedisSerializer via 'setValueSerializer(..)'", e); } - - return (Message) ops.get(); + Object result = ops.get(); + Assert.isInstanceOf(Message.class, result, "Return value is not an instace of Message"); + return (Message) result; } @@ -80,7 +91,7 @@ public class RedisMessageStore implements MessageStore, InitializingBean{ Assert.notNull(id, "'id' must not be null"); Message message = this.getMessage(id); if (message != null){ - this.redisTemplate.delete(id); + this.redisTemplate.delete(id.toString()); return message; } else { @@ -104,7 +115,7 @@ public class RedisMessageStore implements MessageStore, InitializingBean{ this.valueSerializer = valueSerializer; } - private boolean messageExists(final UUID id){ + private boolean keyExists(final UUID id){ return redisTemplate.execute(new RedisCallback() { public Boolean doInRedis(RedisConnection connection) throws DataAccessException { @@ -113,14 +124,14 @@ public class RedisMessageStore implements MessageStore, InitializingBean{ }); } - private static class UuidSerializer implements RedisSerializer { + private static class KeySerializer implements RedisSerializer { - public byte[] serialize(UUID t) throws SerializationException { - return t.toString().getBytes(); + public byte[] serialize(String value) throws SerializationException { + return value.getBytes(); } - public UUID deserialize(byte[] bytes) throws SerializationException { - return UUID.fromString(new String(bytes)); + public String deserialize(byte[] bytes) throws SerializationException { + return new String(bytes); } } @@ -128,4 +139,64 @@ public class RedisMessageStore implements MessageStore, InitializingBean{ public void afterPropertiesSet() throws Exception { Assert.notNull(this.valueSerializer, "'valueSerializer' must not be null"); } + /** + * + */ + public MessageGroup getMessageGroup(Object groupId) { + Assert.notNull(groupId, "'groupId' must not be null"); + + BoundSetOperations mGroupsOps = this.redisTemplate.boundSetOps(MESSAGE_GROUPS_KEY); + if (!mGroupsOps.isMember(groupId)){ + mGroupsOps.add(groupId); + } + return new RedisMessageGroup(this.redisTemplate, groupId); + } + /** + * + */ + public MessageGroup addMessageToGroup(Object groupId, Message message) { + Assert.notNull(groupId, "'groupId' must not be null"); + Assert.notNull(message, "'message' must not be null"); + + MessageGroup mg = this.getMessageGroup(groupId); + + Assert.isInstanceOf(RedisMessageGroup.class, mg, "MessageGroup is not an instance of RedisMessageGroup"); + ((RedisMessageGroup)mg).add(message); + return mg; + } + + public MessageGroup markMessageGroup(MessageGroup group) { + Assert.isInstanceOf(RedisMessageGroup.class, group, "MessageGroup is not an instance of RedisMessageGroup"); + ((RedisMessageGroup)group).markAll(); + return group; + } + + public MessageGroup removeMessageFromGroup(Object key, + Message messageToRemove) { + // TODO Auto-generated method stub + return null; + } + + public MessageGroup markMessageFromGroup(Object key, + Message messageToMark) { + // TODO Auto-generated method stub + return null; + } + + public void removeMessageGroup(Object groupId) { + Assert.notNull(groupId, "'groupId' must not be null"); + RedisMessageGroup messageGroup = (RedisMessageGroup) this.getMessageGroup(groupId); + messageGroup.destroy(); + this.redisTemplate.delete(groupId.toString()); + } + + @Override + public Iterator iterator() { + BoundSetOperations mGroupsOps = this.redisTemplate.boundSetOps(MESSAGE_GROUPS_KEY); + List messageGroups = new ArrayList(); + for (Object msgGroupId : mGroupsOps.members()) { + messageGroups.add(new RedisMessageGroup(this.redisTemplate, msgGroupId)); + } + return messageGroups.iterator(); + } } 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 ab24d9bafb..e898c92986 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 @@ -23,7 +23,6 @@ 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 @@ -38,7 +37,7 @@ public class RedisAvailableTests { JedisConnectionFactory jcf = new JedisConnectionFactory(); jcf.setPort(7379); jcf.afterPropertiesSet(); - RedisTemplate rt = new RedisTemplate>(); + RedisTemplate rt = new RedisTemplate(); rt.setConnectionFactory(jcf); rt.execute(new RedisCallback() { diff --git a/spring-integration-redis/src/test/java/org/springframework/integration/redis/store/RedisMessageGroupStoreTests.java b/spring-integration-redis/src/test/java/org/springframework/integration/redis/store/RedisMessageGroupStoreTests.java new file mode 100644 index 0000000000..215e6d7514 --- /dev/null +++ b/spring-integration-redis/src/test/java/org/springframework/integration/redis/store/RedisMessageGroupStoreTests.java @@ -0,0 +1,88 @@ +/* + * Copyright 2007-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 + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.springframework.integration.redis.store; + +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 org.springframework.integration.store.MessageGroup; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertNotNull; +import static org.junit.Assert.assertTrue; + +/** + * @author Oleg Zhurakousky + * + */ +public class RedisMessageGroupStoreTests extends RedisAvailableTests { + + @Test + @RedisAvailable + public void testNonExistingEmptyMessageGroup(){ + JedisConnectionFactory jcf = this.getConnectionFactoryForTest(); + RedisMessageStore store = new RedisMessageStore(jcf); + + MessageGroup messageGroup = store.getMessageGroup(1); + assertNotNull(messageGroup); + assertTrue(messageGroup instanceof RedisMessageGroup); + assertEquals(0, messageGroup.size()); + } + + @Test + @RedisAvailable + public void testMessageGroupWithAddedMessageViaMessageGroup(){ + JedisConnectionFactory jcf = this.getConnectionFactoryForTest(); + RedisMessageStore store = new RedisMessageStore(jcf); + + MessageGroup messageGroup = store.getMessageGroup(1); + Message message = new GenericMessage("Hello"); + ((RedisMessageGroup)messageGroup).add(message); + assertEquals(1, messageGroup.size()); + } + + @Test + @RedisAvailable + public void testMessageGroupWithAddedMessageViaMessageGroupStore(){ + JedisConnectionFactory jcf = this.getConnectionFactoryForTest(); + RedisMessageStore store = new RedisMessageStore(jcf); + + MessageGroup messageGroup = store.getMessageGroup(1); + Message message = new GenericMessage("Hello"); + store.addMessageToGroup(1, message); + assertEquals(1, messageGroup.size()); + } + + @Test + @RedisAvailable + public void testRemoveMessageGroup(){ + JedisConnectionFactory jcf = this.getConnectionFactoryForTest(); + RedisMessageStore store = new RedisMessageStore(jcf); + + MessageGroup messageGroup = store.getMessageGroup(1); + Message message = new GenericMessage("Hello"); + ((RedisMessageGroup)messageGroup).add(message); + assertEquals(1, messageGroup.size()); + + store.removeMessageGroup(1); + messageGroup = store.getMessageGroup(1); + assertEquals(0, messageGroup.size()); + } + +}