DATAMONGO-2089 - Add fluent change stream API to ReactiveMongoTemplate.

We now offer a fluent API for more intuitive change stream interaction.

Flux<ChangeStreamEvent<User>> flux = reactiveTemplate.changeStream(User.class)
    .watchCollection("people")
    .filter(where("age").gte(38))
    .listen();

Original pull request: #751.

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

@@ -0,0 +1,191 @@
+/*
+ * 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 reactor.core.publisher.Flux;
+
+import java.time.Instant;
+import java.util.function.Consumer;
+
+import org.bson.BsonTimestamp;
+import org.bson.BsonValue;
+import org.springframework.data.mongodb.core.ChangeStreamOptions.ChangeStreamOptionsBuilder;
+import org.springframework.data.mongodb.core.aggregation.Aggregation;
+import org.springframework.data.mongodb.core.query.CriteriaDefinition;
+
+/**
+ * {@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 />
+ * 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}.
+ * However, it is possible to define an different {@literal returnType} via {@code as}.<br />
+ * The collection to operate on is optional in which case call collection with the actual database are watched, use
+ * {@literal watchCollection} to define a fixed collection.
+ *
+ * <pre>
+ *     <code>
+ *         changeStream(Human.class)
+ *             .watchCollection("star-wars")
+ *             .filter(where("operationType").is("insert"))
+ *             .as(Jedi.class)
+ *             .resumeAt(Instant.now())
+ *             .listen();
+ *     </code>
+ * </pre>
+ * 
+ * @author Christoph Strobl
+ * @since 2.2
+ */
+public interface ReactiveChangeStreamOperation {
+
+	/**
+	 * 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 ChangeStreamWithFilterAndProjection#filter(CriteriaDefinition) filter}.
+	 *
+	 * @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}.
+	 */
+	<T> ReactiveChangeStream<T> changeStream(Class<T> domainType);
+
+	/**
+	 * Compose change stream execution by calling one of the terminating methods.
+	 */
+	interface TerminatingChangeStream<T> {
+
+		/**
+		 * Start listening to changes. The stream will not be completed unless the {@link org.reactivestreams.Subscription}
+		 * is {@link org.reactivestreams.Subscription#cancel() canceled}.
+		 * <p />
+		 * However, the stream may become dead, or invalid, if all watched collections, databases are dropped.
+		 */
+		Flux<ChangeStreamEvent<T>> listen();
+	}
+
+	/**
+	 * Collection override (optional).
+	 */
+	interface ChangeStreamWithCollection<T> {
+
+		/**
+		 * Explicitly set the name of the collection to watch.<br />
+		 * 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}.
+		 */
+		ChangeStreamWithFilterAndProjection<T> watchCollection(String collection);
+	}
+
+	/**
+	 * Provide a filter for limiting results (optional).
+	 */
+	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 ChangeStreamWithFilterAndProjection}.
+		 * @throws IllegalArgumentException if the given {@link Aggregation} is {@literal null}.
+		 */
+		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 ChangeStreamWithFilterAndProjection}.
+		 * @throws IllegalArgumentException if the given {@link CriteriaDefinition} is {@literal null}.
+		 */
+		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);
+	}
+
+	/**
+	 * Resume a change stream. (optional).
+	 */
+	interface ResumingChangeStream<T> extends TerminatingChangeStream<T> {
+
+		/**
+		 * Resume the change stream at a given point.
+		 *
+		 * @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 token);
+
+		/**
+		 * Resume the change stream after a given point.
+		 *
+		 * @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 token);
+
+		/**
+		 * Start the change stream after a given point.
+		 *
+		 * @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 token);
+	}
+
+	/**
+	 * Provide some options.
+	 */
+	interface ChangeStreamWithOptions<T> {
+
+		/**
+		 * Provide some options via the callback by modifying the given {@link ChangeStreamOptionsBuilder}. Previously
+		 * defined options like a {@link ResumingChangeStream#resumeAfter(Object) resumeToken} are carried over to the
+		 * builder and can be overwritten via eg. {@link ChangeStreamOptionsBuilder#resumeToken(BsonValue)}.
+		 *
+		 * @param optionsConsumer never {@literal null}.
+		 * @return new instance of {@link ReactiveChangeStream}.
+		 */
+		ReactiveChangeStream<T> withOptions(Consumer<ChangeStreamOptionsBuilder> optionsConsumer);
+	}
+
+	/**
+	 * {@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>, ChangeStreamWithFilterAndProjection<T> {}
+}

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

@@ -0,0 +1,215 @@
+/*
+ * 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 reactor.core.publisher.Flux;
+
+import java.time.Instant;
+import java.util.List;
+import java.util.function.Consumer;
+
+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.aggregation.Aggregation;
+import org.springframework.data.mongodb.core.aggregation.MatchOperation;
+import org.springframework.data.mongodb.core.query.CriteriaDefinition;
+import org.springframework.lang.Nullable;
+import org.springframework.util.Assert;
+
+/**
+ * @author Christoph Strobl
+ * @since 2.2
+ */
+class ReactiveChangeStreamOperationSupport implements ReactiveChangeStreamOperation {
+
+	private final ReactiveMongoTemplate template;
+
+	/**
+	 * @param template must not be {@literal null}.
+	 */
+	ReactiveChangeStreamOperationSupport(ReactiveMongoTemplate template) {
+		this.template = template;
+	}
+
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.mongodb.core.ReactiveChangeStreamOperation#changeStream(java.lang.Class)
+	 */
+	@Override
+	public <T> ReactiveChangeStream<T> changeStream(Class<T> domainType) {
+
+		Assert.notNull(domainType, "DomainType must not be 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<T> returnType;
+		private final @Nullable String collection;
+		private final @Nullable ChangeStreamOptions options;
+
+		private ReactiveChangeStreamSupport(ReactiveMongoTemplate template, Class<?> domainType, Class<T> returnType,
+				@Nullable String collection, @Nullable ChangeStreamOptions options) {
+
+			this.template = template;
+			this.domainType = domainType;
+			this.returnType = returnType;
+			this.collection = collection;
+			this.options = options;
+		}
+
+		/*
+		 * (non-Javadoc)
+		 * @see org.springframework.data.mongodb.core.ReactiveChangeStreamOperation.ChangeStreamWithCollection#watchCollection(java.lang.String)
+		 */
+		@Override
+		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.ResumingChangeStream#resumeAt(java.lang.Object)
+		 */
+		@Override
+		public TerminatingChangeStream<T> resumeAt(Object token) {
+
+			return withOptions(builder -> {
+
+				if (token instanceof Instant) {
+					builder.resumeAt((Instant) token);
+				} else if (token instanceof BsonTimestamp) {
+					builder.resumeAt((BsonTimestamp) token);
+				}
+			});
+		}
+
+		/*
+		 * (non-Javadoc)
+		 * @see org.springframework.data.mongodb.core.ReactiveChangeStreamOperation.ResumingChangeStream#resumeAfter(java.lang.Object)
+		 */
+		@Override
+		public TerminatingChangeStream<T> resumeAfter(Object token) {
+
+			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 token) {
+
+			Assert.isInstanceOf(BsonValue.class, token, "Token must be a BsonValue");
+			return withOptions(builder -> builder.startAfter((BsonValue) token));
+		}
+
+		/*
+		 * (non-Javadoc)
+		 * @see org.springframework.data.mongodb.core.ReactiveChangeStreamOperation.ChangeStreamWithOptions#withOptions(java.util.function.Consumer)
+		 */
+		@Override
+		public ReactiveChangeStreamSupport<T> withOptions(Consumer<ChangeStreamOptionsBuilder> optionsConsumer) {
+
+			ChangeStreamOptionsBuilder builder = initOptionsBuilder();
+			optionsConsumer.accept(builder);
+
+			return new ReactiveChangeStreamSupport<>(template, domainType, returnType, collection, builder.build());
+		}
+
+		/*
+		 * (non-Javadoc)
+		 * @see org.springframework.data.mongodb.core.ReactiveChangeStreamOperation.ChangeStreamWithProjection#as(java.lang.Class)
+		 */
+		@Override
+		public <R> ChangeStreamWithFilterAndProjection<R> as(Class<R> resultType) {
+
+			Assert.notNull(resultType, "ResultType must not be null!");
+			return new ReactiveChangeStreamSupport<>(template, domainType, resultType, collection, options);
+		}
+
+		/*
+		 * (non-Javadoc)
+		 * @see org.springframework.data.mongodb.core.ReactiveChangeStreamOperation.ChangeStreamWithFilter#filter(org.springframework.data.mongodb.core.aggregation.Aggregation)
+		 */
+		@Override
+		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 ChangeStreamWithFilterAndProjection<T> filter(CriteriaDefinition by) {
+
+			MatchOperation $match = Aggregation.match(by);
+			Aggregation aggregation = domainType != null && !Document.class.equals(domainType)
+					? Aggregation.newAggregation(domainType, $match)
+					: Aggregation.newAggregation($match);
+			return filter(aggregation);
+		}
+
+		/*
+		 * (non-Javadoc)
+		 * @see org.springframework.data.mongodb.core.ReactiveChangeStreamOperation.TerminatingChangeStream#listen()
+		 */
+		@Override
+		public Flux<ChangeStreamEvent<T>> listen() {
+			return template.changeStream(collection, options != null ? options : ChangeStreamOptions.empty(), returnType);
+		}
+
+		private ChangeStreamOptionsBuilder initOptionsBuilder() {
+
+			ChangeStreamOptionsBuilder builder = ChangeStreamOptions.builder();
+			if (options == null) {
+				return builder;
+			}
+
+			options.getFilter().ifPresent(it -> {
+				if (it instanceof Aggregation) {
+					builder.filter((Aggregation) it);
+				} else {
+					builder.filter(((List<Document>) it).toArray(new Document[0]));
+				}
+			});
+			options.getFullDocumentLookup().ifPresent(builder::fullDocumentLookup);
+			options.getCollation().ifPresent(builder::collation);
+
+			if (options.isResumeAfter()) {
+				options.getResumeToken().ifPresent(builder::resumeAfter);
+				options.getResumeBsonTimestamp().ifPresent(builder::resumeAfter);
+			} else if (options.isStartAfter()) {
+				options.getResumeToken().ifPresent(builder::startAfter);
+			} else {
+				options.getResumeTimestamp().ifPresent(builder::resumeAt);
+				options.getResumeBsonTimestamp().ifPresent(builder::resumeAt);
+			}
+			return builder;
+
+		}
+	}
+}

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

@@ -23,4 +23,4 @@ package org.springframework.data.mongodb.core;
  * @since 2.0
  */
 public interface ReactiveFluentMongoOperations extends ReactiveFindOperation, ReactiveInsertOperation,
-		ReactiveUpdateOperation, ReactiveRemoveOperation, ReactiveAggregationOperation, ReactiveMapReduceOperation {}
+		ReactiveUpdateOperation, ReactiveRemoveOperation, ReactiveAggregationOperation, ReactiveMapReduceOperation, ReactiveChangeStreamOperation {}

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

@@ -2267,6 +2267,15 @@ public class ReactiveMongoTemplate implements ReactiveMongoOperations, Applicati
 		return new ReactiveMapReduceOperationSupport(this).mapReduce(domainType);
 	}
 
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.mongodb.core.ReactiveChangeStreamOperation#changeStream(java.lang.Class)
+	 */
+	@Override
+	public <T> ReactiveChangeStream<T> changeStream(Class<T> domainType) {
+		return new ReactiveChangeStreamOperationSupport(this).changeStream(domainType);
+	}
+
 	/**
 	 * Retrieve and remove all documents matching the given {@code query} by calling {@link #find(Query, Class, String)}
 	 * and {@link #remove(Query, Class, String)}, whereas the {@link Query} for {@link #remove(Query, Class, String)} is

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

@@ -0,0 +1,165 @@
+/*
+ * 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 static org.springframework.data.mongodb.core.query.Criteria.*;
+
+import lombok.SneakyThrows;
+import reactor.core.Disposable;
+import reactor.core.publisher.Flux;
+import reactor.test.StepVerifier;
+
+import java.time.Duration;
+import java.util.concurrent.BlockingQueue;
+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;
+
+/**
+ * @author Christoph Strobl
+ * @currentRead Dawn Cook - The Decoy Princess
+ */
+public class ReactiveChangeStreamOperationSupportTests {
+
+	public static @ClassRule ReplicaSet REPL_SET_REQUIRED = ReplicaSet.required();
+	static final String DATABASE_NAME = "rx-change-stream";
+
+	MongoClient client;
+	ReactiveMongoTemplate template;
+
+	@Before
+	public void setUp() {
+
+		client = MongoTestUtils.reactiveReplSetClient();
+		template = new ReactiveMongoTemplate(client, DATABASE_NAME);
+
+		MongoTestUtils.createOrReplaceCollectionNow(DATABASE_NAME, "person", client);
+	}
+
+	@After
+	public void tearDown() {
+		MongoTestUtils.dropCollectionNow(DATABASE_NAME, "person", client);
+	}
+
+	@SneakyThrows
+	@Test // DATAMONGO-2089
+	public void changeStreamEventsShouldBeEmittedCorrectly() {
+
+		BlockingQueue<ChangeStreamEvent<Document>> documents = new LinkedBlockingQueue<>(100);
+
+		Disposable disposable = template.changeStream(Document.class) //
+				.watchCollection("person") //
+				.listen() //
+				.doOnNext(documents::add).subscribe();
+
+		Thread.sleep(500); // just give it some time to link to the collection.
+
+		Person person1 = new Person("Spring", 38);
+		Person person2 = new Person("Data", 39);
+		Person person3 = new Person("MongoDB", 37);
+
+		Flux.merge(template.insert(person1).delayElement(Duration.ofMillis(2)),
+				template.insert(person2).delayElement(Duration.ofMillis(2)),
+				template.insert(person3).delayElement(Duration.ofMillis(2))) //
+				.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())).hasSize(3)
+					.allMatch(val -> val instanceof Document);
+		} finally {
+			disposable.dispose();
+		}
+	}
+
+	@Test // DATAMONGO-1803
+	public void changeStreamEventsShouldBeConvertedCorrectly() throws InterruptedException {
+
+		BlockingQueue<ChangeStreamEvent<Person>> documents = new LinkedBlockingQueue<>(100);
+
+		Disposable disposable = template.changeStream(Person.class).listen() //
+				.doOnNext(documents::add).subscribe();
+
+		Thread.sleep(500); // just give it some time to link to the collection.
+
+		Person person1 = new Person("Spring", 38);
+		Person person2 = new Person("Data", 39);
+		Person person3 = new Person("MongoDB", 37);
+
+		Flux.merge(template.insert(person1).delayElement(Duration.ofMillis(2)),
+				template.insert(person2).delayElement(Duration.ofMillis(2)),
+				template.insert(person3).delayElement(Duration.ofMillis(2))) //
+				.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, person2, person3);
+		} finally {
+			disposable.dispose();
+		}
+	}
+
+	@Test // DATAMONGO-1803
+	public void changeStreamEventsShouldBeFilteredCorrectly() throws InterruptedException {
+
+		BlockingQueue<ChangeStreamEvent<Person>> documents = new LinkedBlockingQueue<>(100);
+
+		Disposable disposable = template.changeStream(Person.class) //
+				.watchCollection("person") //
+				.filter(where("age").gte(38)) //
+				.listen() //
+				.doOnNext(documents::add).subscribe();
+
+		Thread.sleep(500); // just give it some time to link to the collection.
+
+		Person person1 = new Person("Spring", 38);
+		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))) //
+				.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);
+		} finally {
+			disposable.dispose();
+		}
+	}
+}

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

