diff --git a/spring-data-mongodb/src/main/java/org/springframework/data/mongodb/core/ChangeStreamOptions.java b/spring-data-mongodb/src/main/java/org/springframework/data/mongodb/core/ChangeStreamOptions.java index bd729dab3..fc6f69a12 100644 --- a/spring-data-mongodb/src/main/java/org/springframework/data/mongodb/core/ChangeStreamOptions.java +++ b/spring-data-mongodb/src/main/java/org/springframework/data/mongodb/core/ChangeStreamOptions.java @@ -25,6 +25,7 @@ import org.bson.BsonDocument; 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 @@ public class ChangeStreamOptions { */ enum Resume { + UNDEFINED, + /** * @see com.mongodb.client.ChangeStreamIterable#startAfter(BsonDocument) */ @@ -185,7 +188,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; private ChangeStreamOptionsBuilder() {} @@ -253,6 +256,11 @@ public class ChangeStreamOptions { 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 class ChangeStreamOptions { public ChangeStreamOptionsBuilder resumeAfter(BsonValue resumeToken) { resumeToken(resumeToken); - resume = Resume.RESUME_AFTER; + this.resume = Resume.RESUME_AFTER; return this; } @@ -334,7 +342,7 @@ public class ChangeStreamOptions { public ChangeStreamOptionsBuilder startAfter(BsonValue resumeToken) { resumeToken(resumeToken); - resume = Resume.START_AFTER; + this.resume = Resume.START_AFTER; return this; } @@ -346,12 +354,12 @@ public class ChangeStreamOptions { 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; } diff --git a/spring-data-mongodb/src/test/java/org/springframework/data/mongodb/core/ChangeStreamOptionsUnitTests.java b/spring-data-mongodb/src/test/java/org/springframework/data/mongodb/core/ChangeStreamOptionsUnitTests.java new file mode 100644 index 000000000..b0a6ba250 --- /dev/null +++ b/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(); + } +} diff --git a/spring-data-mongodb/src/test/java/org/springframework/data/mongodb/core/messaging/ChangeStreamTaskUnitTests.java b/spring-data-mongodb/src/test/java/org/springframework/data/mongodb/core/messaging/ChangeStreamTaskUnitTests.java index 8e42731dd..71ee90638 100644 --- a/spring-data-mongodb/src/test/java/org/springframework/data/mongodb/core/messaging/ChangeStreamTaskUnitTests.java +++ b/spring-data-mongodb/src/test/java/org/springframework/data/mongodb/core/messaging/ChangeStreamTaskUnitTests.java @@ -27,6 +27,7 @@ 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; @@ -71,7 +72,7 @@ public class ChangeStreamTaskUnitTests { } @Test // DATAMONGO-2258 - public void shouldBe2DotOneComplient() { + public void shouldNotBreakLovelaceBehavior() { BsonDocument resumeToken = new BsonDocument("token", new BsonString(UUID.randomUUID().toString()));