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

DATAMONGO-2115 - Polishing.

Browse Source 
Simplify asTimestampOfType(…) retrieval and move the cast to outer method. Simplify test.

Original pull request: #624.
This commit is contained in:
Mark Paluch
2018-12-04 15:44:35 +01:00
parent 9246d3311d
commit 8fe734deb1
2 changed files with 19 additions and 29 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -86,7 +86,7 @@ public class ChangeStreamOptions {
* @return {@link Optional#empty()} if not set.
*/
public Optional<Instant> getResumeTimestamp() {
return Optional.ofNullable(resumeTimestamp).map(this::asInstant);
return Optional.ofNullable(resumeTimestamp).map(timestamp -> asTimestampOfType(timestamp, Instant.class));
}
/**
@@ -94,7 +94,7 @@ public class ChangeStreamOptions {
* @since 2.2
*/
public Optional<BsonTimestamp> getResumeBsonTimestamp() {
return Optional.ofNullable(resumeTimestamp).map(this::asBsonTimestamp);
return Optional.ofNullable(resumeTimestamp).map(timestamp -> asTimestampOfType(timestamp, BsonTimestamp.class));
}
/**
@@ -114,30 +114,26 @@ public class ChangeStreamOptions {
return new ChangeStreamOptionsBuilder();
}
private Instant asInstant(Object timestamp) {
return asTimestampOfType(timestamp, Instant.class);
private static <T> T asTimestampOfType(Object timestamp, Class<T> targetType) {
return targetType.cast(doGetTimestamp(timestamp, targetType));
}
private BsonTimestamp asBsonTimestamp(Object timestamp) {
return asTimestampOfType(timestamp, BsonTimestamp.class);
}
private <T> T asTimestampOfType(Object timestamp, Class<T> targetType) {
private static <T> Object doGetTimestamp(Object timestamp, Class<T> targetType) {
if (ClassUtils.isAssignableValue(targetType, timestamp)) {
return (T) timestamp;
return timestamp;
}
if (timestamp instanceof Instant) {
return (T) new BsonTimestamp((int) ((Instant) timestamp).getEpochSecond(), 0);
return new BsonTimestamp((int) ((Instant) timestamp).getEpochSecond(), 0);
}
if (timestamp instanceof BsonTimestamp) {
return (T) Instant.ofEpochSecond(((BsonTimestamp) timestamp).getTime());
return Instant.ofEpochSecond(((BsonTimestamp) timestamp).getTime());
}
throw new IllegalArgumentException(
"o_O that should actually not happen. The timestampt should be an Instant or a BsonTimestamp but was "
"o_O that should actually not happen. The timestamp should be an Instant or a BsonTimestamp but was "
+ ObjectUtils.nullSafeClassName(timestamp));
}

26
spring-data-mongodb/src/test/java/org/springframework/data/mongodb/core/ReactiveMongoTemplateTests.java
View File

@@ -24,7 +24,6 @@ import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;
import lombok.experimental.Wither;
import org.bson.BsonTimestamp;
import reactor.core.Disposable;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
@@ -46,6 +45,7 @@ import java.util.stream.IntStream;
import org.assertj.core.api.Assertions;
import org.assertj.core.api.Assumptions;
import org.bson.BsonDocument;
import org.bson.BsonTimestamp;
import org.bson.Document;
import org.bson.types.ObjectId;
import org.junit.After;
@@ -1420,27 +1420,21 @@ public class ReactiveMongoTemplateTests {
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
disposable.dispose();
documents.take(); // skip first
BsonTimestamp resumeAt = documents.take().getBsonTimestamp(); // take 2nd
disposable.dispose();
StepVerifier.create(template.save(person3)).expectNextCount(1).verifyComplete();
BlockingQueue<ChangeStreamEvent<Person>> resumeDocuments = new LinkedBlockingQueue<>(100);
template.changeStream("person", ChangeStreamOptions.builder().resumeAt(resumeAt).build(), Person.class)
.doOnNext(resumeDocuments::add).subscribe();
Thread.sleep(500); // just give it some time to link receive all events
try {
Assertions.assertThat(resumeDocuments.stream().map(ChangeStreamEvent::getBody).collect(Collectors.toList()))
.containsExactly(person2, person3);
} finally {
disposable.dispose();
}
.map(ChangeStreamEvent::getBody) //
.buffer(2) //
.as(StepVerifier::create) //
.consumeNextWith(actual -> {
assertThat(actual).containsExactly(person2, person3);
}).thenCancel() //
.verify();
}
private PersonWithAList createPersonWithAList(String firstname, int age) {