DATAMONGO-2089 - Refactor and add Kotlin extension.

Author: christophstrobl

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

@@ -22,9 +22,7 @@
 
 import org.bson.BsonTimestamp;
 import org.bson.BsonValue;
-import org.bson.Document;
 import org.springframework.data.mongodb.core.ChangeStreamOptions.ChangeStreamOptionsBuilder;
-import org.springframework.data.mongodb.core.ReactiveFindOperation.ReactiveFind;
 import org.springframework.data.mongodb.core.aggregation.Aggregation;
 import org.springframework.data.mongodb.core.query.CriteriaDefinition;
 
@@ -54,22 +52,13 @@
  */
 public interface ReactiveChangeStreamOperation {
 
-	/**
-	 * Start creating a change stream operation.watching all collections within the database. <br />
-	 * Consider limiting events be defining a {@link ChangeStreamWithCollection#watchCollection(String) collection} and/or
-	 * {@link ChangeStreamWithFilter#filter(CriteriaDefinition) filter}.
-	 * 
-	 * @return new instance of {@link ReactiveFind}. Never {@literal null}.
-	 */
-	ReactiveChangeStream<Document> changeStream();
-
 	/**
 	 * Start creating a change stream operation for the given {@literal domainType} watching all collections within the
 	 * database. <br />
 	 * Consider limiting events be defining a {@link ChangeStreamWithCollection#watchCollection(String) collection} and/or
-	 * {@link ChangeStreamWithFilter#filter(CriteriaDefinition) filter}.
+	 * {@link ChangeStreamWithFilterAndProjection#filter(CriteriaDefinition) filter}.
 	 *
-	 * @param domainType must not be {@literal null}.
+	 * @param domainType must not be {@literal null}. Use {@link org.bson.Document} to obtain raw elements.
 	 * @return new instance of {@link ReactiveChangeStream}. Never {@literal null}.
 	 * @throws IllegalArgumentException if domainType is {@literal null}.
 	 */
@@ -102,48 +91,42 @@ interface ChangeStreamWithCollection<T> {
 		 * @return new instance of {@link ChangeStreamWithCollection}.
 		 * @throws IllegalArgumentException if collection is {@literal null}.
 		 */
-		ChangeStreamWithProjection<T> watchCollection(String collection);
-	}
-
-	/**
-	 * Result type override (optional).
-	 */
-	interface ChangeStreamWithProjection<T> extends ChangeStreamWithFilter<T> {
-
-		/**
-		 * Define the target type fields should be mapped to.
-		 *
-		 * @param resultType must not be {@literal null}.
-		 * @param <R> result type.
-		 * @return new instance of {@link ChangeStreamWithProjection}.
-		 * @throws IllegalArgumentException if resultType is {@literal null}.
-		 */
-		<R> ChangeStreamWithFilter<R> as(Class<R> resultType);
+		ChangeStreamWithFilterAndProjection<T> watchCollection(String collection);
 	}
 
 	/**
 	 * Provide a filter for limiting results (optional).
 	 */
-	interface ChangeStreamWithFilter<T> extends ResumingChangeStream<T>, TerminatingChangeStream<T> {
+	interface ChangeStreamWithFilterAndProjection<T> extends ResumingChangeStream<T>, TerminatingChangeStream<T> {
 
 		/**
 		 * Use an {@link Aggregation} to filter matching events.
 		 *
 		 * @param by must not be {@literal null}.
-		 * @return new instance of {@link ChangeStreamWithProjection}.
+		 * @return new instance of {@link ChangeStreamWithFilterAndProjection}.
 		 * @throws IllegalArgumentException if the given {@link Aggregation} is {@literal null}.
 		 */
-		ChangeStreamWithProjection<T> filter(Aggregation by);
+		ChangeStreamWithFilterAndProjection<T> filter(Aggregation by);
 
 		/**
 		 * Use a {@link CriteriaDefinition critera} to filter matching events via an
 		 * {@link org.springframework.data.mongodb.core.aggregation.MatchOperation}.
 		 *
 		 * @param by must not be {@literal null}.
-		 * @return new instance of {@link ChangeStreamWithProjection}.
+		 * @return new instance of {@link ChangeStreamWithFilterAndProjection}.
 		 * @throws IllegalArgumentException if the given {@link CriteriaDefinition} is {@literal null}.
 		 */
-		ChangeStreamWithProjection<T> filter(CriteriaDefinition by);
+		ChangeStreamWithFilterAndProjection<T> filter(CriteriaDefinition by);
+
+		/**
+		 * Define the target type fields should be mapped to.
+		 *
+		 * @param resultType must not be {@literal null}.
+		 * @param <R> result type.
+		 * @return new instance of {@link ChangeStreamWithFilterAndProjection}.
+		 * @throws IllegalArgumentException if resultType is {@literal null}.
+		 */
+		<R> ChangeStreamWithFilterAndProjection<R> as(Class<R> resultType);
 	}
 
 	/**
@@ -154,34 +137,34 @@ interface ResumingChangeStream<T> extends TerminatingChangeStream<T> {
 		/**
 		 * Resume the change stream at a given point.
 		 *
-		 * @param beacon an {@link Instant} or {@link BsonTimestamp}
+		 * @param token an {@link Instant} or {@link BsonTimestamp}
 		 * @return new instance of {@link TerminatingChangeStream}.
 		 * @see ChangeStreamOptionsBuilder#resumeAt(Instant)
 		 * @see ChangeStreamOptionsBuilder#resumeAt(BsonTimestamp)
 		 * @throws IllegalArgumentException if the given beacon is neither {@link Instant} nor {@link BsonTimestamp}.
 		 */
-		TerminatingChangeStream<T> resumeAt(Object beacon);
+		TerminatingChangeStream<T> resumeAt(Object token);
 
 		/**
 		 * Resume the change stream after a given point.
 		 *
-		 * @param beacon an {@link Instant} or {@link BsonTimestamp}
+		 * @param token an {@link Instant} or {@link BsonTimestamp}
 		 * @return new instance of {@link TerminatingChangeStream}.
 		 * @see ChangeStreamOptionsBuilder#resumeAfter(BsonValue)
 		 * @see ChangeStreamOptionsBuilder#resumeToken(BsonValue)
 		 * @throws IllegalArgumentException if the given beacon not a {@link BsonValue}.
 		 */
-		TerminatingChangeStream<T> resumeAfter(Object beacon);
+		TerminatingChangeStream<T> resumeAfter(Object token);
 
 		/**
 		 * Start the change stream after a given point.
 		 *
-		 * @param beacon an {@link Instant} or {@link BsonTimestamp}
+		 * @param token an {@link Instant} or {@link BsonTimestamp}
 		 * @return new instance of {@link TerminatingChangeStream}.
 		 * @see ChangeStreamOptionsBuilder#startAfter(BsonValue) (BsonValue)
 		 * @throws IllegalArgumentException if the given beacon not a {@link BsonValue}.
 		 */
-		TerminatingChangeStream<T> startAfter(Object beacon);
+		TerminatingChangeStream<T> startAfter(Object token);
 	}
 
 	/**
@@ -204,5 +187,5 @@ interface ChangeStreamWithOptions<T> {
 	 * {@link ReactiveChangeStream} provides methods for constructing change stream operations in a fluent way.
 	 */
 	interface ReactiveChangeStream<T> extends ChangeStreamWithOptions<T>, ChangeStreamWithCollection<T>,
-			TerminatingChangeStream<T>, ResumingChangeStream<T>, ChangeStreamWithFilter<T> {}
+			TerminatingChangeStream<T>, ResumingChangeStream<T>, ChangeStreamWithFilterAndProjection<T> {}
 }

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

