From 5263ea6dfff72b0e2dfdd2c3387b005fc3430a7a 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 --- .../store/AbstractKeyValueMessageStore.java | 83 ++++++++++++++----- .../gemfire/store/GemfireMessageStore.java | 67 +++++++-------- .../gemfire/store/GemfireGroupStoreTests.java | 19 ----- .../store/GemfireMessageStoreTests.java | 29 +++++-- .../redis/store/RedisMessageStore.java | 39 +++++++-- .../redis/store/RedisMessageStoreTests.java | 21 ++++- src/reference/asciidoc/gemfire.adoc | 27 ++---- src/reference/asciidoc/redis.adoc | 2 + 8 files changed, 176 insertions(+), 111 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 3230846d76..ceadd23047 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 @@ -49,12 +49,56 @@ public abstract class AbstractKeyValueMessageStore extends AbstractMessageGroupS @Deprecated 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 public Message getMessage(UUID messageId) { Assert.notNull(messageId, "'messageId' must not be null"); - Object object = doRetrieve(MESSAGE_KEY_PREFIX + messageId); + Object object = doRetrieve(this.messagePrefix + messageId); if (object != null) { return extractMessage(object); } @@ -80,7 +124,7 @@ public abstract class AbstractKeyValueMessageStore extends AbstractMessageGroupS @Override public MessageMetadata getMessageMetadata(UUID messageId) { Assert.notNull(messageId, "'messageId' must not be null"); - Object object = doRetrieve(MESSAGE_KEY_PREFIX + messageId); + Object object = doRetrieve(this.messagePrefix + messageId); if (object != null) { extractMessage(object); if (object instanceof MessageHolder) { @@ -100,13 +144,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(); - doStoreIfAbsent(MESSAGE_KEY_PREFIX + messageId, new MessageHolder(message)); + doStoreIfAbsent(this.messagePrefix + messageId, new MessageHolder(message)); } @Override public Message removeMessage(UUID id) { Assert.notNull(id, "'id' must not be null"); - Object object = doRemove(MESSAGE_KEY_PREFIX + id); + Object object = doRemove(this.messagePrefix + id); if (object != null) { return extractMessage(object); } @@ -118,7 +162,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; } @@ -147,7 +191,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; @@ -187,7 +231,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); } @Override @@ -195,17 +239,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); } } @@ -216,7 +260,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); } } @@ -226,7 +270,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; @@ -248,7 +292,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 @@ -259,7 +303,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); } } @@ -295,7 +339,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); } @@ -304,11 +348,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); } @@ -359,6 +403,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 c9265b613e..0c47b46400 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 @@ -20,12 +20,8 @@ import java.util.ArrayList; import java.util.Collection; import java.util.List; -import org.apache.geode.cache.Cache; import org.apache.geode.cache.Region; -import org.springframework.beans.factory.InitializingBean; -import org.springframework.data.gemfire.RegionAttributesFactoryBean; -import org.springframework.data.gemfire.RegionFactoryBean; import org.springframework.integration.store.AbstractKeyValueMessageStore; import org.springframework.integration.store.MessageGroupStore; import org.springframework.integration.store.MessageStore; @@ -39,17 +35,13 @@ import org.springframework.util.PatternMatchUtils; * @author Mark Fisher * @author Oleg Zhurakousky * @author David Turanski + * @author Artem Bilan + * * @since 2.1 */ -public class GemfireMessageStore extends AbstractKeyValueMessageStore implements InitializingBean { +public class GemfireMessageStore extends AbstractKeyValueMessageStore { - private static final String MESSAGE_STORE_REGION_NAME = "messageStoreRegion"; - - private volatile Region messageStoreRegion; - - private final Cache cache; - - private volatile boolean ignoreJta = true; + private final Region messageStoreRegion; /** * Provides the region to be used for the message store. This is useful when @@ -58,41 +50,38 @@ public class GemfireMessageStore extends AbstractKeyValueMessageStore implements * @param messageStoreRegion The region. */ public GemfireMessageStore(Region messageStoreRegion) { - this.cache = null; + 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.messageStoreRegion = messageStoreRegion; } + /** + * 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") + /** + * @deprecated in favor of constructor initialization. + */ + @Deprecated public void afterPropertiesSet() { - if (this.messageStoreRegion != null) { - return; - } - try { - if (logger.isDebugEnabled()) { - logger.debug("creating message store region as '" + MESSAGE_STORE_REGION_NAME + "'"); - } - - RegionAttributesFactoryBean attributesFactoryBean = new RegionAttributesFactoryBean(); - attributesFactoryBean.setIgnoreJTA(this.ignoreJta); - attributesFactoryBean.afterPropertiesSet(); - RegionFactoryBean messageRegionFactoryBean = new RegionFactoryBean() { - - }; - messageRegionFactoryBean.setBeanName(MESSAGE_STORE_REGION_NAME); - messageRegionFactoryBean.setAttributes(attributesFactoryBean.getObject()); - messageRegionFactoryBean.setCache(this.cache); - messageRegionFactoryBean.afterPropertiesSet(); - this.messageStoreRegion = messageRegionFactoryBean.getObject(); - } - catch (Exception e) { - throw new IllegalArgumentException("Failed to initialize Gemfire Region", e); - } } @Override 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 02c3dc9388..7d7a828c7a 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 @@ -69,7 +69,6 @@ public class GemfireGroupStoreTests { @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"); 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 2b6f38c8c3..28308c27c6 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 @@ -37,6 +37,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; @@ -58,8 +59,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-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 75c151d472..1517a5762f 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. @@ -35,13 +35,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()); @@ -93,13 +114,6 @@ public class RedisMessageStore extends AbstractKeyValueMessageStore { } } - 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); - } - @Override protected Object doRemove(Object id) { Assert.notNull(id, "'id' must not be null"); @@ -110,10 +124,17 @@ public class RedisMessageStore extends AbstractKeyValueMessageStore { return removedObject; } - @Override protected Collection doListKeys(String keyPattern) { Assert.hasText(keyPattern, "'keyPattern' must not be empty"); return this.redisTemplate.keys(keyPattern); } + + 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 ee05daa398..865da1887b 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