Infrastructure for ReactiveMessageHandler (#3137)

* Infrastructure for ReactiveMessageHandler

We have now a `ReactiveMongoDbStoringMessageHandler` which implements
a `ReactiveMessageHandler`, but not a `MessageHandler` for possible
deferred subscriptions to the returned Reactor type

We don't have a proper application context processing for this
new type of message handlers

* Change a  `ConsumerEndpointFactoryBean` to apply an `MH` and `RMH`
as possible types for handler
* Introduce a `ReactiveMessageHandlerAdapter` to wrap an `RMH`
into a `MH` for synchronous calls in the regular consumer endpoints
* Wrap an `RMH` into a `ReactiveMessageHandlerAdapter` for regular
endpoints and unwrap for `ReactiveStreamsConsumer`
* Add `RMH`-based ctor into `ReactiveStreamsConsumer` for target
reactive streams composition (`flatMap()` on the `RMH`)
* Remove a `DelegatingSubscriber` from the `ReactiveStreamsConsumer`
in favor of direct calls from the `doOnSubscribe()`, `doOnComplete()`
& `doOnNext()`
* Add an `onErrorContinue()` to handle per-message errors, but don't
cancel the whole source `Publisher`
* Use `Disposable` from the `subscribe()` to cancel in the `stop()`
- recommended way in Reactor
* Use `onErrorContinue()` in the `FluxMessageChannel` instead of
`try..catch` in the `doOnNext()` - for possible `onErrorStop()`
in the provided upstream `Publisher`
* Handle `RMH` in the `ServiceActivatorFactoryBean` as a direct handler
as well with wrapping into `ReactiveMessageHandlerAdapter` for return.
The `ConsumerEndpointFactoryBean` extracts an `RMH` from the adapter
for the `ReactiveStreamsConsumer` anyway
* Add XML parsing test for `ReactiveMongoDbStoringMessageHandler`
* Add `log4j-slf4j-impl` for all the test runtime since `slf4j-api`
comes as a transitive dependency from many places

* * Fix conflicts after rebasing to master

* * Fix typo in warn message
* Change `Assert.state()` to `Assert.isTrue()`
for `ConsumerEndpointFactoryBean.setHandler()`

* * Fix `ConsumerEndpointFactoryBean` when reactive and no advice-chain
* Fix race condition in the
`ReactiveMongoDbStoringMessageHandlerTests.testReactiveMongoMessageHandlerFromApplicationContext()`

* * Handle `ReactiveMessageHandler` in Java DSL.
Essentially request a wrapping into `ReactiveMessageHandlerAdapter`.
Describe such a requirements in the `ReactiveMessageHandlerAdapter` JavaDocs
* Some Java DSL test polishing
* Add Java DSL for `ReactiveMongoDbStoringMessageHandler`
* Propagate missed `ApplicationContext` population into an internally
created `ReactiveMongoTemplate` in the `ReactiveMongoDbStoringMessageHandler`

build.gradle

@@ -237,6 +237,7 @@ configure(javaProjects) { subproject ->
 
 		testRuntimeOnly 'org.apache.logging.log4j:log4j-core'
 		testRuntimeOnly 'org.apache.logging.log4j:log4j-jcl'
+		testRuntimeOnly 'org.apache.logging.log4j:log4j-slf4j-impl'
 	}
 
 	// enable all compiler warnings; individual projects may customize further
@@ -471,7 +472,6 @@ project('spring-integration-gemfire') {
 		api "commons-io:commons-io:$commonsIoVersion"
 
 		testImplementation project(':spring-integration-stream')
-		testRuntimeOnly 'org.apache.logging.log4j:log4j-slf4j-impl'
 	}
 }
 

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2019 the original author or authors.
+ * Copyright 2002-2020 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.
@@ -78,14 +78,8 @@ public class FluxMessageChannel extends AbstractMessageChannel
 				Flux.from(publisher)
 						.delaySubscription(this.subscribedSignal.filter(Boolean::booleanValue).next())
 						.publishOn(Schedulers.boundedElastic())
-						.doOnNext((message) -> {
-							try {
-								send(message);
-							}
-							catch (Exception e) {
-								logger.warn("Error during processing event: " + message, e);
-							}
-						})
+						.doOnNext(this::send)
+						.onErrorContinue((ex, message) -> logger.warn("Error during processing event: " + message, ex))
 						.subscribe());
 	}
 

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

@@ -42,12 +42,14 @@ import org.springframework.integration.endpoint.EventDrivenConsumer;
 import org.springframework.integration.endpoint.PollingConsumer;
 import org.springframework.integration.endpoint.ReactiveStreamsConsumer;
 import org.springframework.integration.handler.AbstractReplyProducingMessageHandler;
+import org.springframework.integration.handler.ReactiveMessageHandlerAdapter;
 import org.springframework.integration.handler.advice.HandleMessageAdvice;
 import org.springframework.integration.scheduling.PollerMetadata;
 import org.springframework.integration.support.channel.ChannelResolverUtils;
 import org.springframework.messaging.MessageChannel;
 import org.springframework.messaging.MessageHandler;
 import org.springframework.messaging.PollableChannel;
+import org.springframework.messaging.ReactiveMessageHandler;
 import org.springframework.messaging.SubscribableChannel;
 import org.springframework.messaging.core.DestinationResolver;
 import org.springframework.scheduling.TaskScheduler;
@@ -112,11 +114,17 @@ public class ConsumerEndpointFactoryBean
 
 	private volatile boolean initialized;
 
-	public void setHandler(MessageHandler handler) {
-		Assert.notNull(handler, "handler must not be null");
+	public void setHandler(Object handler) {
+		Assert.isTrue(handler instanceof MessageHandler || handler instanceof ReactiveMessageHandler,
+				"'handler' must be an instance of 'MessageHandler' or 'ReactiveMessageHandler'");
 		synchronized (this.handlerMonitor) {
 			Assert.isNull(this.handler, "handler cannot be overridden");
-			this.handler = handler;
+			if (handler instanceof ReactiveMessageHandler) {
+				this.handler = new ReactiveMessageHandlerAdapter((ReactiveMessageHandler) handler);
+			}
+			else {
+				this.handler = (MessageHandler) handler;
+			}
 		}
 	}
 
@@ -210,7 +218,12 @@ public class ConsumerEndpointFactoryBean
 			}
 		}
 
-		adviceChain();
+		if (!(this.handler instanceof ReactiveMessageHandlerAdapter)) {
+			adviceChain();
+		}
+		else if (!CollectionUtils.isEmpty(this.adviceChain)) {
+			LOGGER.warn("the advice chain cannot be applied to a 'ReactiveMessageHandler'");
+		}
 		if (this.channelResolver == null) {
 			this.channelResolver = ChannelResolverUtils.getChannelResolver(this.beanFactory);
 		}
