Spring/spring-integration
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Add MongoDbMessageSource UPDATE option (#3493)

Browse Source 
* Add MongoDbMessageSource UPDATE option

* Extract `AbstractMongoDbMessageSource` with common options and methods
for both `MongoDbMessageSource` and `ReactiveMongoDbMessageSource`
* Add an `updateExpression` option into MongoDb source implementations
* Implement respective `update` logic after fetching the data from the collection
* Cover both reactive and blocking updates with tests
* Add `MongoDbMessageSourceSpec` into Java DSL for MongoDb channel adapters
* Expose an `update` XML attribute for the `<int-mongo:inbound-channel-adapter>`
* Upgrade MongoDb driver for latest Spring Data compatibility
* Document a new feature
* Upgrade `mongodb.adoc` for code block switch whenever it is appropriate

* * Add a Kotlin sample for `MongoDb.outboundGateway()` DSL

* Apply suggestions from code review

Co-authored-by: Gary Russell <grussell@vmware.com>

Co-authored-by: Gary Russell <grussell@vmware.com>
This commit is contained in:
Artem Bilan
2021-02-16 16:46:45 -05:00
committed by GitHub
parent b4290dd2a5
commit b1cb9069fe
17 changed files with 846 additions and 455 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
build.gradle
View File

@@ -85,7 +85,7 @@ ext {
mailVersion = '1.6.5'
micrometerVersion = '1.6.3'
mockitoVersion = '3.7.0'
mongoDriverVersion = '4.1.1'
mongoDriverVersion = '4.2.0'
mysqlVersion = '8.0.22'
pahoMqttClientVersion = '1.2.5'
postgresVersion = '42.2.18'

12
spring-integration-mongodb/src/main/java/org/springframework/integration/mongodb/config/MongoDbInboundChannelAdapterParser.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2019 the original author or authors.
* Copyright 2002-2021 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.
@@ -25,11 +25,14 @@ import org.springframework.beans.factory.xml.ParserContext;
import org.springframework.integration.config.xml.AbstractPollingInboundChannelAdapterParser;
import org.springframework.integration.config.xml.IntegrationNamespaceUtils;
import org.springframework.integration.mongodb.inbound.MongoDbMessageSource;
/**
* Parser for Mongodb store inbound adapters
* Parser for MongoDb store inbound adapters
*
* @author Amol Nayak
* @author Oleg Zhurakousky
* @author Artem Bilan
*
* @since 2.2
*/
public class MongoDbInboundChannelAdapterParser extends AbstractPollingInboundChannelAdapterParser {
@@ -48,6 +51,11 @@ public class MongoDbInboundChannelAdapterParser extends AbstractPollingInboundCh
builder.addConstructorArgValue(queryExpressionDef);
BeanDefinition expressionDef =
IntegrationNamespaceUtils.createExpressionDefinitionFromValueOrExpression("update", "update-expression",
parserContext, element, false);
builder.addPropertyValue("updateExpression", expressionDef);
IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "entity-class");
IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "expect-single-result");

151
spring-integration-mongodb/src/main/java/org/springframework/integration/mongodb/dsl/AbstractMongoDbMessageSourceSpec.java Normal file
View File

@@ -0,0 +1,151 @@
/*
* Copyright 2021 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.function.Supplier;
import org.springframework.data.mongodb.core.convert.MongoConverter;
import org.springframework.data.mongodb.core.query.Update;
import org.springframework.expression.Expression;
import org.springframework.expression.common.LiteralExpression;
import org.springframework.integration.dsl.MessageSourceSpec;
import org.springframework.integration.expression.SupplierExpression;
import org.springframework.integration.expression.ValueExpression;
import org.springframework.integration.mongodb.inbound.AbstractMongoDbMessageSource;
/**
* A {@link MessageSourceSpec} extension for common MongoDB sources options.
*
* @author Artem Bilan
*
* @since 5.5
*/
public class AbstractMongoDbMessageSourceSpec<S extends AbstractMongoDbMessageSourceSpec<S, H>,
H extends AbstractMongoDbMessageSource<?>>
extends MessageSourceSpec<S, H> {
/**
* Allow you to set the type of the entityClass that will be passed to the the MongoDB query method.
* Default is {@link com.mongodb.DBObject}.
* @param entityClass The entity class.
* @return the spec
* @see AbstractMongoDbMessageSource#setEntityClass(Class)
*/
public S entityClass(Class<?> entityClass) {
this.target.setEntityClass(entityClass);
return _this();
}
/**
* Allow you to manage which find* method to invoke.
* @param expectSingleResult true if a single result is expected.
* @return the spec
* @see AbstractMongoDbMessageSource#setExpectSingleResult(boolean)
*/
public S expectSingleResult(boolean expectSingleResult) {
this.target.setExpectSingleResult(expectSingleResult);
return _this();
}
/**
* Configure a collection name to query against.
* @param collectionName the name of the MongoDb collection
* @return the spec
*/
public S collectionName(String collectionName) {
return collectionNameExpression(new LiteralExpression(collectionName));
}
/**
* Configure a SpEL expression to evaluation a collection name on each {@code receive()} call.
* @param collectionNameExpression the SpEL expression for name of the MongoDb collection
* @return the spec
*/
public S collectionNameExpression(String collectionNameExpression) {
return collectionNameExpression(PARSER.parseExpression(collectionNameExpression));
}
/**
* Configure a {@link Supplier} to obtain a collection name on each {@code receive()} call.
* @param collectionNameSupplier the {@link Supplier} for name of the MongoDb collection
* @return the spec
*/
public S collectionNameSupplier(Supplier<String> collectionNameSupplier) {
return collectionNameExpression(new SupplierExpression<>(collectionNameSupplier));
}
/**
* Configure a SpEL expression to evaluation a collection name on each {@code receive()} call.
* @param collectionNameExpression the SpEL expression for name of the MongoDb collection
* @return the spec
* @see AbstractMongoDbMessageSource#setCollectionNameExpression(Expression)
*/
public S collectionNameExpression(Expression collectionNameExpression) {
this.target.setCollectionNameExpression(collectionNameExpression);
return _this();
}
/**
* Configure a custom {@link MongoConverter} used to assist in deserialization
* data read from MongoDb.
* @param mongoConverter The mongo converter.
* @return the spec
* @see AbstractMongoDbMessageSource#setMongoConverter(MongoConverter)
*/
public S mongoConverter(MongoConverter mongoConverter) {
this.target.setMongoConverter(mongoConverter);
return _this();
}
/**
* Configure a MongoDB update.
* @param update the MongoDB update.
* @return the spec
*/
public S update(String update) {
return update(new LiteralExpression(update));
}
/**
* Configure a MongoDB update.
* @param update the MongoDB update.
* @return the spec
*/
public S update(Update update) {
return update(new ValueExpression<>(update));
}
/**
* Configure a {@link Supplier} to produce a MongoDB update on each receive call.
* @param updateSupplier the {@link Supplier} for MongoDB update.
* @return the spec
*/
public S updateSupplier(Supplier<Update> updateSupplier) {
return update(new SupplierExpression<>(updateSupplier));
}
/**
* Configure a SpEL expression to evaluate a MongoDB update.
* @param updateExpression the expression to evaluate a MongoDB update.
* @return the spec
*/
public S update(Expression updateExpression) {
this.target.setUpdateExpression(updateExpression);
return _this();
}
}

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

@@ -1,5 +1,5 @@
/*
* Copyright 2016-2020 the original author or authors.
* Copyright 2016-2021 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.
@@ -154,6 +154,54 @@ public final class MongoDb {
return new MongoDbChangeStreamMessageProducerSpec(new MongoDbChangeStreamMessageProducer(mongoOperations));
}
/**
* Create a {@link MongoDbMessageSourceSpec} builder instance
* based on the provided {@link MongoDatabaseFactory}.
* @param mongoDbFactory the {@link MongoDatabaseFactory} to use.
* @param query the MongoDb query
* @return the {@link MongoDbMessageSourceSpec} instance
* @since 5.5
*/
public static MongoDbMessageSourceSpec inboundChannelAdapter(MongoDatabaseFactory mongoDbFactory, String query) {
return new MongoDbMessageSourceSpec(mongoDbFactory, new LiteralExpression(query));
}
/**
* Create a {@link MongoDbMessageSourceSpec} builder instance
* based on the provided {@link MongoDatabaseFactory}.
* @param mongoDbFactory the {@link MongoDatabaseFactory} to use.
* @param query the MongoDb query DSL object
* @return the {@link MongoDbMessageSourceSpec} instance
* @since 5.5
*/
public static MongoDbMessageSourceSpec inboundChannelAdapter(MongoDatabaseFactory mongoDbFactory, Query query) {
return new MongoDbMessageSourceSpec(mongoDbFactory, new ValueExpression<>(query));
}
/**
* Create a {@link MongoDbMessageSourceSpec} builder instance
* based on the provided {@link MongoOperations}.
* @param mongoTemplate the {@link MongoOperations} to use.
* @param query the MongoDb query
* @return the {@link MongoDbMessageSourceSpec} instance
* @since 5.5
*/
public static MongoDbMessageSourceSpec inboundChannelAdapter(MongoOperations mongoTemplate, String query) {
return new MongoDbMessageSourceSpec(mongoTemplate, new LiteralExpression(query));
}
/**
* Create a {@link MongoDbMessageSourceSpec} builder instance
* based on the provided {@link MongoOperations}.
* @param mongoTemplate the {@link MongoOperations} to use.
* @param query the MongoDb query DSL object
* @return the {@link MongoDbMessageSourceSpec} instance
* @since 5.5
*/
public static MongoDbMessageSourceSpec reactiveInboundChannelAdapter(MongoOperations mongoTemplate, Query query) {
return new MongoDbMessageSourceSpec(mongoTemplate, new ValueExpression<>(query));
}
private MongoDb() {
}

42
spring-integration-mongodb/src/main/java/org/springframework/integration/mongodb/dsl/MongoDbMessageSourceSpec.java Normal file
View File

@@ -0,0 +1,42 @@
/*
* Copyright 2021 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 org.springframework.data.mongodb.MongoDatabaseFactory;
import org.springframework.data.mongodb.core.MongoOperations;
import org.springframework.expression.Expression;
import org.springframework.integration.mongodb.inbound.MongoDbMessageSource;
/**
* A {@link AbstractMongoDbMessageSourceSpec} implementation for a {@link MongoDbMessageSource}.
*
* @author Artem Bilan
*
* @since 5.5
*/
public class MongoDbMessageSourceSpec
extends AbstractMongoDbMessageSourceSpec<MongoDbMessageSourceSpec, MongoDbMessageSource> {
protected MongoDbMessageSourceSpec(MongoDatabaseFactory mongoDatabaseFactory, Expression queryExpression) {
this.target = new MongoDbMessageSource(mongoDatabaseFactory, queryExpression);
}
protected MongoDbMessageSourceSpec(MongoOperations mongoTemplate, Expression queryExpression) {
this.target = new MongoDbMessageSource(mongoTemplate, queryExpression);
}
}

88
spring-integration-mongodb/src/main/java/org/springframework/integration/mongodb/dsl/ReactiveMongoDbMessageSourceSpec.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2020 the original author or authors.
* Copyright 2020-2021 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.
@@ -16,26 +16,20 @@
package org.springframework.integration.mongodb.dsl;
import java.util.function.Supplier;
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.MessageSourceSpec;
import org.springframework.integration.expression.SupplierExpression;
import org.springframework.integration.mongodb.inbound.ReactiveMongoDbMessageSource;
/**
* A {@link MessageSourceSpec} implementation for a {@link ReactiveMongoDbMessageSource}.
* A {@link AbstractMongoDbMessageSourceSpec} implementation for a {@link ReactiveMongoDbMessageSource}.
*
* @author Artem Bilan
*
* @since 5.3
*/
public class ReactiveMongoDbMessageSourceSpec
extends MessageSourceSpec<ReactiveMongoDbMessageSourceSpec, ReactiveMongoDbMessageSource> {
extends AbstractMongoDbMessageSourceSpec<ReactiveMongoDbMessageSourceSpec, ReactiveMongoDbMessageSource> {
protected ReactiveMongoDbMessageSourceSpec(ReactiveMongoDatabaseFactory reactiveMongoDatabaseFactory,
Expression queryExpression) {
@@ -49,80 +43,4 @@ public class ReactiveMongoDbMessageSourceSpec
this.target = new ReactiveMongoDbMessageSource(reactiveMongoTemplate, queryExpression);
}
/**
* Allow you to set the type of the entityClass that will be passed to the
* {@link ReactiveMongoOperations#find} or {@link ReactiveMongoOperations#findOne}
* method.
* Default is {@link com.mongodb.DBObject}.
* @param entityClass The entity class.
* @return the spec
* @see ReactiveMongoDbMessageSource#setEntityClass(Class)
*/
public ReactiveMongoDbMessageSourceSpec entityClass(Class<?> entityClass) {
this.target.setEntityClass(entityClass);
return this;
}
/**
* Allow you to manage which find* method to invoke on {@link ReactiveMongoOperations}.
* @param expectSingleResult true if a single result is expected.
* @return the spec
* @see ReactiveMongoDbMessageSource#setExpectSingleResult(boolean)
*/
public ReactiveMongoDbMessageSourceSpec expectSingleResult(boolean expectSingleResult) {
this.target.setExpectSingleResult(expectSingleResult);
return this;
}
/**
* Configure a collection name to query against.
* @param collectionName the name of the MongoDb collection
* @return the spec
*/
public ReactiveMongoDbMessageSourceSpec collectionName(String collectionName) {
return collectionNameExpression(new LiteralExpression(collectionName));
}
/**
* Configure a SpEL expression to evaluation a collection name on each {@code receive()} call.
* @param collectionNameExpression the SpEL expression for name of the MongoDb collection
* @return the spec
*/
public ReactiveMongoDbMessageSourceSpec collectionNameExpression(String collectionNameExpression) {
return collectionNameExpression(PARSER.parseExpression(collectionNameExpression));
}
/**
* Configure a {@link Supplier} to obtain a collection name on each {@code receive()} call.
* @param collectionNameSupplier the {@link Supplier} for name of the MongoDb collection
* @return the spec
*/
public ReactiveMongoDbMessageSourceSpec collectionNameSupplier(Supplier<String> collectionNameSupplier) {
return collectionNameExpression(new SupplierExpression<>(collectionNameSupplier));
}
/**
* Configure a SpEL expression to evaluation a collection name on each {@code receive()} call.
* @param collectionNameExpression the SpEL expression for name of the MongoDb collection
* @return the spec
* @see ReactiveMongoDbMessageSource#setCollectionNameExpression(Expression)
*/
public ReactiveMongoDbMessageSourceSpec collectionNameExpression(Expression collectionNameExpression) {
this.target.setCollectionNameExpression(collectionNameExpression);
return this;
}
/**
* Configure 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.
* @return the spec
* @see ReactiveMongoDbMessageSource#setMongoConverter(MongoConverter)
*/
public ReactiveMongoDbMessageSourceSpec mongoConverter(MongoConverter mongoConverter) {
this.target.setMongoConverter(mongoConverter);
return this;
}
}

273
spring-integration-mongodb/src/main/java/org/springframework/integration/mongodb/inbound/AbstractMongoDbMessageSource.java Normal file
View File

@@ -0,0 +1,273 @@
/*
* Copyright 2021 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 java.util.Collection;
import java.util.Map;
import org.bson.Document;
import org.springframework.beans.BeansException;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationContextAware;
import org.springframework.data.mapping.IdentifierAccessor;
import org.springframework.data.mapping.context.MappingContext;
import org.springframework.data.mongodb.ReactiveMongoDatabaseFactory;
import org.springframework.data.mongodb.core.ReactiveMongoTemplate;
import org.springframework.data.mongodb.core.convert.MongoConverter;
import org.springframework.data.mongodb.core.mapping.MongoPersistentEntity;
import org.springframework.data.mongodb.core.query.BasicQuery;
import org.springframework.data.mongodb.core.query.BasicUpdate;
import org.springframework.data.mongodb.core.query.Criteria;
import org.springframework.data.mongodb.core.query.Query;
import org.springframework.data.mongodb.core.query.Update;
import org.springframework.data.mongodb.core.query.UpdateDefinition;
import org.springframework.data.util.Pair;
import org.springframework.expression.Expression;
import org.springframework.expression.TypeLocator;
import org.springframework.expression.common.LiteralExpression;
import org.springframework.expression.spel.support.StandardTypeLocator;
import org.springframework.integration.endpoint.AbstractMessageSource;
import org.springframework.integration.mongodb.support.MongoHeaders;
import org.springframework.lang.Nullable;
import org.springframework.util.Assert;
import org.springframework.util.LinkedMultiValueMap;
import org.springframework.util.MultiValueMap;
import com.mongodb.DBObject;
/**
* An {@link AbstractMessageSource} extension for common MongoDB sources options and support methods.
*
* @param <T> The payload type.
*
* @author Artem Bilan
*
* @since 5.5
*/
public abstract class AbstractMongoDbMessageSource<T> extends AbstractMessageSource<T>
implements ApplicationContextAware {
private static final String ID_FIELD = "_id";
protected final Expression queryExpression;
private Expression collectionNameExpression = new LiteralExpression("data");
private MongoConverter mongoConverter;
private Class<?> entityClass = DBObject.class;
private boolean expectSingleResult = false;
private Expression updateExpression;
private ApplicationContext applicationContext;
private volatile boolean initialized = false;
protected AbstractMongoDbMessageSource(Expression queryExpression) {
Assert.notNull(queryExpression, "'queryExpression' must not be null");
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) {
this.mongoConverter = mongoConverter;
}
/**
* Specify an optional {@code update} for just polled records from the collection.
* @param updateExpression SpEL expression for an {@link UpdateDefinition}.
* @since 5.5
*/
public void setUpdateExpression(Expression updateExpression) {
this.updateExpression = updateExpression;
}
@Override
public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
this.applicationContext = applicationContext;
}
public Expression getCollectionNameExpression() {
return this.collectionNameExpression;
}
public MongoConverter getMongoConverter() {
return this.mongoConverter;
}
public Class<?> getEntityClass() {
return this.entityClass;
}
public boolean isExpectSingleResult() {
return this.expectSingleResult;
}
public Expression getUpdateExpression() {
return this.updateExpression;
}
public ApplicationContext getApplicationContext() {
return this.applicationContext;
}
protected void setInitialized(boolean initialized) {
this.initialized = initialized;
}
protected boolean isInitialized() {
return this.initialized;
}
@Override
protected void onInit() {
super.onInit();
TypeLocator typeLocator = getEvaluationContext().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");
}
}
protected Query evaluateQueryExpression() {
Object value = this.queryExpression.getValue(getEvaluationContext());
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);
}
return query;
}
protected String evaluateCollectionNameExpression() {
String collectionName = getCollectionNameExpression().getValue(getEvaluationContext(), String.class);
Assert.notNull(collectionName, "'collectionNameExpression' must not evaluate to null");
return collectionName;
}
/*
* Inspired by {@code org.springframework.data.mongodb.core.EntityOperations#getByIdInQuery}
*/
protected Query getByIdInQuery(Collection<?> entities) {
MultiValueMap<String, Object> byIds = new LinkedMultiValueMap<>();
entities.stream()
.map(this::idForEntity)
.forEach(it -> byIds.add(it.getFirst(), it.getSecond()));
Criteria[] criterias = byIds.entrySet().stream()
.map(it -> Criteria.where(it.getKey()).in(it.getValue()))
.toArray(Criteria[]::new);
return new Query(criterias.length == 1 ? criterias[0] : new Criteria().orOperator(criterias));
}
@SuppressWarnings("unchecked")
protected Pair<String, Object> idForEntity(Object entity) {
if (entity instanceof String) {
return idFieldFromMap(Document.parse(entity.toString()));
}
if (entity instanceof Map) {
return idFieldFromMap((Map<String, Object>) entity);
}
MappingContext<? extends MongoPersistentEntity<?>, ?> context = this.mongoConverter.getMappingContext();
MongoPersistentEntity<?> persistentEntity = context.getRequiredPersistentEntity(entity.getClass());
String idField = persistentEntity.getRequiredIdProperty().getFieldName();
IdentifierAccessor idAccessor = persistentEntity.getIdentifierAccessor(entity);
return Pair.of(idField, idAccessor.getRequiredIdentifier());
}
@Nullable
protected Update evaluateUpdateExpression() {
if (this.updateExpression != null) {
Object value = this.updateExpression.getValue(getEvaluationContext());
Assert.notNull(value, "'updateExpression' must not evaluate to null");
Update update;
if (value instanceof String) {
update = new BasicUpdate((String) value);
}
else if (value instanceof Update) {
update = ((Update) value);
}
else {
throw new IllegalStateException("'updateExpression' must evaluate to String " +
"or org.springframework.data.mongodb.core.query.Update");
}
return update;
}
return null;
}
private static Pair<String, Object> idFieldFromMap(Map<String, Object> map) {
return Pair.of(ID_FIELD, map.get(ID_FIELD));
}
}

