INT-4012: PriorityChannel refinement

JIRA: https://jira.spring.io/browse/INT-4012

Previously to configure a `priority` for the `MessageChannel` we should configure `QueueChannel` for particular `MessageStore`.
Although the target priority logic is really in the `MessageStore` implementation, it isn't so obvious why we can't use `PriorityChannel` instance for `MessageStore` case as well.

* Add `PriorityCapableChannelMessageStore` and `MessageGroupQueue` based ctors to the `PriorityChannel` for consistency.
* Delegate the logic to the super `QueueChannel` as before
* Rework `PointToPointChannelParser` and Java DSL components to reflect a new state of the `PriorityChannel`
* Improve Docs on the matter

spring-integration-core/src/main/java/org/springframework/integration/channel/PriorityChannel.java

@@ -21,6 +21,8 @@ import java.util.concurrent.PriorityBlockingQueue;
 import java.util.concurrent.atomic.AtomicLong;
 
 import org.springframework.integration.IntegrationMessageHeaderAccessor;
+import org.springframework.integration.store.MessageGroupQueue;
+import org.springframework.integration.store.PriorityCapableChannelMessageStore;
 import org.springframework.integration.util.UpperBound;
 import org.springframework.messaging.Message;
 import org.springframework.messaging.MessageHeaders;
@@ -32,6 +34,7 @@ import org.springframework.messaging.MessageHeaders;
  * @author Mark Fisher
  * @author Oleg Zhurakousky
  * @author Gary Russell
