Polishing.

Simplify reactive composition. Switch to eager operator evaluation.

See #4712
Original pull request: #4717

Author: mp911de

spring-data-mongodb/src/main/java/org/springframework/data/mongodb/repository/query/AbstractReactiveMongoQuery.java

@@ -17,14 +17,14 @@
 
 import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
-import reactor.util.function.Tuple2;
 
 import java.util.ArrayList;
 import java.util.List;
 
 import org.bson.Document;
 import org.bson.codecs.configuration.CodecRegistry;
 import org.reactivestreams.Publisher;
+
 import org.springframework.core.convert.converter.Converter;
 import org.springframework.data.mapping.model.EntityInstantiators;
 import org.springframework.data.mapping.model.SpELExpressionEvaluator;
@@ -231,7 +231,6 @@ private boolean isTailable(MongoQueryMethod method) {
 		return method.getTailableAnnotation() != null;
 	}
 
-
 	Query applyQueryMetaAttributesWhenPresent(Query query) {
 
 		if (method.hasQueryMetaAttributes()) {
@@ -290,7 +289,8 @@ Query applyHintIfPresent(Query query) {
 	}
 
 	/**
-	 * If present apply the {@link com.mongodb.ReadPreference} from the {@link org.springframework.data.mongodb.repository.ReadPreference} annotation.
+	 * If present apply the {@link com.mongodb.ReadPreference} from the
+	 * {@link org.springframework.data.mongodb.repository.ReadPreference} annotation.
 	 *
 	 * @param query must not be {@literal null}.
 	 * @return never {@literal null}.
@@ -339,8 +339,8 @@ protected Mono<UpdateDefinition> createUpdate(MongoParameterAccessor accessor) {
 
 				String updateJson = updateSource.update();
 				return getParameterBindingCodec() //
-						.flatMap(codec -> expressionEvaluator(updateJson, accessor, codec)) //
-						.map(it -> decode(it.getT1(), updateJson, accessor, it.getT2())) //
+						.flatMap(codec -> expressionEvaluator(updateJson, accessor, codec) //
+								.map(evaluator -> decode(evaluator, updateJson, accessor, codec))) //
 						.map(BasicUpdate::fromDocument);
 			}
 			if (!ObjectUtils.isEmpty(updateSource.pipeline())) {
@@ -376,16 +376,17 @@ protected Mono<List<AggregationOperation>> parseAggregationPipeline(String[] pip
 	private Mono<AggregationOperation> computePipelineStage(String source, MongoParameterAccessor accessor,
 			ParameterBindingDocumentCodec codec) {
 
-		return expressionEvaluator(source, accessor, codec).map(
-            it -> new StringAggregationOperation(source, AbstractReactiveMongoQuery.this.getQueryMethod().getDomainClass(), bsonString -> AbstractReactiveMongoQuery.this.decode(it.getT1(), bsonString, accessor, it.getT2())));
+		return expressionEvaluator(source, accessor, codec).map(evaluator -> new StringAggregationOperation(source,
+				AbstractReactiveMongoQuery.this.getQueryMethod().getDomainClass(),
+				bsonString -> AbstractReactiveMongoQuery.this.decode(evaluator, bsonString, accessor, codec)));
 	}
 
-	private Mono<Tuple2<SpELExpressionEvaluator, ParameterBindingDocumentCodec>> expressionEvaluator(String source,
-			MongoParameterAccessor accessor, ParameterBindingDocumentCodec codec) {
+	private Mono<SpELExpressionEvaluator> expressionEvaluator(String source, MongoParameterAccessor accessor,
+			ParameterBindingDocumentCodec codec) {
 
 		ExpressionDependencies dependencies = codec.captureExpressionDependencies(source, accessor::getBindableValue,
 				expressionParser);
-		return getSpelEvaluatorFor(dependencies, accessor).zipWith(Mono.just(codec));
+		return getSpelEvaluatorFor(dependencies, accessor);
 	}
 
 	private Document decode(SpELExpressionEvaluator expressionEvaluator, String source, MongoParameterAccessor accessor,

spring-data-mongodb/src/main/java/org/springframework/data/mongodb/repository/query/StringAggregationOperation.java

@@ -22,9 +22,13 @@
 import org.bson.Document;
 import org.springframework.data.mongodb.core.aggregation.AggregationOperation;
 import org.springframework.data.mongodb.core.aggregation.AggregationOperationContext;
+import org.springframework.lang.Nullable;
 
 /**
+ * String-based aggregation operation for a repository query method.
+ *
  * @author Christoph Strobl
+ * @since 4.2.7
  */
 class StringAggregationOperation implements AggregationOperation {
 
@@ -33,12 +37,16 @@ class StringAggregationOperation implements AggregationOperation {
 	private final String source;
 	private final Class<?> domainType;
 	private final Function<String, Document> bindFunction;
+	private final @Nullable String operator;
 
 	StringAggregationOperation(String source, Class<?> domainType, Function<String, Document> bindFunction) {
 
 		this.source = source;
 		this.domainType = domainType;
 		this.bindFunction = bindFunction;
+
+		Matcher matcher = OPERATOR_PATTERN.matcher(source);
+		this.operator = matcher.find() ? matcher.group() : null;
 	}
 
 	@Override
@@ -48,11 +56,6 @@ public Document toDocument(AggregationOperationContext context) {
 
 	@Override
 	public String getOperator() {
-
-		Matcher matcher = OPERATOR_PATTERN.matcher(source);
-		if (matcher.find()) {
-			return matcher.group();
-		}
-		return AggregationOperation.super.getOperator();
+		return operator != null ? operator : AggregationOperation.super.getOperator();
 	}
 }

spring-data-mongodb/src/test/java/org/springframework/data/mongodb/repository/query/StringBasedAggregationOperationUnitTests.java

@@ -15,7 +15,7 @@
  */
 package org.springframework.data.mongodb.repository.query;
 
-import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.*;
 
 import org.assertj.core.api.Assertions;
 import org.bson.Document;
@@ -24,27 +24,28 @@
 import org.junit.jupiter.params.provider.ValueSource;
 
 /**
+ * Unit tests for {@link StringBasedAggregation}.
+ *
  * @author Christoph Strobl
  */
 public class StringBasedAggregationOperationUnitTests {
 
-    @ParameterizedTest // GH-4712
-    @ValueSource(strings = { "$project", "'$project'", "\"$project\"" })
-    void extractsAggregationOperatorFromAggregationStringWithoutBindingParameters(String operator) {
+	@ParameterizedTest // GH-4712
+	@ValueSource(strings = { "$project", "'$project'", "\"$project\"" })
+	void extractsAggregationOperatorFromAggregationStringWithoutBindingParameters(String operator) {
 
-        StringAggregationOperation agg = new StringAggregationOperation("{ %s : { 'fn' : 1 } }".formatted(operator),
-            Object.class, (it) -> Assertions.fail("o_O Parameter binding"));
+		StringAggregationOperation agg = new StringAggregationOperation("{ %s : { 'fn' : 1 } }".formatted(operator),
+				Object.class, (it) -> Assertions.fail("o_O Parameter binding"));
 
-        assertThat(agg.getOperator()).isEqualTo("$project");
-    }
+		assertThat(agg.getOperator()).isEqualTo("$project");
+	}
 
-    @Test
-        // GH-4712
-    void fallbackToParameterBindingIfAggregationOperatorCannotBeExtractedFromAggregationStringWithoutBindingParameters() {
+	@Test // GH-4712
+	void fallbackToParameterBindingIfAggregationOperatorCannotBeExtractedFromAggregationStringWithoutBindingParameters() {
 
-        StringAggregationOperation agg = new StringAggregationOperation("{ happy-madison : { 'fn' : 1 } }", Object.class,
-            (it) -> new Document("$project", ""));
+		StringAggregationOperation agg = new StringAggregationOperation("{ happy-madison : { 'fn' : 1 } }", Object.class,
+				(it) -> new Document("$project", ""));
 
-        assertThat(agg.getOperator()).isEqualTo("$project");
-    }
+		assertThat(agg.getOperator()).isEqualTo("$project");
+	}
 }