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

Merge pull request #34 from olegz/INT-2030

Browse Source 
INT-2030: RedisMessageStore
This commit is contained in:
Mark Fisher
2011-08-27 10:24:16 -04:00
parent 0aa87c4dc7 bffd40eaf4
commit d1ece40ddc
6 changed files with 324 additions and 458 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
build.gradle
View File

@@ -126,7 +126,7 @@ configure(javaprojects) {
springAmqpVersion = '1.0.0.RELEASE'
springDataMongoVersion = '1.0.0.BUILD-SNAPSHOT'
springDataCommonsVersion = '1.2.0.BUILD-SNAPSHOT'
springDataRedisVersion = '1.0.0.BUILD-SNAPSHOT'
springDataRedisVersion = '1.0.0.M4'
springGemfireVersion = '1.1.0.BUILD-SNAPSHOT'
springSecurityVersion = '3.0.5.RELEASE'
springWsVersion = '2.0.2.RELEASE'

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

@@ -1,250 +0,0 @@
/*
* 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
*
* 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.Collection;
import java.util.LinkedList;
import java.util.List;
import java.util.UUID;
import org.springframework.data.redis.core.BoundListOperations;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.integration.Message;
import org.springframework.integration.store.MessageGroup;
import org.springframework.integration.store.MessageStore;
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 String unmarkedId;
private final String markedId;
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;
this.redisTemplate = redisTemplate;
this.messageStore = messageStore;
this.unmarkedId = UNMARKED_PREFIX + groupId;
this.markedId = MARKED_PREFIX + groupId;
BoundListOperations<String, Object> unmarkedOps = this.redisTemplate.boundListOps(unmarkedId);
BoundListOperations<String, Object> markedOps = this.redisTemplate.boundListOps(markedId);
this.rebuildLocalCache(unmarkedOps, markedOps);
}
public boolean canAdd(Message<?> message) {
return !isMember(message);
}
public Collection<Message<?>> getUnmarked() {
return this.unmarked;
}
public Collection<Message<?>> getMarked() {
return this.marked;
}
public Object getGroupId() {
return this.groupId;
}
public boolean isComplete() {
if (size() == 0) {
return true;
}
int sequenceSize = getSequenceSize();
// If there is no sequence then it must be incomplete....
return sequenceSize > 0 && sequenceSize == size();
}
public int getSequenceSize() {
if (size() == 0) {
return 0;
}
return getOne().getHeaders().getSequenceSize();
}
public int size() {
synchronized (lock) {
return marked.size() + unmarked.size();
}
}
public Message<?> getOne() {
synchronized (lock) {
Message<?> one = unmarked.get(0);
if (one == null) {
one = marked.get(0);
}
return one;
}
}
public long getTimestamp() {
return this.timestamp;
}
protected void markAll() {
BoundListOperations<String, Object> unmarkedOps = this.redisTemplate.boundListOps(unmarkedId);
BoundListOperations<String, Object> markedOps = this.redisTemplate.boundListOps(markedId);
long uSize = unmarkedOps.size();
synchronized (lock) {
unmarkedOps.rename(markedId);
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(unmarkedId);
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(markedId);
markedOps.rightPush(id);
unmarkedOps.remove(0, id);
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(unmarkedId);
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(unmarkedId);
BoundListOperations<String, Object> markedOps = this.redisTemplate.boundListOps(markedId);
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> unmarkedOps = this.redisTemplate.boundListOps(unmarkedId);
List<Object> messageIds = unmarkedOps.range(0, unmarkedOps.size()-1);
for (Object messageId : messageIds) {
this.messageStore.removeMessage(UUID.fromString(messageId.toString()));
}
this.redisTemplate.delete(unmarkedId);
BoundListOperations<String, Object> markedOps = this.redisTemplate.boundListOps(markedId);
messageIds = markedOps.range(0, markedOps.size()-1);
for (Object messageId : messageIds) {
this.messageStore.removeMessage(UUID.fromString(messageId.toString()));
}
this.redisTemplate.delete(markedId);
this.rebuildLocalCache(unmarkedOps, markedOps);
}
private Collection<Message<?>> buildMessageList(BoundListOperations<String, Object> mGroupOps){
List<Message<?>> messages = new LinkedList<Message<?>>();
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(unmarkedId);
Collection<Message<?>> unmarked = this.buildMessageList(mGroupOps);
mGroupOps = this.redisTemplate.boundListOps(markedId);
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));
}
}
}

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

@@ -13,19 +13,22 @@
* 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.HashMap;
import java.util.Collection;
import java.util.Collections;
import java.util.Iterator;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import java.util.Set;
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.BoundListOperations;
import org.springframework.data.redis.core.BoundSetOperations;
import org.springframework.data.redis.core.BoundValueOperations;
import org.springframework.data.redis.core.RedisCallback;
@@ -33,59 +36,70 @@ import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.serializer.JdkSerializationRedisSerializer;
import org.springframework.data.redis.serializer.RedisSerializer;
import org.springframework.data.redis.serializer.SerializationException;
import org.springframework.data.redis.serializer.StringRedisSerializer;
import org.springframework.integration.Message;
import org.springframework.integration.store.AbstractMessageGroupStore;
import org.springframework.integration.store.MessageGroup;
import org.springframework.integration.store.MessageGroupStore;
import org.springframework.integration.store.MessageStore;
import org.springframework.integration.store.MessageStoreException;
import org.springframework.integration.store.SimpleMessageGroup;
import org.springframework.jmx.export.annotation.ManagedAttribute;
import org.springframework.util.Assert;
/**
* An implementation of both the {@link MessageStore} and {@link MessageGroupStore}
* strategies that relies upon Redis for persistence.
*
* @author Oleg Zhurakousky
* @since 2.1
*
*/
public class RedisMessageStore extends AbstractMessageGroupStore implements MessageStore, InitializingBean{
private final String MESSAGE_GROUPS_KEY = "MESSAGE_GROUPS";
public class RedisMessageStore extends AbstractMessageGroupStore implements MessageStore {
private static final String MESSAGE_GROUPS_KEY = "MESSAGE_GROUPS";
private static final String MARKED_PREFIX = "MARKED_";
private static final String UNMARKED_PREFIX = "UNMARKED_";
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){
public RedisMessageStore(RedisConnectionFactory connectionFactory) {
this.redisTemplate = new RedisTemplate<String, Object>();
this.redisTemplate.setConnectionFactory(connectionFactory);
this.redisTemplate.setKeySerializer(new KeySerializer());
this.redisTemplate.setValueSerializer(this.valueSerializer);
this.redisTemplate.setKeySerializer(new StringRedisSerializer());
this.redisTemplate.setValueSerializer(new JdkSerializationRedisSerializer());
}
public void setValueSerializer(RedisSerializer<?> valueSerializer) {
Assert.notNull(valueSerializer, "'valueSerializer' must not be null");
this.redisTemplate.setValueSerializer(valueSerializer);
}
public Message<?> getMessage(final UUID id) {
Assert.notNull(id, "'id' must not be null");
if (this.keyExists(id)){
if (this.redisTemplate.hasKey(id.toString())) {
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;
return null;
}
@SuppressWarnings("unchecked")
public <T> Message<T> addMessage(Message<T> message) {
Assert.notNull(message, "'message' must not be null");
BoundValueOperations<String, Object> ops = redisTemplate.boundValueOps(message.getHeaders().getId().toString());
BoundValueOperations<String, Object> ops = redisTemplate.boundValueOps(message.getHeaders().getId().toString());
try {
ops.set(message);
} catch (SerializationException e) {
throw new MessageStoreException(message, "It seems like while relying on the default RedisSerializer (JdkSerializationRedisSerializer) " +
"the Message contains data that is not Serializable. Either make it Serializable or provide your own implementation of " +
}
catch (SerializationException e) {
throw new MessageStoreException(message, "If relying on the default RedisSerializer (JdkSerializationRedisSerializer) " +
"the Message must be Serializable. Either make it Serializable or provide your own implementation of " +
"RedisSerializer via 'setValueSerializer(..)'", e);
}
Object result = ops.get();
@@ -93,131 +107,187 @@ public class RedisMessageStore extends AbstractMessageGroupStore implements Mess
return (Message<T>) result;
}
public Message<?> removeMessage(UUID id) {
Assert.notNull(id, "'id' must not be null");
Message<?> message = this.getMessage(id);
if (message != null){
if (message != null) {
this.redisTemplate.delete(id.toString());
return message;
}
else {
throw new IllegalArgumentException("Message with id '" + id + "' can not be removed since it does NOT exist");
}
return message;
}
@ManagedAttribute
public long 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;
return redisTemplate.execute(new RedisCallback<Long>() {
public Long doInRedis(RedisConnection connection) throws DataAccessException {
return connection.dbSize();
}
});
}
public void setValueSerializer(RedisSerializer<?> valueSerializer) {
this.valueSerializer = valueSerializer;
}
private boolean keyExists(final UUID id){
return redisTemplate.execute(new RedisCallback<Boolean>() {
public Boolean doInRedis(RedisConnection connection)
throws DataAccessException {
return connection.exists(id.toString().getBytes());
}
});
}
private static class KeySerializer implements RedisSerializer<String> {
public byte[] serialize(String value) throws SerializationException {
return value.getBytes();
}
public String deserialize(byte[] bytes) throws SerializationException {
return new String(bytes);
}
}
// MESSAGE GROUP methods
public void afterPropertiesSet() throws Exception {
Assert.notNull(this.valueSerializer, "'valueSerializer' must not be null");
}
/**
*
* Will create a new instance of SimpleMessageGroup initializing it with
* data collected from the Redis Message Store.
*/
public MessageGroup getMessageGroup(Object groupId) {
Assert.notNull(groupId, "'groupId' must not be null");
RedisMessageGroup group = this.messageGroups.get(groupId);
if (group == null){
synchronized (lock) {
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 group;
long timestamp = System.currentTimeMillis();
Collection<Message<?>> unmarkedMessages = this.buildMessageList(this.redisTemplate.boundListOps(UNMARKED_PREFIX + groupId));
Collection<Message<?>> markedMessages = this.buildMessageList(this.redisTemplate.boundListOps(MARKED_PREFIX + groupId));
this.doCreateMessageGroupIfNecessary(groupId);
return new SimpleMessageGroup(unmarkedMessages, markedMessages, groupId, timestamp);
}
/**
*
* Add a Message to the group with the provided group ID.
*/
public MessageGroup addMessageToGroup(Object groupId, Message<?> message) {
Assert.notNull(groupId, "'groupId' must not be null");
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;
synchronized (groupId) {
this.doAddMessageToGroup(message, groupId);
this.addMessage(message);
return this.getMessageGroup(groupId);
}
}
/**
* Mark all messages in the provided group.
*/
public MessageGroup markMessageGroup(MessageGroup group) {
Assert.isInstanceOf(RedisMessageGroup.class, group, "MessageGroup is not an instance of RedisMessageGroup");
((RedisMessageGroup)group).markAll();
return group;
Assert.notNull(group, "'group' must not be null");
Object groupId = group.getGroupId();
synchronized (groupId) {
this.doMarkMessageGroup(groupId);
return this.getMessageGroup(groupId);
}
}
public MessageGroup removeMessageFromGroup(Object key,
Message<?> messageToRemove) {
RedisMessageGroup messageGroup = (RedisMessageGroup) this.getMessageGroup(key);
messageGroup.remove(messageToRemove);
return messageGroup;
/**
* Remove a Message from the group with the provided group ID.
*/
public MessageGroup removeMessageFromGroup(Object groupId, Message<?> messageToRemove) {
Assert.notNull(groupId, "'groupId' must not be null");
Assert.notNull(messageToRemove, "'messageToRemove' must not be null");
UUID messageId = messageToRemove.getHeaders().getId();
synchronized (groupId) {
this.doRemoveMessageFromGroup(groupId, messageId);
this.removeMessage(messageId);
return this.getMessageGroup(groupId);
}
}
public MessageGroup markMessageFromGroup(Object key,
Message<?> messageToMark) {
RedisMessageGroup messageGroup = (RedisMessageGroup) this.getMessageGroup(key);
messageGroup.markMessage(messageToMark.getHeaders().getId().toString());
return messageGroup;
/**
* Mark the given Message within the group corresponding to the provided group ID.
*/
public MessageGroup markMessageFromGroup(Object groupId, Message<?> messageToMark) {
Assert.notNull(groupId, "'groupId' must not be null");
Assert.notNull(messageToMark, "'messageToMark' must not be null");
String messageIdAsString = messageToMark.getHeaders().getId().toString();
synchronized (groupId) {
this.doMarkMessageFromGroup(messageIdAsString, groupId);
return this.getMessageGroup(groupId);
}
}
/**
* Remove the MessageGroup with the provided group ID.
*/
public void removeMessageGroup(Object groupId) {
Assert.notNull(groupId, "'groupId' must not be null");
RedisMessageGroup messageGroup = (RedisMessageGroup) this.getMessageGroup(groupId);
synchronized (messageGroup) {
messageGroup.destroy();
BoundSetOperations<String, Object> mGroupsOps = this.redisTemplate.boundSetOps(MESSAGE_GROUPS_KEY);
mGroupsOps.remove(groupId);
this.redisTemplate.delete(groupId.toString());
this.messageGroups.remove(groupId);
synchronized (groupId) {
this.doRemoveMessageGroup(groupId);
}
}
@Override
public Iterator<MessageGroup> iterator() {
BoundSetOperations<String, Object> mGroupsOps = this.redisTemplate.boundSetOps(MESSAGE_GROUPS_KEY);
Set<Object> messageGroupIds = mGroupsOps.members();
List<MessageGroup> messageGroups = new ArrayList<MessageGroup>();
for (Object msgGroupId : mGroupsOps.members()) {
messageGroups.add(new RedisMessageGroup(this, this.redisTemplate, msgGroupId));
for (Object messageGroupId : messageGroupIds) {
messageGroups.add(this.getMessageGroup(messageGroupId));
}
return messageGroups.iterator();
}
private Collection<Message<?>> buildMessageList(BoundListOperations<String, Object> messageGroupOps) {
List<Message<?>> messages = new LinkedList<Message<?>>();
if (messageGroupOps.size() == 0) {
return Collections.emptyList();
}
List<Object> messageIds = messageGroupOps.range(0, messageGroupOps.size() - 1);
for (Object messageId : messageIds) {
Message<?> message = this.getMessage(UUID.fromString(messageId.toString()));
if (message != null) {
messages.add((Message<?>) message);
}
}
return messages;
}
/* candidates for future abstract methods */
private void doCreateMessageGroupIfNecessary(Object groupId) {
BoundSetOperations<String, Object> messageGroupsOps = this.redisTemplate.boundSetOps(MESSAGE_GROUPS_KEY);
if (!messageGroupsOps.members().contains(groupId)) {
messageGroupsOps.add(groupId);
}
}
private void doAddMessageToGroup(Message<?> message, Object groupId) {
String messageId = message.getHeaders().getId().toString();
BoundListOperations<String, Object> unmarkedOps = this.redisTemplate.boundListOps(UNMARKED_PREFIX + groupId);
unmarkedOps.rightPush(messageId);
}
private void doMarkMessageGroup(Object groupId) {
BoundListOperations<String, Object> unmarkedOps = this.redisTemplate.boundListOps(UNMARKED_PREFIX + groupId);
unmarkedOps.rename(MARKED_PREFIX + groupId);
}
private void doRemoveMessageFromGroup(Object groupId, UUID messageId) {
BoundListOperations<String, Object> unmarkedOps = this.redisTemplate.boundListOps(UNMARKED_PREFIX + groupId);
BoundListOperations<String, Object> markedOps = this.redisTemplate.boundListOps(MARKED_PREFIX + groupId);
unmarkedOps.remove(0, messageId.toString());
markedOps.remove(0, messageId.toString());
}
private void doMarkMessageFromGroup(String messageIdAsString, Object groupId) {
BoundListOperations<String, Object> unmarkedOps = this.redisTemplate.boundListOps(UNMARKED_PREFIX + groupId);
if (unmarkedOps.size() > 0) {
List<Object> messageIds = unmarkedOps.range(0, unmarkedOps.size() - 1);
int objectIndex = messageIds.indexOf(messageIdAsString);
if (objectIndex > -1) {
BoundListOperations<String, Object> markedOps = this.redisTemplate.boundListOps(MARKED_PREFIX + groupId);
markedOps.rightPush(messageIdAsString);
unmarkedOps.remove(0, messageIdAsString);
}
}
}
private void doRemoveMessageGroup(Object groupId) {
BoundListOperations<String, Object> unmarkedOps = this.redisTemplate.boundListOps(UNMARKED_PREFIX + groupId);
if (unmarkedOps.size() > 0) {
List<Object> messageIds = unmarkedOps.range(0, unmarkedOps.size() - 1);
for (Object messageId : messageIds) {
this.removeMessage(UUID.fromString(messageId.toString()));
}
this.redisTemplate.delete(UNMARKED_PREFIX + groupId);
}
BoundListOperations<String, Object> markedOps = this.redisTemplate.boundListOps(MARKED_PREFIX + groupId);
if (markedOps.size() > 0) {
List<Object> messageIds = markedOps.range(0, markedOps.size() - 1);
for (Object messageId : messageIds) {
this.removeMessage(UUID.fromString(messageId.toString()));
}
this.redisTemplate.delete(MARKED_PREFIX + groupId);
}
BoundSetOperations<String, Object> messageGroupsOps = this.redisTemplate.boundSetOps(MESSAGE_GROUPS_KEY);
messageGroupsOps.remove(groupId);
}
}

1
spring-integration-redis/src/test/java/org/springframework/integration/redis/config/RedisChannelParserTests.java
View File

@@ -65,7 +65,6 @@ public class RedisChannelParserTests extends RedisAvailableTests{
redisChannel.send(m);
Thread.sleep(1000);
Mockito.verify(marker, Mockito.times(1)).mark();
System.out.println("done");
}
interface Marker {

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

@@ -15,6 +15,15 @@
*/
package org.springframework.integration.redis.store;
import java.util.ArrayList;
import java.util.Iterator;
import java.util.List;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import junit.framework.AssertionFailedError;
import org.junit.Test;
import org.springframework.context.support.ClassPathXmlApplicationContext;
import org.springframework.data.redis.connection.jedis.JedisConnectionFactory;
@@ -25,6 +34,7 @@ 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.store.SimpleMessageGroup;
import org.springframework.integration.support.MessageBuilder;
import static org.junit.Assert.assertEquals;
@@ -41,53 +51,57 @@ public class RedisMessageGroupStoreTests extends RedisAvailableTests {
@Test
@RedisAvailable
public void testNonExistingEmptyMessageGroup(){
public void testNonExistingEmptyMessageGroup() throws Exception{
JedisConnectionFactory jcf = this.getConnectionFactoryForTest();
RedisMessageStore store = new RedisMessageStore(jcf);
MessageGroup messageGroup = store.getMessageGroup(1);
assertNotNull(messageGroup);
assertTrue(messageGroup instanceof RedisMessageGroup);
assertTrue(messageGroup instanceof SimpleMessageGroup);
assertEquals(0, messageGroup.size());
}
@Test
@RedisAvailable
public void testMessageGroupWithAddedMessageViaMessageGroup(){
public void testMessageGroupWithAddedMessage() throws Exception{
JedisConnectionFactory jcf = this.getConnectionFactoryForTest();
RedisMessageStore store = new RedisMessageStore(jcf);
MessageGroup messageGroup = store.getMessageGroup(1);
Message<?> message = new GenericMessage<String>("Hello");
((RedisMessageGroup)messageGroup).add(message);
messageGroup = store.addMessageToGroup(1, message);
assertEquals(1, messageGroup.size());
// make sure the store is properly rebuild from Redis
store = new RedisMessageStore(jcf);
messageGroup = store.getMessageGroup(1);
assertEquals(1, messageGroup.size());
}
@Test
@RedisAvailable
public void testMessageGroupWithAddedMessageViaMessageGroupStore(){
public void testRemoveMessageGroup() throws Exception{
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);
messageGroup = store.addMessageToGroup(messageGroup.getGroupId(), message);
assertEquals(1, messageGroup.size());
store.removeMessageGroup(1);
MessageGroup messageGroupA = store.getMessageGroup(1);
assertNotSame(messageGroup, messageGroupA);
assertEquals(0, messageGroupA.getMarked().size());
assertEquals(0, messageGroupA.getUnmarked().size());
assertEquals(0, messageGroupA.size());
// make sure the store is properly rebuild from Redis
store = new RedisMessageStore(jcf);
messageGroup = store.getMessageGroup(1);
assertEquals(0, messageGroup.getMarked().size());
assertEquals(0, messageGroup.getUnmarked().size());
assertEquals(0, messageGroup.size());
@@ -95,136 +109,177 @@ public class RedisMessageGroupStoreTests extends RedisAvailableTests {
@Test
@RedisAvailable
public void testRemoveMessageFromTheGroup(){
public void testRemoveMessageFromTheGroup() throws Exception{
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"));
store.addMessageToGroup(messageGroup.getGroupId(), new GenericMessage<String>("1"));
store.addMessageToGroup(messageGroup.getGroupId(), message);
messageGroup = store.addMessageToGroup(messageGroup.getGroupId(), new GenericMessage<String>("3"));
assertEquals(3, messageGroup.size());
store.removeMessageFromGroup(1, message);
messageGroup = store.removeMessageFromGroup(1, message);
assertEquals(2, messageGroup.size());
// make sure the store is properly rebuild from Redis
store = new RedisMessageStore(jcf);
messageGroup = store.getMessageGroup(1);
assertEquals(2, messageGroup.size());
}
@Test
@RedisAvailable
public void testMarkAllMessagesInMessageGroup(){
public void testMarkAllMessagesInMessageGroup() throws Exception{
JedisConnectionFactory jcf = this.getConnectionFactoryForTest();
RedisMessageStore store = new RedisMessageStore(jcf);
MessageGroup messageGroup = store.getMessageGroup(1);
((RedisMessageGroup)messageGroup).add(new GenericMessage<String>("1"));
((RedisMessageGroup)messageGroup).add(new GenericMessage<String>("2"));
((RedisMessageGroup)messageGroup).add(new GenericMessage<String>("3"));
store.addMessageToGroup(messageGroup.getGroupId(), new GenericMessage<String>("1"));
store.addMessageToGroup(messageGroup.getGroupId(), new GenericMessage<String>("2"));
messageGroup = store.addMessageToGroup(messageGroup.getGroupId(), new GenericMessage<String>("3"));
assertEquals(3, messageGroup.getUnmarked().size());
assertEquals(0, messageGroup.getMarked().size());
((RedisMessageGroup)messageGroup).markAll();
messageGroup = store.markMessageGroup(messageGroup);
assertEquals(0, messageGroup.getUnmarked().size());
assertEquals(3, messageGroup.getMarked().size());
// make sure the store is properly rebuild from Redis
store = new RedisMessageStore(jcf);
messageGroup = store.getMessageGroup(1);
assertEquals(0, messageGroup.getUnmarked().size());
assertEquals(3, messageGroup.getMarked().size());
}
@Test
@RedisAvailable
public void testMarkMessageInMessageGroup(){
public void testMarkMessageInMessageGroup() throws Exception{
JedisConnectionFactory jcf = this.getConnectionFactoryForTest();
RedisMessageStore store = new RedisMessageStore(jcf);
MessageGroup messageGroup = store.getMessageGroup(1);
Message<?> messageToMark = new GenericMessage<String>("1");
((RedisMessageGroup)messageGroup).add(messageToMark);
((RedisMessageGroup)messageGroup).add(new GenericMessage<String>("2"));
((RedisMessageGroup)messageGroup).add(new GenericMessage<String>("3"));
store.addMessageToGroup(messageGroup.getGroupId(), messageToMark);
store.addMessageToGroup(messageGroup.getGroupId(), new GenericMessage<String>("2"));
messageGroup = store.addMessageToGroup(messageGroup.getGroupId(), new GenericMessage<String>("3"));
assertEquals(3, messageGroup.getUnmarked().size());
assertEquals(0, messageGroup.getMarked().size());
store.markMessageFromGroup(1, messageToMark);
messageGroup = store.markMessageFromGroup(1, messageToMark);
assertEquals(2, messageGroup.getUnmarked().size());
assertEquals(1, messageGroup.getMarked().size());
// make sure the store is properly rebuild from Redis
store = new RedisMessageStore(jcf);
messageGroup = store.getMessageGroup(1);
assertEquals(2, messageGroup.getUnmarked().size());
assertEquals(1, messageGroup.getMarked().size());
}
@Test
@RedisAvailable
public void testGetOneFromMessageGroup(){
public void testMultipleInstancesOfGroupStore() throws Exception{
JedisConnectionFactory jcf = this.getConnectionFactoryForTest();
RedisMessageStore store = new RedisMessageStore(jcf);
MessageGroup messageGroup = store.getMessageGroup(1);
((RedisMessageGroup)messageGroup).add(new GenericMessage<String>("1"));
((RedisMessageGroup)messageGroup).add(new GenericMessage<String>("2"));
((RedisMessageGroup)messageGroup).add(new GenericMessage<String>("3"));
Message<?> message = messageGroup.getOne();
assertEquals("1", message.getPayload());
RedisMessageStore store1 = new RedisMessageStore(jcf);
RedisMessageStore store2 = new RedisMessageStore(jcf);
Message<?> message = new GenericMessage<String>("1");
store1.addMessageToGroup(1, message);
MessageGroup messageGroup = store2.addMessageToGroup(1, new GenericMessage<String>("2"));
assertEquals(2, messageGroup.getUnmarked().size());
assertEquals(0, messageGroup.getMarked().size());
RedisMessageStore store3 = new RedisMessageStore(jcf);
messageGroup = store3.markMessageFromGroup(1, message);
assertEquals(1, messageGroup.getUnmarked().size());
assertEquals(1, messageGroup.getMarked().size());
}
@Test
@RedisAvailable
public void testAddToMessageGroup(){
public void testIteratorOfMessageGroups() throws Exception{
JedisConnectionFactory jcf = this.getConnectionFactoryForTest();
RedisMessageStore store = new RedisMessageStore(jcf);
MessageGroup messageGroup = store.getMessageGroup(1);
RedisMessageStore store1 = new RedisMessageStore(jcf);
RedisMessageStore store2 = new RedisMessageStore(jcf);
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());
store1.addMessageToGroup(1, new GenericMessage<String>("1"));
store2.addMessageToGroup(2, new GenericMessage<String>("2"));
store1.addMessageToGroup(3, new GenericMessage<String>("3"));
store2.addMessageToGroup(3, new GenericMessage<String>("3A"));
Iterator<MessageGroup> messageGroups = store1.iterator();
int counter = 0;
while (messageGroups.hasNext()) {
messageGroups.next();
counter++;
}
assertEquals(3, counter);
store2.removeMessageGroup(3);
messageGroups = store1.iterator();
counter = 0;
while (messageGroups.hasNext()) {
messageGroups.next();
counter++;
}
assertEquals(2, counter);
}
@Test
@RedisAvailable
public void testCanAddToMessageGroup(){
public void testConcurrentModifications() throws Exception{
JedisConnectionFactory jcf = this.getConnectionFactoryForTest();
RedisMessageStore store = new RedisMessageStore(jcf);
final RedisMessageStore store1 = new RedisMessageStore(jcf);
final RedisMessageStore store2 = 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);
final Message<?> message = new GenericMessage<String>("1");
MessageGroup mg1 = store.getMessageGroup(1);
MessageGroup mg2 = store.getMessageGroup(1);
ExecutorService executor = null;
assertEquals(mg1, mg2);
final List<Object> failures = new ArrayList<Object>();
store.removeMessageGroup(1);
mg2 = store.getMessageGroup(1);
assertNotSame(mg1, mg2);
for (int i = 0; i < 100; i++) {
executor = Executors.newCachedThreadPool();
executor.execute(new Runnable() {
public void run() {
MessageGroup group = store1.addMessageToGroup(1, message);
if (group.getUnmarked().size() != 1){
failures.add("ADD");
throw new AssertionFailedError("Failed on ADD");
}
}
});
executor.execute(new Runnable() {
public void run() {
MessageGroup group = store2.removeMessageFromGroup(1, message);
if (group.getUnmarked().size() != 0){
failures.add("REMOVE");
throw new AssertionFailedError("Failed on Remove");
}
}
});
executor.shutdown();
executor.awaitTermination(10, TimeUnit.SECONDS);
store2.removeMessageFromGroup(1, message); // ensures that if ADD thread executed after REMOVE, the store is empty for the next cycle
}
assertTrue(failures.size() == 0);
}
@Test
@RedisAvailable
public void testWithAggregatorWithShutdown(){

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

@@ -116,14 +116,6 @@ public class RedisMessageStoreTests extends RedisAvailableTests {
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());
}
@SuppressWarnings("serial")
public static class Person implements Serializable{
private Address address;