INT-4568: Add ReactiveMongoDBMessageSource
JIRA: https://jira.spring.io/browse/INT-4568 Clean up per review and some updates to `mongodb.adoc`
This commit is contained in:
David Turanski
committed by
Artem Bilan
Artem Bilan
parent
4b051684ef
commit
263995fb90
@@ -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.
|
||||
* <p>
|
||||
* 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<Publisher<?>>
|
||||
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);
|
||||
}
|
||||
|
||||
}
|
||||
@@ -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<BasicDBObject>) 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<Person>) messageSource.receive().getPayload())
|
||||
.assertNext(person -> assertThat(person.getName()).isEqualTo("Oleg"))
|
||||
.verifyComplete();
|
||||
}
|
||||
|
||||
@Test
|
||||
@MongoDbAvailable
|
||||
public void validateSuccessfulQueryWithMultipleElements() {
|
||||
final List<String> 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<Person>) 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<BasicDBObject>) messageSource.receive().getPayload());
|
||||
Object id = result.get("_id");
|
||||
result.put("company", "PepBoys");
|
||||
waitFor(template.save(result, "data"));
|
||||
result = waitFor((Mono<BasicDBObject>) messageSource.receive().getPayload());
|
||||
assertThat(result.get("_id")).isEqualTo(id);
|
||||
}
|
||||
|
||||
private Flux<Person> queryMultipleElements(Expression queryExpression) {
|
||||
return this.queryMultipleElements(queryExpression, Optional.empty());
|
||||
}
|
||||
|
||||
@SuppressWarnings("unchecked")
|
||||
private Flux<Person> queryMultipleElements(Expression queryExpression, Optional<MappingMongoConverter> 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<Person>) messageSource.receive().getPayload();
|
||||
}
|
||||
|
||||
private static <T> T waitFor(Mono<T> mono) {
|
||||
return mono.block(Duration.ofSeconds(10));
|
||||
}
|
||||
|
||||
}
|
||||
@@ -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);
|
||||
|
||||
@@ -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"]
|
||||
----
|
||||
<dependency>
|
||||
<groupId>org.mongodb</groupId>
|
||||
<artifactId>mongodb-driver-reactivestreams</artifactId>
|
||||
<version>1.12.0</version>
|
||||
</dependency>
|
||||
----
|
||||
|
||||
.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
|
||||
----
|
||||
<bean id="mongoDbFactory" class="o.s.data.mongodb.core.SimpleMongoDbFactory">
|
||||
<constructor-arg>
|
||||
<bean class="com.mongodb.Mongo"/>
|
||||
<bean class="com.mongodb.client.MongoClients" factory-method="create"/>
|
||||
</constructor-arg>
|
||||
<constructor-arg value="test"/>
|
||||
</bean>
|
||||
@@ -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]
|
||||
----
|
||||
<bean id="mongoDbFactory" class="o.s.data.mongodb.core.SimpleReactiveMongoDatabaseFactory">
|
||||
<constructor-arg>
|
||||
<bean class="com.mongodb.reactivestreams.client.MongoClients" factory-method="create"/>
|
||||
</constructor-arg>
|
||||
<constructor-arg value="test"/>
|
||||
</bean>
|
||||
----
|
||||
====
|
||||
|
||||
[[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.
|
||||
|
||||
Reference in New Issue
Block a user