mindol1004/spring-data-mongodb
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Fix usage of change stream option startAfter.

Browse Source 
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.
This commit is contained in:
Christoph Strobl
2022-09-16 11:16:28 +02:00
committed by Mark Paluch
parent 7c7b05f10d
commit da5f24981c
2 changed files with 25 additions and 1 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -2141,7 +2141,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);

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

@@ -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;
@@ -110,6 +112,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;
@@ -145,6 +148,7 @@ public class ReactiveMongoTemplateUnitTests {
@Mock DistinctPublisher distinctPublisher;
@Mock Publisher deletePublisher;
@Mock MapReducePublisher mapReducePublisher;
@Mock ChangeStreamPublisher changeStreamPublisher;
private MongoExceptionTranslator exceptionTranslator = new MongoExceptionTranslator();
private MappingMongoConverter converter;
@@ -1455,6 +1459,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<Document> realPublisher = Flux.just(document);