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

DATAMONGO-2365 - Extract common functionality to centralized component.

Browse Source 
Share common code paths between reactive and imperative implementation.

Original pull request: #828.
This commit is contained in:
Christoph Strobl
2020-01-21 16:14:08 +01:00
committed by Mark Paluch
parent 0bc0fff24e
commit c7f9274480
3 changed files with 781 additions and 305 deletions
Download Patch File Download Diff File Expand all files Collapse all files

217
spring-data-mongodb/src/main/java/org/springframework/data/mongodb/core/MongoTemplate.java
View File

@@ -30,9 +30,7 @@ import java.util.*;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;
import org.bson.BsonValue;
import org.bson.Document;
import org.bson.codecs.Codec;
import org.bson.conversions.Bson;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -54,8 +52,6 @@ import org.springframework.data.geo.Distance;
import org.springframework.data.geo.GeoResult;
import org.springframework.data.geo.GeoResults;
import org.springframework.data.geo.Metric;
import org.springframework.data.mapping.PropertyPath;
import org.springframework.data.mapping.PropertyReferenceException;
import org.springframework.data.mapping.callback.EntityCallbacks;
import org.springframework.data.mapping.context.MappingContext;
import org.springframework.data.mongodb.MongoDatabaseUtils;
@@ -64,14 +60,15 @@ import org.springframework.data.mongodb.SessionSynchronization;
import org.springframework.data.mongodb.core.BulkOperations.BulkMode;
import org.springframework.data.mongodb.core.DefaultBulkOperations.BulkOperationContext;
import org.springframework.data.mongodb.core.EntityOperations.AdaptibleEntity;
import org.springframework.data.mongodb.core.MappedDocument.MappedUpdate;
import org.springframework.data.mongodb.core.QueryOperations.CountContext;
import org.springframework.data.mongodb.core.QueryOperations.DeleteContext;
import org.springframework.data.mongodb.core.QueryOperations.DistinctQueryContext;
import org.springframework.data.mongodb.core.QueryOperations.QueryContext;
import org.springframework.data.mongodb.core.QueryOperations.UpdateContext;
import org.springframework.data.mongodb.core.aggregation.Aggregation;
import org.springframework.data.mongodb.core.aggregation.AggregationOperationContext;
import org.springframework.data.mongodb.core.aggregation.AggregationOptions;
import org.springframework.data.mongodb.core.aggregation.AggregationResults;
import org.springframework.data.mongodb.core.aggregation.AggregationUpdate;
import org.springframework.data.mongodb.core.aggregation.Fields;
import org.springframework.data.mongodb.core.aggregation.RelaxedTypeBasedAggregationOperationContext;
import org.springframework.data.mongodb.core.aggregation.TypeBasedAggregationOperationContext;
import org.springframework.data.mongodb.core.aggregation.TypedAggregation;
import org.springframework.data.mongodb.core.convert.DbRefResolver;
@@ -196,6 +193,7 @@ public class MongoTemplate implements MongoOperations, ApplicationContextAware,
private final SpelAwareProxyProjectionFactory projectionFactory;
private final EntityOperations operations;
private final PropertyOperations propertyOperations;
private final QueryOperations queryOperations;
private @Nullable WriteConcern writeConcern;
private WriteConcernResolver writeConcernResolver = DefaultWriteConcernResolver.INSTANCE;
@@ -247,6 +245,7 @@ public class MongoTemplate implements MongoOperations, ApplicationContextAware,
this.projectionFactory = new SpelAwareProxyProjectionFactory();
this.operations = new EntityOperations(this.mongoConverter.getMappingContext());
this.propertyOperations = new PropertyOperations(this.mongoConverter.getMappingContext());
this.queryOperations = new QueryOperations(queryMapper, updateMapper, operations, mongoDbFactory);
// We always have a mapping context in the converter, whether it's a simple one or not
mappingContext = this.mongoConverter.getMappingContext();
@@ -285,6 +284,7 @@ public class MongoTemplate implements MongoOperations, ApplicationContextAware,
this.mappingContext = that.mappingContext;
this.operations = that.operations;
this.propertyOperations = that.propertyOperations;
this.queryOperations = that.queryOperations;
}
/**
@@ -827,10 +827,11 @@ public class MongoTemplate implements MongoOperations, ApplicationContextAware,
}
Assert.notNull(collectionName, "CollectionName must not be null!");
Document mappedQuery = queryMapper.getMappedObject(query.getQueryObject(), getPersistentEntity(entityClass));
QueryContext queryContext = queryOperations.createQueryContext(query);
Document mappedQuery = queryContext.getMappedQuery(entityClass, this::getPersistentEntity);
return execute(collectionName, new ExistsCallback(mappedQuery,
operations.forType(entityClass).getCollation(query).map(Collation::toMongoCollation).orElse(null)));
return execute(collectionName,
new ExistsCallback(mappedQuery, queryContext.getCollation(entityClass).orElse(null)));
}
// Find methods that take a Query to express the query and that return a List of objects.
@@ -903,13 +904,11 @@ public class MongoTemplate implements MongoOperations, ApplicationContextAware,
Assert.notNull(resultClass, "ResultClass must not be null!");
MongoPersistentEntity<?> entity = entityClass != Object.class ? getPersistentEntity(entityClass) : null;
DistinctQueryContext distinctQueryContext = queryOperations.distincQueryContext(query, field);
Document mappedQuery = queryMapper.getMappedObject(query.getQueryObject(), entity);
String mappedFieldName = queryMapper.getMappedFields(new Document(field, 1), entity).keySet().iterator().next();
Class<T> mongoDriverCompatibleType = getMongoDbFactory().getCodecFor(resultClass) //
.map(Codec::getEncoderClass) //
.orElse((Class<T>) BsonValue.class);
Document mappedQuery = distinctQueryContext.getMappedQuery(entity);
String mappedFieldName = distinctQueryContext.getMappedFieldName(entity);
Class<T> mongoDriverCompatibleType = distinctQueryContext.getDriverCompatibleClass(resultClass);
MongoIterable<?> result = execute(collectionName, (collection) -> {
@@ -924,12 +923,9 @@ public class MongoTemplate implements MongoOperations, ApplicationContextAware,
}
DistinctIterable<T> iterable = collection.distinct(mappedFieldName, mappedQuery, mongoDriverCompatibleType);
distinctQueryContext.applyCollation(entityClass, iterable::collation);
return operations.forType(entityClass) //
.getCollation(query) //
.map(Collation::toMongoCollation) //
.map(iterable::collation) //
.orElse(iterable);
return iterable;
});
if (resultClass == Object.class || mongoDriverCompatibleType != resultClass) {
@@ -938,7 +934,7 @@ public class MongoTemplate implements MongoOperations, ApplicationContextAware,
DefaultDbRefResolver dbRefResolver = new DefaultDbRefResolver(mongoDbFactory);
result = result.map((source) -> converter.mapValueToTargetType(source,
getMostSpecificConversionTargetType(resultClass, entityClass, field), dbRefResolver));
distinctQueryContext.getMostSpecificConversionTargetType(resultClass, entityClass), dbRefResolver));
}
try {
@@ -948,32 +944,6 @@ public class MongoTemplate implements MongoOperations, ApplicationContextAware,
}
}
/**
* @param userType must not be {@literal null}.
* @param domainType must not be {@literal null}.
* @param field must not be {@literal null}.
* @return the most specific conversion target type depending on user preference and domain type property.
* @since 2.1
*/
private static Class<?> getMostSpecificConversionTargetType(Class<?> userType, Class<?> domainType, String field) {
Class<?> conversionTargetType = userType;
try {
Class<?> propertyType = PropertyPath.from(field, domainType).getLeafProperty().getLeafType();
// use the more specific type but favor UserType over property one
if (ClassUtils.isAssignable(userType, propertyType)) {
conversionTargetType = propertyType;
}
} catch (PropertyReferenceException e) {
// just don't care about it as we default to Object.class anyway.
}
return conversionTargetType;
}
@Override
public <T> GeoResults<T> geoNear(NearQuery near, Class<T> entityClass) {
return geoNear(near, entityClass, getCollectionName(entityClass));
@@ -1088,10 +1058,11 @@ public class MongoTemplate implements MongoOperations, ApplicationContextAware,
Assert.isTrue(query.getSkip() <= 0, "Query must not define skip.");
MongoPersistentEntity<?> entity = mappingContext.getPersistentEntity(entityType);
QueryContext queryContext = queryOperations.createQueryContext(query);
Document mappedQuery = queryMapper.getMappedObject(query.getQueryObject(), entity);
Document mappedFields = queryMapper.getMappedFields(query.getFieldsObject(), entity);
Document mappedSort = queryMapper.getMappedSort(query.getSortObject(), entity);
Document mappedQuery = queryContext.getMappedQuery(entity);
Document mappedFields = queryContext.getMappedFields(entity);
Document mappedSort = queryContext.getMappedSort(entity);
replacement = maybeCallBeforeConvert(replacement, collectionName);
Document mappedReplacement = operations.forEntity(replacement).toMappedDocument(this.mongoConverter).getDocument();
@@ -1100,8 +1071,7 @@ public class MongoTemplate implements MongoOperations, ApplicationContextAware,
maybeCallBeforeSave(replacement, mappedReplacement, collectionName);
return doFindAndReplace(collectionName, mappedQuery, mappedFields, mappedSort,
operations.forType(entityType).getCollation(query).map(Collation::toMongoCollation).orElse(null), entityType,
mappedReplacement, options, resultType);
queryContext.getCollation(entityType).orElse(null), entityType, mappedReplacement, options, resultType);
}
// Find methods that take a Query to express the query and that return a single object that is also removed from the
@@ -1147,29 +1117,12 @@ public class MongoTemplate implements MongoOperations, ApplicationContextAware,
Assert.notNull(query, "Query must not be null!");
Assert.hasText(collectionName, "Collection name must not be null or empty!");
CountOptions options = new CountOptions();
query.getCollation().map(Collation::toMongoCollation).ifPresent(options::collation);
CountContext countContext = queryOperations.countQueryContext(query);
if (query.getLimit() > 0) {
options.limit(query.getLimit());
}
if (query.getSkip() > 0) {
options.skip((int) query.getSkip());
}
if (StringUtils.hasText(query.getHint())) {
CountOptions options = countContext.getCountOptions(entityClass);
Document mappedQuery = countContext.getMappedQuery(entityClass, mappingContext::getPersistentEntity);
String hint = query.getHint();
if(BsonUtils.isJsonDocument(hint)) {
options = options.hint(BsonUtils.parse(hint, mongoDbFactory));
} else {
options = options.hintString(hint);
}
}
Document document = queryMapper.getMappedObject(query.getQueryObject(),
Optional.ofNullable(entityClass).map(it -> mappingContext.getPersistentEntity(entityClass)));
return doCount(collectionName, document, options);
return doCount(collectionName, mappedQuery, options);
}
@SuppressWarnings("ConstantConditions")
@@ -1617,33 +1570,20 @@ public class MongoTemplate implements MongoOperations, ApplicationContextAware,
}
MongoPersistentEntity<?> entity = entityClass == null ? null : getPersistentEntity(entityClass);
increaseVersionForUpdateIfNecessary(entity, update);
UpdateOptions opts = new UpdateOptions();
opts.upsert(upsert);
UpdateContext updateContext = multi ? queryOperations.updateContext(update, query, upsert)
: queryOperations.updateSingleContext(update, query, upsert);
updateContext.increaseVersionForUpdateIfNecessary(entity);
if (update.hasArrayFilters()) {
opts.arrayFilters(update.getArrayFilters().stream().map(ArrayFilter::asDocument).collect(Collectors.toList()));
}
Document queryObj = updateContext.getMappedQuery(entity);
UpdateOptions opts = updateContext.getUpdateOptions(entityClass);
Document queryObj = new Document();
if (updateContext.isAggregationUpdate()) {
if (query != null) {
queryObj.putAll(queryMapper.getMappedObject(query.getQueryObject(), entity));
}
if (multi && update.isIsolated() && !queryObj.containsKey("$isolated")) {
queryObj.put("$isolated", 1);
}
if (update instanceof AggregationUpdate) {
AggregationOperationContext context = entityClass != null
? new RelaxedTypeBasedAggregationOperationContext(entityClass, mappingContext, queryMapper)
: Aggregation.DEFAULT_CONTEXT;
List<Document> pipeline = new AggregationUtil(queryMapper, mappingContext)
.createPipeline((AggregationUpdate) update, context);
List<Document> pipeline = updateContext.getUpdatePipeline(entityClass);
MongoAction mongoAction = new MongoAction(writeConcern, MongoActionOperation.UPDATE, collectionName, entityClass,
update.getUpdateObject(), queryObj);
WriteConcern writeConcernToUse = prepareWriteConcern(mongoAction);
return execute(collectionName, collection -> {
@@ -1652,43 +1592,29 @@ public class MongoTemplate implements MongoOperations, ApplicationContextAware,
serializeToJsonSafely(queryObj), serializeToJsonSafely(pipeline), collectionName);
}
MongoAction mongoAction = new MongoAction(writeConcern, MongoActionOperation.UPDATE, collectionName,
entityClass, update.getUpdateObject(), queryObj);
WriteConcern writeConcernToUse = prepareWriteConcern(mongoAction);
collection = writeConcernToUse != null ? collection.withWriteConcern(writeConcernToUse) : collection;
return multi ? collection.updateMany(queryObj, pipeline, opts) : collection.updateOne(queryObj, pipeline, opts);
});
}
Document updateObj = updateContext.getMappedUpdate(entity);
MongoAction mongoAction = new MongoAction(writeConcern, MongoActionOperation.UPDATE, collectionName, entityClass,
updateObj, queryObj);
WriteConcern writeConcernToUse = prepareWriteConcern(mongoAction);
return execute(collectionName, collection -> {
operations.forType(entityClass) //
.getCollation(query) //
.map(Collation::toMongoCollation) //
.ifPresent(opts::collation);
Document updateObj = update instanceof MappedUpdate ? update.getUpdateObject()
: updateMapper.getMappedObject(update.getUpdateObject(), entity);
if (LOGGER.isDebugEnabled()) {
LOGGER.debug("Calling update using query: {} and update: {} in collection: {}", serializeToJsonSafely(queryObj),
serializeToJsonSafely(updateObj), collectionName);
}
MongoAction mongoAction = new MongoAction(writeConcern, MongoActionOperation.UPDATE, collectionName, entityClass,
updateObj, queryObj);
WriteConcern writeConcernToUse = prepareWriteConcern(mongoAction);
collection = writeConcernToUse != null ? collection.withWriteConcern(writeConcernToUse) : collection;
if (!UpdateMapper.isUpdateObject(updateObj)) {
ReplaceOptions replaceOptions = new ReplaceOptions();
replaceOptions.collation(opts.getCollation());
replaceOptions.upsert(opts.isUpsert());
ReplaceOptions replaceOptions = updateContext.getReplaceOptions(entityClass);
return collection.replaceOne(queryObj, updateObj, replaceOptions);
} else {
return multi ? collection.updateMany(queryObj, updateObj, opts)
@@ -1697,17 +1623,6 @@ public class MongoTemplate implements MongoOperations, ApplicationContextAware,
});
}
private void increaseVersionForUpdateIfNecessary(@Nullable MongoPersistentEntity<?> persistentEntity,
UpdateDefinition update) {
if (persistentEntity != null && persistentEntity.hasVersionProperty()) {
String versionFieldName = persistentEntity.getRequiredVersionProperty().getFieldName();
if (!update.modifies(versionFieldName)) {
update.inc(versionFieldName);
}
}
}
@Override
public DeleteResult remove(Object object) {
@@ -1752,7 +1667,15 @@ public class MongoTemplate implements MongoOperations, ApplicationContextAware,
Assert.hasText(collectionName, "Collection name must not be null or empty!");
MongoPersistentEntity<?> entity = getPersistentEntity(entityClass);
Document queryObject = queryMapper.getMappedObject(query.getQueryObject(), entity);
DeleteContext deleteContext = multi ? queryOperations.deleteQueryContext(query) : queryOperations.deleteSingleContext(query);
Document queryObject = deleteContext.getMappedQuery(entity);
DeleteOptions options = deleteContext.getDeleteOptions(entityClass);
MongoAction mongoAction = new MongoAction(writeConcern, MongoActionOperation.REMOVE, collectionName, entityClass,
null, queryObject);
WriteConcern writeConcernToUse = prepareWriteConcern(mongoAction);
return execute(collectionName, collection -> {
@@ -1760,18 +1683,6 @@ public class MongoTemplate implements MongoOperations, ApplicationContextAware,
Document removeQuery = queryObject;
DeleteOptions options = new DeleteOptions();
operations.forType(entityClass) //
.getCollation(query) //
.map(Collation::toMongoCollation) //
.ifPresent(options::collation);
MongoAction mongoAction = new MongoAction(writeConcern, MongoActionOperation.REMOVE, collectionName, entityClass,
null, queryObject);
WriteConcern writeConcernToUse = prepareWriteConcern(mongoAction);
if (LOGGER.isDebugEnabled()) {
LOGGER.debug("Remove using query: {} in collection: {}.",
new Object[] { serializeToJsonSafely(removeQuery), collectionName });
@@ -2479,6 +2390,7 @@ public class MongoTemplate implements MongoOperations, ApplicationContextAware,
Class<T> entityClass) {
MongoPersistentEntity<?> entity = mappingContext.getPersistentEntity(entityClass);
Document mappedQuery = queryMapper.getMappedObject(query, entity);
Document mappedFields = queryMapper.getMappedObject(fields, entity);
@@ -2670,21 +2582,12 @@ public class MongoTemplate implements MongoOperations, ApplicationContextAware,
MongoPersistentEntity<?> entity = mappingContext.getPersistentEntity(entityClass);
increaseVersionForUpdateIfNecessary(entity, update);
UpdateContext updateContext = queryOperations.updateSingleContext(update, query, false);
updateContext.increaseVersionForUpdateIfNecessary(entity);
Document mappedQuery = queryMapper.getMappedObject(query, entity);
Object mappedUpdate;
if (update instanceof AggregationUpdate) {
AggregationOperationContext context = entityClass != null
? new RelaxedTypeBasedAggregationOperationContext(entityClass, mappingContext, queryMapper)
: Aggregation.DEFAULT_CONTEXT;
mappedUpdate = new AggregationUtil(queryMapper, mappingContext).createPipeline((Aggregation) update, context);
} else {
mappedUpdate = updateMapper.getMappedObject(update.getUpdateObject(), entity);
}
Document mappedQuery = updateContext.getMappedQuery(entity);
Object mappedUpdate = updateContext.isAggregationUpdate() ? updateContext.getUpdatePipeline(entityClass)
: updateContext.getMappedUpdate(entity);
if (LOGGER.isDebugEnabled()) {
LOGGER.debug(
@@ -3286,7 +3189,7 @@ public class MongoTemplate implements MongoOperations, ApplicationContextAware,
String hint = query.getHint();
if(BsonUtils.isJsonDocument(hint)) {
if (BsonUtils.isJsonDocument(hint)) {
cursorToUse = cursorToUse.hint(BsonUtils.parse(hint, mongoDbFactory));
} else {
cursorToUse = cursorToUse.hintString(hint);

661
spring-data-mongodb/src/main/java/org/springframework/data/mongodb/core/QueryOperations.java Normal file
View File

@@ -0,0 +1,661 @@
/*
* Copyright 2020 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* https://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.data.mongodb.core;
import java.util.List;
import java.util.Optional;
import java.util.function.Consumer;
import java.util.function.Function;
import java.util.stream.Collectors;
import org.bson.BsonValue;
import org.bson.Document;
import org.bson.codecs.Codec;
import org.springframework.data.mapping.PropertyPath;
import org.springframework.data.mapping.PropertyReferenceException;
import org.springframework.data.mapping.context.MappingContext;
import org.springframework.data.mongodb.CodecRegistryProvider;
import org.springframework.data.mongodb.core.MappedDocument.MappedUpdate;
import org.springframework.data.mongodb.core.aggregation.Aggregation;
import org.springframework.data.mongodb.core.aggregation.AggregationOperationContext;
import org.springframework.data.mongodb.core.aggregation.AggregationUpdate;
import org.springframework.data.mongodb.core.aggregation.RelaxedTypeBasedAggregationOperationContext;
import org.springframework.data.mongodb.core.convert.QueryMapper;
import org.springframework.data.mongodb.core.convert.UpdateMapper;
import org.springframework.data.mongodb.core.mapping.MongoPersistentEntity;
import org.springframework.data.mongodb.core.mapping.MongoPersistentProperty;
import org.springframework.data.mongodb.core.query.BasicQuery;
import org.springframework.data.mongodb.core.query.Collation;
import org.springframework.data.mongodb.core.query.Query;
import org.springframework.data.mongodb.core.query.UpdateDefinition;
import org.springframework.data.mongodb.core.query.UpdateDefinition.ArrayFilter;
import org.springframework.data.mongodb.util.BsonUtils;
import org.springframework.lang.Nullable;
import org.springframework.util.ClassUtils;
import org.springframework.util.StringUtils;
import com.mongodb.client.model.CountOptions;
import com.mongodb.client.model.DeleteOptions;
import com.mongodb.client.model.ReplaceOptions;
import com.mongodb.client.model.UpdateOptions;
/**
* {@link QueryOperations} centralizes common operations required before an operation is actually ready to be executed.
* This involves mapping {@link Query queries} into their respective MongoDB representation, computing execution options
* for {@literal count}, {@literal remove}, ... <br />
*
* @author Christoph Strobl
* @since 3.0
*/
class QueryOperations {
private final QueryMapper queryMapper;
private final UpdateMapper updateMapper;
private final EntityOperations entityOperations;
private final CodecRegistryProvider codecRegistryProvider;
private final MappingContext<? extends MongoPersistentEntity<?>, MongoPersistentProperty> mappingContext;
private final AggregationUtil aggregationUtil;
/**
* Create a new instance of {@link QueryOperations}.
*
* @param queryMapper must not be {@literal null}.
* @param updateMapper must not be {@literal null}.
* @param entityOperations must not be {@literal null}.
* @param codecRegistryProvider must not be {@literal null}.
*/
QueryOperations(QueryMapper queryMapper, UpdateMapper updateMapper, EntityOperations entityOperations,
CodecRegistryProvider codecRegistryProvider) {
this.queryMapper = queryMapper;
this.updateMapper = updateMapper;
this.entityOperations = entityOperations;
this.codecRegistryProvider = codecRegistryProvider;
this.mappingContext = queryMapper.getMappingContext();
this.aggregationUtil = new AggregationUtil(queryMapper, mappingContext);
}
/**
* Create a new {@link QueryContext} instance.
*
* @param query must not be {@literal null}.
* @return new instance of {@link QueryContext}.
*/
QueryContext createQueryContext(Query query) {
return new QueryContext(query);
}
/**
* Create a new {@link DistinctQueryContext} instance.
*
* @param query must not be {@literal null}.
* @return new instance of {@link DistinctQueryContext}.
*/
DistinctQueryContext distincQueryContext(Query query, String fieldName) {
return new DistinctQueryContext(query, fieldName);
}
/**
* Create a new {@link CountContext} instance.
*
* @param query must not be {@literal null}.
* @return new instance of {@link CountContext}.
*/
CountContext countQueryContext(Query query) {
return new CountContext(query);
}
/**
* Create a new {@link UpdateContext} instance affecting multiple documents.
*
* @param updateDefinition must not be {@literal null}.
* @param query must not be {@literal null}.
* @param upsert use {@literal true} to insert diff when no existing document found.
* @return new instance of {@link UpdateContext}.
*/
UpdateContext updateContext(UpdateDefinition updateDefinition, Query query, boolean upsert) {
return new UpdateContext(updateDefinition, query, true, upsert);
}
/**
* Create a new {@link UpdateContext} instance affecting a single document.
*
* @param updateDefinition must not be {@literal null}.
* @param query must not be {@literal null}.
* @param upsert use {@literal true} to insert diff when no existing document found.
* @return new instance of {@link UpdateContext}.
*/
UpdateContext updateSingleContext(UpdateDefinition updateDefinition, Query query, boolean upsert) {
return new UpdateContext(updateDefinition, query, false, upsert);
}
/**
* Create a new {@link UpdateContext} instance affecting a single document.
*
* @param updateDefinition must not be {@literal null}.
* @param query must not be {@literal null}.
* @param upsert use {@literal true} to insert diff when no existing document found.
* @return new instance of {@link UpdateContext}.
*/
UpdateContext updateSingleContext(UpdateDefinition updateDefinition, Document query, boolean upsert) {
return new UpdateContext(updateDefinition, query, false, upsert);
}
/**
* Create a new {@link DeleteContext} instance removing all matching documents.
*
* @param query must not be {@literal null}.
* @return new instance of {@link QueryContext}.
*/
DeleteContext deleteQueryContext(Query query) {
return new DeleteContext(query, true);
}
/**
* Create a new {@link DeleteContext} instance only the first matching document.
*
* @param query must not be {@literal null}.
* @return new instance of {@link QueryContext}.
*/
DeleteContext deleteSingleContext(Query query) {
return new DeleteContext(query, false);
}
/**
* {@link QueryContext} encapsulates common tasks required to convert a {@link Query} into its MongoDB document
* representation, mapping fieldnames, as well as determinging and applying {@link Collation collations}.
*
* @author Christoph Strobl
*/
class QueryContext {
private final Query query;
/**
* Create new a {@link QueryContext} instance from the given {@literal query} (can be eihter a {@link Query} or a
* plain {@link Document}.
*
* @param query can be {@literal null}.
*/
private QueryContext(@Nullable Query query) {
this.query = query != null ? query : new Query();
}
/**
* @return never {@literal null}.
*/
Query getQuery() {
return query;
}
/**
* Extract the raw {@link Query#getQueryObject() unmapped document} from the {@link Query}.
*
* @return
*/
Document getQueryObject() {
return query.getQueryObject();
}
/**
* Get the already mapped MongoDB query representation.
*
* @param domainType can be {@literal null}.
* @param entityLookup the {@link Function lookup} used to provide the {@link MongoPersistentEntity} for the
* given{@literal domainType}
* @param <T>
* @return never {@literal null}.
*/
<T> Document getMappedQuery(@Nullable Class domainType, Function<Class<T>, MongoPersistentEntity> entityLookup) {
return getMappedQuery(domainType == null ? null : entityLookup.apply(domainType));
}
/**
* Get the already mapped MongoDB query representation.
*
* @param entity the Entity to map field names to. Can be {@literal null}.
* @param <T>
* @return never {@literal null}.
*/
<T> Document getMappedQuery(@Nullable MongoPersistentEntity<T> entity) {
return queryMapper.getMappedObject(getQueryObject(), entity);
}
/**
* Get the already mapped {@link Query#getFieldsObject() fields projection}
*
* @param entity the Entity to map field names to. Can be {@literal null}.
* @return never {@literal null}.
*/
Document getMappedFields(@Nullable MongoPersistentEntity<?> entity) {
return queryMapper.getMappedFields(query.getFieldsObject(), entity);
}
/**
* Get the already mapped {@link Query#getSortObject() sort} option.
*
* @param entity the Entity to map field names to. Can be {@literal null}.
* @return never {@literal null}.
*/
Document getMappedSort(@Nullable MongoPersistentEntity<?> entity) {
return queryMapper.getMappedSort(query.getSortObject(), entity);
}
/**
* Apply the {@link com.mongodb.client.model.Collation} if present extracted from the {@link Query} or fall back to
* the {@literal domain types} default {@link org.springframework.data.mongodb.core.mapping.Document#collation()
* collation}.
*
* @param domainType can be {@literal null}.
* @param consumer must not be {@literal null}.
*/
void applyCollation(@Nullable Class<?> domainType, Consumer<com.mongodb.client.model.Collation> consumer) {
getCollation(domainType).ifPresent(consumer::accept);
}
/**
* Get the {@link com.mongodb.client.model.Collation} extracted from the {@link Query} if present or fall back to
* the {@literal domain types} default {@link org.springframework.data.mongodb.core.mapping.Document#collation()
* collation}.
*
* @param domainType can be {@literal null}.
* @return never {@literal null}.
*/
Optional<com.mongodb.client.model.Collation> getCollation(@Nullable Class<?> domainType) {
return entityOperations.forType(domainType).getCollation(query) //
.map(Collation::toMongoCollation);
}
}
/**
* A {@link QueryContext} that encapsulates common tasks required when running {@literal distinct} queries.
*
* @author Christoph Strobl
*/
class DistinctQueryContext extends QueryContext {
private final String fieldName;
/**
* Create a new {@link DistinctQueryContext} instance.
*
* @param query can be {@literal null}.
* @param fieldName must not be {@literal null}.
*/
private DistinctQueryContext(@Nullable Object query, String fieldName) {
super(query instanceof Document ? new BasicQuery((Document) query) : (Query) query);
this.fieldName = fieldName;
}
@Override
Document getMappedFields(@Nullable MongoPersistentEntity<?> entity) {
return queryMapper.getMappedFields(new Document(fieldName, 1), entity);
}
/**
* Get the mapped field name to project to.
*
* @param entity can be {@literal null}.
* @return never {@literal null}.
*/
String getMappedFieldName(@Nullable MongoPersistentEntity<?> entity) {
return getMappedFields(entity).keySet().iterator().next();
}
/**
* Get the MongoDB native representation of the given {@literal type}.
*
* @param type must not be {@literal null}.
* @param <T>
* @return never {@literal null}.
*/
<T> Class<T> getDriverCompatibleClass(Class<T> type) {
return codecRegistryProvider.getCodecFor(type) //
.map(Codec::getEncoderClass) //
.orElse((Class<T>) BsonValue.class);
}
/**
* Get the most speficic read target type based on the user {@literal requestedTargetType} an the property type
* based on meta information extracted from the {@literal domainType}.
*
* @param requestedTargetType must not be {@literal null}.
* @param domainType must not be {@literal null}.
* @return never {@literal null}.
*/
Class<?> getMostSpecificConversionTargetType(Class<?> requestedTargetType, Class<?> domainType) {
Class<?> conversionTargetType = requestedTargetType;
try {
Class<?> propertyType = PropertyPath.from(fieldName, domainType).getLeafProperty().getLeafType();
// use the more specific type but favor UserType over property one
if (ClassUtils.isAssignable(requestedTargetType, propertyType)) {
conversionTargetType = propertyType;
}
} catch (PropertyReferenceException e) {
// just don't care about it as we default to Object.class anyway.
}
return conversionTargetType;
}
}
/**
* A {@link QueryContext} that encapsulates common tasks required when running {@literal count} queries.
*
* @author Christoph Strobl
*/
class CountContext extends QueryContext {
/**
* Creates a new {@link CountContext} instance.
*
* @param query can be {@literal null}.
*/
CountContext(@Nullable Query query) {
super(query);
}
/**
* Get the {@link CountOptions} applicable for the {@link Query}.
*
* @param domainType must not be {@literal null}.
* @return never {@literal null}.
*/
CountOptions getCountOptions(@Nullable Class<?> domainType) {
return getCountOptions(domainType, null);
}
/**
* Get the {@link CountOptions} applicable for the {@link Query}.
*
* @param domainType can be {@literal null}.
* @param callback a callback to modify the generated options. Can be {@literal null}.
* @return
*/
CountOptions getCountOptions(@Nullable Class<?> domainType, @Nullable Consumer<CountOptions> callback) {
CountOptions options = new CountOptions();
Query query = getQuery();
applyCollation(domainType, options::collation);
if (query.getLimit() > 0) {
options.limit(query.getLimit());
}
if (query.getSkip() > 0) {
options.skip((int) query.getSkip());
}
if (StringUtils.hasText(query.getHint())) {
String hint = query.getHint();
if (BsonUtils.isJsonDocument(hint)) {
options.hint(BsonUtils.parse(hint, codecRegistryProvider));
} else {
options.hintString(hint);
}
}
if (callback != null) {
callback.accept(options);
}
return options;
}
}
/**
* A {@link QueryContext} that encapsulates common tasks required when running {@literal delete} queries.
*
* @author Christoph Strobl
*/
class DeleteContext extends QueryContext {
private boolean multi;
/**
* Crate a new {@link DeleteContext} instance.
*
* @param query can be {@literal null}.
* @param multi use {@literal true} to remove all matching documents, {@literal false} for just the first one.
*/
DeleteContext(@Nullable Query query, boolean multi) {
super(query);
this.multi = multi;
}
/**
* Get the {@link DeleteOptions} applicable for the {@link Query}.
*
* @param domainType must not be {@literal null}.
* @return never {@literal null}.
*/
DeleteOptions getDeleteOptions(@Nullable Class<?> domainType) {
return getDeleteOptions(domainType, null);
}
/**
* Get the {@link DeleteOptions} applicable for the {@link Query}.
*
* @param domainType can be {@literal null}.
* @param callback a callback to modify the generated options. Can be {@literal null}.
* @return
*/
DeleteOptions getDeleteOptions(@Nullable Class<?> domainType, @Nullable Consumer<DeleteOptions> callback) {
DeleteOptions options = new DeleteOptions();
applyCollation(domainType, options::collation);
if (callback != null) {
callback.accept(options);
}
return options;
}
/**
* @return {@literal true} if all matching documents shall be deleted.
*/
boolean isMulti() {
return multi;
}
}
/**
* A {@link QueryContext} that encapsulates common tasks required when running {@literal updates}.
*/
class UpdateContext extends QueryContext {
private final boolean multi;
private final boolean upsert;
private final UpdateDefinition update;
/**
* Create a new {@link UpdateContext} instance.
*
* @param update must not be {@literal null}.
* @param query must not be {@literal null}.
* @param multi use {@literal true} to update all matching documents.
* @param upsert use {@literal true} to insert a new document if none match.
*/
UpdateContext(UpdateDefinition update, Document query, boolean multi, boolean upsert) {
this(update, new BasicQuery(query), multi, upsert);
}
/**
* Create a new {@link UpdateContext} instance.
*
* @param update must not be {@literal null}.
* @param query can be {@literal null}.
* @param multi use {@literal true} to update all matching documents.
* @param upsert use {@literal true} to insert a new document if none match.
*/
UpdateContext(UpdateDefinition update, @Nullable Query query, boolean multi, boolean upsert) {
super(query);
this.multi = multi;
this.upsert = upsert;
this.update = update;
}
/**
* Get the {@link UpdateOptions} applicable for the {@link Query}.
*
* @param domainType must not be {@literal null}.
* @return never {@literal null}.
*/
UpdateOptions getUpdateOptions(@Nullable Class<?> domainType) {
return getUpdateOptions(domainType, null);
}
/**
* Get the {@link UpdateOptions} applicable for the {@link Query}.
*
* @param domainType can be {@literal null}.
* @param callback a callback to modify the generated options. Can be {@literal null}.
* @return
*/
UpdateOptions getUpdateOptions(@Nullable Class<?> domainType, @Nullable Consumer<UpdateOptions> callback) {
UpdateOptions options = new UpdateOptions();
options.upsert(upsert);
if (update.hasArrayFilters()) {
options
.arrayFilters(update.getArrayFilters().stream().map(ArrayFilter::asDocument).collect(Collectors.toList()));
}
applyCollation(domainType, options::collation);
if (callback != null) {
callback.accept(options);
}
return options;
}
/**
* Get the {@link ReplaceOptions} applicable for the {@link Query}.
*
* @param domainType must not be {@literal null}.
* @return never {@literal null}.
*/
ReplaceOptions getReplaceOptions(@Nullable Class<?> domainType) {
return getReplaceOptions(domainType, null);
}
/**
* Get the {@link ReplaceOptions} applicable for the {@link Query}.
*
* @param domainType can be {@literal null}.
* @param callback a callback to modify the generated options. Can be {@literal null}.
* @return
*/
ReplaceOptions getReplaceOptions(@Nullable Class<?> domainType, @Nullable Consumer<ReplaceOptions> callback) {
UpdateOptions updateOptions = getUpdateOptions(domainType);
ReplaceOptions options = new ReplaceOptions();
options.collation(updateOptions.getCollation());
options.upsert(updateOptions.isUpsert());
if (callback != null) {
callback.accept(options);
}
return options;
}
@Override
<T> Document getMappedQuery(@Nullable MongoPersistentEntity<T> domainType) {
Document mappedQuery = super.getMappedQuery(domainType);
if (multi && update.isIsolated() && !mappedQuery.containsKey("$isolated")) {
mappedQuery.put("$isolated", 1);
}
return mappedQuery;
}
/**
* Get the already mapped aggregation pipeline to use with an {@link #isAggregationUpdate()}.
*
* @param domainType must not be {@literal null}.
* @return never {@literal null}.
*/
List<Document> getUpdatePipeline(@Nullable Class<?> domainType) {
AggregationOperationContext context = domainType != null
? new RelaxedTypeBasedAggregationOperationContext(domainType, mappingContext, queryMapper)
: Aggregation.DEFAULT_CONTEXT;
return aggregationUtil.createPipeline((AggregationUpdate) update, context);
}
/**
* Get the already mapped update {@link Document}.
*
* @param entity
* @return
*/
Document getMappedUpdate(@Nullable MongoPersistentEntity<?> entity) {
return update instanceof MappedUpdate ? update.getUpdateObject()
: updateMapper.getMappedObject(update.getUpdateObject(), entity);
}
/**
* Increase a potential {@link MongoPersistentEntity#getVersionProperty() version property} prior to update if not
* already done in the actual {@link UpdateDefinition}
*
* @param persistentEntity can be {@literal null}.
*/
void increaseVersionForUpdateIfNecessary(@Nullable MongoPersistentEntity<?> persistentEntity) {
if (persistentEntity != null && persistentEntity.hasVersionProperty()) {
String versionFieldName = persistentEntity.getRequiredVersionProperty().getFieldName();
if (!update.modifies(versionFieldName)) {
update.inc(versionFieldName);
}
}
}
/**
* @return {@literal true} if the update holds an aggregation pipeline.
*/
boolean isAggregationUpdate() {
return update instanceof AggregationUpdate;
}
/**
* @return {@literal true} if all matching documents should be updated.
*/
boolean isMulti() {
return multi;
}
}
}

208
spring-data-mongodb/src/main/java/org/springframework/data/mongodb/core/ReactiveMongoTemplate.java
View File

@@ -21,7 +21,6 @@ import com.mongodb.client.result.InsertOneResult;
import lombok.AccessLevel;
import lombok.NonNull;
import lombok.RequiredArgsConstructor;
import org.springframework.data.mongodb.util.BsonUtils;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import reactor.util.function.Tuple2;
@@ -35,7 +34,6 @@ import java.util.stream.Collectors;
import org.bson.BsonValue;
import org.bson.Document;
import org.bson.codecs.Codec;
import org.bson.conversions.Bson;
import org.bson.types.ObjectId;
import org.reactivestreams.Publisher;
@@ -58,8 +56,6 @@ import org.springframework.data.geo.Distance;
import org.springframework.data.geo.GeoResult;
import org.springframework.data.geo.Metric;
import org.springframework.data.mapping.PersistentEntity;
import org.springframework.data.mapping.PropertyPath;
import org.springframework.data.mapping.PropertyReferenceException;
import org.springframework.data.mapping.callback.ReactiveEntityCallbacks;
import org.springframework.data.mapping.context.MappingContext;
import org.springframework.data.mapping.context.MappingContextEvent;
@@ -68,12 +64,15 @@ import org.springframework.data.mongodb.ReactiveMongoDatabaseFactory;
import org.springframework.data.mongodb.ReactiveMongoDatabaseUtils;
import org.springframework.data.mongodb.SessionSynchronization;
import org.springframework.data.mongodb.core.EntityOperations.AdaptibleEntity;
import org.springframework.data.mongodb.core.QueryOperations.CountContext;
import org.springframework.data.mongodb.core.QueryOperations.DeleteContext;
import org.springframework.data.mongodb.core.QueryOperations.DistinctQueryContext;
import org.springframework.data.mongodb.core.QueryOperations.QueryContext;
import org.springframework.data.mongodb.core.QueryOperations.UpdateContext;
import org.springframework.data.mongodb.core.aggregation.Aggregation;
import org.springframework.data.mongodb.core.aggregation.AggregationOperationContext;
import org.springframework.data.mongodb.core.aggregation.AggregationOptions;
import org.springframework.data.mongodb.core.aggregation.AggregationUpdate;
import org.springframework.data.mongodb.core.aggregation.PrefixingDelegatingAggregationOperationContext;
import org.springframework.data.mongodb.core.aggregation.RelaxedTypeBasedAggregationOperationContext;
import org.springframework.data.mongodb.core.aggregation.TypeBasedAggregationOperationContext;
import org.springframework.data.mongodb.core.aggregation.TypedAggregation;
import org.springframework.data.mongodb.core.convert.DbRefResolver;
@@ -112,6 +111,7 @@ import org.springframework.data.mongodb.core.query.Query;
import org.springframework.data.mongodb.core.query.UpdateDefinition;
import org.springframework.data.mongodb.core.query.UpdateDefinition.ArrayFilter;
import org.springframework.data.mongodb.core.validation.Validator;
import org.springframework.data.mongodb.util.BsonUtils;
import org.springframework.data.projection.SpelAwareProxyProjectionFactory;
import org.springframework.data.util.Optionals;
import org.springframework.lang.Nullable;
@@ -195,6 +195,7 @@ public class ReactiveMongoTemplate implements ReactiveMongoOperations, Applicati
private final ApplicationListener<MappingContextEvent<?, ?>> indexCreatorListener;
private final EntityOperations operations;
private final PropertyOperations propertyOperations;
private final QueryOperations queryOperations;
private @Nullable WriteConcern writeConcern;
private WriteConcernResolver writeConcernResolver = DefaultWriteConcernResolver.INSTANCE;
@@ -264,6 +265,7 @@ public class ReactiveMongoTemplate implements ReactiveMongoOperations, Applicati
this.mappingContext = this.mongoConverter.getMappingContext();
this.operations = new EntityOperations(this.mappingContext);
this.propertyOperations = new PropertyOperations(this.mappingContext);
this.queryOperations = new QueryOperations(queryMapper, updateMapper, operations, mongoDatabaseFactory);
// We create indexes based on mapping events
if (this.mappingContext instanceof MongoMappingContext) {
@@ -296,6 +298,7 @@ public class ReactiveMongoTemplate implements ReactiveMongoOperations, Applicati
this.operations = that.operations;
this.propertyOperations = that.propertyOperations;
this.sessionSynchronization = that.sessionSynchronization;
this.queryOperations = that.queryOperations;
}
private void onCheckForIndexes(MongoPersistentEntity<?> entity, Consumer<Throwable> subscriptionExceptionHandler) {
@@ -839,7 +842,9 @@ public class ReactiveMongoTemplate implements ReactiveMongoOperations, Applicati
return createFlux(collectionName, collection -> {
Document filter = queryMapper.getMappedObject(query.getQueryObject(), getPersistentEntity(entityClass));
QueryContext queryContext = queryOperations.createQueryContext(query);
Document filter = queryContext.getMappedQuery(entityClass, this::getPersistentEntity);
FindPublisher<Document> findPublisher = collection.find(filter, Document.class)
.projection(new Document("_id", 1));
@@ -847,8 +852,7 @@ public class ReactiveMongoTemplate implements ReactiveMongoOperations, Applicati
LOGGER.debug("exists: {} in collection: {}", serializeToJsonSafely(filter), collectionName);
}
findPublisher = operations.forType(entityClass).getCollation(query).map(Collation::toMongoCollation)
.map(findPublisher::collation).orElse(findPublisher);
queryContext.applyCollation(entityClass, findPublisher::collation);
return findPublisher.limit(1);
}).hasElements();
@@ -918,13 +922,11 @@ public class ReactiveMongoTemplate implements ReactiveMongoOperations, Applicati
Assert.notNull(resultClass, "ResultClass must not be null!");
MongoPersistentEntity<?> entity = getPersistentEntity(entityClass);
DistinctQueryContext distinctQueryContext = queryOperations.distincQueryContext(query, field);
Document mappedQuery = queryMapper.getMappedObject(query.getQueryObject(), entity);
String mappedFieldName = queryMapper.getMappedFields(new Document(field, 1), entity).keySet().iterator().next();
Class<T> mongoDriverCompatibleType = mongoDatabaseFactory.getCodecFor(resultClass) //
.map(Codec::getEncoderClass) //
.orElse((Class<T>) BsonValue.class);
Document mappedQuery = distinctQueryContext.getMappedQuery(entity);
String mappedFieldName = distinctQueryContext.getMappedFieldName(entity);
Class<T> mongoDriverCompatibleType = distinctQueryContext.getDriverCompatibleClass(resultClass);
Flux<?> result = execute(collectionName, collection -> {
@@ -939,15 +941,13 @@ public class ReactiveMongoTemplate implements ReactiveMongoOperations, Applicati
}
DistinctPublisher<T> publisher = collection.distinct(mappedFieldName, mappedQuery, mongoDriverCompatibleType);
return operations.forType(entityClass).getCollation(query) //
.map(Collation::toMongoCollation) //
.map(publisher::collation) //
.orElse(publisher);
distinctQueryContext.applyCollation(entityClass, publisher::collation);
return publisher;
});
if (resultClass == Object.class || mongoDriverCompatibleType != resultClass) {
Class<?> targetType = getMostSpecificConversionTargetType(resultClass, entityClass, field);
Class<?> targetType = distinctQueryContext.getMostSpecificConversionTargetType(resultClass, entityClass);
MongoConverter converter = getConverter();
result = result.map(it -> converter.mapValueToTargetType(it, targetType, NO_OP_REF_RESOLVER));
@@ -956,32 +956,6 @@ public class ReactiveMongoTemplate implements ReactiveMongoOperations, Applicati
return (Flux<T>) result;
}
/**
* @param userType must not be {@literal null}.
* @param domainType must not be {@literal null}.
* @param field must not be {@literal null}.
* @return the most specific conversion target type depending on user preference and domain type property.
* @since 2.1
*/
private static Class<?> getMostSpecificConversionTargetType(Class<?> userType, Class<?> domainType, String field) {
Class<?> conversionTargetType = userType;
try {
Class<?> propertyType = PropertyPath.from(field, domainType).getLeafProperty().getLeafType();
// use the more specific type but favor UserType over property one
if (ClassUtils.isAssignable(userType, propertyType)) {
conversionTargetType = propertyType;
}
} catch (PropertyReferenceException e) {
// just don't care about it as we default to Object.class anyway.
}
return conversionTargetType;
}
/*
* (non-Javadoc)
* @see org.springframework.data.mongodb.core.ReactiveMongoOperations#aggregate(org.springframework.data.mongodb.core.aggregation.TypedAggregation, java.lang.String, java.lang.Class)
@@ -1191,10 +1165,11 @@ public class ReactiveMongoTemplate implements ReactiveMongoOperations, Applicati
Assert.isTrue(query.getSkip() <= 0, "Query must not define skip.");
MongoPersistentEntity<?> entity = mappingContext.getPersistentEntity(entityType);
QueryContext queryContext = queryOperations.createQueryContext(query);
Document mappedQuery = queryMapper.getMappedObject(query.getQueryObject(), entity);
Document mappedFields = queryMapper.getMappedFields(query.getFieldsObject(), entity);
Document mappedSort = queryMapper.getMappedSort(query.getSortObject(), entity);
Document mappedQuery = queryContext.getMappedQuery(entity);
Document mappedFields = queryContext.getMappedFields(entity);
Document mappedSort = queryContext.getMappedSort(entity);
return Mono.just(PersistableEntityModel.of(replacement, collectionName)) //
.doOnNext(it -> maybeEmitEvent(new BeforeConvertEvent<>(it.getSource(), it.getCollection()))) //
@@ -1212,8 +1187,8 @@ public class ReactiveMongoTemplate implements ReactiveMongoOperations, Applicati
PersistableEntityModel<S> flowObject = (PersistableEntityModel<S>) it;
return doFindAndReplace(flowObject.getCollection(), mappedQuery, mappedFields, mappedSort,
operations.forType(entityType).getCollation(query).map(Collation::toMongoCollation).orElse(null),
entityType, flowObject.getTarget(), options, resultType);
queryContext.getCollation(entityType).orElse(null), entityType, flowObject.getTarget(), options,
resultType);
});
}
@@ -1267,30 +1242,10 @@ public class ReactiveMongoTemplate implements ReactiveMongoOperations, Applicati
return createMono(collectionName, collection -> {
Document filter = queryMapper.getMappedObject(query.getQueryObject(),
entityClass == null ? null : mappingContext.getPersistentEntity(entityClass));
CountContext countContext = queryOperations.countQueryContext(query);
CountOptions options = new CountOptions();
query.getCollation().map(Collation::toMongoCollation).ifPresent(options::collation);
if (query.getLimit() > 0) {
options.limit(query.getLimit());
}
if (query.getSkip() > 0) {
options.skip((int) query.getSkip());
}
if (StringUtils.hasText(query.getHint())) {
String hint = query.getHint();
if(BsonUtils.isJsonDocument(hint)) {
options = options.hint(BsonUtils.parse(hint, mongoDatabaseFactory));
} else {
options = options.hintString(hint);
}
}
operations.forType(entityClass).getCollation(query).map(Collation::toMongoCollation) //
.ifPresent(options::collation);
CountOptions options = countContext.getCountOptions(entityClass);
Document filter = countContext.getMappedQuery(entityClass, mappingContext::getPersistentEntity);
if (LOGGER.isDebugEnabled()) {
LOGGER.debug("Executing count: {} in collection: {}", serializeToJsonSafely(filter), collectionName);
@@ -1775,35 +1730,22 @@ public class ReactiveMongoTemplate implements ReactiveMongoOperations, Applicati
}
MongoPersistentEntity<?> entity = entityClass == null ? null : getPersistentEntity(entityClass);
increaseVersionForUpdateIfNecessary(entity, update);
Document queryObj = queryMapper.getMappedObject(query.getQueryObject(), entity);
UpdateContext updateContext = multi ? queryOperations.updateContext(update, query, upsert)
: queryOperations.updateSingleContext(update, query, upsert);
updateContext.increaseVersionForUpdateIfNecessary(entity);
UpdateOptions updateOptions = new UpdateOptions().upsert(upsert);
operations.forType(entityClass).getCollation(query) //
.map(Collation::toMongoCollation) //
.ifPresent(updateOptions::collation);
if (update.hasArrayFilters()) {
updateOptions.arrayFilters(update.getArrayFilters().stream().map(ArrayFilter::asDocument)
.map(it -> queryMapper.getMappedObject(it, entity)).collect(Collectors.toList()));
}
if (multi && update.isIsolated() && !queryObj.containsKey("$isolated")) {
queryObj.put("$isolated", 1);
}
Document queryObj = updateContext.getMappedQuery(entity);
UpdateOptions updateOptions = updateContext.getUpdateOptions(entityClass);
Flux<UpdateResult> result;
if (update instanceof AggregationUpdate) {
if (updateContext.isAggregationUpdate()) {
AggregationOperationContext context = entityClass != null
? new RelaxedTypeBasedAggregationOperationContext(entityClass, mappingContext, queryMapper)
: Aggregation.DEFAULT_CONTEXT;
List<Document> pipeline = new AggregationUtil(queryMapper, mappingContext)
.createPipeline((AggregationUpdate) update, context);
List<Document> pipeline = updateContext.getUpdatePipeline(entityClass);
MongoAction mongoAction = new MongoAction(writeConcern, MongoActionOperation.UPDATE, collectionName, entityClass,
update.getUpdateObject(), queryObj);
WriteConcern writeConcernToUse = prepareWriteConcern(mongoAction);
result = execute(collectionName, collection -> {
@@ -1812,10 +1754,6 @@ public class ReactiveMongoTemplate implements ReactiveMongoOperations, Applicati
serializeToJsonSafely(queryObj), serializeToJsonSafely(pipeline), collectionName));
}
MongoAction mongoAction = new MongoAction(writeConcern, MongoActionOperation.UPDATE, collectionName,
entityClass, update.getUpdateObject(), queryObj);
WriteConcern writeConcernToUse = prepareWriteConcern(mongoAction);
collection = writeConcernToUse != null ? collection.withWriteConcern(writeConcernToUse) : collection;
return multi ? collection.updateMany(queryObj, pipeline, updateOptions)
@@ -1823,26 +1761,23 @@ public class ReactiveMongoTemplate implements ReactiveMongoOperations, Applicati
});
} else {
result = execute(collectionName, collection -> {
Document updateObj = updateContext.getMappedUpdate(entity);
MongoAction mongoAction = new MongoAction(writeConcern, MongoActionOperation.UPDATE, collectionName, entityClass,
updateObj, queryObj);
WriteConcern writeConcernToUse = prepareWriteConcern(mongoAction);
Document updateObj = updateMapper.getMappedObject(update.getUpdateObject(), entity);
result = execute(collectionName, collection -> {
if (LOGGER.isDebugEnabled()) {
LOGGER.debug(String.format("Calling update using query: %s and update: %s in collection: %s",
serializeToJsonSafely(queryObj), serializeToJsonSafely(updateObj), collectionName));
}
MongoAction mongoAction = new MongoAction(writeConcern, MongoActionOperation.UPDATE, collectionName,
entityClass, updateObj, queryObj);
WriteConcern writeConcernToUse = prepareWriteConcern(mongoAction);
MongoCollection<Document> collectionToUse = prepareCollection(collection, writeConcernToUse);
if (!UpdateMapper.isUpdateObject(updateObj)) {
ReplaceOptions replaceOptions = new ReplaceOptions();
replaceOptions.upsert(updateOptions.isUpsert());
replaceOptions.collation(updateOptions.getCollation());
ReplaceOptions replaceOptions = updateContext.getReplaceOptions(entityClass);
return collectionToUse.replaceOne(queryObj, updateObj, replaceOptions);
}
@@ -1856,7 +1791,7 @@ public class ReactiveMongoTemplate implements ReactiveMongoOperations, Applicati
if (entity != null && entity.hasVersionProperty() && !multi) {
if (updateResult.wasAcknowledged() && updateResult.getMatchedCount() == 0) {
Document updateObj = updateMapper.getMappedObject(update.getUpdateObject(), entity);
Document updateObj = updateContext.getMappedUpdate(entity);
if (containsVersionProperty(queryObj, entity))
throw new OptimisticLockingFailureException("Optimistic lock exception on saving entity: "
+ updateObj.toString() + " to collection " + collectionName);
@@ -1867,17 +1802,6 @@ public class ReactiveMongoTemplate implements ReactiveMongoOperations, Applicati
return result.next();
}
private void increaseVersionForUpdateIfNecessary(@Nullable MongoPersistentEntity<?> persistentEntity,
UpdateDefinition update) {
if (persistentEntity != null && persistentEntity.hasVersionProperty()) {
String versionFieldName = persistentEntity.getRequiredVersionProperty().getFieldName();
if (!update.modifies(versionFieldName)) {
update.inc(versionFieldName);
}
}
}
private boolean containsVersionProperty(Document document, @Nullable MongoPersistentEntity<?> persistentEntity) {
if (persistentEntity == null || !persistentEntity.hasVersionProperty()) {
@@ -1981,24 +1905,20 @@ public class ReactiveMongoTemplate implements ReactiveMongoOperations, Applicati
Assert.hasText(collectionName, "Collection name must not be null or empty!");
Document queryObject = query.getQueryObject();
MongoPersistentEntity<?> entity = getPersistentEntity(entityClass);
Document removeQuery = queryMapper.getMappedObject(queryObject, entity);
DeleteContext deleteContext = queryOperations.deleteQueryContext(query);
Document queryObject = deleteContext.getMappedQuery(entity);
DeleteOptions deleteOptions = deleteContext.getDeleteOptions(entityClass);
Document removeQuery = deleteContext.getMappedQuery(entity);
MongoAction mongoAction = new MongoAction(writeConcern, MongoActionOperation.REMOVE, collectionName, entityClass,
null, removeQuery);
WriteConcern writeConcernToUse = prepareWriteConcern(mongoAction);
return execute(collectionName, collection -> {
maybeEmitEvent(new BeforeDeleteEvent<>(removeQuery, entityClass, collectionName));
MongoAction mongoAction = new MongoAction(writeConcern, MongoActionOperation.REMOVE, collectionName, entityClass,
null, removeQuery);
DeleteOptions deleteOptions = new DeleteOptions();
operations.forType(entityClass).getCollation(query) //
.map(Collation::toMongoCollation) //
.ifPresent(deleteOptions::collation);
WriteConcern writeConcernToUse = prepareWriteConcern(mongoAction);
MongoCollection<Document> collectionToUse = prepareCollection(collection, writeConcernToUse);
if (LOGGER.isDebugEnabled()) {
@@ -2586,22 +2506,14 @@ public class ReactiveMongoTemplate implements ReactiveMongoOperations, Applicati
Class<T> entityClass, UpdateDefinition update, FindAndModifyOptions options) {
MongoPersistentEntity<?> entity = mappingContext.getPersistentEntity(entityClass);
increaseVersionForUpdateIfNecessary(entity, update);
UpdateContext updateContext = queryOperations.updateSingleContext(update, query, false);
updateContext.increaseVersionForUpdateIfNecessary(entity);
return Mono.defer(() -> {
Document mappedQuery = queryMapper.getMappedObject(query, entity);
Object mappedUpdate;
if (update instanceof AggregationUpdate) {
AggregationOperationContext context = new RelaxedTypeBasedAggregationOperationContext(entityClass,
mappingContext, queryMapper);
mappedUpdate = new AggregationUtil(queryMapper, mappingContext).createPipeline((Aggregation) update, context);
} else {
mappedUpdate = updateMapper.getMappedObject(update.getUpdateObject(), entity);
}
Document mappedQuery = updateContext.getMappedQuery(entity);
Object mappedUpdate = updateContext.isAggregationUpdate() ? updateContext.getUpdatePipeline(entityClass)
: updateContext.getMappedUpdate(entity);
if (LOGGER.isDebugEnabled()) {
LOGGER.debug(String.format(
@@ -3286,7 +3198,7 @@ public class ReactiveMongoTemplate implements ReactiveMongoOperations, Applicati
String hint = query.getHint();
if(BsonUtils.isJsonDocument(hint)) {
if (BsonUtils.isJsonDocument(hint)) {
findPublisherToUse = findPublisherToUse.hint(BsonUtils.parse(hint, mongoDatabaseFactory));
} else {
findPublisherToUse = findPublisherToUse.hintString(hint);