DATAMONGO-2089 - Polishing.

Add watchCollection(…) accepting an entity class. Use static import for assertions. Tweak javadoc.

Original pull request: #751.

Author: mp911de

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

@@ -28,7 +28,7 @@
 
 /**
  * {@link ReactiveChangeStreamOperation} allows creation and execution of reactive MongoDB
- * <a href="https://docs.mongodb.com/manual/changeStreams/">Change Stream</a> operations in a fluent API * style. <br />
+ * <a href="https://docs.mongodb.com/manual/changeStreams/">Change Stream</a> operations in a fluent API style. <br />
  * The starting {@literal domainType} is used for mapping a potentially given
  * {@link org.springframework.data.mongodb.core.aggregation.TypedAggregation} used for filtering. By default, the
  * originating {@literal domainType} is also used for mapping back the result from the {@link org.bson.Document}.
@@ -38,15 +38,14 @@
  *
  * <pre>
  *     <code>
- *         changeStream(Human.class)
+ *         changeStream(Jedi.class)
  *             .watchCollection("star-wars")
  *             .filter(where("operationType").is("insert"))
- *             .as(Jedi.class)
  *             .resumeAt(Instant.now())
  *             .listen();
  *     </code>
  * </pre>
- * 
+ *
  * @author Christoph Strobl
  * @since 2.2
  */
@@ -88,10 +87,20 @@ interface ChangeStreamWithCollection<T> {
 		 * Skip this step to watch all collections within the database.
 		 *
 		 * @param collection must not be {@literal null} nor {@literal empty}.
-		 * @return new instance of {@link ChangeStreamWithCollection}.
-		 * @throws IllegalArgumentException if collection is {@literal null}.
+		 * @return new instance of {@link ChangeStreamWithFilterAndProjection}.
+		 * @throws IllegalArgumentException if {@code collection} is {@literal null}.
 		 */
 		ChangeStreamWithFilterAndProjection<T> watchCollection(String collection);
+
+		/**
+		 * Set the the collection to watch. Collection name is derived from the {@link Class entityClass}.<br />
+		 * Skip this step to watch all collections within the database.
+		 *
+		 * @param entityClass must not be {@literal null}.
+		 * @return new instance of {@link ChangeStreamWithFilterAndProjection}.
+		 * @throws IllegalArgumentException if {@code entityClass} is {@literal null}.
+		 */
+		ChangeStreamWithFilterAndProjection<T> watchCollection(Class<?> entityClass);
 	}
 
 	/**

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

@@ -54,14 +54,14 @@ class ReactiveChangeStreamOperationSupport implements ReactiveChangeStreamOperat
 	public <T> ReactiveChangeStream<T> changeStream(Class<T> domainType) {
 
 		Assert.notNull(domainType, "DomainType must not be null!");
-		return new ReactiveChangeStreamSupport(template, domainType, domainType, null, null);
+		return new ReactiveChangeStreamSupport<>(template, domainType, domainType, null, null);
 	}
 
 	static class ReactiveChangeStreamSupport<T>
 			implements ReactiveChangeStream<T>, ChangeStreamWithFilterAndProjection<T> {
 
 		private final ReactiveMongoTemplate template;
-		private final @Nullable Class<?> domainType;
+		private final Class<?> domainType;
 		private final Class<T> returnType;
 		private final @Nullable String collection;
 		private final @Nullable ChangeStreamOptions options;
@@ -84,9 +84,22 @@ private ReactiveChangeStreamSupport(ReactiveMongoTemplate template, Class<?> dom
 		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);
 		}
 
+		/*
+		 * (non-Javadoc)
+		 * @see org.springframework.data.mongodb.core.ReactiveChangeStreamOperation.ChangeStreamWithCollection#watchCollection(java.lang.Class)
+		 */
+		@Override
+		public ChangeStreamWithFilterAndProjection<T> watchCollection(Class<?> entityClass) {
+
+			Assert.notNull(entityClass, "Collection type not be null!");
+
+			return watchCollection(template.getCollectionName(entityClass));
+		}
+
 		/*
 		 * (non-Javadoc)
 		 * @see org.springframework.data.mongodb.core.ReactiveChangeStreamOperation.ResumingChangeStream#resumeAt(java.lang.Object)
@@ -112,6 +125,7 @@ public TerminatingChangeStream<T> resumeAt(Object token) {
 		public TerminatingChangeStream<T> resumeAfter(Object token) {
 
 			Assert.isInstanceOf(BsonValue.class, token, "Token must be a BsonValue");
+
 			return withOptions(builder -> builder.resumeAfter((BsonValue) token));
 		}
 
@@ -123,6 +137,7 @@ public TerminatingChangeStream<T> resumeAfter(Object token) {
 		public TerminatingChangeStream<T> startAfter(Object token) {
 
 			Assert.isInstanceOf(BsonValue.class, token, "Token must be a BsonValue");
+
 			return withOptions(builder -> builder.startAfter((BsonValue) token));
 		}
 
@@ -147,6 +162,7 @@ public ReactiveChangeStreamSupport<T> withOptions(Consumer<ChangeStreamOptionsBu
 		public <R> ChangeStreamWithFilterAndProjection<R> as(Class<R> resultType) {
 
 			Assert.notNull(resultType, "ResultType must not be null!");
+
 			return new ReactiveChangeStreamSupport<>(template, domainType, resultType, collection, options);
 		}
 
@@ -167,8 +183,7 @@ public ChangeStreamWithFilterAndProjection<T> filter(Aggregation filter) {
 		public ChangeStreamWithFilterAndProjection<T> filter(CriteriaDefinition by) {
 
 			MatchOperation $match = Aggregation.match(by);
-			Aggregation aggregation = domainType != null && !Document.class.equals(domainType)
-					? Aggregation.newAggregation(domainType, $match)
+			Aggregation aggregation = !Document.class.equals(domainType) ? Aggregation.newAggregation(domainType, $match)
 					: Aggregation.newAggregation($match);
 			return filter(aggregation);
 		}
@@ -208,8 +223,8 @@ private ChangeStreamOptionsBuilder initOptionsBuilder() {
 				options.getResumeTimestamp().ifPresent(builder::resumeAt);
 				options.getResumeBsonTimestamp().ifPresent(builder::resumeAt);
 			}
-			return builder;
 
+			return builder;
 		}
 	}
 }

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

@@ -19,9 +19,8 @@ import kotlinx.coroutines.FlowPreview
 import kotlinx.coroutines.flow.Flow
 import kotlinx.coroutines.reactive.flow.asFlow
 
-
 /**
- * Extension for [RactiveChangeStreamOperation. changeStream] leveraging reified type parameters.
+ * Extension for [RactiveChangeStreamOperation.changeStream] leveraging reified type parameters.
  *
  * @author Christoph Strobl
  * @since 2.2
@@ -30,7 +29,7 @@ inline fun <reified T : Any> ReactiveChangeStreamOperation.changeStream(): React
 		changeStream(T::class.java)
 
 /**
- * Extension for [ReactiveChangeStreamOperation.ChangeStreamWithFilterAndProjection. as] leveraging reified type parameters.
+ * Extension for [ReactiveChangeStreamOperation.ChangeStreamWithFilterAndProjection.as] leveraging reified type parameters.
  *
  * @author Christoph Strobl
  * @since 2.2
@@ -39,7 +38,7 @@ inline fun <reified T : Any> ReactiveChangeStreamOperation.ChangeStreamWithFilte
 		`as`(T::class.java)
 
 /**
- * Coroutines [Flow] variant of [ReactiveChangeStreamOperation.TerminatingChangeStream. listen].
+ * 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.

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

@@ -15,6 +15,7 @@
  */
 package org.springframework.data.mongodb.core;
 