@@ -282,7 +295,13 @@ public class ConsumerEndpointFactoryBean
 				pollingConsumer(channel);
 			}
 			else {
-				this.endpoint = new ReactiveStreamsConsumer(channel, this.handler);
+				if (this.handler instanceof ReactiveMessageHandlerAdapter) {
+					this.endpoint = new ReactiveStreamsConsumer(channel,
+							((ReactiveMessageHandlerAdapter) this.handler).getDelegate());
+				}
+				else {
+					this.endpoint = new ReactiveStreamsConsumer(channel, this.handler);
+				}
 			}
 			this.endpoint.setBeanName(this.beanName);
 			this.endpoint.setBeanFactory(this.beanFactory);

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2019 the original author or authors.
+ * Copyright 2002-2020 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.
@@ -22,9 +22,11 @@ import org.springframework.expression.Expression;
 import org.springframework.integration.handler.AbstractMessageProducingHandler;
 import org.springframework.integration.handler.ExpressionEvaluatingMessageProcessor;
 import org.springframework.integration.handler.MessageProcessor;
+import org.springframework.integration.handler.ReactiveMessageHandlerAdapter;
 import org.springframework.integration.handler.ReplyProducingMessageHandlerWrapper;
 import org.springframework.integration.handler.ServiceActivatingHandler;
 import org.springframework.messaging.MessageHandler;
