From 955597bb547e3d23b7851cca58fcce5e8629dd16 Mon Sep 17 00:00:00 2001 From: Mark Paluch Date: Fri, 24 Mar 2017 17:42:31 +0100 Subject: [PATCH] DATAMONGO-1559 - Migrate reactive tests from TestSubscriber to StepVerifier. --- .../ReactiveMongoTemplateExecuteTests.java | 145 +- .../core/ReactiveMongoTemplateIndexTests.java | 127 +- .../core/ReactiveMongoTemplateTests.java | 590 +++++---- ...onvertingReactiveMongoRepositoryTests.java | 170 ++- .../ReactiveMongoRepositoryTests.java | 165 +-- .../SimpleReactiveMongoRepositoryTests.java | 224 +--- .../java/reactor/test/TestSubscriber.java | 1180 ----------------- 7 files changed, 673 insertions(+), 1928 deletions(-) delete mode 100644 spring-data-mongodb/src/test/java/reactor/test/TestSubscriber.java diff --git a/spring-data-mongodb/src/test/java/org/springframework/data/mongodb/core/ReactiveMongoTemplateExecuteTests.java b/spring-data-mongodb/src/test/java/org/springframework/data/mongodb/core/ReactiveMongoTemplateExecuteTests.java index b2257a8bb..aae70605b 100644 --- a/spring-data-mongodb/src/test/java/org/springframework/data/mongodb/core/ReactiveMongoTemplateExecuteTests.java +++ b/spring-data-mongodb/src/test/java/org/springframework/data/mongodb/core/ReactiveMongoTemplateExecuteTests.java @@ -13,7 +13,6 @@ * See the License for the specific language governing permissions and * limitations under the License. */ - package org.springframework.data.mongodb.core; import static com.sun.prism.impl.Disposer.*; @@ -21,7 +20,8 @@ import static org.hamcrest.Matchers.*; import static org.junit.Assert.*; import static org.junit.Assume.*; -import java.util.List; +import reactor.core.publisher.Flux; +import reactor.test.StepVerifier; import org.bson.Document; import org.junit.After; @@ -39,11 +39,9 @@ import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; import com.mongodb.MongoException; import com.mongodb.ReadPreference; +import com.mongodb.reactivestreams.client.MongoCollection; import com.mongodb.reactivestreams.client.MongoDatabase; -import reactor.core.publisher.Flux; -import reactor.test.TestSubscriber; - /** * Integration test for {@link ReactiveMongoTemplate} execute methods. * @@ -55,11 +53,11 @@ public class ReactiveMongoTemplateExecuteTests { private static final Version THREE = Version.parse("3.0"); + @Rule public ExpectedException thrown = ExpectedException.none(); + @Autowired SimpleReactiveMongoDatabaseFactory factory; @Autowired ReactiveMongoOperations operations; - @Rule public ExpectedException thrown = ExpectedException.none(); - Version mongoVersion; @Before @@ -67,99 +65,104 @@ public class ReactiveMongoTemplateExecuteTests { cleanUp(); if (mongoVersion == null) { - Document result = operations.executeCommand("{ buildInfo: 1 }").block(); - mongoVersion = Version.parse(result.get("version").toString()); + mongoVersion = operations.executeCommand("{ buildInfo: 1 }") // + .map(it -> it.get("version").toString())// + .map(Version::parse) // + .block(); } } @After public void tearDown() { - operations.dropCollection("person").block(); - operations.dropCollection(Person.class).block(); - operations.dropCollection("execute_test").block(); - operations.dropCollection("execute_test1").block(); - operations.dropCollection("execute_test2").block(); - operations.dropCollection("execute_index_test").block(); + Flux cleanup = operations.dropCollection("person") // + .mergeWith(operations.dropCollection(Person.class)) // + .mergeWith(operations.dropCollection("execute_test")) // + .mergeWith(operations.dropCollection("execute_test1")) // + .mergeWith(operations.dropCollection("execute_test2")) // + .mergeWith(operations.dropCollection("execute_index_test")); + + StepVerifier.create(cleanup).verifyComplete(); } @Test // DATAMONGO-1444 - public void executeCommandJsonCommandShouldReturnSingleResponse() throws Exception { + public void executeCommandJsonCommandShouldReturnSingleResponse() { - Document document = operations.executeCommand("{ buildInfo: 1 }").block(); + StepVerifier.create(operations.executeCommand("{ buildInfo: 1 }")).consumeNextWith(actual -> { - assertThat(document, hasKey("version")); + assertThat(actual, hasKey("version")); + }).verifyComplete(); } @Test // DATAMONGO-1444 - public void executeCommandDocumentCommandShouldReturnSingleResponse() throws Exception { + public void executeCommandDocumentCommandShouldReturnSingleResponse() { - Document document = operations.executeCommand(new Document("buildInfo", 1)).block(); + StepVerifier.create(operations.executeCommand(new Document("buildInfo", 1))).consumeNextWith(actual -> { - assertThat(document, hasKey("version")); + assertThat(actual, hasKey("version")); + }).verifyComplete(); } @Test // DATAMONGO-1444 - public void executeCommandJsonCommandShouldReturnMultipleResponses() throws Exception { + public void executeCommandJsonCommandShouldReturnMultipleResponses() { assumeTrue(mongoVersion.isGreaterThan(THREE)); - operations.executeCommand("{ insert: 'execute_test', documents: [{},{},{}]}").block(); + StepVerifier.create(operations.executeCommand("{ insert: 'execute_test', documents: [{},{},{}]}")) + .expectNextCount(1).verifyComplete(); - TestSubscriber subscriber = TestSubscriber.create(); - operations.executeCommand("{ find: 'execute_test'}").subscribe(subscriber); + StepVerifier.create(operations.executeCommand("{ find: 'execute_test'}")) // + .consumeNextWith(actual -> { - subscriber.awaitAndAssertNextValueCount(1); - subscriber.assertValuesWith(document -> { - - assertThat(document.get("ok", Double.class), is(closeTo(1D, 0D))); - assertThat(document, hasKey("cursor")); - }); + assertThat(actual.get("ok", Double.class), is(closeTo(1D, 0D))); + assertThat(actual, hasKey("cursor")); + }) // + .verifyComplete(); } @Test // DATAMONGO-1444 - public void executeCommandJsonCommandShouldTranslateExceptions() throws Exception { + public void executeCommandJsonCommandShouldTranslateExceptions() { - TestSubscriber testSubscriber = TestSubscriber.subscribe(operations.executeCommand("{ unknown: 1 }")); - - testSubscriber.await().assertError(InvalidDataAccessApiUsageException.class); + StepVerifier.create(operations.executeCommand("{ unknown: 1 }")) // + .expectError(InvalidDataAccessApiUsageException.class) // + .verify(); } @Test // DATAMONGO-1444 - public void executeCommandDocumentCommandShouldTranslateExceptions() throws Exception { + public void executeCommandDocumentCommandShouldTranslateExceptions() { - TestSubscriber testSubscriber = TestSubscriber - .subscribe(operations.executeCommand(new Document("unknown", 1))); + StepVerifier.create(operations.executeCommand(new Document("unknown", 1))) // + .expectError(InvalidDataAccessApiUsageException.class) // + .verify(); - testSubscriber.await().assertError(InvalidDataAccessApiUsageException.class); } @Test // DATAMONGO-1444 - public void executeCommandWithReadPreferenceCommandShouldTranslateExceptions() throws Exception { + public void executeCommandWithReadPreferenceCommandShouldTranslateExceptions() { - TestSubscriber testSubscriber = TestSubscriber - .subscribe(operations.executeCommand(new Document("unknown", 1), ReadPreference.nearest())); - - testSubscriber.await().assertError(InvalidDataAccessApiUsageException.class); + StepVerifier.create(operations.executeCommand(new Document("unknown", 1), ReadPreference.nearest())) // + .expectError(InvalidDataAccessApiUsageException.class) // + .verify(); } @Test // DATAMONGO-1444 - public void executeOnDatabaseShouldExecuteCommand() throws Exception { + public void executeOnDatabaseShouldExecuteCommand() { - operations.executeCommand("{ insert: 'execute_test', documents: [{},{},{}]}").block(); - operations.executeCommand("{ insert: 'execute_test1', documents: [{},{},{}]}").block(); - operations.executeCommand("{ insert: 'execute_test2', documents: [{},{},{}]}").block(); + Flux documentFlux = operations.executeCommand("{ insert: 'execute_test', documents: [{},{},{}]}") + .mergeWith(operations.executeCommand("{ insert: 'execute_test1', documents: [{},{},{}]}")) + .mergeWith(operations.executeCommand("{ insert: 'execute_test2', documents: [{},{},{}]}")); + + StepVerifier.create(documentFlux).expectNextCount(3).verifyComplete(); Flux execute = operations.execute(MongoDatabase::listCollections); - List documents = execute.filter(document -> document.getString("name").startsWith("execute_test")) - .collectList().block(); - - assertThat(documents, hasSize(3)); + StepVerifier.create(execute.filter(document -> document.getString("name").startsWith("execute_test"))) // + .expectNextCount(3) // + .verifyComplete(); } @Test // DATAMONGO-1444 - public void executeOnDatabaseShouldDeferExecution() throws Exception { + public void executeOnDatabaseShouldDeferExecution() { operations.execute(db -> { throw new MongoException(50, "hi there"); @@ -169,42 +172,34 @@ public class ReactiveMongoTemplateExecuteTests { } @Test // DATAMONGO-1444 - public void executeOnDatabaseShouldShouldTranslateExceptions() throws Exception { - - TestSubscriber testSubscriber = TestSubscriber.create(); + public void executeOnDatabaseShouldShouldTranslateExceptions() { Flux execute = operations.execute(db -> { throw new MongoException(50, "hi there"); }); - execute.subscribe(testSubscriber); - - testSubscriber.await().assertError(UncategorizedMongoDbException.class); + StepVerifier.create(execute).expectError(UncategorizedMongoDbException.class).verify(); } @Test // DATAMONGO-1444 - public void executeOnCollectionWithTypeShouldReturnFindResults() throws Exception { + public void executeOnCollectionWithTypeShouldReturnFindResults() { - operations.executeCommand("{ insert: 'person', documents: [{},{},{}]}").block(); + StepVerifier.create(operations.executeCommand("{ insert: 'person', documents: [{},{},{}]}")) // + .expectNextCount(1) // + .verifyComplete(); - TestSubscriber testSubscriber = TestSubscriber.create(); - - Flux execute = operations.execute(Person.class, collection -> collection.find()); - execute.subscribe(testSubscriber); - - testSubscriber.awaitAndAssertNextValueCount(3).assertComplete(); + StepVerifier.create(operations.execute(Person.class, MongoCollection::find)).expectNextCount(3).verifyComplete(); } @Test // DATAMONGO-1444 - public void executeOnCollectionWithNameShouldReturnFindResults() throws Exception { + public void executeOnCollectionWithNameShouldReturnFindResults() { - operations.executeCommand("{ insert: 'execute_test', documents: [{},{},{}]}").block(); + StepVerifier.create(operations.executeCommand("{ insert: 'execute_test', documents: [{},{},{}]}")) // + .expectNextCount(1) // + .verifyComplete(); - TestSubscriber testSubscriber = TestSubscriber.create(); - - Flux execute = operations.execute("execute_test", collection -> collection.find()); - execute.subscribe(testSubscriber); - - testSubscriber.awaitAndAssertNextValueCount(3).assertComplete(); + StepVerifier.create(operations.execute("execute_test", MongoCollection::find)) // + .expectNextCount(3) // + .verifyComplete(); } } diff --git a/spring-data-mongodb/src/test/java/org/springframework/data/mongodb/core/ReactiveMongoTemplateIndexTests.java b/spring-data-mongodb/src/test/java/org/springframework/data/mongodb/core/ReactiveMongoTemplateIndexTests.java index 2e86b0969..0aff819a7 100644 --- a/spring-data-mongodb/src/test/java/org/springframework/data/mongodb/core/ReactiveMongoTemplateIndexTests.java +++ b/spring-data-mongodb/src/test/java/org/springframework/data/mongodb/core/ReactiveMongoTemplateIndexTests.java @@ -18,6 +18,10 @@ package org.springframework.data.mongodb.core; import static org.hamcrest.Matchers.*; import static org.junit.Assert.*; +import lombok.Data; +import reactor.core.publisher.Flux; +import reactor.test.StepVerifier; + import java.util.List; import org.bson.Document; @@ -33,18 +37,12 @@ import org.springframework.data.domain.Sort.Direction; import org.springframework.data.mongodb.core.index.Index; import org.springframework.data.mongodb.core.index.IndexField; import org.springframework.data.mongodb.core.index.IndexInfo; -import org.springframework.data.util.Version; import org.springframework.test.context.ContextConfiguration; import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; import com.mongodb.reactivestreams.client.ListIndexesPublisher; import com.mongodb.reactivestreams.client.MongoCollection; -import lombok.Data; -import reactor.core.publisher.Flux; -import reactor.core.publisher.Mono; -import reactor.test.TestSubscriber; - /** * Integration test for {@link MongoTemplate}. * @@ -55,25 +53,19 @@ import reactor.test.TestSubscriber; @ContextConfiguration("classpath:reactive-infrastructure.xml") public class ReactiveMongoTemplateIndexTests { + @Rule public ExpectedException thrown = ExpectedException.none(); + @Autowired SimpleReactiveMongoDatabaseFactory factory; @Autowired ReactiveMongoTemplate template; - @Rule public ExpectedException thrown = ExpectedException.none(); - - Version mongoVersion; - @Before public void setUp() { - cleanDb(); + StepVerifier.create(template.dropCollection(Person.class)).verifyComplete(); } @After public void cleanUp() {} - private void cleanDb() { - template.dropCollection(Person.class).block(); - } - @Test // DATAMONGO-1444 public void testEnsureIndexShouldCreateIndex() { @@ -84,23 +76,28 @@ public class ReactiveMongoTemplateIndexTests { p2.setAge(40); template.insert(p2); - template.indexOps(Person.class).ensureIndex(new Index().on("age", Direction.DESC).unique()).block(); + StepVerifier + .create(template.indexOps(Person.class) // + .ensureIndex(new Index().on("age", Direction.DESC).unique())) // + .expectNextCount(1) // + .verifyComplete(); MongoCollection coll = template.getCollection(template.getCollectionName(Person.class)); - List indexInfo = Flux.from(coll.listIndexes()).collectList().block(); + StepVerifier.create(Flux.from(coll.listIndexes()).collectList()).consumeNextWith(indexInfo -> { - assertThat(indexInfo.size(), is(2)); - Object indexKey = null; - boolean unique = false; - for (Document ix : indexInfo) { + assertThat(indexInfo.size(), is(2)); + Object indexKey = null; + boolean unique = false; + for (Document ix : indexInfo) { - if ("age_-1".equals(ix.get("name"))) { - indexKey = ix.get("key"); - unique = (Boolean) ix.get("unique"); + if ("age_-1".equals(ix.get("name"))) { + indexKey = ix.get("key"); + unique = (Boolean) ix.get("unique"); + } } - } - assertThat(((Document) indexKey), hasEntry("age", -1)); - assertThat(unique, is(true)); + assertThat(((Document) indexKey), hasEntry("age", -1)); + assertThat(unique, is(true)); + }).verifyComplete(); } @Test // DATAMONGO-1444 @@ -108,22 +105,27 @@ public class ReactiveMongoTemplateIndexTests { Person p1 = new Person("Oliver"); p1.setAge(25); - template.insert(p1).block(); + StepVerifier.create(template.insert(p1)).expectNextCount(1).verifyComplete(); - template.indexOps(Person.class).ensureIndex(new Index().on("age", Direction.DESC).unique()).block(); + StepVerifier + .create(template.indexOps(Person.class) // + .ensureIndex(new Index().on("age", Direction.DESC).unique())) // + .expectNextCount(1) // + .verifyComplete(); - List indexInfoList = Flux.from(template.indexOps(Person.class).getIndexInfo()).collectList() - .block(); - assertThat(indexInfoList.size(), is(2)); + StepVerifier.create(template.indexOps(Person.class).getIndexInfo().collectList()).consumeNextWith(indexInfos -> { - IndexInfo ii = indexInfoList.get(1); - assertThat(ii.isUnique(), is(true)); - assertThat(ii.isSparse(), is(false)); + assertThat(indexInfos.size(), is(2)); - List indexFields = ii.getIndexFields(); - IndexField field = indexFields.get(0); + IndexInfo ii = indexInfos.get(1); + assertThat(ii.isUnique(), is(true)); + assertThat(ii.isSparse(), is(false)); - assertThat(field, is(IndexField.create("age", Direction.DESC))); + List indexFields = ii.getIndexFields(); + IndexField field = indexFields.get(0); + + assertThat(field, is(IndexField.create("age", Direction.DESC))); + }).verifyComplete(); } @Test // DATAMONGO-1444 @@ -131,41 +133,46 @@ public class ReactiveMongoTemplateIndexTests { String command = "db." + template.getCollectionName(Person.class) + ".createIndex({'age':-1}, {'unique':true, 'sparse':true}), 1"; - template.indexOps(Person.class).dropAllIndexes().block(); + StepVerifier.create(template.indexOps(Person.class).dropAllIndexes()).verifyComplete(); - TestSubscriber subscriber = TestSubscriber - .subscribe(template.indexOps(Person.class).getIndexInfo()); - subscriber.await().assertComplete().assertNoValues(); + StepVerifier.create(template.indexOps(Person.class).getIndexInfo()).verifyComplete(); - Mono.from(factory.getMongoDatabase().runCommand(new org.bson.Document("eval", command))).block(); + StepVerifier.create(factory.getMongoDatabase().runCommand(new org.bson.Document("eval", command))) // + .expectNextCount(1) // + .verifyComplete(); ListIndexesPublisher listIndexesPublisher = template .getCollection(template.getCollectionName(Person.class)).listIndexes(); - List indexInfo = Flux.from(listIndexesPublisher).collectList().block(); - Document indexKey = null; - boolean unique = false; - for (Document document : indexInfo) { + StepVerifier.create(Flux.from(listIndexesPublisher).collectList()).consumeNextWith(indexInfos -> { - if ("age_-1".equals(document.get("name"))) { - indexKey = (org.bson.Document) document.get("key"); - unique = (Boolean) document.get("unique"); + Document indexKey = null; + boolean unique = false; + + for (Document document : indexInfos) { + + if ("age_-1".equals(document.get("name"))) { + indexKey = (org.bson.Document) document.get("key"); + unique = (Boolean) document.get("unique"); + } } - } - assertThat(indexKey, hasEntry("age", -1D)); - assertThat(unique, is(true)); + assertThat(indexKey, hasEntry("age", -1D)); + assertThat(unique, is(true)); + }).verifyComplete(); - List indexInfos = template.indexOps(Person.class).getIndexInfo().collectList().block(); + StepVerifier.create(Flux.from(template.indexOps(Person.class).getIndexInfo().collectList())) + .consumeNextWith(indexInfos -> { - IndexInfo info = indexInfos.get(1); - assertThat(info.isUnique(), is(true)); - assertThat(info.isSparse(), is(true)); + IndexInfo info = indexInfos.get(1); + assertThat(info.isUnique(), is(true)); + assertThat(info.isSparse(), is(true)); - List indexFields = info.getIndexFields(); - IndexField field = indexFields.get(0); + List indexFields = info.getIndexFields(); + IndexField field = indexFields.get(0); - assertThat(field, is(IndexField.create("age", Direction.DESC))); + assertThat(field, is(IndexField.create("age", Direction.DESC))); + }).verifyComplete(); } @Data diff --git a/spring-data-mongodb/src/test/java/org/springframework/data/mongodb/core/ReactiveMongoTemplateTests.java b/spring-data-mongodb/src/test/java/org/springframework/data/mongodb/core/ReactiveMongoTemplateTests.java index d78194e91..36d690c87 100644 --- a/spring-data-mongodb/src/test/java/org/springframework/data/mongodb/core/ReactiveMongoTemplateTests.java +++ b/spring-data-mongodb/src/test/java/org/springframework/data/mongodb/core/ReactiveMongoTemplateTests.java @@ -13,7 +13,6 @@ * See the License for the specific language governing permissions and * limitations under the License. */ - package org.springframework.data.mongodb.core; import static org.hamcrest.Matchers.*; @@ -21,7 +20,12 @@ import static org.junit.Assert.*; import static org.springframework.data.mongodb.core.query.Criteria.*; import static org.springframework.data.mongodb.core.query.Query.*; -import java.util.ArrayList; +import lombok.Data; +import reactor.core.Cancellation; +import reactor.core.publisher.Flux; +import reactor.core.publisher.Mono; +import reactor.test.StepVerifier; + import java.util.Arrays; import java.util.Collections; import java.util.HashMap; @@ -42,12 +46,10 @@ import org.junit.runner.RunWith; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.dao.DataIntegrityViolationException; import org.springframework.dao.DuplicateKeyException; -import org.springframework.dao.InvalidDataAccessApiUsageException; import org.springframework.dao.OptimisticLockingFailureException; import org.springframework.data.annotation.Id; import org.springframework.data.domain.Sort; import org.springframework.data.domain.Sort.Direction; -import org.springframework.data.domain.Sort.Order; import org.springframework.data.geo.Metrics; import org.springframework.data.mapping.model.MappingException; import org.springframework.data.mongodb.core.MongoTemplateTests.PersonWithConvertedId; @@ -59,18 +61,11 @@ import org.springframework.data.mongodb.core.query.Criteria; import org.springframework.data.mongodb.core.query.NearQuery; import org.springframework.data.mongodb.core.query.Query; import org.springframework.data.mongodb.core.query.Update; -import org.springframework.data.util.Version; import org.springframework.test.context.ContextConfiguration; import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; import com.mongodb.WriteConcern; -import lombok.Data; -import reactor.core.Cancellation; -import reactor.core.publisher.Flux; -import reactor.core.publisher.Mono; -import reactor.test.TestSubscriber; - /** * Integration test for {@link MongoTemplate}. * @@ -81,93 +76,86 @@ import reactor.test.TestSubscriber; @ContextConfiguration("classpath:reactive-infrastructure.xml") public class ReactiveMongoTemplateTests { + @Rule public ExpectedException thrown = ExpectedException.none(); + @Autowired SimpleReactiveMongoDatabaseFactory factory; @Autowired ReactiveMongoTemplate template; - @Rule public ExpectedException thrown = ExpectedException.none(); - - Version mongoVersion; - @Before public void setUp() { - cleanDb(); + + StepVerifier + .create(template.dropCollection("people") // + .mergeWith(template.dropCollection("collection")) // + .mergeWith(template.dropCollection(Person.class)) // + .mergeWith(template.dropCollection(Venue.class)) // + .mergeWith(template.dropCollection(PersonWithAList.class)) // + .mergeWith(template.dropCollection(PersonWithIdPropertyOfTypeObjectId.class)) // + .mergeWith(template.dropCollection(PersonWithVersionPropertyOfTypeInteger.class)) // + .mergeWith(template.dropCollection(Sample.class))) // + .verifyComplete(); } @After public void cleanUp() {} - private void cleanDb() { - template.dropCollection("people") // - .and(template.dropCollection("collection")) // - .and(template.dropCollection(Person.class)) // - .and(template.dropCollection(Venue.class)) // - .and(template.dropCollection(PersonWithAList.class)) // - .and(template.dropCollection(PersonWithIdPropertyOfTypeObjectId.class)) // - .and(template.dropCollection(PersonWithVersionPropertyOfTypeInteger.class)) // - .and(template.dropCollection(Sample.class)).block(); - } - @Test // DATAMONGO-1444 - public void insertSetsId() throws Exception { + public void insertSetsId() { PersonWithAList person = new PersonWithAList(); assert person.getId() == null; - template.insert(person).block(); + StepVerifier.create(template.insert(person)).expectNextCount(1).verifyComplete(); assertThat(person.getId(), is(notNullValue())); } @Test // DATAMONGO-1444 - public void insertAllSetsId() throws Exception { + public void insertAllSetsId() { PersonWithAList person = new PersonWithAList(); - assert person.getId() == null; - template.insertAll(Collections.singletonList(person)).next().block(); + StepVerifier.create(template.insertAll(Collections.singleton(person))).expectNextCount(1).verifyComplete(); assertThat(person.getId(), is(notNullValue())); } @Test // DATAMONGO-1444 - public void insertCollectionSetsId() throws Exception { + public void insertCollectionSetsId() { PersonWithAList person = new PersonWithAList(); - assert person.getId() == null; - template.insert(Collections.singletonList(person), PersonWithAList.class).next().block(); + StepVerifier.create(template.insert(Collections.singleton(person), PersonWithAList.class)).expectNextCount(1) + .verifyComplete(); assertThat(person.getId(), is(notNullValue())); } @Test // DATAMONGO-1444 - public void saveSetsId() throws Exception { + public void saveSetsId() { PersonWithAList person = new PersonWithAList(); assert person.getId() == null; - template.save(person).block(); + StepVerifier.create(template.save(person)).expectNextCount(1).verifyComplete(); assertThat(person.getId(), is(notNullValue())); } @Test // DATAMONGO-1444 - public void insertsSimpleEntityCorrectly() throws Exception { + public void insertsSimpleEntityCorrectly() { Person person = new Person("Mark"); person.setAge(35); - template.insert(person).block(); + StepVerifier.create(template.insert(person)).expectNextCount(1).verifyComplete(); - TestSubscriber testSubscriber = TestSubscriber.create(); - Flux flux = template.find(new Query(where("_id").is(person.getId())), Person.class); - flux.subscribe(testSubscriber); - - testSubscriber.awaitAndAssertNextValueCount(1); - testSubscriber.assertValues(person); + StepVerifier.create(template.find(new Query(where("_id").is(person.getId())), Person.class)) // + .expectNext(person) // + .verifyComplete(); } @Test // DATAMONGO-1444 - public void simpleInsertDoesNotAllowArrays() throws Exception { + public void simpleInsertDoesNotAllowArrays() { thrown.expect(IllegalArgumentException.class); @@ -177,7 +165,7 @@ public class ReactiveMongoTemplateTests { } @Test // DATAMONGO-1444 - public void simpleInsertDoesNotAllowCollections() throws Exception { + public void simpleInsertDoesNotAllowCollections() { thrown.expect(IllegalArgumentException.class); @@ -187,193 +175,183 @@ public class ReactiveMongoTemplateTests { } @Test // DATAMONGO-1444 - public void insertsSimpleEntityWithSuppliedCollectionNameCorrectly() throws Exception { + public void insertsSimpleEntityWithSuppliedCollectionNameCorrectly() { Person person = new Person("Homer"); person.setAge(35); - template.insert(person, "people").block(); + StepVerifier.create(template.insert(person, "people")).expectNextCount(1).verifyComplete(); - TestSubscriber testSubscriber = TestSubscriber.create(); - Flux flux = template.find(new Query(where("_id").is(person.getId())), Person.class, "people"); - flux.subscribe(testSubscriber); - - testSubscriber.awaitAndAssertNextValueCount(1); - testSubscriber.assertValues(person); + StepVerifier.create(template.find(new Query(where("_id").is(person.getId())), Person.class, "people")) // + .expectNext(person) // + .verifyComplete(); } @Test // DATAMONGO-1444 - public void insertBatchCorrectly() throws Exception { + public void insertBatchCorrectly() { - List persons = Arrays.asList(new Person("Dick", 22), new Person("Harry", 23), new Person("Tom", 21)); + List people = Arrays.asList(new Person("Dick", 22), new Person("Harry", 23), new Person("Tom", 21)); - template.insertAll(persons).next().block(); + StepVerifier.create(template.insertAll(people)).expectNextCount(3).verifyComplete(); - TestSubscriber testSubscriber = TestSubscriber.create(); - Flux flux = template.find(new Query().with(new Sort(new Order("firstname"))), Person.class); - flux.subscribe(testSubscriber); - - testSubscriber.awaitAndAssertNextValueCount(3); - testSubscriber.assertValues(persons.toArray(new Person[persons.size()])); + StepVerifier.create(template.find(new Query().with(Sort.by("firstname")), Person.class)) // + .expectNextSequence(people) // + .verifyComplete(); } @Test // DATAMONGO-1444 - public void insertBatchWithSuppliedCollectionNameCorrectly() throws Exception { + public void insertBatchWithSuppliedCollectionNameCorrectly() { - List persons = Arrays.asList(new Person("Dick", 22), new Person("Harry", 23), new Person("Tom", 21)); + List people = Arrays.asList(new Person("Dick", 22), new Person("Harry", 23), new Person("Tom", 21)); - template.insert(persons, "people").then().block(); + StepVerifier.create(template.insert(people, "people")).expectNextCount(3).verifyComplete(); - TestSubscriber testSubscriber = TestSubscriber.create(); - Flux flux = template.find(new Query().with(new Sort(new Order("firstname"))), Person.class, "people"); - flux.subscribe(testSubscriber); - - testSubscriber.awaitAndAssertNextValueCount(3); - testSubscriber.assertValues(persons.toArray(new Person[persons.size()])); + StepVerifier.create(template.find(new Query().with(Sort.by("firstname")), Person.class, "people")) // + .expectNextSequence(people) // + .verifyComplete(); } @Test // DATAMONGO-1444 - public void insertBatchWithSuppliedEntityTypeCorrectly() throws Exception { + public void insertBatchWithSuppliedEntityTypeCorrectly() { - List persons = Arrays.asList(new Person("Dick", 22), new Person("Harry", 23), new Person("Tom", 21)); + List people = Arrays.asList(new Person("Dick", 22), new Person("Harry", 23), new Person("Tom", 21)); - template.insert(persons, Person.class).then().block(); + StepVerifier.create(template.insert(people, Person.class)).expectNextCount(3).verifyComplete(); - TestSubscriber testSubscriber = TestSubscriber.create(); - Flux flux = template.find(new Query().with(new Sort(new Order("firstname"))), Person.class); - flux.subscribe(testSubscriber); - - testSubscriber.awaitAndAssertNextValueCount(3); - testSubscriber.assertValues(persons.toArray(new Person[persons.size()])); + StepVerifier.create(template.find(new Query().with(Sort.by("firstname")), Person.class)) // + .expectNextSequence(people) // + .verifyComplete(); } @Test // DATAMONGO-1444 public void testAddingToList() { - PersonWithAList p = new PersonWithAList(); - p.setFirstName("Sven"); - p.setAge(22); - template.insert(p).block(); + PersonWithAList person = createPersonWithAList("Sven", 22); + StepVerifier.create(template.insert(person)).expectNextCount(1).verifyComplete(); - Query q1 = new Query(where("id").is(p.getId())); - PersonWithAList p2 = template.findOne(q1, PersonWithAList.class).block(); - assertThat(p2, notNullValue()); - assertThat(p2.getWishList().size(), is(0)); + Query query = new Query(where("id").is(person.getId())); - p2.addToWishList("please work!"); + StepVerifier.create(template.findOne(query, PersonWithAList.class)).consumeNextWith(actual -> { - template.save(p2).block(); + assertThat(actual.getWishList().size(), is(0)); + }).verifyComplete(); - PersonWithAList p3 = template.findOne(q1, PersonWithAList.class).block(); - assertThat(p3, notNullValue()); - assertThat(p3.getWishList().size(), is(1)); + person.addToWishList("please work!"); - Friend f = new Friend(); - p.setFirstName("Erik"); - p.setAge(21); + StepVerifier.create(template.save(person)).expectNextCount(1).verifyComplete(); - p3.addFriend(f); - template.save(p3).block(); + StepVerifier.create(template.findOne(query, PersonWithAList.class)).consumeNextWith(actual -> { - PersonWithAList p4 = template.findOne(q1, PersonWithAList.class).block(); - assertThat(p4, notNullValue()); - assertThat(p4.getWishList().size(), is(1)); - assertThat(p4.getFriends().size(), is(1)); + assertThat(actual.getWishList().size(), is(1)); + }).verifyComplete(); + Friend friend = new Friend(); + person.setFirstName("Erik"); + person.setAge(21); + + person.addFriend(friend); + StepVerifier.create(template.save(person)).expectNextCount(1).verifyComplete(); + + StepVerifier.create(template.findOne(query, PersonWithAList.class)).consumeNextWith(actual -> { + + assertThat(actual.getWishList().size(), is(1)); + assertThat(actual.getFriends().size(), is(1)); + }).verifyComplete(); } @Test // DATAMONGO-1444 public void testFindOneWithSort() { - PersonWithAList p = new PersonWithAList(); - p.setFirstName("Sven"); - p.setAge(22); - template.insert(p).block(); - PersonWithAList p2 = new PersonWithAList(); - p2.setFirstName("Erik"); - p2.setAge(21); - template.insert(p2).block(); + PersonWithAList sven = createPersonWithAList("Sven", 22); + PersonWithAList erik = createPersonWithAList("Erik", 21); + PersonWithAList mark = createPersonWithAList("Mark", 40); - PersonWithAList p3 = new PersonWithAList(); - p3.setFirstName("Mark"); - p3.setAge(40); - template.insert(p3).block(); + StepVerifier.create(template.insertAll(Arrays.asList(sven, erik, mark))).expectNextCount(3).verifyComplete(); // test query with a sort - Query q2 = new Query(where("age").gt(10)); - q2.with(new Sort(Direction.DESC, "age")); - PersonWithAList p5 = template.findOne(q2, PersonWithAList.class).block(); - assertThat(p5.getFirstName(), is("Mark")); + Query query = new Query(where("age").gt(10)); + query.with(Sort.by(Direction.DESC, "age")); + + StepVerifier.create(template.findOne(query, PersonWithAList.class)).consumeNextWith(actual -> { + + assertThat(actual.getFirstName(), is("Mark")); + }).verifyComplete(); } @Test // DATAMONGO-1444 - public void bogusUpdateDoesNotTriggerException() throws Exception { + public void bogusUpdateDoesNotTriggerException() { ReactiveMongoTemplate mongoTemplate = new ReactiveMongoTemplate(factory); mongoTemplate.setWriteResultChecking(WriteResultChecking.EXCEPTION); - Person person = new Person("Oliver2"); - person.setAge(25); - mongoTemplate.insert(person).block(); + Person oliver = new Person("Oliver2", 25); + StepVerifier.create(template.insert(oliver)).expectNextCount(1).verifyComplete(); Query q = new Query(where("BOGUS").gt(22)); Update u = new Update().set("firstName", "Sven"); - mongoTemplate.updateFirst(q, u, Person.class).block(); + + StepVerifier.create(mongoTemplate.updateFirst(q, u, Person.class)).expectNextCount(1).verifyComplete(); } @Test // DATAMONGO-1444 - public void updateFirstByEntityTypeShouldUpdateObject() throws Exception { + public void updateFirstByEntityTypeShouldUpdateObject() { Person person = new Person("Oliver2", 25); - template.insert(person) // + StepVerifier.create(template.insert(person) // .then(template.updateFirst(new Query(where("age").is(25)), new Update().set("firstName", "Sven"), Person.class)) // - .flatMap(p -> template.find(new Query(where("age").is(25)), Person.class)) - .subscribeWith(TestSubscriber.create()) // - .await() // - .assertValuesWith(result -> { - assertThat(result.getFirstName(), is(equalTo("Sven"))); - }); + .flatMap(p -> template.find(new Query(where("age").is(25)), Person.class))).consumeNextWith(actual -> { + + assertThat(actual.getFirstName(), is(equalTo("Sven"))); + }).verifyComplete(); } @Test // DATAMONGO-1444 - public void updateFirstByCollectionNameShouldUpdateObjects() throws Exception { + public void updateFirstByCollectionNameShouldUpdateObjects() { Person person = new Person("Oliver2", 25); - template.insert(person, "people") // - .then(template.updateFirst(new Query(where("age").is(25)), new Update().set("firstName", "Sven"), "people")) // - .flatMap(p -> template.find(new Query(where("age").is(25)), Person.class, "people")) - .subscribeWith(TestSubscriber.create()) // - .await() // - .assertValuesWith(result -> { - assertThat(result.getFirstName(), is(equalTo("Sven"))); - }); + StepVerifier + .create(template.insert(person, "people") // + .then(template.updateFirst(new Query(where("age").is(25)), new Update().set("firstName", "Sven"), "people")) // + .flatMap(p -> template.find(new Query(where("age").is(25)), Person.class, "people"))) + .consumeNextWith(actual -> { + + assertThat(actual.getFirstName(), is(equalTo("Sven"))); + }).verifyComplete(); } @Test // DATAMONGO-1444 - public void updateMultiByEntityTypeShouldUpdateObjects() throws Exception { + public void updateMultiByEntityTypeShouldUpdateObjects() { Query query = new Query( new Criteria().orOperator(where("firstName").is("Walter Jr"), Criteria.where("firstName").is("Walter"))); - template.insertAll(Mono.just(Arrays.asList(new Person("Walter", 50), new Person("Skyler", 43), new Person("Walter Jr", 16)))) // - .flatMap(a -> template.updateMulti(query, new Update().set("firstName", "Walt"), Person.class)) // - .flatMap(p -> template.find(new Query(where("firstName").is("Walt")), Person.class)) // - .subscribeWith(TestSubscriber.create()) // - .awaitAndAssertNextValueCount(2); + StepVerifier + .create(template + .insertAll(Mono + .just(Arrays.asList(new Person("Walter", 50), new Person("Skyler", 43), new Person("Walter Jr", 16)))) // + .flatMap(a -> template.updateMulti(query, new Update().set("firstName", "Walt"), Person.class)) // + .thenMany(template.find(new Query(where("firstName").is("Walt")), Person.class))) // + .expectNextCount(2) // + .verifyComplete(); } @Test // DATAMONGO-1444 - public void updateMultiByCollectionNameShouldUpdateObject() throws Exception { + public void updateMultiByCollectionNameShouldUpdateObject() { Query query = new Query( new Criteria().orOperator(where("firstName").is("Walter Jr"), Criteria.where("firstName").is("Walter"))); - template - .insertAll(Mono.just(Arrays.asList(new Person("Walter", 50), new Person("Skyler", 43), new Person("Walter Jr", 16))), "people") // + List people = Arrays.asList(new Person("Walter", 50), // + new Person("Skyler", 43), // + new Person("Walter Jr", 16)); + + Flux personFlux = template.insertAll(Mono.just(people), "people") // .collectList() // .flatMap(a -> template.updateMulti(query, new Update().set("firstName", "Walt"), Person.class, "people")) // - .flatMap(p -> template.find(new Query(where("firstName").is("Walt")), Person.class, "people")) // - .subscribeWith(TestSubscriber.create()) // - .awaitAndAssertNextValueCount(2); + .flatMap(p -> template.find(new Query(where("firstName").is("Walt")), Person.class, "people")); + + StepVerifier.create(personFlux) // + .expectNextCount(2) // + .verifyComplete(); } @Test // DATAMONGO-1444 @@ -385,14 +363,9 @@ public class ReactiveMongoTemplateTests { Person person = new Person(new ObjectId(), "Amol"); person.setAge(28); - template.insert(person).block(); + StepVerifier.create(template.insert(person)).expectNextCount(1).verifyComplete(); - try { - template.insert(person).block(); - fail("Expected DataIntegrityViolationException!"); - } catch (DataIntegrityViolationException e) { - assertThat(e.getMessage(), containsString("E11000 duplicate key error")); - } + StepVerifier.create(template.insert(person)).expectError(DataIntegrityViolationException.class).verify(); } @Test // DATAMONGO-1444 @@ -405,24 +378,19 @@ public class ReactiveMongoTemplateTests { Person person = new Person(id, "Amol"); person.setAge(28); - template.insert(person).block(); - - thrown.expect(DataIntegrityViolationException.class); - thrown.expectMessage("array"); - thrown.expectMessage("age"); - // thrown.expectMessage("failed"); + StepVerifier.create(template.insert(person)).expectNextCount(1).verifyComplete(); Query query = new Query(where("firstName").is("Amol")); Update upd = new Update().push("age", 29); - template.updateFirst(query, upd, Person.class).block(); + + StepVerifier.create(template.updateFirst(query, upd, Person.class)) // + .expectError(DataIntegrityViolationException.class) // + .verify(); } @Test // DATAMONGO-1444 public void rejectsDuplicateIdInInsertAll() { - thrown.expect(DataIntegrityViolationException.class); - thrown.expectMessage("E11000 duplicate key error"); - ReactiveMongoTemplate template = new ReactiveMongoTemplate(factory); template.setWriteResultChecking(WriteResultChecking.EXCEPTION); @@ -430,18 +398,19 @@ public class ReactiveMongoTemplateTests { Person person = new Person(id, "Amol"); person.setAge(28); - List records = new ArrayList<>(); - records.add(person); - records.add(person); - - template.insertAll(records).next().block(); + StepVerifier.create(template.insertAll(Arrays.asList(person, person))) // + .expectError(DataIntegrityViolationException.class) // + .verify(); } @Test // DATAMONGO-1444 public void testFindAndUpdate() { - template.insertAll(Arrays.asList(new Person("Tom", 21), new Person("Dick", 22), new Person("Harry", 23))).next() - .block(); + StepVerifier + .create( + template.insertAll(Arrays.asList(new Person("Tom", 21), new Person("Dick", 22), new Person("Harry", 23)))) // + .expectNextCount(3) // + .verifyComplete(); Query query = new Query(Criteria.where("firstName").is("Harry")); Update update = new Update().inc("age", 1); @@ -478,62 +447,52 @@ public class ReactiveMongoTemplateTests { Sample spring = new Sample("100", "spring"); Sample data = new Sample("200", "data"); Sample mongodb = new Sample("300", "mongodb"); - template.insert(Arrays.asList(spring, data, mongodb), Sample.class).then().block(); + + StepVerifier.create(template.insert(Arrays.asList(spring, data, mongodb), Sample.class)) // + .expectNextCount(3) // + .verifyComplete(); Query qry = query(where("field").in("spring", "mongodb")); - TestSubscriber testSubscriber = TestSubscriber.create(); - template.findAllAndRemove(qry, Sample.class).subscribe(testSubscriber); + StepVerifier.create(template.findAllAndRemove(qry, Sample.class)).expectNextCount(2).verifyComplete(); - testSubscriber.awaitAndAssertNextValueCount(2); - testSubscriber.assertValues(spring, mongodb); - - assertThat(template.findOne(new Query(), Sample.class).block(), is(equalTo(data))); + StepVerifier.create(template.findOne(new Query(), Sample.class)).expectNext(data).verifyComplete(); } - @Test(expected = OptimisticLockingFailureException.class) // DATAMONGO-1444 + @Test // DATAMONGO-1444 public void optimisticLockingHandling() { // Init version PersonWithVersionPropertyOfTypeInteger person = new PersonWithVersionPropertyOfTypeInteger(); person.age = 29; person.firstName = "Patryk"; - template.save(person).block(); - List result = Flux - .from(template.findAll(PersonWithVersionPropertyOfTypeInteger.class)).collectList().block(); + StepVerifier.create(template.save(person)).expectNextCount(1).verifyComplete(); - assertThat(result, hasSize(1)); - assertThat(result.get(0).version, is(0)); + StepVerifier.create(template.findAll(PersonWithVersionPropertyOfTypeInteger.class)).consumeNextWith(actual -> { - // Version change - person = result.get(0); - person.firstName = "Patryk2"; + assertThat(actual.version, is(0)); + }).verifyComplete(); - template.save(person).block(); + StepVerifier.create(template.findAll(PersonWithVersionPropertyOfTypeInteger.class).flatMap(p -> { + + // Version change + person.firstName = "Patryk2"; + return template.save(person); + })).expectNextCount(1).verifyComplete(); assertThat(person.version, is(1)); - result = Flux.from(template.findAll(PersonWithVersionPropertyOfTypeInteger.class)).collectList().block(); + StepVerifier.create(template.findAll(PersonWithVersionPropertyOfTypeInteger.class)).consumeNextWith(actual -> { - assertThat(result, hasSize(1)); - assertThat(result.get(0).version, is(1)); + assertThat(actual.version, is(1)); + }).verifyComplete(); // Optimistic lock exception person.version = 0; person.firstName = "Patryk3"; - template.save(person).block(); - } - - @Test // DATAMONGO-1444 - public void optimisticLockingHandlingWithExistingId() { - - PersonWithVersionPropertyOfTypeInteger person = new PersonWithVersionPropertyOfTypeInteger(); - person.id = new ObjectId().toString(); - person.age = 29; - person.firstName = "Patryk"; - template.save(person); + StepVerifier.create(template.save(person)).expectError(OptimisticLockingFailureException.class).verify(); } @Test // DATAMONGO-1444 @@ -542,22 +501,30 @@ public class ReactiveMongoTemplateTests { Document dbObject = new Document(); dbObject.put("firstName", "Oliver"); - template.insert(dbObject, template.determineCollectionName(PersonWithVersionPropertyOfTypeInteger.class)); + StepVerifier + .create(template.insert(dbObject, // + template.determineCollectionName(PersonWithVersionPropertyOfTypeInteger.class))) // + .expectNextCount(1) // + .verifyComplete(); } @Test // DATAMONGO-1444 public void removesObjectFromExplicitCollection() { String collectionName = "explicit"; - template.remove(new Query(), collectionName).block(); + StepVerifier.create(template.remove(new Query(), collectionName)).expectNextCount(1).verifyComplete(); PersonWithConvertedId person = new PersonWithConvertedId(); person.name = "Dave"; - template.save(person, collectionName).block(); - assertThat(template.findAll(PersonWithConvertedId.class, collectionName).next().block(), is(notNullValue())); - template.remove(person, collectionName).block(); - assertThat(template.findAll(PersonWithConvertedId.class, collectionName).next().block(), is(nullValue())); + StepVerifier.create(template.save(person, collectionName)).expectNextCount(1).verifyComplete(); + + StepVerifier.create(template.findAll(PersonWithConvertedId.class, collectionName)).expectNextCount(1) + .verifyComplete(); + + StepVerifier.create(template.remove(person, collectionName)).expectNextCount(1).verifyComplete(); + + StepVerifier.create(template.findAll(PersonWithConvertedId.class, collectionName)).verifyComplete(); } @Test // DATAMONGO-1444 @@ -566,24 +533,23 @@ public class ReactiveMongoTemplateTests { Map map = new HashMap<>(); map.put("key", "value"); - template.save(map, "maps").block(); + StepVerifier.create(template.save(map, "maps")).expectNextCount(1).verifyComplete(); } - @Test(expected = IllegalArgumentException.class) // DATAMONGO-1444 + @Test // DATAMONGO-1444 public void savesMongoPrimitiveObjectCorrectly() { - template.save(new Object(), "collection").block(); - } - @Test(expected = IllegalArgumentException.class) // DATAMONGO-1444 - public void rejectsNullObjectToBeSaved() { - template.save((Object) null); + StepVerifier.create(template.save(new Object(), "collection")) // + .expectError(IllegalArgumentException.class) // + .verify(); } @Test // DATAMONGO-1444 public void savesPlainDbObjectCorrectly() { Document dbObject = new Document("foo", "bar"); - template.save(dbObject, "collection").block(); + + StepVerifier.create(template.save(dbObject, "collection")).expectNextCount(1).verifyComplete(); assertThat(dbObject.containsKey("_id"), is(true)); } @@ -592,20 +558,27 @@ public class ReactiveMongoTemplateTests { public void rejectsPlainObjectWithOutExplicitCollection() { Document dbObject = new Document("foo", "bar"); - template.save(dbObject, "collection").block(); - template.findById(dbObject.get("_id"), Document.class).block(); + StepVerifier.create(template.save(dbObject, "collection")).expectNextCount(1).verifyComplete(); + + StepVerifier.create(template.findById(dbObject.get("_id"), Document.class)) // + .expectError(IllegalArgumentException.class) // + .verify(); + } @Test // DATAMONGO-1444 public void readsPlainDbObjectById() { Document dbObject = new Document("foo", "bar"); - template.save(dbObject, "collection").block(); + StepVerifier.create(template.save(dbObject, "collection")).expectNextCount(1).verifyComplete(); - Document result = template.findById(dbObject.get("_id"), Document.class, "collection").block(); - assertThat(result.get("foo"), is(dbObject.get("foo"))); - assertThat(result.get("_id"), is(dbObject.get("_id"))); + StepVerifier.create(template.findById(dbObject.get("_id"), Document.class, "collection")) // + .consumeNextWith(actual -> { + + assertThat(actual.get("foo"), is(dbObject.get("foo"))); + assertThat(actual.get("_id"), is(dbObject.get("_id"))); + }).verifyComplete(); } @Test // DATAMONGO-1444 @@ -616,32 +589,38 @@ public class ReactiveMongoTemplateTests { new Venue("Flatiron Building", -73.988135, 40.741404), // new Venue("Maplewood, NJ", -74.2713, 40.73137)); - template.insertAll(venues).blockLast(); + StepVerifier.create(template.insertAll(venues)).expectNextCount(4).verifyComplete(); + IndexOperationsAdapter.blocking(template.indexOps(Venue.class)) .ensureIndex(new GeospatialIndex("location").typed(GeoSpatialIndexType.GEO_2D)); NearQuery geoFar = NearQuery.near(-73, 40, Metrics.KILOMETERS).num(10).maxDistance(150, Metrics.KILOMETERS); - template.geoNear(geoFar, Venue.class) // - .subscribeWith(TestSubscriber.create()) // - .awaitAndAssertNextValueCount(4); + StepVerifier.create(template.geoNear(geoFar, Venue.class)) // + .expectNextCount(4) // + .verifyComplete(); NearQuery geoNear = NearQuery.near(-73, 40, Metrics.KILOMETERS).num(10).maxDistance(120, Metrics.KILOMETERS); - template.geoNear(geoNear, Venue.class) // - .subscribeWith(TestSubscriber.create()) // - .await() // - .assertValueCount(3); + StepVerifier.create(template.geoNear(geoNear, Venue.class)) // + .expectNextCount(3) // + .verifyComplete(); } @Test // DATAMONGO-1444 public void writesPlainString() { - template.save("{ 'foo' : 'bar' }", "collection").block(); + + StepVerifier.create(template.save("{ 'foo' : 'bar' }", "collection")) // + .expectNextCount(1) // + .verifyComplete(); } - @Test(expected = MappingException.class) // DATAMONGO-1444 + @Test // DATAMONGO-1444 public void rejectsNonJsonStringForSave() { - template.save("Foobar!", "collection").block(); + + StepVerifier.create(template.save("Foobar!", "collection")) // + .expectError(MappingException.class) // + .verify(); } @Test // DATAMONGO-1444 @@ -650,7 +629,7 @@ public class ReactiveMongoTemplateTests { PersonWithVersionPropertyOfTypeInteger person = new PersonWithVersionPropertyOfTypeInteger(); person.firstName = "Dave"; - template.insert(person).block(); + StepVerifier.create(template.insert(person)).expectNextCount(1).verifyComplete(); assertThat(person.version, is(0)); } @@ -661,17 +640,17 @@ public class ReactiveMongoTemplateTests { PersonWithVersionPropertyOfTypeInteger person = new PersonWithVersionPropertyOfTypeInteger(); person.firstName = "Dave"; - template.insertAll(Collections.singletonList(person)).next().block(); + StepVerifier.create(template.insertAll(Collections.singleton(person))).expectNextCount(1).verifyComplete(); assertThat(person.version, is(0)); } @Test // DATAMONGO-1444 - public void queryCantBeNull() { + public void queryCanBeNull() { - List result = Flux - .from(template.findAll(PersonWithIdPropertyOfTypeObjectId.class)).collectList().block(); - assertThat(template.find(null, PersonWithIdPropertyOfTypeObjectId.class).collectList().block(), is(result)); + StepVerifier.create(template.findAll(PersonWithIdPropertyOfTypeObjectId.class)).verifyComplete(); + + StepVerifier.create(template.find(null, PersonWithIdPropertyOfTypeObjectId.class)).verifyComplete(); } @Test // DATAMONGO-1444 @@ -680,10 +659,10 @@ public class ReactiveMongoTemplateTests { PersonWithVersionPropertyOfTypeInteger person = new PersonWithVersionPropertyOfTypeInteger(); person.firstName = "Dave"; - template.save(person, "personX").block(); + StepVerifier.create(template.save(person, "personX")).expectNextCount(1).verifyComplete(); assertThat(person.version, is(0)); - template.save(person, "personX").block(); + StepVerifier.create(template.save(person, "personX")).expectNextCount(1).verifyComplete(); assertThat(person.version, is(1)); } @@ -693,7 +672,7 @@ public class ReactiveMongoTemplateTests { PersonWithVersionPropertyOfTypeLong person = new PersonWithVersionPropertyOfTypeLong(); person.firstName = "Dave"; - template.save(person).block(); + StepVerifier.create(template.save(person, "personX")).expectNextCount(1).verifyComplete(); assertThat(person.version, is(0L)); } @@ -702,25 +681,24 @@ public class ReactiveMongoTemplateTests { ReactiveMongoTemplate template = new ReactiveMongoTemplate(factory); template.setWriteResultChecking(WriteResultChecking.EXCEPTION); - template.indexOps(Person.class).ensureIndex(new Index().on("firstName", Direction.DESC).unique()).block(); + StepVerifier + .create(template.indexOps(Person.class) // + .ensureIndex(new Index().on("firstName", Direction.DESC).unique())) // + .expectNextCount(1) // + .verifyComplete(); Person person = new Person(new ObjectId(), "Amol"); person.setAge(28); - template.save(person).block(); + StepVerifier.create(template.save(person)).expectNextCount(1).verifyComplete(); person = new Person(new ObjectId(), "Amol"); person.setAge(28); - try { - template.save(person).block(); - fail("Expected DataIntegrityViolationException!"); - } catch (DataIntegrityViolationException e) { - assertThat(e.getMessage(), containsString("E11000 duplicate key error")); - } + StepVerifier.create(template.save(person)).expectError(DataIntegrityViolationException.class).verify(); } - @Test(expected = DuplicateKeyException.class) // DATAMONGO-1444 + @Test // DATAMONGO-1444 public void preventsDuplicateInsert() { template.setWriteConcern(WriteConcern.MAJORITY); @@ -728,24 +706,25 @@ public class ReactiveMongoTemplateTests { PersonWithVersionPropertyOfTypeInteger person = new PersonWithVersionPropertyOfTypeInteger(); person.firstName = "Dave"; - template.save(person).block(); + StepVerifier.create(template.save(person)).expectNextCount(1).verifyComplete(); assertThat(person.version, is(0)); person.version = null; - template.save(person).block(); + StepVerifier.create(template.save(person)).expectError(DuplicateKeyException.class).verify(); } @Test // DATAMONGO-1444 public void countAndFindWithoutTypeInformation() { Person person = new Person(); - template.save(person).block(); + StepVerifier.create(template.save(person)).expectNextCount(1).verifyComplete(); Query query = query(where("_id").is(person.getId())); String collectionName = template.getCollectionName(Person.class); - assertThat(Flux.from(template.find(query, HashMap.class, collectionName)).collectList().block(), hasSize(1)); - assertThat(template.count(query, collectionName).block(), is(1L)); + StepVerifier.create(template.find(query, HashMap.class, collectionName)).expectNextCount(1).verifyComplete(); + + StepVerifier.create(template.count(query, collectionName)).expectNext(1L).verifyComplete(); } @Test // DATAMONGO-1444 @@ -755,27 +734,36 @@ public class ReactiveMongoTemplateTests { person.firstname = "Dave"; person.lastname = "Matthews"; - template.save(person).block(); + StepVerifier.create(template.save(person)).expectNextCount(1).verifyComplete(); + assertThat(person.id, is(notNullValue())); person.lastname = null; - template.save(person).block(); + StepVerifier.create(template.save(person)).expectNextCount(1).verifyComplete(); - person = template.findOne(query(where("id").is(person.id)), VersionedPerson.class).block(); - assertThat(person.lastname, is(nullValue())); + StepVerifier.create(template.findOne(query(where("id").is(person.id)), VersionedPerson.class)) // + .consumeNextWith(actual -> { + + assertThat(actual.lastname, is(nullValue())); + }) // + .verifyComplete(); } @Test // DATAMONGO-1444 public void nullsValuesForUpdatesOfUnversionedEntity() { Person person = new Person("Dave"); - template.save(person).block(); + StepVerifier.create(template.save(person)).expectNextCount(1).verifyComplete(); person.setFirstName(null); - template.save(person).block(); + StepVerifier.create(template.save(person)).expectNextCount(1).verifyComplete(); - person = template.findOne(query(where("id").is(person.getId())), Person.class).block(); - assertThat(person.getFirstName(), is(nullValue())); + StepVerifier.create(template.findOne(query(where("id").is(person.getId())), Person.class)) // + .consumeNextWith(actual -> { + + assertThat(actual.getFirstName(), is(nullValue())); + }) // + .verifyComplete(); } @Test // DATAMONGO-1444 @@ -783,32 +771,42 @@ public class ReactiveMongoTemplateTests { Document dbObject = new Document().append("first", "first").append("second", "second"); - template.save(dbObject, "collection").block(); + StepVerifier.create(template.save(dbObject, "collection")).expectNextCount(1).verifyComplete(); - Document result = template.findAll(Document.class, "collection").next().block(); - assertThat(result.containsKey("first"), is(true)); + StepVerifier.create(template.findAll(Document.class, "collection")) // + .consumeNextWith(actual -> { + + assertThat(actual.containsKey("first"), is(true)); + }) // + .verifyComplete(); } @Test // DATAMONGO-1444 public void executesExistsCorrectly() { Sample sample = new Sample(); - template.save(sample).block(); + StepVerifier.create(template.save(sample)).expectNextCount(1).verifyComplete(); Query query = query(where("id").is(sample.id)); - assertThat(template.exists(query, Sample.class).block(), is(true)); - assertThat(template.exists(query(where("_id").is(sample.id)), template.getCollectionName(Sample.class)).block(), - is(true)); - assertThat(template.exists(query, Sample.class, template.getCollectionName(Sample.class)).block(), is(true)); + StepVerifier.create(template.exists(query, Sample.class)).expectNext(true).verifyComplete(); + + StepVerifier.create(template.exists(query(where("_id").is(sample.id)), template.getCollectionName(Sample.class))) + .expectNext(true).verifyComplete(); + + StepVerifier.create(template.exists(query, Sample.class, template.getCollectionName(Sample.class))).expectNext(true) + .verifyComplete(); } @Test // DATAMONGO-1444 public void tailStreamsData() throws InterruptedException { - template.dropCollection("capped").block(); - template.createCollection("capped", new CollectionOptions(1000, 10, true)).block(); - template.insert(new Document("random", Math.random()).append("key", "value"), "capped").block(); + StepVerifier.create(template.dropCollection("capped") + .then(template.createCollection("capped", // + new CollectionOptions(1000, 10, true))) + .then(template.insert(new Document("random", Math.random()).append("key", "value"), // + "capped"))) + .expectNextCount(1).verifyComplete(); BlockingQueue documents = new LinkedBlockingQueue<>(1000); @@ -825,9 +823,12 @@ public class ReactiveMongoTemplateTests { @Test // DATAMONGO-1444 public void tailStreamsDataUntilCancellation() throws InterruptedException { - template.dropCollection("capped").block(); - template.createCollection("capped", new CollectionOptions(1000, 10, true)).block(); - template.insert(new Document("random", Math.random()).append("key", "value"), "capped").block(); + StepVerifier.create(template.dropCollection("capped") + .then(template.createCollection("capped", // + new CollectionOptions(1000, 10, true))) + .then(template.insert(new Document("random", Math.random()).append("key", "value"), // + "capped"))) + .expectNextCount(1).verifyComplete(); BlockingQueue documents = new LinkedBlockingQueue<>(1000); @@ -838,15 +839,30 @@ public class ReactiveMongoTemplateTests { assertThat(documents.poll(5, TimeUnit.SECONDS), is(notNullValue())); assertThat(documents.isEmpty(), is(true)); - template.insert(new Document("random", Math.random()).append("key", "value"), "capped").block(); + StepVerifier.create(template.insert(new Document("random", Math.random()).append("key", "value"), "capped")) // + .expectNextCount(1) // + .verifyComplete(); + assertThat(documents.poll(5, TimeUnit.SECONDS), is(notNullValue())); cancellation.dispose(); - template.insert(new Document("random", Math.random()).append("key", "value"), "capped").block(); + StepVerifier.create(template.insert(new Document("random", Math.random()).append("key", "value"), "capped")) // + .expectNextCount(1) // + .verifyComplete(); + assertThat(documents.poll(1, TimeUnit.SECONDS), is(nullValue())); } + private PersonWithAList createPersonWithAList(String firstname, int age) { + + PersonWithAList p = new PersonWithAList(); + p.setFirstName(firstname); + p.setAge(age); + + return p; + } + @Data static class Sample { diff --git a/spring-data-mongodb/src/test/java/org/springframework/data/mongodb/repository/ConvertingReactiveMongoRepositoryTests.java b/spring-data-mongodb/src/test/java/org/springframework/data/mongodb/repository/ConvertingReactiveMongoRepositoryTests.java index 741694cf5..af13cf100 100644 --- a/spring-data-mongodb/src/test/java/org/springframework/data/mongodb/repository/ConvertingReactiveMongoRepositoryTests.java +++ b/spring-data-mongodb/src/test/java/org/springframework/data/mongodb/repository/ConvertingReactiveMongoRepositoryTests.java @@ -13,12 +13,19 @@ * See the License for the specific language governing permissions and * limitations under the License. */ - package org.springframework.data.mongodb.repository; import static org.hamcrest.Matchers.*; import static org.junit.Assert.*; +import lombok.Data; +import lombok.NoArgsConstructor; +import reactor.core.publisher.Flux; +import reactor.core.publisher.Mono; +import reactor.test.StepVerifier; +import rx.Observable; +import rx.Single; + import java.util.Arrays; import java.util.List; @@ -28,28 +35,17 @@ import org.junit.runner.RunWith; import org.reactivestreams.Publisher; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.context.annotation.ComponentScan.Filter; -import org.springframework.context.annotation.Configuration; import org.springframework.context.annotation.ImportResource; import org.springframework.data.annotation.Id; import org.springframework.data.domain.Sort; -import org.springframework.data.mongodb.core.ReactiveMongoTemplate; import org.springframework.data.mongodb.core.mapping.Document; import org.springframework.data.mongodb.repository.config.EnableReactiveMongoRepositories; -import org.springframework.data.repository.RepositoryDefinition; import org.springframework.data.repository.reactive.ReactiveSortingRepository; import org.springframework.data.repository.reactive.RxJava1SortingRepository; import org.springframework.stereotype.Repository; import org.springframework.test.context.ContextConfiguration; import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; -import lombok.Data; -import lombok.NoArgsConstructor; -import reactor.core.publisher.Flux; -import reactor.core.publisher.Mono; -import reactor.test.TestSubscriber; -import rx.Observable; -import rx.Single; - /** * Test for {@link ReactiveMongoRepository} using reactive wrapper type conversion. * @@ -59,11 +55,11 @@ import rx.Single; @ContextConfiguration(classes = ConvertingReactiveMongoRepositoryTests.Config.class) public class ConvertingReactiveMongoRepositoryTests { - @EnableReactiveMongoRepositories(includeFilters = @Filter(value = Repository.class), considerNestedRepositories = true) + @EnableReactiveMongoRepositories(includeFilters = @Filter(value = Repository.class), + considerNestedRepositories = true) @ImportResource("classpath:reactive-infrastructure.xml") static class Config {} - @Autowired ReactiveMongoTemplate template; @Autowired MixedReactivePersonRepostitory reactiveRepository; @Autowired ReactivePersonRepostitory reactivePersonRepostitory; @Autowired RxJavaPersonRepostitory rxJavaPersonRepostitory; @@ -71,9 +67,9 @@ public class ConvertingReactiveMongoRepositoryTests { ReactivePerson dave, oliver, carter, boyd, stefan, leroi, alicia; @Before - public void setUp() throws Exception { + public void setUp() { - reactiveRepository.deleteAll().block(); + StepVerifier.create(reactiveRepository.deleteAll()).verifyComplete(); dave = new ReactivePerson("Dave", "Matthews", 42); oliver = new ReactivePerson("Oliver August", "Matthews", 4); @@ -83,123 +79,121 @@ public class ConvertingReactiveMongoRepositoryTests { leroi = new ReactivePerson("Leroi", "Moore", 41); alicia = new ReactivePerson("Alicia", "Keys", 30); - TestSubscriber subscriber = TestSubscriber.create(); - reactiveRepository.save(Arrays.asList(oliver, dave, carter, boyd, stefan, leroi, alicia)).subscribe(subscriber); - - subscriber.await().assertComplete().assertNoError(); + StepVerifier.create(reactiveRepository.save(Arrays.asList(oliver, dave, carter, boyd, stefan, leroi, alicia))) // + .expectNextCount(7) // + .verifyComplete(); } @Test // DATAMONGO-1444 - public void reactiveStreamsMethodsShouldWork() throws Exception { - - TestSubscriber subscriber = TestSubscriber.subscribe(reactivePersonRepostitory.exists(dave.getId())); - - subscriber.awaitAndAssertNextValueCount(1).assertValues(true); + public void reactiveStreamsMethodsShouldWork() { + StepVerifier.create(reactivePersonRepostitory.exists(dave.getId())).expectNext(true).verifyComplete(); } @Test // DATAMONGO-1444 - public void reactiveStreamsQueryMethodsShouldWork() throws Exception { - - TestSubscriber subscriber = TestSubscriber - .subscribe(reactivePersonRepostitory.findByLastname(boyd.getLastname())); - - subscriber.awaitAndAssertNextValueCount(1).assertValues(boyd); + public void reactiveStreamsQueryMethodsShouldWork() { + StepVerifier.create(reactivePersonRepostitory.findByLastname(boyd.getLastname())).expectNext(boyd).verifyComplete(); } @Test // DATAMONGO-1444 - public void simpleRxJavaMethodsShouldWork() throws Exception { + public void simpleRxJavaMethodsShouldWork() { - rx.observers.TestSubscriber subscriber = new rx.observers.TestSubscriber<>(); - rxJavaPersonRepostitory.exists(dave.getId()).subscribe(subscriber); - - subscriber.awaitTerminalEvent(); - subscriber.assertCompleted(); - subscriber.assertNoErrors(); - subscriber.assertValue(true); + rxJavaPersonRepostitory.exists(dave.getId()) // + .test() // + .awaitTerminalEvent() // + .assertValue(true) // + .assertNoErrors() // + .assertCompleted(); } @Test // DATAMONGO-1444 - public void existsWithSingleRxJavaIdMethodsShouldWork() throws Exception { + public void existsWithSingleRxJavaIdMethodsShouldWork() { - rx.observers.TestSubscriber subscriber = new rx.observers.TestSubscriber<>(); - rxJavaPersonRepostitory.exists(Single.just(dave.getId())).subscribe(subscriber); - - subscriber.awaitTerminalEvent(); - subscriber.assertCompleted(); - subscriber.assertNoErrors(); - subscriber.assertValue(true); + rxJavaPersonRepostitory.exists(Single.just(dave.getId())) // + .test() // + .awaitTerminalEvent() // + .assertValue(true) // + .assertNoErrors() // + .assertCompleted(); } @Test // DATAMONGO-1444 - public void singleRxJavaQueryMethodShouldWork() throws Exception { + public void singleRxJavaQueryMethodShouldWork() { - rx.observers.TestSubscriber subscriber = new rx.observers.TestSubscriber<>(); - rxJavaPersonRepostitory.findByFirstnameAndLastname(dave.getFirstname(), dave.getLastname()).subscribe(subscriber); - - subscriber.awaitTerminalEvent(); - subscriber.assertCompleted(); - subscriber.assertNoErrors(); - subscriber.assertValue(dave); + rxJavaPersonRepostitory.findByFirstnameAndLastname(dave.getFirstname(), dave.getLastname()) // + .test() // + .awaitTerminalEvent() // + .assertValue(dave) // + .assertNoErrors() // + .assertCompleted(); } @Test // DATAMONGO-1444 - public void singleProjectedRxJavaQueryMethodShouldWork() throws Exception { + public void singleProjectedRxJavaQueryMethodShouldWork() { - rx.observers.TestSubscriber subscriber = new rx.observers.TestSubscriber<>(); - rxJavaPersonRepostitory.findProjectedByLastname(carter.getLastname()).subscribe(subscriber); + List people = rxJavaPersonRepostitory.findProjectedByLastname(carter.getLastname()) // + .test() // + .awaitTerminalEvent() // + .assertValueCount(1) // + .assertNoErrors() // + .assertCompleted() // + .getOnNextEvents(); - subscriber.awaitTerminalEvent(); - subscriber.assertCompleted(); - subscriber.assertNoErrors(); - - ProjectedPerson projectedPerson = subscriber.getOnNextEvents().get(0); + ProjectedPerson projectedPerson = people.get(0); assertThat(projectedPerson.getFirstname(), is(equalTo(carter.getFirstname()))); } @Test // DATAMONGO-1444 - public void observableRxJavaQueryMethodShouldWork() throws Exception { + public void observableRxJavaQueryMethodShouldWork() { - rx.observers.TestSubscriber subscriber = new rx.observers.TestSubscriber<>(); - rxJavaPersonRepostitory.findByLastname(boyd.getLastname()).subscribe(subscriber); - - subscriber.awaitTerminalEvent(); - subscriber.assertCompleted(); - subscriber.assertNoErrors(); - subscriber.assertValue(boyd); + rxJavaPersonRepostitory.findByLastname(boyd.getLastname()) // + .test() // + .awaitTerminalEvent() // + .assertValue(boyd) // + .assertNoErrors() // + .assertCompleted() // + .getOnNextEvents(); } @Test // DATAMONGO-1444 - public void mixedRepositoryShouldWork() throws Exception { + public void mixedRepositoryShouldWork() { - ReactivePerson value = reactiveRepository.findByLastname(boyd.getLastname()).toBlocking().value(); - - assertThat(value, is(equalTo(boyd))); + reactiveRepository.findByLastname(boyd.getLastname()) // + .test() // + .awaitTerminalEvent() // + .assertValue(boyd) // + .assertNoErrors() // + .assertCompleted() // + .getOnNextEvents(); } @Test // DATAMONGO-1444 - public void shouldFindOneBySingleOfLastName() throws Exception { + public void shouldFindOneBySingleOfLastName() { - ReactivePerson carter = reactiveRepository.findByLastname(Single.just("Beauford")).block(); - - assertThat(carter.getFirstname(), is(equalTo("Carter"))); + StepVerifier.create(reactiveRepository.findByLastname(Single.just(carter.getLastname()))) // + .expectNext(carter) // + .verifyComplete(); } @Test // DATAMONGO-1444 - public void shouldFindByObservableOfLastNameIn() throws Exception { + public void shouldFindByObservableOfLastNameIn() { - List persons = reactiveRepository.findByLastnameIn(Observable.just("Beauford", "Matthews")) - .collectList().block(); - - assertThat(persons, hasItems(carter, dave, oliver)); + StepVerifier.create(reactiveRepository.findByLastnameIn(Observable.just(carter.getLastname(), dave.getLastname()))) // + .expectNextCount(3) // + .verifyComplete(); } @Test // DATAMONGO-1444 - public void shouldFindByPublisherOfLastNameInAndAgeGreater() throws Exception { + public void shouldFindByPublisherOfLastNameInAndAgeGreater() { - List persons = reactiveRepository - .findByLastnameInAndAgeGreaterThan(Flux.just("Beauford", "Matthews"), 41).toList().toBlocking().single(); + List people = reactiveRepository + .findByLastnameInAndAgeGreaterThan(Flux.just(carter.getLastname(), dave.getLastname()), 41).test() // + .awaitTerminalEvent() // + .assertValueCount(2) // + .assertNoErrors() // + .assertCompleted() // + .getOnNextEvents(); - assertThat(persons, hasItems(carter, dave)); + assertThat(people, hasItems(carter, dave)); } @Repository diff --git a/spring-data-mongodb/src/test/java/org/springframework/data/mongodb/repository/ReactiveMongoRepositoryTests.java b/spring-data-mongodb/src/test/java/org/springframework/data/mongodb/repository/ReactiveMongoRepositoryTests.java index 642e83d23..e87f8d90d 100644 --- a/spring-data-mongodb/src/test/java/org/springframework/data/mongodb/repository/ReactiveMongoRepositoryTests.java +++ b/spring-data-mongodb/src/test/java/org/springframework/data/mongodb/repository/ReactiveMongoRepositoryTests.java @@ -19,8 +19,13 @@ import static org.hamcrest.Matchers.*; import static org.junit.Assert.*; import static org.springframework.data.domain.Sort.Direction.*; +import lombok.NoArgsConstructor; +import reactor.core.Cancellation; +import reactor.core.publisher.Flux; +import reactor.core.publisher.Mono; +import reactor.test.StepVerifier; + import java.util.Arrays; -import java.util.List; import java.util.concurrent.BlockingQueue; import java.util.concurrent.LinkedBlockingDeque; import java.util.concurrent.TimeUnit; @@ -34,12 +39,9 @@ import org.springframework.beans.factory.BeanClassLoaderAware; import org.springframework.beans.factory.BeanFactory; import org.springframework.beans.factory.BeanFactoryAware; import org.springframework.beans.factory.annotation.Autowired; -import org.springframework.data.domain.Page; import org.springframework.data.domain.PageRequest; import org.springframework.data.domain.Pageable; -import org.springframework.data.domain.Slice; import org.springframework.data.domain.Sort; -import org.springframework.data.domain.Sort.Order; import org.springframework.data.geo.Circle; import org.springframework.data.geo.Distance; import org.springframework.data.geo.GeoResult; @@ -55,13 +57,7 @@ import org.springframework.data.repository.Repository; import org.springframework.data.repository.query.DefaultEvaluationContextProvider; import org.springframework.test.context.ContextConfiguration; import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; - -import lombok.NoArgsConstructor; import org.springframework.util.ClassUtils; -import reactor.core.Cancellation; -import reactor.core.publisher.Flux; -import reactor.core.publisher.Mono; -import reactor.test.TestSubscriber; /** * Test for {@link ReactiveMongoRepository} query methods. @@ -75,10 +71,10 @@ public class ReactiveMongoRepositoryTests implements BeanClassLoaderAware, BeanF @Autowired ReactiveMongoTemplate template; ReactiveMongoRepositoryFactory factory; - private ClassLoader classLoader; - private BeanFactory beanFactory; - private ReactivePersonRepository repository; - private ReactiveCappedCollectionRepository cappedRepository; + ClassLoader classLoader; + BeanFactory beanFactory; + ReactivePersonRepository repository; + ReactiveCappedCollectionRepository cappedRepository; Person dave, oliver, carter, boyd, stefan, leroi, alicia; @@ -104,7 +100,7 @@ public class ReactiveMongoRepositoryTests implements BeanClassLoaderAware, BeanF repository = factory.getRepository(ReactivePersonRepository.class); cappedRepository = factory.getRepository(ReactiveCappedCollectionRepository.class); - repository.deleteAll().block(); + StepVerifier.create(repository.deleteAll()).verifyComplete(); dave = new Person("Dave", "Matthews", 42); oliver = new Person("Oliver August", "Matthews", 4); @@ -118,78 +114,72 @@ public class ReactiveMongoRepositoryTests implements BeanClassLoaderAware, BeanF alicia = new Person("Alicia", "Keys", 30, Sex.FEMALE); - TestSubscriber subscriber = TestSubscriber.create(); - repository.save(Arrays.asList(oliver, dave, carter, boyd, stefan, leroi, alicia)).subscribe(subscriber); - - subscriber.await().assertComplete().assertNoError(); + StepVerifier.create(repository.save(Arrays.asList(oliver, dave, carter, boyd, stefan, leroi, alicia))) // + .expectNextCount(7) // + .verifyComplete(); } @Test // DATAMONGO-1444 public void shouldFindByLastName() { - - List list = repository.findByLastname("Matthews").collectList().block(); - - assertThat(list, hasSize(2)); + StepVerifier.create(repository.findByLastname(dave.getLastname())).expectNextCount(2).verifyComplete(); } @Test // DATAMONGO-1444 public void shouldFindOneByLastName() { - - Person carter = repository.findOneByLastname("Beauford").block(); - - assertThat(carter.getFirstname(), is(equalTo("Carter"))); + StepVerifier.create(repository.findOneByLastname(carter.getLastname())).expectNext(carter); } @Test // DATAMONGO-1444 public void shouldFindOneByPublisherOfLastName() { - - Person carter = repository.findByLastname(Mono.just("Beauford")).block(); - - assertThat(carter.getFirstname(), is(equalTo("Carter"))); + StepVerifier.create(repository.findByLastname(Mono.just(carter.getLastname()))).expectNext(carter); } @Test // DATAMONGO-1444 public void shouldFindByPublisherOfLastNameIn() { - - List persons = repository.findByLastnameIn(Flux.just("Beauford", "Matthews")).collectList().block(); - - assertThat(persons, hasItems(carter, dave, oliver)); + StepVerifier.create(repository.findByLastnameIn(Flux.just(carter.getLastname(), dave.getLastname()))) // + .expectNextCount(3) // + .verifyComplete(); } @Test // DATAMONGO-1444 public void shouldFindByPublisherOfLastNameInAndAgeGreater() { - List persons = repository.findByLastnameInAndAgeGreaterThan(Flux.just("Beauford", "Matthews"), 41) - .collectList().block(); - - assertThat(persons, hasItems(carter, dave)); + StepVerifier + .create(repository.findByLastnameInAndAgeGreaterThan(Flux.just(carter.getLastname(), dave.getLastname()), 41)) // + .expectNextCount(2) // + .verifyComplete(); } @Test // DATAMONGO-1444 public void shouldFindUsingPublishersInStringQuery() { - List persons = repository.findStringQuery(Flux.just("Beauford", "Matthews"), Mono.just(41)).collectList() - .block(); - - assertThat(persons, hasItems(carter, dave)); + StepVerifier.create(repository.findStringQuery(Flux.just("Beauford", "Matthews"), Mono.just(41))) // + .expectNextCount(2) // + .verifyComplete(); } @Test // DATAMONGO-1444 public void shouldFindByLastNameAndSort() { + StepVerifier.create(repository.findByLastname("Matthews", Sort.by(ASC, "age"))) // + .expectNext(oliver, dave) // + .verifyComplete(); - List persons = repository.findByLastname("Matthews", new Sort(new Order(ASC, "age"))).collectList().block(); - assertThat(persons, contains(oliver, dave)); - - persons = repository.findByLastname("Matthews", new Sort(new Order(DESC, "age"))).collectList().block(); - assertThat(persons, contains(dave, oliver)); + StepVerifier.create(repository.findByLastname("Matthews", Sort.by(DESC, "age"))) // + .expectNext(dave, oliver) // + .verifyComplete(); } @Test // DATAMONGO-1444 public void shouldUseInfiniteStream() throws Exception { - template.dropCollection(Capped.class).block(); - template.createCollection(Capped.class, new CollectionOptions(1000, 100, true)).block(); - template.insert(new Capped("value", Math.random())).block(); + StepVerifier + .create(template.dropCollection(Capped.class) // + .then(template.createCollection(Capped.class, // + new CollectionOptions(1000, 100, true)))) // + .expectNextCount(1) // + .verifyComplete(); + + StepVerifier.create(template.insert(new Capped("value", Math.random()))).expectNextCount(1).verifyComplete(); BlockingQueue documents = new LinkedBlockingDeque<>(100); @@ -197,7 +187,7 @@ public class ReactiveMongoRepositoryTests implements BeanClassLoaderAware, BeanF assertThat(documents.poll(5, TimeUnit.SECONDS), is(notNullValue())); - template.insert(new Capped("value", Math.random())).block(); + StepVerifier.create(template.insert(new Capped("value", Math.random()))).expectNextCount(1).verifyComplete(); assertThat(documents.poll(5, TimeUnit.SECONDS), is(notNullValue())); assertThat(documents.isEmpty(), is(true)); @@ -207,9 +197,14 @@ public class ReactiveMongoRepositoryTests implements BeanClassLoaderAware, BeanF @Test // DATAMONGO-1444 public void shouldUseInfiniteStreamWithProjection() throws Exception { - template.dropCollection(Capped.class).block(); - template.createCollection(Capped.class, new CollectionOptions(1000, 100, true)).block(); - template.insert(new Capped("value", Math.random())).block(); + StepVerifier + .create(template.dropCollection(Capped.class) // + .then(template.createCollection(Capped.class, // + new CollectionOptions(1000, 100, true)))) // + .expectNextCount(1) // + .verifyComplete(); + + StepVerifier.create(template.insert(new Capped("value", Math.random()))).expectNextCount(1).verifyComplete(); BlockingQueue documents = new LinkedBlockingDeque<>(100); @@ -219,7 +214,7 @@ public class ReactiveMongoRepositoryTests implements BeanClassLoaderAware, BeanF assertThat(projection1, is(notNullValue())); assertThat(projection1.getRandom(), is(not(0))); - template.insert(new Capped("value", Math.random())).block(); + StepVerifier.create(template.insert(new Capped("value", Math.random()))).expectNextCount(1).verifyComplete(); CappedProjection projection2 = documents.poll(5, TimeUnit.SECONDS); assertThat(projection2, is(notNullValue())); @@ -235,11 +230,11 @@ public class ReactiveMongoRepositoryTests implements BeanClassLoaderAware, BeanF Point point = new Point(-73.99171, 40.738868); dave.setLocation(point); - repository.save(dave).block(); + StepVerifier.create(repository.save(dave)).expectNextCount(1).verifyComplete(); - repository.findByLocationWithin(new Circle(-78.99171, 45.738868, 170)) // - .subscribeWith(TestSubscriber.create()) // - .awaitAndAssertNextValues(dave); + StepVerifier.create(repository.findByLocationWithin(new Circle(-78.99171, 45.738868, 170))) // + .expectNext(dave) // + .verifyComplete(); } @Test // DATAMONGO-1444 @@ -247,11 +242,13 @@ public class ReactiveMongoRepositoryTests implements BeanClassLoaderAware, BeanF Point point = new Point(-73.99171, 40.738868); dave.setLocation(point); - repository.save(dave).block(); + StepVerifier.create(repository.save(dave)).expectNextCount(1).verifyComplete(); - repository.findByLocationWithin(new Circle(-78.99171, 45.738868, 170), new PageRequest(0, 10)) // - .subscribeWith(TestSubscriber.create()) // - .awaitAndAssertNextValues(dave); + StepVerifier + .create(repository.findByLocationWithin(new Circle(-78.99171, 45.738868, 170), // + PageRequest.of(0, 10))) // + .expectNext(dave) // + .verifyComplete(); } @Test // DATAMONGO-1444 @@ -259,15 +256,15 @@ public class ReactiveMongoRepositoryTests implements BeanClassLoaderAware, BeanF Point point = new Point(-73.99171, 40.738868); dave.setLocation(point); - repository.save(dave).block(); + StepVerifier.create(repository.save(dave)).expectNextCount(1).verifyComplete(); - repository.findByLocationNear(new Point(-73.99, 40.73), new Distance(2000, Metrics.KILOMETERS)) // - .subscribeWith(TestSubscriber.create()) // - .awaitAndAssertNextValuesWith(personGeoResult -> { + StepVerifier.create(repository.findByLocationNear(new Point(-73.99, 40.73), // + new Distance(2000, Metrics.KILOMETERS)) // + ).consumeNextWith(actual -> { - assertThat(personGeoResult.getDistance().getValue(), is(closeTo(1, 1))); - assertThat(personGeoResult.getContent(), is(equalTo(dave))); - }); + assertThat(actual.getDistance().getValue(), is(closeTo(1, 1))); + assertThat(actual.getContent(), is(equalTo(dave))); + }).verifyComplete(); } @Test // DATAMONGO-1444 @@ -275,15 +272,17 @@ public class ReactiveMongoRepositoryTests implements BeanClassLoaderAware, BeanF Point point = new Point(-73.99171, 40.738868); dave.setLocation(point); - repository.save(dave).block(); + StepVerifier.create(repository.save(dave)).expectNextCount(1).verifyComplete(); - repository.findByLocationNear(new Point(-73.99, 40.73), new Distance(2000, Metrics.KILOMETERS), new PageRequest(0, 10)) // - .subscribeWith(TestSubscriber.create()) // - .awaitAndAssertNextValuesWith(personGeoResult -> { + StepVerifier + .create(repository.findByLocationNear(new Point(-73.99, 40.73), // + new Distance(2000, Metrics.KILOMETERS), // + PageRequest.of(0, 10))) // + .consumeNextWith(actual -> { - assertThat(personGeoResult.getDistance().getValue(), is(closeTo(1, 1))); - assertThat(personGeoResult.getContent(), is(equalTo(dave))); - }); + assertThat(actual.getDistance().getValue(), is(closeTo(1, 1))); + assertThat(actual.getContent(), is(equalTo(dave))); + }).verifyComplete(); } @Test // DATAMONGO-1444 @@ -291,11 +290,13 @@ public class ReactiveMongoRepositoryTests implements BeanClassLoaderAware, BeanF Point point = new Point(-73.99171, 40.738868); dave.setLocation(point); - repository.save(dave).block(); + StepVerifier.create(repository.save(dave)).expectNextCount(1).verifyComplete(); - repository.findPersonByLocationNear(new Point(-73.99, 40.73), new Distance(2000, Metrics.KILOMETERS)) // - .subscribeWith(TestSubscriber.create()) // - .awaitAndAssertNextValues(dave); + StepVerifier + .create(repository.findPersonByLocationNear(new Point(-73.99, 40.73), // + new Distance(2000, Metrics.KILOMETERS))) // + .expectNext(dave) // + .verifyComplete(); } interface ReactivePersonRepository extends ReactiveMongoRepository { diff --git a/spring-data-mongodb/src/test/java/org/springframework/data/mongodb/repository/SimpleReactiveMongoRepositoryTests.java b/spring-data-mongodb/src/test/java/org/springframework/data/mongodb/repository/SimpleReactiveMongoRepositoryTests.java index fa6d4d473..8d3a1aead 100644 --- a/spring-data-mongodb/src/test/java/org/springframework/data/mongodb/repository/SimpleReactiveMongoRepositoryTests.java +++ b/spring-data-mongodb/src/test/java/org/springframework/data/mongodb/repository/SimpleReactiveMongoRepositoryTests.java @@ -13,15 +13,18 @@ * See the License for the specific language governing permissions and * limitations under the License. */ - package org.springframework.data.mongodb.repository; import static org.hamcrest.Matchers.*; import static org.junit.Assert.*; +import lombok.Data; +import lombok.NoArgsConstructor; +import reactor.core.publisher.Flux; +import reactor.core.publisher.Mono; +import reactor.test.StepVerifier; + import java.util.Arrays; -import java.util.List; -import java.util.stream.Collectors; import org.junit.Before; import org.junit.Test; @@ -32,8 +35,6 @@ import org.springframework.beans.factory.BeanFactory; import org.springframework.beans.factory.BeanFactoryAware; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.data.annotation.Id; -import org.springframework.data.domain.Page; -import org.springframework.data.domain.PageRequest; import org.springframework.data.domain.Sort; import org.springframework.data.domain.Sort.Direction; import org.springframework.data.domain.Sort.Order; @@ -45,12 +46,6 @@ import org.springframework.test.context.ContextConfiguration; import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; import org.springframework.util.ClassUtils; -import lombok.Data; -import lombok.NoArgsConstructor; -import reactor.core.publisher.Flux; -import reactor.core.publisher.Mono; -import reactor.test.TestSubscriber; - /** * Test for {@link ReactiveMongoRepository}. * @@ -62,10 +57,10 @@ public class SimpleReactiveMongoRepositoryTests implements BeanClassLoaderAware, @Autowired private ReactiveMongoTemplate template; - private ReactiveMongoRepositoryFactory factory; - private ClassLoader classLoader; - private BeanFactory beanFactory; - private ReactivePersonRepostitory repository; + ReactiveMongoRepositoryFactory factory; + ClassLoader classLoader; + BeanFactory beanFactory; + ReactivePersonRepostitory repository; private ReactivePerson dave, oliver, carter, boyd, stefan, leroi, alicia; @@ -90,7 +85,7 @@ public class SimpleReactiveMongoRepositoryTests implements BeanClassLoaderAware, repository = factory.getRepository(ReactivePersonRepostitory.class); - repository.deleteAll().block(); + StepVerifier.create(repository.deleteAll()).verifyComplete(); dave = new ReactivePerson("Dave", "Matthews", 42); oliver = new ReactivePerson("Oliver August", "Matthews", 4); @@ -100,134 +95,92 @@ public class SimpleReactiveMongoRepositoryTests implements BeanClassLoaderAware, leroi = new ReactivePerson("Leroi", "Moore", 41); alicia = new ReactivePerson("Alicia", "Keys", 30); - TestSubscriber subscriber = TestSubscriber.create(); - repository.save(Arrays.asList(oliver, dave, carter, boyd, stefan, leroi, alicia)).subscribe(subscriber); - - subscriber.await().assertComplete().assertNoError(); + StepVerifier.create(repository.save(Arrays.asList(oliver, dave, carter, boyd, stefan, leroi, alicia))) // + .expectNextCount(7) // + .verifyComplete(); } @Test // DATAMONGO-1444 public void existsByIdShouldReturnTrueForExistingObject() { - - Boolean exists = repository.exists(dave.id).block(); - - assertThat(exists, is(true)); + StepVerifier.create(repository.exists(dave.id)).expectNext(true).verifyComplete(); } @Test // DATAMONGO-1444 public void existsByIdShouldReturnFalseForAbsentObject() { - - TestSubscriber testSubscriber = TestSubscriber.subscribe(repository.exists("unknown")); - - testSubscriber.await().assertComplete().assertValues(false).assertNoError(); + StepVerifier.create(repository.exists("unknown")).expectNext(false).verifyComplete(); } @Test // DATAMONGO-1444 public void existsByMonoOfIdShouldReturnTrueForExistingObject() { - - Boolean exists = repository.exists(Mono.just(dave.id)).block(); - assertThat(exists, is(true)); + StepVerifier.create(repository.exists(Mono.just(dave.id))).expectNext(true).verifyComplete(); } @Test // DATAMONGO-1444 public void existsByEmptyMonoOfIdShouldReturnEmptyMono() { - - TestSubscriber testSubscriber = TestSubscriber.subscribe(repository.exists(Mono.empty())); - - testSubscriber.await().assertComplete().assertNoValues().assertNoError(); + StepVerifier.create(repository.exists(Mono.empty())).verifyComplete(); } @Test // DATAMONGO-1444 public void findOneShouldReturnObject() { - - ReactivePerson person = repository.findOne(dave.id).block(); - - assertThat(person.getFirstname(), is(equalTo("Dave"))); + StepVerifier.create(repository.findOne(dave.id)).expectNext(dave).verifyComplete(); } @Test // DATAMONGO-1444 public void findOneShouldCompleteWithoutValueForAbsentObject() { - - TestSubscriber testSubscriber = TestSubscriber.subscribe(repository.findOne("unknown")); - - testSubscriber.await().assertComplete().assertNoValues().assertNoError(); + StepVerifier.create(repository.findOne("unknown")).verifyComplete(); } @Test // DATAMONGO-1444 public void findOneByMonoOfIdShouldReturnTrueForExistingObject() { - - ReactivePerson person = repository.findOne(Mono.just(dave.id)).block(); - - assertThat(person.id, is(equalTo(dave.id))); + StepVerifier.create(repository.findOne(Mono.just(dave.id))).expectNext(dave).verifyComplete(); } @Test // DATAMONGO-1444 public void findOneByEmptyMonoOfIdShouldReturnEmptyMono() { - - TestSubscriber testSubscriber = TestSubscriber.subscribe(repository.findOne(Mono.empty())); - - testSubscriber.await().assertComplete().assertNoValues().assertNoError(); + StepVerifier.create(repository.findOne(Mono.empty())).verifyComplete(); } @Test // DATAMONGO-1444 public void findAllShouldReturnAllResults() { - - List persons = repository.findAll().collectList().block(); - - assertThat(persons, hasSize(7)); + StepVerifier.create(repository.findAll()).expectNextCount(7).verifyComplete(); } @Test // DATAMONGO-1444 public void findAllByIterableOfIdShouldReturnResults() { - - List persons = repository.findAll(Arrays.asList(dave.id, boyd.id)).collectList().block(); - - assertThat(persons, hasSize(2)); + StepVerifier.create(repository.findAll(Arrays.asList(dave.id, boyd.id))).expectNextCount(2).verifyComplete(); } @Test // DATAMONGO-1444 public void findAllByPublisherOfIdShouldReturnResults() { - - List persons = repository.findAll(Flux.just(dave.id, boyd.id)).collectList().block(); - - assertThat(persons, hasSize(2)); + StepVerifier.create(repository.findAll(Flux.just(dave.id, boyd.id))).expectNextCount(2).verifyComplete(); } @Test // DATAMONGO-1444 public void findAllByEmptyPublisherOfIdShouldReturnResults() { - - TestSubscriber testSubscriber = TestSubscriber.subscribe(repository.findAll(Flux.empty())); - - testSubscriber.await().assertComplete().assertNoValues().assertNoError(); + StepVerifier.create(repository.findAll(Flux.empty())).verifyComplete(); } @Test // DATAMONGO-1444 public void findAllWithSortShouldReturnResults() { - List persons = repository.findAll(new Sort(new Order(Direction.ASC, "age"))).collectList().block(); - - assertThat(persons, hasSize(7)); - assertThat(persons.get(0).getId(), is(equalTo(oliver.getId()))); + StepVerifier.create(repository.findAll(new Sort(new Order(Direction.ASC, "age")))) // + .expectNextCount(7) // + .verifyComplete(); } @Test // DATAMONGO-1444 public void countShouldReturnNumberOfRecords() { - - TestSubscriber testSubscriber = TestSubscriber.subscribe(repository.count()); - - testSubscriber.await().assertComplete().assertValueCount(1).assertValues(7L).assertNoError(); + StepVerifier.create(repository.count()).expectNext(7L).verifyComplete(); } @Test // DATAMONGO-1444 public void insertEntityShouldInsertEntity() { - repository.deleteAll().block(); + StepVerifier.create(repository.deleteAll()).verifyComplete(); ReactivePerson person = new ReactivePerson("Homer", "Simpson", 36); - TestSubscriber testSubscriber = TestSubscriber.subscribe(repository.insert(person)); - - testSubscriber.await().assertComplete().assertValueCount(1).assertValues(person); + StepVerifier.create(repository.insert(person)).expectNext(person).verifyComplete(); assertThat(person.getId(), is(notNullValue())); } @@ -245,16 +198,15 @@ public class SimpleReactiveMongoRepositoryTests implements BeanClassLoaderAware, @Test // DATAMONGO-1444 public void insertIterableOfEntitiesShouldInsertEntity() { - repository.deleteAll().block(); + StepVerifier.create(repository.deleteAll()).verifyComplete(); dave.setId(null); oliver.setId(null); boyd.setId(null); - TestSubscriber testSubscriber = TestSubscriber - .subscribe(repository.insert(Arrays.asList(dave, oliver, boyd))); - - testSubscriber.await().assertComplete().assertValueCount(3).assertValues(dave, oliver, boyd); + StepVerifier.create(repository.insert(Arrays.asList(dave, oliver, boyd))) // + .expectNext(dave, oliver, boyd) // + .verifyComplete(); assertThat(dave.getId(), is(notNullValue())); assertThat(oliver.getId(), is(notNullValue())); @@ -264,16 +216,13 @@ public class SimpleReactiveMongoRepositoryTests implements BeanClassLoaderAware, @Test // DATAMONGO-1444 public void insertPublisherOfEntitiesShouldInsertEntity() { - repository.deleteAll().block(); + StepVerifier.create(repository.deleteAll()).verifyComplete(); dave.setId(null); oliver.setId(null); boyd.setId(null); - TestSubscriber testSubscriber = TestSubscriber - .subscribe(repository.insert(Flux.just(dave, oliver, boyd))); - - testSubscriber.await().assertComplete().assertValueCount(3); + StepVerifier.create(repository.insert(Flux.just(dave, oliver, boyd))).expectNextCount(3).verifyComplete(); assertThat(dave.getId(), is(notNullValue())); assertThat(oliver.getId(), is(notNullValue())); @@ -286,19 +235,15 @@ public class SimpleReactiveMongoRepositoryTests implements BeanClassLoaderAware, dave.setFirstname("Hello, Dave"); dave.setLastname("Bowman"); - TestSubscriber testSubscriber = TestSubscriber.subscribe(repository.save(dave)); + StepVerifier.create(repository.save(dave)).expectNext(dave).verifyComplete(); - testSubscriber.await().assertComplete().assertValueCount(1).assertValues(dave); + StepVerifier.create(repository.findByLastname("Matthews")).expectNext(oliver).verifyComplete(); - List matthews = repository.findByLastname("Matthews").collectList().block(); - assertThat(matthews, hasSize(1)); - assertThat(matthews, contains(oliver)); - assertThat(matthews, not(contains(dave))); + StepVerifier.create(repository.findOne(dave.id)).consumeNextWith(actual -> { - ReactivePerson reactivePerson = repository.findOne(dave.id).block(); - - assertThat(reactivePerson.getFirstname(), is(equalTo(dave.getFirstname()))); - assertThat(reactivePerson.getLastname(), is(equalTo(dave.getLastname()))); + assertThat(actual.getFirstname(), is(equalTo(dave.getFirstname()))); + assertThat(actual.getLastname(), is(equalTo(dave.getLastname()))); + }).verifyComplete(); } @Test // DATAMONGO-1444 @@ -306,29 +251,25 @@ public class SimpleReactiveMongoRepositoryTests implements BeanClassLoaderAware, ReactivePerson person = new ReactivePerson("Homer", "Simpson", 36); - TestSubscriber testSubscriber = TestSubscriber.subscribe(repository.save(person)); + StepVerifier.create(repository.save(person)).expectNext(person).verifyComplete(); - testSubscriber.await().assertComplete().assertValueCount(1).assertValues(person); + StepVerifier.create(repository.findOne(person.id)).consumeNextWith(actual -> { - ReactivePerson reactivePerson = repository.findOne(person.id).block(); - - assertThat(reactivePerson.getFirstname(), is(equalTo(person.getFirstname()))); - assertThat(reactivePerson.getLastname(), is(equalTo(person.getLastname()))); + assertThat(actual.getFirstname(), is(equalTo(person.getFirstname()))); + assertThat(actual.getLastname(), is(equalTo(person.getLastname()))); + }).verifyComplete(); } @Test // DATAMONGO-1444 public void saveIterableOfNewEntitiesShouldInsertEntity() { - repository.deleteAll().block(); + StepVerifier.create(repository.deleteAll()).verifyComplete(); dave.setId(null); oliver.setId(null); boyd.setId(null); - TestSubscriber testSubscriber = TestSubscriber - .subscribe(repository.save(Arrays.asList(dave, oliver, boyd))); - - testSubscriber.await().assertComplete().assertValueCount(3).assertValues(dave, oliver, boyd); + StepVerifier.create(repository.save(Arrays.asList(dave, oliver, boyd))).expectNextCount(3).verifyComplete(); assertThat(dave.getId(), is(notNullValue())); assertThat(oliver.getId(), is(notNullValue())); @@ -343,32 +284,24 @@ public class SimpleReactiveMongoRepositoryTests implements BeanClassLoaderAware, dave.setFirstname("Hello, Dave"); dave.setLastname("Bowman"); - TestSubscriber testSubscriber = TestSubscriber - .subscribe(repository.save(Arrays.asList(person, dave))); + StepVerifier.create(repository.save(Arrays.asList(person, dave))).expectNextCount(2).verifyComplete(); - testSubscriber.await().assertComplete().assertValueCount(2); - - ReactivePerson persistentDave = repository.findOne(dave.id).block(); - assertThat(persistentDave, is(equalTo(dave))); + StepVerifier.create(repository.findOne(dave.id)).expectNext(dave).verifyComplete(); assertThat(person.id, is(notNullValue())); - ReactivePerson persistentHomer = repository.findOne(person.id).block(); - assertThat(persistentHomer, is(equalTo(person))); + StepVerifier.create(repository.findOne(person.id)).expectNext(person).verifyComplete(); } @Test // DATAMONGO-1444 public void savePublisherOfEntitiesShouldInsertEntity() { - repository.deleteAll().block(); + StepVerifier.create(repository.deleteAll()).verifyComplete(); dave.setId(null); oliver.setId(null); boyd.setId(null); - TestSubscriber testSubscriber = TestSubscriber - .subscribe(repository.save(Flux.just(dave, oliver, boyd))); - - testSubscriber.await().assertComplete().assertValueCount(3); + StepVerifier.create(repository.save(Flux.just(dave, oliver, boyd))).expectNextCount(3).verifyComplete(); assertThat(dave.getId(), is(notNullValue())); assertThat(oliver.getId(), is(notNullValue())); @@ -378,67 +311,46 @@ public class SimpleReactiveMongoRepositoryTests implements BeanClassLoaderAware, @Test // DATAMONGO-1444 public void deleteAllShouldRemoveEntities() { - repository.deleteAll().block(); + StepVerifier.create(repository.deleteAll()).verifyComplete(); - TestSubscriber testSubscriber = TestSubscriber.subscribe(repository.findAll()); - - testSubscriber.await().assertComplete().assertValueCount(0); + StepVerifier.create(repository.findAll()).verifyComplete(); } @Test // DATAMONGO-1444 public void deleteByIdShouldRemoveEntity() { - TestSubscriber testSubscriber = TestSubscriber.subscribe(repository.delete(dave.id)); + StepVerifier.create(repository.delete(dave.id)).verifyComplete(); - testSubscriber.await().assertComplete().assertNoValues(); - - TestSubscriber verificationSubscriber = TestSubscriber.subscribe(repository.findOne(dave.id)); - - verificationSubscriber.await().assertComplete().assertNoValues(); + StepVerifier.create(repository.findOne(dave.id)).verifyComplete(); } @Test // DATAMONGO-1444 public void deleteShouldRemoveEntity() { - TestSubscriber testSubscriber = TestSubscriber.subscribe(repository.delete(dave)); + StepVerifier.create(repository.delete(dave)).verifyComplete(); - testSubscriber.await().assertComplete().assertNoValues(); + StepVerifier.create(repository.findOne(dave.id)).verifyComplete(); - TestSubscriber verificationSubscriber = TestSubscriber.subscribe(repository.findOne(dave.id)); - - verificationSubscriber.await().assertComplete().assertNoValues(); } @Test // DATAMONGO-1444 public void deleteIterableOfEntitiesShouldRemoveEntities() { - TestSubscriber testSubscriber = TestSubscriber.subscribe(repository.delete(Arrays.asList(dave, boyd))); + StepVerifier.create(repository.delete(Arrays.asList(dave, boyd))).verifyComplete(); - testSubscriber.await().assertComplete().assertNoValues(); - - TestSubscriber verificationSubscriber = TestSubscriber.subscribe(repository.findOne(boyd.id)); - verificationSubscriber.await().assertComplete().assertNoValues(); - - List matthews = repository.findByLastname("Matthews").collectList().block(); - assertThat(matthews, hasSize(1)); - assertThat(matthews, contains(oliver)); + StepVerifier.create(repository.findOne(boyd.id)).verifyComplete(); + StepVerifier.create(repository.findByLastname("Matthews")).expectNext(oliver).verifyComplete(); } @Test // DATAMONGO-1444 public void deletePublisherOfEntitiesShouldRemoveEntities() { - TestSubscriber testSubscriber = TestSubscriber.subscribe(repository.delete(Flux.just(dave, boyd))); + StepVerifier.create(repository.delete(Flux.just(dave, boyd))).verifyComplete(); - testSubscriber.await().assertComplete().assertNoValues(); - - TestSubscriber verificationSubscriber = TestSubscriber.subscribe(repository.findOne(boyd.id)); - verificationSubscriber.await().assertComplete().assertNoValues(); - - List matthews = repository.findByLastname("Matthews").collectList().block(); - assertThat(matthews, hasSize(1)); - assertThat(matthews, contains(oliver)); + StepVerifier.create(repository.findOne(boyd.id)).verifyComplete(); + StepVerifier.create(repository.findByLastname("Matthews")).expectNext(oliver).verifyComplete(); } interface ReactivePersonRepostitory extends ReactiveMongoRepository { diff --git a/spring-data-mongodb/src/test/java/reactor/test/TestSubscriber.java b/spring-data-mongodb/src/test/java/reactor/test/TestSubscriber.java deleted file mode 100644 index b6ca1cde6..000000000 --- a/spring-data-mongodb/src/test/java/reactor/test/TestSubscriber.java +++ /dev/null @@ -1,1180 +0,0 @@ -/* - * Copyright 2016 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. - * You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - */ - -package reactor.test; - -import java.time.Duration; -import java.util.ArrayList; -import java.util.Arrays; -import java.util.Iterator; -import java.util.LinkedList; -import java.util.List; -import java.util.Objects; -import java.util.Set; -import java.util.concurrent.CountDownLatch; -import java.util.concurrent.TimeUnit; -import java.util.concurrent.atomic.AtomicLongFieldUpdater; -import java.util.concurrent.atomic.AtomicReferenceFieldUpdater; -import java.util.function.BooleanSupplier; -import java.util.function.Consumer; -import java.util.function.Supplier; - -import org.reactivestreams.Publisher; -import org.reactivestreams.Subscriber; -import org.reactivestreams.Subscription; -import reactor.core.Fuseable; -import reactor.core.Receiver; -import reactor.core.Trackable; -import reactor.core.publisher.Operators; - - - -/** - * 
- *  ###############################################################
- *  ###############################################################
- *  ###############################################################
- *
- *  	THIS CODE IS IMPORTED FROM REACTOR-CORE BECAUSE OF
- *  	https://github.com/reactor/reactor-core/issues/135
- *
- *  ###############################################################
- *  ###############################################################
- *  ###############################################################
- *
- * - * - * A Subscriber implementation that hosts assertion tests for its state and allows - * asynchronous cancellation and requesting. - * - *

To create a new instance of {@link TestSubscriber}, you have the choice between - * these static methods: - *

    - *
  • {@link TestSubscriber#subscribe(Publisher)}: create a new {@link TestSubscriber}, - * subscribe to it with the specified {@link Publisher} and requests an unbounded - * number of elements.
    • - *
  • {@link TestSubscriber#subscribe(Publisher, long)}: create a new {@link TestSubscriber}, - * subscribe to it with the specified {@link Publisher} and requests {@code n} elements - * (can be 0 if you want no initial demand). - *
  • {@link TestSubscriber#create()}: create a new {@link TestSubscriber} and requests - * an unbounded number of elements.
    • - *
  • {@link TestSubscriber#create(long)}: create a new {@link TestSubscriber} and - * requests {@code n} elements (can be 0 if you want no initial demand). - *
- * - *

If you are testing asynchronous publishers, don't forget to use one of the - * {@code await*()} methods to wait for the data to assert. - * - *

You can extend this class but only the onNext, onError and onComplete can be overridden. - * You can call {@link #request(long)} and {@link #cancel()} from any thread or from within - * the overridable methods but you should avoid calling the assertXXX methods asynchronously. - * - *

Usage: - * 

- * {@code
- * TestSubscriber
- *   .subscribe(publisher)
- *   .await()
- *   .assertValues("ABC", "DEF");
- * }
- *
- * - * @param the value type. - * - * @author Sebastien Deleuze - * @author David Karnok - * @author Anatoly Kadyshev - * @author Stephane Maldini - * @author Brian Clozel - */ -public class TestSubscriber - implements Subscriber, Subscription, Trackable, Receiver { - - /** - * Default timeout for waiting next values to be received - */ - public static final Duration DEFAULT_VALUES_TIMEOUT = Duration.ofSeconds(3); - - @SuppressWarnings("rawtypes") - private static final AtomicLongFieldUpdater REQUESTED = - AtomicLongFieldUpdater.newUpdater(TestSubscriber.class, "requested"); - - @SuppressWarnings("rawtypes") - private static final AtomicReferenceFieldUpdater NEXT_VALUES = - AtomicReferenceFieldUpdater.newUpdater(TestSubscriber.class, List.class, - "values"); - - @SuppressWarnings("rawtypes") - private static final AtomicReferenceFieldUpdater S = - AtomicReferenceFieldUpdater.newUpdater(TestSubscriber.class, Subscription.class, "s"); - - - private final List errors = new LinkedList<>(); - - private final CountDownLatch cdl = new CountDownLatch(1); - - volatile Subscription s; - - volatile long requested; - - volatile List values = new LinkedList<>(); - - /** - * The fusion mode to request. - */ - private int requestedFusionMode = -1; - - /** - * The established fusion mode. - */ - private volatile int establishedFusionMode = -1; - - /** - * The fuseable QueueSubscription in case a fusion mode was specified. - */ - private Fuseable.QueueSubscription qs; - - private int subscriptionCount = 0; - - private int completionCount = 0; - - private volatile long valueCount = 0L; - - private volatile long nextValueAssertedCount = 0L; - - private Duration valuesTimeout = DEFAULT_VALUES_TIMEOUT; - - private boolean valuesStorage = true; - -// ============================================================================================================== -// Static methods -// ============================================================================================================== - - /** - * Blocking method that waits until {@code conditionSupplier} returns true, or if it - * does not before the specified timeout, throws an {@link AssertionError} with the - * specified error message supplier. - * - * @param timeout the timeout duration - * @param errorMessageSupplier the error message supplier - * @param conditionSupplier condition to break out of the wait loop - * - * @throws AssertionError - */ - public static void await(Duration timeout, Supplier errorMessageSupplier, - BooleanSupplier conditionSupplier) { - - Objects.requireNonNull(errorMessageSupplier); - Objects.requireNonNull(conditionSupplier); - Objects.requireNonNull(timeout); - - long timeoutNs = timeout.toNanos(); - long startTime = System.nanoTime(); - do { - if (conditionSupplier.getAsBoolean()) { - return; - } - try { - Thread.sleep(100); - } - catch (InterruptedException e) { - Thread.currentThread().interrupt(); - throw new RuntimeException(e); - } - } - while (System.nanoTime() - startTime < timeoutNs); - throw new AssertionError(errorMessageSupplier.get()); - } - - /** - * Blocking method that waits until {@code conditionSupplier} returns true, or if it - * does not before the specified timeout, throw an {@link AssertionError} with the - * specified error message. - * - * @param timeout the timeout duration - * @param errorMessage the error message - * @param conditionSupplier condition to break out of the wait loop - * - * @throws AssertionError - */ - public static void await(Duration timeout, - final String errorMessage, - BooleanSupplier conditionSupplier) { - await(timeout, new Supplier() { - @Override - public String get() { - return errorMessage; - } - }, conditionSupplier); - } - - /** - * Create a new {@link TestSubscriber} that requests an unbounded number of elements. - *

Be sure at least a publisher has subscribed to it via {@link Publisher#subscribe(Subscriber)} - * before use assert methods. - * @see #subscribe(Publisher) - * @param the observed value type - * @return a fresh TestSubscriber instance - */ - public static TestSubscriber create() { - return new TestSubscriber<>(); - } - - /** - * Create a new {@link TestSubscriber} that requests initially {@code n} elements. You - * can then manage the demand with {@link Subscription#request(long)}. - *

Be sure at least a publisher has subscribed to it via {@link Publisher#subscribe(Subscriber)} - * before use assert methods. - * @param n Number of elements to request (can be 0 if you want no initial demand). - * @see #subscribe(Publisher, long) - * @param the observed value type - * @return a fresh TestSubscriber instance - */ - public static TestSubscriber create(long n) { - return new TestSubscriber<>(n); - } - - /** - * Create a new {@link TestSubscriber} that requests an unbounded number of elements, - * and make the specified {@code publisher} subscribe to it. - * @param publisher The publisher to subscribe with - * @param the observed value type - * @return a fresh TestSubscriber instance - */ - public static TestSubscriber subscribe(Publisher publisher) { - TestSubscriber subscriber = new TestSubscriber<>(); - publisher.subscribe(subscriber); - return subscriber; - } - - /** - * Create a new {@link TestSubscriber} that requests initially {@code n} elements, - * and make the specified {@code publisher} subscribe to it. You can then manage the - * demand with {@link Subscription#request(long)}. - * @param publisher The publisher to subscribe with - * @param n Number of elements to request (can be 0 if you want no initial demand). - * @param the observed value type - * @return a fresh TestSubscriber instance - */ - public static TestSubscriber subscribe(Publisher publisher, long n) { - TestSubscriber subscriber = new TestSubscriber<>(n); - publisher.subscribe(subscriber); - return subscriber; - } - -// ============================================================================================================== -// Private constructors -// ============================================================================================================== - - private TestSubscriber() { - this(Long.MAX_VALUE); - } - - private TestSubscriber(long n) { - if (n < 0) { - throw new IllegalArgumentException("initialRequest >= required but it was " + n); - } - REQUESTED.lazySet(this, n); - } - -// ============================================================================================================== -// Configuration -// ============================================================================================================== - - - /** - * Enable or disabled the values storage. It is enabled by default, and can be disable - * in order to be able to perform performance benchmarks or tests with a huge amount - * values. - * @param enabled enable value storage? - * @return this - */ - public final TestSubscriber configureValuesStorage(boolean enabled) { - this.valuesStorage = enabled; - return this; - } - - /** - * Configure the timeout in seconds for waiting next values to be received (3 seconds - * by default). - * @param timeout the new default value timeout duration - * @return this - */ - public final TestSubscriber configureValuesTimeout(Duration timeout) { - this.valuesTimeout = timeout; - return this; - } - - /** - * Returns the established fusion mode or -1 if it was not enabled - * - * @return the fusion mode, see Fuseable constants - */ - public final int establishedFusionMode() { - return establishedFusionMode; - } - -// ============================================================================================================== -// Assertions -// ============================================================================================================== - - /** - * Assert a complete successfully signal has been received. - * @return this - */ - public final TestSubscriber assertComplete() { - assertNoError(); - int c = completionCount; - if (c == 0) { - throw new AssertionError("Not completed", null); - } - if (c > 1) { - throw new AssertionError("Multiple completions: " + c, null); - } - return this; - } - - /** - * Assert the specified values have been received. Values storage should be enabled to - * use this method. - * @param expectedValues the values to assert - * @see #configureValuesStorage(boolean) - * @return this - */ - public final TestSubscriber assertContainValues(Set expectedValues) { - if (!valuesStorage) { - throw new IllegalStateException( - "Using assertNoValues() requires enabling values storage"); - } - if (expectedValues.size() > values.size()) { - throw new AssertionError("Actual contains fewer elements" + values, null); - } - - Iterator expected = expectedValues.iterator(); - - for (; ; ) { - boolean n2 = expected.hasNext(); - if (n2) { - T t2 = expected.next(); - if (!values.contains(t2)) { - throw new AssertionError("The element is not contained in the " + - "received resuls" + - " = " + valueAndClass(t2), null); - } - } - else{ - break; - } - } - return this; - } - - /** - * Assert an error signal has been received. - * @return this - */ - public final TestSubscriber assertError() { - assertNotComplete(); - int s = errors.size(); - if (s == 0) { - throw new AssertionError("No error", null); - } - if (s > 1) { - throw new AssertionError("Multiple errors: " + s, null); - } - return this; - } - - /** - * Assert an error signal has been received. - * @param clazz The class of the exception contained in the error signal - * @return this - */ - public final TestSubscriber assertError(Class clazz) { - assertNotComplete(); - int s = errors.size(); - if (s == 0) { - throw new AssertionError("No error", null); - } - if (s == 1) { - Throwable e = errors.get(0); - if (!clazz.isInstance(e)) { - throw new AssertionError("Error class incompatible: expected = " + - clazz + ", actual = " + e, null); - } - } - if (s > 1) { - throw new AssertionError("Multiple errors: " + s, null); - } - return this; - } - - public final TestSubscriber assertErrorMessage(String message) { - assertNotComplete(); - int s = errors.size(); - if (s == 0) { - assertionError("No error", null); - } - if (s == 1) { - if (!Objects.equals(message, - errors.get(0) - .getMessage())) { - assertionError("Error class incompatible: expected = \"" + message + - "\", actual = \"" + errors.get(0).getMessage() + "\"", null); - } - } - if (s > 1) { - assertionError("Multiple errors: " + s, null); - } - - return this; - } - - /** - * Assert an error signal has been received. - * @param expectation A method that can verify the exception contained in the error signal - * and throw an exception (like an {@link AssertionError}) if the exception is not valid. - * @return this - */ - public final TestSubscriber assertErrorWith(Consumer expectation) { - assertNotComplete(); - int s = errors.size(); - if (s == 0) { - throw new AssertionError("No error", null); - } - if (s == 1) { - expectation.accept(errors.get(0)); - } - if (s > 1) { - throw new AssertionError("Multiple errors: " + s, null); - } - return this; - } - - /** - * Assert that the upstream was a Fuseable source. - * - * @return this - */ - public final TestSubscriber assertFuseableSource() { - if (qs == null) { - throw new AssertionError("Upstream was not Fuseable"); - } - return this; - } - - /** - * Assert that the fusion mode was granted. - * - * @return this - */ - public final TestSubscriber assertFusionEnabled() { - if (establishedFusionMode != Fuseable.SYNC && establishedFusionMode != Fuseable.ASYNC) { - throw new AssertionError("Fusion was not enabled"); - } - return this; - } - - public final TestSubscriber assertFusionMode(int expectedMode) { - if (establishedFusionMode != expectedMode) { - throw new AssertionError("Wrong fusion mode: expected: " + fusionModeName( - expectedMode) + ", actual: " + fusionModeName(establishedFusionMode)); - } - return this; - } - - /** - * Assert that the fusion mode was granted. - * - * @return this - */ - public final TestSubscriber assertFusionRejected() { - if (establishedFusionMode != Fuseable.NONE) { - throw new AssertionError("Fusion was granted"); - } - return this; - } - - /** - * Assert no error signal has been received. - * @return this - */ - public final TestSubscriber assertNoError() { - int s = errors.size(); - if (s == 1) { - Throwable e = errors.get(0); - String valueAndClass = e == null ? null : e + " (" + e.getClass().getSimpleName() + ")"; - throw new AssertionError("Error present: " + valueAndClass, null); - } - if (s > 1) { - throw new AssertionError("Multiple errors: " + s, null); - } - return this; - } - - /** - * Assert no values have been received. - * - * @return this - */ - public final TestSubscriber assertNoValues() { - if (valueCount != 0) { - throw new AssertionError("No values expected but received: [length = " + values.size() + "] " + values, - null); - } - return this; - } - - /** - * Assert that the upstream was not a Fuseable source. - * @return this - */ - public final TestSubscriber assertNonFuseableSource() { - if (qs != null) { - throw new AssertionError("Upstream was Fuseable"); - } - return this; - } - - /** - * Assert no complete successfully signal has been received. - * @return this - */ - public final TestSubscriber assertNotComplete() { - int c = completionCount; - if (c == 1) { - throw new AssertionError("Completed", null); - } - if (c > 1) { - throw new AssertionError("Multiple completions: " + c, null); - } - return this; - } - - /** - * Assert no subscription occurred. - * - * @return this - */ - public final TestSubscriber assertNotSubscribed() { - int s = subscriptionCount; - - if (s == 1) { - throw new AssertionError("OnSubscribe called once", null); - } - if (s > 1) { - throw new AssertionError("OnSubscribe called multiple times: " + s, null); - } - - return this; - } - - /** - * Assert no complete successfully or error signal has been received. - * @return this - */ - public final TestSubscriber assertNotTerminated() { - if (cdl.getCount() == 0) { - throw new AssertionError("Terminated", null); - } - return this; - } - - /** - * Assert subscription occurred (once). - * @return this - */ - public final TestSubscriber assertSubscribed() { - int s = subscriptionCount; - - if (s == 0) { - throw new AssertionError("OnSubscribe not called", null); - } - if (s > 1) { - throw new AssertionError("OnSubscribe called multiple times: " + s, null); - } - - return this; - } - - /** - * Assert either complete successfully or error signal has been received. - * @return this - */ - public final TestSubscriber assertTerminated() { - if (cdl.getCount() != 0) { - throw new AssertionError("Not terminated", null); - } - return this; - } - - /** - * Assert {@code n} values has been received. - * - * @param n the expected value count - * - * @return this - */ - public final TestSubscriber assertValueCount(long n) { - if (valueCount != n) { - throw new AssertionError("Different value count: expected = " + n + ", actual = " + valueCount, - null); - } - return this; - } - - /** - * Assert the specified values have been received in the same order read by the - * passed {@link Iterable}. Values storage - * should be enabled to - * use this method. - * @param expectedSequence the values to assert - * @see #configureValuesStorage(boolean) - * @return this - */ - public final TestSubscriber assertValueSequence(Iterable expectedSequence) { - if (!valuesStorage) { - throw new IllegalStateException("Using assertNoValues() requires enabling values storage"); - } - Iterator actual = values.iterator(); - Iterator expected = expectedSequence.iterator(); - int i = 0; - for (; ; ) { - boolean n1 = actual.hasNext(); - boolean n2 = expected.hasNext(); - if (n1 && n2) { - T t1 = actual.next(); - T t2 = expected.next(); - if (!Objects.equals(t1, t2)) { - throw new AssertionError("The element with index " + i + " does not match: expected = " + valueAndClass(t2) + ", actual = " - + valueAndClass( - t1), null); - } - i++; - } else if (n1 && !n2) { - throw new AssertionError("Actual contains more elements" + values, null); - } else if (!n1 && n2) { - throw new AssertionError("Actual contains fewer elements: " + values, null); - } else { - break; - } - } - return this; - } - - /** - * Assert the specified values have been received in the declared order. Values - * storage should be enabled to use this method. - * - * @param expectedValues the values to assert - * - * @return this - * - * @see #configureValuesStorage(boolean) - */ - @SafeVarargs - public final TestSubscriber assertValues(T... expectedValues) { - return assertValueSequence(Arrays.asList(expectedValues)); - } - - /** - * Assert the specified values have been received in the declared order. Values - * storage should be enabled to use this method. - * - * @param expectations One or more methods that can verify the values and throw a - * exception (like an {@link AssertionError}) if the value is not valid. - * - * @return this - * - * @see #configureValuesStorage(boolean) - */ - @SafeVarargs - public final TestSubscriber assertValuesWith(Consumer... expectations) { - if (!valuesStorage) { - throw new IllegalStateException( - "Using assertNoValues() requires enabling values storage"); - } - final int expectedValueCount = expectations.length; - if (expectedValueCount != values.size()) { - throw new AssertionError("Different value count: expected = " + expectedValueCount + ", actual = " + valueCount, null); - } - for (int i = 0; i < expectedValueCount; i++) { - Consumer consumer = expectations[i]; - T actualValue = values.get(i); - consumer.accept(actualValue); - } - return this; - } - -// ============================================================================================================== -// Await methods -// ============================================================================================================== - - /** - * Blocking method that waits until a complete successfully or error signal is received. - * @return this - */ - public final TestSubscriber await() { - if (cdl.getCount() == 0) { - return this; - } - try { - cdl.await(); - } catch (InterruptedException ex) { - throw new AssertionError("Wait interrupted", ex); - } - return this; - } - - /** - * Blocking method that waits until a complete successfully or error signal is received - * or until a timeout occurs. - * @param timeout The timeout value - * @return this - */ - public final TestSubscriber await(Duration timeout) { - if (cdl.getCount() == 0) { - return this; - } - try { - if (!cdl.await(timeout.toMillis(), TimeUnit.MILLISECONDS)) { - throw new AssertionError("No complete or error signal before timeout"); - } - return this; - } - catch (InterruptedException ex) { - throw new AssertionError("Wait interrupted", ex); - } - } - - /** - * Blocking method that waits until {@code n} next values have been received. - * - * @param n the value count to assert - * - * @return this - */ - public final TestSubscriber awaitAndAssertNextValueCount(final long n) { - await(valuesTimeout, () -> { - if(valuesStorage){ - return String.format("%d out of %d next values received within %d, " + - "values : %s", - valueCount - nextValueAssertedCount, - n, - valuesTimeout.toMillis(), - values.toString() - ); - } - return String.format("%d out of %d next values received within %d", - valueCount - nextValueAssertedCount, - n, - valuesTimeout.toMillis()); - }, () -> valueCount >= (nextValueAssertedCount + n)); - nextValueAssertedCount += n; - return this; - } - - /** - * Blocking method that waits until {@code n} next values have been received (n is the - * number of values provided) to assert them. - * - * @param values the values to assert - * - * @return this - */ - @SafeVarargs - @SuppressWarnings("unchecked") - public final TestSubscriber awaitAndAssertNextValues(T... values) { - final int expectedNum = values.length; - final List> expectations = new ArrayList<>(); - for (int i = 0; i < expectedNum; i++) { - final T expectedValue = values[i]; - expectations.add(actualValue -> { - if (!actualValue.equals(expectedValue)) { - throw new AssertionError(String.format( - "Expected Next signal: %s, but got: %s", - expectedValue, - actualValue)); - } - }); - } - awaitAndAssertNextValuesWith(expectations.toArray((Consumer[]) new Consumer[0])); - return this; - } - - /** - * Blocking method that waits until {@code n} next values have been received - * (n is the number of expectations provided) to assert them. - * @param expectations One or more methods that can verify the values and throw a - * exception (like an {@link AssertionError}) if the value is not valid. - * @return this - */ - @SafeVarargs - public final TestSubscriber awaitAndAssertNextValuesWith(Consumer... expectations) { - valuesStorage = true; - final int expectedValueCount = expectations.length; - await(valuesTimeout, () -> { - if(valuesStorage){ - return String.format("%d out of %d next values received within %d, " + - "values : %s", - valueCount - nextValueAssertedCount, - expectedValueCount, - valuesTimeout.toMillis(), - values.toString() - ); - } - return String.format("%d out of %d next values received within %d ms", - valueCount - nextValueAssertedCount, - expectedValueCount, - valuesTimeout.toMillis()); - }, () -> valueCount >= (nextValueAssertedCount + expectedValueCount)); - List nextValuesSnapshot; - List empty = new ArrayList<>(); - for(;;){ - nextValuesSnapshot = values; - if(NEXT_VALUES.compareAndSet(this, values, empty)){ - break; - } - } - if (nextValuesSnapshot.size() < expectedValueCount) { - throw new AssertionError(String.format("Expected %d number of signals but received %d", - expectedValueCount, - nextValuesSnapshot.size())); - } - for (int i = 0; i < expectedValueCount; i++) { - Consumer consumer = expectations[i]; - T actualValue = nextValuesSnapshot.get(i); - consumer.accept(actualValue); - } - nextValueAssertedCount += expectedValueCount; - return this; - } - -// ============================================================================================================== -// Overrides -// ============================================================================================================== - - @Override - public void cancel() { - Subscription a = s; - if (a != Operators.cancelledSubscription()) { - a = S.getAndSet(this, Operators.cancelledSubscription()); - if (a != null && a != Operators.cancelledSubscription()) { - a.cancel(); - } - } - } - - @Override - public final boolean isCancelled() { - return s == Operators.cancelledSubscription(); - } - - @Override - public final boolean isStarted() { - return s != null; - } - - @Override - public final boolean isTerminated() { - return isCancelled(); - } - - @Override - public void onComplete() { - completionCount++; - cdl.countDown(); - } - - @Override - public void onError(Throwable t) { - errors.add(t); - cdl.countDown(); - } - - @Override - public void onNext(T t) { - if (establishedFusionMode == Fuseable.ASYNC) { - for (; ; ) { - t = qs.poll(); - if (t == null) { - break; - } - valueCount++; - if (valuesStorage) { - List nextValuesSnapshot; - for (; ; ) { - nextValuesSnapshot = values; - nextValuesSnapshot.add(t); - if (NEXT_VALUES.compareAndSet(this, - nextValuesSnapshot, - nextValuesSnapshot)) { - break; - } - } - } - } - } - else { - valueCount++; - if (valuesStorage) { - List nextValuesSnapshot; - for (; ; ) { - nextValuesSnapshot = values; - nextValuesSnapshot.add(t); - if (NEXT_VALUES.compareAndSet(this, - nextValuesSnapshot, - nextValuesSnapshot)) { - break; - } - } - } - } - } - - @Override - @SuppressWarnings("unchecked") - public void onSubscribe(Subscription s) { - subscriptionCount++; - int requestMode = requestedFusionMode; - if (requestMode >= 0) { - if (!setWithoutRequesting(s)) { - if (!isCancelled()) { - errors.add(new IllegalStateException("Subscription already set: " + - subscriptionCount)); - } - } else { - if (s instanceof Fuseable.QueueSubscription) { - this.qs = (Fuseable.QueueSubscription)s; - - int m = qs.requestFusion(requestMode); - establishedFusionMode = m; - - if (m == Fuseable.SYNC) { - for (;;) { - T v = qs.poll(); - if (v == null) { - onComplete(); - break; - } - - onNext(v); - } - } - else { - requestDeferred(); - } - } - else { - requestDeferred(); - } - } - } else { - if (!set(s)) { - if (!isCancelled()) { - errors.add(new IllegalStateException("Subscription already set: " + - subscriptionCount)); - } - } - } - } - - @Override - public void request(long n) { - if (Operators.validate(n)) { - if (establishedFusionMode != Fuseable.SYNC) { - normalRequest(n); - } - } - } - - @Override - public final long requestedFromDownstream() { - return requested; - } - - /** - * Setup what fusion mode should be requested from the incomining - * Subscription if it happens to be QueueSubscription - * @param requestMode the mode to request, see Fuseable constants - * @return this - */ - public final TestSubscriber requestedFusionMode(int requestMode) { - this.requestedFusionMode = requestMode; - return this; - } - - @Override - public Subscription upstream() { - return s; - } - - -// ============================================================================================================== -// Non public methods -// ============================================================================================================== - - protected final void normalRequest(long n) { - Subscription a = s; - if (a != null) { - a.request(n); - } else { - Operators.addAndGet(REQUESTED, this, n); - - a = s; - - if (a != null) { - long r = REQUESTED.getAndSet(this, 0L); - - if (r != 0L) { - a.request(r); - } - } - } - } - - /** - * Requests the deferred amount if not zero. - */ - protected final void requestDeferred() { - long r = REQUESTED.getAndSet(this, 0L); - - if (r != 0L) { - s.request(r); - } - } - - /** - * Atomically sets the single subscription and requests the missed amount from it. - * - * @param s - * @return false if this arbiter is cancelled or there was a subscription already set - */ - protected final boolean set(Subscription s) { - Objects.requireNonNull(s, "s"); - Subscription a = this.s; - if (a == Operators.cancelledSubscription()) { - s.cancel(); - return false; - } - if (a != null) { - s.cancel(); - Operators.reportSubscriptionSet(); - return false; - } - - if (S.compareAndSet(this, null, s)) { - - long r = REQUESTED.getAndSet(this, 0L); - - if (r != 0L) { - s.request(r); - } - - return true; - } - - a = this.s; - - if (a != Operators.cancelledSubscription()) { - s.cancel(); - return false; - } - - Operators.reportSubscriptionSet(); - return false; - } - - /** - * Sets the Subscription once but does not request anything. - * @param s the Subscription to set - * @return true if successful, false if the current subscription is not null - */ - protected final boolean setWithoutRequesting(Subscription s) { - Objects.requireNonNull(s, "s"); - for (;;) { - Subscription a = this.s; - if (a == Operators.cancelledSubscription()) { - s.cancel(); - return false; - } - if (a != null) { - s.cancel(); - Operators.reportSubscriptionSet(); - return false; - } - - if (S.compareAndSet(this, null, s)) { - return true; - } - } - } - - /** - * Prepares and throws an AssertionError exception based on the message, cause, the - * active state and the potential errors so far. - * - * @param message the message - * @param cause the optional Throwable cause - * - * @throws AssertionError as expected - */ - protected final void assertionError(String message, Throwable cause) { - StringBuilder b = new StringBuilder(); - - if (cdl.getCount() != 0) { - b.append("(active) "); - } - b.append(message); - - List err = errors; - if (!err.isEmpty()) { - b.append(" (+ ") - .append(err.size()) - .append(" errors)"); - } - AssertionError e = new AssertionError(b.toString(), cause); - - for (Throwable t : err) { - e.addSuppressed(t); - } - - throw e; - } - - protected final String fusionModeName(int mode) { - switch (mode) { - case -1: - return "Disabled"; - case Fuseable.NONE: - return "None"; - case Fuseable.SYNC: - return "Sync"; - case Fuseable.ASYNC: - return "Async"; - default: - return "Unknown(" + mode + ")"; - } - } - - protected final String valueAndClass(Object o) { - if (o == null) { - return null; - } - return o + " (" + o.getClass().getSimpleName() + ")"; - } - -} \ No newline at end of file