From a2f963df14b0368bda53ccad6c53795552eef6ae Mon Sep 17 00:00:00 2001 From: Mark Paluch Date: Thu, 30 Nov 2017 17:18:10 -0800 Subject: [PATCH] #328 - Fix reactive MongoDB tailable cursor examples. MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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. --- ...ReactivePersonRepositoryIntegrationTest.java | 16 ++++++++++++---- .../RxJava2PersonRepositoryIntegrationTest.java | 17 ++++++++++++----- 2 files changed, 24 insertions(+), 9 deletions(-) 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 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 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); } /**