Add reactive CollectionPreparer variant.

Author: mp911de

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

@@ -15,6 +15,7 @@
  */
 package org.springframework.data.mongodb.core;
 
+import java.util.Arrays;
 import java.util.List;
 import java.util.function.BiFunction;
 import java.util.function.Function;
@@ -34,7 +35,7 @@
  */
 class CollectionPreparerSupport implements ReadConcernAware, ReadPreferenceAware {
 
-	final List<Object> sources;
+	private final List<Object> sources;
 
 	private CollectionPreparerSupport(List<Object> sources) {
 		this.sources = sources;
@@ -134,7 +135,8 @@ public static CollectionPreparerDelegate of(Object... mixedAwares) {
 			if (mixedAwares.length == 1 && mixedAwares[0] instanceof CollectionPreparerDelegate) {
 				return (CollectionPreparerDelegate) mixedAwares[0];
 			}
-			return new CollectionPreparerDelegate(List.of(mixedAwares));
+
+			return new CollectionPreparerDelegate(Arrays.asList(mixedAwares));
 		}
 
 		@Override
@@ -145,4 +147,36 @@ public MongoCollection<Document> prepare(MongoCollection<Document> collection) {
 
 	}
 
+	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/ReactiveFindOperationSupport.java

@@ -20,6 +20,7 @@
 
 import org.bson.Document;
 import org.springframework.dao.IncorrectResultSizeDataAccessException;
+import org.springframework.data.mongodb.core.CollectionPreparerSupport.ReactiveCollectionPreparerDelegate;
 import org.springframework.data.mongodb.core.query.NearQuery;
 import org.springframework.data.mongodb.core.query.Query;
 import org.springframework.data.mongodb.core.query.SerializationUtils;
@@ -67,8 +68,8 @@ static class ReactiveFindSupport<T>
 		private final String collection;
 		private final Query query;
 
-		ReactiveFindSupport(ReactiveMongoTemplate template, Class<?> domainType, Class<T> returnType,
-				String collection, Query query) {
+		ReactiveFindSupport(ReactiveMongoTemplate template, Class<?> domainType, Class<T> returnType, String collection,
+				Query query) {
 
 			this.template = template;
 			this.domainType = domainType;
@@ -169,8 +170,8 @@ private Flux<T> doFind(@Nullable FindPublisherPreparer preparer) {
 			Document queryObject = query.getQueryObject();
 			Document fieldsObject = query.getFieldsObject();
 
-			return template.doFind(getCollectionName(), queryObject, fieldsObject, domainType, returnType,
-					preparer != null ? preparer : getCursorPreparer(query));
+			return template.doFind(ReactiveCollectionPreparerDelegate.of(query), getCollectionName(), queryObject,
+					fieldsObject, domainType, returnType, preparer != null ? preparer : getCursorPreparer(query));
 		}
 
 		@SuppressWarnings("unchecked")