DATAMONGO-2422 - Release version 2.2.3 (Moore SR3).

Reformat code.

Original pull request: #814.

pom.xml

@@ -5,7 +5,7 @@
 
 	<groupId>org.springframework.data</groupId>
 	<artifactId>spring-data-mongodb-parent</artifactId>
-	<version>2.2.1.RELEASE</version>
+	<version>2.2.3.RELEASE</version>
 	<packaging>pom</packaging>
 
 	<name>Spring Data MongoDB</name>
@@ -15,7 +15,7 @@
 	<parent>
 		<groupId>org.springframework.data.build</groupId>
 		<artifactId>spring-data-parent</artifactId>
-		<version>2.2.1.RELEASE</version>
+		<version>2.2.3.RELEASE</version>
 	</parent>
 
 	<modules>
@@ -26,8 +26,8 @@
 	<properties>
 		<project.type>multi</project.type>
 		<dist.id>spring-data-mongodb</dist.id>
-		<springdata.commons>2.2.1.RELEASE</springdata.commons>
-		<mongo>3.11.1</mongo>
+		<springdata.commons>2.2.3.RELEASE</springdata.commons>
+		<mongo>3.11.2</mongo>
 		<mongo.reactivestreams>1.12.0</mongo.reactivestreams>
 		<jmh.version>1.19</jmh.version>
 	</properties>

spring-data-mongodb-benchmarks/pom.xml

@@ -7,7 +7,7 @@
 	<parent>
 		<groupId>org.springframework.data</groupId>
 		<artifactId>spring-data-mongodb-parent</artifactId>
-		<version>2.2.1.RELEASE</version>
+		<version>2.2.3.RELEASE</version>
 		<relativePath>../pom.xml</relativePath>
 	</parent>
 

spring-data-mongodb-distribution/pom.xml

@@ -14,7 +14,7 @@
 	<parent>
 		<groupId>org.springframework.data</groupId>
 		<artifactId>spring-data-mongodb-parent</artifactId>
-		<version>2.2.1.RELEASE</version>
+		<version>2.2.3.RELEASE</version>
 		<relativePath>../pom.xml</relativePath>
 	</parent>
 

spring-data-mongodb/pom.xml

@@ -11,7 +11,7 @@
 	<parent>
 		<groupId>org.springframework.data</groupId>
 		<artifactId>spring-data-mongodb-parent</artifactId>
-		<version>2.2.1.RELEASE</version>
+		<version>2.2.3.RELEASE</version>
 		<relativePath>../pom.xml</relativePath>
 	</parent>
 

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

@@ -239,11 +239,20 @@ public class MappingMongoConverter extends AbstractMongoConverter implements App
 			return conversionService.convert(bson, rawType);
 		}
 
