diff --git a/pom.xml b/pom.xml index 28dc5c96c1..b0281d7ece 100644 --- a/pom.xml +++ b/pom.xml @@ -5,7 +5,7 @@ org.springframework.data spring-data-mongodb-parent - 2.2.0.BUILD-SNAPSHOT + 2.2.0.DATAMONGO-2113-SNAPSHOT pom Spring Data MongoDB diff --git a/spring-data-mongodb-benchmarks/pom.xml b/spring-data-mongodb-benchmarks/pom.xml index c2ff37b35c..3293be68a0 100644 --- a/spring-data-mongodb-benchmarks/pom.xml +++ b/spring-data-mongodb-benchmarks/pom.xml @@ -7,7 +7,7 @@ org.springframework.data spring-data-mongodb-parent - 2.2.0.BUILD-SNAPSHOT + 2.2.0.DATAMONGO-2113-SNAPSHOT ../pom.xml diff --git a/spring-data-mongodb-cross-store/pom.xml b/spring-data-mongodb-cross-store/pom.xml index fd36f227c0..df5bb50c1f 100644 --- a/spring-data-mongodb-cross-store/pom.xml +++ b/spring-data-mongodb-cross-store/pom.xml @@ -6,7 +6,7 @@ org.springframework.data spring-data-mongodb-parent - 2.2.0.BUILD-SNAPSHOT + 2.2.0.DATAMONGO-2113-SNAPSHOT ../pom.xml @@ -50,7 +50,7 @@ org.springframework.data spring-data-mongodb - 2.2.0.BUILD-SNAPSHOT + 2.2.0.DATAMONGO-2113-SNAPSHOT diff --git a/spring-data-mongodb-distribution/pom.xml b/spring-data-mongodb-distribution/pom.xml index cb441dd8ef..40a0251cf1 100644 --- a/spring-data-mongodb-distribution/pom.xml +++ b/spring-data-mongodb-distribution/pom.xml @@ -13,7 +13,7 @@ org.springframework.data spring-data-mongodb-parent - 2.2.0.BUILD-SNAPSHOT + 2.2.0.DATAMONGO-2113-SNAPSHOT ../pom.xml diff --git a/spring-data-mongodb/pom.xml b/spring-data-mongodb/pom.xml index f3c85a046a..ec9596e561 100644 --- a/spring-data-mongodb/pom.xml +++ b/spring-data-mongodb/pom.xml @@ -11,7 +11,7 @@ org.springframework.data spring-data-mongodb-parent - 2.2.0.BUILD-SNAPSHOT + 2.2.0.DATAMONGO-2113-SNAPSHOT ../pom.xml diff --git a/spring-data-mongodb/src/main/java/org/springframework/data/mongodb/core/ChangeStreamEvent.java b/spring-data-mongodb/src/main/java/org/springframework/data/mongodb/core/ChangeStreamEvent.java index b25b3eb4fb..8f661c5af2 100644 --- a/spring-data-mongodb/src/main/java/org/springframework/data/mongodb/core/ChangeStreamEvent.java +++ b/spring-data-mongodb/src/main/java/org/springframework/data/mongodb/core/ChangeStreamEvent.java @@ -83,7 +83,9 @@ public ChangeStreamDocument getRaw() { */ @Nullable public Instant getTimestamp() { - return raw != null && raw.getClusterTime() != null ? Instant.ofEpochMilli(raw.getClusterTime().getValue()) : null; + + return raw != null && raw.getClusterTime() != null + ? converter.getConversionService().convert(raw.getClusterTime(), Instant.class) : null; } /** diff --git a/spring-data-mongodb/src/main/java/org/springframework/data/mongodb/core/ReactiveMongoTemplate.java b/spring-data-mongodb/src/main/java/org/springframework/data/mongodb/core/ReactiveMongoTemplate.java index 67d41f04d0..81e7d821d2 100644 --- a/spring-data-mongodb/src/main/java/org/springframework/data/mongodb/core/ReactiveMongoTemplate.java +++ b/spring-data-mongodb/src/main/java/org/springframework/data/mongodb/core/ReactiveMongoTemplate.java @@ -1915,7 +1915,7 @@ public Flux> changeStream(@Nullable String database, @N publisher = options.getResumeToken().map(BsonValue::asDocument).map(publisher::resumeAfter).orElse(publisher); publisher = options.getCollation().map(Collation::toMongoCollation).map(publisher::collation).orElse(publisher); - publisher = options.getResumeTimestamp().map(it -> new BsonTimestamp(it.toEpochMilli())) + publisher = options.getResumeTimestamp().map(it -> new BsonTimestamp((int) it.getEpochSecond(), 0)) .map(publisher::startAtOperationTime).orElse(publisher); publisher = publisher.fullDocument(options.getFullDocumentLookup().orElse(fullDocument)); diff --git a/spring-data-mongodb/src/main/java/org/springframework/data/mongodb/core/convert/MongoConverters.java b/spring-data-mongodb/src/main/java/org/springframework/data/mongodb/core/convert/MongoConverters.java index 83d27510a2..018eba49c5 100644 --- a/spring-data-mongodb/src/main/java/org/springframework/data/mongodb/core/convert/MongoConverters.java +++ b/spring-data-mongodb/src/main/java/org/springframework/data/mongodb/core/convert/MongoConverters.java @@ -19,6 +19,7 @@ import java.math.BigInteger; import java.net.MalformedURLException; import java.net.URL; +import java.time.Instant; import java.util.ArrayList; import java.util.Collection; import java.util.Currency; @@ -26,6 +27,7 @@ import java.util.concurrent.atomic.AtomicInteger; import java.util.concurrent.atomic.AtomicLong; +import org.bson.BsonTimestamp; import org.bson.Document; import org.bson.types.Binary; import org.bson.types.Code; @@ -86,6 +88,7 @@ static Collection getConvertersToRegister() { converters.add(LongToAtomicLongConverter.INSTANCE); converters.add(IntegerToAtomicIntegerConverter.INSTANCE); converters.add(BinaryToByteArrayConverter.INSTANCE); + converters.add(BsonTimestampToInstantConverter.INSTANCE); return converters; } @@ -465,4 +468,22 @@ public byte[] convert(Binary source) { return source.getData(); } } + + /** + * {@link Converter} implementation converting {@link BsonTimestamp} into {@link Instant}. + * + * @author Christoph Strobl + * @since 2.1.2 + */ + @ReadingConverter + enum BsonTimestampToInstantConverter implements Converter { + + INSTANCE; + + @Nullable + @Override + public Instant convert(BsonTimestamp source) { + return Instant.ofEpochSecond(source.getTime(), 0); + } + } } diff --git a/spring-data-mongodb/src/main/java/org/springframework/data/mongodb/core/messaging/ChangeStreamTask.java b/spring-data-mongodb/src/main/java/org/springframework/data/mongodb/core/messaging/ChangeStreamTask.java index 6e887548ab..5d0ff694d7 100644 --- a/spring-data-mongodb/src/main/java/org/springframework/data/mongodb/core/messaging/ChangeStreamTask.java +++ b/spring-data-mongodb/src/main/java/org/springframework/data/mongodb/core/messaging/ChangeStreamTask.java @@ -115,7 +115,7 @@ protected MongoCursor> initCursor(MongoTemplate t .orElseGet(() -> ClassUtils.isAssignable(Document.class, targetType) ? FullDocument.DEFAULT : FullDocument.UPDATE_LOOKUP); - startAt = changeStreamOptions.getResumeTimestamp().map(Instant::toEpochMilli).map(BsonTimestamp::new) + startAt = changeStreamOptions.getResumeTimestamp().map(it -> new BsonTimestamp((int) it.getEpochSecond(), 0)) .orElse(null); } diff --git a/spring-data-mongodb/src/test/java/org/springframework/data/mongodb/core/ReactiveMongoTemplateTests.java b/spring-data-mongodb/src/test/java/org/springframework/data/mongodb/core/ReactiveMongoTemplateTests.java index 0324628db8..22fc49317d 100644 --- a/spring-data-mongodb/src/test/java/org/springframework/data/mongodb/core/ReactiveMongoTemplateTests.java +++ b/spring-data-mongodb/src/test/java/org/springframework/data/mongodb/core/ReactiveMongoTemplateTests.java @@ -29,6 +29,7 @@ import reactor.core.publisher.Mono; import reactor.test.StepVerifier; +import java.time.Duration; import java.time.Instant; import java.util.Arrays; import java.util.Collections; @@ -1355,7 +1356,7 @@ public void watchesDatabaseCorrectly() throws InterruptedException { } } - @Test // DATAMONGO-2012 + @Test // DATAMONGO-2012, DATAMONGO-2113 public void resumesAtTimestampCorrectly() throws InterruptedException { Assumptions.assumeThat(ReplicaSet.required().runsAsReplicaSet()).isTrue(); @@ -1372,7 +1373,7 @@ public void resumesAtTimestampCorrectly() throws InterruptedException { Person person2 = new Person("Data", 37); Person person3 = new Person("MongoDB", 39); - StepVerifier.create(template.save(person1)).expectNextCount(1).verifyComplete(); + StepVerifier.create(template.save(person1).delayElement(Duration.ofSeconds(1))).expectNextCount(1).verifyComplete(); StepVerifier.create(template.save(person2)).expectNextCount(1).verifyComplete(); Thread.sleep(500); // just give it some time to link receive all events diff --git a/spring-data-mongodb/src/test/java/org/springframework/data/mongodb/core/convert/MongoConvertersUnitTests.java b/spring-data-mongodb/src/test/java/org/springframework/data/mongodb/core/convert/MongoConvertersUnitTests.java index 753f033caa..78e990ae9a 100644 --- a/spring-data-mongodb/src/test/java/org/springframework/data/mongodb/core/convert/MongoConvertersUnitTests.java +++ b/spring-data-mongodb/src/test/java/org/springframework/data/mongodb/core/convert/MongoConvertersUnitTests.java @@ -15,14 +15,18 @@ */ package org.springframework.data.mongodb.core.convert; -import static org.hamcrest.Matchers.*; -import static org.junit.Assert.*; +import static org.assertj.core.api.Assertions.*; import java.math.BigDecimal; +import java.time.Instant; +import java.time.temporal.ChronoUnit; import java.util.Currency; import java.util.concurrent.atomic.AtomicInteger; import java.util.concurrent.atomic.AtomicLong; +import org.assertj.core.data.TemporalUnitLessThanOffset; +import org.bson.BsonTimestamp; +import org.bson.Document; import org.junit.Test; import org.springframework.data.geo.Box; import org.springframework.data.geo.Circle; @@ -32,6 +36,7 @@ import org.springframework.data.mongodb.core.convert.MongoConverters.AtomicIntegerToIntegerConverter; import org.springframework.data.mongodb.core.convert.MongoConverters.AtomicLongToLongConverter; import org.springframework.data.mongodb.core.convert.MongoConverters.BigDecimalToStringConverter; +import org.springframework.data.mongodb.core.convert.MongoConverters.BsonTimestampToInstantConverter; import org.springframework.data.mongodb.core.convert.MongoConverters.CurrencyToStringConverter; import org.springframework.data.mongodb.core.convert.MongoConverters.IntegerToAtomicIntegerConverter; import org.springframework.data.mongodb.core.convert.MongoConverters.LongToAtomicLongConverter; @@ -39,8 +44,6 @@ import org.springframework.data.mongodb.core.convert.MongoConverters.StringToCurrencyConverter; import org.springframework.data.mongodb.core.geo.Sphere; -import org.bson.Document; - /** * Unit tests for {@link MongoConverters}. * @@ -55,10 +58,10 @@ public void convertsBigDecimalToStringAndBackCorrectly() { BigDecimal bigDecimal = BigDecimal.valueOf(254, 1); String value = BigDecimalToStringConverter.INSTANCE.convert(bigDecimal); - assertThat(value, is("25.4")); + assertThat(value).isEqualTo("25.4"); BigDecimal reference = StringToBigDecimalConverter.INSTANCE.convert(value); - assertThat(reference, is(bigDecimal)); + assertThat(reference).isEqualTo(bigDecimal); } @Test // DATAMONGO-858 @@ -69,7 +72,7 @@ public void convertsBoxToDocumentAndBackCorrectly() { Document document = GeoConverters.BoxToDocumentConverter.INSTANCE.convert(box); Shape shape = GeoConverters.DocumentToBoxConverter.INSTANCE.convert(document); - assertThat(shape, is((org.springframework.data.geo.Shape) box)); + assertThat(shape).isEqualTo(box); } @Test // DATAMONGO-858 @@ -80,7 +83,7 @@ public void convertsCircleToDocumentAndBackCorrectly() { Document document = GeoConverters.CircleToDocumentConverter.INSTANCE.convert(circle); Shape shape = GeoConverters.DocumentToCircleConverter.INSTANCE.convert(document); - assertThat(shape, is((org.springframework.data.geo.Shape) circle)); + assertThat(shape).isEqualTo(circle); } @Test // DATAMONGO-858 @@ -91,7 +94,7 @@ public void convertsPolygonToDocumentAndBackCorrectly() { Document document = GeoConverters.PolygonToDocumentConverter.INSTANCE.convert(polygon); Shape shape = GeoConverters.DocumentToPolygonConverter.INSTANCE.convert(document); - assertThat(shape, is((org.springframework.data.geo.Shape) polygon)); + assertThat(shape).isEqualTo(polygon); } @Test // DATAMONGO-858 @@ -102,7 +105,7 @@ public void convertsSphereToDocumentAndBackCorrectly() { Document document = GeoConverters.SphereToDocumentConverter.INSTANCE.convert(sphere); org.springframework.data.geo.Shape shape = GeoConverters.DocumentToSphereConverter.INSTANCE.convert(document); - assertThat(shape, is((org.springframework.data.geo.Shape) sphere)); + assertThat(shape).isEqualTo(sphere); } @Test // DATAMONGO-858 @@ -113,36 +116,44 @@ public void convertsPointToListAndBackCorrectly() { Document document = GeoConverters.PointToDocumentConverter.INSTANCE.convert(point); org.springframework.data.geo.Point converted = GeoConverters.DocumentToPointConverter.INSTANCE.convert(document); - assertThat(converted, is((org.springframework.data.geo.Point) point)); + assertThat(converted).isEqualTo(point); } @Test // DATAMONGO-1372 public void convertsCurrencyToStringCorrectly() { - assertThat(CurrencyToStringConverter.INSTANCE.convert(Currency.getInstance("USD")), is("USD")); + assertThat(CurrencyToStringConverter.INSTANCE.convert(Currency.getInstance("USD"))).isEqualTo("USD"); } @Test // DATAMONGO-1372 public void convertsStringToCurrencyCorrectly() { - assertThat(StringToCurrencyConverter.INSTANCE.convert("USD"), is(Currency.getInstance("USD"))); + assertThat(StringToCurrencyConverter.INSTANCE.convert("USD")).isEqualTo(Currency.getInstance("USD")); } @Test // DATAMONGO-1416 public void convertsAtomicLongToLongCorrectly() { - assertThat(AtomicLongToLongConverter.INSTANCE.convert(new AtomicLong(100L)), is(100L)); + assertThat(AtomicLongToLongConverter.INSTANCE.convert(new AtomicLong(100L))).isEqualTo(100L); } @Test // DATAMONGO-1416 public void convertsAtomicIntegerToIntegerCorrectly() { - assertThat(AtomicIntegerToIntegerConverter.INSTANCE.convert(new AtomicInteger(100)), is(100)); + assertThat(AtomicIntegerToIntegerConverter.INSTANCE.convert(new AtomicInteger(100))).isEqualTo(100); } @Test // DATAMONGO-1416 public void convertsLongToAtomicLongCorrectly() { - assertThat(LongToAtomicLongConverter.INSTANCE.convert(100L), is(instanceOf(AtomicLong.class))); + assertThat(LongToAtomicLongConverter.INSTANCE.convert(100L)).isInstanceOf(AtomicLong.class); } @Test // DATAMONGO-1416 public void convertsIntegerToAtomicIntegerCorrectly() { - assertThat(IntegerToAtomicIntegerConverter.INSTANCE.convert(100), is(instanceOf(AtomicInteger.class))); + assertThat(IntegerToAtomicIntegerConverter.INSTANCE.convert(100)).isInstanceOf(AtomicInteger.class); } + + @Test // DATAMONGO-2113 + public void convertsBsonTimestampToInstantCorrectly() { + + assertThat(BsonTimestampToInstantConverter.INSTANCE.convert(new BsonTimestamp(6615900307735969796L))) + .isCloseTo(Instant.ofEpochSecond(1540384327), new TemporalUnitLessThanOffset(100, ChronoUnit.MILLIS)); + } + } diff --git a/spring-data-mongodb/src/test/java/org/springframework/data/mongodb/core/messaging/ChangeStreamTests.java b/spring-data-mongodb/src/test/java/org/springframework/data/mongodb/core/messaging/ChangeStreamTests.java index c76db2b593..a1d3361c08 100644 --- a/spring-data-mongodb/src/test/java/org/springframework/data/mongodb/core/messaging/ChangeStreamTests.java +++ b/spring-data-mongodb/src/test/java/org/springframework/data/mongodb/core/messaging/ChangeStreamTests.java @@ -405,7 +405,7 @@ public void readsFullDocumentForUpdateWhenNotMappedToDomainTypeButLookupSpecifie .append("user_name", "jellyBelly").append("age", 8).append("_class", User.class.getName())); } - @Test // DATAMONGO-2012 + @Test // DATAMONGO-2012, DATAMONGO-2113 public void resumeAtTimestampCorrectly() throws InterruptedException { CollectingMessageListener, User> messageListener1 = new CollectingMessageListener<>(); @@ -415,6 +415,9 @@ public void resumeAtTimestampCorrectly() throws InterruptedException { awaitSubscription(subscription1); template.save(jellyBelly); + + Thread.sleep(1000); // cluster timestamp is in seconds, so we need to wait at least one. + template.save(sugarSplashy); awaitMessages(messageListener1, 12);