DATAMONGO-2012 - Polishing.

Simplify conditional flow. Replace AtomicReference construction in ChangeStreamEvent with AtomicReferenceFieldUpdater usage to reduce object allocations to streamline lazy body conversion usage. Tweak Javadoc and reference docs.

Original pull request: spring-projects#576.

Author: mp911de

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

@@ -18,7 +18,7 @@
 import lombok.EqualsAndHashCode;
 
 import java.time.Instant;
-import java.util.concurrent.atomic.AtomicReference;
+import java.util.concurrent.atomic.AtomicReferenceFieldUpdater;
 
 import org.bson.BsonValue;
 import org.bson.Document;
@@ -41,11 +41,17 @@
 @EqualsAndHashCode
 public class ChangeStreamEvent<T> {
 
+	@SuppressWarnings("rawtypes") //
+	private static final AtomicReferenceFieldUpdater<ChangeStreamEvent, Object> CONVERTED_UPDATER = AtomicReferenceFieldUpdater
+			.newUpdater(ChangeStreamEvent.class, Object.class, "converted");
+
 	private final @Nullable ChangeStreamDocument<Document> raw;
 
 	private final Class<T> targetType;
 	private final MongoConverter converter;
-	private final AtomicReference<T> converted = new AtomicReference<>();
+
+	// accessed through CONVERTED_UPDATER.
+	private volatile @Nullable T converted;
 
 	/**
 	 * @param raw can be {@literal null}.
@@ -77,7 +83,7 @@ public ChangeStreamDocument<Document> getRaw() {
 	 */
 	@Nullable
 	public Instant getTimestamp() {
-		return raw != null ? Instant.ofEpochMilli(raw.getClusterTime().getValue()) : null;
+		return raw != null && raw.getClusterTime() != null ? Instant.ofEpochMilli(raw.getClusterTime().getValue()) : null;
 	}
 
 	/**
@@ -133,36 +139,48 @@ public T getBody() {
 			return null;
 		}
 
-		if (raw.getFullDocument() == null) {
-			return targetType.cast(raw.getFullDocument());
+		Document fullDocument = raw.getFullDocument();
+
+		if (fullDocument == null) {
+			return targetType.cast(fullDocument);
 		}
 
-		return getConverted();
+		return getConverted(fullDocument);
 	}
 
-	private T getConverted() {
+	@SuppressWarnings("unchecked")
+	private T getConverted(Document fullDocument) {
+		return (T) doGetConverted(fullDocument);
+	}
+
+	private Object doGetConverted(Document fullDocument) {
+
+		Object result = CONVERTED_UPDATER.get(this);
 
-		T result = converted.get();
 		if (result != null) {
 			return result;
 		}
 
-		if (ClassUtils.isAssignable(Document.class, raw.getFullDocument().getClass())) {
+		if (ClassUtils.isAssignable(Document.class, fullDocument.getClass())) {
 
-			result = converter.read(targetType, raw.getFullDocument());
-			return converted.compareAndSet(null, result) ? result : converted.get();
+			result = converter.read(targetType, fullDocument);
+			return CONVERTED_UPDATER.compareAndSet(this, null, result) ? result : CONVERTED_UPDATER.get(this);
 		}
 
-		if (converter.getConversionService().canConvert(raw.getFullDocument().getClass(), targetType)) {
+		if (converter.getConversionService().canConvert(fullDocument.getClass(), targetType)) {
 
-			result = converter.getConversionService().convert(raw.getFullDocument(), targetType);
-			return converted.compareAndSet(null, result) ? result : converted.get();
+			result = converter.getConversionService().convert(fullDocument, targetType);
+			return CONVERTED_UPDATER.compareAndSet(this, null, result) ? result : CONVERTED_UPDATER.get(this);
 		}
 
 		throw new IllegalArgumentException(String.format("No converter found capable of converting %s to %s",
-				raw.getFullDocument().getClass(), targetType));
+				fullDocument.getClass(), targetType));
 	}
 
+	/*
+	 * (non-Javadoc)
+	 * @see java.lang.Object#toString()
+	 */
 	@Override
 	public String toString() {
 		return "ChangeStreamEvent {" + "raw=" + raw + ", targetType=" + targetType + '}';

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

@@ -26,6 +26,7 @@
 import org.reactivestreams.Publisher;
 import org.reactivestreams.Subscription;
 import org.springframework.data.geo.GeoResult;
+import org.springframework.data.mongodb.ReactiveMongoDatabaseFactory;
 import org.springframework.data.mongodb.core.aggregation.Aggregation;
 import org.springframework.data.mongodb.core.aggregation.AggregationOptions;
 import org.springframework.data.mongodb.core.aggregation.TypedAggregation;
@@ -1356,9 +1357,9 @@ <S, T> Mono<T> findAndReplace(Query query, S replacement, FindAndReplaceOptions
 	<T> Flux<T> tail(Query query, Class<T> entityClass, String collectionName);
 
 	/**
-	 * Subscribe to a MongoDB <a href="https://docs.mongodb.com/manual/changeStreams/">Change Streams</a> for all events
-	 * in the configured default database via the reactive infrastructure. Use the optional provided {@link Aggregation}
-	 * to filter events. The stream will not be completed unless the {@link org.reactivestreams.Subscription} is
+	 * Subscribe to a MongoDB <a href="https://docs.mongodb.com/manual/changeStreams/">Change Stream</a> for all events in
+	 * the configured default database via the reactive infrastructure. Use the optional provided {@link Aggregation} to
+	 * filter events. The stream will not be completed unless the {@link org.reactivestreams.Subscription} is
 	 * {@link Subscription#cancel() canceled}.
 	 * <p />
 	 * The {@link ChangeStreamEvent#getBody()} is mapped to the {@literal resultType} while the
@@ -1372,14 +1373,16 @@ <S, T> Mono<T> findAndReplace(Query query, S replacement, FindAndReplaceOptions
 	 * @param <T>
 	 * @return the {@link Flux} emitting {@link ChangeStreamEvent events} as they arrive.
 	 * @since 2.1
+	 * @see ReactiveMongoDatabaseFactory#getMongoDatabase()
+	 * @see ChangeStreamOptions#getFilter()
 	 */
 	default <T> Flux<ChangeStreamEvent<T>> changeStream(ChangeStreamOptions options, Class<T> targetType) {
 		return changeStream(null, options, targetType);
 	}
 
 	/**
-	 * Subscribe to a MongoDB <a href="https://docs.mongodb.com/manual/changeStreams/">Change Streams</a> for all events
-	 * in the given collection via the reactive infrastructure. Use the optional provided {@link Aggregation} to filter
+	 * Subscribe to a MongoDB <a href="https://docs.mongodb.com/manual/changeStreams/">Change Stream</a> for all events in
+	 * the given collection via the reactive infrastructure. Use the optional provided {@link Aggregation} to filter
 	 * events. The stream will not be completed unless the {@link org.reactivestreams.Subscription} is
 	 * {@link Subscription#cancel() canceled}.
 	 * <p />
@@ -1389,12 +1392,13 @@ default <T> Flux<ChangeStreamEvent<T>> changeStream(ChangeStreamOptions options,
 	 * Use {@link ChangeStreamOptions} to set arguments like {@link ChangeStreamOptions#getResumeToken() the resumseToken}
 	 * for resuming change streams.
 	 *
-	 * @param collectionName the collection to watch. Can be {@literal null}, watches all collections if so.
+	 * @param collectionName the collection to watch. Can be {@literal null} to watch all collections.
 	 * @param options must not be {@literal null}. Use {@link ChangeStreamOptions#empty()}.
 	 * @param targetType the result type to use.
 	 * @param <T>
 	 * @return the {@link Flux} emitting {@link ChangeStreamEvent events} as they arrive.
 	 * @since 2.1
+	 * @see ChangeStreamOptions#getFilter()
 	 */
 	default <T> Flux<ChangeStreamEvent<T>> changeStream(@Nullable String collectionName, ChangeStreamOptions options,
 			Class<T> targetType) {
@@ -1403,7 +1407,7 @@ default <T> Flux<ChangeStreamEvent<T>> changeStream(@Nullable String collectionN
 	}
 
 	/**
-	 * Subscribe to a MongoDB <a href="https://docs.mongodb.com/manual/changeStreams/">Change Streams</a> via the reactive
+	 * Subscribe to a MongoDB <a href="https://docs.mongodb.com/manual/changeStreams/">Change Stream</a> via the reactive
 	 * infrastructure. Use the optional provided {@link Aggregation} to filter events. The stream will not be completed
 	 * unless the {@link org.reactivestreams.Subscription} is {@link Subscription#cancel() canceled}.
 	 * <p />
@@ -1420,6 +1424,7 @@ default <T> Flux<ChangeStreamEvent<T>> changeStream(@Nullable String collectionN
 	 * @param <T>
 	 * @return the {@link Flux} emitting {@link ChangeStreamEvent events} as they arrive.
 	 * @since 2.1
+	 * @see ChangeStreamOptions#getFilter()
 	 */
 	<T> Flux<ChangeStreamEvent<T>> changeStream(@Nullable String database, @Nullable String collectionName,
 			ChangeStreamOptions options, Class<T> targetType);

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

@@ -2028,11 +2028,8 @@ public <T> Flux<ChangeStreamEvent<T>> changeStream(@Nullable String database, @N
 
 	List<Document> prepareFilter(ChangeStreamOptions options) {
 
-		if (!options.getFilter().isPresent()) {
-			return Collections.emptyList();
-		}
+		Object filter = options.getFilter().orElse(Collections.emptyList());
 
-		Object filter = options.getFilter().get();
 		if (filter instanceof Aggregation) {
 			Aggregation agg = (Aggregation) filter;
 			AggregationOperationContext context = agg instanceof TypedAggregation
@@ -2042,12 +2039,14 @@ List<Document> prepareFilter(ChangeStreamOptions options) {
 
 			return agg.toPipeline(new PrefixingDelegatingAggregationOperationContext(context, "fullDocument",
 					Arrays.asList("operationType", "fullDocument", "documentKey", "updateDescription", "ns")));
-		} else if (filter instanceof List) {
+		}
+
+		if (filter instanceof List) {
 			return (List<Document>) filter;
-		} else {
-			throw new IllegalArgumentException(
-					"ChangeStreamRequestOptions.filter mut be either an Aggregation or a plain list of Documents");
 		}
+
+		throw new IllegalArgumentException(
+				"ChangeStreamRequestOptions.filter mut be either an Aggregation or a plain list of Documents");
 	}
 
 	/*

spring-data-mongodb/src/main/java/org/springframework/data/mongodb/core/messaging/ChangeStreamRequest.java

@@ -46,13 +46,13 @@
  *
  * or {@link com.mongodb.client.MongoDatabase} which receives events from all {@link com.mongodb.client.MongoCollection
  * collections} in that database.
- * 
+ *
  * <pre>
  * <code>
  *     ChangeStreamRequest<Document> request = new ChangeStreamRequest<>(System.out::println, RequestOptions.justDatabase("test"));
  * </code>
  * </pre>
- * 
+ *
  * For more advanced scenarios {@link ChangeStreamOptions} offers abstractions for options like filtering, resuming,...
  *
  * <pre>

spring-data-mongodb/src/main/java/org/springframework/data/mongodb/core/messaging/ChangeStreamTask.java

@@ -58,6 +58,7 @@
  * {@link Task} implementation for obtaining {@link ChangeStreamDocument ChangeStreamDocuments} from MongoDB.
  *
  * @author Christoph Strobl
+ * @author Mark Paluch
  * @since 2.1
  */
 class ChangeStreamTask extends CursorReadingTask<ChangeStreamDocument<Document>, Object> {
@@ -77,7 +78,7 @@ class ChangeStreamTask extends CursorReadingTask<ChangeStreamDocument<Document>,
 		mongoConverter = template.getConverter();
 	}
 
-	/* 
+	/*
 	 * (non-Javadoc)
 	 * @see org.springframework.data.mongodb.core.messaging.CursorReadingTask#initCursor(org.springframework.data.mongodb.core.MongoTemplate, org.springframework.data.mongodb.core.messaging.SubscriptionRequest.RequestOptions, java.lang.Class)
 	 */
@@ -114,9 +115,8 @@ protected MongoCursor<ChangeStreamDocument<Document>> initCursor(MongoTemplate t
 					.orElseGet(() -> ClassUtils.isAssignable(Document.class, targetType) ? FullDocument.DEFAULT
 							: FullDocument.UPDATE_LOOKUP);
 
-			if (changeStreamOptions.getResumeTimestamp().isPresent()) {
-				startAt = new BsonTimestamp(changeStreamOptions.getResumeTimestamp().get().toEpochMilli());
-			}
+			startAt = changeStreamOptions.getResumeTimestamp().map(Instant::toEpochMilli).map(BsonTimestamp::new)
+					.orElse(null);
 		}
 
 		MongoDatabase db = StringUtils.hasText(options.getDatabaseName())
@@ -149,13 +149,15 @@ protected MongoCursor<ChangeStreamDocument<Document>> initCursor(MongoTemplate t
 		return iterable.iterator();
 	}
 
+	@SuppressWarnings("unchecked")
 	List<Document> prepareFilter(MongoTemplate template, ChangeStreamOptions options) {
 
 		if (!options.getFilter().isPresent()) {
 			return Collections.emptyList();
 		}
 
-		Object filter = options.getFilter().get();
+		Object filter = options.getFilter().orElse(null);
+
 		if (filter instanceof Aggregation) {
 			Aggregation agg = (Aggregation) filter;
 			AggregationOperationContext context = agg instanceof TypedAggregation
@@ -164,14 +166,20 @@ List<Document> prepareFilter(MongoTemplate template, ChangeStreamOptions options
 					: Aggregation.DEFAULT_CONTEXT;
 
 			return agg.toPipeline(new PrefixingDelegatingAggregationOperationContext(context, "fullDocument", blacklist));
-		} else if (filter instanceof List) {
+		}
+
+		if (filter instanceof List) {
 			return (List<Document>) filter;
-		} else {
-			throw new IllegalArgumentException(
-					"ChangeStreamRequestOptions.filter mut be either an Aggregation or a plain list of Documents");
 		}
+
+		throw new IllegalArgumentException(
+				"ChangeStreamRequestOptions.filter mut be either an Aggregation or a plain list of Documents");
 	}
 
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.mongodb.core.messaging.CursorReadingTask#createMessage(java.lang.Object, java.lang.Class, org.springframework.data.mongodb.core.messaging.SubscriptionRequest.RequestOptions)
+	 */
 	@Override
 	protected Message<ChangeStreamDocument<Document>, Object> createMessage(ChangeStreamDocument<Document> source,
 			Class<Object> targetType, RequestOptions options) {
@@ -202,7 +210,7 @@ static class ChangeStreamEventMessage<T> implements Message<ChangeStreamDocument
 		private final ChangeStreamEvent<T> delegate;
 		private final MessageProperties messageProperties;
 
-		/* 
+		/*
 		 * (non-Javadoc)
 		 * @see org.springframework.data.mongodb.core.messaging.Message#getRaw()
 		 */
@@ -212,7 +220,7 @@ public ChangeStreamDocument<Document> getRaw() {
 			return delegate.getRaw();
 		}
 
-		/* 
+		/*
 		 * (non-Javadoc)
 		 * @see org.springframework.data.mongodb.core.messaging.Message#getBody()
 		 */
@@ -222,7 +230,7 @@ public T getBody() {
 			return delegate.getBody();
 		}
 
-		/* 
+		/*
 		 * (non-Javadoc)
 		 * @see org.springframework.data.mongodb.core.messaging.Message#getProperties()
 		 */
@@ -232,7 +240,7 @@ public MessageProperties getProperties() {
 		}
 
 		/**
-		 * @return the resume token or {@litearl null} if not set.
+		 * @return the resume token or {@literal null} if not set.
 		 * @see ChangeStreamEvent#getResumeToken()
 		 */
 		@Nullable

spring-data-mongodb/src/main/java/org/springframework/data/mongodb/core/messaging/SubscriptionRequest.java

@@ -57,7 +57,7 @@ interface RequestOptions {
 		 * Get the database name of the db.
 		 *
 		 * @return the name of the database to subscribe to. Can be {@literal null} in which case the default
-		 *         {@link MongoDbFactory#getDb()} is used.
+		 *         {@link MongoDbFactory#getDb() database} is used.
 		 */
 		@Nullable
 		default String getDatabaseName() {

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

@@ -56,16 +56,16 @@ public class DefaultMessageListenerContainerTests {
 	public static final String DATABASE_NAME = "change-stream-events";
 	public static final String COLLECTION_NAME = "collection-1";
 	public static final String COLLECTION_2_NAME = "collection-2";
-	MongoDbFactory dbFactory;
 
+	public @Rule TestRule replSet = ReplicaSet.none();
+
+	MongoDbFactory dbFactory;
 	MongoCollection<Document> collection;
-	private MongoCollection<Document> collection2;
+	MongoCollection<Document> collection2;
 
 	private CollectingMessageListener<Object, Object> messageListener;
 	private MongoTemplate template;
 
-	public @Rule TestRule replSet = ReplicaSet.none();
-
 	@Before
 	public void setUp() {