diff --git a/spring-integration-mongodb/src/main/java/org/springframework/integration/mongodb/inbound/ReactiveMongoDbMessageSource.java b/spring-integration-mongodb/src/main/java/org/springframework/integration/mongodb/inbound/ReactiveMongoDbMessageSource.java new file mode 100644 index 0000000000..1885c11347 --- /dev/null +++ b/spring-integration-mongodb/src/main/java/org/springframework/integration/mongodb/inbound/ReactiveMongoDbMessageSource.java @@ -0,0 +1,235 @@ +/* + * 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.inbound; + +import org.reactivestreams.Publisher; + +import org.springframework.beans.BeansException; +import org.springframework.context.ApplicationContext; +import org.springframework.context.ApplicationContextAware; +import org.springframework.data.mongodb.ReactiveMongoDatabaseFactory; +import org.springframework.data.mongodb.core.ReactiveMongoOperations; +import org.springframework.data.mongodb.core.ReactiveMongoTemplate; +import org.springframework.data.mongodb.core.convert.MongoConverter; +import org.springframework.data.mongodb.core.query.BasicQuery; +import org.springframework.data.mongodb.core.query.Query; +import org.springframework.expression.Expression; +import org.springframework.expression.TypeLocator; +import org.springframework.expression.common.LiteralExpression; +import org.springframework.expression.spel.support.StandardEvaluationContext; +import org.springframework.expression.spel.support.StandardTypeLocator; +import org.springframework.integration.endpoint.AbstractMessageSource; +import org.springframework.integration.expression.ExpressionUtils; +import org.springframework.integration.mongodb.support.MongoHeaders; +import org.springframework.util.Assert; + +import com.mongodb.DBObject; +import reactor.core.publisher.Flux; +import reactor.core.publisher.Mono; + +/** + * An instance of {@link org.springframework.integration.core.MessageSource} which returns + * a {@link org.springframework.messaging.Message} with a payload which is the result of + * execution of a {@link Query}. When {@code expectSingleResult} is false (default), the MongoDb + * {@link Query} is executed using {@link ReactiveMongoOperations#find(Query, Class)} method which + * returns a {@link Flux}. The returned {@link Flux} will be used as the payload of the + * {@link org.springframework.messaging.Message} returned by the {@link #receive()} + * method. + *

+ * When {@code expectSingleResult} is true, the {@link ReactiveMongoOperations#findOne(Query, Class)} is + * used instead, and the message payload will be a {@link Mono} for the single object returned from the + * query. + * + * @author David Turanski + * + * @since 5.3 + */ +public class ReactiveMongoDbMessageSource extends AbstractMessageSource> + implements ApplicationContextAware { + + private final Expression queryExpression; + + private Expression collectionNameExpression = new LiteralExpression("data"); + + private StandardEvaluationContext evaluationContext; + + private ReactiveMongoOperations reactiveMongoTemplate; + + private MongoConverter mongoConverter; + + private ReactiveMongoDatabaseFactory reactiveMongoDatabaseFactory; + + private Class entityClass = DBObject.class; + + private boolean expectSingleResult = false; + + private ApplicationContext applicationContext; + + private volatile boolean initialized = false; + + /** + * Create an instance with the provided {@link ReactiveMongoDatabaseFactory} and SpEL expression + * which should resolve to a MongoDb 'query' string (see https://www.mongodb.org/display/DOCS/Querying). + * The 'queryExpression' will be evaluated on every call to the {@link #receive()} method. + * @param reactiveMongoDatabaseFactory The reactiveMongoDatabaseFactory factory. + * @param queryExpression The query expression. + */ + public ReactiveMongoDbMessageSource(ReactiveMongoDatabaseFactory reactiveMongoDatabaseFactory, + Expression queryExpression) { + + Assert.notNull(reactiveMongoDatabaseFactory, "'reactiveMongoDatabaseFactory' must not be null"); + Assert.notNull(queryExpression, "'queryExpression' must not be null"); + + this.reactiveMongoDatabaseFactory = reactiveMongoDatabaseFactory; + this.queryExpression = queryExpression; + } + + /** + * Create an instance with the provided {@link ReactiveMongoOperations} and SpEL expression + * which should resolve to a Mongo 'query' string (see https://www.mongodb.org/display/DOCS/Querying). + * It assumes that the {@link ReactiveMongoOperations} is fully initialized and ready to be used. + * The 'queryExpression' will be evaluated on every call to the {@link #receive()} method. + * @param reactiveMongoTemplate The reactive Mongo template. + * @param queryExpression The query expression. + */ + public ReactiveMongoDbMessageSource(ReactiveMongoOperations reactiveMongoTemplate, Expression queryExpression) { + Assert.notNull(reactiveMongoTemplate, "'reactiveMongoTemplate' must not be null"); + Assert.notNull(queryExpression, "'queryExpression' must not be null"); + this.reactiveMongoTemplate = reactiveMongoTemplate; + this.queryExpression = queryExpression; + } + + /** + * Allow you to set the type of the entityClass that will be passed to the + * {@link ReactiveMongoTemplate#find(Query, Class)} or {@link ReactiveMongoTemplate#findOne(Query, Class)} + * method. + * Default is {@link DBObject}. + * @param entityClass The entity class. + */ + public void setEntityClass(Class entityClass) { + Assert.notNull(entityClass, "'entityClass' must not be null"); + this.entityClass = entityClass; + } + + /** + * Allow you to manage which find* method to invoke on {@link ReactiveMongoTemplate}. + * Default is 'false', which means the {@link #receive()} method will use + * the {@link ReactiveMongoTemplate#find(Query, Class)} method. If set to 'true', + * {@link #receive()} will use {@link ReactiveMongoTemplate#findOne(Query, Class)}, + * and the payload of the returned {@link org.springframework.messaging.Message} + * will be the returned target Object of type + * identified by {@link #entityClass} instead of a List. + * @param expectSingleResult true if a single result is expected. + */ + public void setExpectSingleResult(boolean expectSingleResult) { + this.expectSingleResult = expectSingleResult; + } + + /** + * Set the SpEL {@link Expression} that should resolve to a collection name + * used by the {@link Query}. The resulting collection name will be included + * in the {@link MongoHeaders#COLLECTION_NAME} header. + * @param collectionNameExpression The collection name expression. + */ + public void setCollectionNameExpression(Expression collectionNameExpression) { + Assert.notNull(collectionNameExpression, "'collectionNameExpression' must not be null"); + this.collectionNameExpression = collectionNameExpression; + } + + /** + * Allow you to provide a custom {@link MongoConverter} used to assist in deserialization + * data read from MongoDb. Only allowed if this instance was constructed with a + * {@link ReactiveMongoDatabaseFactory}. + * @param mongoConverter The mongo converter. + */ + public void setMongoConverter(MongoConverter mongoConverter) { + Assert.isNull(this.reactiveMongoTemplate, + "'mongoConverter' can not be set when instance was constructed with ReactiveMongoTemplate"); + this.mongoConverter = mongoConverter; + } + + @Override + public void setApplicationContext(ApplicationContext applicationContext) throws BeansException { + this.applicationContext = applicationContext; + } + + @Override + public String getComponentType() { + return "mongo:reactive-inbound-channel-adapter"; + } + + @Override + protected void onInit() { + this.evaluationContext = ExpressionUtils.createStandardEvaluationContext(getBeanFactory()); + TypeLocator typeLocator = this.evaluationContext.getTypeLocator(); + if (typeLocator instanceof StandardTypeLocator) { + //Register MongoDB query API package so FQCN can be avoided in query-expression. + ((StandardTypeLocator) typeLocator).registerImport("org.springframework.data.mongodb.core.query"); + } + if (this.reactiveMongoTemplate == null) { + ReactiveMongoTemplate template = + new ReactiveMongoTemplate(this.reactiveMongoDatabaseFactory, this.mongoConverter); + if (this.applicationContext != null) { + template.setApplicationContext(this.applicationContext); + } + this.reactiveMongoTemplate = template; + } + this.initialized = true; + } + + /** + * Execute a {@link Query} returning its results as the Message payload. + * The payload can be either {@link Flux} or {@link Mono} of objects of type + * identified by {@link #entityClass}, or a single element of type identified by {@link #entityClass} + * based on the value of {@link #expectSingleResult} attribute which defaults to 'false' resulting + * {@link org.springframework.messaging.Message} with payload of type + * {@link Flux}. The collection name used in the + * query will be provided in the {@link MongoHeaders#COLLECTION_NAME} header. + */ + @Override + public Object doReceive() { + Assert.isTrue(this.initialized, "This class is not yet initialized. Invoke its afterPropertiesSet() method"); + Object value = this.queryExpression.getValue(this.evaluationContext); + Assert.notNull(value, "'queryExpression' must not evaluate to null"); + Query query = null; + if (value instanceof String) { + query = new BasicQuery((String) value); + } + else if (value instanceof Query) { + query = ((Query) value); + } + else { + throw new IllegalStateException("'queryExpression' must evaluate to String " + + "or org.springframework.data.mongodb.core.query.Query, but not: " + query); + } + + String collectionName = this.collectionNameExpression.getValue(this.evaluationContext, String.class); + Assert.notNull(collectionName, "'collectionNameExpression' must not evaluate to null"); + + Object result; + if (this.expectSingleResult) { + result = this.reactiveMongoTemplate.findOne(query, this.entityClass, collectionName); + } + else { + result = this.reactiveMongoTemplate.find(query, this.entityClass, collectionName); + } + return getMessageBuilderFactory() + .withPayload(result) + .setHeader(MongoHeaders.COLLECTION_NAME, collectionName); + } + +} diff --git a/spring-integration-mongodb/src/test/java/org/springframework/integration/mongodb/inbound/ReactiveMongoDbMessageSourceTests.java b/spring-integration-mongodb/src/test/java/org/springframework/integration/mongodb/inbound/ReactiveMongoDbMessageSourceTests.java new file mode 100644 index 0000000000..49564fdc5c --- /dev/null +++ b/spring-integration-mongodb/src/test/java/org/springframework/integration/mongodb/inbound/ReactiveMongoDbMessageSourceTests.java @@ -0,0 +1,214 @@ +/* + * 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.inbound; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatIllegalArgumentException; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.spy; +import static org.mockito.Mockito.times; +import static org.mockito.Mockito.verify; + +import java.time.Duration; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.List; +import java.util.Optional; + +import org.bson.conversions.Bson; +import org.junit.Test; +import org.mockito.Mockito; + +import org.springframework.beans.factory.BeanFactory; +import org.springframework.data.mongodb.ReactiveMongoDatabaseFactory; +import org.springframework.data.mongodb.core.ReactiveMongoTemplate; +import org.springframework.data.mongodb.core.convert.MappingMongoConverter; +import org.springframework.data.mongodb.core.mapping.MongoMappingContext; +import org.springframework.expression.Expression; +import org.springframework.expression.common.LiteralExpression; +import org.springframework.integration.mongodb.rules.MongoDbAvailable; +import org.springframework.integration.mongodb.rules.MongoDbAvailableTests; + +import com.mongodb.BasicDBObject; +import reactor.core.publisher.Flux; +import reactor.core.publisher.Mono; +import reactor.test.StepVerifier; + +/** + * @author David Turanski + * + * @since 5.3 + */ +public class ReactiveMongoDbMessageSourceTests extends MongoDbAvailableTests { + + @Test + public void withNullMongoDBFactory() { + assertThatIllegalArgumentException() + .isThrownBy(() -> new ReactiveMongoDbMessageSource((ReactiveMongoDatabaseFactory) null, + mock(Expression.class))); + } + + @Test + public void withNullMongoTemplate() { + assertThatIllegalArgumentException() + .isThrownBy(() -> new ReactiveMongoDbMessageSource((ReactiveMongoTemplate) null, + mock(Expression.class))); + } + + @Test + public void withNullQueryExpression() { + assertThatIllegalArgumentException() + .isThrownBy(() -> new ReactiveMongoDbMessageSource(mock(ReactiveMongoDatabaseFactory.class), + null)); + } + + @Test + @MongoDbAvailable + @SuppressWarnings("unchecked") + public void validateSuccessfulQueryWithSingleElementFluxOfDbObject() { + ReactiveMongoDatabaseFactory reactiveMongoDatabaseFactory = this.prepareReactiveMongoFactory(); + + ReactiveMongoTemplate template = new ReactiveMongoTemplate(reactiveMongoDatabaseFactory); + waitFor(template.save(this.createPerson(), "data")); + + Expression queryExpression = new LiteralExpression("{'name' : 'Oleg'}"); + ReactiveMongoDbMessageSource messageSource = new ReactiveMongoDbMessageSource(reactiveMongoDatabaseFactory, + queryExpression); + messageSource.setBeanFactory(mock(BeanFactory.class)); + messageSource.afterPropertiesSet(); + + StepVerifier.create((Flux) messageSource.receive().getPayload()) + .assertNext(basicDBObject -> assertThat(basicDBObject.get("name")).isEqualTo("Oleg")) + .verifyComplete(); + } + + @Test + @MongoDbAvailable + @SuppressWarnings("unchecked") + public void validateSuccessfulQueryWithSingleElementFluxOfPerson() { + ReactiveMongoDatabaseFactory reactiveMongoDatabaseFactory = this.prepareReactiveMongoFactory(); + + ReactiveMongoTemplate template = new ReactiveMongoTemplate(reactiveMongoDatabaseFactory); + waitFor(template.save(this.createPerson(), "data")); + + Expression queryExpression = new LiteralExpression("{'name' : 'Oleg'}"); + ReactiveMongoDbMessageSource messageSource = new ReactiveMongoDbMessageSource(reactiveMongoDatabaseFactory, + queryExpression); + messageSource.setBeanFactory(mock(BeanFactory.class)); + messageSource.afterPropertiesSet(); + messageSource.setEntityClass(Person.class); + + StepVerifier.create((Flux) messageSource.receive().getPayload()) + .assertNext(person -> assertThat(person.getName()).isEqualTo("Oleg")) + .verifyComplete(); + } + + @Test + @MongoDbAvailable + public void validateSuccessfulQueryWithMultipleElements() { + final List names = new ArrayList<>(Arrays.asList("Manny", "Moe", "Jack")); + StepVerifier.create(queryMultipleElements(new LiteralExpression("{'address.state' : 'PA'}"))) + .expectNextMatches(person -> { + names.remove(person.getName()); + return names.size() == 2; + }) + .expectNextMatches(person -> { + names.remove(person.getName()); + return names.size() == 1; + }) + .expectNextMatches(person -> { + names.remove(person.getName()); + return names.size() == 0; + }) + .verifyComplete(); + } + + @Test + @MongoDbAvailable + public void validateSuccessfulQueryWithEmptyReturn() { + StepVerifier.create(queryMultipleElements(new LiteralExpression("{'address.state' : 'NJ'}"))) + .verifyComplete(); + } + + @Test + @MongoDbAvailable + @SuppressWarnings("unchecked") + public void validateSuccessfulQueryWithCustomConverter() { + MappingMongoConverter converter = new ReactiveTestMongoConverter(this.prepareReactiveMongoFactory(), + new MongoMappingContext()); + converter.afterPropertiesSet(); + converter = spy(converter); + StepVerifier.create(queryMultipleElements(new LiteralExpression("{'address.state' : 'PA'}"), + Optional.of(converter))).expectNextCount(3) + .verifyComplete(); + + verify(converter, times(3)).read((Class) Mockito.any(), Mockito.any(Bson.class)); + } + + @Test + @MongoDbAvailable + @SuppressWarnings("unchecked") + public void validatePipelineInModifyOut() { + + ReactiveMongoDatabaseFactory reactiveMongoDatabaseFactory = this.prepareReactiveMongoFactory(); + + ReactiveMongoTemplate template = new ReactiveMongoTemplate(reactiveMongoDatabaseFactory); + + waitFor(template.save(BasicDBObject.parse("{'name' : 'Manny', 'id' : 1}"), "data")); + + Expression queryExpression = new LiteralExpression("{'name' : 'Manny'}"); + ReactiveMongoDbMessageSource messageSource = new ReactiveMongoDbMessageSource(reactiveMongoDatabaseFactory, + queryExpression); + messageSource.setExpectSingleResult(true); + messageSource.setBeanFactory(mock(BeanFactory.class)); + messageSource.afterPropertiesSet(); + BasicDBObject result = waitFor((Mono) messageSource.receive().getPayload()); + Object id = result.get("_id"); + result.put("company", "PepBoys"); + waitFor(template.save(result, "data")); + result = waitFor((Mono) messageSource.receive().getPayload()); + assertThat(result.get("_id")).isEqualTo(id); + } + + private Flux queryMultipleElements(Expression queryExpression) { + return this.queryMultipleElements(queryExpression, Optional.empty()); + } + + @SuppressWarnings("unchecked") + private Flux queryMultipleElements(Expression queryExpression, Optional converter) { + ReactiveMongoDatabaseFactory reactiveMongoDatabaseFactory = this.prepareReactiveMongoFactory(); + + ReactiveMongoTemplate template = new ReactiveMongoTemplate(reactiveMongoDatabaseFactory); + waitFor(template.save(this.createPerson("Manny"), "data")); + waitFor(template.save(this.createPerson("Moe"), "data")); + waitFor(template.save(this.createPerson("Jack"), "data")); + + ReactiveMongoDbMessageSource messageSource = new ReactiveMongoDbMessageSource(reactiveMongoDatabaseFactory, + queryExpression); + messageSource.setBeanFactory(mock(BeanFactory.class)); + messageSource.setEntityClass(Person.class); + converter.ifPresent(messageSource::setMongoConverter); + messageSource.afterPropertiesSet(); + + return (Flux) messageSource.receive().getPayload(); + } + + private static T waitFor(Mono mono) { + return mono.block(Duration.ofSeconds(10)); + } + +} diff --git a/spring-integration-mongodb/src/test/java/org/springframework/integration/mongodb/rules/MongoDbAvailableTests.java b/spring-integration-mongodb/src/test/java/org/springframework/integration/mongodb/rules/MongoDbAvailableTests.java index bd143a2655..8478e44eb1 100644 --- a/spring-integration-mongodb/src/test/java/org/springframework/integration/mongodb/rules/MongoDbAvailableTests.java +++ b/spring-integration-mongodb/src/test/java/org/springframework/integration/mongodb/rules/MongoDbAvailableTests.java @@ -38,7 +38,6 @@ import org.springframework.data.mongodb.core.mapping.MongoPersistentProperty; import org.springframework.integration.mongodb.outbound.MessageCollectionCallback; import org.springframework.messaging.Message; -import com.mongodb.MongoClientSettings; import com.mongodb.MongoException; import com.mongodb.client.MongoClients; import com.mongodb.client.MongoCollection; @@ -60,16 +59,11 @@ public abstract class MongoDbAvailableTests { public MongoDbAvailableRule mongoDbAvailableRule = new MongoDbAvailableRule(); public static final MongoDbFactory MONGO_DATABASE_FACTORY = - new SimpleMongoClientDbFactory( - MongoClients.create( - MongoClientSettings.builder().build()), - "test"); + new SimpleMongoClientDbFactory(MongoClients.create(), "test"); public static final ReactiveMongoDatabaseFactory REACTIVE_MONGO_DATABASE_FACTORY = new SimpleReactiveMongoDatabaseFactory( - com.mongodb.reactivestreams.client.MongoClients.create( - MongoClientSettings.builder().build()), - "test"); + com.mongodb.reactivestreams.client.MongoClients.create(), "test"); protected MongoDbFactory prepareMongoFactory(String... additionalCollectionsToDrop) { cleanupCollections(MONGO_DATABASE_FACTORY, additionalCollectionsToDrop); diff --git a/src/reference/asciidoc/mongodb.adoc b/src/reference/asciidoc/mongodb.adoc index 9c909b4d4c..e47a37e3ff 100644 --- a/src/reference/asciidoc/mongodb.adoc +++ b/src/reference/asciidoc/mongodb.adoc @@ -28,9 +28,32 @@ To download, install, and run MongoDB, see the https://www.mongodb.org/downloads [[mongodb-connection]] === Connecting to MongoDb +==== Blocking or Reactive? + +Beginning with version 5.3, Spring Integration provides support for reactive MongoDB drivers to enable non-blocking I/O when accessing MongoDB. +To enable reactive support, add the MongoDB reactive streams driver to your dependencies: + +.Maven +[source, xml, subs="normal"] +---- + + org.mongodb + mongodb-driver-reactivestreams + 1.12.0 + +---- + +.Gradle +[source, groovy, subs="normal"] +---- +compile "org.mongodb:mongodb-driver-reactivestreams:1.12.0" +---- + To begin interacting with MongoDB, you first need to connect to it. Spring Integration builds on the support provided by another Spring project, https://projects.spring.io/spring-data-mongodb/[Spring Data MongoDB]. -It provides a factory class called `MongoDbFactory`, which simplifies integration with the MongoDB Client API. +It provides factory classes called `MongoDbFactory` and `ReactiveMongoDatabaseFactory`, which simplify integration with the MongoDB Client API. + +TIP: Spring Data provides provides the blocking MongoDB driver by default but you may opt-in for reactive usage by including the above dependency. ==== Using `MongoDbFactory` @@ -68,7 +91,7 @@ The following example shows how to use `SimpleMongoDbFactory`, the out-of-the-bo ==== [source,java] ---- -MongoDbFactory mongoDbFactory = new SimpleMongoDbFactory(new Mongo(), "test"); +MongoDbFactory mongoDbFactory = new SimpleMongoDbFactory(com.mongodb.client.MongoClients.create(), "test"); ---- ==== @@ -79,7 +102,7 @@ The following example shows how to use `SimpleMongoDbFactory` in XML configurati ---- - + @@ -87,9 +110,36 @@ The following example shows how to use `SimpleMongoDbFactory` in XML configurati ==== `SimpleMongoDbFactory` takes two arguments: a `Mongo` instance and a `String` that specifies the name of the database. -If you need to configure properties such as `host`, `port`, and others, you can pass those by using one of the constructors provided by the underlying `Mongo` class. +If you need to configure properties such as `host`, `port`, and others, you can pass those by using one of the constructors provided by the underlying `MongoClients` class. For more information on how to configure MongoDB, see the https://docs.spring.io/spring-data/data-mongo/docs/current/reference/html/[Spring-Data-MongoDB] reference. +==== Using `ReactiveMongoDatabaseFactory` + +To connect to MongoDB with the reactive driver, you can use an implementation of the `ReactiveMongoDatabaseFactory` interface. + +The following example shows how to use `SimpleReactiveMongoDatabaseFactory`, the out-of-the-box implementation, in Java: + +==== +[source,java] +---- +new SimpleReactiveMongoDatabaseFactory(com.mongodb.reactivestreams.client.MongoClients.create(), "test"); +---- +==== + +The following example shows how to use `SimpleReactiveMongoDatabaseFactory` in XML configuration: + +==== +[source,xml] +---- + + + + + + +---- +==== + [[mongodb-message-store]] === MongoDB Message Store @@ -471,9 +521,10 @@ private MongoDbOutboundGatewaySpec collectionCallbackOutboundGateway() { [[mongodb-reactive-channel-adapters]] === MongoDB Reactive Channel Adapters -Starting with version 5.3, the `ReactiveMongoDbStoringMessageHandler` implementation is provided. -It is based on the `ReactiveMongoOperations` from Spring Data and requires a `org.mongodb:mongodb-driver-reactivestreams` dependency. -This is an implementation of the `ReactiveMessageHandler` which is supported natively in the framework when reactive streams composition is involved in the integration flow definition. +Starting with version 5.3, the `ReactiveMongoDbStoringMessageHandler` and `ReactiveMongoDbMessageSource` implementations are provided. +They are based on the `ReactiveMongoOperations` from Spring Data and requires a `org.mongodb:mongodb-driver-reactivestreams` dependency. + +The `ReactiveMongoDbStoringMessageHandler` is an implementation of the `ReactiveMessageHandler` which is supported natively in the framework when reactive streams composition is involved in the integration flow definition. See more information in the <<./reactive-streams.adoc/reactive-message-handler,ReactiveMessageHandler>>. From configuration perspective there is no difference with many other standard channel adapters.