diff --git a/spring-integration-core/src/main/java/org/springframework/integration/channel/PriorityChannel.java b/spring-integration-core/src/main/java/org/springframework/integration/channel/PriorityChannel.java index c61775cbfd..6817265e68 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/channel/PriorityChannel.java +++ b/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 null, 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> comparator) { - super(new PriorityBlockingQueue>(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 null, 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> 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 { + 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; } + } + } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/config/xml/PointToPointChannelParser.java b/spring-integration-core/src/main/java/org/springframework/integration/config/xml/PointToPointChannelParser.java index 218c623d94..c66f6a08b3 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/config/xml/PointToPointChannelParser.java +++ b/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); } } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/dsl/Channels.java b/spring-integration-core/src/main/java/org/springframework/integration/dsl/Channels.java index 17e9cb70e5..2d1475ad9e 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/dsl/Channels.java +++ b/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); } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/dsl/channel/MessageChannels.java b/spring-integration-core/src/main/java/org/springframework/integration/dsl/channel/MessageChannels.java index 611ac4376f..f1cd95ad5a 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/dsl/channel/MessageChannels.java +++ b/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 > PublishSubscribeChannelSpec publishSubscribe() { - return new PublishSubscribeChannelSpec(); + return new PublishSubscribeChannelSpec<>(); } public static > PublishSubscribeChannelSpec publishSubscribe( @@ -116,7 +115,7 @@ public final class MessageChannels { public static > PublishSubscribeChannelSpec publishSubscribe( Executor executor) { - return new PublishSubscribeChannelSpec(executor); + return new PublishSubscribeChannelSpec<>(executor); } public static > PublishSubscribeChannelSpec publishSubscribe(String id, @@ -125,6 +124,7 @@ public final class MessageChannels { } private MessageChannels() { + super(); } } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/dsl/channel/PriorityChannelSpec.java b/spring-integration-core/src/main/java/org/springframework/integration/dsl/channel/PriorityChannelSpec.java index 5aeb49dea0..33fdb647b5 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/dsl/channel/PriorityChannelSpec.java +++ b/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> comparator; - public PriorityChannelSpec setCapacity(int capacity) { - this.capacity = capacity; - return this; - } - - public PriorityChannelSpec setComparator(Comparator> 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> 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(); + } + } diff --git a/spring-integration-core/src/test/java/org/springframework/integration/config/ChannelWithMessageStoreParserTests.java b/spring-integration-core/src/test/java/org/springframework/integration/config/ChannelWithMessageStoreParserTests.java index df19ce7a19..4838a04bc4 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/config/ChannelWithMessageStoreParserTests.java +++ b/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 Message createMessage(T payload, Object correlationId, int sequenceSize, int sequenceNumber, diff --git a/spring-integration-mongodb/src/test/java/org/springframework/integration/mongodb/store/ConfigurableMongoDbMessageGroupStoreTests.java b/spring-integration-mongodb/src/test/java/org/springframework/integration/mongodb/store/ConfigurableMongoDbMessageGroupStoreTests.java index f1f8ab913d..24a81c6ec6 100644 --- a/spring-integration-mongodb/src/test/java/org/springframework/integration/mongodb/store/ConfigurableMongoDbMessageGroupStoreTests.java +++ b/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; diff --git a/src/reference/asciidoc/channel.adoc b/src/reference/asciidoc/channel.adoc index 27a8a41b17..d6081281f5 100644 --- a/src/reference/asciidoc/channel.adoc +++ b/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 `` sub-element. +Another option to customize the `QueueChannel` environment is provided by the `ref` attribute of the `` 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>() - .codec(new JavaSerializationCodec>()) - .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 <> and <> for more information. You can find sample configuration in <>.