From e097198d5e2ba169828e6542d541bdeb001ea5bf Mon Sep 17 00:00:00 2001 From: Artem Bilan Date: Tue, 5 Sep 2017 18:23:03 -0400 Subject: [PATCH] INT-4123: Add Prefix to the Key-Value MSs Fixes spring-projects/spring-integration#2213 JIRA: https://jira.spring.io/browse/INT-4123 Fully different `MessageStore`s can be configured for the same shared Key-Value data-base. Since the retrieval logic is based on the keys, that may cause the unexpected messages expiration via `MessageGroupStoreReaper`. * To distinguish store instances on the shared store add `prefix` option to the `AbstractKeyValueMessageStore` * Deprecate the `GemfireMessageStore` `Cache`-based configuration - `setIgnoreJta()` and `afterPropertiesSet()`. The `GemfireMessageStore` relies only on an externally configured `Region`. **Cherry-pick to 4.3.x** Doc Polishing # Conflicts: # spring-integration-core/src/main/java/org/springframework/integration/store/AbstractKeyValueMessageStore.java # spring-integration-gemfire/src/main/java/org/springframework/integration/gemfire/store/GemfireMessageStore.java # spring-integration-redis/src/main/java/org/springframework/integration/redis/store/RedisMessageStore.java * Fix tests `GemfireMessageStore` tests conflicts --- .../store/AbstractKeyValueMessageStore.java | 83 ++++++++++++++----- .../gemfire/store/GemfireMessageStore.java | 27 +++++- ...dlerRescheduleIntegrationTests-context.xml | 2 +- ...ayerHandlerRescheduleIntegrationTests.java | 16 +++- .../gemfire/store/GemfireGroupStoreTests.java | 23 +---- .../store/GemfireMessageStoreTests.java | 29 +++++-- .../store/gemfire-aggregator-config-a.xml | 3 +- .../store/gemfire-aggregator-config.xml | 3 +- .../gemfire/store/gemfire-queue-config.xml | 3 +- .../redis/store/RedisMessageStore.java | 32 ++++++- .../redis/store/RedisMessageStoreTests.java | 21 ++++- src/reference/asciidoc/gemfire.adoc | 27 ++---- src/reference/asciidoc/redis.adoc | 2 + 13 files changed, 191 insertions(+), 80 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 a691c06af9..96b3c71bee 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 @@ -47,6 +47,50 @@ public abstract class AbstractKeyValueMessageStore extends AbstractMessageGroupS protected static final String CREATED_DATE = "CREATED_DATE"; + private final String messagePrefix; + + private final String groupPrefix; + + protected AbstractKeyValueMessageStore() { + this(""); + } + + /** + * Construct an instance based on the provided prefix for keys to distinguish between + * different store instances in the same target key-value data base. Defaults to an + * empty string - no prefix. The actual prefix for messages is + * {@code prefix + MESSAGE_}; for message groups - {@code prefix + MESSAGE_GROUP_} + * @param prefix the prefix to use + * @since 4.3.12 + */ + protected AbstractKeyValueMessageStore(String prefix) { + Assert.notNull(prefix, "'prefix' must not be null"); + this.messagePrefix = prefix + MESSAGE_KEY_PREFIX; + this.groupPrefix = prefix + MESSAGE_GROUP_KEY_PREFIX; + } + + /** + * Return the configured prefix for message keys to distinguish between different + * store instances in the same target key-value data base. Defaults to the + * {@value MESSAGE_KEY_PREFIX} - without a custom prefix. + * @return the prefix for keys + * @since 4.3.12 + */ + protected String getMessagePrefix() { + return this.messagePrefix; + } + + /** + * Return the configured prefix for message group keys to distinguish between + * different store instances in the same target key-value data base. Defaults to the + * {@value MESSAGE_GROUP_KEY_PREFIX} - without custom prefix. + * @return the prefix for keys + * @since 4.3.12 + */ + public String getGroupPrefix() { + return this.groupPrefix; + } + // MessageStore methods @Override @@ -68,13 +112,13 @@ public abstract class AbstractKeyValueMessageStore extends AbstractMessageGroupS protected void doAddMessage(Message message) { Assert.notNull(message, "'message' must not be null"); UUID messageId = message.getHeaders().getId(); - doStore(MESSAGE_KEY_PREFIX + messageId, message); + doStore(this.messagePrefix + messageId, message); } @Override public Message removeMessage(UUID id) { Assert.notNull(id, "'id' must not be null"); - Object message = doRemove(MESSAGE_KEY_PREFIX + id); + Object message = doRemove(this.messagePrefix + id); if (message != null) { Assert.isInstanceOf(Message.class, message); } @@ -87,7 +131,7 @@ public abstract class AbstractKeyValueMessageStore extends AbstractMessageGroupS @Override @ManagedAttribute public long getMessageCount() { - Collection messageIds = doListKeys(MESSAGE_KEY_PREFIX + "*"); + Collection messageIds = doListKeys(this.messagePrefix + "*"); return (messageIds != null) ? messageIds.size() : 0; } @@ -116,7 +160,7 @@ public abstract class AbstractKeyValueMessageStore extends AbstractMessageGroupS @Override public MessageGroupMetadata getGroupMetadata(Object groupId) { Assert.notNull(groupId, "'groupId' must not be null"); - Object mgm = this.doRetrieve(MESSAGE_GROUP_KEY_PREFIX + groupId); + Object mgm = this.doRetrieve(this.groupPrefix + groupId); if (mgm != null) { Assert.isInstanceOf(MessageGroupMetadata.class, mgm); return (MessageGroupMetadata) mgm; @@ -158,7 +202,7 @@ public abstract class AbstractKeyValueMessageStore extends AbstractMessageGroupS // store MessageGroupMetadata built from enriched MG - doStore(MESSAGE_GROUP_KEY_PREFIX + groupId, metadata); + doStore(this.groupPrefix + groupId, metadata); } /** @@ -177,7 +221,7 @@ public abstract class AbstractKeyValueMessageStore extends AbstractMessageGroupS if (metadata != null) { metadata.remove(id); metadata.setLastModified(System.currentTimeMillis()); - doStore(MESSAGE_GROUP_KEY_PREFIX + groupId, metadata); + doStore(this.groupPrefix + groupId, metadata); } return getMessageGroup(groupId); @@ -189,17 +233,17 @@ public abstract class AbstractKeyValueMessageStore extends AbstractMessageGroupS Assert.notNull(groupId, "'groupId' must not be null"); Assert.notNull(messages, "'messages' must not be null"); - Object mgm = doRetrieve(MESSAGE_GROUP_KEY_PREFIX + groupId); + Object mgm = doRetrieve(this.groupPrefix + groupId); if (mgm != null) { Assert.isInstanceOf(MessageGroupMetadata.class, mgm); MessageGroupMetadata messageGroupMetadata = (MessageGroupMetadata) mgm; for (Message messageToRemove : messages) { UUID messageId = messageToRemove.getHeaders().getId(); messageGroupMetadata.remove(messageId); - doRemove(MESSAGE_KEY_PREFIX + messageId); + doRemove(this.messagePrefix + messageId); } messageGroupMetadata.setLastModified(System.currentTimeMillis()); - doStore(MESSAGE_GROUP_KEY_PREFIX + groupId, messageGroupMetadata); + doStore(this.groupPrefix + groupId, messageGroupMetadata); } } @@ -210,7 +254,7 @@ public abstract class AbstractKeyValueMessageStore extends AbstractMessageGroupS if (metadata != null) { metadata.complete(); metadata.setLastModified(System.currentTimeMillis()); - doStore(MESSAGE_GROUP_KEY_PREFIX + groupId, metadata); + doStore(this.groupPrefix + groupId, metadata); } } @@ -220,7 +264,7 @@ public abstract class AbstractKeyValueMessageStore extends AbstractMessageGroupS @Override public void removeMessageGroup(Object groupId) { Assert.notNull(groupId, "'groupId' must not be null"); - Object mgm = doRemove(MESSAGE_GROUP_KEY_PREFIX + groupId); + Object mgm = doRemove(this.groupPrefix + groupId); if (mgm != null) { Assert.isInstanceOf(MessageGroupMetadata.class, mgm); MessageGroupMetadata messageGroupMetadata = (MessageGroupMetadata) mgm; @@ -242,7 +286,7 @@ public abstract class AbstractKeyValueMessageStore extends AbstractMessageGroupS } metadata.setLastReleasedMessageSequenceNumber(sequenceNumber); metadata.setLastModified(System.currentTimeMillis()); - doStore(MESSAGE_GROUP_KEY_PREFIX + groupId, metadata); + doStore(this.groupPrefix + groupId, metadata); } @Override @@ -253,7 +297,7 @@ public abstract class AbstractKeyValueMessageStore extends AbstractMessageGroupS if (firstId != null) { groupMetadata.remove(firstId); groupMetadata.setLastModified(System.currentTimeMillis()); - doStore(MESSAGE_GROUP_KEY_PREFIX + groupId, groupMetadata); + doStore(this.groupPrefix + groupId, groupMetadata); return removeMessage(firstId); } } @@ -289,7 +333,7 @@ public abstract class AbstractKeyValueMessageStore extends AbstractMessageGroupS @SuppressWarnings("unchecked") public Iterator iterator() { final Iterator idIterator = normalizeKeys( - (Collection) doListKeys(MESSAGE_GROUP_KEY_PREFIX + "*")) + (Collection) doListKeys(this.groupPrefix + "*")) .iterator(); return new MessageGroupIterator(idIterator); } @@ -298,11 +342,11 @@ public abstract class AbstractKeyValueMessageStore extends AbstractMessageGroupS 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, ""); + if (strKey.startsWith(this.groupPrefix)) { + strKey = strKey.replace(this.groupPrefix, ""); } - else if (strKey.startsWith(MESSAGE_KEY_PREFIX)) { - strKey = strKey.replace(MESSAGE_KEY_PREFIX, ""); + else if (strKey.startsWith(this.messagePrefix)) { + strKey = strKey.replace(this.messagePrefix, ""); } normalizedKeys.add(strKey); } @@ -355,7 +399,7 @@ public abstract class AbstractKeyValueMessageStore extends AbstractMessageGroupS private Message getRawMessage(UUID id) { Assert.notNull(id, "'id' must not be null"); - Object message = doRetrieve(MESSAGE_KEY_PREFIX + id); + Object message = doRetrieve(this.messagePrefix + id); return (Message) message; } @@ -382,6 +426,7 @@ public abstract class AbstractKeyValueMessageStore extends AbstractMessageGroupS public void remove() { throw new UnsupportedOperationException(); } + } } diff --git a/spring-integration-gemfire/src/main/java/org/springframework/integration/gemfire/store/GemfireMessageStore.java b/spring-integration-gemfire/src/main/java/org/springframework/integration/gemfire/store/GemfireMessageStore.java index a305a0e496..fa020b358f 100644 --- a/spring-integration-gemfire/src/main/java/org/springframework/integration/gemfire/store/GemfireMessageStore.java +++ b/spring-integration-gemfire/src/main/java/org/springframework/integration/gemfire/store/GemfireMessageStore.java @@ -39,16 +39,18 @@ import com.gemstone.gemfire.cache.Region; * @author Mark Fisher * @author Oleg Zhurakousky * @author David Turanski + * @author Artem Bilan + * * @since 2.1 */ public class GemfireMessageStore extends AbstractKeyValueMessageStore implements InitializingBean { private static final String MESSAGE_STORE_REGION_NAME = "messageStoreRegion"; - private volatile Region messageStoreRegion; - private final Cache cache; + private volatile Region messageStoreRegion; + private volatile boolean ignoreJta = true; /** @@ -58,6 +60,19 @@ public class GemfireMessageStore extends AbstractKeyValueMessageStore implements * @param messageStoreRegion The region. */ public GemfireMessageStore(Region messageStoreRegion) { + this(messageStoreRegion, ""); + } + + /** + * Construct a {@link GemfireMessageStore} instance based on the provided + * @param messageStoreRegion the region to use. + * @param prefix the key prefix to use, allowing the same region to be used for + * multiple stores. + * @since 4.3.12 + */ + public GemfireMessageStore(Region messageStoreRegion, String prefix) { + super(prefix); + Assert.notNull(messageStoreRegion, "'messageStoreRegion' must not be null"); this.cache = null; this.messageStoreRegion = messageStoreRegion; } @@ -66,7 +81,6 @@ public class GemfireMessageStore extends AbstractKeyValueMessageStore implements * Provides a cache reference used to create a message store region named * 'messageStoreRegion' * @param cache The cache. - * * @deprecated - use the other constructor and provide a region directly. */ @Deprecated @@ -75,12 +89,19 @@ public class GemfireMessageStore extends AbstractKeyValueMessageStore implements this.cache = cache; } + /** + * The boolean flag to ignore JTA on the Gemfire Region. + * @param ignoreJta boolean flag to ignore JTA on the Gemfire Region. + * @deprecated with no-op, in favor of externally configured region. + */ + @Deprecated public void setIgnoreJta(boolean ignoreJta) { this.ignoreJta = ignoreJta; } @Override @SuppressWarnings({ "unchecked", "deprecation" }) + @Deprecated public void afterPropertiesSet() { if (this.messageStoreRegion != null) { return; diff --git a/spring-integration-gemfire/src/test/java/org/springframework/integration/gemfire/store/DelayerHandlerRescheduleIntegrationTests-context.xml b/spring-integration-gemfire/src/test/java/org/springframework/integration/gemfire/store/DelayerHandlerRescheduleIntegrationTests-context.xml index fa4f0dd2a3..840194ba32 100644 --- a/spring-integration-gemfire/src/test/java/org/springframework/integration/gemfire/store/DelayerHandlerRescheduleIntegrationTests-context.xml +++ b/spring-integration-gemfire/src/test/java/org/springframework/integration/gemfire/store/DelayerHandlerRescheduleIntegrationTests-context.xml @@ -7,7 +7,7 @@ + value="#{T (org.springframework.integration.gemfire.store.DelayerHandlerRescheduleIntegrationTests).region}"/> diff --git a/spring-integration-gemfire/src/test/java/org/springframework/integration/gemfire/store/DelayerHandlerRescheduleIntegrationTests.java b/spring-integration-gemfire/src/test/java/org/springframework/integration/gemfire/store/DelayerHandlerRescheduleIntegrationTests.java index 12cf1585e5..5b65d2f598 100644 --- a/spring-integration-gemfire/src/test/java/org/springframework/integration/gemfire/store/DelayerHandlerRescheduleIntegrationTests.java +++ b/spring-integration-gemfire/src/test/java/org/springframework/integration/gemfire/store/DelayerHandlerRescheduleIntegrationTests.java @@ -44,6 +44,9 @@ import org.springframework.messaging.MessageChannel; import org.springframework.messaging.PollableChannel; import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler; +import com.gemstone.gemfire.cache.Cache; +import com.gemstone.gemfire.cache.Region; +import com.gemstone.gemfire.cache.Scope; /** * @author Artem Bilan @@ -55,6 +58,8 @@ public class DelayerHandlerRescheduleIntegrationTests { public static final String DELAYER_ID = "delayerWithGemfireMS"; + public static Region region; + public static CacheFactoryBean cacheFactoryBean; @ClassRule @@ -67,10 +72,15 @@ public class DelayerHandlerRescheduleIntegrationTests { gemfireProperties.setProperty("mcast-port", "0"); cacheFactoryBean.setProperties(gemfireProperties); cacheFactoryBean.afterPropertiesSet(); + Cache cache = cacheFactoryBean.getObject(); + region = cache.createRegionFactory().setScope(Scope.LOCAL).create("sig-tests"); } @AfterClass public static void cleanUp() throws Exception { + if (region != null) { + region.close(); + } if (cacheFactoryBean != null) { cacheFactoryBean.destroy(); } @@ -95,7 +105,7 @@ public class DelayerHandlerRescheduleIntegrationTests { (ThreadPoolTaskScheduler) IntegrationContextUtils.getTaskScheduler(context); taskScheduler.shutdown(); taskScheduler.getScheduledExecutor().awaitTermination(10, TimeUnit.SECONDS); - context.destroy(); + context.close(); try { context.getBean("input", MessageChannel.class); @@ -135,11 +145,11 @@ public class DelayerHandlerRescheduleIntegrationTests { assertEquals(1, messageStore.getMessageGroupCount()); int n = 0; while (n++ < 200 && messageStore.messageGroupSize(delayerMessageGroupId) > 0) { - Thread.sleep(50); + Thread.sleep(100); } assertEquals(0, messageStore.messageGroupSize(delayerMessageGroupId)); - context.destroy(); + context.close(); } } diff --git a/spring-integration-gemfire/src/test/java/org/springframework/integration/gemfire/store/GemfireGroupStoreTests.java b/spring-integration-gemfire/src/test/java/org/springframework/integration/gemfire/store/GemfireGroupStoreTests.java index 10c68cbd41..557cbfcde0 100644 --- a/spring-integration-gemfire/src/test/java/org/springframework/integration/gemfire/store/GemfireGroupStoreTests.java +++ b/spring-integration-gemfire/src/test/java/org/springframework/integration/gemfire/store/GemfireGroupStoreTests.java @@ -64,12 +64,11 @@ public class GemfireGroupStoreTests { public static CacheFactoryBean cacheFactoryBean; - private static Region region; + public static Region region; @Test public void testNonExistingEmptyMessageGroup() throws Exception { GemfireMessageStore store = new GemfireMessageStore(region); - store.afterPropertiesSet(); MessageGroup messageGroup = store.getMessageGroup(1); assertNotNull(messageGroup); assertTrue(messageGroup instanceof SimpleMessageGroup); @@ -79,7 +78,6 @@ public class GemfireGroupStoreTests { @Test public void testMessageGroupWithAddedMessage() throws Exception { GemfireMessageStore store = new GemfireMessageStore(region); - store.afterPropertiesSet(); MessageGroup messageGroup = store.getMessageGroup(1); Message message = new GenericMessage("Hello"); messageGroup = store.addMessageToGroup(1, message); @@ -87,7 +85,6 @@ public class GemfireGroupStoreTests { // make sure the store is properly rebuild from Gemfire store = new GemfireMessageStore(region); - store.afterPropertiesSet(); messageGroup = store.getMessageGroup(1); assertEquals(1, messageGroup.size()); @@ -96,7 +93,6 @@ public class GemfireGroupStoreTests { @Test public void testRemoveMessageFromTheGroup() throws Exception { GemfireMessageStore store = new GemfireMessageStore(region); - store.afterPropertiesSet(); MessageGroup messageGroup = store.getMessageGroup(1); Message message = new GenericMessage("2"); @@ -121,7 +117,6 @@ public class GemfireGroupStoreTests { // make sure the store is properly rebuild from Gemfire store = new GemfireMessageStore(region); - store.afterPropertiesSet(); messageGroup = store.getMessageGroup(1); assertEquals(2, messageGroup.size()); @@ -131,7 +126,6 @@ public class GemfireGroupStoreTests { @Test public void testRemoveMessageGroup() throws Exception { GemfireMessageStore store = new GemfireMessageStore(region); - store.afterPropertiesSet(); MessageGroup messageGroup = store.getMessageGroup(1); Message message = new GenericMessage("Hello"); messageGroup = store.addMessageToGroup(messageGroup.getGroupId(), message); @@ -145,7 +139,6 @@ public class GemfireGroupStoreTests { // make sure the store is properly rebuild from Gemfire store = new GemfireMessageStore(region); - store.afterPropertiesSet(); messageGroup = store.getMessageGroup(1); @@ -156,7 +149,6 @@ public class GemfireGroupStoreTests { @Test public void testRemoveNonExistingMessageFromTheGroup() throws Exception { GemfireMessageStore store = new GemfireMessageStore(region); - store.afterPropertiesSet(); MessageGroup messageGroup = store.getMessageGroup(1); store.addMessagesToGroup(messageGroup.getGroupId(), new GenericMessage("1")); store.removeMessagesFromGroup(1, new GenericMessage("2")); @@ -165,14 +157,12 @@ public class GemfireGroupStoreTests { @Test public void testRemoveNonExistingMessageFromNonExistingTheGroup() throws Exception { GemfireMessageStore store = new GemfireMessageStore(region); - store.afterPropertiesSet(); store.removeMessagesFromGroup(1, new GenericMessage("2")); } @Test public void testCompleteMessageGroup() throws Exception { GemfireMessageStore store = new GemfireMessageStore(region); - store.afterPropertiesSet(); MessageGroup messageGroup = store.getMessageGroup(1); Message messageToMark = new GenericMessage("1"); store.addMessagesToGroup(messageGroup.getGroupId(), messageToMark); @@ -184,7 +174,6 @@ public class GemfireGroupStoreTests { @Test public void testLastReleasedSequenceNumber() throws Exception { GemfireMessageStore store = new GemfireMessageStore(region); - store.afterPropertiesSet(); MessageGroup messageGroup = store.getMessageGroup(1); Message messageToMark = new GenericMessage("1"); store.addMessagesToGroup(messageGroup.getGroupId(), messageToMark); @@ -196,10 +185,8 @@ public class GemfireGroupStoreTests { @Test public void testMultipleInstancesOfGroupStore() throws Exception { GemfireMessageStore store1 = new GemfireMessageStore(region); - store1.afterPropertiesSet(); GemfireMessageStore store2 = new GemfireMessageStore(region); - store2.afterPropertiesSet(); Message message = new GenericMessage("1"); store1.addMessagesToGroup(1, message); @@ -208,7 +195,6 @@ public class GemfireGroupStoreTests { assertEquals(2, messageGroup.getMessages().size()); GemfireMessageStore store3 = new GemfireMessageStore(region); - store3.afterPropertiesSet(); store3.removeMessagesFromGroup(1, message); messageGroup = store3.getMessageGroup(1); @@ -219,7 +205,6 @@ public class GemfireGroupStoreTests { @Test public void testWithMessageHistory() throws Exception { GemfireMessageStore store = new GemfireMessageStore(region); - store.afterPropertiesSet(); store.getMessageGroup(1); @@ -246,9 +231,7 @@ public class GemfireGroupStoreTests { @Test public void testIteratorOfMessageGroups() throws Exception { GemfireMessageStore store1 = new GemfireMessageStore(region); - store1.afterPropertiesSet(); GemfireMessageStore store2 = new GemfireMessageStore(region); - store2.afterPropertiesSet(); store1.addMessagesToGroup(1, new GenericMessage("1")); store2.addMessagesToGroup(2, new GenericMessage("2")); @@ -278,9 +261,7 @@ public class GemfireGroupStoreTests { public void testConcurrentModifications() throws Exception { final GemfireMessageStore store1 = new GemfireMessageStore(region); - store1.afterPropertiesSet(); final GemfireMessageStore store2 = new GemfireMessageStore(region); - store2.afterPropertiesSet(); final Message message = new GenericMessage("1"); @@ -345,7 +326,7 @@ public class GemfireGroupStoreTests { Message m3 = MessageBuilder.withPayload("3").setSequenceNumber(3).setSequenceSize(3).setCorrelationId(1) .build(); inputA.send(m3); - assertNotNull(outputA.receive(1000)); + assertNotNull(outputA.receive(10000)); context1.close(); context2.close(); } diff --git a/spring-integration-gemfire/src/test/java/org/springframework/integration/gemfire/store/GemfireMessageStoreTests.java b/spring-integration-gemfire/src/test/java/org/springframework/integration/gemfire/store/GemfireMessageStoreTests.java index bb187984dd..b657363351 100644 --- a/spring-integration-gemfire/src/test/java/org/springframework/integration/gemfire/store/GemfireMessageStoreTests.java +++ b/spring-integration-gemfire/src/test/java/org/springframework/integration/gemfire/store/GemfireMessageStoreTests.java @@ -34,6 +34,7 @@ import org.springframework.data.gemfire.RegionFactoryBean; import org.springframework.integration.channel.DirectChannel; import org.springframework.integration.history.MessageHistory; import org.springframework.integration.store.MessageGroup; +import org.springframework.integration.store.MessageGroupMetadata; import org.springframework.integration.support.MessageBuilder; import org.springframework.integration.test.util.TestUtils; import org.springframework.messaging.Message; @@ -60,8 +61,6 @@ public class GemfireMessageStoreTests { @Test public void addAndGetMessage() throws Exception { GemfireMessageStore store = new GemfireMessageStore(region); - store.afterPropertiesSet(); - Message message = MessageBuilder.withPayload("test").build(); store.addMessage(message); Message retrieved = store.getMessage(message.getHeaders().getId()); @@ -76,7 +75,6 @@ public class GemfireMessageStoreTests { region.afterPropertiesSet(); GemfireMessageStore store = new GemfireMessageStore(region.getObject()); - store.afterPropertiesSet(); assertSame(region.getObject(), TestUtils.getPropertyValue(store, "messageStoreRegion")); region.destroy(); @@ -85,7 +83,6 @@ public class GemfireMessageStoreTests { @Test public void testWithMessageHistory() throws Exception { GemfireMessageStore store = new GemfireMessageStore(region); - store.afterPropertiesSet(); Message message = new GenericMessage("Hello"); DirectChannel fooChannel = new DirectChannel(); @@ -108,7 +105,6 @@ public class GemfireMessageStoreTests { @Test public void testAddAndRemoveMessagesFromMessageGroup() throws Exception { GemfireMessageStore messageStore = new GemfireMessageStore(region); - messageStore.afterPropertiesSet(); String groupId = "X"; List> messages = new ArrayList>(); @@ -124,6 +120,29 @@ public class GemfireMessageStoreTests { assertEquals(0, group.size()); } + @Test + public void testAddAndRemoveMessagesFromMessageGroupWithPrefix() throws Exception { + GemfireMessageStore messageStore = new GemfireMessageStore(region, "foo_"); + + String groupId = "X"; + List> messages = new ArrayList>(); + for (int i = 0; i < 25; i++) { + Message message = MessageBuilder.withPayload("foo").setCorrelationId(groupId).build(); + messageStore.addMessagesToGroup(groupId, message); + messages.add(message); + } + + MessageGroupMetadata messageGroupMetadata = + (MessageGroupMetadata) region.get("foo_" + "MESSAGE_GROUP_" + groupId); + + assertNotNull(messageGroupMetadata); + assertEquals(25, messageGroupMetadata.size()); + + messageStore.removeMessagesFromGroup(groupId, messages); + MessageGroup group = messageStore.getMessageGroup(groupId); + assertEquals(0, group.size()); + } + @Before public void prepare() { if (region != null) { diff --git a/spring-integration-gemfire/src/test/java/org/springframework/integration/gemfire/store/gemfire-aggregator-config-a.xml b/spring-integration-gemfire/src/test/java/org/springframework/integration/gemfire/store/gemfire-aggregator-config-a.xml index 500f4f5315..6e6ac71d14 100644 --- a/spring-integration-gemfire/src/test/java/org/springframework/integration/gemfire/store/gemfire-aggregator-config-a.xml +++ b/spring-integration-gemfire/src/test/java/org/springframework/integration/gemfire/store/gemfire-aggregator-config-a.xml @@ -12,8 +12,7 @@ - + diff --git a/spring-integration-gemfire/src/test/java/org/springframework/integration/gemfire/store/gemfire-aggregator-config.xml b/spring-integration-gemfire/src/test/java/org/springframework/integration/gemfire/store/gemfire-aggregator-config.xml index 500f4f5315..6e6ac71d14 100644 --- a/spring-integration-gemfire/src/test/java/org/springframework/integration/gemfire/store/gemfire-aggregator-config.xml +++ b/spring-integration-gemfire/src/test/java/org/springframework/integration/gemfire/store/gemfire-aggregator-config.xml @@ -12,8 +12,7 @@ - + diff --git a/spring-integration-gemfire/src/test/java/org/springframework/integration/gemfire/store/gemfire-queue-config.xml b/spring-integration-gemfire/src/test/java/org/springframework/integration/gemfire/store/gemfire-queue-config.xml index 632100b8f4..d01a30fada 100644 --- a/spring-integration-gemfire/src/test/java/org/springframework/integration/gemfire/store/gemfire-queue-config.xml +++ b/spring-integration-gemfire/src/test/java/org/springframework/integration/gemfire/store/gemfire-queue-config.xml @@ -19,8 +19,7 @@ - + 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 94e7a203d1..20c79248fb 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-2016 the original author or authors. + * Copyright 2007-2017 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. @@ -36,13 +36,34 @@ import org.springframework.util.Assert; * * @author Oleg Zhurakousky * @author Gary Russell + * @author Artem Bilan + * * @since 2.1 */ public class RedisMessageStore extends AbstractKeyValueMessageStore { private final RedisTemplate redisTemplate; + /** + * Construct {@link RedisMessageStore} based on the provided + * {@link RedisConnectionFactory} and default empty prefix. + * @param connectionFactory the RedisConnectionFactory to use + */ public RedisMessageStore(RedisConnectionFactory connectionFactory) { + this(connectionFactory, ""); + } + + /** + * Construct {@link RedisMessageStore} based on the provided + * {@link RedisConnectionFactory} and prefix. + * @param connectionFactory the RedisConnectionFactory to use + * @param prefix the key prefix to use, allowing the same broker to be used for + * multiple stores. + * @since 4.3.12 + * @see AbstractKeyValueMessageStore#AbstractKeyValueMessageStore(String) + */ + public RedisMessageStore(RedisConnectionFactory connectionFactory, String prefix) { + super(prefix); this.redisTemplate = new RedisTemplate(); this.redisTemplate.setConnectionFactory(connectionFactory); this.redisTemplate.setKeySerializer(new StringRedisSerializer()); @@ -89,11 +110,18 @@ public class RedisMessageStore extends AbstractKeyValueMessageStore { return removedObject; } - @Override protected Collection doListKeys(String keyPattern) { Assert.hasText(keyPattern, "'keyPattern' must not be empty"); Set keys = this.redisTemplate.keys(keyPattern); return keys; } + + private void rethrowAsIllegalArgumentException(SerializationException e) { + throw new IllegalArgumentException("If relying on the default RedisSerializer " + + "(JdkSerializationRedisSerializer) the Object must be Serializable. " + + "Either make it Serializable or provide your own implementation of " + + "RedisSerializer via 'setValueSerializer(..)'", e); + } + } diff --git a/spring-integration-redis/src/test/java/org/springframework/integration/redis/store/RedisMessageStoreTests.java b/spring-integration-redis/src/test/java/org/springframework/integration/redis/store/RedisMessageStoreTests.java index 5d501cace2..ca179597b7 100644 --- a/spring-integration-redis/src/test/java/org/springframework/integration/redis/store/RedisMessageStoreTests.java +++ b/spring-integration-redis/src/test/java/org/springframework/integration/redis/store/RedisMessageStoreTests.java @@ -32,6 +32,7 @@ import org.junit.Before; import org.junit.Test; import org.springframework.data.redis.connection.RedisConnectionFactory; +import org.springframework.data.redis.core.BoundValueOperations; import org.springframework.data.redis.core.StringRedisTemplate; import org.springframework.integration.channel.DirectChannel; import org.springframework.integration.history.MessageHistory; @@ -53,7 +54,7 @@ public class RedisMessageStoreTests extends RedisAvailableTests { @After public void setUpTearDown() { StringRedisTemplate template = this.createStringRedisTemplate(this.getConnectionFactoryForTest()); - template.delete(template.keys("MESSAGE_*")); + template.delete(template.keys("*MESSAGE_*")); } @Test @@ -122,6 +123,24 @@ public class RedisMessageStoreTests extends RedisAvailableTests { assertEquals("Hello Redis", retrievedMessage.getPayload()); } + @SuppressWarnings("unchecked") + @Test + @RedisAvailable + public void testAddAndGetWithPrefix() { + RedisConnectionFactory jcf = this.getConnectionFactoryForTest(); + RedisMessageStore store = new RedisMessageStore(jcf, "foo"); + Message stringMessage = new GenericMessage("Hello Redis"); + store.addMessage(stringMessage); + Message retrievedMessage = (Message) store.getMessage(stringMessage.getHeaders().getId()); + assertNotNull(retrievedMessage); + assertEquals("Hello Redis", retrievedMessage.getPayload()); + + StringRedisTemplate template = createStringRedisTemplate(getConnectionFactoryForTest()); + BoundValueOperations ops = + template.boundValueOps("foo" + "MESSAGE_" + stringMessage.getHeaders().getId()); + assertNotNull(ops.get()); + } + @SuppressWarnings("unchecked") @Test @RedisAvailable diff --git a/src/reference/asciidoc/gemfire.adoc b/src/reference/asciidoc/gemfire.adoc index a114370777..313a8eb84a 100644 --- a/src/reference/asciidoc/gemfire.adoc +++ b/src/reference/asciidoc/gemfire.adoc @@ -135,10 +135,13 @@ Spring Integration's Gemfire module provides the `GemfireMessageStore` which is [source,xml] ---- - + - + + + + @@ -148,23 +151,7 @@ Spring Integration's Gemfire module provides the `GemfireMessageStore` which is message-store="gemfireMessageStore"/> ---- -Above is a sample `GemfireMessageStore` configuration that shows its usage by a _QueueChannel_ and an _Aggregator_. -As you can see it is a normal Spring bean configuration. -The simplest configuration requires a reference to a `GemFireCache` (created by `CacheFactoryBean`) as a constructor argument. -If the cache is standalone, i.e., embedded in the same JVM, the MessageStore will create a message store region named "messageStoreRegion". -If your application requires customization of the messageStore region, for example, multiple Gemfire message stores each with its own region, you can configure a region for each message store instance and use the `Region` as the constructor argument: -[source,xml] ----- - - - - - - - ----- - -In the above examle, the cache and region are configured using the spring-gemfire namespace (not to be confused with the spring-integration-gemfire namespace). +In the above example, the cache and region are configured using the spring-gemfire namespace (not to be confused with the spring-integration-gemfire namespace). Often it is desirable for the message store to be maintained in one or more remote cache servers in a client-server configuration (See the http://www.vmware.com/support/pubs/vfabric-gemfire.html[GemFire product documentation] for more details). In this case, you configure a client cache, client region, and client pool and inject the region into the MessageStore. Here is an example: @@ -188,6 +175,8 @@ Note the _pool_ element is configured with the address of a cache server (a loca The region is configured as a 'PROXY' so that no data will be stored locally. The region's id corresponds to a region with the same name configured in the cache server. +Starting with version _4.3.12_, the `GemfireMessageStore` supports the key `prefix` option to allow distinguishing between instances of the store on the same Gemfire region. + [[gemfire-lock-registry]] === Gemfire Lock Registry diff --git a/src/reference/asciidoc/redis.adoc b/src/reference/asciidoc/redis.adoc index 4e2b6cd59d..bf19bcbfc2 100644 --- a/src/reference/asciidoc/redis.adoc +++ b/src/reference/asciidoc/redis.adoc @@ -354,6 +354,8 @@ RedisSerializer serializer = new GenericJackson2JsonRedisSerializer(mapp store.setValueSerializer(serializer); ---- +Starting with version _4.3.12_, the `RedisMessageStore` supports the key `prefix` option to allow distinguishing between instances of the store on the same Redis server. + [[redis-cms]] ==== Redis Channel Message Stores