diff --git a/mongodb/reactive/src/test/java/example/springdata/mongodb/people/ReactivePersonRepositoryIntegrationTest.java b/mongodb/reactive/src/test/java/example/springdata/mongodb/people/ReactivePersonRepositoryIntegrationTest.java
index 2f14eca7..090bbd87 100644
--- a/mongodb/reactive/src/test/java/example/springdata/mongodb/people/ReactivePersonRepositoryIntegrationTest.java
+++ b/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.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;
@@ -50,8 +52,10 @@ public class ReactivePersonRepositoryIntegrationTest {
 
 		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 class ReactivePersonRepositoryIntegrationTest {
 						new Person("Jesse", "Pinkman", 27))) //
 				.then() //
 				.block();
-
 	}
 
 	/**
@@ -109,8 +112,11 @@ public class ReactivePersonRepositoryIntegrationTest {
 	@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 class ReactivePersonRepositoryIntegrationTest {
 
 		repository.save(new Person("Gus", "Fring", 53)).subscribe();
 		Thread.sleep(100);
+
+		assertThat(people).hasSize(6);
 	}
 
 	/**
diff --git a/mongodb/reactive/src/test/java/example/springdata/mongodb/people/RxJava2PersonRepositoryIntegrationTest.java b/mongodb/reactive/src/test/java/example/springdata/mongodb/people/RxJava2PersonRepositoryIntegrationTest.java
index 603957cd..88884ed6 100644
--- a/mongodb/reactive/src/test/java/example/springdata/mongodb/people/RxJava2PersonRepositoryIntegrationTest.java
+++ b/mongodb/reactive/src/test/java/example/springdata/mongodb/people/RxJava2PersonRepositoryIntegrationTest.java
@@ -20,10 +20,11 @@ import static org.assertj.core.api.Assertions.*;
 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 class RxJava2PersonRepositoryIntegrationTest {
 
 		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 class RxJava2PersonRepositoryIntegrationTest {
 	}
 
 	/**
-	 * 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 class RxJava2PersonRepositoryIntegrationTest {
 
 		repository.save(new Person("Gus", "Fring", 53)).subscribe();
 		Thread.sleep(100);
+
+		assertThat(people).hasSize(6);
 	}
 
 	/**