Spring/spring-data-mongodb
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

DATAMONGO-2258 - Add startAfter option to change stream support.

Browse Source 
Original pull request: #739.
This commit is contained in:
Christoph Strobl
2019-04-15 12:49:21 +02:00
committed by Mark Paluch
parent a30ee07b4a
commit 7f4d3f27e6
4 changed files with 234 additions and 2 deletions
Download Patch File Download Diff File Expand all files Collapse all files

67
spring-data-mongodb/src/main/java/org/springframework/data/mongodb/core/ChangeStreamOptions.java
View File

@@ -21,6 +21,7 @@ import java.time.Instant;
import java.util.Arrays;
import java.util.Optional;
import org.bson.BsonDocument;
import org.bson.BsonTimestamp;
import org.bson.BsonValue;
import org.bson.Document;
@@ -51,6 +52,7 @@ public class ChangeStreamOptions {
private @Nullable FullDocument fullDocumentLookup;
private @Nullable Collation collation;
private @Nullable Object resumeTimestamp;
private Resume resume = Resume.RESUME_AFTER;
protected ChangeStreamOptions() {}
@@ -97,6 +99,22 @@ public class ChangeStreamOptions {
return Optional.ofNullable(resumeTimestamp).map(timestamp -> asTimestampOfType(timestamp, BsonTimestamp.class));
}
/**
* @return {@literal true} if the change stream should be started after the {@link #getResumeToken() token}.
* @since 2.2
*/
public boolean isStartAfter() {
return Resume.START_AFTER.equals(resume);
}
/**
* @return {@literal true} if the change stream should be resumed after the {@link #getResumeToken() token}.
* @since 2.2
*/
public boolean isResumeAfter() {
return Resume.RESUME_AFTER.equals(resume);
}
/**
* @return empty {@link ChangeStreamOptions}.
*/
@@ -137,6 +155,23 @@ public class ChangeStreamOptions {
+ ObjectUtils.nullSafeClassName(timestamp));
}
/**
* @author Christoph Strobl
* @since 2.2
*/
enum Resume {
/**
* @see com.mongodb.client.ChangeStreamIterable#startAfter(BsonDocument)
*/
START_AFTER,
/**
* @see com.mongodb.client.ChangeStreamIterable#resumeAfter(BsonDocument)
*/
RESUME_AFTER
}
/**
* Builder for creating {@link ChangeStreamOptions}.
*
@@ -150,6 +185,7 @@ public class ChangeStreamOptions {
private @Nullable FullDocument fullDocumentLookup;
private @Nullable Collation collation;
private @Nullable Object resumeTimestamp;
private Resume resume = Resume.RESUME_AFTER;
private ChangeStreamOptionsBuilder() {}
@@ -273,6 +309,36 @@ public class ChangeStreamOptions {
return this;
}
/**
* Set the resume token after which to continue emitting notifications.
*
* @param resumeToken must not be {@literal null}.
* @return this.
* @since 2.2
*/
public ChangeStreamOptionsBuilder resumeAfter(BsonValue resumeToken) {
resumeToken(resumeToken);
resume = Resume.RESUME_AFTER;
return this;
}
/**
* Set the resume token after which to start emitting notifications.
*
* @param resumeToken must not be {@literal null}.
* @return this.
* @since 2.2
*/
public ChangeStreamOptionsBuilder startAfter(BsonValue resumeToken) {
resumeToken(resumeToken);
resume = Resume.START_AFTER;
return this;
}
/**
* @return the built {@link ChangeStreamOptions}
*/
@@ -285,6 +351,7 @@ public class ChangeStreamOptions {
options.fullDocumentLookup = fullDocumentLookup;
options.collation = collation;
options.resumeTimestamp = resumeTimestamp;
options.resume = resume;
return options;
}

30
spring-data-mongodb/src/main/java/org/springframework/data/mongodb/core/messaging/ChangeStreamRequest.java
View File