+import static org.assertj.core.api.Assertions.*;
 import static org.springframework.data.mongodb.core.query.Criteria.*;
 
 import lombok.SneakyThrows;
@@ -27,18 +28,20 @@
 import java.util.concurrent.LinkedBlockingQueue;
 import java.util.stream.Collectors;
 
-import org.assertj.core.api.Assertions;
 import org.bson.Document;
 import org.junit.After;
 import org.junit.Before;
 import org.junit.ClassRule;
 import org.junit.Test;
+
 import org.springframework.data.mongodb.test.util.MongoTestUtils;
 import org.springframework.data.mongodb.test.util.ReplicaSet;
 
 import com.mongodb.reactivestreams.client.MongoClient;
 
 /**
+ * Tests for {@link ReactiveChangeStreamOperation}.
+ *
  * @author Christoph Strobl
  * @currentRead Dawn Cook - The Decoy Princess
  */
@@ -91,8 +94,8 @@ public void changeStreamEventsShouldBeEmittedCorrectly() {
 		Thread.sleep(500); // just give it some time to link receive all events
 
 		try {
-			Assertions.assertThat(documents.stream().map(ChangeStreamEvent::getBody).collect(Collectors.toList())).hasSize(3)
-					.allMatch(val -> val instanceof Document);
+			assertThat(documents.stream().map(ChangeStreamEvent::getBody).collect(Collectors.toList())).hasSize(3)
+					.allMatch(Document.class::isInstance);
 		} finally {
 			disposable.dispose();
 		}
@@ -122,7 +125,7 @@ public void changeStreamEventsShouldBeConvertedCorrectly() throws InterruptedExc
 		Thread.sleep(500); // just give it some time to link receive all events
 
 		try {
-			Assertions.assertThat(documents.stream().map(ChangeStreamEvent::getBody).collect(Collectors.toList()))
+			assertThat(documents.stream().map(ChangeStreamEvent::getBody).collect(Collectors.toList()))
 					.containsExactly(person1, person2, person3);
 		} finally {
 			disposable.dispose();
@@ -135,7 +138,7 @@ public void changeStreamEventsShouldBeFilteredCorrectly() throws InterruptedExce
 		BlockingQueue<ChangeStreamEvent<Person>> documents = new LinkedBlockingQueue<>(100);
 
 		Disposable disposable = template.changeStream(Person.class) //
-				.watchCollection("person") //
+				.watchCollection(Person.class) //
 				.filter(where("age").gte(38)) //
 				.listen() //
 				.doOnNext(documents::add).subscribe();
@@ -146,18 +149,17 @@ public void changeStreamEventsShouldBeFilteredCorrectly() throws InterruptedExce
 		Person person2 = new Person("Data", 37);
 		Person person3 = new Person("MongoDB", 39);
 
-		Flux.merge(template.save(person1).delayElement(Duration.ofMillis(2)),
-				template.save(person2).delayElement(Duration.ofMillis(2)),
-				template.save(person3).delayElement(Duration.ofMillis(2))) //
+		Flux.merge(template.save(person1), template.save(person2).delayElement(Duration.ofMillis(50)),
+				template.save(person3).delayElement(Duration.ofMillis(100))) //
 				.as(StepVerifier::create) //
 				.expectNextCount(3) //
 				.verifyComplete();
 
 		Thread.sleep(500); // just give it some time to link receive all events
 
 		try {
-			Assertions.assertThat(documents.stream().map(ChangeStreamEvent::getBody).collect(Collectors.toList()))
-					.containsExactly(person1, person3);
+			assertThat(documents.stream().map(ChangeStreamEvent::getBody).collect(Collectors.toList())).containsOnly(person1,
+					person3);
 		} finally {
 			disposable.dispose();
 		}

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

@@ -16,7 +16,7 @@
 package org.springframework.data.mongodb.core;
 
 import static org.assertj.core.api.Assertions.*;
-import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.*;
 import static org.mockito.Mockito.*;
 import static org.mockito.Mockito.eq;
 import static org.mockito.Mockito.isNull;
@@ -36,11 +36,14 @@
 import org.mockito.ArgumentCaptor;
 import org.mockito.Mock;
 import org.mockito.junit.MockitoJUnitRunner;
+
 import org.springframework.data.mongodb.core.aggregation.Aggregation;
 import org.springframework.data.mongodb.core.aggregation.TypedAggregation;
 import org.springframework.data.mongodb.core.query.Criteria;
 
 /**
+ * Unit tests for {@link ReactiveChangeStreamOperationSupport}.
+ *
  * @author Christoph Strobl
  * @currentRead Dawn Cook - The Decoy Princess
  */

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

@@ -1390,7 +1390,7 @@ public void changeStreamEventsShouldBeEmittedCorrectly() throws InterruptedExcep
 		Thread.sleep(500); // just give it some time to link receive all events
 
 		try {
-			Assertions.assertThat(documents.stream().map(ChangeStreamEvent::getBody).collect(Collectors.toList())).hasSize(3)
+			assertThat(documents.stream().map(ChangeStreamEvent::getBody).collect(Collectors.toList())).hasSize(3)
 					.allMatch(val -> val instanceof Document);
 		} finally {
 			disposable.dispose();
@@ -1422,7 +1422,7 @@ public void changeStreamEventsShouldBeConvertedCorrectly() throws InterruptedExc
 		Thread.sleep(500); // just give it some time to link receive all events
 
 		try {
-			Assertions.assertThat(documents.stream().map(ChangeStreamEvent::getBody).collect(Collectors.toList()))
+			assertThat(documents.stream().map(ChangeStreamEvent::getBody).collect(Collectors.toList()))
 					.containsExactly(person1, person2, person3);
 		} finally {
 			disposable.dispose();
@@ -1455,7 +1455,7 @@ public void changeStreamEventsShouldBeFilteredCorrectly() throws InterruptedExce
 		Thread.sleep(500); // just give it some time to link receive all events
 
 		try {
-			Assertions.assertThat(documents.stream().map(ChangeStreamEvent::getBody).collect(Collectors.toList()))
+			assertThat(documents.stream().map(ChangeStreamEvent::getBody).collect(Collectors.toList()))
 					.containsExactly(person1, person3);
 		} finally {
 			disposable.dispose();
@@ -1499,7 +1499,7 @@ public void mapsReservedWordsCorrectly() throws InterruptedException {
 		Thread.sleep(500); // just give it some time to link receive all events
 
 		try {
-			Assertions.assertThat(documents.stream().map(ChangeStreamEvent::getBody).collect(Collectors.toList()))
+			assertThat(documents.stream().map(ChangeStreamEvent::getBody).collect(Collectors.toList()))
 					.containsExactly(replacement);
 		} finally {
 			disposable.dispose();
@@ -1541,7 +1541,7 @@ public void changeStreamEventsShouldBeResumedCorrectly() throws InterruptedExcep
 		Thread.sleep(500); // just give it some time to link receive all events
 
 		try {
-			Assertions.assertThat(resumeDocuments.stream().map(ChangeStreamEvent::getBody).collect(Collectors.toList()))
+			assertThat(resumeDocuments.stream().map(ChangeStreamEvent::getBody).collect(Collectors.toList()))
 					.containsExactly(person2, person3);
 		} finally {
 			disposable.dispose();
@@ -1563,7 +1563,7 @@ public void removeShouldConsiderLimit() {
 
 		template.remove(query(where("field").is("lannister")).limit(25), Sample.class) //
 				.as(StepVerifier::create) //
-				.assertNext(wr -> Assertions.assertThat(wr.getDeletedCount()).isEqualTo(25L)).verifyComplete();
+				.assertNext(wr -> assertThat(wr.getDeletedCount()).isEqualTo(25L)).verifyComplete();
 	}
 
 	@Test // DATAMONGO-1870
@@ -1577,7 +1577,7 @@ public void removeShouldConsiderSkipAndSort() {
 
 		template.remove(new Query().skip(25).with(Sort.by("field")), Sample.class) //
 				.as(StepVerifier::create) //
-				.assertNext(wr -> Assertions.assertThat(wr.getDeletedCount()).isEqualTo(75L)).verifyComplete();
+				.assertNext(wr -> assertThat(wr.getDeletedCount()).isEqualTo(75L)).verifyComplete();
 
 		template.count(query(where("field").is("lannister")), Sample.class).as(StepVerifier::create).expectNext(25L)
 				.verifyComplete();
@@ -1651,7 +1651,7 @@ public void watchesDatabaseCorrectly() throws InterruptedException {
 		Thread.sleep(500); // just give it some time to link receive all events
 
 		try {
-			Assertions.assertThat(documents.stream().map(ChangeStreamEvent::getBody).collect(Collectors.toList()))
+			assertThat(documents.stream().map(ChangeStreamEvent::getBody).collect(Collectors.toList()))
 					.containsExactly(person1, person2, person3);
 		} finally {
 			disposable.dispose();
@@ -1701,7 +1701,7 @@ public void resumesAtTimestampCorrectly() throws InterruptedException {
 		Thread.sleep(500); // just give it some time to link receive all events
 
 		try {
-			Assertions.assertThat(resumeDocuments.stream().map(ChangeStreamEvent::getBody).collect(Collectors.toList()))
+			assertThat(resumeDocuments.stream().map(ChangeStreamEvent::getBody).collect(Collectors.toList()))
 					.containsExactly(person2, person3);
 		} finally {
 			disposable.dispose();

spring-data-mongodb/src/test/kotlin/org/springframework/data/mongodb/core/ReactiveChangeStreamOperationExtensionsTests.kt

@@ -34,6 +34,7 @@ import reactor.core.publisher.Flux
 class ReactiveChangeStreamOperationExtensionsTests {
 
 	val operation = mockk<ReactiveChangeStreamOperation>(relaxed = true)
+	val changestream = mockk<ReactiveChangeStreamOperation.ReactiveChangeStream<First>>(relaxed = true)
 
 	@Test // DATAMONGO-2089
 	fun `ReactiveChangeStreamOperation#changeStream() with reified type parameter extension should call its Java counterpart`() {
@@ -61,4 +62,6 @@ class ReactiveChangeStreamOperationExtensionsTests {
 			spec.listen()
 		}
 	}
+
+	data class Last(val id: String)
 }

src/main/asciidoc/reference/change-streams.adoc

@@ -52,7 +52,7 @@ Subscribing to Change Streams with the reactive API is a more natural approach t
 [source,java]
 ----
 Flux<ChangeStreamEvent<User>> flux = reactiveTemplate.changeStream(User.class) <1>
-    .watchCollection("persons")
+    .watchCollection("people")
     .filter(where("age").gte(38))                                              <2>
     .listen();                                                                 <3>
 ----
@@ -72,9 +72,8 @@ The following example shows how to set the resume offset using server time:
 ====
 [source,java]
 ----
-Flux<ChangeStreamEvent<Person>> resumed = template.changeStream()
-    .watchCollection("persons")
-    .as(User.class)
+Flux<ChangeStreamEvent<User>> resumed = template.changeStream(User.class)
+    .watchCollection("people")
     .resumeAt(Instant.now().minusSeconds(1)) <1>
     .listen();
 ----