Add support for ReadConcern.

Author: mp911de

spring-data-mongodb/src/main/java/org/springframework/data/mongodb/core/CollectionPreparerDelegate.java

@@ -0,0 +1,131 @@
+/*
+ * Copyright 2023 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 org.bson.Document;
+
+import com.mongodb.ReadConcern;
+import com.mongodb.ReadPreference;
+import com.mongodb.client.MongoCollection;
+
+/**
+ * Delegate to apply {@link ReadConcern} and {@link ReadPreference} settings upon {@link CollectionPreparer preparing a
+ * collection}.
+ *
+ * @author Mark Paluch
+ * @since 4.1
+ */
+class CollectionPreparerDelegate implements ReadConcernAware, ReadPreferenceAware, CollectionPreparer {
+
+	List<Object> sources;
+
+	private CollectionPreparerDelegate(List<Object> sources) {
+		this.sources = sources;
+	}
+
+	public static CollectionPreparerDelegate of(ReadPreferenceAware... awares) {
+		return of((Object[]) awares);
+	}
+
+	public static CollectionPreparerDelegate of(Object... mixedAwares) {
+
+		if (mixedAwares.length == 1 && mixedAwares[0] instanceof CollectionPreparerDelegate) {
+			return (CollectionPreparerDelegate) mixedAwares[0];
+		}
+		return new CollectionPreparerDelegate(List.of(mixedAwares));
+	}
+
+	@Override
+	public MongoCollection<Document> prepare(MongoCollection<Document> collection) {
+
+		MongoCollection<Document> collectionToUse = collection;
+
+		for (Object source : sources) {
+			if (source instanceof ReadConcernAware rca && rca.hasReadConcern()) {
+
+				ReadConcern concern = rca.getReadConcern();
+				if (collection.getReadConcern() != concern) {
+					collectionToUse = collectionToUse.withReadConcern(concern);
+				}
+				break;
+			}
+		}
+
+		for (Object source : sources) {
+			if (source instanceof ReadPreferenceAware rpa && rpa.hasReadPreference()) {
+
+				ReadPreference preference = rpa.getReadPreference();
+				if (collection.getReadPreference() != preference) {
+					collectionToUse = collectionToUse.withReadPreference(preference);
+				}
+				break;
+			}
+		}
+
+		return collectionToUse;
+	}
+
+	@Override
+	public boolean hasReadConcern() {
+
+		for (Object aware : sources) {
+			if (aware instanceof ReadConcernAware rca && rca.hasReadConcern()) {
+				return true;
+			}
+		}
+
+		return false;
+	}
+
+	@Override
+	public ReadConcern getReadConcern() {
+
+		for (Object aware : sources) {
+			if (aware instanceof ReadConcernAware rca && rca.hasReadConcern()) {
+				return rca.getReadConcern();
+			}
+		}
+
+		return null;
+	}
+
+	@Override
+	public boolean hasReadPreference() {
+
+		for (Object aware : sources) {
+			if (aware instanceof ReadPreferenceAware rpa && rpa.hasReadPreference()) {
+				return true;
+			}
+		}
+
+		return false;
+	}
+
+	@Override
+	public ReadPreference getReadPreference() {
+
+		for (Object aware : sources) {
+			if (aware instanceof ReadPreferenceAware rpa && rpa.hasReadPreference()) {
+				return rpa.getReadPreference();
+			}
+		}
+
+		return null;
+	}
+
+}

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