@@ -0,0 +1,164 @@
+/*
+ * 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 static org.assertj.core.api.Assertions.*;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.*;
+import static org.mockito.Mockito.eq;
+import static org.mockito.Mockito.isNull;
+import static org.springframework.data.mongodb.core.query.Criteria.*;
+
+import reactor.core.publisher.Flux;
+
+import java.time.Instant;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+
+import org.bson.Document;
+import org.junit.Before;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+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;
+
+/**
+ * @author Christoph Strobl
+ * @currentRead Dawn Cook - The Decoy Princess
+ */
+@RunWith(MockitoJUnitRunner.class)
+public class ReactiveChangeStreamOperationSupportUnitTests {
+
+	@Mock ReactiveMongoTemplate template;
+	ReactiveChangeStreamOperationSupport changeStreamSupport;
+
+	@Before
+	public void setUp() {
+		when(template.changeStream(any(), any(), any())).thenReturn(Flux.empty());
+		changeStreamSupport = new ReactiveChangeStreamOperationSupport(template);
+	}
+
+	@Test // DATAMONGO-2089
+	public void listenWithoutDomainTypeUsesDocumentAsDefault() {
+
+		changeStreamSupport.changeStream(Document.class).listen().subscribe();
+
+		verify(template).changeStream(isNull(), eq(ChangeStreamOptions.empty()), eq(Document.class));
+	}
+
+	@Test // DATAMONGO-2089
+	public void listenWithDomainTypeUsesSourceAsTarget() {
+
+		changeStreamSupport.changeStream(Person.class).listen().subscribe();
+
+		verify(template).changeStream(isNull(), eq(ChangeStreamOptions.empty()), eq(Person.class));
+	}
+
+	@Test // DATAMONGO-2089
+	public void collectionNameIsPassedOnCorrectly() {
+
+		changeStreamSupport.changeStream(Person.class).watchCollection("star-wars").listen().subscribe();
+
+		verify(template).changeStream(eq("star-wars"), eq(ChangeStreamOptions.empty()), eq(Person.class));
+	}
+
+	@Test // DATAMONGO-2089
+	public void listenWithDomainTypeCreatesTypedAggregation() {
+
+		Criteria criteria = where("operationType").is("insert");
+		changeStreamSupport.changeStream(Person.class).filter(criteria).listen().subscribe();
+
+		ArgumentCaptor<ChangeStreamOptions> optionsArgumentCaptor = ArgumentCaptor.forClass(ChangeStreamOptions.class);
+		verify(template).changeStream(isNull(), optionsArgumentCaptor.capture(), eq(Person.class));
+
+		assertThat(optionsArgumentCaptor.getValue().getFilter()).hasValueSatisfying(it -> {
+
+			assertThat(it).isInstanceOf(TypedAggregation.class);
+			TypedAggregation<?> aggregation = (TypedAggregation<?>) it;
+
+			assertThat(aggregation.getInputType()).isEqualTo(Person.class);
+			assertThat(extractPipeline(aggregation))
+					.containsExactly(new Document("$match", new Document("operationType", "insert")));
+		});
+	}
+
+	@Test // DATAMONGO-2089
+	public void listenWithoutDomainTypeCreatesUntypedAggregation() {
+
+		Criteria criteria = where("operationType").is("insert");
+		changeStreamSupport.changeStream(Document.class).filter(criteria).listen().subscribe();
+
+		ArgumentCaptor<ChangeStreamOptions> optionsArgumentCaptor = ArgumentCaptor.forClass(ChangeStreamOptions.class);
+		verify(template).changeStream(isNull(), optionsArgumentCaptor.capture(), eq(Document.class));
+
+		assertThat(optionsArgumentCaptor.getValue().getFilter()).hasValueSatisfying(it -> {
+
+			assertThat(it).isInstanceOf(Aggregation.class);
+			assertThat(it).isNotInstanceOf(TypedAggregation.class);
+
+			Aggregation aggregation = (Aggregation) it;
+
+			assertThat(extractPipeline(aggregation))
+					.containsExactly(new Document("$match", new Document("operationType", "insert")));
+		});
+	}
+
+	@Test // DATAMONGO-2089
+	public void optionsShouldBePassedOnCorrectly() {
+
+		Document filter = new Document("$match", new Document("operationType", "insert"));
+
+		changeStreamSupport.changeStream(Document.class).withOptions(options -> {
+			options.filter(filter);
+		}).listen().subscribe();
+
+		ArgumentCaptor<ChangeStreamOptions> optionsArgumentCaptor = ArgumentCaptor.forClass(ChangeStreamOptions.class);
+		verify(template).changeStream(isNull(), optionsArgumentCaptor.capture(), eq(Document.class));
+
+		assertThat(optionsArgumentCaptor.getValue()).satisfies(it -> {
+			assertThat(it.getFilter().get()).isEqualTo(Collections.singletonList(filter));
+		});
+	}
+
+	@Test // DATAMONGO-2089
+	public void optionsShouldBeCombinedCorrectly() {
+
+		Document filter = new Document("$match", new Document("operationType", "insert"));
+		Instant resumeTimestamp = Instant.now();
+
+		changeStreamSupport.changeStream(Document.class).withOptions(options -> {
+			options.filter(filter);
+		}).resumeAt(resumeTimestamp).listen().subscribe();
+
+		ArgumentCaptor<ChangeStreamOptions> optionsArgumentCaptor = ArgumentCaptor.forClass(ChangeStreamOptions.class);
+		verify(template).changeStream(isNull(), optionsArgumentCaptor.capture(), eq(Document.class));
+
+		assertThat(optionsArgumentCaptor.getValue()).satisfies(it -> {
+
+			assertThat(it.getFilter().get()).isEqualTo(Collections.singletonList(filter));
+			assertThat(it.getResumeTimestamp()).contains(resumeTimestamp);
+		});
+	}
+
+	private static List<Document> extractPipeline(Aggregation aggregation) {
+		return aggregation.toDocument("person", Aggregation.DEFAULT_CONTEXT).get("pipeline", ArrayList.class);
+	}
+}

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