@@ -46,15 +46,6 @@ class ReactiveChangeStreamOperationSupport implements ReactiveChangeStreamOperat
 		this.template = template;
 	}
 
-	/*
-	 * (non-Javadoc)
-	 * @see org.springframework.data.mongodb.core.ReactiveChangeStreamOperation#changeStream()
-	 */
-	@Override
-	public ReactiveChangeStream<Document> changeStream() {
-		return new ReactiveChangeStreamSupport(template, null, Document.class, null, null);
-	}
-
 	/*
 	 * (non-Javadoc)
 	 * @see org.springframework.data.mongodb.core.ReactiveChangeStreamOperation#changeStream(java.lang.Class)
@@ -67,7 +58,7 @@ public <T> ReactiveChangeStream<T> changeStream(Class<T> domainType) {
 	}
 
 	static class ReactiveChangeStreamSupport<T>
-			implements ReactiveChangeStream<T>, ChangeStreamWithProjection<T>, ChangeStreamWithFilter<T> {
+			implements ReactiveChangeStream<T>, ChangeStreamWithFilterAndProjection<T> {
 
 		private final ReactiveMongoTemplate template;
 		private final @Nullable Class<?> domainType;
@@ -90,7 +81,7 @@ private ReactiveChangeStreamSupport(ReactiveMongoTemplate template, Class<?> dom
 		 * @see org.springframework.data.mongodb.core.ReactiveChangeStreamOperation.ChangeStreamWithCollection#watchCollection(java.lang.String)
 		 */
 		@Override
