From 38390d347531c3f48dfa33008ecbe393193b03ee Mon Sep 17 00:00:00 2001 From: Christoph Strobl Date: Fri, 16 Sep 2022 11:16:28 +0200 Subject: [PATCH] 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. --- .../mongodb/core/ReactiveMongoTemplate.java | 6 +++++- .../core/ReactiveMongoTemplateUnitTests.java | 20 +++++++++++++++++++ 2 files changed, 25 insertions(+), 1 deletion(-) diff --git a/spring-data-mongodb/src/main/java/org/springframework/data/mongodb/core/ReactiveMongoTemplate.java b/spring-data-mongodb/src/main/java/org/springframework/data/mongodb/core/ReactiveMongoTemplate.java index e8fad963c..9af3b5518 100644 --- a/spring-data-mongodb/src/main/java/org/springframework/data/mongodb/core/ReactiveMongoTemplate.java +++ b/spring-data-mongodb/src/main/java/org/springframework/data/mongodb/core/ReactiveMongoTemplate.java @@ -1894,7 +1894,11 @@ public class ReactiveMongoTemplate implements ReactiveMongoOperations, Applicati 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); diff --git a/spring-data-mongodb/src/test/java/org/springframework/data/mongodb/core/ReactiveMongoTemplateUnitTests.java b/spring-data-mongodb/src/test/java/org/springframework/data/mongodb/core/ReactiveMongoTemplateUnitTests.java index 0d9bca468..b6fedd86e 100644 --- a/spring-data-mongodb/src/test/java/org/springframework/data/mongodb/core/ReactiveMongoTemplateUnitTests.java +++ b/spring-data-mongodb/src/test/java/org/springframework/data/mongodb/core/ReactiveMongoTemplateUnitTests.java @@ -39,6 +39,8 @@ import java.util.concurrent.TimeUnit; 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.InsertManyResult; 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 @@ public class ReactiveMongoTemplateUnitTests { .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 realPublisher = Flux.just(document);