INT-3325 Add Redis Channel Message Group Store

JIRA: https://jira.spring.io/browse/INT-3325
JIRA: https://jira.spring.io/browse/INT-1870

Optimized MGS for QueueChannel - uses a LIST for
each channel and LPUSH, RPOP.

* Also fix MutableMessage to be Serializable

INT-1870 Priority Redis Channel Message Store

Supports priorities 0-9 (+ no priority).

Priorities out of that range are treated as no priority.

Polishing - Add Marker Interfaces

* Emit a `WARN` log if a channel is used with a regular MessageGroupStore
* Allow message-store on namespace when defining a priority channel

INT-3325 Polishing; PR Comments

Fix some typos in JavaDocs and Docs

spring-integration-core/src/main/java/org/springframework/integration/config/xml/PointToPointChannelParser.java

@@ -50,9 +50,10 @@ public class PointToPointChannelParser extends AbstractChannelParser {
 		boolean isFixedSubscriber = "true".equals(fixedSubscriberChannel.trim().toLowerCase());
 
 		// configure a queue-based channel if any queue sub-element is defined
+		String channel = element.getAttribute(ID_ATTRIBUTE);
 		if ((queueElement = DomUtils.getChildElementByTagName(element, "queue")) != null) {
 			builder = BeanDefinitionBuilder.genericBeanDefinition(QueueChannel.class);
-			boolean hasStoreRef = this.parseStoreRef(builder, queueElement, element.getAttribute(ID_ATTRIBUTE));
+			boolean hasStoreRef = this.parseStoreRef(builder, queueElement, channel, false);
 			boolean hasQueueRef = this.parseQueueRef(builder, queueElement);
 			if (!hasStoreRef) {
 				boolean hasCapacity = this.parseQueueCapacity(builder, queueElement);
@@ -75,6 +76,15 @@ public class PointToPointChannelParser extends AbstractChannelParser {
 			if (StringUtils.hasText(comparatorRef)) {
 				builder.addConstructorArgReference(comparatorRef);
 			}
+			if (parseStoreRef(builder, queueElement, channel, true)) {
+				if (StringUtils.hasText(comparatorRef)) {
+					parserContext.getReaderContext().error(
+							"The 'message-store' attribute is not allowed" + " when providing a 'comparator' to a priority queue.",
+							element);
+				}
+				builder.getRawBeanDefinition().setBeanClass(QueueChannel.class);
+			}
+
 		}
 		else if ((queueElement = DomUtils.getChildElementByTagName(element, "rendezvous-queue")) != null) {
 			builder = BeanDefinitionBuilder.genericBeanDefinition(RendezvousChannel.class);
@@ -158,13 +168,14 @@ public class PointToPointChannelParser extends AbstractChannelParser {
 		return false;
 	}
 
-	private boolean parseStoreRef(BeanDefinitionBuilder builder, Element queueElement, String channel) {
+	private boolean parseStoreRef(BeanDefinitionBuilder builder, Element queueElement, String channel, boolean priority) {
 		String storeRef = queueElement.getAttribute("message-store");
 		if (StringUtils.hasText(storeRef)) {
 			BeanDefinitionBuilder queueBuilder = BeanDefinitionBuilder
 					.genericBeanDefinition(MessageGroupQueue.class);
 			queueBuilder.addConstructorArgReference(storeRef);
 			queueBuilder.addConstructorArgValue(new TypedStringValue(storeRef).getValue() + ":" + channel);
+			queueBuilder.addPropertyValue("priority", priority);
 			parseQueueCapacity(queueBuilder, queueElement);
 			builder.addConstructorArgValue(queueBuilder.getBeanDefinition());
 			return true;

spring-integration-core/src/main/java/org/springframework/integration/message/MutableMessage.java

@@ -15,6 +15,7 @@
  */
 package org.springframework.integration.message;
 
+import java.io.Serializable;
 import java.util.Map;
 
 import org.springframework.beans.DirectFieldAccessor;
@@ -36,7 +37,9 @@ import org.springframework.util.ObjectUtils;
  * @since 4.0
  *
  */
-public class MutableMessage<T> implements Message<T> {
+public class MutableMessage<T> implements Message<T>, Serializable {
+
+	private static final long serialVersionUID = -636635024258737500L;
 
 	private T payload;
 

spring-integration-core/src/main/java/org/springframework/integration/store/BasicMessageGroupStore.java

@@ -0,0 +1,66 @@
+/*
+ * Copyright 2014 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.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.springframework.integration.store;
+
+import org.springframework.jmx.export.annotation.ManagedAttribute;
+import org.springframework.messaging.Message;
+
+/**
+ * Defines a minimal message group store with basic capabilities.
+ *
+ * @author Gary Russell
+ * @since 4.0
+ *
+ */
+public interface BasicMessageGroupStore {
+
+	/**
+	 * Returns the size of this MessageGroup.
+	 *
+	 * @param groupId The group identifier.
+	 * @return The size.
+	 */
+	@ManagedAttribute
+	int messageGroupSize(Object groupId);
+
+	/**
+	 * Return all Messages currently in the MessageStore that were stored using
+	 * {@link #addMessageToGroup(Object, Message)} with this group id.
+	 *
+	 * @param groupId The group identifier.
+	 * @return A group of messages, empty if none exists for this key.
+	 */
+	MessageGroup getMessageGroup(Object groupId);
+
+	/**
+	 * Store a message with an association to a group id. This can be used to group messages together.
+	 *
+	 * @param groupId The group id to store the message under.
+	 * @param message A message.
+	 * @return The message group.
+	 */
+	MessageGroup addMessageToGroup(Object groupId, Message<?> message);
+
+	/**
+	 * Polls Message from this {@link MessageGroup} (in FIFO style if supported by the implementation)
+	 * while also removing the polled {@link Message}
+	 *
+	 * @param groupId The group identifier.
+	 * @return The message.
+	 */
+	Message<?> pollMessageFromGroup(Object groupId);
+
+}

spring-integration-core/src/main/java/org/springframework/integration/store/ChannelMessageStore.java

@@ -0,0 +1,30 @@
+/*
+ * Copyright 2014 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.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.springframework.integration.store;
+
+import org.springframework.integration.channel.QueueChannel;
+
+/**
+ * A marker interface that indicates this message store has optimizations for
+ * use in a {@link QueueChannel}.
+ *
+ * @author Gary Russell
+ * @since 4.0
+ *
+ */
+public interface ChannelMessageStore extends BasicMessageGroupStore {
+
+}

spring-integration-core/src/main/java/org/springframework/integration/store/MessageGroupQueue.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2012 the original author or authors.
+ * Copyright 2002-2014 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.
@@ -40,6 +40,7 @@ import org.springframework.util.Assert;
  * @author Dave Syer
  * @author Oleg Zhurakousky
  * @author Gunnar Hillert
+ * @author Gary Russell
  *
  * @since 2.0
  *
@@ -50,7 +51,7 @@ public class MessageGroupQueue extends AbstractQueue<Message<?>> implements Bloc
 
 	private static final int DEFAULT_CAPACITY = Integer.MAX_VALUE;
 
-	private final MessageGroupStore messageGroupStore;
+	private final BasicMessageGroupStore messageGroupStore;
 
 	private final Object groupId;
 
@@ -63,19 +64,19 @@ public class MessageGroupQueue extends AbstractQueue<Message<?>> implements Bloc
 
 	private final Condition messageStoreNotEmpty;
 
-	public MessageGroupQueue(MessageGroupStore messageGroupStore, Object groupId) {
+	public MessageGroupQueue(BasicMessageGroupStore messageGroupStore, Object groupId) {
 		this(messageGroupStore, groupId, DEFAULT_CAPACITY, new ReentrantLock(true));
 	}
 
-	public MessageGroupQueue(MessageGroupStore messageGroupStore, Object groupId, int capacity) {
+	public MessageGroupQueue(BasicMessageGroupStore messageGroupStore, Object groupId, int capacity) {
 		this(messageGroupStore, groupId, capacity, new ReentrantLock(true));
 	}
 
-	public MessageGroupQueue(MessageGroupStore messageGroupStore, Object groupId, Lock storeLock) {
+	public MessageGroupQueue(BasicMessageGroupStore messageGroupStore, Object groupId, Lock storeLock) {
 		this(messageGroupStore, groupId, DEFAULT_CAPACITY, storeLock);
 	}
 
-	public MessageGroupQueue(MessageGroupStore messageGroupStore, Object groupId, int capacity, Lock storeLock) {
+	public MessageGroupQueue(BasicMessageGroupStore messageGroupStore, Object groupId, int capacity, Lock storeLock) {
 		Assert.isTrue(capacity > 0, "'capacity' must be greater than 0");
 		Assert.notNull(storeLock, "'storeLock' must not be null");
 		Assert.notNull(messageGroupStore, "'messageGroupStore' must not be null");
@@ -86,16 +87,45 @@ public class MessageGroupQueue extends AbstractQueue<Message<?>> implements Bloc
 		this.messageGroupStore = messageGroupStore;
 		this.groupId = groupId;
 		this.capacity = capacity;
+		if (logger.isWarnEnabled() && !(messageGroupStore instanceof ChannelMessageStore)) {
+			logger.warn(messageGroupStore.getClass().getSimpleName() + " is not optimized for use "
+					+ "in a 'MessageGroupQueue'; consider using a `ChannelMessageStore'");
+		}
 	}
 
+	/**
+	 * If true, ensures that the message store supports priority. If false WARNs if the
+	 * message store uses priority to determine the message order when receiving.
+	 * @param priority true if priority is expected to be used.
+	 */
+	public void setPriority(boolean priority) {
+		if (priority) {
+			Assert.isInstanceOf(PriorityCapableChannelMessageStore.class, this.messageGroupStore);
+			Assert.isTrue(((PriorityCapableChannelMessageStore) this.messageGroupStore).isPriorityEnabled(),
+					"When using priority, the 'PriorityCapableChannelMessageStore' must have priority enabled.");
+		}
+		else {
+			if (logger.isWarnEnabled() && this.messageGroupStore instanceof PriorityCapableChannelMessageStore
+					&& ((PriorityCapableChannelMessageStore) this.messageGroupStore).isPriorityEnabled()) {
+				logger.warn("It's not recommended to use a priority-based message store " +
+						"when declaring a non-priority 'MessageGroupQueue'; message retrieval may not be FIFO; " +
+						"set 'priority' to 'true' if that is your intent. If you are using the namespace to " +
+						"define a channel, use '<priority-queue message-store.../> instead.");
+			}
+		}
+	}
+
+	@Override
 	public Iterator<Message<?>> iterator() {
 		return getMessages().iterator();
 	}
 
+	@Override
 	public int size() {
 		return messageGroupStore.messageGroupSize(groupId);
 	}
 
+	@Override
 	public Message<?> peek() {
 		Message<?> message = null;
 		final Lock storeLock = this.storeLock;
@@ -117,6 +147,7 @@ public class MessageGroupQueue extends AbstractQueue<Message<?>> implements Bloc
 		return message;
 	}
 
+	@Override
 	public Message<?> poll(long timeout, TimeUnit unit) throws InterruptedException {
 		Message<?> message = null;
 		long timeoutInNanos = unit.toNanos(timeout);
@@ -136,6 +167,7 @@ public class MessageGroupQueue extends AbstractQueue<Message<?>> implements Bloc
 		return message;
 	}
 
+	@Override
 	public Message<?> poll() {
 		Message<?> message = null;
 		final Lock storeLock = this.storeLock;
@@ -154,10 +186,12 @@ public class MessageGroupQueue extends AbstractQueue<Message<?>> implements Bloc
 		return message;
 	}
 
+	@Override
 	public int drainTo(Collection<? super Message<?>> c) {
 		return this.drainTo(c, Integer.MAX_VALUE);
 	}
 
+	@Override
 	public int drainTo(Collection<? super Message<?>> collection, int maxElements) {
 		Assert.notNull(collection, "'collection' must not be null");
 		int originalSize = collection.size();
@@ -185,6 +219,7 @@ public class MessageGroupQueue extends AbstractQueue<Message<?>> implements Bloc
 		return collection.size() - originalSize;
 	}
 
+	@Override
 	public boolean offer(Message<?> message) {
 		boolean offered = true;
 		final Lock storeLock = this.storeLock;
@@ -203,6 +238,7 @@ public class MessageGroupQueue extends AbstractQueue<Message<?>> implements Bloc
 		return offered;
 	}
 
+	@Override
 	public boolean offer(Message<?> message, long timeout, TimeUnit unit) throws InterruptedException {
 		long timeoutInNanos = unit.toNanos(timeout);
 		boolean offered = false;
@@ -225,6 +261,7 @@ public class MessageGroupQueue extends AbstractQueue<Message<?>> implements Bloc
 		return offered;
 	}
 
+	@Override
 	public void put(Message<?> message) throws InterruptedException {
 		final Lock storeLock = this.storeLock;
 		storeLock.lockInterruptibly();
@@ -241,6 +278,7 @@ public class MessageGroupQueue extends AbstractQueue<Message<?>> implements Bloc
 		}
 	}
 
+	@Override
 	public int remainingCapacity() {
 		if (capacity == Integer.MAX_VALUE) {
 			return Integer.MAX_VALUE;
@@ -248,6 +286,7 @@ public class MessageGroupQueue extends AbstractQueue<Message<?>> implements Bloc
 		return capacity - this.size();
 	}
 
+	@Override
 	public Message<?> take() throws InterruptedException {
 		Message<?> message = null;
 		final Lock storeLock = this.storeLock;

spring-integration-core/src/main/java/org/springframework/integration/store/MessageGroupStore.java

@@ -18,7 +18,7 @@ import org.springframework.jmx.export.annotation.ManagedAttribute;
 import org.springframework.messaging.Message;
 
 /**
- * Interface for storage operations on groups of messages linked by a group id.
+ * Defines additional storage operations on groups of messages linked by a group id.
  *
  * @author Dave Syer
  * @author Oleg Zhurakousky
@@ -27,7 +27,7 @@ import org.springframework.messaging.Message;
  * @since 2.0
  *
  */
-public interface MessageGroupStore {
+public interface MessageGroupStore extends BasicMessageGroupStore {
 
 	/**
 	 * Optional attribute giving the number of messages in the store over all groups. Implementations may decline to
@@ -38,6 +38,7 @@ public interface MessageGroupStore {
 	 */
 	@ManagedAttribute
 	int getMessageCountForAllMessageGroups();
+
 	/**
 	 * Optional attribute giving the number of  message groups. Implementations may decline
 	 * to respond by throwing an exception.
@@ -48,33 +49,6 @@ public interface MessageGroupStore {
 	@ManagedAttribute
 	int getMessageGroupCount();
 
-	/**
-	 * Returns the size of this MessageGroup.
-	 *
-	 * @param groupId The group identifier.
-	 * @return The size.
-	 */
-	@ManagedAttribute
-	int messageGroupSize(Object groupId);
-
-	/**
-	 * Return all Messages currently in the MessageStore that were stored using
-	 * {@link #addMessageToGroup(Object, Message)} with this group id.
-	 *
-	 * @param groupId The group identifier.
-	 * @return A group of messages, empty if none exists for this key.
-	 */
-	MessageGroup getMessageGroup(Object groupId);
-
-	/**
-	 * Store a message with an association to a group id. This can be used to group messages together.
-	 *
-	 * @param groupId The group id to store the message under.
-	 * @param message A message.
-	 * @return The message group.
-	 */
-	MessageGroup addMessageToGroup(Object groupId, Message<?> message);
-
 	/**
 	 * Persist a deletion on a single message from the group. The group is modified to reflect that 'messageToRemove' is
 	 * no longer present in the group.
@@ -125,16 +99,6 @@ public interface MessageGroupStore {
 	 */
 	Iterator<MessageGroup> iterator();
 
-
-	/**
-	 * Polls Message from this {@link MessageGroup} (in FIFO style if supported by the implementation)
-	 * while also removing the polled {@link Message}
-	 *
-	 * @param groupId The group identifier.
-	 * @return The message.
-	 */
-	Message<?> pollMessageFromGroup(Object groupId);
-
 	/**
 	 * Completes this MessageGroup. Completion of the MessageGroup generally means
 	 * that this group should not be allowing any more mutating operation to be performed on it.

spring-integration-core/src/main/java/org/springframework/integration/store/PriorityCapableChannelMessageStore.java

@@ -0,0 +1,34 @@
+/*
+ * Copyright 2014 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.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.springframework.integration.store;
+
+/**
+ * A {@link ChannelMessageStore} that supports the
+ * notion of message priority. It is left to implementations to determine what
+ * that means and whether all or a subset of priorities are supported.
+ *
+ * @author Gary Russell
+ * @since 4.0
+ *
+ */
+public interface PriorityCapableChannelMessageStore extends ChannelMessageStore {
+
+	/**
+	 * @return true if message priority is enabled in this channel message store.
+	 */
+	boolean isPriorityEnabled();
+
+}

spring-integration-core/src/main/java/org/springframework/integration/store/SimpleMessageStore.java

@@ -44,7 +44,8 @@ import org.springframework.util.CollectionUtils;
  * @since 2.0
  */
 @ManagedResource
-public class SimpleMessageStore extends AbstractMessageGroupStore implements MessageStore, MessageGroupStore {
+public class SimpleMessageStore extends AbstractMessageGroupStore
+		implements MessageStore, ChannelMessageStore {
 
 	private volatile LockRegistry lockRegistry;
 

spring-integration-core/src/main/resources/org/springframework/integration/config/xml/spring-integration-4.0.xsd

@@ -274,11 +274,26 @@
 				<xsd:documentation>
 					<![CDATA[
 	Allows you to specify the reference to the bean which implements java.util.Comparator&lt;Message&lt;?&gt;&gt;
-	interface and provides logic based on which Messages will be prioritized.
+	interface and provides logic based on which Messages will be prioritized. Not allowed if `message-store` is set.
 					]]>
 				</xsd:documentation>
 			</xsd:annotation>
 		</xsd:attribute>
+		<xsd:attribute name="message-store" type="xsd:string">
+				<xsd:annotation>
+						<xsd:appinfo>
+								<tool:annotation kind="ref">
+										<tool:expected-type type="org.springframework.integration.store.PriorityCapableChannelMessageStore" />
+								</tool:annotation>
+						</xsd:appinfo>
+						<xsd:documentation>
+						A reference to a bean that implements 'org.springframework.integration.store.PriorityCapableChannelMessageStore'.
+						A message store that supports priority in a manner defined by the store. When set, the underlying
+						channel will be a 'QueueChannel` that delegates to a `MessageGroupQueue' backed by the store.
+						Not allowed if 'comparator' is set.
+						</xsd:documentation>
+				</xsd:annotation>
+		</xsd:attribute>
 	</xsd:complexType>
 
 	<xsd:complexType name="rendezvousQueueType">

spring-integration-core/src/test/java/org/springframework/integration/config/ChannelWithMessageStoreParserTests-context.xml

@@ -23,4 +23,10 @@
 		<beans:property name="replyMessageText" value="hello"/>
 	</beans:bean>
 
+	<channel id="priority">
+		<priority-queue message-store="priorityMessageStore" />
+	</channel>
+
+	<beans:bean id="priorityMessageStore" class="org.springframework.integration.config.ChannelWithMessageStoreParserTests$DummyPriorityMS" />
+
 </beans:beans>

spring-integration-core/src/test/java/org/springframework/integration/config/ChannelWithMessageStoreParserTests.java

@@ -17,18 +17,23 @@
 package org.springframework.integration.config;
 
 import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertSame;
 
 import java.util.concurrent.TimeUnit;
 
 import org.junit.Test;
 import org.junit.runner.RunWith;
+
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.beans.factory.annotation.Qualifier;
+import org.springframework.integration.store.MessageGroupStore;
+import org.springframework.integration.store.PriorityCapableChannelMessageStore;
+import org.springframework.integration.store.SimpleMessageStore;
+import org.springframework.integration.support.MessageBuilder;
+import org.springframework.integration.test.util.TestUtils;
 import org.springframework.messaging.Message;
 import org.springframework.messaging.MessageChannel;
 import org.springframework.messaging.PollableChannel;
-import org.springframework.integration.store.MessageGroupStore;
-import org.springframework.integration.support.MessageBuilder;
 import org.springframework.test.annotation.DirtiesContext;
 import org.springframework.test.context.ContextConfiguration;
 import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
@@ -39,7 +44,7 @@ import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
 @ContextConfiguration
 @RunWith(SpringJUnit4ClassRunner.class)
 public class ChannelWithMessageStoreParserTests {
-	
+
 	private static final String BASE_PACKAGE = "org.springframework.integration";
 
 	@Autowired
@@ -53,9 +58,15 @@ public class ChannelWithMessageStoreParserTests {
 	@Autowired
 	private TestHandler handler;
 
-	@Autowired
+	@Autowired @Qualifier("messageStore")
 	private MessageGroupStore messageGroupStore;
 
+	@Autowired @Qualifier("priority")
+	private PollableChannel priorityChannel;
+
+	@Autowired @Qualifier("priorityMessageStore")
+	private MessageGroupStore priorityMessageStore;
+
 	@Test
 	@DirtiesContext
 	public void testActivatorSendsToPersistentQueue() throws Exception {
@@ -65,17 +76,32 @@ public class ChannelWithMessageStoreParserTests {
 		assertEquals("The message payload is not correct", "123", handler.getMessageString());
 		// The group id for buffered messages is the channel name
 		assertEquals(1, messageGroupStore.getMessageGroup("messageStore:output").size());
-		
+
 		Message<?> result = output.receive(100);
 		assertEquals("hello", result.getPayload());
 		assertEquals(0, messageGroupStore.getMessageGroup(BASE_PACKAGE+".store:output").size());
 
 	}
 
+	@Test
+	@DirtiesContext
+	public void testPriorityMessageStore() {
+		assertSame(this.priorityMessageStore, TestUtils.getPropertyValue(this.priorityChannel, "queue.messageGroupStore"));
+	}
+
 	private static <T> Message<T> createMessage(T payload, Object correlationId, int sequenceSize, int sequenceNumber,
 			MessageChannel outputChannel) {
 		return MessageBuilder.withPayload(payload).setCorrelationId(correlationId).setSequenceSize(sequenceSize)
 				.setSequenceNumber(sequenceNumber).setReplyChannel(outputChannel).build();
 	}
 
+	public static class DummyPriorityMS extends SimpleMessageStore implements PriorityCapableChannelMessageStore {
+
+		@Override
+		public boolean isPriorityEnabled() {
+			return true;
+		}
+
+	}
+
 }