DATAMONGO-2403 - Polishing.

Use handle(…) to skip values instead of flatMap(…) to reduce overhead.

Original pull request: #804.

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

@@ -36,6 +36,7 @@ import org.springframework.data.repository.query.ParameterAccessor;
 import org.springframework.data.repository.query.QueryMethodEvaluationContextProvider;
 import org.springframework.data.repository.query.RepositoryQuery;
 import org.springframework.data.repository.query.ResultProcessor;
+import org.springframework.data.util.TypeInformation;
 import org.springframework.expression.spel.standard.SpelExpressionParser;
 import org.springframework.lang.Nullable;
 import org.springframework.util.Assert;
@@ -117,17 +118,18 @@ public abstract class AbstractReactiveMongoQuery implements RepositoryQuery {
 
 	private Object execute(MongoParameterAccessor parameterAccessor) {
 
-		ConvertingParameterAccessor convertingParamterAccessor = new ConvertingParameterAccessor(operations.getConverter(),
+		ConvertingParameterAccessor accessor = new ConvertingParameterAccessor(operations.getConverter(),
 				parameterAccessor);
 
-		ResultProcessor processor = method.getResultProcessor().withDynamicProjection(convertingParamterAccessor);
+		TypeInformation<?> returnType = method.getReturnType();
+		ResultProcessor processor = method.getResultProcessor().withDynamicProjection(accessor);
 		Class<?> typeToRead = processor.getReturnedType().getTypeToRead();
 
-		if(typeToRead == null && method.getReturnType().getComponentType() != null) {
-			typeToRead = method.getReturnType().getComponentType().getType();
+		if (typeToRead == null && returnType.getComponentType() != null) {
+			typeToRead = returnType.getComponentType().getType();
 		}
 
-		return doExecute(method, processor, convertingParamterAccessor, typeToRead);
+		return doExecute(method, processor, accessor, typeToRead);
 	}
 
 	/**

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

@@ -16,11 +16,11 @@
 package org.springframework.data.mongodb.repository.query;
 
 import reactor.core.publisher.Flux;
-import reactor.core.publisher.Mono;
 
 import java.util.List;
 
 import org.bson.Document;
+
 import org.springframework.data.mongodb.core.ReactiveMongoOperations;
 import org.springframework.data.mongodb.core.aggregation.Aggregation;
 import org.springframework.data.mongodb.core.aggregation.AggregationOperation;
@@ -95,8 +95,14 @@ public class ReactiveStringBasedAggregation extends AbstractReactiveMongoQuery {
 		Flux<?> flux = reactiveMongoOperations.aggregate(aggregation, targetType);
 
 		if (isSimpleReturnType && !isRawReturnType) {
-			flux = flux.flatMap(
-					it -> Mono.justOrEmpty(AggregationUtils.extractSimpleTypeResult((Document) it, typeToRead, mongoConverter)));
+			flux = flux.handle((it, sink) -> {
+
+				Object result = AggregationUtils.extractSimpleTypeResult((Document) it, typeToRead, mongoConverter);
+
+				if (result != null) {
+					sink.next(result);
+				}
+			});
 		}
 
 		if (method.isCollectionQuery()) {

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

@@ -546,7 +546,7 @@ public class ReactiveMongoRepositoryTests {
 	}
 
 	@Test // DATAMONGO-2403
-	public void annotatedAggregationExtractingSimpleValueEmitsEmptyMonoForEmptyDocument() {
+	public void annotatedAggregationExtractingSimpleValueIsEmptyForEmptyDocument() {
 
 		Person p = new Person("project-on-lastanme", null);
 		repository.save(p).then().as(StepVerifier::create).verifyComplete();