162
spring-integration-mongodb/src/main/java/org/springframework/integration/mongodb/inbound/MongoDbMessageSource.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2007-2020 the original author or authors.
* Copyright 2007-2021 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.
@@ -16,21 +16,18 @@
package org.springframework.integration.mongodb.inbound;
import java.util.Collection;
import java.util.List;
import org.springframework.context.ApplicationContext;
import org.springframework.data.mongodb.MongoDatabaseFactory;
import org.springframework.data.mongodb.core.MongoOperations;
import org.springframework.data.mongodb.core.MongoTemplate;
import org.springframework.data.mongodb.core.convert.MongoConverter;
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.data.mongodb.core.query.Update;
import org.springframework.data.util.Pair;
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.integration.support.AbstractIntegrationMessageBuilder;
import org.springframework.integration.transaction.IntegrationResourceHolder;
@@ -39,8 +36,6 @@ import org.springframework.transaction.support.TransactionSynchronizationManager
import org.springframework.util.Assert;
import org.springframework.util.CollectionUtils;
import com.mongodb.DBObject;
/**
* 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
@@ -64,43 +59,28 @@ import com.mongodb.DBObject;
*
* @since 2.2
*/
public class MongoDbMessageSource extends AbstractMessageSource<Object> {
public class MongoDbMessageSource extends AbstractMongoDbMessageSource<Object> {
@Nullable
private final MongoDatabaseFactory mongoDbFactory;
private final Expression queryExpression;
private Expression collectionNameExpression = new LiteralExpression("data");
private StandardEvaluationContext evaluationContext;
private MongoOperations mongoTemplate;
private MongoConverter mongoConverter;
private Class<?> entityClass = DBObject.class;
private boolean expectSingleResult = false;
private volatile boolean initialized = false;
/**
* Creates an instance with the provided {@link MongoDatabaseFactory} and SpEL expression
* Create an instance with the provided {@link MongoDatabaseFactory} 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 mongoDbFactory The mongodb factory.
* @param queryExpression The query expression.
*/
public MongoDbMessageSource(MongoDatabaseFactory mongoDbFactory, Expression queryExpression) {
super(queryExpression);
Assert.notNull(mongoDbFactory, "'mongoDbFactory' must not be null");
Assert.notNull(queryExpression, "'queryExpression' must not be null");
this.mongoDbFactory = mongoDbFactory;
this.queryExpression = queryExpression;
}
/**
* Creates an instance with the provided {@link MongoOperations} and SpEL expression
* Create an instance with the provided {@link MongoOperations} and SpEL expression
* which should resolve to a Mongo 'query' string (see https://www.mongodb.org/display/DOCS/Querying).
* It assumes that the {@link MongoOperations} is fully initialized and ready to be used.
* The 'queryExpression' will be evaluated on every call to the {@link #receive()} method.
@@ -108,59 +88,10 @@ public class MongoDbMessageSource extends AbstractMessageSource<Object> {
* @param queryExpression The query expression.
*/
public MongoDbMessageSource(MongoOperations mongoTemplate, Expression queryExpression) {
super(queryExpression);
Assert.notNull(mongoTemplate, "'mongoTemplate' must not be null");
Assert.notNull(queryExpression, "'queryExpression' must not be null");
this.mongoDbFactory = null;
this.mongoTemplate = mongoTemplate;
this.queryExpression = queryExpression;
}
/**
* Allows you to set the type of the entityClass that will be passed to the
* {@link MongoTemplate#find(Query, Class)} or {@link MongoTemplate#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;
}
/**
* Allows you to manage which find* method to invoke on {@link MongoTemplate}.
* Default is 'false', which means the {@link #receive()} method will use
* the {@link MongoTemplate#find(Query, Class)} method. If set to 'true',
* {@link #receive()} will use {@link MongoTemplate#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;
}
/**
* Sets 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;
}
/**
* Allows 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 MongoDatabaseFactory}.
* @param mongoConverter The mongo converter.
*/
public void setMongoConverter(MongoConverter mongoConverter) {
Assert.isNull(this.mongoTemplate,
"'mongoConverter' can not be set when instance was constructed with MongoTemplate");
this.mongoConverter = mongoConverter;
}
@Override
@@ -170,63 +101,53 @@ public class MongoDbMessageSource extends AbstractMessageSource<Object> {
@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");
}
super.onInit();
if (this.mongoDbFactory != null) {
this.mongoTemplate = new MongoTemplate(this.mongoDbFactory, this.mongoConverter);
MongoTemplate template = new MongoTemplate(this.mongoDbFactory, getMongoConverter());
ApplicationContext applicationContext = getApplicationContext();
if (applicationContext != null) {
template.setApplicationContext(applicationContext);
}
this.mongoTemplate = template;
}
this.initialized = true;
setMongoConverter(this.mongoTemplate.getConverter());
setInitialized(true);
}
/**
* Will execute a {@link Query} returning its results as the Message payload.
* The payload can be either {@link List} of elements 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
* identified by {@link #getEntityClass()}, or a single element of type identified by {@link #getEntityClass()}
* based on the value of {@link #isExpectSingleResult()} attribute which defaults to 'false' resulting
* {@link org.springframework.messaging.Message} with payload of type
* {@link List}. The collection name used in the
* query will be provided in the {@link MongoHeaders#COLLECTION_NAME} header.
*/
@Override
protected Object doReceive() {
Assert.isTrue(this.initialized, "This class is not yet initialized. Invoke its afterPropertiesSet() method");
Assert.isTrue(isInitialized(), "This class is not yet initialized. Invoke its afterPropertiesSet() method");
AbstractIntegrationMessageBuilder<Object> messageBuilder = null;
Object value = this.queryExpression.getValue(this.evaluationContext);
Assert.notNull(value, "'queryExpression' must not evaluate to null");
Query query;
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");
}
Assert.notNull(query, "'queryExpression' must not evaluate to null");
String collectionName = this.collectionNameExpression.getValue(this.evaluationContext, String.class);
Assert.notNull(collectionName, "'collectionNameExpression' must not evaluate to null");
Query query = evaluateQueryExpression();
String collectionName = evaluateCollectionNameExpression();
Object result = null;
if (this.expectSingleResult) {
result = this.mongoTemplate.findOne(query, this.entityClass, collectionName);
if (isExpectSingleResult()) {
result = this.mongoTemplate.findOne(query, getEntityClass(), collectionName);
}
else {
List<?> results = this.mongoTemplate.find(query, this.entityClass, collectionName);
List<?> results = this.mongoTemplate.find(query, getEntityClass(), collectionName);
if (!CollectionUtils.isEmpty(results)) {
result = results;
}
}
if (result != null) {
messageBuilder = this.getMessageBuilderFactory().withPayload(result)
.setHeader(MongoHeaders.COLLECTION_NAME, collectionName);
updateIfAny(result, collectionName);
messageBuilder =
getMessageBuilderFactory()
.withPayload(result)
.setHeader(MongoHeaders.COLLECTION_NAME, collectionName);
}
Object holder = TransactionSynchronizationManager.getResource(this);
@@ -238,4 +159,19 @@ public class MongoDbMessageSource extends AbstractMessageSource<Object> {
return messageBuilder;
}
private void updateIfAny(Object result, String collectionName) {
Update update = evaluateUpdateExpression();
if (update != null) {
if (result instanceof List) {
this.mongoTemplate.updateMulti(getByIdInQuery((Collection<?>) result), update, collectionName);
}
else {
Pair<String, Object> idPair = idForEntity(result);
Query query = new Query(Criteria.where(idPair.getFirst()).is(idPair.getSecond()));
this.mongoTemplate.updateFirst(query, update, collectionName);
}
}
}
}

183
spring-integration-mongodb/src/main/java/org/springframework/integration/mongodb/inbound/ReactiveMongoDbMessageSource.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2020 the original author or authors.
* Copyright 2020-2021 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,27 +18,21 @@ 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.Criteria;
import org.springframework.data.mongodb.core.query.Query;
import org.springframework.data.mongodb.core.query.Update;
import org.springframework.data.util.Pair;
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.lang.Nullable;
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
@@ -59,30 +53,13 @@ import com.mongodb.DBObject;
*
* @since 5.3
*/
public class ReactiveMongoDbMessageSource extends AbstractMessageSource<Publisher<?>>
implements ApplicationContextAware {
public class ReactiveMongoDbMessageSource extends AbstractMongoDbMessageSource<Publisher<?>> {
@Nullable
private final ReactiveMongoDatabaseFactory reactiveMongoDatabaseFactory;
private final Expression queryExpression;
private Expression collectionNameExpression = new LiteralExpression("data");
private StandardEvaluationContext evaluationContext;
private ReactiveMongoOperations reactiveMongoTemplate;
private MongoConverter mongoConverter;
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).
@@ -93,10 +70,9 @@ public class ReactiveMongoDbMessageSource extends AbstractMessageSource<Publishe
public ReactiveMongoDbMessageSource(ReactiveMongoDatabaseFactory reactiveMongoDatabaseFactory,
Expression queryExpression) {
super(queryExpression);
Assert.notNull(reactiveMongoDatabaseFactory, "'reactiveMongoDatabaseFactory' must not be null");
Assert.notNull(queryExpression, "'queryExpression' must not be null");
this.reactiveMongoDatabaseFactory = reactiveMongoDatabaseFactory;
this.queryExpression = queryExpression;
}
/**
@@ -108,65 +84,10 @@ public class ReactiveMongoDbMessageSource extends AbstractMessageSource<Publishe
* @param queryExpression The query expression.
*/
public ReactiveMongoDbMessageSource(ReactiveMongoOperations reactiveMongoTemplate, Expression queryExpression) {
super(queryExpression);
Assert.notNull(reactiveMongoTemplate, "'reactiveMongoTemplate' must not be null");
Assert.notNull(queryExpression, "'queryExpression' must not be null");
this.reactiveMongoDatabaseFactory = 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
@@ -176,63 +97,83 @@ public class ReactiveMongoDbMessageSource extends AbstractMessageSource<Publishe
@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");
}
super.onInit();
if (this.reactiveMongoDatabaseFactory != null) {
ReactiveMongoTemplate template =
new ReactiveMongoTemplate(this.reactiveMongoDatabaseFactory, this.mongoConverter);
if (this.applicationContext != null) {
template.setApplicationContext(this.applicationContext);
new ReactiveMongoTemplate(this.reactiveMongoDatabaseFactory, getMongoConverter());
ApplicationContext applicationContext = getApplicationContext();
if (applicationContext != null) {
template.setApplicationContext(applicationContext);
}
this.reactiveMongoTemplate = template;
}
this.initialized = true;
setMongoConverter(this.reactiveMongoTemplate.getConverter());
setInitialized(true);
}
/**
* Execute a {@link Query} returning its results as the Message payload.
* The payload can be either {@link reactor.core.publisher.Flux} or
* {@link reactor.core.publisher.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 reactor.core.publisher.Mono} of objects of type identified by {@link #getEntityClass()},
* or a single element of type identified by {@link #getEntityClass()}
* based on the value of {@link #isExpectSingleResult()} attribute which defaults to 'false' resulting
* {@link org.springframework.messaging.Message} with payload of type
* {@link reactor.core.publisher.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);
Assert.isTrue(isInitialized(), "This class is not yet initialized. Invoke its afterPropertiesSet() method");
Query query = evaluateQueryExpression();
String collectionName = evaluateCollectionNameExpression();
Publisher<?> result;
if (isExpectSingleResult()) {
result = this.reactiveMongoTemplate.findOne(query, getEntityClass(), collectionName);
}
else {
throw new IllegalStateException("'queryExpression' must evaluate to String " +
"or org.springframework.data.mongodb.core.query.Query, but not: " + query);
result = this.reactiveMongoTemplate.find(query, getEntityClass(), collectionName);
}
String collectionName = this.collectionNameExpression.getValue(this.evaluationContext, String.class);
Assert.notNull(collectionName, "'collectionNameExpression' must not evaluate to null");
result = updateIfAny(result, collectionName);
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);
}
private Publisher<?> updateIfAny(Publisher<?> result, String collectionName) {
Update update = evaluateUpdateExpression();
if (update != null) {
if (result instanceof Mono) {
return updateSingle((Mono<?>) result, update, collectionName);
}
else {
return updateMulti((Flux<?>) result, update, collectionName);
}
}
else {
return result;
}
}
private Publisher<?> updateSingle(Mono<?> result, Update update, String collectionName) {
return result.flatMap((entity) -> {
Pair<String, Object> idPair = idForEntity(entity);
Query query = new Query(Criteria.where(idPair.getFirst()).is(idPair.getSecond()));
return this.reactiveMongoTemplate.updateFirst(query, update, collectionName)
.thenReturn(entity);
});
}
private Publisher<?> updateMulti(Flux<?> result, Update update, String collectionName) {
return result.collectList()
.flatMapMany((entities) ->
this.reactiveMongoTemplate.updateMulti(getByIdInQuery(entities), update, collectionName)
.thenMany(Flux.fromIterable(entities))
);
}
}

22
spring-integration-mongodb/src/main/resources/org/springframework/integration/mongodb/config/spring-integration-mongodb.xsd
View File

@@ -53,6 +53,28 @@
</xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="update" type="xsd:string">
<xsd:annotation>
<xsd:documentation>
String representation of MongoDb Update (e.g., update="{$set: {'name' : 'Bob'}").
Please refer to MongoDb documentation for more query samples
https://www.mongodb.org/display/DOCS/Querying
This attribute is
mutually exclusive with 'update-expression' attribute.
</xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="update-expression" type="xsd:string">
<xsd:annotation>
<xsd:documentation>
SpEL expression which should resolve to a String update (please refer to the 'update'
attribute),
or to an instance of MongoDb Update (e.q.,
update-expression="T(Update).update('name', 'Bob')").
This attribute is mutually exclusive with 'update' attribute.
</xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="entity-class" type="xsd:string">
<xsd:annotation>
<xsd:appinfo>

1
spring-integration-mongodb/src/test/java/org/springframework/integration/mongodb/config/MongoDbInboundChannelAdapterParserTests-context.xml
View File

@@ -29,6 +29,7 @@
<int-mongodb:inbound-channel-adapter id="fullConfigWithQueryExpression"
collection-name-expression="'foo'"
query-expression="new BasicQuery('{''address.state'' : ''PA''}').limit(2)"
update="{ $set: {'address.state' : 'NJ'} }"
mongo-converter="mongoConverter"
mongodb-factory="mongoDbFactory"
auto-startup="false">

30
spring-integration-mongodb/src/test/java/org/springframework/integration/mongodb/config/MongoDbInboundChannelAdapterParserTests.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2020 the original author or authors.
* Copyright 2002-2021 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.
@@ -17,9 +17,9 @@
package org.springframework.integration.mongodb.config;
import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.assertThatExceptionOfType;
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;
@@ -34,8 +34,7 @@ import org.springframework.integration.endpoint.SourcePollingChannelAdapter;
import org.springframework.integration.mongodb.inbound.MongoDbMessageSource;
import org.springframework.integration.test.util.TestUtils;
import org.springframework.test.annotation.DirtiesContext;
import org.springframework.test.context.ContextConfiguration;
import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
import org.springframework.test.context.junit.jupiter.SpringJUnitConfig;
/**
* @author Oleg Zhurakousky
@@ -43,8 +42,7 @@ import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
* @author Artem Bilan
* @author Yaron Yamin
*/
@ContextConfiguration
@RunWith(SpringJUnit4ClassRunner.class)
@SpringJUnitConfig
@DirtiesContext
public class MongoDbInboundChannelAdapterParserTests {
@@ -112,6 +110,8 @@ public class MongoDbInboundChannelAdapterParserTests {
assertThat(TestUtils.getPropertyValue(source, "queryExpression") instanceof SpelExpression).isTrue();
assertThat(TestUtils.getPropertyValue(source, "queryExpression.expression"))
.isEqualTo("new BasicQuery('{''address.state'' : ''PA''}').limit(2)");
assertThat(TestUtils.getPropertyValue(source, "updateExpression.literalValue"))
.isEqualTo("{ $set: {'address.state' : 'NJ'} }");
}
@Test
@@ -152,16 +152,20 @@ public class MongoDbInboundChannelAdapterParserTests {
assertThat(TestUtils.getPropertyValue(source, "collectionNameExpression.literalValue")).isEqualTo("foo");
}
@Test(expected = BeanDefinitionParsingException.class)
@Test
public void templateAndFactoryFail() {
new ClassPathXmlApplicationContext("inbound-adapter-parser-fail-template-factory-config.xml", this.getClass())
.close();
assertThatExceptionOfType(BeanDefinitionParsingException.class)
.isThrownBy(() ->
new ClassPathXmlApplicationContext("inbound-adapter-parser-fail-template-factory-config.xml",
getClass()));
}
@Test(expected = BeanDefinitionParsingException.class)
@Test
public void templateAndConverterFail() {
new ClassPathXmlApplicationContext("inbound-adapter-parser-fail-template-converter-config.xml", this.getClass())
.close();
assertThatExceptionOfType(BeanDefinitionParsingException.class)
.isThrownBy(() ->
new ClassPathXmlApplicationContext("inbound-adapter-parser-fail-template-converter-config.xml",
getClass()));
}
private MongoDbMessageSource assertMongoDbMessageSource(Object testedBean) {

23
spring-integration-mongodb/src/test/java/org/springframework/integration/mongodb/inbound/MongoDbMessageSourceTests.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2007-2020 the original author or authors.
* Copyright 2007-2021 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.
@@ -35,6 +35,7 @@ import org.springframework.data.mongodb.core.MongoOperations;
import org.springframework.data.mongodb.core.MongoTemplate;
import org.springframework.data.mongodb.core.convert.MappingMongoConverter;
import org.springframework.data.mongodb.core.mapping.MongoMappingContext;
import org.springframework.data.mongodb.core.query.BasicQuery;
import org.springframework.expression.Expression;
import org.springframework.expression.common.LiteralExpression;
import org.springframework.expression.spel.standard.SpelExpressionParser;
@@ -241,10 +242,8 @@ public class MongoDbMessageSourceTests extends MongoDbAvailableTests {
@SuppressWarnings("unchecked")
@Test
@MongoDbAvailable
public void validateSuccessfulQueryWithMongoTemplate() {
MongoDatabaseFactory mongoDbFactory = this.prepareMongoFactory();
public void validateSuccessfulQueryWithMongoTemplateAndUpdate() {
MongoDatabaseFactory mongoDbFactory = prepareMongoFactory();
MappingMongoConverter converter = new TestMongoConverter(mongoDbFactory, new MongoMappingContext());
converter.afterPropertiesSet();
converter = spy(converter);
@@ -253,16 +252,22 @@ public class MongoDbMessageSourceTests extends MongoDbAvailableTests {
Expression queryExpression = new LiteralExpression("{'address.state' : 'PA'}");
MongoDbMessageSource messageSource = new MongoDbMessageSource(template, queryExpression);
messageSource.setBeanFactory(mock(BeanFactory.class));
messageSource.setUpdateExpression(new LiteralExpression("{ $set: {'address.state' : 'NJ'} }"));
messageSource.afterPropertiesSet();
MongoTemplate writingTemplate = new MongoTemplate(mongoDbFactory, converter);
writingTemplate.save(this.createPerson("Manny"), "data");
writingTemplate.save(this.createPerson("Moe"), "data");
writingTemplate.save(this.createPerson("Jack"), "data");
writingTemplate.save(createPerson("Manny"), "data");
writingTemplate.save(createPerson("Moe"), "data");
writingTemplate.save(createPerson("Jack"), "data");
List<Person> persons = (List<Person>) messageSource.receive().getPayload();
assertThat(persons.size()).isEqualTo(3);
assertThat(persons).hasSize(3);
verify(converter, times(3)).read((Class<Person>) Mockito.any(), Mockito.any(Bson.class));
assertThat(messageSource.receive()).isNull();
assertThat(template.find(new BasicQuery("{'address.state' : 'NJ'}"), Object.class, "data"))
.hasSize(3);
}
@Test

13
spring-integration-mongodb/src/test/java/org/springframework/integration/mongodb/inbound/ReactiveMongoDbMessageSourceTests.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2020 the original author or authors.
* Copyright 2020-2021 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.
@@ -42,6 +42,7 @@ 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.data.mongodb.core.query.Update;
import org.springframework.expression.Expression;
import org.springframework.expression.common.LiteralExpression;
import org.springframework.integration.channel.FluxMessageChannel;
@@ -91,7 +92,7 @@ public class ReactiveMongoDbMessageSourceTests extends MongoDbAvailableTests {
@MongoDbAvailable
@SuppressWarnings("unchecked")
public void validateSuccessfulQueryWithSingleElementFluxOfDbObject() {
ReactiveMongoDatabaseFactory reactiveMongoDatabaseFactory = this.prepareReactiveMongoFactory();
ReactiveMongoDatabaseFactory reactiveMongoDatabaseFactory = prepareReactiveMongoFactory();
ReactiveMongoTemplate template = new ReactiveMongoTemplate(reactiveMongoDatabaseFactory);
waitFor(template.save(createPerson(), "data"));
@@ -111,7 +112,7 @@ public class ReactiveMongoDbMessageSourceTests extends MongoDbAvailableTests {
@MongoDbAvailable
@SuppressWarnings("unchecked")
public void validateSuccessfulQueryWithSingleElementFluxOfPerson() {
ReactiveMongoDatabaseFactory reactiveMongoDatabaseFactory = this.prepareReactiveMongoFactory();
ReactiveMongoDatabaseFactory reactiveMongoDatabaseFactory = prepareReactiveMongoFactory();
ReactiveMongoTemplate template = new ReactiveMongoTemplate(reactiveMongoDatabaseFactory);
waitFor(template.save(createPerson(), "data"));
@@ -183,6 +184,7 @@ public class ReactiveMongoDbMessageSourceTests extends MongoDbAvailableTests {
StepVerifier.create(output)
.assertNext(
message -> assertThat(((Person) message.getPayload()).getName()).isEqualTo("Oleg"))
.expectNoEvent(Duration.ofMillis(100))
.thenCancel()
.verify(Duration.ofSeconds(10));
@@ -247,9 +249,10 @@ public class ReactiveMongoDbMessageSourceTests extends MongoDbAvailableTests {
public IntegrationFlow pollingFlow() {
return IntegrationFlows
.from(MongoDb.reactiveInboundChannelAdapter(
MongoDbAvailableTests.REACTIVE_MONGO_DATABASE_FACTORY, "{'name' : 'Oleg'}")
REACTIVE_MONGO_DATABASE_FACTORY, "{'name' : 'Oleg'}")
.update(Update.update("name", "DONE"))
.entityClass(Person.class),
c -> c.poller(Pollers.fixedDelay(100).maxMessagesPerPoll(1)))
c -> c.poller(Pollers.fixedDelay(100)))
.split()
.channel(c -> c.flux("output"))
.get();

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

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2020 the original author or authors.
* Copyright 2002-2021 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.
@@ -24,6 +24,7 @@ import org.bson.conversions.Bson;
import org.junit.Rule;
import org.springframework.dao.DataAccessException;
import org.springframework.data.annotation.Id;
import org.springframework.data.mapping.context.MappingContext;
import org.springframework.data.mongodb.MongoDatabaseFactory;
import org.springframework.data.mongodb.ReactiveMongoDatabaseFactory;
@@ -130,6 +131,9 @@ public abstract class MongoDbAvailableTests {
public static class Person {
@Id
private String id;
private Address address;
private String name;

235
src/reference/asciidoc/mongodb.adoc
View File

@@ -5,9 +5,10 @@ Version 2.1 introduced support for https://www.mongodb.org/[MongoDB]: a "`high-p
You need to include this dependency into your project:
====
[source, xml, subs="normal", role="primary"]
.Maven
[source, xml, subs="normal"]
----
<dependency>
<groupId>org.springframework.integration</groupId>
@@ -15,9 +16,8 @@ You need to include this dependency into your project:
<version>{project-version}</version>
</dependency>
----
[source, groovy, subs="normal", role="secondary"]
.Gradle
[source, groovy, subs="normal"]
----
compile "org.springframework.integration:spring-integration-mongodb:{project-version}"
----
@@ -33,37 +33,40 @@ To download, install, and run MongoDB, see the https://www.mongodb.org/downloads
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:
====
[source, xml, subs="normal", role="primary"]
.Maven
[source, xml, subs="normal"]
----
<dependency>
<groupId>org.mongodb</groupId>
<artifactId>mongodb-driver-reactivestreams</artifactId>
</dependency>
</dependency>
----
[source, groovy, subs="normal", role="secondary"]
.Gradle
[source, groovy, subs="normal"]
----
compile "org.mongodb:mongodb-driver-reactivestreams"
----
====
For regular synchronous client you need to add its respective driver into dependencies:
====
[source, xml, subs="normal", role="primary"]
.Maven
[source, xml, subs="normal"]
----
<dependency>
<groupId>org.mongodb</groupId>
<artifactId>mongodb-driver-sync</artifactId>
</dependency>
</dependency>
----
[source, groovy, subs="normal", role="secondary"]
.Gradle
[source, groovy, subs="normal"]
----
compile "org.mongodb:mongodb-driver-sync"
----
====
Both of them are `optional` in the framework for better end-user choice support.
@@ -77,20 +80,17 @@ TIP: Spring Data provides provides the blocking MongoDB driver by default but yo
To connect to MongoDB you can use an implementation of the `MongoDatabaseFactory` interface.
The following example shows how to use `SimpleMongoClientDatabaseFactory`, the out-of-the-box implementation, in Java:
The following example shows how to use `SimpleMongoClientDatabaseFactory`:
====
[source,java]
[source, java, role="primary"]
.Java
----
MongoDatabaseFactory mongoDbFactory =
new SimpleMongoClientDatabaseFactory(com.mongodb.client.MongoClients.create(), "test");
----
====
The following example shows how to use `SimpleMongoClientDatabaseFactory` in XML configuration:
====
[source,xml]
[source, xml, role="secondary"]
.XML
----
<bean id="mongoDbFactory" class="o.s.data.mongodb.core.SimpleMongoClientDatabaseFactory">
<constructor-arg>
@@ -109,19 +109,18 @@ For more information on how to configure MongoDB, see the https://docs.spring.io
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:
The following example shows how to use `SimpleReactiveMongoDatabaseFactory`:
====
[source,java]
[source, java, role="primary"]
.Java
----
new SimpleReactiveMongoDatabaseFactory(com.mongodb.reactivestreams.client.MongoClients.create(), "test");
ReactiveMongoDatabaseFactory mongoDbFactory =
new SimpleReactiveMongoDatabaseFactory(com.mongodb.reactivestreams.client.MongoClients.create(), "test");
----
====
The following example shows how to use `SimpleReactiveMongoDatabaseFactory` in XML configuration:
====
[source,xml]
[source, xml, role="secondary"]
.XML
----
<bean id="mongoDbFactory" class="o.s.data.mongodb.core.SimpleReactiveMongoDatabaseFactory">
<constructor-arg>
@@ -302,7 +301,7 @@ You can do so by using that transaction synchronization feature Spring Integrati
<int:transaction-synchronization-factory id="syncFactory">
<int:after-commit
expression="@documentCleaner.remove(#mongoTemplate, payload, headers.mongo_collectionName)"
channe="someChannel"/>
channel="someChannel"/>
</int:transaction-synchronization-factory>
<bean id="documentCleaner" class="thing1.thing2.DocumentCleaner"/>
@@ -348,6 +347,10 @@ If the result of an expression is null or void, no message is generated.
For more information about transaction synchronization, see <<./transactions.adoc#transaction-synchronization,Transaction Synchronization>>.
Starting with version 5.5, the `MongoDbMessageSource` can be configured with an `updateExpression`, which must evaluate to a `String` with the MongoDb `update` syntax or to an `org.springframework.data.mongodb.core.query.Update` instance.
It can be used as an alternative to abov described post-processing procedure and it modifies those entities that were fetched from the collection, so they won't be pulled from the collection again on the next polling cycle (assuming the update changes some value used in the query).
It is still recommended to use transactions to achieve execution isolation and data consistency, when several instances of the `MongoDbMessageSource` for the same collection are used in the cluster.
[[mongodb-change-stream-channel-adapter]]
=== MongoDB Change Stream Inbound Channel Adapter
@@ -415,82 +418,10 @@ It allows you query a database by sending a message to its request channel.
The gateway then send the response to the reply channel.
You can use the message payload and headers to specify the query and the collection name, as the following example shows:
====
[source,xml]
----
<int-mongodb:outbound-gateway id="gatewayQuery"
mongodb-factory="mongoDbFactory"
mongo-converter="mongoConverter"
query="{firstName: 'Bob'}"
collection-name="myCollection"
request-channel="in"
reply-channel="out"
entity-class="org.springframework.integration.mongodb.test.entity$Person"/>
----
====
You can use the following attributes with a MongoDB outbound Gateway:
* `collection-name` or `collection-name-expression`: Identifies the name of the MongoDB collection to use.
* `mongo-converter`: Reference to an instance of `o.s.data.mongodb.core.convert.MongoConverter` that assists with converting a raw Java object to a JSON document representation.
* `mongodb-factory`: Reference to an instance of `o.s.data.mongodb.MongoDbFactory`.
* `mongo-template`: Reference to an instance of `o.s.data.mongodb.core.MongoTemplate`.
NOTE: you can not set both `mongo-template` and `mongodb-factory`.
* `entity-class`: The fully qualified name of the entity class to be passed to the `find(..)` and `findOne(..)` methods in MongoTemplate.
If this attribute is not provided, the default value is `org.bson.Document`.
* `query` or `query-expression`: Specifies the MongoDB query.
See the https://www.mongodb.org/display/DOCS/Querying[MongoDB documentation] for more query samples.
* `collection-callback`: Reference to an instance of `org.springframework.data.mongodb.core.CollectionCallback`.
Preferable an instance of `o.s.i.mongodb.outbound.MessageCollectionCallback` since 5.0.11 with the request message context.
See its Javadocs for more information.
NOTE: You can not have both `collection-callback` and any of the query attributes.
==== Configuring with Java Configuration
The following Spring Boot application shows an example of how to configure the outbound gateway with Java configuration:
====
[source, java]
----
@SpringBootApplication
public class MongoDbJavaApplication {
public static void main(String[] args) {
new SpringApplicationBuilder(MongoDbJavaApplication.class)
.web(false)
.run(args);
}
@Autowired
private MongoDbFactory mongoDbFactory;
@Bean
@ServiceActivator(inputChannel = "requestChannel")
public MessageHandler mongoDbOutboundGateway() {
MongoDbOutboundGateway gateway = new MongoDbOutboundGateway(this.mongoDbFactory);
gateway.setCollectionNameExpressionString("'myCollection'");
gateway.setQueryExpressionString("'{''name'' : ''Bob''}'");
gateway.setExpectSingleResult(true);
gateway.setEntityClass(Person.class);
gateway.setOutputChannelName("replyChannel");
return gateway;
}
@Bean
@ServiceActivator(inputChannel = "replyChannel")
public MessageHandler handler() {
return message -> System.out.println(message.getPayload());
}
}
----
====
==== Configuring with the Java DSL
The following Spring Boot application show an example of how to configure the outbound gateway with the Java DSL:
====
[source, java]
[source, java, role="primary"]
.Java DSL
----
@SpringBootApplication
public class MongoDbJavaApplication {
@@ -525,8 +456,100 @@ public class MongoDbJavaApplication {
}
----
[source, kotlin, role="secondary"]
.Kotlin DSL
----
class MongoDbKotlinApplication {
fun main(args: Array<String>) = runApplication<MongoDbKotlinApplication>(*args)
@Autowired
lateinit var mongoDbFactory: MongoDatabaseFactory;
@Autowired
lateinit var mongoConverter: MongoConverter;
@Bean
fun gatewaySingleQueryFlow() =
integrationFlow {
handle(queryOutboundGateway())
channel { queue("retrieveResults") }
}
private fun queryOutboundGateway(): MongoDbOutboundGatewaySpec {
return MongoDb.outboundGateway(this.mongoDbFactory, this.mongoConverter)
.query("{name : 'Bob'}")
.collectionNameFunction<Any> { m -> m.headers["collection"] as String }
.expectSingleResult(true)
.entityClass(Person::class.java)
}
}
----
[source, java, role="secondary"]
.Java
----
@SpringBootApplication
public class MongoDbJavaApplication {
public static void main(String[] args) {
new SpringApplicationBuilder(MongoDbJavaApplication.class)
.web(false)
.run(args);
}
@Autowired
private MongoDbFactory mongoDbFactory;
@Bean
@ServiceActivator(inputChannel = "requestChannel")
public MessageHandler mongoDbOutboundGateway() {
MongoDbOutboundGateway gateway = new MongoDbOutboundGateway(this.mongoDbFactory);
gateway.setCollectionNameExpressionString("'myCollection'");
gateway.setQueryExpressionString("'{''name'' : ''Bob''}'");
gateway.setExpectSingleResult(true);
gateway.setEntityClass(Person.class);
gateway.setOutputChannelName("replyChannel");
return gateway;
}
@Bean
@ServiceActivator(inputChannel = "replyChannel")
public MessageHandler handler() {
return message -> System.out.println(message.getPayload());
}
}
----
[source, xml, role="secondary"]
.XML
----
<int-mongodb:outbound-gateway id="gatewayQuery"
mongodb-factory="mongoDbFactory"
mongo-converter="mongoConverter"
query="{firstName: 'Bob'}"
collection-name="myCollection"
request-channel="in"
reply-channel="out"
entity-class="org.springframework.integration.mongodb.test.entity$Person"/>
----
====
You can use the following attributes with a MongoDB outbound Gateway:
* `collection-name` or `collection-name-expression`: Identifies the name of the MongoDB collection to use.
* `mongo-converter`: Reference to an instance of `o.s.data.mongodb.core.convert.MongoConverter` that assists with converting a raw Java object to a JSON document representation.
* `mongodb-factory`: Reference to an instance of `o.s.data.mongodb.MongoDbFactory`.
* `mongo-template`: Reference to an instance of `o.s.data.mongodb.core.MongoTemplate`.
NOTE: you can not set both `mongo-template` and `mongodb-factory`.
* `entity-class`: The fully qualified name of the entity class to be passed to the `find(..)` and `findOne(..)` methods in MongoTemplate.
If this attribute is not provided, the default value is `org.bson.Document`.
* `query` or `query-expression`: Specifies the MongoDB query.
See the https://www.mongodb.org/display/DOCS/Querying[MongoDB documentation] for more query samples.
* `collection-callback`: Reference to an instance of `org.springframework.data.mongodb.core.CollectionCallback`.
Preferable an instance of `o.s.i.mongodb.outbound.MessageCollectionCallback` since 5.0.11 with the request message context.
See its Javadocs for more information.
NOTE: You can not have both `collection-callback` and any of the query attributes.
As an alternate to the `query` and `query-expression` properties, you can specify other database operations by using the `collectionCallback` property as a reference to the `MessageCollectionCallback` functional interface implementation.
The following example specifies a count operation:
@@ -537,7 +560,7 @@ private MongoDbOutboundGatewaySpec collectionCallbackOutboundGateway() {
return MongoDb.outboundGateway(this.mongoDbFactory, this.mongoConverter)
.collectionCallback((collection, requestMessage) -> collection.count())
.collectionName("myCollection");
}
}
----
====
@@ -590,3 +613,7 @@ public IntegrationFlow reactiveMongoDbFlow(ReactiveMongoDatabaseFactory mongoDbF
}
----
====
Starting with version 5.5, the `ReactiveMongoDbMessageSource` can be configured with an `updateExpression`.
It has the same functionality as the blocking `MongoDbMessageSource`.
See <<mongodb-inbound-channel-adapter>> and `AbstractMongoDbMessageSourceSpec` JavaDocs for more information.

8
src/reference/asciidoc/whats-new.adoc
View File

@@ -57,3 +57,11 @@ This is to solve a problem where changes deep in the directory tree were not det
In addition, `forRecursion=true` causes the full path to files to be used as the metadata store keys; this solves a problem where the filter did not work properly if a file with the same name appears multiple times in different directories.
IMPORTANT: This means that existing keys in a persistent metadata store will not be found for files beneath the top level directory.
For this reason, the property is `false` by default; this may change in a future release.
[[x5.5-mongodb]]
==== MongoDb Changes
The `MongoDbMessageSourceSpec` was added into MongoDd Java DSL.
An `update` option is now exposed on both the `MongoDbMessageSource` and `ReactiveMongoDbMessageSource` implementations.
See <<./mongodb.adoc#mongodb,MongoDb Support>> for more information.