@@ -371,6 +371,36 @@ public class ChangeStreamRequest<T>
return this;
}
/**
* Set the resume token after which to continue emitting notifications.
*
* @param resumeToken must not be {@literal null}.
* @return this.
* @since 2.2
*/
public ChangeStreamRequestBuilder<T> resumeAfter(BsonValue resumeToken) {
Assert.notNull(resumeToken, "ResumeToken must not be null!");
this.delegate.resumeAfter(resumeToken);
return this;
}
/**
* Set the resume token after which to start emitting notifications.
*
* @param resumeToken must not be {@literal null}.
* @return this.
* @since 2.2
*/
public ChangeStreamRequestBuilder<T> startAfter(BsonValue resumeToken) {
Assert.notNull(resumeToken, "ResumeToken must not be null!");
this.delegate.startAfter(resumeToken);
return this;
}
/**
* Set the {@link FullDocument} lookup to {@link FullDocument#UPDATE_LOOKUP}.
*

13
spring-data-mongodb/src/main/java/org/springframework/data/mongodb/core/messaging/ChangeStreamTask.java
View File

@@ -92,6 +92,7 @@ class ChangeStreamTask extends CursorReadingTask<ChangeStreamDocument<Document>,
FullDocument fullDocument = ClassUtils.isAssignable(Document.class, targetType) ? FullDocument.DEFAULT
: FullDocument.UPDATE_LOOKUP;
BsonTimestamp startAt = null;
boolean resumeAfter = true;
if (options instanceof ChangeStreamRequest.ChangeStreamRequestOptions) {
@@ -108,7 +109,9 @@ class ChangeStreamTask extends CursorReadingTask<ChangeStreamDocument<Document>,
}
if (changeStreamOptions.getResumeToken().isPresent()) {
resumeToken = changeStreamOptions.getResumeToken().get().asDocument();
resumeAfter = changeStreamOptions.isResumeAfter();
}
fullDocument = changeStreamOptions.getFullDocumentLookup()
@@ -119,7 +122,8 @@ class ChangeStreamTask extends CursorReadingTask<ChangeStreamDocument<Document>,
}
MongoDatabase db = StringUtils.hasText(options.getDatabaseName())
? template.getMongoDbFactory().getDb(options.getDatabaseName()) : template.getDb();
? template.getMongoDbFactory().getDb(options.getDatabaseName())
: template.getDb();
ChangeStreamIterable<Document> iterable;
@@ -132,7 +136,12 @@ class ChangeStreamTask extends CursorReadingTask<ChangeStreamDocument<Document>,
}
if (!resumeToken.isEmpty()) {
iterable = iterable.resumeAfter(resumeToken);
if (resumeAfter) {
iterable = iterable.resumeAfter(resumeToken);
} else {
iterable = iterable.startAfter(resumeToken);
}
}
if (startAt != null) {

126
spring-data-mongodb/src/test/java/org/springframework/data/mongodb/core/messaging/ChangeStreamTaskUnitTests.java Normal file
View File

@@ -0,0 +1,126 @@
/*
* 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.messaging;
import static org.mockito.Mockito.*;
import java.util.UUID;
import org.bson.BsonDocument;
import org.bson.BsonString;
import org.bson.Document;
import org.junit.Before;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.mockito.Mock;
import org.mockito.junit.MockitoJUnitRunner;
import org.springframework.data.mongodb.core.MongoTemplate;
import org.springframework.data.mongodb.core.convert.MappingMongoConverter;
import org.springframework.data.mongodb.core.convert.MongoConverter;
import org.springframework.data.mongodb.core.convert.NoOpDbRefResolver;
import org.springframework.data.mongodb.core.mapping.MongoMappingContext;
import com.mongodb.client.ChangeStreamIterable;
import com.mongodb.client.MongoCollection;
import com.mongodb.client.MongoCursor;
import com.mongodb.client.MongoDatabase;
import com.mongodb.client.model.changestream.ChangeStreamDocument;
/**
* @author Christoph Strobl
*/
@RunWith(MockitoJUnitRunner.class)
public class ChangeStreamTaskUnitTests {
ChangeStreamTask task;
@Mock MongoTemplate template;
@Mock MongoDatabase mongoDatabase;
@Mock MongoCollection<Document> mongoCollection;
@Mock ChangeStreamIterable<Document> changeStreamIterable;
MongoConverter converter;
@Before
public void setUp() {
MongoMappingContext mappingContext = new MongoMappingContext();
converter = new MappingMongoConverter(NoOpDbRefResolver.INSTANCE, mappingContext);
when(template.getConverter()).thenReturn(converter);
when(template.getDb()).thenReturn(mongoDatabase);
when(mongoDatabase.getCollection(any())).thenReturn(mongoCollection);
when(mongoCollection.watch(eq(Document.class))).thenReturn(changeStreamIterable);
when(changeStreamIterable.startAfter(any())).thenReturn(changeStreamIterable);
when(changeStreamIterable.resumeAfter(any())).thenReturn(changeStreamIterable);
when(changeStreamIterable.fullDocument(any())).thenReturn(changeStreamIterable);
}
@Test // DATAMONGO-2258
public void shouldBe2DotOneComplient() {
BsonDocument resumeToken = new BsonDocument("token", new BsonString(UUID.randomUUID().toString()));
ChangeStreamRequest request = ChangeStreamRequest.builder() //
.collection("start-wars") //
.resumeToken(resumeToken) //
.publishTo(message -> {}) //
.build();
initTask(request, Document.class);
verify(changeStreamIterable).resumeAfter(eq(resumeToken));
}
@Test // DATAMONGO-2258
public void shouldApplyResumeAfterToChangeStream() {
BsonDocument resumeToken = new BsonDocument("token", new BsonString(UUID.randomUUID().toString()));
ChangeStreamRequest request = ChangeStreamRequest.builder() //
.collection("start-wars") //
.resumeAfter(resumeToken) //
.publishTo(message -> {}) //
.build();
initTask(request, Document.class);
verify(changeStreamIterable).resumeAfter(eq(resumeToken));
}
@Test // DATAMONGO-2258
public void shouldApplyStartAfterToChangeStream() {
BsonDocument resumeToken = new BsonDocument("token", new BsonString(UUID.randomUUID().toString()));
ChangeStreamRequest request = ChangeStreamRequest.builder() //
.collection("start-wars") //
.startAfter(resumeToken) //
.publishTo(message -> {}) //
.build();
initTask(request, Document.class);
verify(changeStreamIterable).startAfter(eq(resumeToken));
}
private MongoCursor<ChangeStreamDocument<Document>> initTask(ChangeStreamRequest request, Class<?> targetType) {
ChangeStreamTask task = new ChangeStreamTask(template, request, targetType, er -> {});
return task.initCursor(template, request.getRequestOptions(), targetType);
}
}