@@ -0,0 +1,64 @@
+/*
+ * 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 example.first.First
+import io.mockk.every
+import io.mockk.mockk
+import io.mockk.verify
+import kotlinx.coroutines.FlowPreview
+import kotlinx.coroutines.flow.toList
+import kotlinx.coroutines.runBlocking
+import org.assertj.core.api.Assertions.assertThat
+import org.bson.Document
+import org.junit.Test
+import reactor.core.publisher.Flux
+
+/**
+ * @author Christoph Strobl
+ * @soundtrack Rage Against The Machine - Take the Power Back
+ */
+class ReactiveChangeStreamOperationExtensionsTests {
+
+	val operation = mockk<ReactiveChangeStreamOperation>(relaxed = true)
+
+	@Test // DATAMONGO-2089
+	fun `ReactiveChangeStreamOperation#changeStream() with reified type parameter extension should call its Java counterpart`() {
+
+		operation.changeStream<First>()
+		verify { operation.changeStream(First::class.java) }
+	}
+
+	@Test // DATAMONGO-2089
+	@FlowPreview
+	fun `TerminatingChangeStream#listen() flow extension`() {
+
+		val doc1 = mockk<ChangeStreamEvent<Document>>()
+		val doc2 = mockk<ChangeStreamEvent<Document>>()
+		val doc3 = mockk<ChangeStreamEvent<Document>>()
+
+		val spec = mockk<ReactiveChangeStreamOperation.TerminatingChangeStream<Document>>()
+		every { spec.listen() } returns Flux.just(doc1, doc2, doc3)
+
+		runBlocking {
+			assertThat(spec.flow().toList()).contains(doc1, doc2, doc3)
+		}
+
+		verify {
+			spec.listen()
+		}
+	}
+}

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