@@ -454,7 +454,7 @@ protected <T> Stream<T> doStream(Query query, Class<?> entityType, String collec
 			Document mappedQuery = queryContext.getMappedQuery(persistentEntity);
 			Document mappedFields = queryContext.getMappedFields(persistentEntity, projection);
 
-			ReadPreferenceDelegate readPreference = createDelegate(query);
+			CollectionPreparerDelegate readPreference = createDelegate(query);
 			FindIterable<Document> cursor = new QueryCursorPreparer(query, entityType).initiateFind(collection,
 					col -> readPreference.prepare(col).find(mappedQuery, Document.class).projection(mappedFields));
 
@@ -1025,7 +1025,7 @@ public <S, T> T findAndReplace(Query query, S replacement, FindAndReplaceOptions
 		QueryContext queryContext = queryOperations.createQueryContext(query);
 
 		EntityProjection<T, S> projection = operations.introspectProjection(resultType, entityType);
-		ReadPreferenceDelegate collectionPreparer = createDelegate(query);
+		CollectionPreparerDelegate collectionPreparer = createDelegate(query);
 		Document mappedQuery = queryContext.getMappedQuery(entity);
 		Document mappedFields = queryContext.getMappedFields(entity, projection);
 		Document mappedSort = queryContext.getMappedSort(entity);
@@ -1096,7 +1096,7 @@ public long count(Query query, @Nullable Class<?> entityClass, String collection
 		CountOptions options = countContext.getCountOptions(entityClass);
 		Document mappedQuery = countContext.getMappedQuery(entityClass, mappingContext::getPersistentEntity);
 
-		ReadPreferenceDelegate readPreference = createDelegate(query);
+		CollectionPreparerDelegate readPreference = createDelegate(query);
 		return doCount(readPreference, collectionName, mappedQuery, options);
 	}
 
@@ -1117,7 +1117,7 @@ protected long doCount(CollectionPreparer collectionPreparer, String collectionN
 	 */
 	@Override
 	public long estimatedCount(String collectionName) {
-		return doEstimatedCount(ReadPreferenceDelegate.of(this), collectionName, new EstimatedDocumentCountOptions());
+		return doEstimatedCount(CollectionPreparerDelegate.of(this), collectionName, new EstimatedDocumentCountOptions());
 	}
 
 	protected long doEstimatedCount(CollectionPreparer collectionPreparer, String collectionName,
@@ -1836,7 +1836,7 @@ public <T> List<T> mapReduce(Query query, Class<?> domainType, String inputColle
 
 		String mapFunc = replaceWithResourceIfNecessary(mapFunction);
 		String reduceFunc = replaceWithResourceIfNecessary(reduceFunction);
-		ReadPreferenceDelegate readPreference = createDelegate(query);
+		CollectionPreparerDelegate readPreference = createDelegate(query);
 		MongoCollection<Document> inputCollection = readPreference
 				.prepare(getAndPrepareCollection(doGetDatabase(), inputCollectionName));
 
@@ -2061,7 +2061,7 @@ protected <O> AggregationResults<O> doAggregate(Aggregation aggregation, String
 		return execute(collectionName, collection -> {
 
 			List<Document> rawResult = new ArrayList<>();
-			ReadPreferenceDelegate delegate = ReadPreferenceDelegate.of(options);
+			CollectionPreparerDelegate delegate = CollectionPreparerDelegate.of(options);
 			Class<?> domainType = aggregation instanceof TypedAggregation ? ((TypedAggregation<?>) aggregation).getInputType()
 					: null;
 
@@ -2132,7 +2132,8 @@ protected <O> Stream<O> aggregateStream(Aggregation aggregation, String collecti
 
 		return execute(collectionName, (CollectionCallback<Stream<O>>) collection -> {
 
-			ReadPreferenceDelegate delegate = ReadPreferenceDelegate.of(options);
+			CollectionPreparerDelegate delegate = CollectionPreparerDelegate.of(options);
+
 			AggregateIterable<Document> cursor = delegate.prepare(collection).aggregate(pipeline, Document.class) //
 					.allowDiskUse(options.isAllowDiskUse());
 
@@ -2649,8 +2650,8 @@ protected <T> T doFindAndReplace(CollectionPreparer collectionPreparer, String c
 				options, projection);
 	}
 
-	ReadPreferenceDelegate createDelegate(Query query) {
-		return ReadPreferenceDelegate.of(query);
+	CollectionPreparerDelegate createDelegate(Query query) {
+		return CollectionPreparerDelegate.of(query);
 	}
 
 	/**

spring-data-mongodb/src/main/java/org/springframework/data/mongodb/core/ReadConcernAware.java

@@ -0,0 +1,44 @@
+/*
+ * Copyright 2023 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 org.springframework.lang.Nullable;
+
+import com.mongodb.ReadConcern;
+
+/**
+ * Interface to be implemented by any object that wishes to expose the {@link ReadConcern}.
+ * <p>
+ * Typically implemented by cursor or query preparer objects.
+ *
+ * @author Mark Paluch
+ * @since 4.1
+ */
+public interface ReadConcernAware {
+
+	/**
+	 * @return {@literal true} if a {@link ReadConcern} is set.
+	 */
+	default boolean hasReadConcern() {
+		return getReadConcern() != null;
+	}
+
+	/**
+	 * @return the {@link ReadConcern} to apply or {@literal null} if none set.
+	 */
+	@Nullable
+	ReadConcern getReadConcern();
+}