INT-2030: RedisMessageStore
refactored RedisMessageStore to use SimpleMessageGroup. Deleted RedisMessageGroup polished RedisMessageStore, made it more inline with JdbcMessageStore, added additional tests more polishing on RedisMessageStore, added more tests removed test that was no longer valid. Downgraded spring-data-redis dependency to M4 changed UUID serializer to be StringRedisSerializer instead of custom class which essentially did the same thing polished code based on the review comments. Rename onX methods to doX, removed this.hasKey in favor of this.redisTemplate.hasKey polishing removed afterPropertiesSet() calls in test
This commit is contained in:
Oleg Zhurakousky
committed by
Mark Fisher
Mark Fisher
parent
0aa87c4dc7
commit
bffd40eaf4
@@ -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'
|
||||
|
||||
@@ -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));
|
||||
}
|
||||
}
|
||||
|
||||
}
|
||||
@@ -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);
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
@@ -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 {
|
||||
|
||||
@@ -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(){
|
||||
|
||||
@@ -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;
|
||||
|
||||
Reference in New Issue
Block a user