From 108baedd53e01efc18af216731ddfb6e4074fa96 Mon Sep 17 00:00:00 2001 From: Oleg Zhurakousky Date: Thu, 6 Sep 2012 12:24:08 -0400 Subject: [PATCH] INT-2737 RedisMessageStore Fix Fix RedisMessageStore to ensure that it strips prefix from keys wheh they are returned via iterator(). INT-2737 polishing INT-2737 polishing --- .../store/AbstractKeyValueMessageStore.java | 108 +++++++----- .../redis/store/RedisMessageStore.java | 8 +- .../store/RedisMessageGroupStoreTests.java | 162 ++++++++++-------- 3 files changed, 154 insertions(+), 124 deletions(-) diff --git a/spring-integration-core/src/main/java/org/springframework/integration/store/AbstractKeyValueMessageStore.java b/spring-integration-core/src/main/java/org/springframework/integration/store/AbstractKeyValueMessageStore.java index 10902b69f6..7cccd5131b 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/store/AbstractKeyValueMessageStore.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/store/AbstractKeyValueMessageStore.java @@ -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 Message addMessage(Message 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) 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 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 iterator() { - final Iterator idIterator = this.doListKeys(MESSAGE_GROUP_KEY_PREFIX + "*").iterator(); + final Iterator idIterator = this.normalizeKeys( + (Collection) this.doListKeys(MESSAGE_GROUP_KEY_PREFIX + "*")) + .iterator(); return new MessageGroupIterator(idIterator); } - + + private Collection normalizeKeys(Collection keys){ + Set normalizedKeys = new HashSet(); + 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> messages = new ArrayList>(); - + Iterator 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; } diff --git a/spring-integration-redis/src/main/java/org/springframework/integration/redis/store/RedisMessageStore.java b/spring-integration-redis/src/main/java/org/springframework/integration/redis/store/RedisMessageStore.java index 9e373f2c91..95a3b82e4b 100644 --- a/spring-integration-redis/src/main/java/org/springframework/integration/redis/store/RedisMessageStore.java +++ b/spring-integration-redis/src/main/java/org/springframework/integration/redis/store/RedisMessageStore.java @@ -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; } diff --git a/spring-integration-redis/src/test/java/org/springframework/integration/redis/store/RedisMessageGroupStoreTests.java b/spring-integration-redis/src/test/java/org/springframework/integration/redis/store/RedisMessageGroupStoreTests.java index 863da108f4..99e2ee90c4 100644 --- a/spring-integration-redis/src/test/java/org/springframework/integration/redis/store/RedisMessageGroupStoreTests.java +++ b/spring-integration-redis/src/test/java/org/springframework/integration/redis/store/RedisMessageGroupStoreTests.java @@ -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("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("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("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("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("1")); store.removeMessageFromGroup(1, new GenericMessage("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("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("1"); store1.addMessageToGroup(1, message); MessageGroup messageGroup = store2.addMessageToGroup(1, new GenericMessage("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("1")); store2.addMessageToGroup(2, new GenericMessage("2")); store1.addMessageToGroup(3, new GenericMessage("3")); store2.addMessageToGroup(3, new GenericMessage("3A")); - + Iterator 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("1"); + final Message message = new GenericMessage("1"); ExecutorService executor = null; - + final List failures = new ArrayList(); - + 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)); } - + }