Support ReadConcern & ReadPreference via the Query and Aggregation API.

Add support for setting the ReadConcern and ReadPreference via the Query and Aggregation API.

Closes: #4277, #4286
Original Pull Request: #4288

Author: mp911de

Diff for: spring-data-mongodb/src/main/java/org/springframework/data/mongodb/core/CollectionPreparer.java

@@ -0,0 +1,61 @@
+/*
+ * 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.util.Assert;
+
+import com.mongodb.client.MongoCollection;
+
+/**
+ * Interface for functional preparation of a {@link MongoCollection}.
+ *
+ * @author Mark Paluch
+ * @since 4.1
+ */
+public interface CollectionPreparer<T> {
+
+	/**
+	 * Returns a preparer that always returns its input collection.
+	 *
+	 * @return a preparer that always returns its input collection.
+	 */
+	static <T> CollectionPreparer<T> identity() {
+		return it -> it;
+	}
+
+	/**
+	 * Prepare the {@code collection}.
+	 *
+	 * @param collection the collection to prepare.
+	 * @return the prepared collection.
+	 */
+	T prepare(T collection);
+
+	/**
+	 * Returns a composed {@code CollectionPreparer} that first applies this preparer to the collection, and then applies
+	 * the {@code after} preparer to the result. If evaluation of either function throws an exception, it is relayed to
+	 * the caller of the composed function.
+	 *
+	 * @param after the collection preparer to apply after this function is applied.
+	 * @return a composed {@code CollectionPreparer} that first applies this preparer and then applies the {@code after}
+	 *         preparer.
+	 */
+	default CollectionPreparer<T> andThen(CollectionPreparer<T> after) {
+		Assert.notNull(after, "After CollectionPreparer must not be null");
+		return c -> after.prepare(prepare(c));
+	}
+
+}

Diff for: spring-data-mongodb/src/main/java/org/springframework/data/mongodb/core/CollectionPreparerSupport.java

@@ -0,0 +1,182 @@
+/*
+ * 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.Arrays;
+import java.util.List;
+import java.util.function.BiFunction;
+import java.util.function.Function;
+
+import org.bson.Document;
+
+import com.mongodb.ReadConcern;
+import com.mongodb.ReadPreference;
+import com.mongodb.client.MongoCollection;
+
+/**
+ * Support class for delegate implementations to apply {@link ReadConcern} and {@link ReadPreference} settings upon
+ * {@link CollectionPreparer preparing a collection}.
+ *
+ * @author Mark Paluch
+ * @since 4.1
+ */
+class CollectionPreparerSupport implements ReadConcernAware, ReadPreferenceAware {
+
+	private final List<Object> sources;
+
+	private CollectionPreparerSupport(List<Object> sources) {
+		this.sources = sources;
+	}
+
+	<T> T doPrepare(T collection, Function<T, ReadConcern> concernAccessor, BiFunction<T, ReadConcern, T> concernFunction,
+			Function<T, ReadPreference> preferenceAccessor, BiFunction<T, ReadPreference, T> preferenceFunction) {
+
+		T collectionToUse = collection;
+
+		for (Object source : sources) {
+			if (source instanceof ReadConcernAware rca && rca.hasReadConcern()) {
+
+				ReadConcern concern = rca.getReadConcern();
+				if (concernAccessor.apply(collectionToUse) != concern) {
+					collectionToUse = concernFunction.apply(collectionToUse, concern);
+				}
+				break;
+			}
+		}
+
+		for (Object source : sources) {
+			if (source instanceof ReadPreferenceAware rpa && rpa.hasReadPreference()) {
+
+				ReadPreference preference = rpa.getReadPreference();
+				if (preferenceAccessor.apply(collectionToUse) != preference) {
+					collectionToUse = preferenceFunction.apply(collectionToUse, 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;
+	}
+
+	static class CollectionPreparerDelegate extends CollectionPreparerSupport
+			implements CollectionPreparer<MongoCollection<Document>> {
+
+		private CollectionPreparerDelegate(List<Object> sources) {
+			super(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(Arrays.asList(mixedAwares));
+		}
+
+		@Override
+		public MongoCollection<Document> prepare(MongoCollection<Document> collection) {
+			return doPrepare(collection, MongoCollection::getReadConcern, MongoCollection::withReadConcern,
+					MongoCollection::getReadPreference, MongoCollection::withReadPreference);
+		}
+
+	}
+
+	static class ReactiveCollectionPreparerDelegate extends CollectionPreparerSupport
+			implements CollectionPreparer<com.mongodb.reactivestreams.client.MongoCollection<Document>> {
+
+		private ReactiveCollectionPreparerDelegate(List<Object> sources) {
+			super(sources);
+		}
+
+		public static ReactiveCollectionPreparerDelegate of(ReadPreferenceAware... awares) {
+			return of((Object[]) awares);
+		}
+
+		public static ReactiveCollectionPreparerDelegate of(Object... mixedAwares) {
+
+			if (mixedAwares.length == 1 && mixedAwares[0] instanceof CollectionPreparerDelegate) {
+				return (ReactiveCollectionPreparerDelegate) mixedAwares[0];
+			}
+
+			return new ReactiveCollectionPreparerDelegate(Arrays.asList(mixedAwares));
+		}
+
+		@Override
+		public com.mongodb.reactivestreams.client.MongoCollection<Document> prepare(
+				com.mongodb.reactivestreams.client.MongoCollection<Document> collection) {
+			return doPrepare(collection, //
+					com.mongodb.reactivestreams.client.MongoCollection::getReadConcern,
+					com.mongodb.reactivestreams.client.MongoCollection::withReadConcern,
+					com.mongodb.reactivestreams.client.MongoCollection::getReadPreference,
+					com.mongodb.reactivestreams.client.MongoCollection::withReadPreference);
+		}
+
+	}
+
+}

Diff for: spring-data-mongodb/src/main/java/org/springframework/data/mongodb/core/ExecutableFindOperationSupport.java

@@ -20,7 +20,6 @@
 import java.util.stream.Stream;
 
 import org.bson.Document;
-
 import org.springframework.dao.IncorrectResultSizeDataAccessException;
 import org.springframework.data.mongodb.core.query.NearQuery;
 import org.springframework.data.mongodb.core.query.Query;
@@ -168,7 +167,8 @@ private List<T> doFind(@Nullable CursorPreparer preparer) {
 			Document queryObject = query.getQueryObject();
 			Document fieldsObject = query.getFieldsObject();
 
-			return template.doFind(getCollectionName(), queryObject, fieldsObject, domainType, returnType,
+			return template.doFind(template.createDelegate(query), getCollectionName(), queryObject, fieldsObject, domainType,
+					returnType,
 					getCursorPreparer(query, preparer));
 		}