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

INT-2737 RedisMessageStore Fix

Browse Source 
Fix RedisMessageStore to ensure that it
strips prefix from keys wheh they are returned
via iterator().

INT-2737 polishing

INT-2737 polishing
This commit is contained in:
Oleg Zhurakousky
2012-09-06 12:24:08 -04:00
committed by Gary Russell
parent 5df766b93d
commit 108baedd53
3 changed files with 154 additions and 124 deletions
Download Patch File Download Diff File Expand all files Collapse all files

108
spring-integration-core/src/main/java/org/springframework/integration/store/AbstractKeyValueMessageStore.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2011 the original author or authors
* Copyright 2002-2012 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.
@@ -18,8 +18,10 @@ package org.springframework.integration.store;
import java.util.ArrayList;
import java.util.Collection;
import java.util.HashSet;
import java.util.Iterator;
import java.util.Map;
import java.util.Set;
import java.util.UUID;
import org.springframework.beans.DirectFieldAccessor;
@@ -31,7 +33,7 @@ import org.springframework.util.Assert;
/**
* Base class for implementations of Key/Value style {@link MessageGroupStore} and {@link MessageStore}
*
*
* @author Oleg Zhurakousky
* @since 2.1
*/
@@ -40,11 +42,11 @@ public abstract class AbstractKeyValueMessageStore extends AbstractMessageGroupS
protected static final String MESSAGE_KEY_PREFIX = "MESSAGE_";
protected static final String MESSAGE_GROUP_KEY_PREFIX = "MESSAGE_GROUP_";
protected static final String CREATED_DATE = "CREATED_DATE";
// MessageStore methods
public Message<?> getMessage(UUID id) {
Message<?> message = this.getRawMessage(id);
if (message != null){
@@ -56,7 +58,7 @@ public abstract class AbstractKeyValueMessageStore extends AbstractMessageGroupS
@SuppressWarnings("unchecked")
public <T> Message<T> addMessage(Message<T> message) {
Assert.notNull(message, "'message' must not be null");
UUID messageId = message.getHeaders().getId();
UUID messageId = message.getHeaders().getId();
this.doStore(MESSAGE_KEY_PREFIX + messageId, message);
return (Message<T>) this.getRawMessage(messageId);
}
@@ -88,50 +90,50 @@ public abstract class AbstractKeyValueMessageStore extends AbstractMessageGroupS
public MessageGroup getMessageGroup(Object groupId) {
return this.buildMessageGroup(groupId, false);
}
/**
* Add a Message to the group with the provided group ID.
* 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(message, "'message' must not be null");
// add message as is to the MG accessible by the caller
SimpleMessageGroup messageGroup = this.getSimpleMessageGroup(this.getMessageGroup(groupId));
messageGroup.add(message);
SimpleMessageGroup messageGroup = this.getSimpleMessageGroup(this.getMessageGroup(groupId));
messageGroup.add(message);
// enrich Message with additional headers and add it to MS
Message<?> enrichedMessage = this.enrichMessage(message);
this.addMessage(enrichedMessage);
// build raw MessageGroup and add enriched Message to it
SimpleMessageGroup rawGroup = this.buildMessageGroup(groupId, true);
rawGroup.setLastModified(System.currentTimeMillis());
rawGroup.add(enrichedMessage);
// store MessageGroupMetadata built from enriched MG
this.doStore(MESSAGE_GROUP_KEY_PREFIX + groupId, new MessageGroupMetadata(rawGroup));
this.doStore(MESSAGE_GROUP_KEY_PREFIX + groupId, new MessageGroupMetadata(rawGroup));
// return clean MG
return this.getMessageGroup(groupId);
}
/**
* Remove a Message from the group with the provided group ID.
* 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");
// build raw MG
SimpleMessageGroup rawGroup = this.buildMessageGroup(groupId, true);
// create a clean instance of
// create a clean instance of
SimpleMessageGroup messageGroup = this.normalizeSimpleMessageGroup(rawGroup);
for (Message<?> message : rawGroup.getMessages()) {
if (message.getHeaders().getId().equals(messageToRemove.getHeaders().getId())){
rawGroup.remove(message);
@@ -139,13 +141,13 @@ public abstract class AbstractKeyValueMessageStore extends AbstractMessageGroupS
}
this.removeMessage(messageToRemove.getHeaders().getId());
rawGroup.setLastModified(System.currentTimeMillis());
this.doStore(MESSAGE_GROUP_KEY_PREFIX + groupId, new MessageGroupMetadata(rawGroup));
messageGroup = this.getSimpleMessageGroup(this.getMessageGroup(groupId));
return messageGroup;
}
public void completeGroup(Object groupId) {
Assert.notNull(groupId, "'groupId' must not be null");
@@ -156,7 +158,7 @@ public abstract class AbstractKeyValueMessageStore extends AbstractMessageGroupS
}
/**
* Remove the MessageGroup with the provided group ID.
* Remove the MessageGroup with the provided group ID.
*/
public void removeMessageGroup(Object groupId) {
Assert.notNull(groupId, "'groupId' must not be null");
@@ -164,7 +166,7 @@ public abstract class AbstractKeyValueMessageStore extends AbstractMessageGroupS
if (mgm != null) {
Assert.isInstanceOf(MessageGroupMetadata.class, mgm);
MessageGroupMetadata messageGroupMetadata = (MessageGroupMetadata) mgm;
Iterator<UUID> messageIds = messageGroupMetadata.messageIdIterator();
while (messageIds.hasNext()){
this.removeMessage(messageIds.next());
@@ -179,14 +181,14 @@ public abstract class AbstractKeyValueMessageStore extends AbstractMessageGroupS
messageGroup.setLastModified(System.currentTimeMillis());
this.doStore(MESSAGE_GROUP_KEY_PREFIX + groupId, new MessageGroupMetadata(messageGroup));
}
public Message<?> pollMessageFromGroup(Object groupId) {
Assert.notNull(groupId, "'groupId' must not be null");
Object mgm = this.doRetrieve(MESSAGE_GROUP_KEY_PREFIX + groupId);
if (mgm != null) {
Assert.isInstanceOf(MessageGroupMetadata.class, mgm);
MessageGroupMetadata messageGroupMetadata = (MessageGroupMetadata) mgm;
UUID firstId = messageGroupMetadata.firstId();
if (firstId != null){
messageGroupMetadata.remove(firstId);
@@ -198,11 +200,29 @@ public abstract class AbstractKeyValueMessageStore extends AbstractMessageGroupS
return null;
}
@SuppressWarnings("unchecked")
public Iterator<MessageGroup> iterator() {
final Iterator<?> idIterator = this.doListKeys(MESSAGE_GROUP_KEY_PREFIX + "*").iterator();
final Iterator<?> idIterator = this.normalizeKeys(
(Collection<String>) this.doListKeys(MESSAGE_GROUP_KEY_PREFIX + "*"))
.iterator();
return new MessageGroupIterator(idIterator);
}
private Collection<String> normalizeKeys(Collection<String> keys){
Set<String> normalizedKeys = new HashSet<String>();
for (Object key : keys) {
String strKey = (String) key;
if (strKey.startsWith(MESSAGE_GROUP_KEY_PREFIX)){
strKey = strKey.replace(MESSAGE_GROUP_KEY_PREFIX, "");
}
else if (strKey.startsWith(MESSAGE_KEY_PREFIX)){
strKey = strKey.replace(MESSAGE_KEY_PREFIX, "");
}
normalizedKeys.add(strKey);
}
return normalizedKeys;
}
public int messageGroupSize(Object groupId) {
Object mgm = this.doRetrieve(MESSAGE_GROUP_KEY_PREFIX + groupId);
if (mgm != null) {
@@ -212,15 +232,15 @@ public abstract class AbstractKeyValueMessageStore extends AbstractMessageGroupS
}
return 0;
}
protected abstract Object doRetrieve(Object id);
protected abstract void doStore(Object id, Object objectToStore);
protected abstract Object doRemove(Object id);
protected abstract Object doRemove(Object id);
protected abstract Collection<?> doListKeys(String keyPattern);
@SuppressWarnings({ "rawtypes", "unchecked" })
private Message<?> normalizeMessage(Message<?> message){
Message<?> normalizedMessage = MessageBuilder.fromMessage(message).removeHeader("CREATED_DATE").build();
@@ -229,7 +249,7 @@ public abstract class AbstractKeyValueMessageStore extends AbstractMessageGroupS
innerMap.put(MessageHeaders.TIMESTAMP, message.getHeaders().getTimestamp());
return normalizedMessage;
}
/**
* Will enrich Message with additional meta headers
* @param message
@@ -243,7 +263,7 @@ public abstract class AbstractKeyValueMessageStore extends AbstractMessageGroupS
innerMap.put(MessageHeaders.TIMESTAMP, message.getHeaders().getTimestamp());
return enrichedMessage;
}
private SimpleMessageGroup buildMessageGroup(Object groupId, boolean raw){
Assert.notNull(groupId, "'groupId' must not be null");
Object mgm = this.doRetrieve(MESSAGE_GROUP_KEY_PREFIX + groupId);
@@ -251,7 +271,7 @@ public abstract class AbstractKeyValueMessageStore extends AbstractMessageGroupS
Assert.isInstanceOf(MessageGroupMetadata.class, mgm);
MessageGroupMetadata messageGroupMetadata = (MessageGroupMetadata) mgm;
ArrayList<Message<?>> messages = new ArrayList<Message<?>>();
Iterator<UUID> messageIds = messageGroupMetadata.messageIdIterator();
while (messageIds.hasNext()){
if (raw){
@@ -261,8 +281,8 @@ public abstract class AbstractKeyValueMessageStore extends AbstractMessageGroupS
messages.add(this.getMessage(messageIds.next()));
}
}
SimpleMessageGroup messageGroup = new SimpleMessageGroup(messages,
SimpleMessageGroup messageGroup = new SimpleMessageGroup(messages,
groupId, messageGroupMetadata.getTimestamp(), messageGroupMetadata.isComplete());
messageGroup.setLastModified(messageGroupMetadata.getLastModified());
messageGroup.setLastReleasedMessageSequenceNumber(messageGroupMetadata.getLastReleasedMessageSequenceNumber());
@@ -281,7 +301,7 @@ public abstract class AbstractKeyValueMessageStore extends AbstractMessageGroupS
return new SimpleMessageGroup(messageGroup);
}
}
private SimpleMessageGroup normalizeSimpleMessageGroup(SimpleMessageGroup messageGroup){
SimpleMessageGroup normalizedGroup = new SimpleMessageGroup(messageGroup.getGroupId());
for (Message<?> message : messageGroup.getMessages()) {
@@ -290,10 +310,10 @@ public abstract class AbstractKeyValueMessageStore extends AbstractMessageGroupS
}
return normalizedGroup;
}
private Message<?> getRawMessage(UUID id) {
Assert.notNull(id, "'id' must not be null");
Object message = this.doRetrieve(MESSAGE_KEY_PREFIX + id);
Object message = this.doRetrieve(MESSAGE_KEY_PREFIX + id);
return (Message<?>) message;
}

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

@@ -1,5 +1,5 @@
/*
* Copyright 2007-2011 the original author or authors
* Copyright 2007-2012 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.
@@ -33,7 +33,7 @@ import org.springframework.util.Assert;
/**
* Redis implementation of the key/value style {@link MessageStore} and {@link MessageGroupStore}
*
*
* @author Oleg Zhurakousky
* @since 2.1
*/
@@ -52,7 +52,7 @@ public class RedisMessageStore extends AbstractKeyValueMessageStore {
Assert.notNull(valueSerializer, "'valueSerializer' must not be null");
this.redisTemplate.setValueSerializer(valueSerializer);
}
@Override
protected Object doRetrieve(Object id){
Assert.notNull(id, "'id' must not be null");
@@ -83,7 +83,7 @@ public class RedisMessageStore extends AbstractKeyValueMessageStore {
Object removedObject = this.doRetrieve(id);
if (removedObject != null){
redisTemplate.delete(id);
}
}
return removedObject;
}

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

@@ -1,5 +1,5 @@
/*
* Copyright 2007-2011 the original author or authors
* Copyright 2007-2012 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.
@@ -15,6 +15,12 @@
*/
package org.springframework.integration.redis.store;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertNotSame;
import static org.junit.Assert.assertNull;
import static org.junit.Assert.assertTrue;
import java.util.ArrayList;
import java.util.Iterator;
import java.util.List;
@@ -42,12 +48,6 @@ import org.springframework.integration.store.MessageGroup;
import org.springframework.integration.store.SimpleMessageGroup;
import org.springframework.integration.support.MessageBuilder;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertNotSame;
import static org.junit.Assert.assertNull;
import static org.junit.Assert.assertTrue;
/**
* @author Oleg Zhurakousky
*
@@ -56,19 +56,19 @@ public class RedisMessageGroupStoreTests extends RedisAvailableTests {
@Test
@RedisAvailable
public void testNonExistingEmptyMessageGroup() throws Exception{
public void testNonExistingEmptyMessageGroup() throws Exception{
JedisConnectionFactory jcf = this.getConnectionFactoryForTest();
RedisMessageStore store = new RedisMessageStore(jcf);
MessageGroup messageGroup = store.getMessageGroup(1);
assertNotNull(messageGroup);
assertTrue(messageGroup instanceof SimpleMessageGroup);
assertEquals(0, messageGroup.size());
}
@Test
@RedisAvailable
public void testMessageGroupUpdatedDateChangesWithEachAddedMessage() throws Exception{
@RedisAvailable
public void testMessageGroupUpdatedDateChangesWithEachAddedMessage() throws Exception{
JedisConnectionFactory jcf = this.getConnectionFactoryForTest();
RedisMessageStore store = new RedisMessageStore(jcf);
@@ -85,17 +85,17 @@ public class RedisMessageGroupStoreTests extends RedisAvailableTests {
createdTimestamp = messageGroup.getTimestamp();
updatedTimestamp = messageGroup.getLastModified();
assertTrue(updatedTimestamp > createdTimestamp);
// 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 testMessageGroupWithAddedMessage() throws Exception{
public void testMessageGroupWithAddedMessage() throws Exception{
JedisConnectionFactory jcf = this.getConnectionFactoryForTest();
RedisMessageStore store = new RedisMessageStore(jcf);
@@ -103,17 +103,17 @@ public class RedisMessageGroupStoreTests extends RedisAvailableTests {
Message<?> message = new GenericMessage<String>("Hello");
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 testRemoveMessageGroup() throws Exception{
public void testRemoveMessageGroup() throws Exception{
JedisConnectionFactory jcf = this.getConnectionFactoryForTest();
RedisMessageStore store = new RedisMessageStore(jcf);
@@ -121,26 +121,26 @@ public class RedisMessageGroupStoreTests extends RedisAvailableTests {
Message<?> message = new GenericMessage<String>("Hello");
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.getMessages().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.getMessages().size());
assertEquals(0, messageGroup.size());
}
@Test
@RedisAvailable
public void testCompleteMessageGroup() throws Exception{
public void testCompleteMessageGroup() throws Exception{
JedisConnectionFactory jcf = this.getConnectionFactoryForTest();
RedisMessageStore store = new RedisMessageStore(jcf);
@@ -151,10 +151,10 @@ public class RedisMessageGroupStoreTests extends RedisAvailableTests {
messageGroup = store.getMessageGroup(1);
assertTrue(messageGroup.isComplete());
}
@Test
@RedisAvailable
public void testLastReleasedSequenceNumber() throws Exception{
public void testLastReleasedSequenceNumber() throws Exception{
JedisConnectionFactory jcf = this.getConnectionFactoryForTest();
RedisMessageStore store = new RedisMessageStore(jcf);
@@ -165,10 +165,10 @@ public class RedisMessageGroupStoreTests extends RedisAvailableTests {
messageGroup = store.getMessageGroup(1);
assertEquals(5, messageGroup.getLastReleasedMessageSequenceNumber());
}
@Test
@RedisAvailable
public void testRemoveMessageFromTheGroup() throws Exception{
public void testRemoveMessageFromTheGroup() throws Exception{
JedisConnectionFactory jcf = this.getConnectionFactoryForTest();
RedisMessageStore store = new RedisMessageStore(jcf);
@@ -178,37 +178,37 @@ public class RedisMessageGroupStoreTests extends RedisAvailableTests {
store.addMessageToGroup(messageGroup.getGroupId(), message);
messageGroup = store.addMessageToGroup(messageGroup.getGroupId(), new GenericMessage<String>("3"));
assertEquals(3, messageGroup.size());
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 testWithMessageHistory() throws Exception{
public void testWithMessageHistory() throws Exception{
JedisConnectionFactory jcf = this.getConnectionFactoryForTest();
RedisMessageStore store = new RedisMessageStore(jcf);
store.getMessageGroup(1);
Message<?> message = new GenericMessage<String>("Hello");
DirectChannel fooChannel = new DirectChannel();
fooChannel.setBeanName("fooChannel");
DirectChannel barChannel = new DirectChannel();
barChannel.setBeanName("barChannel");
message = MessageHistory.write(message, fooChannel);
message = MessageHistory.write(message, barChannel);
store.addMessageToGroup(1, message);
message = store.getMessageGroup(1).getMessages().iterator().next();
MessageHistory messageHistory = MessageHistory.read(message);
assertNotNull(messageHistory);
assertEquals(2, messageHistory.size());
@@ -218,7 +218,7 @@ public class RedisMessageGroupStoreTests extends RedisAvailableTests {
}
@Test
@RedisAvailable
public void testRemoveNonExistingMessageFromTheGroup() throws Exception{
public void testRemoveNonExistingMessageFromTheGroup() throws Exception{
JedisConnectionFactory jcf = this.getConnectionFactoryForTest();
RedisMessageStore store = new RedisMessageStore(jcf);
@@ -226,61 +226,71 @@ public class RedisMessageGroupStoreTests extends RedisAvailableTests {
store.addMessageToGroup(messageGroup.getGroupId(), new GenericMessage<String>("1"));
store.removeMessageFromGroup(1, new GenericMessage<String>("2"));
}
@Test
@RedisAvailable
public void testRemoveNonExistingMessageFromNonExistingTheGroup() throws Exception{
public void testRemoveNonExistingMessageFromNonExistingTheGroup() throws Exception{
JedisConnectionFactory jcf = this.getConnectionFactoryForTest();
RedisMessageStore store = new RedisMessageStore(jcf);
store.removeMessageFromGroup(1, new GenericMessage<String>("2"));
}
@Test
@RedisAvailable
public void testMultipleInstancesOfGroupStore() throws Exception{
public void testMultipleInstancesOfGroupStore() throws Exception{
JedisConnectionFactory jcf = this.getConnectionFactoryForTest();
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.getMessages().size());
RedisMessageStore store3 = new RedisMessageStore(jcf);
messageGroup = store3.removeMessageFromGroup(1, message);
assertEquals(1, messageGroup.getMessages().size());
}
@Test
@RedisAvailable
public void testIteratorOfMessageGroups() throws Exception{
public void testIteratorOfMessageGroups() throws Exception{
JedisConnectionFactory jcf = this.getConnectionFactoryForTest();
RedisMessageStore store1 = new RedisMessageStore(jcf);
RedisMessageStore store2 = new RedisMessageStore(jcf);
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();
MessageGroup group = messageGroups.next();
String groupId = (String) group.getGroupId();
if (groupId.equals("1")){
assertEquals(1, group.getMessages().size());
}
else if (groupId.equals("2")) {
assertEquals(1, group.getMessages().size());
}
else if (groupId.equals("3")) {
assertEquals(2, group.getMessages().size());
}
counter++;
}
assertEquals(3, counter);
store2.removeMessageGroup(3);
messageGroups = store1.iterator();
counter = 0;
while (messageGroups.hasNext()) {
@@ -289,58 +299,58 @@ public class RedisMessageGroupStoreTests extends RedisAvailableTests {
}
assertEquals(2, counter);
}
@Test
@RedisAvailable @Ignore
public void testConcurrentModifications() throws Exception{
public void testConcurrentModifications() throws Exception{
JedisConnectionFactory jcf = this.getConnectionFactoryForTest();
final RedisMessageStore store1 = new RedisMessageStore(jcf);
final RedisMessageStore store2 = new RedisMessageStore(jcf);
final Message<?> message = new GenericMessage<String>("1");
final Message<?> message = new GenericMessage<String>("1");
ExecutorService executor = null;
final List<Object> failures = new ArrayList<Object>();
for (int i = 0; i < 100; i++) {
executor = Executors.newCachedThreadPool();
executor.execute(new Runnable() {
public void run() {
executor.execute(new Runnable() {
public void run() {
MessageGroup group = store1.addMessageToGroup(1, message);
if (group.getMessages().size() != 1){
failures.add("ADD");
throw new AssertionFailedError("Failed on ADD");
}
}
}
});
executor.execute(new Runnable() {
executor.execute(new Runnable() {
public void run() {
MessageGroup group = store2.removeMessageFromGroup(1, message);
if (group.getMessages().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(){
public void testWithAggregatorWithShutdown(){
this.getConnectionFactoryForTest(); // for this test it only ensures that DB was flushed before test
ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext("redis-aggregator-config.xml", this.getClass());
MessageChannel input = context.getBean("inputChannel", MessageChannel.class);
QueueChannel output = context.getBean("outputChannel", QueueChannel.class);
Message<?> m1 = MessageBuilder.withPayload("1").setSequenceNumber(1).setSequenceSize(3).setCorrelationId(1).build();
Message<?> m2 = MessageBuilder.withPayload("2").setSequenceNumber(2).setSequenceSize(3).setCorrelationId(1).build();
input.send(m1);
@@ -348,14 +358,14 @@ public class RedisMessageGroupStoreTests extends RedisAvailableTests {
input.send(m2);
assertNull(output.receive(1000));
context.close();
context = new ClassPathXmlApplicationContext("redis-aggregator-config.xml", this.getClass());
input = context.getBean("inputChannel", MessageChannel.class);
output = context.getBean("outputChannel", QueueChannel.class);
Message<?> m3 = MessageBuilder.withPayload("3").setSequenceNumber(3).setSequenceSize(3).setCorrelationId(1).build();
input.send(m3);
assertNotNull(output.receive(1000));
}
}