+ * @author Artem Bilan
  */
 public class PriorityChannel extends QueueChannel {
 
@@ -39,19 +42,14 @@ public class PriorityChannel extends QueueChannel {
 
 	private final AtomicLong sequenceCounter = new AtomicLong();
 
+	private final boolean useMessageStore;
+
 	/**
-	 * Create a channel with the specified queue capacity. If the capacity
-	 * is a non-positive value, the queue will be unbounded. Message priority
-	 * will be determined by the provided {@link Comparator}. If the comparator
-	 * is <code>null</code>, the priority will be based upon the value of
-	 * {@link IntegrationMessageHeaderAccessor#getPriority()}.
-	 *
-	 * @param capacity The capacity.
-	 * @param comparator The comparator.
+	 * Create a channel with an unbounded queue. Message priority will be
+	 * based on the value of {@link IntegrationMessageHeaderAccessor#getPriority()}.
 	 */
-	public PriorityChannel(int capacity, Comparator<Message<?>> comparator) {
-		super(new PriorityBlockingQueue<Message<?>>(11, new SequenceFallbackComparator(comparator)));
-		this.upperBound = new UpperBound(capacity);
+	public PriorityChannel() {
+		this(0, null);
 	}
 
 	/**
@@ -77,11 +75,41 @@ public class PriorityChannel extends QueueChannel {
 	}
 
 	/**
-	 * Create a channel with an unbounded queue. Message priority will be
-	 * based on the value of {@link IntegrationMessageHeaderAccessor#getPriority()}.
+	 * Create a channel with the specified queue capacity. If the capacity
+	 * is a non-positive value, the queue will be unbounded. Message priority
+	 * will be determined by the provided {@link Comparator}. If the comparator
+	 * is <code>null</code>, the priority will be based upon the value of
+	 * {@link IntegrationMessageHeaderAccessor#getPriority()}.
+	 *
+	 * @param capacity The capacity.
+	 * @param comparator The comparator.
 	 */
-	public PriorityChannel() {
-		this(0, null);
+	public PriorityChannel(int capacity, Comparator<Message<?>> comparator) {
+		super(new PriorityBlockingQueue<>(11, new SequenceFallbackComparator(comparator)));
+		this.upperBound = new UpperBound(capacity);
+		this.useMessageStore = false;
+	}
+
+	/**
+	 * Create a channel based on the provided {@link PriorityCapableChannelMessageStore}
+	 * and group id for message store operations.
+	 * @param messageGroupStore the {@link PriorityCapableChannelMessageStore} to use.
+	 * @param groupId to group message for this channel in the message store.
+	 * @since 5.0
+	 */
+	public PriorityChannel(PriorityCapableChannelMessageStore messageGroupStore, Object groupId) {
+		this(new MessageGroupQueue(messageGroupStore, groupId));
+	}
+
+	/**
+	 * Create a channel based on the provided {@link MessageGroupQueue}.
+	 * @param messageGroupQueue the {@link MessageGroupQueue} to use.
+	 * @since 5.0
+	 */
+	public PriorityChannel(MessageGroupQueue messageGroupQueue) {
+		super(messageGroupQueue);
+		this.upperBound = new UpperBound(0);
+		this.useMessageStore = true;
 	}
 
 	@Override
@@ -94,7 +122,9 @@ public class PriorityChannel extends QueueChannel {
 		if (!this.upperBound.tryAcquire(timeout)) {
 			return false;
 		}
-		message = new MessageWrapper(message);
+		if (!this.useMessageStore) {
+			message = new MessageWrapper(message);
+		}
 		return super.doSend(message, 0);
 	}
 
@@ -102,7 +132,9 @@ public class PriorityChannel extends QueueChannel {
 	protected Message<?> doReceive(long timeout) {
 		Message<?> message = super.doReceive(timeout);
 		if (message != null) {
-			message = ((MessageWrapper) message).getRootMessage();
+			if (!this.useMessageStore) {
+				message = ((MessageWrapper) message).getRootMessage();
+			}
 			this.upperBound.release();
 		}
 		return message;
@@ -138,11 +170,14 @@ public class PriorityChannel extends QueueChannel {
 			}
 			return compareResult;
 		}
+
 	}
 
 	//we need this because of INT-2508
 	private final class MessageWrapper implements Message<Object> {
+
 		private final Message<?> rootMessage;
+
 		private final long sequence;
 
 		MessageWrapper(Message<?> rootMessage) {
@@ -167,5 +202,7 @@ public class PriorityChannel extends QueueChannel {
 		long getSequence() {
 			return this.sequence;
 		}
+
 	}
+
 }

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

@@ -94,7 +94,6 @@ public class PointToPointChannelParser extends AbstractChannelParser {
 					parserContext.getReaderContext().error("The 'capacity' attribute is not allowed"
 							+ " when providing a 'message-store' to a custom MessageGroupStore.", element);
 				}
-				builder.getRawBeanDefinition().setBeanClass(QueueChannel.class);
 			}
 
 		}

spring-integration-core/src/main/java/org/springframework/integration/dsl/Channels.java

@@ -86,20 +86,19 @@ public class Channels {
 		return MessageChannels.priority(id);
 	}
 
-	public QueueChannelSpec.MessageStoreSpec priority(String id, PriorityCapableChannelMessageStore messageGroupStore,
+	public PriorityChannelSpec priority(String id, PriorityCapableChannelMessageStore messageGroupStore,
 			Object groupId) {
 		return MessageChannels.priority(id, messageGroupStore, groupId);
 	}
 
-	public QueueChannelSpec.MessageStoreSpec priority(PriorityCapableChannelMessageStore messageGroupStore,
-			Object groupId) {
-		return MessageChannels.priority(messageGroupStore, groupId);
-	}
-
 	public RendezvousChannelSpec rendezvous() {
 		return MessageChannels.rendezvous();
 	}
 
+	public PriorityChannelSpec priority(PriorityCapableChannelMessageStore messageGroupStore, Object groupId) {
+		return MessageChannels.priority(messageGroupStore, groupId);
+	}
+
 	public RendezvousChannelSpec rendezvous(String id) {
 		return MessageChannels.rendezvous(id);
 	}

spring-integration-core/src/main/java/org/springframework/integration/dsl/channel/MessageChannels.java

@@ -95,18 +95,17 @@ public final class MessageChannels {
 		return priority().id(id);
 	}
 
-	public static QueueChannelSpec.MessageStoreSpec priority(PriorityCapableChannelMessageStore messageGroupStore,
-			Object groupId) {
-		return new QueueChannelSpec.MessageStoreSpec(messageGroupStore, groupId);
+	public static PriorityChannelSpec priority(PriorityCapableChannelMessageStore messageGroupStore, Object groupId) {
+		return priority().messageStore(messageGroupStore, groupId);
 	}
 
-	public static QueueChannelSpec.MessageStoreSpec priority(String id,
+	public static PriorityChannelSpec priority(String id,
 			PriorityCapableChannelMessageStore messageGroupStore, Object groupId) {
-		return queue(messageGroupStore, groupId).id(id);
+		return priority(messageGroupStore, groupId).id(id);
 	}
 
 	public static <S extends PublishSubscribeChannelSpec<S>> PublishSubscribeChannelSpec<S> publishSubscribe() {
-		return new PublishSubscribeChannelSpec<S>();
+		return new PublishSubscribeChannelSpec<>();
 	}
 
 	public static <S extends PublishSubscribeChannelSpec<S>> PublishSubscribeChannelSpec<S> publishSubscribe(
@@ -116,7 +115,7 @@ public final class MessageChannels {
 
 	public static <S extends PublishSubscribeChannelSpec<S>> PublishSubscribeChannelSpec<S> publishSubscribe(
 			Executor executor) {
-		return new PublishSubscribeChannelSpec<S>(executor);
+		return new PublishSubscribeChannelSpec<>(executor);
 	}
 
 	public static <S extends PublishSubscribeChannelSpec<S>> PublishSubscribeChannelSpec<S> publishSubscribe(String id,
@@ -125,6 +124,7 @@ public final class MessageChannels {
 	}
 
 	private MessageChannels() {
+		super();
 	}
 
 }

spring-integration-core/src/main/java/org/springframework/integration/dsl/channel/PriorityChannelSpec.java

@@ -19,7 +19,10 @@ package org.springframework.integration.dsl.channel;
 import java.util.Comparator;
 
 import org.springframework.integration.channel.PriorityChannel;
+import org.springframework.integration.store.MessageGroupQueue;
+import org.springframework.integration.store.PriorityCapableChannelMessageStore;
 import org.springframework.messaging.Message;
+import org.springframework.util.Assert;
 
 /**
  * @author Artem Bilan
@@ -32,25 +35,38 @@ public class PriorityChannelSpec extends MessageChannelSpec<PriorityChannelSpec,
 
 	private Comparator<Message<?>> comparator;
 
-	public PriorityChannelSpec setCapacity(int capacity) {
-		this.capacity = capacity;
-		return this;
-	}
-
-	public PriorityChannelSpec setComparator(Comparator<Message<?>> comparator) {
-		this.comparator = comparator;
-		return this;
-	}
-
-	@Override
-	protected PriorityChannel doGet() {
-		this.channel = new PriorityChannel(this.capacity, this.comparator);
-		return super.doGet();
-	}
-
+	private MessageGroupQueue messageGroupQueue;
 
 	PriorityChannelSpec() {
 		super();
 	}
 
+	public PriorityChannelSpec capacity(int capacity) {
+		this.capacity = capacity;
+		return this;
+	}
+
+	public PriorityChannelSpec comparator(Comparator<Message<?>> comparator) {
+		this.comparator = comparator;
+		return this;
+	}
+
+	public PriorityChannelSpec messageStore(PriorityCapableChannelMessageStore messageGroupStore, Object groupId) {
+		this.messageGroupQueue = new MessageGroupQueue(messageGroupStore, groupId);
+		return this;
+	}
+
+	@Override
+	protected PriorityChannel doGet() {
+		Assert.state(this.comparator != null && this.messageGroupQueue != null,
+				"Only one of 'comparator' or 'messageGroupStore' can be specified.");
+		if (this.messageGroupQueue != null) {
+			this.channel = new PriorityChannel(this.messageGroupQueue);
+		}
+		else {
+			this.channel = new PriorityChannel(this.capacity, this.comparator);
+		}
+		return super.doGet();
+	}
+
 }

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

@@ -16,8 +16,10 @@
 
 package org.springframework.integration.config;
 
+import static org.hamcrest.Matchers.instanceOf;
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertSame;
+import static org.junit.Assert.assertThat;
 
 import java.util.concurrent.TimeUnit;
 
@@ -26,6 +28,7 @@ import org.junit.runner.RunWith;
 
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.beans.factory.annotation.Qualifier;
+import org.springframework.integration.channel.PriorityChannel;
 import org.springframework.integration.store.MessageGroupStore;
 import org.springframework.integration.store.PriorityCapableChannelMessageStore;
 import org.springframework.integration.store.SimpleMessageStore;
@@ -35,14 +38,14 @@ import org.springframework.messaging.Message;
 import org.springframework.messaging.MessageChannel;
 import org.springframework.messaging.PollableChannel;
 import org.springframework.test.annotation.DirtiesContext;
-import org.springframework.test.context.ContextConfiguration;
-import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
+import org.springframework.test.context.junit4.SpringRunner;
 
 /**
  * @author Dave Syer
+ * @author Artem Bilan
  */
-@ContextConfiguration
-@RunWith(SpringJUnit4ClassRunner.class)
+@RunWith(SpringRunner.class)
+@DirtiesContext
 public class ChannelWithMessageStoreParserTests {
 
 	private static final String BASE_PACKAGE = "org.springframework.integration";
@@ -58,13 +61,16 @@ public class ChannelWithMessageStoreParserTests {
 	@Autowired
 	private TestHandler handler;
 
-	@Autowired @Qualifier("messageStore")
+	@Autowired
+	@Qualifier("messageStore")
 	private MessageGroupStore messageGroupStore;
 
-	@Autowired @Qualifier("priority")
+	@Autowired
+	@Qualifier("priority")
 	private PollableChannel priorityChannel;
 
-	@Autowired @Qualifier("priorityMessageStore")
+	@Autowired
+	@Qualifier("priorityMessageStore")
 	private MessageGroupStore priorityMessageStore;
 
 	@Test
@@ -87,6 +93,7 @@ public class ChannelWithMessageStoreParserTests {
 	@DirtiesContext
 	public void testPriorityMessageStore() {
 		assertSame(this.priorityMessageStore, TestUtils.getPropertyValue(this.priorityChannel, "queue.messageGroupStore"));
+		assertThat(this.priorityChannel, instanceOf(PriorityChannel.class));
 	}
 
 	private static <T> Message<T> createMessage(T payload, Object correlationId, int sequenceSize, int sequenceNumber,

spring-integration-mongodb/src/test/java/org/springframework/integration/mongodb/store/ConfigurableMongoDbMessageGroupStoreTests.java

@@ -140,8 +140,7 @@ public class ConfigurableMongoDbMessageGroupStoreTests extends AbstractMongoDbMe
 		context.refresh();
 
 		Object priorityChannel = context.getBean("priorityChannel");
-		assertThat(priorityChannel, Matchers.not(Matchers.instanceOf(PriorityChannel.class)));
-		assertThat(priorityChannel, Matchers.instanceOf(QueueChannel.class));
+		assertThat(priorityChannel, Matchers.instanceOf(PriorityChannel.class));
 
 		QueueChannel channel = (QueueChannel) priorityChannel;
 

src/reference/asciidoc/channel.adoc

@@ -520,7 +520,7 @@ public BasicMessageGroupStore mongoDbChannelMessageStore(MongoDbFactory mongoDbF
 
 @Bean
 public PollableChannel priorityQueue(BasicMessageGroupStore mongoDbChannelMessageStore) {
-    return new QueueChannel(new MessageGroupQueue(mongoDbChannelMessageStore, "priorityQueue"));
+    return new PriorityChannel(new MessageGroupQueue(mongoDbChannelMessageStore, "priorityQueue"));
 }
 ----
 
@@ -540,18 +540,22 @@ public IntegrationFlow priorityFlow(PriorityCapableChannelMessageStore mongoDbCh
 }
 ----
 
-Another option to customize the QueueChannel environment is provided by the `ref` attribute of the `<int:queue>` sub-element.
+Another option to customize the `QueueChannel` environment is provided by the `ref` attribute of the `<int:queue>` sub-element or particular constructor.
 This attribute implies the reference to any `java.util.Queue` implementation.
-An implementation is provided by the https://github.com/reactor/reactor[Project Reactor] and its `reactor.queue.PersistentQueue` implementation for the https://github.com/OpenHFT/Chronicle-Queue[IndexedChronicle]:
+For example Hazelcast distributed https://hazelcast.com/use-cases/imdg/imdg-messaging/[`IQueue`]:
 
 [source,java]
 ----
 @Bean
-public QueueChannel reactorQueue() {
-    return new QueueChannel(new PersistentQueueSpec<Message<?>>()
-                    .codec(new JavaSerializationCodec<Message<?>>())
-                    .basePath(System.getProperty("java.io.tmpdir") + "/reactor-queue")
-                    .get());
+public HazelcastInstance hazelcastInstance() {
+    return Hazelcast.newHazelcastInstance(new Config()
+                                           .setProperty("hazelcast.logging.type", "log4j"));
+}
+
+@Bean
+public PollableChannel distributedQueue() {
+    return new QueueChannel(hazelcastInstance()
+                              .getQueue("springIntegrationQueue"));
 }
 ----
 
@@ -632,7 +636,7 @@ The following example demonstrates all of these:
 ----
 
 Since _version 4.0_, the `priority-channel` child element supports the `message-store` option (`comparator` and `capacity` are not allowed in that case).
-The message store must be a `PriorityCapableChannelMessageStore` and, in this case, the namespace parser will declare a `QueueChannel` instead of a `PriorityChannel`.
+The message store must be a `PriorityCapableChannelMessageStore` and, in this case.
 Implementations of the `PriorityCapableChannelMessageStore` are currently provided for `Redis`, `JDBC` and `MongoDB`.
 See <<channel-configuration-queuechannel>> and <<message-store>> for more information.
 You can find sample configuration in <<jdbc-message-store-channels>>.