+import org.springframework.messaging.ReactiveMessageHandler;
 import org.springframework.util.StringUtils;
 
 /**
@@ -47,7 +49,7 @@ public class ServiceActivatorFactoryBean extends AbstractStandardMessageHandlerF
 
 	@Override
 	protected MessageHandler createMethodInvokingHandler(Object targetObject, String targetMethodName) {
-		MessageHandler handler = null;
+		MessageHandler handler;
 		handler = createDirectHandlerIfPossible(targetObject, targetMethodName);
 		if (handler == null) {
 			handler = configureHandler(
@@ -67,26 +69,30 @@ public class ServiceActivatorFactoryBean extends AbstractStandardMessageHandlerF
 	 */
 	protected MessageHandler createDirectHandlerIfPossible(final Object targetObject, String targetMethodName) {
 		MessageHandler handler = null;
-		if (targetObject instanceof MessageHandler
-				&& this.methodIsHandleMessageOrEmpty(targetMethodName)) {
+		if ((targetObject instanceof MessageHandler || targetObject instanceof ReactiveMessageHandler)
+				&& methodIsHandleMessageOrEmpty(targetMethodName)) {
 			if (targetObject instanceof AbstractMessageProducingHandler) {
 				// should never happen but just return it if it's already an AMPH
 				return (MessageHandler) targetObject;
 			}
-			/*
-			 * Return a reply-producing message handler so that we still get 'produced no reply' messages
-			 * and the super class will inject the advice chain to advise the handler method if needed.
-			 */
-			handler = new ReplyProducingMessageHandlerWrapper((MessageHandler) targetObject);
-
+			if (targetObject instanceof ReactiveMessageHandler) {
+				handler = new ReactiveMessageHandlerAdapter((ReactiveMessageHandler) targetObject);
+			}
+			else {
+				/*
+				 * Return a reply-producing message handler so that we still get 'produced no reply' messages
+				 * and the super class will inject the advice chain to advise the handler method if needed.
+				 */
+				handler = new ReplyProducingMessageHandlerWrapper((MessageHandler) targetObject);
+			}
 		}
 		return handler;
 	}
 
 	@Override
 	protected MessageHandler createExpressionEvaluatingHandler(Expression expression) {
-		ExpressionEvaluatingMessageProcessor<Object> processor = new ExpressionEvaluatingMessageProcessor<Object>(expression);
-		processor.setBeanFactory(this.getBeanFactory());
+		ExpressionEvaluatingMessageProcessor<Object> processor = new ExpressionEvaluatingMessageProcessor<>(expression);
+		processor.setBeanFactory(getBeanFactory());
 		ServiceActivatingHandler handler = new ServiceActivatingHandler(processor);
 		handler.setPrimaryExpression(expression);
 		return this.configureHandler(handler);
@@ -94,7 +100,7 @@ public class ServiceActivatorFactoryBean extends AbstractStandardMessageHandlerF
 
 	@Override
 	protected <T> MessageHandler createMessageProcessingHandler(MessageProcessor<T> processor) {
-		return this.configureHandler(new ServiceActivatingHandler(processor));
+		return configureHandler(new ServiceActivatingHandler(processor));
 	}
 
 	protected MessageHandler configureHandler(ServiceActivatingHandler handler) {
@@ -115,7 +121,6 @@ public class ServiceActivatorFactoryBean extends AbstractStandardMessageHandlerF
 	@Override
 	protected void postProcessReplyProducer(AbstractMessageProducingHandler handler) {
 		super.postProcessReplyProducer(handler);
-
 		if (this.headers != null) {
 			handler.setNotPropagatedHeaders(this.headers);
 		}

spring-integration-core/src/main/java/org/springframework/integration/endpoint/ReactiveStreamsConsumer.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2016-2019 the original author or authors.
+ * Copyright 2016-2020 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.
@@ -27,19 +27,25 @@ import org.springframework.integration.channel.ChannelUtils;
 import org.springframework.integration.channel.MessageChannelReactiveUtils;
 import org.springframework.integration.channel.NullChannel;
 import org.springframework.integration.core.MessageProducer;
+import org.springframework.integration.handler.ReactiveMessageHandlerAdapter;
 import org.springframework.integration.router.MessageRouter;
+import org.springframework.lang.Nullable;
 import org.springframework.messaging.Message;
 import org.springframework.messaging.MessageChannel;
 import org.springframework.messaging.MessageHandler;
+import org.springframework.messaging.ReactiveMessageHandler;
 import org.springframework.util.Assert;
 import org.springframework.util.ErrorHandler;
 
 import reactor.core.CoreSubscriber;
 import reactor.core.Disposable;
-import reactor.core.publisher.BaseSubscriber;
+import reactor.core.publisher.Flux;
 
 
 /**
+ * An {@link AbstractEndpoint} implementation for Reactive Streams subscription into an
+ * input channel and reactive consumption of messages from that channel.
+ *
  * @author Artem Bilan
  *
  * @since 5.0
@@ -48,17 +54,22 @@ public class ReactiveStreamsConsumer extends AbstractEndpoint implements Integra
 
 	private final MessageChannel inputChannel;
 
-	private final MessageHandler handler;
-
 	private final Publisher<Message<Object>> publisher;
 
+	private final MessageHandler handler;
+
+	@Nullable
+	private final ReactiveMessageHandler reactiveMessageHandler;
+
+	@Nullable
 	private final Subscriber<Message<?>> subscriber;
 
+	@Nullable
 	private final Lifecycle lifecycleDelegate;
 
 	private ErrorHandler errorHandler;
 
-	private volatile Subscription subscription;
+	private volatile Disposable subscription;
 
 	@SuppressWarnings("unchecked")
 	public ReactiveStreamsConsumer(MessageChannel inputChannel, MessageHandler messageHandler) {
@@ -68,10 +79,10 @@ public class ReactiveStreamsConsumer extends AbstractEndpoint implements Integra
 						: new MessageHandlerSubscriber(messageHandler));
 	}
 
-	public ReactiveStreamsConsumer(MessageChannel inputChannel, final Subscriber<Message<?>> subscriber) {
-		this.inputChannel = inputChannel;
+	public ReactiveStreamsConsumer(MessageChannel inputChannel, Subscriber<Message<?>> subscriber) {
 		Assert.notNull(inputChannel, "'inputChannel' must not be null");
 		Assert.notNull(subscriber, "'subscriber' must not be null");
+		this.inputChannel = inputChannel;
 
 		if (inputChannel instanceof NullChannel && logger.isWarnEnabled()) {
 			logger.warn("The consuming from the NullChannel does not have any effects: " +
@@ -90,6 +101,24 @@ public class ReactiveStreamsConsumer extends AbstractEndpoint implements Integra
 		else {
 			this.handler = this.subscriber::onNext;
 		}
+		this.reactiveMessageHandler = null;
+	}
+
+	/**
+	 * Instantiate an endpoint based on the provided {@link MessageChannel} and {@link ReactiveMessageHandler}.
+	 * @param inputChannel the channel to consume in reactive manner.
+	 * @param reactiveMessageHandler the {@link ReactiveMessageHandler} to process messages.
+	 * @since 5.3
+	 */
+	public ReactiveStreamsConsumer(MessageChannel inputChannel, ReactiveMessageHandler reactiveMessageHandler) {
+		Assert.notNull(inputChannel, "'inputChannel' must not be null");
+		this.inputChannel = inputChannel;
+		this.handler = new ReactiveMessageHandlerAdapter(reactiveMessageHandler);
+		this.reactiveMessageHandler = reactiveMessageHandler;
+		this.publisher = MessageChannelReactiveUtils.toPublisher(inputChannel);
+		this.subscriber = null;
+		this.lifecycleDelegate =
+				reactiveMessageHandler instanceof Lifecycle ? (Lifecycle) reactiveMessageHandler : null;
 	}
 
 	public void setErrorHandler(ErrorHandler errorHandler) {
@@ -132,55 +161,35 @@ public class ReactiveStreamsConsumer extends AbstractEndpoint implements Integra
 		if (this.lifecycleDelegate != null) {
 			this.lifecycleDelegate.start();
 		}
-		this.publisher.subscribe(new DelegatingSubscriber());
+
+		Flux<?> flux = null;
+		if (this.reactiveMessageHandler != null) {
+			flux = Flux.from(this.publisher)
+					.flatMap(this.reactiveMessageHandler::handleMessage);
+		}
+		else if (this.subscriber != null) {
+			flux = Flux.from(this.publisher)
+					.doOnSubscribe(this.subscriber::onSubscribe)
+					.doOnComplete(this.subscriber::onComplete)
+					.doOnNext(this.subscriber::onNext);
+		}
+		if (flux != null) {
+			this.subscription =
+					flux.onErrorContinue((ex, data) -> this.errorHandler.handleError(ex))
+							.subscribe();
+		}
 	}
 
 	@Override
 	protected void doStop() {
 		if (this.subscription != null) {
-			this.subscription.cancel();
+			this.subscription.dispose();
 		}
 		if (this.lifecycleDelegate != null) {
 			this.lifecycleDelegate.stop();
 		}
 	}
 
-	private final class DelegatingSubscriber extends BaseSubscriber<Message<?>> {
-
-		private final Subscriber<Message<?>> delegate = ReactiveStreamsConsumer.this.subscriber;
-
-		DelegatingSubscriber() {
-		}
-
-		@Override
-		public void hookOnSubscribe(Subscription s) {
-			ReactiveStreamsConsumer.this.subscription = s;
-			this.delegate.onSubscribe(s);
-		}
-
-		@Override
-		public void hookOnNext(Message<?> message) {
-			try {
-				this.delegate.onNext(message);
-			}
-			catch (Exception e) {
-				ReactiveStreamsConsumer.this.errorHandler.handleError(e);
-				hookOnError(e);
-			}
-		}
-
-		@Override
-		public void hookOnError(Throwable t) {
-			this.delegate.onError(t);
-		}
-
-		@Override
-		public void hookOnComplete() {
-			this.delegate.onComplete();
-		}
-
-	}
-
 	private static final class MessageHandlerSubscriber
 			implements CoreSubscriber<Message<?>>, Disposable, Lifecycle {
 

spring-integration-core/src/main/java/org/springframework/integration/handler/ReactiveMessageHandlerAdapter.java

@@ -0,0 +1,74 @@
+/*
+ * Copyright 2020 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
+ *
+ *      https://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.handler;
+
+import org.springframework.messaging.Message;
+import org.springframework.messaging.MessageHandler;
+import org.springframework.messaging.MessagingException;
+import org.springframework.messaging.ReactiveMessageHandler;
+import org.springframework.util.Assert;
+
+/**
+ * A {@link MessageHandler} implementation to adapt a {@link ReactiveMessageHandler}
+ * for synchronous invocations.
+ * A subscription to the returned reactive type from {@link ReactiveMessageHandler#handleMessage(Message)}
+ * call is done directly in the {@link #handleMessage} implementation.
+ * <p>
+ * The framework wraps a target {@link ReactiveMessageHandler} into this instance automatically
+ * for XML and Annotation configuration. For Java DSL it is recommended to wrap for generic usage
+ * ({@code .handle(MessageHandle)}) or it has to be done in the
+ * {@link org.springframework.integration.dsl.MessageHandlerSpec}
+ * implementation for protocol-specif {@link ReactiveMessageHandler}.
+ * <p>
+ * The framework unwraps a delegate {@link ReactiveMessageHandler} whenever it can compose
+ * reactive streams, e.g. {@link org.springframework.integration.endpoint.ReactiveStreamsConsumer}.
+ *
+ * @author Artem Bilan
+ *
+ * @since 5.3
+ *
+ * @see org.springframework.integration.endpoint.ReactiveStreamsConsumer
+ */
+public class ReactiveMessageHandlerAdapter implements MessageHandler {
+
+	private final ReactiveMessageHandler delegate;
+
+	/**
+	 * Instantiate based on the provided {@link ReactiveMessageHandler}.
+	 * @param reactiveMessageHandler the {@link ReactiveMessageHandler} to delegate to.
+	 */
+	public ReactiveMessageHandlerAdapter(ReactiveMessageHandler reactiveMessageHandler) {
+		Assert.notNull(reactiveMessageHandler, "'reactiveMessageHandler' must not be null");
+		this.delegate = reactiveMessageHandler;
+	}
+
+	/**
+	 * Get access to the delegate {@link ReactiveMessageHandler}.
+	 * Typically used in the framework internally in components which can compose reactive streams internally
+	 * and allow us to avoid an explicit {@code subscribe()} call.
+	 * @return the {@link ReactiveMessageHandler} this instance is delegating to.
+	 */
+	public ReactiveMessageHandler getDelegate() {
+		return this.delegate;
+	}
+
+	@Override
+	public void handleMessage(Message<?> message) throws MessagingException {
+		this.delegate.handleMessage(message).subscribe();
+	}
+
+}

spring-integration-core/src/test/java/org/springframework/integration/channel/reactive/ReactiveStreamsConsumerTests.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2016-2019 the original author or authors.
+ * Copyright 2016-2020 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.
@@ -32,7 +32,7 @@ import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.LinkedBlockingQueue;
 import java.util.concurrent.TimeUnit;
 
-import org.junit.Test;
+import org.junit.jupiter.api.Test;
 import org.mockito.ArgumentCaptor;
 import org.mockito.Mockito;
 import org.reactivestreams.Subscriber;
@@ -49,8 +49,13 @@ import org.springframework.integration.handler.MethodInvokingMessageHandler;
 import org.springframework.messaging.Message;
 import org.springframework.messaging.MessageDeliveryException;
 import org.springframework.messaging.MessageHandler;
+import org.springframework.messaging.ReactiveMessageHandler;
 import org.springframework.messaging.support.GenericMessage;
 
+import reactor.core.publisher.EmitterProcessor;
+import reactor.core.publisher.Mono;
+import reactor.test.StepVerifier;
+
 /**
  * @author Artem Bilan
  *
@@ -261,4 +266,43 @@ public class ReactiveStreamsConsumerTests {
 		assertThat(result).containsExactly(testMessage, testMessage2, testMessage2);
 	}
 
+	@Test
+	public void testReactiveStreamsConsumerFluxMessageChannelReactiveMessageHandler() {
+		FluxMessageChannel testChannel = new FluxMessageChannel();
+
+		EmitterProcessor<Message<?>> processor = EmitterProcessor.create(2, false);
+
+		ReactiveMessageHandler messageHandler =
+				m -> {
+					processor.onNext(m);
+					return Mono.empty();
+				};
+
+		ReactiveStreamsConsumer reactiveConsumer = new ReactiveStreamsConsumer(testChannel, messageHandler);
+		reactiveConsumer.setBeanFactory(mock(BeanFactory.class));
+		reactiveConsumer.afterPropertiesSet();
+		reactiveConsumer.start();
+
+		Message<?> testMessage = new GenericMessage<>("test");
+		testChannel.send(testMessage);
+
+		reactiveConsumer.stop();
+
+		assertThatExceptionOfType(MessageDeliveryException.class)
+				.isThrownBy(() -> testChannel.send(testMessage))
+				.withCauseInstanceOf(IllegalStateException.class)
+				.withMessageContaining("doesn't have subscribers to accept messages");
+
+		reactiveConsumer.start();
+
+		Message<?> testMessage2 = new GenericMessage<>("test2");
+		testChannel.send(testMessage2);
+
+		StepVerifier.create(processor)
+				.expectNext(testMessage, testMessage2)
+				.thenCancel()
+				.verify();
+	}
+
+
 }

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2016-2019 the original author or authors.
+ * Copyright 2016-2020 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.
@@ -18,7 +18,6 @@ package org.springframework.integration.dsl.flows;
 
 import static org.assertj.core.api.Assertions.assertThat;
 import static org.assertj.core.api.Assertions.assertThatExceptionOfType;
-import static org.assertj.core.api.Assertions.fail;
 
 import java.io.Serializable;
 import java.util.concurrent.CountDownLatch;
@@ -33,8 +32,7 @@ import java.util.function.Supplier;
 import org.aopalliance.aop.Advice;
 import org.aopalliance.intercept.MethodInterceptor;
 import org.aopalliance.intercept.MethodInvocation;
-import org.junit.Test;
-import org.junit.runner.RunWith;
+import org.junit.jupiter.api.Test;
 
 import org.springframework.beans.factory.BeanCreationException;
 import org.springframework.beans.factory.ListableBeanFactory;
@@ -64,8 +62,8 @@ import org.springframework.integration.dsl.Pollers;
 import org.springframework.integration.dsl.Transformers;
 import org.springframework.integration.endpoint.EventDrivenConsumer;
 import org.springframework.integration.handler.AbstractReplyProducingMessageHandler;
-import org.springframework.integration.handler.GenericHandler;
 import org.springframework.integration.handler.LoggingHandler;
+import org.springframework.integration.handler.ReactiveMessageHandlerAdapter;
 import org.springframework.integration.handler.advice.ErrorMessageSendingRecoverer;
 import org.springframework.integration.handler.advice.ExpressionEvaluatingRequestHandlerAdvice;
 import org.springframework.integration.handler.advice.RequestHandlerRetryAdvice;
@@ -93,7 +91,9 @@ import org.springframework.stereotype.Component;
 import org.springframework.stereotype.Service;
 import org.springframework.test.annotation.DirtiesContext;
 import org.springframework.test.context.ContextConfiguration;
-import org.springframework.test.context.junit4.SpringRunner;
+import org.springframework.test.context.junit.jupiter.SpringJUnitConfig;
+
+import reactor.core.publisher.Mono;
 
 /**
  * @author Artem Bilan
@@ -104,7 +104,7 @@ import org.springframework.test.context.junit4.SpringRunner;
  * @since 5.0
  */
 @ContextConfiguration(loader = NoBeansOverrideAnnotationConfigContextLoader.class)
-@RunWith(SpringRunner.class)
+@SpringJUnitConfig
 @DirtiesContext
 public class IntegrationFlowTests {
 
@@ -191,15 +191,12 @@ public class IntegrationFlowTests {
 		assertThat(this.beanFactory.containsBean("expressionFilter.handler")).isTrue();
 		QueueChannel replyChannel = new QueueChannel();
 		Message<String> message = MessageBuilder.withPayload("100").setReplyChannel(replyChannel).build();
-		try {
-			this.inputChannel.send(message);
-			fail("Expected MessageDispatchingException");
-		}
-		catch (Exception e) {
-			assertThat(e).isInstanceOf(MessageDeliveryException.class);
-			assertThat(e.getCause()).isInstanceOf(MessageDispatchingException.class);
-			assertThat(e.getMessage()).contains("Dispatcher has no subscribers");
-		}
+
+		assertThatExceptionOfType(MessageDeliveryException.class)
+				.isThrownBy(() -> this.inputChannel.send(message))
+				.withCauseInstanceOf(MessageDispatchingException.class)
+				.withMessageContaining("Dispatcher has no subscribers");
+
 		this.controlBus.send("@payloadSerializingTransformer.start()");
 
 		final AtomicBoolean used = new AtomicBoolean();
@@ -670,7 +667,7 @@ public class IntegrationFlowTests {
 		public IntegrationFlow wireTapFlow1() {
 			return IntegrationFlows.from("tappedChannel1")
 					.wireTap("tapChannel", wt -> wt.selector(m -> m.getPayload().equals("foo")))
-					.handle(loggingMessageHandler())
+					.handle(new ReactiveMessageHandlerAdapter((message) -> Mono.just(message).log().then()))
 					.get();
 		}
 
@@ -820,9 +817,10 @@ public class IntegrationFlowTests {
 		@Bean
 		public IntegrationFlow errorRecovererFlow() {
 			return IntegrationFlows.from(Function.class, (gateway) -> gateway.beanName("errorRecovererFunction"))
-					.handle((GenericHandler<?>) (p, h) -> {
-						throw new RuntimeException("intentional");
-					}, e -> e.advice(retryAdvice()))
+					.<Object>handle((p, h) -> {
+								throw new RuntimeException("intentional");
+							},
+							e -> e.advice(retryAdvice()))
 					.get();
 		}
 
@@ -903,7 +901,7 @@ public class IntegrationFlowTests {
 			return IntegrationFlows.from(Consumer.class,
 					(gateway) -> gateway.beanName("globalErrorChannelResolutionFunction"))
 					.channel(c -> c.executor(taskExecutor))
-					.handle((GenericHandler<?>) (p, h) -> {
+					.handle((p, h) -> {
 						throw new RuntimeException("intentional");
 					})
 					.get();

spring-integration-core/src/test/java/org/springframework/integration/handler/advice/ExpressionEvaluatingRequestHandlerAdviceTests.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2017-2019 the original author or authors.
+ * Copyright 2017-2020 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.
@@ -19,8 +19,7 @@ package org.springframework.integration.handler.advice;
 import static org.assertj.core.api.Assertions.assertThat;
 
 import org.aopalliance.aop.Advice;
-import org.junit.Test;
-import org.junit.runner.RunWith;
+import org.junit.jupiter.api.Test;
 
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.beans.factory.annotation.Qualifier;
@@ -28,21 +27,22 @@ import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
 import org.springframework.integration.config.EnableIntegration;
 import org.springframework.integration.dsl.IntegrationFlow;
-import org.springframework.integration.handler.GenericHandler;
 import org.springframework.integration.handler.advice.ExpressionEvaluatingRequestHandlerAdvice.MessageHandlingExpressionEvaluatingAdviceException;
 import org.springframework.integration.message.AdviceMessage;
 import org.springframework.messaging.Message;
 import org.springframework.messaging.MessageChannel;
 import org.springframework.messaging.support.ErrorMessage;
 import org.springframework.messaging.support.GenericMessage;
-import org.springframework.test.context.junit4.SpringRunner;
+import org.springframework.test.context.junit.jupiter.SpringJUnitConfig;
 
 /**
  * @author Gary Russell
+ * @author Artem Bilan
+ *
  * @since 5.0
  *
  */
-@RunWith(SpringRunner.class)
+@SpringJUnitConfig
 public class ExpressionEvaluatingRequestHandlerAdviceTests {
 
 	@Autowired
@@ -70,14 +70,16 @@ public class ExpressionEvaluatingRequestHandlerAdviceTests {
 
 		@Bean
 		public IntegrationFlow advised() {
-			return f -> f.handle((GenericHandler<String>) (payload, headers) -> {
-				if (payload.equals("good")) {
-					return null;
-				}
-				else {
-					throw new RuntimeException("some failure");
-				}
-			}, c -> c.advice(expressionAdvice()));
+			return f -> f
+					.<String>handle((payload, headers) -> {
+								if (payload.equals("good")) {
+									return null;
+								}
+								else {
+									throw new RuntimeException("some failure");
+								}
+							},
+							c -> c.advice(expressionAdvice()));
 		}
 
 		@Bean

spring-integration-mongodb/src/main/java/org/springframework/integration/mongodb/dsl/MongoDb.java

@@ -17,7 +17,9 @@
 package org.springframework.integration.mongodb.dsl;
 
 import org.springframework.data.mongodb.MongoDbFactory;
+import org.springframework.data.mongodb.ReactiveMongoDatabaseFactory;
 import org.springframework.data.mongodb.core.MongoOperations;
+import org.springframework.data.mongodb.core.ReactiveMongoOperations;
 import org.springframework.data.mongodb.core.convert.MongoConverter;
 
 /**
@@ -52,6 +54,30 @@ public final class MongoDb {
 		return new MongoDbOutboundGatewaySpec(mongoTemplate);
 	}
 
+	/**
+	 * Create a {@link ReactiveMongoDbMessageHandlerSpec} builder instance
+	 * based on the provided {@link ReactiveMongoDatabaseFactory}.
+	 * @param mongoDbFactory the {@link ReactiveMongoDatabaseFactory} to use.
+	 * @return the {@link MongoDbOutboundGatewaySpec} instance
+	 * @since 5.3
+	 */
+	public static ReactiveMongoDbMessageHandlerSpec reactiveOutboundChannelAdapter(
+			ReactiveMongoDatabaseFactory mongoDbFactory) {
+
+		return new ReactiveMongoDbMessageHandlerSpec(mongoDbFactory);
+	}
+
+	/**
+	 * Create a {@link ReactiveMongoDbMessageHandlerSpec} builder instance
+	 * based on the provided {@link ReactiveMongoOperations}.
+	 * @param mongoTemplate the {@link ReactiveMongoOperations} to use.
+	 * @return the {@link ReactiveMongoDbMessageHandlerSpec} instance
+	 * @since 5.3
+	 */
+	public static ReactiveMongoDbMessageHandlerSpec reactiveOutboundChannelAdapter(ReactiveMongoOperations mongoTemplate) {
+		return new ReactiveMongoDbMessageHandlerSpec(mongoTemplate);
+	}
+
 	private MongoDb() {
 	}
 

spring-integration-mongodb/src/main/java/org/springframework/integration/mongodb/dsl/MongoDbOutboundGatewaySpec.java

@@ -32,7 +32,7 @@ import org.springframework.messaging.Message;
 /**
  * A {@link MessageHandlerSpec} extension for the MongoDb Outbound endpoint {@link MongoDbOutboundGateway}
  *
- * @author Xavier Padr?
+ * @author Xavier Padro
  * @author Artem Bilan
  *
  * @since 5.0

spring-integration-mongodb/src/main/java/org/springframework/integration/mongodb/dsl/ReactiveMongoDbMessageHandlerSpec.java

@@ -0,0 +1,108 @@
+/*
+ * Copyright 2020 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
+ *
+ *      https://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.mongodb.dsl;
+
+import java.util.Collections;
+import java.util.Map;
+import java.util.function.Function;
+
+import org.springframework.data.mongodb.ReactiveMongoDatabaseFactory;
+import org.springframework.data.mongodb.core.ReactiveMongoOperations;
+import org.springframework.data.mongodb.core.convert.MongoConverter;
+import org.springframework.expression.Expression;
+import org.springframework.expression.common.LiteralExpression;
+import org.springframework.integration.dsl.ComponentsRegistration;
+import org.springframework.integration.dsl.MessageHandlerSpec;
+import org.springframework.integration.expression.FunctionExpression;
+import org.springframework.integration.handler.ReactiveMessageHandlerAdapter;
+import org.springframework.integration.mongodb.outbound.ReactiveMongoDbStoringMessageHandler;
+import org.springframework.messaging.Message;
+
+/**
+ * A {@link MessageHandlerSpec} extension for the Reactive MongoDb Outbound endpoint
+ * {@link ReactiveMongoDbStoringMessageHandler}.
+ *
+ * @author Artem Bilan
+ *
+ * @since 5.3
+ */
+public class ReactiveMongoDbMessageHandlerSpec
+		extends MessageHandlerSpec<ReactiveMongoDbMessageHandlerSpec, ReactiveMessageHandlerAdapter>
+		implements ComponentsRegistration {
+
+	private final ReactiveMongoDbStoringMessageHandler messageHandler;
+
+	ReactiveMongoDbMessageHandlerSpec(ReactiveMongoDatabaseFactory mongoDbFactory) {
+		this(new ReactiveMongoDbStoringMessageHandler(mongoDbFactory));
+	}
+
+	ReactiveMongoDbMessageHandlerSpec(ReactiveMongoOperations reactiveMongoOperations) {
+		this(new ReactiveMongoDbStoringMessageHandler(reactiveMongoOperations));
+	}
+
+	private ReactiveMongoDbMessageHandlerSpec(ReactiveMongoDbStoringMessageHandler messageHandler) {
+		this.messageHandler = messageHandler;
+		this.target = new ReactiveMessageHandlerAdapter(this.messageHandler);
+	}
+
+	/**
+	 * Configure a {@link MongoConverter}.
+	 * @param mongoConverter the {@link MongoConverter} to use.
+	 * @return the spec
+	 */
+	public ReactiveMongoDbMessageHandlerSpec mongoConverter(MongoConverter mongoConverter) {
+		this.messageHandler.setMongoConverter(mongoConverter);
+		return this;
+	}
+
+	/**
+	 * Configure a collection name to store data.
+	 * @param collectionName the explicit collection name to use.
+	 * @return the spec
+	 */
+	public ReactiveMongoDbMessageHandlerSpec collectionName(String collectionName) {
+		return collectionNameExpression(new LiteralExpression(collectionName));
+	}
+
+	/**
+	 * Configure a {@link Function} for evaluation a collection against request message.
+	 * @param collectionNameFunction the {@link Function} to determine a collection name at runtime.
+	 * @param <P> an expected payload type
+	 * @return the spec
+	 */
+	public <P> ReactiveMongoDbMessageHandlerSpec collectionNameFunction(
+			Function<Message<P>, String> collectionNameFunction) {
+
+		return collectionNameExpression(new FunctionExpression<>(collectionNameFunction));
+	}
+
+	/**
+	 * Configure a SpEL expression to evaluate a collection name against a request message.
+	 * @param collectionNameExpression the SpEL expression to use.
+	 * @return the spec
+	 */
+	public ReactiveMongoDbMessageHandlerSpec collectionNameExpression(Expression collectionNameExpression) {
+		this.messageHandler.setCollectionNameExpression(collectionNameExpression);
+		return this;
+	}
+
+	@Override
+	public Map<Object, String> getComponentsToRegister() {
+		return Collections.singletonMap(this.messageHandler, null);
+	}
+
+}

spring-integration-mongodb/src/main/java/org/springframework/integration/mongodb/outbound/ReactiveMongoDbStoringMessageHandler.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2019 the original author or authors.
+ * Copyright 2019-2020 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.
@@ -36,6 +36,7 @@ import reactor.core.publisher.Mono;
  * collection is identified by evaluation of the {@link #collectionNameExpression}.
  *
  * @author David Turanski
+ * @author Artme Bilan
  *
  * @since 5.3
  */
@@ -104,7 +105,9 @@ public class ReactiveMongoDbStoringMessageHandler extends AbstractReactiveMessag
 		super.onInit();
 		this.evaluationContext = ExpressionUtils.createStandardEvaluationContext(getBeanFactory());
 		if (this.mongoTemplate == null) {
-			this.mongoTemplate = new ReactiveMongoTemplate(this.mongoDbFactory, this.mongoConverter);
+			ReactiveMongoTemplate mongoTemplate = new ReactiveMongoTemplate(this.mongoDbFactory, this.mongoConverter);
+			mongoTemplate.setApplicationContext(getApplicationContext());
+			this.mongoTemplate = mongoTemplate;
 		}
 		this.initialized = true;
 	}

spring-integration-mongodb/src/test/java/org/springframework/integration/mongodb/dsl/MongoDbTests.java

@@ -17,7 +17,9 @@
 package org.springframework.integration.mongodb.dsl;
 
 import static org.assertj.core.api.Assertions.assertThat;
+import static org.awaitility.Awaitility.await;
 
+import java.time.Duration;
 import java.util.Arrays;
 import java.util.List;
 
@@ -34,8 +36,10 @@ import org.springframework.data.mongodb.MongoDbFactory;
 import org.springframework.data.mongodb.core.BulkOperations;
 import org.springframework.data.mongodb.core.MongoOperations;
 import org.springframework.data.mongodb.core.MongoTemplate;
+import org.springframework.data.mongodb.core.ReactiveMongoTemplate;
 import org.springframework.data.mongodb.core.convert.MongoConverter;
 import org.springframework.data.mongodb.core.mapping.MongoMappingContext;
+import org.springframework.data.mongodb.core.query.BasicQuery;
 import org.springframework.data.mongodb.core.query.Criteria;
 import org.springframework.data.mongodb.core.query.Query;
 import org.springframework.integration.config.EnableIntegration;
@@ -255,6 +259,25 @@ public class MongoDbTests extends MongoDbAvailableTests {
 		bulkOperations.execute();
 	}
 
+	@Autowired
+	@Qualifier("reactiveStore.input")
+	private MessageChannel reactiveStoreInput;
+
+	@Test
+	@MongoDbAvailable
+	public void testReactiveMongoDbMessageHandler() {
+		this.reactiveStoreInput.send(MessageBuilder.withPayload(createPerson("Bob")).build());
+
+		ReactiveMongoTemplate reactiveMongoTemplate = new ReactiveMongoTemplate(REACTIVE_MONGO_DATABASE_FACTORY);
+
+		await().untilAsserted(() ->
+				assertThat(
+						reactiveMongoTemplate.findOne(new BasicQuery("{'name' : 'Bob'}"), Person.class, "data")
+								.block(Duration.ofSeconds(10)))
+						.isNotNull()
+						.extracting("name", "address.state").contains("Bob", "PA"));
+	}
+
 	@Configuration
 	@EnableIntegration
 	public static class ContextConfiguration {
@@ -395,6 +418,14 @@ public class MongoDbTests extends MongoDbAvailableTests {
 					.entityClass(Person.class);
 		}
 
+		@Bean
+		public IntegrationFlow reactiveStore() {
+			return f -> f
+					.channel(MessageChannels.flux())
+					.handle(MongoDb.reactiveOutboundChannelAdapter(REACTIVE_MONGO_DATABASE_FACTORY));
+		}
+
+
 	}
 
 }

spring-integration-mongodb/src/test/java/org/springframework/integration/mongodb/outbound/MongoDbOutboundGatewayXmlTests.java

@@ -39,7 +39,9 @@ import org.springframework.test.annotation.DirtiesContext;
 import org.springframework.test.context.junit4.SpringRunner;
 
 /**
- * @author Xavier Padr?
+ * @author Xavier Padro
+ * @author Artem Bilan
+ *
  * @since 5.0
  */
 @RunWith(SpringRunner.class)
@@ -52,7 +54,7 @@ public class MongoDbOutboundGatewayXmlTests extends MongoDbAvailableTests {
 	private ApplicationContext context;
 
 	@Before
-	public void setUp() throws Exception {
+	public void setUp() {
 		MongoDbFactory mongoDbFactory = this.prepareMongoFactory();
 		MongoTemplate mongoTemplate = new MongoTemplate(mongoDbFactory);
 
@@ -63,7 +65,7 @@ public class MongoDbOutboundGatewayXmlTests extends MongoDbAvailableTests {
 	}
 
 	@After
-	public void cleanUp() throws Exception {
+	public void cleanUp() {
 		MongoDbFactory mongoDbFactory = this.prepareMongoFactory();
 		MongoTemplate mongoTemplate = new MongoTemplate(mongoDbFactory);
 
@@ -73,7 +75,7 @@ public class MongoDbOutboundGatewayXmlTests extends MongoDbAvailableTests {
 
 	@Test
 	@MongoDbAvailable
-	public void testSingleQuery() throws Exception {
+	public void testSingleQuery() {
 		EventDrivenConsumer consumer = context.getBean("gatewaySingleQuery", EventDrivenConsumer.class);
 		PollableChannel outChannel = context.getBean("out", PollableChannel.class);
 
@@ -87,7 +89,7 @@ public class MongoDbOutboundGatewayXmlTests extends MongoDbAvailableTests {
 
 	@Test
 	@MongoDbAvailable
-	public void testSingleQueryWithTemplate() throws Exception {
+	public void testSingleQueryWithTemplate() {
 		EventDrivenConsumer consumer = context.getBean("gatewayWithTemplate", EventDrivenConsumer.class);
 		PollableChannel outChannel = context.getBean("out", PollableChannel.class);
 
@@ -101,7 +103,7 @@ public class MongoDbOutboundGatewayXmlTests extends MongoDbAvailableTests {
 
 	@Test
 	@MongoDbAvailable
-	public void testSingleQueryExpression() throws Exception {
+	public void testSingleQueryExpression() {
 		EventDrivenConsumer consumer = context.getBean("gatewaySingleQueryExpression", EventDrivenConsumer.class);
 		PollableChannel outChannel = context.getBean("out", PollableChannel.class);
 
@@ -120,7 +122,7 @@ public class MongoDbOutboundGatewayXmlTests extends MongoDbAvailableTests {
 
 	@Test
 	@MongoDbAvailable
-	public void testQueryExpression() throws Exception {
+	public void testQueryExpression() {
 		EventDrivenConsumer consumer = context.getBean("gatewayQueryExpression", EventDrivenConsumer.class);
 		PollableChannel outChannel = context.getBean("out", PollableChannel.class);
 
@@ -139,7 +141,7 @@ public class MongoDbOutboundGatewayXmlTests extends MongoDbAvailableTests {
 
 	@Test
 	@MongoDbAvailable
-	public void testQueryExpressionWithLimit() throws Exception {
+	public void testQueryExpressionWithLimit() {
 		EventDrivenConsumer consumer = context.getBean("gatewayQueryExpressionLimit", EventDrivenConsumer.class);
 		PollableChannel outChannel = context.getBean("out", PollableChannel.class);
 
@@ -157,7 +159,7 @@ public class MongoDbOutboundGatewayXmlTests extends MongoDbAvailableTests {
 
 	@Test
 	@MongoDbAvailable
-	public void testCollectionCallback() throws Exception {
+	public void testCollectionCallback() {
 		EventDrivenConsumer consumer = context.getBean("gatewayCollectionCallback", EventDrivenConsumer.class);
 		PollableChannel outChannel = context.getBean("out", PollableChannel.class);
 

spring-integration-mongodb/src/test/java/org/springframework/integration/mongodb/outbound/ReactiveMongoDbStoringMessageHandlerTests-context.xml

@@ -0,0 +1,18 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<beans xmlns="http://www.springframework.org/schema/beans"
+	xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+	xmlns:int="http://www.springframework.org/schema/integration"
+	xsi:schemaLocation="http://www.springframework.org/schema/integration https://www.springframework.org/schema/integration/spring-integration.xsd
+		http://www.springframework.org/schema/beans https://www.springframework.org/schema/beans/spring-beans.xsd">
+
+	<bean id="input" class="org.springframework.integration.channel.FluxMessageChannel"/>
+
+
+	<int:service-activator input-channel="input">
+		<bean class="org.springframework.integration.mongodb.outbound.ReactiveMongoDbStoringMessageHandler">
+			<constructor-arg
+					value="#{T (org.springframework.integration.mongodb.outbound.ReactiveMongoDbStoringMessageHandlerTests).REACTIVE_MONGO_DATABASE_FACTORY}"/>
+		</bean>
+	</int:service-activator>
+
+</beans>

spring-integration-mongodb/src/test/java/org/springframework/integration/mongodb/outbound/ReactiveMongoDbStoringMessageHandlerTests.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2019 the original author or authors.
+ * Copyright 2019-2020 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.
@@ -18,6 +18,7 @@ package org.springframework.integration.mongodb.outbound;
 
 import static org.assertj.core.api.Assertions.assertThat;
 import static org.assertj.core.api.Assertions.assertThatIllegalArgumentException;
+import static org.awaitility.Awaitility.await;
 import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.spy;
 
@@ -26,8 +27,12 @@ import java.util.concurrent.atomic.AtomicBoolean;
 
 import org.junit.Before;
 import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.mockito.Answers;
 
 import org.springframework.beans.factory.BeanFactory;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.context.ApplicationContext;
 import org.springframework.data.mongodb.ReactiveMongoDatabaseFactory;
 import org.springframework.data.mongodb.core.ReactiveMongoOperations;
 import org.springframework.data.mongodb.core.ReactiveMongoTemplate;
@@ -40,6 +45,9 @@ import org.springframework.integration.mongodb.rules.MongoDbAvailable;
 import org.springframework.integration.mongodb.rules.MongoDbAvailableTests;
 import org.springframework.integration.support.MessageBuilder;
 import org.springframework.messaging.Message;
+import org.springframework.messaging.MessageChannel;
+import org.springframework.test.annotation.DirtiesContext;
+import org.springframework.test.context.junit4.SpringRunner;
 
 import reactor.core.publisher.Mono;
 
@@ -48,19 +56,25 @@ import reactor.core.publisher.Mono;
  * @author Oleg Zhurakousky
  * @author Gary Russell
  * @author David Turanski
+ * @author Artem Bilan
  *
  * @since 5.3
  */
+@RunWith(SpringRunner.class)
+@DirtiesContext
 public class ReactiveMongoDbStoringMessageHandlerTests extends MongoDbAvailableTests {
 
 	private ReactiveMongoTemplate template;
 
 	private ReactiveMongoDatabaseFactory mongoDbFactory;
 
+	@Autowired
+	private MessageChannel input;
+
 	@Before
 	public void setUp() {
-		mongoDbFactory = this.prepareReactiveMongoFactory("foo");
-		template = new ReactiveMongoTemplate(mongoDbFactory);
+		this.mongoDbFactory = prepareReactiveMongoFactory("foo");
+		this.template = new ReactiveMongoTemplate(this.mongoDbFactory);
 	}
 
 	@Test
@@ -82,6 +96,7 @@ public class ReactiveMongoDbStoringMessageHandlerTests extends MongoDbAvailableT
 	public void validateMessageHandlingWithDefaultCollection() {
 		ReactiveMongoDbStoringMessageHandler handler = new ReactiveMongoDbStoringMessageHandler(this.mongoDbFactory);
 		handler.setBeanFactory(mock(BeanFactory.class));
+		handler.setApplicationContext(mock(ApplicationContext.class, Answers.RETURNS_MOCKS));
 		handler.afterPropertiesSet();
 		Message<Person> message = MessageBuilder.withPayload(this.createPerson("Bob")).build();
 		waitFor(handler.handleMessage(message));
@@ -99,6 +114,7 @@ public class ReactiveMongoDbStoringMessageHandlerTests extends MongoDbAvailableT
 		ReactiveMongoDbStoringMessageHandler handler = new ReactiveMongoDbStoringMessageHandler(this.mongoDbFactory);
 		handler.setCollectionNameExpression(new LiteralExpression("foo"));
 		handler.setBeanFactory(mock(BeanFactory.class));
+		handler.setApplicationContext(mock(ApplicationContext.class, Answers.RETURNS_MOCKS));
 		handler.afterPropertiesSet();
 
 		Message<Person> message = MessageBuilder.withPayload(this.createPerson("Bob")).build();
@@ -119,6 +135,7 @@ public class ReactiveMongoDbStoringMessageHandlerTests extends MongoDbAvailableT
 		ReactiveMongoDbStoringMessageHandler handler = new ReactiveMongoDbStoringMessageHandler(this.mongoDbFactory);
 		handler.setCollectionNameExpression(new LiteralExpression(null));
 		handler.setBeanFactory(mock(BeanFactory.class));
+		handler.setApplicationContext(mock(ApplicationContext.class, Answers.RETURNS_MOCKS));
 		handler.afterPropertiesSet();
 
 		Message<Person> message = MessageBuilder.withPayload(createPerson("Bob")).build();
@@ -139,6 +156,7 @@ public class ReactiveMongoDbStoringMessageHandlerTests extends MongoDbAvailableT
 		converter = spy(converter);
 		handler.setMongoConverter(converter);
 		handler.setBeanFactory(mock(BeanFactory.class));
+		handler.setApplicationContext(mock(ApplicationContext.class, Answers.RETURNS_MOCKS));
 		handler.afterPropertiesSet();
 		Message<Person> message = MessageBuilder.withPayload(this.createPerson("Bob")).build();
 		waitFor(handler.handleMessage(message));
@@ -161,6 +179,7 @@ public class ReactiveMongoDbStoringMessageHandlerTests extends MongoDbAvailableT
 		ReactiveMongoDbStoringMessageHandler handler = new ReactiveMongoDbStoringMessageHandler(writingTemplate);
 		handler.setCollectionNameExpression(new LiteralExpression("foo"));
 		handler.setBeanFactory(mock(BeanFactory.class));
+		handler.setApplicationContext(mock(ApplicationContext.class, Answers.RETURNS_MOCKS));
 		handler.afterPropertiesSet();
 		Message<Person> message = MessageBuilder.withPayload(this.createPerson("Bob")).build();
 		waitFor(handler.handleMessage(message));
@@ -172,8 +191,22 @@ public class ReactiveMongoDbStoringMessageHandlerTests extends MongoDbAvailableT
 		assertThat(person.getAddress().getState()).isEqualTo("PA");
 	}
 
+	@Test
+	@MongoDbAvailable
+	public void testReactiveMongoMessageHandlerFromApplicationContext() {
+		Message<Person> message = MessageBuilder.withPayload(createPerson("Bob")).build();
+		this.input.send(message);
+
+		Query query = new BasicQuery("{'name' : 'Bob'}");
+
+		await().untilAsserted(() ->
+				assertThat(waitFor(this.template.findOne(query, Person.class, "data")))
+						.isNotNull()
+						.extracting("name", "address.state").contains("Bob", "PA"));
+	}
+
 	private static <T> T waitFor(Mono<T> mono) {
-		return mono.block(Duration.ofSeconds(3));
+		return mono.block(Duration.ofSeconds(10));
 	}
 
 }

spring-integration-mongodb/src/test/java/org/springframework/integration/mongodb/rules/MongoDbAvailableTests.java

@@ -65,16 +65,20 @@ public abstract class MongoDbAvailableTests {
 							MongoClientSettings.builder().build()),
 					"test");
 
+	public static final ReactiveMongoDatabaseFactory REACTIVE_MONGO_DATABASE_FACTORY =
+			new SimpleReactiveMongoDatabaseFactory(
+					com.mongodb.reactivestreams.client.MongoClients.create(
+							MongoClientSettings.builder().build()),
+					"test");
+
 	protected MongoDbFactory prepareMongoFactory(String... additionalCollectionsToDrop) {
 		cleanupCollections(MONGO_DATABASE_FACTORY, additionalCollectionsToDrop);
 		return MONGO_DATABASE_FACTORY;
 	}
 
 	protected ReactiveMongoDatabaseFactory prepareReactiveMongoFactory(String... additionalCollectionsToDrop) {
-		ReactiveMongoDatabaseFactory mongoDbFactory = new SimpleReactiveMongoDatabaseFactory(
-				com.mongodb.reactivestreams.client.MongoClients.create(), "test");
-		cleanupCollections(mongoDbFactory, additionalCollectionsToDrop);
-		return mongoDbFactory;
+		cleanupCollections(REACTIVE_MONGO_DATABASE_FACTORY, additionalCollectionsToDrop);
+		return REACTIVE_MONGO_DATABASE_FACTORY;
 	}
 
 	protected void cleanupCollections(ReactiveMongoDatabaseFactory mongoDbFactory,