@@ -51,14 +51,14 @@ Subscribing to Change Streams with the reactive API is a more natural approach t
 ====
 [source,java]
 ----
-ChangeStreamOptions options = ChangeStreamOptions.builder()
-    .filter(newAggregation(User.class, match(where("age").gte(38)))                              <1>
-    .build();
-
-Flux<ChangeStreamEvent<User>> flux = reactiveTemplate.changeStream("user", options, User.class); <2>
+Flux<ChangeStreamEvent<User>> flux = reactiveTemplate.changeStream(User.class) <1>
+    .watchCollection("persons")
+    .filter(where("age").gte(38))                                              <2>
+    .listen();                                                                 <3>
 ----
-<1> Use an aggregation pipeline to filter events.
-<2> Obtain a `Flux` of change stream events. The `ChangeStreamEvent#getBody()` is converted to the requested domain type. Use `Document` to receive raw results without conversion.
+<1> The event target type the underlying document should be converted to. Leave this out to receive raw results without conversion.
+<2> Use an aggregation pipeline or just a query `Criteria` to filter events.
+<3> Obtain a `Flux` of change stream events. The `ChangeStreamEvent#getBody()` is converted to the requested domain type from (2).
 ====
 
 === Resuming Change Streams
@@ -72,11 +72,11 @@ The following example shows how to set the resume offset using server time:
 ====
 [source,java]
 ----
-ChangeStreamOptions = ChangeStreamOptions.builder()
-    .resumeAt(Instant.now().minusSeconds(1))        <1>
-    .build()
-
-Flux<ChangeStreamEvent<Person>> resumed = template.changeStream("person", options, User.class)
+Flux<ChangeStreamEvent<Person>> resumed = template.changeStream()
+    .watchCollection("persons")
+    .as(User.class)
+    .resumeAt(Instant.now().minusSeconds(1)) <1>
+    .listen();
 ----
 <1> You may obtain the server time of an `ChangeStreamEvent` through the `getTimestamp` method or use the `resumeToken`
 exposed through `getResumeToken`.