Upgrade to SI 4.0

* Add new correlation endpoint options
* Add `messageStore` to the `PriorityChannelSpec`
* Add generic to `EnricherSpec#property`
* Remove `type` from `EnricherSpec.headerExpression`
* Add `initialDelay` for `PeriodicTriggerSpec`
* Fix `IntegrationFlowBuilder` bugs, when there is no relevant the end of the flow
* Add embedded MongoDb support for tests
* Add `priority` test with `MongoDbChannelMessageStore`

spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/CorrelationHandlerSpec.java

@@ -16,6 +16,9 @@
 
 package org.springframework.integration.dsl;
 
+import org.springframework.expression.Expression;
+import org.springframework.expression.common.LiteralExpression;
+import org.springframework.expression.spel.standard.SpelExpressionParser;
 import org.springframework.integration.aggregator.AbstractCorrelatingMessageHandler;
 import org.springframework.integration.aggregator.CorrelationStrategy;
 import org.springframework.integration.aggregator.ExpressionEvaluatingCorrelationStrategy;
@@ -26,6 +29,7 @@ import org.springframework.integration.config.ReleaseStrategyFactoryBean;
 import org.springframework.integration.dsl.core.IntegrationComponentSpec;
 import org.springframework.integration.store.MessageGroupStore;
 import org.springframework.messaging.MessageChannel;
