DATAMONGO-2258 - Polishing.

Consider undefined state in ChangeStreamOptions.isResumeAfter()/isStartAfter() when resume token is not set.

Original pull request: #739.

Author: mp911de

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

@@ -25,6 +25,7 @@
 import org.bson.BsonTimestamp;
 import org.bson.BsonValue;
 import org.bson.Document;
+
 import org.springframework.data.mongodb.core.aggregation.Aggregation;
 import org.springframework.data.mongodb.core.query.Collation;
 import org.springframework.lang.Nullable;
@@ -52,7 +53,7 @@ public class ChangeStreamOptions {
 	private @Nullable FullDocument fullDocumentLookup;
 	private @Nullable Collation collation;
 	private @Nullable Object resumeTimestamp;
-	private Resume resume = Resume.RESUME_AFTER;
+	private Resume resume = Resume.UNDEFINED;
 
 	protected ChangeStreamOptions() {}
 
@@ -161,6 +162,8 @@ private static <T> Object doGetTimestamp(Object timestamp, Class<T> targetType)
 	 */
 	enum Resume {
 
+		UNDEFINED,
+
 		/**
 		 * @see com.mongodb.client.ChangeStreamIterable#startAfter(BsonDocument)
 		 */
@@ -185,7 +188,7 @@ public static class ChangeStreamOptionsBuilder {
 		private @Nullable FullDocument fullDocumentLookup;
 		private @Nullable Collation collation;
 		private @Nullable Object resumeTimestamp;
-		private Resume resume = Resume.RESUME_AFTER;
+		private Resume resume = Resume.UNDEFINED;
 
 		private ChangeStreamOptionsBuilder() {}
 
@@ -253,6 +256,11 @@ public ChangeStreamOptionsBuilder resumeToken(BsonValue resumeToken) {
 			Assert.notNull(resumeToken, "ResumeToken must not be null!");
 
 			this.resumeToken = resumeToken;
+
+			if (this.resume == Resume.UNDEFINED) {
+				this.resume = Resume.RESUME_AFTER;
+			}
+
 			return this;
 		}
 
@@ -319,7 +327,7 @@ public ChangeStreamOptionsBuilder resumeAt(BsonTimestamp resumeTimestamp) {
 		public ChangeStreamOptionsBuilder resumeAfter(BsonValue resumeToken) {
 
 			resumeToken(resumeToken);
-			resume = Resume.RESUME_AFTER;
+			this.resume = Resume.RESUME_AFTER;
 
 			return this;
 		}
@@ -334,7 +342,7 @@ public ChangeStreamOptionsBuilder resumeAfter(BsonValue resumeToken) {
 		public ChangeStreamOptionsBuilder startAfter(BsonValue resumeToken) {
 
 			resumeToken(resumeToken);
-			resume = Resume.START_AFTER;
+			this.resume = Resume.START_AFTER;
 
 			return this;
 		}
@@ -346,12 +354,12 @@ public ChangeStreamOptions build() {
 
 			ChangeStreamOptions options = new ChangeStreamOptions();
 
-			options.filter = filter;
-			options.resumeToken = resumeToken;
-			options.fullDocumentLookup = fullDocumentLookup;
-			options.collation = collation;
-			options.resumeTimestamp = resumeTimestamp;
-			options.resume = resume;
+			options.filter = this.filter;
+			options.resumeToken = this.resumeToken;
+			options.fullDocumentLookup = this.fullDocumentLookup;
+			options.collation = this.collation;
+			options.resumeTimestamp = this.resumeTimestamp;
+			options.resume = this.resume;
 
 			return options;
 		}

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

@@ -0,0 +1,56 @@
+/*
+ * 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;
+
+import static org.assertj.core.api.Assertions.*;
+
+import org.bson.BsonDocument;
+import org.junit.Test;
+
+/**
+ * Unit tests for {@link ChangeStreamOptions}.
+ *
+ * @author Mark Paluch
+ */
+public class ChangeStreamOptionsUnitTests {
+
+	@Test // DATAMONGO-2258
+	public void shouldReportResumeAfter() {
+
+		ChangeStreamOptions options = ChangeStreamOptions.builder().resumeAfter(new BsonDocument()).build();
+
+		assertThat(options.isResumeAfter()).isTrue();
+		assertThat(options.isStartAfter()).isFalse();
+	}
+
+	@Test // DATAMONGO-2258
+	public void shouldReportStartAfter() {
+
+		ChangeStreamOptions options = ChangeStreamOptions.builder().startAfter(new BsonDocument()).build();
+
+		assertThat(options.isResumeAfter()).isFalse();
+		assertThat(options.isStartAfter()).isTrue();
+	}
+
+	@Test // DATAMONGO-2258
+	public void shouldNotReportResumeStartAfter() {
+
+		ChangeStreamOptions options = ChangeStreamOptions.empty();
+
+		assertThat(options.isResumeAfter()).isFalse();
+		assertThat(options.isStartAfter()).isFalse();
+	}
+}

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

@@ -27,6 +27,7 @@
 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;
@@ -71,7 +72,7 @@ public void setUp() {
 	}
 
 	@Test // DATAMONGO-2258
-	public void shouldBe2DotOneComplient() {
+	public void shouldNotBreakLovelaceBehavior() {
 
 		BsonDocument resumeToken = new BsonDocument("token", new BsonString(UUID.randomUUID().toString()));