-		if (DBObject.class.isAssignableFrom(rawType)) {
+		if (Document.class.isAssignableFrom(rawType)) {
 			return (S) bson;
 		}
 
-		if (Document.class.isAssignableFrom(rawType)) {
+		if (DBObject.class.isAssignableFrom(rawType)) {
+
+			if (bson instanceof DBObject) {
+				return (S) bson;
+			}
+
+			if (bson instanceof Document) {
+				return (S) new BasicDBObject((Document) bson);
+			}
+
 			return (S) bson;
 		}
 
@@ -273,9 +282,8 @@ public class MappingMongoConverter extends AbstractMongoConverter implements App
 			if (codecRegistryProvider != null) {
 
 				Optional<? extends Codec<? extends S>> codec = codecRegistryProvider.getCodecFor(rawType);
-				if(codec.isPresent()) {
-					return codec.get().decode(new JsonReader(target.toJson()),
-							DecoderContext.builder().build());
+				if (codec.isPresent()) {
+					return codec.get().decode(new JsonReader(target.toJson()), DecoderContext.builder().build());
 				}
 			}
 

spring-data-mongodb/src/main/java/org/springframework/data/mongodb/gridfs/DataBufferPublisherAdapter.java

@@ -62,7 +62,7 @@ class DataBufferPublisherAdapter {
 
 	/**
 	 * Use an {@link AsyncInputStreamHandler} to read data from the given {@link AsyncInputStream}.
-	 * 
+	 *
 	 * @param inputStream the source stream.
 	 * @return a {@link Flux} emitting data chunks one by one.
 	 * @since 2.2.1
@@ -71,7 +71,6 @@ class DataBufferPublisherAdapter {
 
 		AsyncInputStreamHandler streamHandler = new AsyncInputStreamHandler(inputStream, inputStream.dataBufferFactory,
 				inputStream.bufferSize);
-
 		return Flux.create((sink) -> {
 
 			sink.onDispose(streamHandler::close);
@@ -87,7 +86,7 @@ class DataBufferPublisherAdapter {
 	 * An {@link AsyncInputStream} also holding a {@link DataBufferFactory} and default {@literal bufferSize} for reading
 	 * from it, delegating operations on the {@link AsyncInputStream} to the reference instance. <br />
 	 * Used to pass on the {@link AsyncInputStream} and parameters to avoid capturing lambdas.
-	 * 
+	 *
 	 * @author Christoph Strobl
 	 * @since 2.2.1
 	 */
@@ -146,12 +145,18 @@ class DataBufferPublisherAdapter {
 		private static final AtomicIntegerFieldUpdater<AsyncInputStreamHandler> STATE = AtomicIntegerFieldUpdater
 				.newUpdater(AsyncInputStreamHandler.class, "state");
 
+		private static final AtomicIntegerFieldUpdater<AsyncInputStreamHandler> DRAIN = AtomicIntegerFieldUpdater
+				.newUpdater(AsyncInputStreamHandler.class, "drain");
+
 		private static final AtomicIntegerFieldUpdater<AsyncInputStreamHandler> READ = AtomicIntegerFieldUpdater
 				.newUpdater(AsyncInputStreamHandler.class, "read");
 
 		private static final int STATE_OPEN = 0;
 		private static final int STATE_CLOSED = 1;
 
+		private static final int DRAIN_NONE = 0;
+		private static final int DRAIN_COMPLETION = 1;
+
 		private static final int READ_NONE = 0;
 		private static final int READ_IN_PROGRESS = 1;
 
@@ -165,6 +170,9 @@ class DataBufferPublisherAdapter {
 		// see STATE
 		volatile int state = STATE_OPEN;
 
+		// see DRAIN
+		volatile int drain = DRAIN_NONE;
+
 		// see READ_IN_PROGRESS
 		volatile int read = READ_NONE;
 
@@ -209,6 +217,14 @@ class DataBufferPublisherAdapter {
 			STATE.compareAndSet(this, STATE_OPEN, STATE_CLOSED);
 		}
 
+		boolean enterDrainLoop() {
+			return DRAIN.compareAndSet(this, DRAIN_NONE, DRAIN_COMPLETION);
+		}
+
+		void leaveDrainLoop() {
+			DRAIN.set(this, DRAIN_NONE);
+		}
+
 		boolean isClosed() {
 			return STATE.get(this) == STATE_CLOSED;
 		}
@@ -235,7 +251,6 @@ class DataBufferPublisherAdapter {
 			private final FluxSink<DataBuffer> sink;
 			private final DataBufferFactory factory;
 			private final ByteBuffer transport;
-			private final Thread subscribeThread = Thread.currentThread();
 			private volatile Subscription subscription;
 
 			BufferCoreSubscriber(FluxSink<DataBuffer> sink, DataBufferFactory factory, ByteBuffer transport) {
@@ -261,8 +276,6 @@ class DataBufferPublisherAdapter {
 			public void onNext(Integer bytes) {
 
 				if (isClosed()) {
-
-					onReadDone();
 					return;
 				}
 
@@ -273,13 +286,9 @@ class DataBufferPublisherAdapter {
 					decrementDemand();
 				}
 
-				try {
-					if (bytes == -1) {
-						sink.complete();
-						return;
-					}
-				} finally {
-					onReadDone();
+				if (bytes == -1) {
+					sink.complete();
+					return;
 				}
 
 				subscription.request(1);
@@ -306,15 +315,25 @@ class DataBufferPublisherAdapter {
 					return;
 				}
 
-				onReadDone();
+				close();
 				sink.error(t);
 			}
 
 			@Override
 			public void onComplete() {
 
-				if (subscribeThread != Thread.currentThread()) {
-					drainLoop(sink);
+				onReadDone();
+
+				if (!isClosed()) {
+
+					if (enterDrainLoop()) {
+						try {
+							drainLoop(sink);
+						} finally {
+							leaveDrainLoop();
+						}
+					}
+
 				}
 			}
 		}

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,13 +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();
 
-		return doExecute(method, processor, convertingParamterAccessor, typeToRead);
+		if (typeToRead == null && returnType.getComponentType() != null) {
+			typeToRead = returnType.getComponentType().getType();
+		}
+
+		return doExecute(method, processor, accessor, typeToRead);
 	}
 
 	/**

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

@@ -113,7 +113,7 @@ class AggregationUtils {
 			SpelExpressionParser expressionParser, QueryMethodEvaluationContextProvider evaluationContextProvider) {
 
 		ParameterBindingContext bindingContext = new ParameterBindingContext((accessor::getBindableValue), expressionParser,
-				evaluationContextProvider.getEvaluationContext(method.getParameters(), accessor.getValues()));
+				() -> evaluationContextProvider.getEvaluationContext(method.getParameters(), accessor.getValues()));
 
 		List<AggregationOperation> target = new ArrayList<>(method.getAnnotatedAggregation().length);
 		for (String source : method.getAnnotatedAggregation()) {

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

@@ -73,7 +73,7 @@ class CollationUtils {
 		if (StringUtils.trimLeadingWhitespace(collationExpression).startsWith("{")) {
 
 			ParameterBindingContext bindingContext = new ParameterBindingContext((accessor::getBindableValue),
-					expressionParser, evaluationContextProvider.getEvaluationContext(parameters, accessor.getValues()));
+					expressionParser, () -> evaluationContextProvider.getEvaluationContext(parameters, accessor.getValues()));
 
 			return Collation.from(CODEC.decode(collationExpression, bindingContext));
 		}

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

@@ -95,7 +95,14 @@ public class ReactiveStringBasedAggregation extends AbstractReactiveMongoQuery {
 		Flux<?> flux = reactiveMongoOperations.aggregate(aggregation, targetType);
 
 		if (isSimpleReturnType && !isRawReturnType) {
-			flux = flux.map(it -> 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/main/java/org/springframework/data/mongodb/repository/query/ReactiveStringBasedMongoQuery.java

@@ -117,7 +117,7 @@ public class ReactiveStringBasedMongoQuery extends AbstractReactiveMongoQuery {
 	protected Query createQuery(ConvertingParameterAccessor accessor) {
 
 		ParameterBindingContext bindingContext = new ParameterBindingContext((accessor::getBindableValue), expressionParser,
-				evaluationContextProvider.getEvaluationContext(getQueryMethod().getParameters(), accessor.getValues()));
+				() -> evaluationContextProvider.getEvaluationContext(getQueryMethod().getParameters(), accessor.getValues()));
 
 		Document queryObject = CODEC.decode(this.query, bindingContext);
 		Document fieldsObject = CODEC.decode(this.fieldSpec, bindingContext);

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

@@ -116,7 +116,7 @@ public class StringBasedMongoQuery extends AbstractMongoQuery {
 	protected Query createQuery(ConvertingParameterAccessor accessor) {
 
 		ParameterBindingContext bindingContext = new ParameterBindingContext((accessor::getBindableValue), expressionParser,
-				evaluationContextProvider.getEvaluationContext(getQueryMethod().getParameters(), accessor.getValues()));
+				() -> evaluationContextProvider.getEvaluationContext(getQueryMethod().getParameters(), accessor.getValues()));
 
 		Document queryObject = CODEC.decode(this.query, bindingContext);
 		Document fieldsObject = CODEC.decode(this.fieldSpec, bindingContext);

spring-data-mongodb/src/main/java/org/springframework/data/mongodb/util/json/ParameterBindingContext.java

@@ -15,28 +15,54 @@
  */
 package org.springframework.data.mongodb.util.json;
 
-import lombok.Getter;
-import lombok.RequiredArgsConstructor;
+import java.util.function.Supplier;
 
+import org.springframework.data.util.Lazy;
 import org.springframework.expression.EvaluationContext;
 import org.springframework.expression.Expression;
 import org.springframework.expression.spel.standard.SpelExpressionParser;
 import org.springframework.lang.Nullable;
 
 /**
- * Reusable context for binding parameters to an placeholder or a SpEL expression within a JSON structure. <br />
+ * Reusable context for binding parameters to a placeholder or a SpEL expression within a JSON structure. <br />
  * To be used along with {@link ParameterBindingDocumentCodec#decode(String, ParameterBindingContext)}.
  *
  * @author Christoph Strobl
  * @since 2.2
  */
-@RequiredArgsConstructor
-@Getter
 public class ParameterBindingContext {
 
 	private final ValueProvider valueProvider;
 	private final SpelExpressionParser expressionParser;
-	private final EvaluationContext evaluationContext;
+	private final Lazy<EvaluationContext> evaluationContext;
+
+	/**
+	 * @param valueProvider
+	 * @param expressionParser
+	 * @param evaluationContext
+	 * @deprecated since 2.2.3 - Please use
+	 *             {@link #ParameterBindingContext(ValueProvider, SpelExpressionParser, Supplier)} instead.
+	 */
+	@Deprecated
+	public ParameterBindingContext(ValueProvider valueProvider, SpelExpressionParser expressionParser,
+			EvaluationContext evaluationContext) {
+
+		this(valueProvider, expressionParser, () -> evaluationContext);
+	}
+
+	/**
+	 * @param valueProvider
+	 * @param expressionParser
+	 * @param evaluationContext a {@link Supplier} for {@link Lazy} context retrieval.
+	 * @since 2.2.3
+	 */
+	public ParameterBindingContext(ValueProvider valueProvider, SpelExpressionParser expressionParser,
+			Supplier<EvaluationContext> evaluationContext) {
+
+		this.valueProvider = valueProvider;
+		this.expressionParser = expressionParser;
+		this.evaluationContext = evaluationContext instanceof Lazy ? (Lazy) evaluationContext : Lazy.of(evaluationContext);
+	}
 
 	@Nullable
 	public Object bindableValueForIndex(int index) {
@@ -47,6 +73,18 @@ public class ParameterBindingContext {
 	public Object evaluateExpression(String expressionString) {
 
 		Expression expression = expressionParser.parseExpression(expressionString);
-		return expression.getValue(this.evaluationContext, Object.class);
+		return expression.getValue(getEvaluationContext(), Object.class);
+	}
+
+	public EvaluationContext getEvaluationContext() {
+		return this.evaluationContext.get();
+	}
+
+	public SpelExpressionParser getExpressionParser() {
+		return expressionParser;
+	}
+
+	public ValueProvider getValueProvider() {
+		return valueProvider;
 	}
 }

spring-data-mongodb/src/main/java/org/springframework/data/mongodb/util/json/ParameterBindingDocumentCodec.java

@@ -162,7 +162,7 @@ public class ParameterBindingDocumentCodec implements CollectibleCodec<Document>
 	public Document decode(@Nullable String json, Object[] values) {
 
 		return decode(json, new ParameterBindingContext((index) -> values[index], new SpelExpressionParser(),
-				EvaluationContextProvider.DEFAULT.getEvaluationContext(values)));
+				() -> EvaluationContextProvider.DEFAULT.getEvaluationContext(values)));
 	}
 
 	public Document decode(@Nullable String json, ParameterBindingContext bindingContext) {

spring-data-mongodb/src/main/java/org/springframework/data/mongodb/util/json/ParameterBindingJsonReader.java

@@ -26,6 +26,7 @@ import java.util.Calendar;
 import java.util.Date;
 import java.util.Locale;
 import java.util.TimeZone;
+import java.util.function.Supplier;
 import java.util.regex.Matcher;
 import java.util.regex.Pattern;
 
@@ -96,6 +97,15 @@ public class ParameterBindingJsonReader extends AbstractBsonReader {
 	public ParameterBindingJsonReader(String json, ValueProvider accessor, SpelExpressionParser spelExpressionParser,
 			EvaluationContext evaluationContext) {
 
+		this(json, accessor, spelExpressionParser, () -> evaluationContext);
+	}
+
+	/**
+	 * @since 2.2.3
+	 */
+	public ParameterBindingJsonReader(String json, ValueProvider accessor, SpelExpressionParser spelExpressionParser,
+			Supplier<EvaluationContext> evaluationContext) {
+
 		this.scanner = new JsonScanner(json);
 		setContext(new Context(null, BsonContextType.TOP_LEVEL));
 

spring-data-mongodb/src/main/kotlin/org/springframework/data/mongodb/core/ExecutableFindOperationExtensions.kt

@@ -73,7 +73,8 @@ fun <T : Any> ExecutableFindOperation.DistinctWithProjection.asType(resultType:
  * Extension for [ExecutableFindOperation.DistinctWithProjection.as] leveraging reified type parameters.
  *
  * @author Christoph Strobl
+ * @author Mark Paluch
  * @since 2.1
  */
-inline fun <reified T : Any> ExecutableFindOperation.DistinctWithProjection.asType(): ExecutableFindOperation.DistinctWithProjection =
+inline fun <reified T : Any> ExecutableFindOperation.DistinctWithProjection.asType(): ExecutableFindOperation.TerminatingDistinct<T> =
 		`as`(T::class.java)

spring-data-mongodb/src/main/kotlin/org/springframework/data/mongodb/core/ReactiveFindOperationExtensions.kt

@@ -76,7 +76,7 @@ fun <T : Any> ReactiveFindOperation.DistinctWithProjection.asType(resultType: KC
  * @author Christoph Strobl
  * @since 2.1
  */
-inline fun <reified T : Any> ReactiveFindOperation.DistinctWithProjection.asType(): ReactiveFindOperation.DistinctWithProjection =
+inline fun <reified T : Any> ReactiveFindOperation.DistinctWithProjection.asType(): ReactiveFindOperation.TerminatingDistinct<T> =
 		`as`(T::class.java)
 
 /**

spring-data-mongodb/src/test/java/org/springframework/data/mongodb/core/convert/MappingMongoConverterUnitTests.java

@@ -20,6 +20,7 @@ import static org.assertj.core.api.Assertions.*;
 import static org.mockito.Mockito.*;
 import static org.springframework.data.mongodb.core.DocumentTestUtils.*;
 
+import com.mongodb.DBObject;
 import lombok.EqualsAndHashCode;
 import lombok.RequiredArgsConstructor;
 
@@ -77,6 +78,7 @@ import org.springframework.data.util.ClassTypeInformation;
 import org.springframework.test.util.ReflectionTestUtils;
 
 import com.mongodb.BasicDBList;
+import com.mongodb.BasicDBObject;
 import com.mongodb.DBRef;
 
 /**
@@ -2067,6 +2069,15 @@ public class MappingMongoConverterUnitTests {
 		assertThat(target.dateAsObjectId).isEqualTo(new Date(reference.getTimestamp()));
 	}
 
+	@Test // DATAMONGO-2410
+	public void shouldAllowReadingBackDbObject() {
+
+		assertThat(converter.read(BasicDBObject.class, new org.bson.Document("property", "value")))
+				.isEqualTo(new BasicDBObject("property", "value"));
+		assertThat(converter.read(DBObject.class, new org.bson.Document("property", "value")))
+				.isEqualTo(new BasicDBObject("property", "value"));
+	}
+
 	static class GenericType<T> {
 		T content;
 	}

spring-data-mongodb/src/test/java/org/springframework/data/mongodb/core/messaging/DefaultMessageListenerContainerTests.java

@@ -31,21 +31,18 @@ import org.junit.Before;
 import org.junit.Rule;
 import org.junit.Test;
 import org.junit.rules.TestRule;
-
 import org.springframework.core.task.SimpleAsyncTaskExecutor;
 import org.springframework.dao.DataAccessException;
 import org.springframework.data.annotation.Id;
 import org.springframework.data.mongodb.MongoDbFactory;
 import org.springframework.data.mongodb.core.MongoTemplate;
 import org.springframework.data.mongodb.core.SimpleMongoClientDbFactory;
-import org.springframework.data.mongodb.core.SimpleMongoDbFactory;
 import org.springframework.data.mongodb.core.messaging.SubscriptionRequest.RequestOptions;
 import org.springframework.data.mongodb.test.util.MongoTestUtils;
 import org.springframework.data.mongodb.test.util.ReplicaSet;
 import org.springframework.test.annotation.IfProfileValue;
 import org.springframework.util.ErrorHandler;
 
-import com.mongodb.MongoClient;
 import com.mongodb.client.MongoCollection;
 import com.mongodb.client.model.CreateCollectionOptions;
 import com.mongodb.client.model.changestream.ChangeStreamDocument;
@@ -61,6 +58,8 @@ public class DefaultMessageListenerContainerTests {
 	public static final String COLLECTION_NAME = "collection-1";
 	public static final String COLLECTION_2_NAME = "collection-2";
 
+	public static final Duration TIMEOUT = Duration.ofSeconds(2);
+
 	public @Rule TestRule replSet = ReplicaSet.none();
 
 	MongoDbFactory dbFactory;
@@ -94,12 +93,12 @@ public class DefaultMessageListenerContainerTests {
 				Person.class);
 		container.start();
 
-		awaitSubscription(subscription, Duration.ofMillis(500));
+		awaitSubscription(subscription, TIMEOUT);
 
 		collection.insertOne(new Document("_id", "id-1").append("firstname", "foo"));
 		collection.insertOne(new Document("_id", "id-2").append("firstname", "bar"));
 
-		awaitMessages(messageListener, 2, Duration.ofMillis(500));
+		awaitMessages(messageListener, 2, TIMEOUT);
 
 		assertThat(messageListener.getMessages().stream().map(Message::getBody).collect(Collectors.toList()))
 				.containsExactly(new Person("id-1", "foo"), new Person("id-2", "bar"));
@@ -125,12 +124,12 @@ public class DefaultMessageListenerContainerTests {
 		}, () -> COLLECTION_NAME), Person.class, errorHandler);
 		container.start();
 
-		awaitSubscription(subscription, Duration.ofMillis(500));
+		awaitSubscription(subscription, TIMEOUT);
 
 		collection.insertOne(new Document("_id", "id-1").append("firstname", "foo"));
 		collection.insertOne(new Document("_id", "id-2").append("firstname", "bar"));
 
-		awaitMessages(messageListener, 2, Duration.ofMillis(500));
+		awaitMessages(messageListener, 2, TIMEOUT);
 
 		verify(errorHandler, atLeast(1)).handleError(any(IllegalStateException.class));
 		assertThat(messageListener.getTotalNumberMessagesReceived()).isEqualTo(2);
@@ -145,12 +144,12 @@ public class DefaultMessageListenerContainerTests {
 				Document.class);
 		container.start();
 
-		awaitSubscription(subscription, Duration.ofMillis(500));
+		awaitSubscription(subscription, TIMEOUT);
 
 		collection.insertOne(new Document("_id", "id-1").append("value", "foo"));
 		collection.insertOne(new Document("_id", "id-2").append("value", "bar"));
 
-		awaitMessages(messageListener, 2, Duration.ofMillis(500));
+		awaitMessages(messageListener, 2, TIMEOUT);
 
 		container.stop();
 
@@ -174,12 +173,12 @@ public class DefaultMessageListenerContainerTests {
 		Subscription subscription = container.register(new ChangeStreamRequest(messageListener, () -> COLLECTION_NAME),
 				Document.class);
 
-		awaitSubscription(subscription, Duration.ofMillis(500));
+		awaitSubscription(subscription, TIMEOUT);
 
 		Document expected = new Document("_id", "id-2").append("value", "bar");
 		collection.insertOne(expected);
 
-		awaitMessages(messageListener, 1, Duration.ofMillis(500));
+		awaitMessages(messageListener, 1, TIMEOUT);
 		container.stop();
 
 		assertThat(messageListener.getMessages().stream().map(Message::getBody).collect(Collectors.toList()))
@@ -226,11 +225,11 @@ public class DefaultMessageListenerContainerTests {
 
 		awaitSubscription(
 				container.register(new TailableCursorRequest(messageListener, () -> COLLECTION_NAME), Document.class),
-				Duration.ofMillis(500));
+				TIMEOUT);
 
 		collection.insertOne(new Document("_id", "id-2").append("value", "bar"));
 
-		awaitMessages(messageListener, 2, Duration.ofSeconds(2));
+		awaitMessages(messageListener, 2, TIMEOUT);
 		container.stop();
 
 		assertThat(messageListener.getTotalNumberMessagesReceived()).isEqualTo(2);
@@ -247,12 +246,12 @@ public class DefaultMessageListenerContainerTests {
 
 		awaitSubscription(
 				container.register(new TailableCursorRequest(messageListener, () -> COLLECTION_NAME), Document.class),
-				Duration.ofMillis(500));
+				TIMEOUT);
 
 		collection.insertOne(new Document("_id", "id-1").append("value", "foo"));
 		collection.insertOne(new Document("_id", "id-2").append("value", "bar"));
 
-		awaitMessages(messageListener, 2, Duration.ofSeconds(2));
+		awaitMessages(messageListener, 2, TIMEOUT);
 		container.stop();
 
 		assertThat(messageListener.getTotalNumberMessagesReceived()).isEqualTo(2);
@@ -359,7 +358,7 @@ public class DefaultMessageListenerContainerTests {
 
 		container.start();
 
-		awaitSubscription(subscription, Duration.ofMillis(500));
+		awaitSubscription(subscription, TIMEOUT);
 
 		collection.insertOne(new Document("_id", "col-1-id-1").append("firstname", "foo"));
 		collection.insertOne(new Document("_id", "col-1-id-2").append("firstname", "bar"));
@@ -367,7 +366,7 @@ public class DefaultMessageListenerContainerTests {
 		collection2.insertOne(new Document("_id", "col-2-id-1").append("firstname", "bar"));
 		collection2.insertOne(new Document("_id", "col-2-id-2").append("firstname", "foo"));
 
-		awaitMessages(messageListener, 4, Duration.ofMillis(500));
+		awaitMessages(messageListener, 4, TIMEOUT);
 
 		assertThat(messageListener.getMessages().stream().map(Message::getBody).collect(Collectors.toList()))
 				.containsExactly(new Person("col-1-id-1", "foo"), new Person("col-1-id-2", "bar"),

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

@@ -27,6 +27,7 @@ import reactor.core.publisher.Mono;
 import reactor.test.StepVerifier;
 
 import java.util.Arrays;
+import java.util.Map;
 import java.util.concurrent.BlockingQueue;
 import java.util.concurrent.LinkedBlockingDeque;
 import java.util.concurrent.TimeUnit;
@@ -36,7 +37,6 @@ import org.junit.BeforeClass;
 import org.junit.Test;
 import org.junit.runner.RunWith;
 import org.reactivestreams.Publisher;
-
 import org.springframework.beans.factory.BeanFactory;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.context.annotation.Bean;
@@ -535,6 +535,46 @@ public class ReactiveMongoRepositoryTests {
 				}).verifyComplete();
 	}
 
+	@Test // DATAMONGO-2153
+	public void annotatedAggregationWithAggregationResultAsMap() {
+
+		repository.sumAgeAndReturnSumAsMap() //
+				.as(StepVerifier::create) //
+				.consumeNextWith(it -> {
+					assertThat(it).isInstanceOf(Map.class);
+				}).verifyComplete();
+	}
+
+	@Test // DATAMONGO-2403
+	public void annotatedAggregationExtractingSimpleValueIsEmptyForEmptyDocument() {
+
+		Person p = new Person("project-on-lastanme", null);
+		repository.save(p).then().as(StepVerifier::create).verifyComplete();
+
+		repository.projectToLastnameAndRemoveId(p.getFirstname()) //
+				.as(StepVerifier::create) //
+				.verifyComplete();
+	}
+
+	@Test // DATAMONGO-2403
+	public void annotatedAggregationSkipsEmptyDocumentsWhenExtractingSimpleValue() {
+
+		String firstname = "project-on-lastanme";
+
+		Person p1 = new Person(firstname, null);
+		p1.setEmail("p1@example.com");
+		Person p2 = new Person(firstname, "lastname");
+		p2.setEmail("p2@example.com");
+		Person p3 = new Person(firstname, null);
+		p3.setEmail("p3@example.com");
+
+		repository.saveAll(Arrays.asList(p1, p2, p3)).then().as(StepVerifier::create).verifyComplete();
+
+		repository.projectToLastnameAndRemoveId(firstname) //
+				.as(StepVerifier::create) //
+				.expectNext("lastname").verifyComplete();
+	}
+
 	interface ReactivePersonRepository
 			extends ReactiveMongoRepository<Person, String>, ReactiveQuerydslPredicateExecutor<Person> {
 
@@ -596,6 +636,13 @@ public class ReactiveMongoRepositoryTests {
 		@Aggregation(pipeline = "{ '$group' : { '_id' : null, 'total' : { $sum: '$age' } } }")
 		Mono<SumAge> sumAgeAndReturnSumWrapper();
 
+		@Aggregation(pipeline = "{ '$group' : { '_id' : null, 'total' : { $sum: '$age' } } }")
+		Mono<Map> sumAgeAndReturnSumAsMap();
+
+		@Aggregation(
+				pipeline = { "{ '$match' : { 'firstname' : '?0' } }", "{ '$project' : { '_id' : 0, 'lastname' : 1 } }" })
+		Mono<String> projectToLastnameAndRemoveId(String firstname);
+
 		@Query(value = "{_id:?0}")
 		Mono<org.bson.Document> findDocumentById(String id);
 	}

spring-data-mongodb/src/test/java/org/springframework/data/mongodb/util/json/ParameterBindingJsonReaderUnitTests.java

@@ -26,6 +26,10 @@ import java.util.List;
 import org.bson.Document;
 import org.bson.codecs.DecoderContext;
 import org.junit.Test;
+import org.springframework.expression.EvaluationContext;
+import org.springframework.expression.TypedValue;
+import org.springframework.expression.spel.standard.SpelExpressionParser;
+import org.springframework.expression.spel.support.StandardEvaluationContext;
 
 /**
  * Unit tests for {@link ParameterBindingJsonReader}.
@@ -197,6 +201,25 @@ public class ParameterBindingJsonReaderUnitTests {
 		assertThat(target).isEqualTo(Document.parse("{ 'end_date' : { $gte : { $date : " + time + " } } } "));
 	}
 
+	@Test // DATAMONGO-2418
+	public void shouldNotAccessSpElEvaluationContextWhenNoSpElPresentInBindableTarget() {
+
+		Object[] args = new Object[] { "value" };
+		EvaluationContext evaluationContext = new StandardEvaluationContext() {
+
+			@Override
+			public TypedValue getRootObject() {
+				throw new RuntimeException("o_O");
+			}
+		};
+
+		ParameterBindingJsonReader reader = new ParameterBindingJsonReader("{ 'name':'?0' }",
+				new ParameterBindingContext((index) -> args[index], new SpelExpressionParser(), evaluationContext));
+		Document target = new ParameterBindingDocumentCodec().decode(reader, DecoderContext.builder().build());
+
+		assertThat(target).isEqualTo(new Document("name", "value"));
+	}
+
 	private static Document parse(String json, Object... args) {
 
 		ParameterBindingJsonReader reader = new ParameterBindingJsonReader(json, args);

src/main/resources/changelog.txt

@@ -1,6 +1,34 @@
 Spring Data MongoDB Changelog
 =============================
 
+Changes in version 2.2.3.RELEASE (2019-12-04)
+---------------------------------------------
+* DATAMONGO-2430 - Upgrade to mongo-java-driver 3.11.2.
+* DATAMONGO-2422 - Release 2.2.3 (Moore SR3).
+* DATAMONGO-2418 - Application Context Doesn't start with @Query.
+* DATAMONGO-2410 - Using BasicDBObject as an entity caused java.lang.ClassCastException in runtime.
+
+
+Changes in version 2.1.14.RELEASE (2019-12-04)
+----------------------------------------------
+* DATAMONGO-2421 - Release 2.1.14 (Lovelace SR14).
+* DATAMONGO-2410 - Using BasicDBObject as an entity caused java.lang.ClassCastException in runtime.
+
+
+Changes in version 2.2.2.RELEASE (2019-11-18)
+---------------------------------------------
+* DATAMONGO-2414 - ReactiveGridFsResource.getDownloadStream(…) hang if completion happens on event loop.
+* DATAMONGO-2409 - Extension Function ReactiveFindOperation.DistinctWithProjection.asType() has wrong return type.
+* DATAMONGO-2403 - ReactiveStringBasedAggregation / AggregationUtils fails on NPE because source or value is null.
+* DATAMONGO-2402 - Release 2.2.2 (Moore SR2).
+
+
+Changes in version 2.1.13.RELEASE (2019-11-18)
+----------------------------------------------
+* DATAMONGO-2409 - Extension Function ReactiveFindOperation.DistinctWithProjection.asType() has wrong return type.
+* DATAMONGO-2401 - Release 2.1.13 (Lovelace SR13).
+
+
 Changes in version 2.2.1.RELEASE (2019-11-04)
 ---------------------------------------------
 * DATAMONGO-2399 - Upgrade to mongo-java-driver 3.11.1.
@@ -2800,3 +2828,5 @@ Repository
 * Namespace support for Mongo repositories
 * Allow usage of pagination and sorting with repositories
 
+
+

src/main/resources/notice.txt

@@ -1,4 +1,4 @@
-Spring Data MongoDB 2.2.1
+Spring Data MongoDB 2.2.3
 Copyright (c) [2010-2019] Pivotal Software, Inc.
 
 This product is licensed to you under the Apache License, Version 2.0 (the "License").
@@ -8,3 +8,4 @@ This product may include a number of subcomponents with
 separate copyright notices and license terms. Your use of the source
 code for the these subcomponents is subject to the terms and
 conditions of the subcomponent's license, as noted in the LICENSE file.
+