+import org.springframework.scheduling.TaskScheduler;
 
 /**
  * @author Artem Bilan
@@ -33,12 +37,18 @@ import org.springframework.messaging.MessageChannel;
 public abstract class CorrelationHandlerSpec<S extends CorrelationHandlerSpec<S, H>, H extends AbstractCorrelatingMessageHandler>
 		extends IntegrationComponentSpec<S, H> {
 
+	protected final static SpelExpressionParser PARSER = new SpelExpressionParser();
+
 	protected MessageGroupStore messageStore;
 
 	protected boolean sendPartialResultOnExpiry;
 
 	private long minimumTimeoutForEmptyGroups;
 
+	private Expression groupTimeoutExpression;
+
+	private TaskScheduler taskScheduler;
+
 	private MessageChannel discardChannel;
 
 	private String discardChannelName;
@@ -62,6 +72,21 @@ public abstract class CorrelationHandlerSpec<S extends CorrelationHandlerSpec<S,
 		return _this();
 	}
 
+	public S groupTimeout(long groupTimeout) {
+		this.groupTimeoutExpression = new LiteralExpression("" + groupTimeout);
+		return _this();
+	}
+
+	public S groupTimeoutExpression(String expression) {
+		this.groupTimeoutExpression = PARSER.parseExpression(expression);
+		return _this();
+	}
+
+	public S taskScheduler(TaskScheduler taskScheduler) {
+		this.taskScheduler = taskScheduler;
+		return _this();
+	}
+
 	public S discardChannel(MessageChannel discardChannel) {
 		this.discardChannel = discardChannel;
 		return _this();
@@ -127,6 +152,10 @@ public abstract class CorrelationHandlerSpec<S extends CorrelationHandlerSpec<S,
 			handler.setMessageStore(this.messageStore);
 		}
 		handler.setMinimumTimeoutForEmptyGroups(this.minimumTimeoutForEmptyGroups);
+		handler.setGroupTimeoutExpression(this.groupTimeoutExpression);
+		if (this.taskScheduler != null) {
+			handler.setTaskScheduler(this.taskScheduler);
+		}
 		handler.setSendPartialResultOnExpiry(this.sendPartialResultOnExpiry);
 		if (this.correlationStrategy != null) {
 			handler.setCorrelationStrategy(this.correlationStrategy);

spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/EnricherSpec.java

@@ -20,9 +20,9 @@ import java.util.HashMap;
 import java.util.Map;
 
 import org.springframework.expression.Expression;
-import org.springframework.expression.common.LiteralExpression;
 import org.springframework.expression.spel.standard.SpelExpressionParser;
 import org.springframework.integration.dsl.core.IntegrationComponentSpec;
+import org.springframework.integration.expression.ValueExpression;
 import org.springframework.integration.transformer.ContentEnricher;
 import org.springframework.integration.transformer.support.AbstractHeaderValueMessageProcessor;
 import org.springframework.integration.transformer.support.ExpressionEvaluatingHeaderValueMessageProcessor;
@@ -87,8 +87,8 @@ public class EnricherSpec extends IntegrationComponentSpec<EnricherSpec, Content
 		return _this();
 	}
 
-	public EnricherSpec property(String key, String value) {
-		this.propertyExpressions.put(key, new LiteralExpression(value));
+	public <V> EnricherSpec property(String key, V value) {
+		this.propertyExpressions.put(key, new ValueExpression<V>(value));
 		return _this();
 	}
 
@@ -109,20 +109,12 @@ public class EnricherSpec extends IntegrationComponentSpec<EnricherSpec, Content
 	}
 
 	public EnricherSpec headerExpression(String name, String expression) {
-		return this.headerExpression(name, expression, null, null);
+		return this.headerExpression(name, expression, null);
 	}
 
 	public EnricherSpec headerExpression(String name, String expression, Boolean overwrite) {
-		return this.headerExpression(name, expression, overwrite, null);
-	}
-
-	public EnricherSpec headerExpression(String name, String expression, Class<?> type) {
-		return this.headerExpression(name, expression, null, type);
-	}
-
-	public <T> EnricherSpec headerExpression(String name, String expression, Boolean overwrite, Class<T> type) {
-		AbstractHeaderValueMessageProcessor<T> headerValueMessageProcessor =
-				new ExpressionEvaluatingHeaderValueMessageProcessor<T>(expression, type);
+		AbstractHeaderValueMessageProcessor<?> headerValueMessageProcessor =
+				new ExpressionEvaluatingHeaderValueMessageProcessor<Object>(expression, null);
 		headerValueMessageProcessor.setOverwrite(overwrite);
 		return this.header(name, headerValueMessageProcessor);
 	}
@@ -133,7 +125,6 @@ public class EnricherSpec extends IntegrationComponentSpec<EnricherSpec, Content
 		return _this();
 	}
 
-
 	@Override
 	protected ContentEnricher doGet() {
 		this.enricher.setPropertyExpressions(this.propertyExpressions);

spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/HeaderEnricherSpec.java

@@ -87,7 +87,7 @@ public class HeaderEnricherSpec extends IntegrationComponentSpec<HeaderEnricherS
 		return this.headerExpression(name, expression, overwrite, null);
 	}
 
-	public HeaderEnricherSpec headerExpression(String name, String expression, Class<?> type) {
+	public <T> HeaderEnricherSpec headerExpression(String name, String expression, Class<T> type) {
 		return this.headerExpression(name, expression, null, type);
 	}
 
@@ -104,7 +104,6 @@ public class HeaderEnricherSpec extends IntegrationComponentSpec<HeaderEnricherS
 		return _this();
 	}
 
-
 	@Override
 	protected HeaderEnricher doGet() {
 		return this.headerEnricher;

spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/IntegrationFlowBuilder.java

@@ -555,6 +555,19 @@ public final class IntegrationFlowBuilder {
 					" for FixedSubscriberChannel which can't be created without MessageHandler constructor argument. " +
 					"That means that '.fixedSubscriberChannel()' can't be the last EIP-method in the IntegrationFlow definition.");
 		}
+
+		if (this.flow.getIntegrationComponents().size() == 1) {
+			if (this.currentComponent != null) {
+				if (this.currentComponent instanceof SourcePollingChannelAdapterFactoryBean) {
+					throw new BeanCreationException("The 'SourcePollingChannelAdapter' (" + this.currentComponent + ") " +
+							"must be configured with at least one 'MessageChanel' or 'MessageHandler'.");
+				}
+			}
+			else if (this.currentMessageChannel != null) {
+				throw new BeanCreationException("The 'IntegrationFlow' can't consist of only one 'MessageChannel'. " +
+						"Add at lest '.bridge()' EIP-method before the end of flow.");
+			}
+		}
 		return this.flow;
 	}
 

spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/IntegrationFlows.java

@@ -62,13 +62,16 @@ public final class IntegrationFlows {
 		return from(messageSource, null);
 	}
 
-	public static IntegrationFlowBuilder from(MessageSource<?> messageSource, EndpointConfigurer<SourcePollingChannelAdapterSpec> endpointConfigurer) {
+	public static IntegrationFlowBuilder from(MessageSource<?> messageSource,
+			EndpointConfigurer<SourcePollingChannelAdapterSpec> endpointConfigurer) {
 		SourcePollingChannelAdapterSpec spec = new SourcePollingChannelAdapterSpec(messageSource);
 		if (endpointConfigurer != null) {
 			endpointConfigurer.configure(spec);
 		}
 		SourcePollingChannelAdapterFactoryBean sourcePollingChannelAdapterFactoryBean = spec.get().getT1();
-		return new IntegrationFlowBuilder().addComponent(sourcePollingChannelAdapterFactoryBean).currentComponent(sourcePollingChannelAdapterFactoryBean);
+		return new IntegrationFlowBuilder()
+				.addComponent(sourcePollingChannelAdapterFactoryBean)
+				.currentComponent(sourcePollingChannelAdapterFactoryBean);
 	}
 
 	/*public static IntegrationFlowBuilder from(AbstractEndpoint endpoint) {

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

@@ -23,6 +23,7 @@ import java.util.List;
 
 import org.springframework.integration.channel.AbstractMessageChannel;
 import org.springframework.integration.dsl.core.IntegrationComponentSpec;
+import org.springframework.messaging.converter.MessageConverter;
 import org.springframework.messaging.support.ChannelInterceptor;
 import org.springframework.util.Assert;
 
@@ -37,6 +38,8 @@ public abstract class MessageChannelSpec<S extends MessageChannelSpec<S, C>, C e
 
 	private final List<ChannelInterceptor> interceptors = new LinkedList<ChannelInterceptor>();
 
+	private MessageConverter messageConverter;
+
 	@Override
 	protected S id(String id) {
 		return super.id(id);
@@ -54,11 +57,17 @@ public abstract class MessageChannelSpec<S extends MessageChannelSpec<S, C>, C e
 		return _this();
 	}
 
+	public S messageConverter(MessageConverter messageConverter) {
+		this.messageConverter = messageConverter;
+		return _this();
+	}
+
 	@Override
 	protected C doGet() {
 		this.channel.setDatatypes(this.datatypes.toArray(new Class<?>[this.datatypes.size()]));
 		this.channel.setBeanName(this.id);
 		this.channel.setInterceptors(this.interceptors);
+		this.channel.setMessageConverter(this.messageConverter);
 		return this.channel;
 	}
 

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

@@ -19,7 +19,8 @@ package org.springframework.integration.dsl.channel;
 import java.util.concurrent.BlockingQueue;
 import java.util.concurrent.Executor;
 
-import org.springframework.integration.store.MessageGroupStore;
+import org.springframework.integration.store.ChannelMessageStore;
+import org.springframework.integration.store.PriorityCapableChannelMessageStore;
 import org.springframework.messaging.Message;
 
 /**
@@ -59,11 +60,11 @@ public final class MessageChannels {
 		return queue(capacity).id(id);
 	}
 
-	public static QueueChannelSpec.MessageStoreSpec queue(MessageGroupStore messageGroupStore, Object groupId) {
+	public static QueueChannelSpec.MessageStoreSpec queue(ChannelMessageStore messageGroupStore, Object groupId) {
 		return new QueueChannelSpec.MessageStoreSpec(messageGroupStore, groupId);
 	}
 
-	public static QueueChannelSpec.MessageStoreSpec queue(String id, MessageGroupStore messageGroupStore, Object groupId) {
+	public static QueueChannelSpec.MessageStoreSpec queue(String id, ChannelMessageStore messageGroupStore, Object groupId) {
 		return queue(messageGroupStore, groupId).id(id);
 	}
 
@@ -91,6 +92,16 @@ 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 QueueChannelSpec.MessageStoreSpec priority(String id, PriorityCapableChannelMessageStore messageGroupStore,
+			Object groupId) {
+		return queue(messageGroupStore, groupId).id(id);
+	}
+
 	public static PublishSubscribeChannelSpec publishSubscribe() {
 		return new PublishSubscribeChannelSpec();
 	}

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

@@ -20,8 +20,9 @@ import java.util.concurrent.BlockingQueue;
 import java.util.concurrent.locks.Lock;
 
 import org.springframework.integration.channel.QueueChannel;
+import org.springframework.integration.store.ChannelMessageStore;
 import org.springframework.integration.store.MessageGroupQueue;
-import org.springframework.integration.store.MessageGroupStore;
+import org.springframework.integration.store.PriorityCapableChannelMessageStore;
 import org.springframework.messaging.Message;
 
 /**
@@ -60,13 +61,13 @@ public class QueueChannelSpec extends MessageChannelSpec<QueueChannelSpec, Queue
 
 	public static class MessageStoreSpec extends QueueChannelSpec {
 
-		private final MessageGroupStore messageGroupStore;
+		private final ChannelMessageStore messageGroupStore;
 
 		private final Object groupId;
 
 		private Lock storeLock;
 
-		MessageStoreSpec(MessageGroupStore messageGroupStore, Object groupId) {
+		MessageStoreSpec(ChannelMessageStore messageGroupStore, Object groupId) {
 			super();
 			this.messageGroupStore = messageGroupStore;
 			this.groupId = groupId;
@@ -77,6 +78,7 @@ public class QueueChannelSpec extends MessageChannelSpec<QueueChannelSpec, Queue
 			return (MessageStoreSpec) super.id(id);
 		}
 
+
 		public MessageStoreSpec capacity(Integer capacity) {
 			this.capacity = capacity;
 			return this;
@@ -91,18 +93,21 @@ public class QueueChannelSpec extends MessageChannelSpec<QueueChannelSpec, Queue
 		protected QueueChannel doGet() {
 			if (this.capacity != null) {
 				if (this.storeLock != null) {
-					this.queue = new MessageGroupQueue(messageGroupStore, groupId, this.capacity, this.storeLock);
+					this.queue = new MessageGroupQueue(this.messageGroupStore, this.groupId, this.capacity, this.storeLock);
 				}
 				else {
-					this.queue = new MessageGroupQueue(messageGroupStore, groupId, this.capacity);
+					this.queue = new MessageGroupQueue(this.messageGroupStore, this.groupId, this.capacity);
 				}
 			}
 			else if (this.storeLock != null) {
-				this.queue = new MessageGroupQueue(messageGroupStore, groupId, this.storeLock);
+				this.queue = new MessageGroupQueue(this.messageGroupStore, this.groupId, this.storeLock);
 			}
 			else {
-				this.queue = new MessageGroupQueue(messageGroupStore, groupId);
+				this.queue = new MessageGroupQueue(this.messageGroupStore, this.groupId);
 			}
+
+			((MessageGroupQueue) this.queue).setPriority(this.messageGroupStore instanceof PriorityCapableChannelMessageStore);
+
 			return super.doGet();
 		}
 

spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/support/Pollers.java

@@ -37,7 +37,14 @@ public final class Pollers {
 	}
 
 	public static PollerSpec fixedRate(long period, TimeUnit timeUnit) {
-		return periodicTrigger(period, timeUnit, true);
+		return fixedRate(period, timeUnit, 0);
+	}
+	public static PollerSpec fixedRate(long period, long initialDelay) {
+		return periodicTrigger(period, null, true, initialDelay);
+	}
+
+	public static PollerSpec fixedRate(long period, TimeUnit timeUnit, long initialDelay) {
+		return periodicTrigger(period, timeUnit, true, initialDelay);
 	}
 
 	public static PollerSpec fixedDelay(long period) {
@@ -45,12 +52,21 @@ public final class Pollers {
 	}
 
 	public static PollerSpec fixedDelay(long period, TimeUnit timeUnit) {
-		return periodicTrigger(period, timeUnit, false);
+		return fixedDelay(period, timeUnit, 0);
 	}
 
-	private static PollerSpec periodicTrigger(long period, TimeUnit timeUnit, boolean fixedRate) {
+	public static PollerSpec fixedDelay(long period, long initialDelay) {
+		return periodicTrigger(period, null, false, initialDelay);
+	}
+
+	public static PollerSpec fixedDelay(long period, TimeUnit timeUnit, long initialDelay) {
+		return periodicTrigger(period, timeUnit, false, initialDelay);
+	}
+
+	private static PollerSpec periodicTrigger(long period, TimeUnit timeUnit, boolean fixedRate, long initialDelay) {
 		PeriodicTrigger periodicTrigger = new PeriodicTrigger(period, timeUnit);
 		periodicTrigger.setFixedRate(fixedRate);
+		periodicTrigger.setInitialDelay(initialDelay);
 		return new PollerSpec(periodicTrigger);
 	}
 

spring-integration-java-dsl/src/test/java/org/springframework/integration/dsl/test/IntegrationFlowTests.java

@@ -16,16 +16,10 @@
 
 package org.springframework.integration.dsl.test;
 
-import static org.junit.Assert.assertEquals;
-import static org.junit.Assert.assertFalse;
-import static org.junit.Assert.assertNotNull;
-import static org.junit.Assert.assertNull;
-import static org.junit.Assert.assertSame;
-import static org.junit.Assert.assertThat;
-import static org.junit.Assert.assertTrue;
-import static org.junit.Assert.fail;
+import static org.junit.Assert.*;
 
 import java.io.File;
+import java.io.IOException;
 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.Collection;
@@ -39,10 +33,21 @@ import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicInteger;
 import java.util.concurrent.atomic.AtomicReference;
 
+import com.mongodb.Mongo;
+import com.mongodb.MongoURI;
+import de.flapdoodle.embed.mongo.MongodExecutable;
+import de.flapdoodle.embed.mongo.MongodProcess;
+import de.flapdoodle.embed.mongo.MongodStarter;
+import de.flapdoodle.embed.mongo.config.MongodConfigBuilder;
+import de.flapdoodle.embed.mongo.config.Net;
+import de.flapdoodle.embed.mongo.distribution.Version;
+import de.flapdoodle.embed.process.runtime.Network;
 import org.aopalliance.aop.Advice;
 import org.aopalliance.intercept.MethodInterceptor;
 import org.aopalliance.intercept.MethodInvocation;
 import org.hamcrest.Matchers;
+import org.junit.AfterClass;
+import org.junit.BeforeClass;
 import org.junit.Test;
 import org.junit.runner.RunWith;
 
@@ -55,6 +60,8 @@ import org.springframework.context.annotation.AnnotationConfigApplicationContext
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.ComponentScan;
 import org.springframework.context.annotation.Configuration;
+import org.springframework.data.mongodb.MongoDbFactory;
+import org.springframework.data.mongodb.core.SimpleMongoDbFactory;
 import org.springframework.integration.IntegrationMessageHeaderAccessor;
 import org.springframework.integration.MessageDispatchingException;
 import org.springframework.integration.annotation.Header;
@@ -85,10 +92,12 @@ import org.springframework.integration.file.DefaultFileNameGenerator;
 import org.springframework.integration.file.FileHeaders;
 import org.springframework.integration.file.FileWritingMessageHandler;
 import org.springframework.integration.handler.advice.ExpressionEvaluatingRequestHandlerAdvice;
+import org.springframework.integration.mongodb.store.MongoDbChannelMessageStore;
 import org.springframework.integration.router.MethodInvokingRouter;
 import org.springframework.integration.scheduling.PollerMetadata;
 import org.springframework.integration.splitter.DefaultMessageSplitter;
 import org.springframework.integration.store.MessageStore;
+import org.springframework.integration.store.PriorityCapableChannelMessageStore;
 import org.springframework.integration.store.SimpleMessageStore;
 import org.springframework.integration.support.MessageBuilder;
 import org.springframework.integration.support.MutableMessageBuilder;
@@ -106,6 +115,7 @@ import org.springframework.messaging.core.DestinationResolutionException;
 import org.springframework.messaging.support.ChannelInterceptorAdapter;
 import org.springframework.messaging.support.GenericMessage;
 import org.springframework.stereotype.Component;
+import org.springframework.test.annotation.DirtiesContext;
 import org.springframework.test.context.ContextConfiguration;
 import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
 
@@ -114,10 +124,15 @@ import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
  */
 @ContextConfiguration
 @RunWith(SpringJUnit4ClassRunner.class)
