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

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<MessageGroup> iterator() {
 		final Iterator<?> idIterator = normalizeKeys(
-				(Collection<String>) doListKeys(MESSAGE_GROUP_KEY_PREFIX + "*"))
+				(Collection<String>) doListKeys(this.groupPrefix + "*"))
 				.iterator();
 		return new MessageGroupIterator(idIterator);
 	}
@@ -304,11 +348,11 @@ public abstract class AbstractKeyValueMessageStore extends AbstractMessageGroupS
 		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, "");
+			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();
 		}
+
 	}
 
 }

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<Object, Object> messageStoreRegion;
-
-	private final Cache cache;
-
-	private volatile boolean ignoreJta = true;
+	private final Region<Object, Object> 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<Object, Object> 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<Object, Object> 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<Object, Object> messageRegionFactoryBean = new RegionFactoryBean<Object, Object>() {
-
-			};
-			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

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<String>("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<String>("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<String>("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<String>("1"));
 		store.removeMessagesFromGroup(1, new GenericMessage<String>("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<String>("2"));
 	}
 
 	@Test
 	public void testCompleteMessageGroup() throws Exception {
 		GemfireMessageStore store = new GemfireMessageStore(region);
-		store.afterPropertiesSet();
 		MessageGroup messageGroup = store.getMessageGroup(1);
 		Message<?> messageToMark = new GenericMessage<String>("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<String>("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<String>("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<String>("1"));
 		store2.addMessagesToGroup(2, new GenericMessage<String>("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<String>("1");
 

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<String>("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<Message<?>> messages = new ArrayList<Message<?>>();
@@ -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<Message<?>> messages = new ArrayList<Message<?>>();
+		for (int i = 0; i < 25; i++) {
+			Message<String> 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) {

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<Object, Object> 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<Object, Object>();
 		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);
+	}
+
 }

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<String> stringMessage = new GenericMessage<String>("Hello Redis");
+		store.addMessage(stringMessage);
+		Message<String> retrievedMessage = (Message<String>) store.getMessage(stringMessage.getHeaders().getId());
+		assertNotNull(retrievedMessage);
+		assertEquals("Hello Redis", retrievedMessage.getPayload());
+
+		StringRedisTemplate template = createStringRedisTemplate(getConnectionFactoryForTest());
+		BoundValueOperations<String, String> ops =
+				template.boundValueOps("foo" + "MESSAGE_" + stringMessage.getHeaders().getId());
+		assertNotNull(ops.get());
+	}
+
 	@SuppressWarnings("unchecked")
 	@Test
 	@RedisAvailable

src/reference/asciidoc/gemfire.adoc

@@ -135,10 +135,13 @@ Spring Integration's Gemfire module provides the `GemfireMessageStore` which is
 [source,xml]
 ----
 <bean id="gemfireMessageStore" class="o.s.i.gemfire.store.GemfireMessageStore">
-    <constructor-arg ref="myCache"/>
+    <constructor-arg ref="myRegion"/>
 </bean>
 
-<bean id="myCache" class="org.springframework.data.gemfire.CacheFactoryBean"/>
+<gfe:cache/>
+
+<gfe:replicated-region id="myRegion"/>
+
 
 <int:channel id="somePersistentQueueChannel">
     <int:queue message-store="gemfireMessageStore"/>
@@ -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]
-----
-<bean id="gemfireMessageStore" class="o.s.i.gemfire.store.GemfireMessageStore">
-    <constructor-arg ref="myRegion"/>
-</bean>
-
-<gfe:cache/>
-
-<gfe:replicated-region id="myRegion"/>
-----
-
-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
 

src/reference/asciidoc/redis.adoc

@@ -354,6 +354,8 @@ RedisSerializer<Object> 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