DATAMONGO-2115 - Polishing.

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

Original pull request: #624.

Author: mp911de

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

@@ -86,15 +86,15 @@ public Optional<Collation> getCollation() {
 	 * @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));
 	}
 
 	/**
 	 * @return {@link Optional#empty()} if not set.
 	 * @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 static ChangeStreamOptionsBuilder builder() {
 		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));
 	}
 

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

@@ -24,7 +24,6 @@
 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 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 void resumesAtBsonTimestampCorrectly() throws InterruptedException {
 		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) {