+@DirtiesContext
 public class IntegrationFlowTests {
 
 	private static final File tmpDir = new File(System.getProperty("java.io.tmpdir"));
 
+	private static MongodExecutable mongodExe;
+
+	private static MongodProcess mongod;
+
 	@Autowired
 	private ListableBeanFactory beanFactory;
 
@@ -249,6 +264,30 @@ public class IntegrationFlowTests {
 	@Qualifier("claimCheckInput")
 	private MessageChannel claimCheckInput;
 
+	@Autowired
+	@Qualifier("priorityChannel")
+	private MessageChannel priorityChannel;
+
+	@Autowired
+	@Qualifier("priorityReplyChannel")
+	private PollableChannel priorityReplyChannel;
+
+	@BeforeClass
+	public static void setup() throws IOException {
+		mongodExe = MongodStarter.getDefaultInstance()
+				.prepare(new MongodConfigBuilder()
+						.version(Version.Main.PRODUCTION)
+						.net(new Net(12345, Network.localhostIsIPv6()))
+						.build());
+		mongod = mongodExe.start();
+	}
+
+	@AfterClass
+	public static void tearDown() {
+		mongod.stop();
+		mongodExe.stop();
+	}
+
 	@Test
 	public void testPollingFlow() {
 		assertThat(this.beanFactory.getBean("integerChannel"), Matchers.instanceOf(FixedSubscriberChannel.class));
@@ -664,6 +703,61 @@ public class IntegrationFlowTests {
 		assertSame(message, this.messageStore.getMessage(message.getHeaders().getId()));
 	}
 
+	@Test
+	public void testPriority() throws InterruptedException {
+
+		Message<String> message = MessageBuilder.withPayload("1").setPriority(1).build();
+		this.priorityChannel.send(message);
+
+		message = MessageBuilder.withPayload("-1").setPriority(-1).build();
+		this.priorityChannel.send(message);
+
+		message = MessageBuilder.withPayload("3").setPriority(3).build();
+		this.priorityChannel.send(message);
+
+		message = MessageBuilder.withPayload("0").setPriority(0).build();
+		this.priorityChannel.send(message);
+
+		message = MessageBuilder.withPayload("2").setPriority(2).build();
+		this.priorityChannel.send(message);
+
+		message = MessageBuilder.withPayload("none").build();
+		this.priorityChannel.send(message);
+
+		message = MessageBuilder.withPayload("31").setPriority(3).build();
+		this.priorityChannel.send(message);
+
+		Thread.sleep(1000);
+
+		Message<?> receive = this.priorityReplyChannel.receive(1000);
+		assertNotNull(receive);
+		assertEquals("3", receive.getPayload());
+
+		receive = this.priorityReplyChannel.receive(1000);
+		assertNotNull(receive);
+		assertEquals("31", receive.getPayload());
+
+		receive = this.priorityReplyChannel.receive(1000);
+		assertNotNull(receive);
+		assertEquals("2", receive.getPayload());
+
+		receive = this.priorityReplyChannel.receive(1000);
+		assertNotNull(receive);
+		assertEquals("1", receive.getPayload());
+
+		receive = this.priorityReplyChannel.receive(1000);
+		assertNotNull(receive);
+		assertEquals("0", receive.getPayload());
+
+		receive = this.priorityReplyChannel.receive(1000);
+		assertNotNull(receive);
+		assertEquals("-1", receive.getPayload());
+
+		receive = this.priorityReplyChannel.receive(1000);
+		assertNotNull(receive);
+		assertEquals("none", receive.getPayload());
+	}
+
 
 	@MessagingGateway(defaultRequestChannel = "controlBus")
 	private static interface ControlBusGateway {
@@ -752,6 +846,27 @@ public class IntegrationFlowTests {
 					.get();
 		}
 
+		@Bean
+		public MongoDbFactory mongoDbFactory() throws Exception {
+			return new SimpleMongoDbFactory(new MongoURI("mongodb://localhost:12345/local"));
+		}
+
+		@Bean
+		public MongoDbChannelMessageStore mongoDbChannelMessageStore(MongoDbFactory mongoDbFactory) {
+			MongoDbChannelMessageStore mongoDbChannelMessageStore = new MongoDbChannelMessageStore(mongoDbFactory);
+			mongoDbChannelMessageStore.setPriorityEnabled(true);
+			return mongoDbChannelMessageStore;
+		}
+
+		@Bean
+		public IntegrationFlow priorityFlow(PriorityCapableChannelMessageStore mongoDbChannelMessageStore) {
+			return IntegrationFlows.from(MessageChannels.priority("priorityChannel",
+					mongoDbChannelMessageStore,	"priorityGroup"))
+					.bridge(s -> s.poller(Pollers.fixedDelay(1000, 2000)))
+					.channel(MessageChannels.queue("priorityReplyChannel"))
+					.get();
+		}
+
 	}
 
 	@MessageEndpoint
@@ -851,7 +966,8 @@ public class IntegrationFlowTests {
 									r.defaultOutputChannel(defaultOutputChannel())
 											.recipient("foo-channel", "'foo' == payload")
 											.recipient("bar-channel", m ->
-													m.getHeaders().containsKey("recipient") && (boolean) m.getHeaders().get("recipient"))
+													m.getHeaders().containsKey("recipient")
+															&& (boolean) m.getHeaders().get("recipient"))
 					)
 					.get();
 		}
@@ -1175,3 +1291,4 @@ public class IntegrationFlowTests {
 	}
 
 }
+