spring-projects#328 - Fix reactive MongoDB tailable cursor examples.

Replace flatMap(…) operator after collection drop with then(…) to create the collection regardless of whether the previous operation emitted an element or not.

Collect emitted elements to add an assertion and fail if the tailing does not work.

Author: mp911de

mongodb/reactive/src/test/java/example/springdata/mongodb/people/ReactivePersonRepositoryIntegrationTest.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2015-2016 the original author or authors.
+ * Copyright 2015-2017 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.
@@ -22,6 +22,8 @@
 import reactor.core.publisher.Mono;
 
 import java.util.List;
+import java.util.Queue;
+import java.util.concurrent.ConcurrentLinkedQueue;
 import java.util.concurrent.CountDownLatch;
 
 import org.junit.Before;
@@ -50,8 +52,10 @@ public void setUp() {
 
 		operations.collectionExists(Person.class) //
 				.flatMap(exists -> exists ? operations.dropCollection(Person.class) : Mono.just(exists)) //
-				.flatMap(o -> operations.createCollection(Person.class, CollectionOptions.empty().size(1024 * 1024).maxDocuments( 100).capped())) //
-				.then() //
+				.then(operations.createCollection(Person.class, CollectionOptions.empty() //
+						.size(1024 * 1024) //
+						.maxDocuments(100) //
+						.capped())) //
 				.block();
 
 		repository
@@ -61,7 +65,6 @@ public void setUp() {
 						new Person("Jesse", "Pinkman", 27))) //
 				.then() //
 				.block();
-
 	}
 
 	/**
@@ -109,8 +112,11 @@ public void shouldPerformConversionBeforeResultProcessing() throws Exception {
 	@Test
 	public void shouldStreamDataWithTailableCursor() throws Exception {
 
+		Queue<Person> people = new ConcurrentLinkedQueue<>();
+
 		Disposable disposable = repository.findWithTailableCursorBy() //
 				.doOnNext(System.out::println) //
+				.doOnNext(people::add) //
 				.doOnComplete(() -> System.out.println("Complete")) //
 				.doOnTerminate(() -> System.out.println("Terminated")) //
 				.subscribe();
@@ -127,6 +133,8 @@ public void shouldStreamDataWithTailableCursor() throws Exception {
 
 		repository.save(new Person("Gus", "Fring", 53)).subscribe();
 		Thread.sleep(100);
+
+		assertThat(people).hasSize(6);
 	}
 
 	/**

mongodb/reactive/src/test/java/example/springdata/mongodb/people/RxJava2PersonRepositoryIntegrationTest.java

@@ -20,10 +20,11 @@
 import io.reactivex.Flowable;
 import io.reactivex.Single;
 import io.reactivex.disposables.Disposable;
-import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
 
 import java.util.List;
+import java.util.Queue;
+import java.util.concurrent.ConcurrentLinkedQueue;
 import java.util.concurrent.CountDownLatch;
 
 import org.junit.Before;
@@ -56,9 +57,10 @@ public void setUp() {
 
 		operations.collectionExists(Person.class) //
 				.flatMap(exists -> exists ? operations.dropCollection(Person.class) : Mono.just(exists)) //
-				.flatMap(o -> operations.createCollection(Person.class,
-						CollectionOptions.empty().size(1024 * 1024).maxDocuments(100).capped())) //
-				.then() //
+				.then(operations.createCollection(Person.class, CollectionOptions.empty() //
+						.size(1024 * 1024) //
+						.maxDocuments(100) //
+						.capped())) //
 				.block();
 
 		repository.saveAll(Flowable.just(new Person("Walter", "White", 50), //
@@ -112,13 +114,16 @@ public void shouldPerformConversionBeforeResultProcessing() throws Exception {
 	}
 
 	/**
-	 * A tailable cursor streams data using {@link Flux} as it arrives inside the capped collection.
+	 * A tailable cursor streams data using {@link Flowable} as it arrives inside the capped collection.
 	 */
 	@Test
 	public void shouldStreamDataWithTailableCursor() throws Exception {
 
+		Queue<Person> people = new ConcurrentLinkedQueue<>();
+
 		Disposable subscription = repository.findWithTailableCursorBy() //
 				.doOnNext(System.out::println) //
+				.doOnNext(people::add) //
 				.doOnComplete(() -> System.out.println("Complete")) //
 				.doOnTerminate(() -> System.out.println("Terminated")) //
 				.subscribe();
@@ -135,6 +140,8 @@ public void shouldStreamDataWithTailableCursor() throws Exception {
 
 		repository.save(new Person("Gus", "Fring", 53)).subscribe();
 		Thread.sleep(100);
+
+		assertThat(people).hasSize(6);
 	}
 
 	/**