Fix usage of change stream option startAfter.

We now make sure to apply the token to startAfter method of the driver. Before this change it had been incorrectly applied to resumeAfter.

Closes #4167.
Original pull request: #4168.

Author: christophstrobl

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

@@ -2221,7 +2221,11 @@ public <T> Flux<ChangeStreamEvent<T>> changeStream(@Nullable String database, @N
 						publisher = filter.isEmpty() ? db.watch(Document.class) : db.watch(filter, Document.class);
 					}
 
-					publisher = options.getResumeToken().map(BsonValue::asDocument).map(publisher::resumeAfter).orElse(publisher);
+					if(options.isResumeAfter()) {
+						publisher = options.getResumeToken().map(BsonValue::asDocument).map(publisher::resumeAfter).orElse(publisher);
+					} else if (options.isStartAfter()) {
+						publisher = options.getResumeToken().map(BsonValue::asDocument).map(publisher::startAfter).orElse(publisher);
+					}
 					publisher = options.getCollation().map(Collation::toMongoCollation).map(publisher::collation)
 							.orElse(publisher);
 					publisher = options.getResumeBsonTimestamp().map(publisher::startAtOperationTime).orElse(publisher);

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

@@ -39,6 +39,8 @@
 import java.util.stream.Collectors;
 
 import org.assertj.core.api.Assertions;
+import org.bson.BsonDocument;
+import org.bson.BsonString;
 import org.bson.Document;
 import org.bson.conversions.Bson;
 import org.bson.types.ObjectId;
@@ -111,6 +113,7 @@
 import com.mongodb.client.result.InsertOneResult;
 import com.mongodb.client.result.UpdateResult;
 import com.mongodb.reactivestreams.client.AggregatePublisher;
+import com.mongodb.reactivestreams.client.ChangeStreamPublisher;
 import com.mongodb.reactivestreams.client.DistinctPublisher;
 import com.mongodb.reactivestreams.client.FindPublisher;
 import com.mongodb.reactivestreams.client.MapReducePublisher;
@@ -146,6 +149,7 @@ public class ReactiveMongoTemplateUnitTests {
 	@Mock DistinctPublisher distinctPublisher;
 	@Mock Publisher deletePublisher;
 	@Mock MapReducePublisher mapReducePublisher;
+	@Mock ChangeStreamPublisher changeStreamPublisher;
 
 	private MongoExceptionTranslator exceptionTranslator = new MongoExceptionTranslator();
 	private MappingMongoConverter converter;
@@ -1485,6 +1489,22 @@ void createCollectionShouldSetUpTimeSeries() {
 						.granularity(TimeSeriesGranularity.HOURS).toString());
 	}
 
+	@Test // GH-4167
+	void changeStreamOptionStartAftershouldApplied() {
+
+		when(factory.getMongoDatabase(anyString())).thenReturn(Mono.just(db));
+
+		when(collection.watch(any(Class.class))).thenReturn(changeStreamPublisher);
+		when(changeStreamPublisher.batchSize(anyInt())).thenReturn(changeStreamPublisher);
+		when(changeStreamPublisher.startAfter(any())).thenReturn(changeStreamPublisher);
+		when(changeStreamPublisher.fullDocument(any())).thenReturn(changeStreamPublisher);
+
+		BsonDocument token = new BsonDocument("token", new BsonString("id"));
+		template.changeStream("database", "collection", ChangeStreamOptions.builder().startAfter(token).build(), Object.class).subscribe();
+
+		verify(changeStreamPublisher).startAfter(eq(token));
+	}
+
 	private void stubFindSubscribe(Document document) {
 
 		Publisher<Document> realPublisher = Flux.just(document);