From fcd61878494742b1435bbed2978fcb5924a6a826 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 eed68281f..4333052db 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 @@ -2221,7 +2221,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);