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

pom.xml

@@ -5,7 +5,7 @@
 
 	<groupId>org.springframework.data</groupId>
 	<artifactId>spring-data-mongodb-parent</artifactId>
-	<version>2.2.0.BUILD-SNAPSHOT</version>
+	<version>2.2.0.DATAMONGO-2258-SNAPSHOT</version>
 	<packaging>pom</packaging>
 
 	<name>Spring Data MongoDB</name>

spring-data-mongodb-benchmarks/pom.xml

@@ -7,7 +7,7 @@
 	<parent>
 		<groupId>org.springframework.data</groupId>
 		<artifactId>spring-data-mongodb-parent</artifactId>
-		<version>2.2.0.BUILD-SNAPSHOT</version>
+		<version>2.2.0.DATAMONGO-2258-SNAPSHOT</version>
 		<relativePath>../pom.xml</relativePath>
 	</parent>
 

spring-data-mongodb-distribution/pom.xml

@@ -14,7 +14,7 @@
 	<parent>
 		<groupId>org.springframework.data</groupId>
 		<artifactId>spring-data-mongodb-parent</artifactId>
-		<version>2.2.0.BUILD-SNAPSHOT</version>
+		<version>2.2.0.DATAMONGO-2258-SNAPSHOT</version>
 		<relativePath>../pom.xml</relativePath>
 	</parent>
 

spring-data-mongodb/pom.xml

@@ -11,7 +11,7 @@
 	<parent>
 		<groupId>org.springframework.data</groupId>
 		<artifactId>spring-data-mongodb-parent</artifactId>
-		<version>2.2.0.BUILD-SNAPSHOT</version>
+		<version>2.2.0.DATAMONGO-2258-SNAPSHOT</version>
 		<relativePath>../pom.xml</relativePath>
 	</parent>
 

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

@@ -21,6 +21,7 @@
 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 Optional<BsonTimestamp> getResumeBsonTimestamp() {
 		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 @@ private static <T> Object doGetTimestamp(Object timestamp, Class<T> targetType)
 						+ 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 static class ChangeStreamOptionsBuilder {
 		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 ChangeStreamOptionsBuilder resumeAt(BsonTimestamp resumeTimestamp) {
 			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 ChangeStreamOptions build() {
 			options.fullDocumentLookup = fullDocumentLookup;
 			options.collation = collation;
 			options.resumeTimestamp = resumeTimestamp;
+			options.resume = resume;
 
 			return options;
 		}

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

@@ -371,6 +371,36 @@ public ChangeStreamRequestBuilder<T> resumeAt(Instant clusterTime) {
 			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}.
 		 *

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

@@ -92,6 +92,7 @@ protected MongoCursor<ChangeStreamDocument<Document>> initCursor(MongoTemplate t
 		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 @@ protected MongoCursor<ChangeStreamDocument<Document>> initCursor(MongoTemplate t
 			}
 
 			if (changeStreamOptions.getResumeToken().isPresent()) {
+
 				resumeToken = changeStreamOptions.getResumeToken().get().asDocument();
+				resumeAfter = changeStreamOptions.isResumeAfter();
 			}
 
 			fullDocument = changeStreamOptions.getFullDocumentLookup()
@@ -119,7 +122,8 @@ protected MongoCursor<ChangeStreamDocument<Document>> initCursor(MongoTemplate t
 		}
 
 		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 @@ protected MongoCursor<ChangeStreamDocument<Document>> initCursor(MongoTemplate t
 		}
 
 		if (!resumeToken.isEmpty()) {
-			iterable = iterable.resumeAfter(resumeToken);
+
+			if (resumeAfter) {
+				iterable = iterable.resumeAfter(resumeToken);
+			} else {
+				iterable = iterable.startAfter(resumeToken);
+			}
 		}
 
 		if (startAt != null) {

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

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