diff --git a/build.gradle b/build.gradle index 5d2340e1e2..58d40d82b8 100644 --- a/build.gradle +++ b/build.gradle @@ -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' diff --git a/spring-integration-mongodb/src/main/java/org/springframework/integration/mongodb/config/MongoDbInboundChannelAdapterParser.java b/spring-integration-mongodb/src/main/java/org/springframework/integration/mongodb/config/MongoDbInboundChannelAdapterParser.java index 3a53dfe158..3f49688497 100644 --- a/spring-integration-mongodb/src/main/java/org/springframework/integration/mongodb/config/MongoDbInboundChannelAdapterParser.java +++ b/spring-integration-mongodb/src/main/java/org/springframework/integration/mongodb/config/MongoDbInboundChannelAdapterParser.java @@ -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"); diff --git a/spring-integration-mongodb/src/main/java/org/springframework/integration/mongodb/dsl/AbstractMongoDbMessageSourceSpec.java b/spring-integration-mongodb/src/main/java/org/springframework/integration/mongodb/dsl/AbstractMongoDbMessageSourceSpec.java new file mode 100644 index 0000000000..ac0128c5ef --- /dev/null +++ b/spring-integration-mongodb/src/main/java/org/springframework/integration/mongodb/dsl/AbstractMongoDbMessageSourceSpec.java @@ -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, + H extends AbstractMongoDbMessageSource> + extends MessageSourceSpec { + + /** + * 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 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 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(); + } + +} diff --git a/spring-integration-mongodb/src/main/java/org/springframework/integration/mongodb/dsl/MongoDb.java b/spring-integration-mongodb/src/main/java/org/springframework/integration/mongodb/dsl/MongoDb.java index c46de2c582..2ea1ab06f4 100644 --- a/spring-integration-mongodb/src/main/java/org/springframework/integration/mongodb/dsl/MongoDb.java +++ b/spring-integration-mongodb/src/main/java/org/springframework/integration/mongodb/dsl/MongoDb.java @@ -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() { } diff --git a/spring-integration-mongodb/src/main/java/org/springframework/integration/mongodb/dsl/MongoDbMessageSourceSpec.java b/spring-integration-mongodb/src/main/java/org/springframework/integration/mongodb/dsl/MongoDbMessageSourceSpec.java new file mode 100644 index 0000000000..08adb85355 --- /dev/null +++ b/spring-integration-mongodb/src/main/java/org/springframework/integration/mongodb/dsl/MongoDbMessageSourceSpec.java @@ -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 { + + protected MongoDbMessageSourceSpec(MongoDatabaseFactory mongoDatabaseFactory, Expression queryExpression) { + this.target = new MongoDbMessageSource(mongoDatabaseFactory, queryExpression); + } + + protected MongoDbMessageSourceSpec(MongoOperations mongoTemplate, Expression queryExpression) { + this.target = new MongoDbMessageSource(mongoTemplate, queryExpression); + } + +} diff --git a/spring-integration-mongodb/src/main/java/org/springframework/integration/mongodb/dsl/ReactiveMongoDbMessageSourceSpec.java b/spring-integration-mongodb/src/main/java/org/springframework/integration/mongodb/dsl/ReactiveMongoDbMessageSourceSpec.java index 174686fc56..67a5cf6677 100644 --- a/spring-integration-mongodb/src/main/java/org/springframework/integration/mongodb/dsl/ReactiveMongoDbMessageSourceSpec.java +++ b/spring-integration-mongodb/src/main/java/org/springframework/integration/mongodb/dsl/ReactiveMongoDbMessageSourceSpec.java @@ -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 { + extends AbstractMongoDbMessageSourceSpec { 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 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; - } - } diff --git a/spring-integration-mongodb/src/main/java/org/springframework/integration/mongodb/inbound/AbstractMongoDbMessageSource.java b/spring-integration-mongodb/src/main/java/org/springframework/integration/mongodb/inbound/AbstractMongoDbMessageSource.java new file mode 100644 index 0000000000..797fd69653 --- /dev/null +++ b/spring-integration-mongodb/src/main/java/org/springframework/integration/mongodb/inbound/AbstractMongoDbMessageSource.java @@ -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 The payload type. + * + * @author Artem Bilan + * + * @since 5.5 + */ +public abstract class AbstractMongoDbMessageSource extends AbstractMessageSource + 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 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 idForEntity(Object entity) { + if (entity instanceof String) { + return idFieldFromMap(Document.parse(entity.toString())); + } + if (entity instanceof Map) { + return idFieldFromMap((Map) entity); + } + + MappingContext, ?> 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 idFieldFromMap(Map map) { + return Pair.of(ID_FIELD, map.get(ID_FIELD)); + } + +} diff --git a/spring-integration-mongodb/src/main/java/org/springframework/integration/mongodb/inbound/MongoDbMessageSource.java b/spring-integration-mongodb/src/main/java/org/springframework/integration/mongodb/inbound/MongoDbMessageSource.java index ced393184d..28e1213f16 100644 --- a/spring-integration-mongodb/src/main/java/org/springframework/integration/mongodb/inbound/MongoDbMessageSource.java +++ b/spring-integration-mongodb/src/main/java/org/springframework/integration/mongodb/inbound/MongoDbMessageSource.java @@ -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 { +public class MongoDbMessageSource extends AbstractMongoDbMessageSource { @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 { * @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 { @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 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 { 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 idPair = idForEntity(result); + Query query = new Query(Criteria.where(idPair.getFirst()).is(idPair.getSecond())); + this.mongoTemplate.updateFirst(query, update, collectionName); + } + } + } + + } diff --git a/spring-integration-mongodb/src/main/java/org/springframework/integration/mongodb/inbound/ReactiveMongoDbMessageSource.java b/spring-integration-mongodb/src/main/java/org/springframework/integration/mongodb/inbound/ReactiveMongoDbMessageSource.java index 5e8e1386f4..f4637877c9 100644 --- a/spring-integration-mongodb/src/main/java/org/springframework/integration/mongodb/inbound/ReactiveMongoDbMessageSource.java +++ b/spring-integration-mongodb/src/main/java/org/springframework/integration/mongodb/inbound/ReactiveMongoDbMessageSource.java @@ -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> - implements ApplicationContextAware { +public class ReactiveMongoDbMessageSource extends AbstractMongoDbMessageSource> { @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 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 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 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)) + ); + } + } diff --git a/spring-integration-mongodb/src/main/resources/org/springframework/integration/mongodb/config/spring-integration-mongodb.xsd b/spring-integration-mongodb/src/main/resources/org/springframework/integration/mongodb/config/spring-integration-mongodb.xsd index 5c810f164d..59b8649ea6 100644 --- a/spring-integration-mongodb/src/main/resources/org/springframework/integration/mongodb/config/spring-integration-mongodb.xsd +++ b/spring-integration-mongodb/src/main/resources/org/springframework/integration/mongodb/config/spring-integration-mongodb.xsd @@ -53,6 +53,28 @@ + + + + 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. + + + + + + + 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. + + + diff --git a/spring-integration-mongodb/src/test/java/org/springframework/integration/mongodb/config/MongoDbInboundChannelAdapterParserTests-context.xml b/spring-integration-mongodb/src/test/java/org/springframework/integration/mongodb/config/MongoDbInboundChannelAdapterParserTests-context.xml index df71797cb7..7c9c896850 100644 --- a/spring-integration-mongodb/src/test/java/org/springframework/integration/mongodb/config/MongoDbInboundChannelAdapterParserTests-context.xml +++ b/spring-integration-mongodb/src/test/java/org/springframework/integration/mongodb/config/MongoDbInboundChannelAdapterParserTests-context.xml @@ -29,6 +29,7 @@ diff --git a/spring-integration-mongodb/src/test/java/org/springframework/integration/mongodb/config/MongoDbInboundChannelAdapterParserTests.java b/spring-integration-mongodb/src/test/java/org/springframework/integration/mongodb/config/MongoDbInboundChannelAdapterParserTests.java index 8004db2c90..325f63d7bb 100644 --- a/spring-integration-mongodb/src/test/java/org/springframework/integration/mongodb/config/MongoDbInboundChannelAdapterParserTests.java +++ b/spring-integration-mongodb/src/test/java/org/springframework/integration/mongodb/config/MongoDbInboundChannelAdapterParserTests.java @@ -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) { diff --git a/spring-integration-mongodb/src/test/java/org/springframework/integration/mongodb/inbound/MongoDbMessageSourceTests.java b/spring-integration-mongodb/src/test/java/org/springframework/integration/mongodb/inbound/MongoDbMessageSourceTests.java index 40a6a62762..84b497dc63 100644 --- a/spring-integration-mongodb/src/test/java/org/springframework/integration/mongodb/inbound/MongoDbMessageSourceTests.java +++ b/spring-integration-mongodb/src/test/java/org/springframework/integration/mongodb/inbound/MongoDbMessageSourceTests.java @@ -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 persons = (List) messageSource.receive().getPayload(); - assertThat(persons.size()).isEqualTo(3); + assertThat(persons).hasSize(3); verify(converter, times(3)).read((Class) Mockito.any(), Mockito.any(Bson.class)); + + assertThat(messageSource.receive()).isNull(); + + assertThat(template.find(new BasicQuery("{'address.state' : 'NJ'}"), Object.class, "data")) + .hasSize(3); } @Test diff --git a/spring-integration-mongodb/src/test/java/org/springframework/integration/mongodb/inbound/ReactiveMongoDbMessageSourceTests.java b/spring-integration-mongodb/src/test/java/org/springframework/integration/mongodb/inbound/ReactiveMongoDbMessageSourceTests.java index 4155aa01ed..1ccbcd436f 100644 --- a/spring-integration-mongodb/src/test/java/org/springframework/integration/mongodb/inbound/ReactiveMongoDbMessageSourceTests.java +++ b/spring-integration-mongodb/src/test/java/org/springframework/integration/mongodb/inbound/ReactiveMongoDbMessageSourceTests.java @@ -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(); diff --git a/spring-integration-mongodb/src/test/java/org/springframework/integration/mongodb/rules/MongoDbAvailableTests.java b/spring-integration-mongodb/src/test/java/org/springframework/integration/mongodb/rules/MongoDbAvailableTests.java index cfc5181304..8438d2ebe4 100644 --- a/spring-integration-mongodb/src/test/java/org/springframework/integration/mongodb/rules/MongoDbAvailableTests.java +++ b/spring-integration-mongodb/src/test/java/org/springframework/integration/mongodb/rules/MongoDbAvailableTests.java @@ -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; diff --git a/src/reference/asciidoc/mongodb.adoc b/src/reference/asciidoc/mongodb.adoc index 266ba22cbb..9ba0de3f8a 100644 --- a/src/reference/asciidoc/mongodb.adoc +++ b/src/reference/asciidoc/mongodb.adoc @@ -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"] ---- org.springframework.integration @@ -15,9 +16,8 @@ You need to include this dependency into your project: {project-version} ---- - +[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"] ---- org.mongodb mongodb-driver-reactivestreams - + ---- - +[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"] ---- org.mongodb mongodb-driver-sync - + ---- - +[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 ---- @@ -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 ---- @@ -302,7 +301,7 @@ You can do so by using that transaction synchronization feature Spring Integrati + channel="someChannel"/> @@ -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] ----- - ----- -==== - -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) = runApplication(*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 { 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 +---- + +---- ==== +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 <> and `AbstractMongoDbMessageSourceSpec` JavaDocs for more information. diff --git a/src/reference/asciidoc/whats-new.adoc b/src/reference/asciidoc/whats-new.adoc index c04ef1645b..500c478860 100644 --- a/src/reference/asciidoc/whats-new.adoc +++ b/src/reference/asciidoc/whats-new.adoc @@ -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.