Spring/spring-integration
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

INT-2030 added initial support for Redis MessageGroupStore

Browse Source
This commit is contained in:
Oleg Zhurakousky
2011-08-10 12:39:36 -04:00
parent 835dab4683
commit 665e76e50a
4 changed files with 317 additions and 19 deletions
Download Patch File Download Diff File Expand all files Collapse all files

140
spring-integration-redis/src/main/java/org/springframework/integration/redis/store/RedisMessageGroup.java Normal file
View File

@@ -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<String, Object> redisTemplate;
public RedisMessageGroup(RedisTemplate<String, Object> 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<String, Object> mGroupOps = this.redisTemplate.boundSetOps(UNMARKED_PREFIX + this.groupId.toString());
String key = message.getHeaders().getId().toString();
mGroupOps.add(key);
BoundValueOperations<String, Object> bvOps = this.redisTemplate.boundValueOps(key);
bvOps.set(message);
}
public void destroy(){
// delete unmarked
BoundSetOperations<String, Object> 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<Message<?>> getUnmarked() {
BoundSetOperations<String, Object> mGroupOps = this.redisTemplate.boundSetOps(UNMARKED_PREFIX + this.groupId.toString());
return this.getMessages(mGroupOps);
}
public Collection<Message<?>> getMarked() {
BoundSetOperations<String, Object> 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<String, Object> 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<Message<?>> getMessages(BoundSetOperations<String, Object> mGroupOps){
List<Message<?>> messages = new ArrayList<Message<?>>();
for (Object messageId : mGroupOps.members()) {
BoundValueOperations<String, Object> 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;
}
}

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

@@ -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<UUID, Message<?>> redisTemplate;
private final RedisTemplate<String, Object> redisTemplate;
private volatile RedisSerializer<?> valueSerializer = new JdkSerializationRedisSerializer();
public RedisMessageStore(RedisConnectionFactory connectionFactory){
this.redisTemplate = new RedisTemplate<UUID, Message<?>>();
this.redisTemplate = new RedisTemplate<String, Object>();
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<UUID, Message<?>> ops = redisTemplate.boundValueOps(id);
return ops.get();
if (this.keyExists(id)){
BoundValueOperations<String, Object> 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 <T> Message<T> addMessage(Message<T> message) {
Assert.notNull(message, "'message' must not be null");
BoundValueOperations<UUID, Message<?>> ops = redisTemplate.boundValueOps(message.getHeaders().getId());
BoundValueOperations<String, Object> 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<T>) ops.get();
Object result = ops.get();
Assert.isInstanceOf(Message.class, result, "Return value is not an instace of Message");
return (Message<T>) 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<Boolean>() {
public Boolean doInRedis(RedisConnection connection)
throws DataAccessException {
@@ -113,14 +124,14 @@ public class RedisMessageStore implements MessageStore, InitializingBean{
});
}
private static class UuidSerializer implements RedisSerializer<UUID> {
private static class KeySerializer implements RedisSerializer<String> {
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<String, Object> 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<MessageGroup> iterator() {
BoundSetOperations<String, Object> mGroupsOps = this.redisTemplate.boundSetOps(MESSAGE_GROUPS_KEY);
List<MessageGroup> messageGroups = new ArrayList<MessageGroup>();
for (Object msgGroupId : mGroupsOps.members()) {
messageGroups.add(new RedisMessageGroup(this.redisTemplate, msgGroupId));
}
return messageGroups.iterator();
}
}

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

@@ -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<UUID, Message<?>>();
RedisTemplate rt = new RedisTemplate<UUID, Object>();
rt.setConnectionFactory(jcf);
rt.execute(new RedisCallback() {

88
spring-integration-redis/src/test/java/org/springframework/integration/redis/store/RedisMessageGroupStoreTests.java Normal file
View File

@@ -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<String>("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<String>("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<String>("Hello");
((RedisMessageGroup)messageGroup).add(message);
assertEquals(1, messageGroup.size());
store.removeMessageGroup(1);
messageGroup = store.getMessageGroup(1);
assertEquals(0, messageGroup.size());
}
}