-		public ChangeStreamWithProjection<T> watchCollection(String collection) {
+		public ChangeStreamWithFilterAndProjection<T> watchCollection(String collection) {
 
 			Assert.hasText(collection, "Collection name must not be null nor empty!");
 			return new ReactiveChangeStreamSupport<>(template, domainType, returnType, collection, options);
@@ -101,14 +92,14 @@ public ChangeStreamWithProjection<T> watchCollection(String collection) {
 		 * @see org.springframework.data.mongodb.core.ReactiveChangeStreamOperation.ResumingChangeStream#resumeAt(java.lang.Object)
 		 */
 		@Override
-		public TerminatingChangeStream<T> resumeAt(Object beacon) {
+		public TerminatingChangeStream<T> resumeAt(Object token) {
 
 			return withOptions(builder -> {
 
-				if (beacon instanceof Instant) {
-					builder.resumeAt((Instant) beacon);
-				} else if (beacon instanceof BsonTimestamp) {
-					builder.resumeAt((BsonTimestamp) beacon);
+				if (token instanceof Instant) {
+					builder.resumeAt((Instant) token);
+				} else if (token instanceof BsonTimestamp) {
+					builder.resumeAt((BsonTimestamp) token);
 				}
 			});
 		}
@@ -118,21 +109,21 @@ public TerminatingChangeStream<T> resumeAt(Object beacon) {
 		 * @see org.springframework.data.mongodb.core.ReactiveChangeStreamOperation.ResumingChangeStream#resumeAfter(java.lang.Object)
 		 */
 		@Override
-		public TerminatingChangeStream<T> resumeAfter(Object beacon) {
+		public TerminatingChangeStream<T> resumeAfter(Object token) {
 
-			Assert.isInstanceOf(BsonValue.class, beacon, "Beacon must be a BsonValue");
-			return withOptions(builder -> builder.resumeAfter((BsonValue) beacon));
+			Assert.isInstanceOf(BsonValue.class, token, "Token must be a BsonValue");
+			return withOptions(builder -> builder.resumeAfter((BsonValue) token));
 		}
 
 		/*
 		 * (non-Javadoc)
 		 * @see org.springframework.data.mongodb.core.ReactiveChangeStreamOperation.ResumingChangeStream#startAfter(java.lang.Object)
 		 */
 		@Override
-		public TerminatingChangeStream<T> startAfter(Object beacon) {
+		public TerminatingChangeStream<T> startAfter(Object token) {
 
-			Assert.isInstanceOf(BsonValue.class, beacon, "Beacon must be a BsonValue");
-			return withOptions(builder -> builder.startAfter((BsonValue) beacon));
+			Assert.isInstanceOf(BsonValue.class, token, "Token must be a BsonValue");
+			return withOptions(builder -> builder.startAfter((BsonValue) token));
 		}
 
 		/*
@@ -153,7 +144,9 @@ public ReactiveChangeStreamSupport<T> withOptions(Consumer<ChangeStreamOptionsBu
 		 * @see org.springframework.data.mongodb.core.ReactiveChangeStreamOperation.ChangeStreamWithProjection#as(java.lang.Class)
 		 */
 		@Override
-		public <R> ChangeStreamWithFilter<R> as(Class<R> resultType) {
+		public <R> ChangeStreamWithFilterAndProjection<R> as(Class<R> resultType) {
+
+			Assert.notNull(resultType, "ResultType must not be null!");
 			return new ReactiveChangeStreamSupport<>(template, domainType, resultType, collection, options);
 		}
 
@@ -162,19 +155,20 @@ public <R> ChangeStreamWithFilter<R> as(Class<R> resultType) {
 		 * @see org.springframework.data.mongodb.core.ReactiveChangeStreamOperation.ChangeStreamWithFilter#filter(org.springframework.data.mongodb.core.aggregation.Aggregation)
 		 */
 		@Override
-		public ChangeStreamWithProjection<T> filter(Aggregation aggregation) {
-			return withOptions(builder -> builder.filter(aggregation));
+		public ChangeStreamWithFilterAndProjection<T> filter(Aggregation filter) {
+			return withOptions(builder -> builder.filter(filter));
 		}
 
 		/*
 		 * (non-Javadoc)
 		 * @see org.springframework.data.mongodb.core.ReactiveChangeStreamOperation.ChangeStreamWithFilter#filter(org.springframework.data.mongodb.core.query.CriteriaDefinition)
 		 */
 		@Override
-		public ChangeStreamWithProjection<T> filter(CriteriaDefinition by) {
+		public ChangeStreamWithFilterAndProjection<T> filter(CriteriaDefinition by) {
 
 			MatchOperation $match = Aggregation.match(by);
-			Aggregation aggregation = domainType != null ? Aggregation.newAggregation(domainType, $match)
+			Aggregation aggregation = domainType != null && !Document.class.equals(domainType)
+					? Aggregation.newAggregation(domainType, $match)
 					: Aggregation.newAggregation($match);
 			return filter(aggregation);
 		}

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

@@ -2269,15 +2269,6 @@ public <T> ReactiveMapReduce<T> mapReduce(Class<T> domainType) {
 		return new ReactiveMapReduceOperationSupport(this).mapReduce(domainType);
 	}
 
-	/*
-	 * (non-Javadoc)
-	 * @see org.springframework.data.mongodb.core.ReactiveChangeStreamOperation#changeStream()
-	 */
-	@Override
-	public ReactiveChangeStream<Document> changeStream() {
-		return new ReactiveChangeStreamOperationSupport(this).changeStream();
-	}
-
 	/*
 	 * (non-Javadoc)
 	 * @see org.springframework.data.mongodb.core.ReactiveChangeStreamOperation#changeStream(java.lang.Class)

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

@@ -0,0 +1,53 @@
+/*
+ * Copyright 2019 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.springframework.data.mongodb.core
+
+import kotlinx.coroutines.FlowPreview
+import kotlinx.coroutines.flow.Flow
+import kotlinx.coroutines.reactive.flow.asFlow
+
+
+/**
+ * Extension for [RactiveChangeStreamOperation. changeStream] leveraging reified type parameters.
+ *
+ * @author Christoph Strobl
+ * @since 2.2
+ */
+inline fun <reified T : Any> ReactiveChangeStreamOperation.changeStream(): ReactiveChangeStreamOperation.ReactiveChangeStream<T> =
+		changeStream(T::class.java)
+
+/**
+ * Extension for [ReactiveChangeStreamOperation.ChangeStreamWithFilterAndProjection. as] leveraging reified type parameters.
+ *
+ * @author Christoph Strobl
+ * @since 2.2
+ */
+inline fun <reified T : Any> ReactiveChangeStreamOperation.ChangeStreamWithFilterAndProjection<*>.asType(): ReactiveChangeStreamOperation.ChangeStreamWithFilterAndProjection<T> =
+		`as`(T::class.java)
+
+/**
+ * Coroutines [Flow] variant of [ReactiveChangeStreamOperation.TerminatingChangeStream. listen].
+ *
+ * Backpressure is controlled by [batchSize] parameter that controls the size of in-flight elements
+ * and [org.reactivestreams.Subscription.request] size.
+ *
+ * @author Christoph Strobl
+ * @since 2.2
+ */
+@FlowPreview
+fun <T : Any> ReactiveChangeStreamOperation.TerminatingChangeStream<T>.flow(batchSize: Int = 1): Flow<ChangeStreamEvent<T>> =
+		listen().asFlow(batchSize)
+

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

@@ -70,7 +70,7 @@ public void changeStreamEventsShouldBeEmittedCorrectly() {
 
 		BlockingQueue<ChangeStreamEvent<Document>> documents = new LinkedBlockingQueue<>(100);
 
-		Disposable disposable = template.changeStream() //
+		Disposable disposable = template.changeStream(Document.class) //
 				.watchCollection("person") //
 				.listen() //
 				.doOnNext(documents::add).subscribe();

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

@@ -59,7 +59,7 @@ public void setUp() {
 	@Test // DATAMONGO-2089
 	public void listenWithoutDomainTypeUsesDocumentAsDefault() {
 
-		changeStreamSupport.changeStream().listen().subscribe();
+		changeStreamSupport.changeStream(Document.class).listen().subscribe();
 
 		verify(template).changeStream(isNull(), eq(ChangeStreamOptions.empty()), eq(Document.class));
 	}
@@ -104,7 +104,7 @@ public void listenWithDomainTypeCreatesTypedAggregation() {
 	public void listenWithoutDomainTypeCreatesUntypedAggregation() {
 
 		Criteria criteria = where("operationType").is("insert");
-		changeStreamSupport.changeStream().filter(criteria).listen().subscribe();
+		changeStreamSupport.changeStream(Document.class).filter(criteria).listen().subscribe();
 
 		ArgumentCaptor<ChangeStreamOptions> optionsArgumentCaptor = ArgumentCaptor.forClass(ChangeStreamOptions.class);
 		verify(template).changeStream(isNull(), optionsArgumentCaptor.capture(), eq(Document.class));
@@ -126,7 +126,7 @@ public void optionsShouldBePassedOnCorrectly() {
 
 		Document filter = new Document("$match", new Document("operationType", "insert"));
 
-		changeStreamSupport.changeStream().withOptions(options -> {
+		changeStreamSupport.changeStream(Document.class).withOptions(options -> {
 			options.filter(filter);
 		}).listen().subscribe();
 
@@ -144,7 +144,7 @@ public void optionsShouldBeCombinedCorrectly() {
 		Document filter = new Document("$match", new Document("operationType", "insert"));
 		Instant resumeTimestamp = Instant.now();
 
-		changeStreamSupport.changeStream().withOptions(options -> {
+		changeStreamSupport.changeStream(Document.class).withOptions(options -> {
 			options.filter(filter);
 		}).resumeAt(resumeTimestamp).listen().subscribe();