From acbb3ba5f93651b9ef55aa41ec1d6bcd57d49fb2 Mon Sep 17 00:00:00 2001 From: Mark Paluch Date: Fri, 24 Mar 2017 14:38:56 +0100 Subject: [PATCH] DATACASS-368 - Migrate reactive tests from TestSubscriber to StepVerifier. Replace TestSubscriber and .block() calls in test with StepVerifier. --- spring-cql/pom.xml | 7 + ...ridgedReactiveSessionIntegrationTests.java | 40 +- .../ReactiveCqlTemplateIntegrationTests.java | 77 +- .../core/ReactiveCqlTemplateUnitTests.java | 210 ++- spring-data-cassandra/pom.xml | 7 + ...tiveCassandraTemplateIntegrationTests.java | 50 +- .../ReactiveCassandraTemplateUnitTests.java | 103 +- ...rtingReactiveCassandraRepositoryTests.java | 125 +- ...veCassandraRepositoryIntegrationTests.java | 50 +- ...veCassandraRepositoryIntegrationTests.java | 183 +-- .../java/reactor/test/TestSubscriber.java | 1129 ----------------- 11 files changed, 353 insertions(+), 1628 deletions(-) delete mode 100644 spring-data-cassandra/src/test/java/reactor/test/TestSubscriber.java diff --git a/spring-cql/pom.xml b/spring-cql/pom.xml index b9602d377..c300b99aa 100644 --- a/spring-cql/pom.xml +++ b/spring-cql/pom.xml @@ -53,6 +53,13 @@ true + + io.projectreactor.addons + reactor-test + ${reactor} + test + + com.datastax.cassandra cassandra-driver-core diff --git a/spring-cql/src/test/java/org/springframework/cassandra/core/DefaultBridgedReactiveSessionIntegrationTests.java b/spring-cql/src/test/java/org/springframework/cassandra/core/DefaultBridgedReactiveSessionIntegrationTests.java index 1e1b39a35..1836ea212 100644 --- a/spring-cql/src/test/java/org/springframework/cassandra/core/DefaultBridgedReactiveSessionIntegrationTests.java +++ b/spring-cql/src/test/java/org/springframework/cassandra/core/DefaultBridgedReactiveSessionIntegrationTests.java @@ -19,6 +19,7 @@ import static org.assertj.core.api.Assertions.*; import reactor.core.publisher.Mono; import reactor.core.scheduler.Schedulers; +import reactor.test.StepVerifier; import org.junit.Before; import org.junit.Test; @@ -27,8 +28,6 @@ import org.springframework.cassandra.core.session.ReactiveResultSet; import org.springframework.cassandra.test.integration.AbstractKeyspaceCreatingIntegrationTest; import com.datastax.driver.core.KeyspaceMetadata; -import com.datastax.driver.core.PreparedStatement; -import com.datastax.driver.core.Row; import com.datastax.driver.core.exceptions.SyntaxError; /** @@ -38,7 +37,7 @@ import com.datastax.driver.core.exceptions.SyntaxError; */ public class DefaultBridgedReactiveSessionIntegrationTests extends AbstractKeyspaceCreatingIntegrationTest { - private DefaultBridgedReactiveSession reactiveSession; + DefaultBridgedReactiveSession reactiveSession; @Before public void before() throws Exception { @@ -59,18 +58,17 @@ public class DefaultBridgedReactiveSessionIntegrationTests extends AbstractKeysp assertThat(keyspace.getTable("users")).isNull(); - ReactiveResultSet resultSet = execution.block(); + StepVerifier.create(execution).consumeNextWith(actual -> { + + assertThat(actual.wasApplied()).isTrue(); + }).verifyComplete(); - assertThat(resultSet.wasApplied()).isTrue(); assertThat(keyspace.getTable("users")).isNotNull(); } - @Test(expected = SyntaxError.class) // DATACASS-335 - public void executeShouldTransportExceptionsInMono() throws Exception { - - Mono execution = reactiveSession.execute("INSERT INTO dummy;"); - - execution.block(); + @Test // DATACASS-335 + public void executeShouldTransportExceptionsInMono() { + StepVerifier.create(reactiveSession.execute("INSERT INTO dummy;")).expectError(SyntaxError.class).verify(); } @Test // DATACASS-335 @@ -79,12 +77,13 @@ public class DefaultBridgedReactiveSessionIntegrationTests extends AbstractKeysp session.execute("CREATE TABLE users (\n" + " userid text PRIMARY KEY,\n" + " first_name text\n" + ");"); session.execute("INSERT INTO users (userid, first_name) VALUES ('White', 'Walter');"); - Mono execution = reactiveSession.execute("SELECT * FROM users;"); - ReactiveResultSet resultSet = execution.block(); - Row row = resultSet.rows().blockFirst(); + StepVerifier.create(reactiveSession.execute("SELECT * FROM users;")).consumeNextWith(actual -> { - assertThat(row).isNotNull(); - assertThat(row.getString("userid")).isEqualTo("White"); + StepVerifier.create(actual.rows()).consumeNextWith(row -> { + + assertThat(row.getString("userid")).isEqualTo("White"); + }).verifyComplete(); + }).verifyComplete(); } @Test // DATACASS-335 @@ -92,12 +91,11 @@ public class DefaultBridgedReactiveSessionIntegrationTests extends AbstractKeysp session.execute("CREATE TABLE users (\n" + " userid text PRIMARY KEY,\n" + " first_name text\n" + ");"); - Mono execution = reactiveSession - .prepare("INSERT INTO users (userid, first_name) VALUES (?, ?);"); - PreparedStatement preparedStatement = execution.block(); + StepVerifier.create(reactiveSession.prepare("INSERT INTO users (userid, first_name) VALUES (?, ?);")) + .consumeNextWith(actual -> { - assertThat(preparedStatement).isNotNull(); - assertThat(preparedStatement.getQueryString()).isEqualTo("INSERT INTO users (userid, first_name) VALUES (?, ?);"); + assertThat(actual.getQueryString()).isEqualTo("INSERT INTO users (userid, first_name) VALUES (?, ?);"); + }).verifyComplete(); } private KeyspaceMetadata getKeyspaceMetadata() { diff --git a/spring-cql/src/test/java/org/springframework/cassandra/core/ReactiveCqlTemplateIntegrationTests.java b/spring-cql/src/test/java/org/springframework/cassandra/core/ReactiveCqlTemplateIntegrationTests.java index e2f13978e..b19ebded0 100644 --- a/spring-cql/src/test/java/org/springframework/cassandra/core/ReactiveCqlTemplateIntegrationTests.java +++ b/spring-cql/src/test/java/org/springframework/cassandra/core/ReactiveCqlTemplateIntegrationTests.java @@ -18,8 +18,8 @@ package org.springframework.cassandra.core; import static org.assertj.core.api.Assertions.*; import reactor.core.scheduler.Schedulers; +import reactor.test.StepVerifier; -import java.util.Map; import java.util.concurrent.atomic.AtomicBoolean; import org.junit.Before; @@ -39,11 +39,12 @@ import com.datastax.driver.core.querybuilder.QueryBuilder; public class ReactiveCqlTemplateIntegrationTests extends AbstractKeyspaceCreatingIntegrationTest { private static final AtomicBoolean initialized = new AtomicBoolean(); - private ReactiveSession reactiveSession; - private ReactiveCqlTemplate template; + + ReactiveSession reactiveSession; + ReactiveCqlTemplate template; @Before - public void before() throws Exception { + public void before() { reactiveSession = new DefaultBridgedReactiveSession(getSession(), Schedulers.elastic()); @@ -59,74 +60,88 @@ public class ReactiveCqlTemplateIntegrationTests extends AbstractKeyspaceCreatin } @Test // DATACASS-335 - public void executeShouldRemoveRecords() throws Exception { + public void executeShouldRemoveRecords() { - template.execute("DELETE FROM user WHERE id = 'WHITE'").block(); + StepVerifier.create(template.execute("DELETE FROM user WHERE id = 'WHITE'")).expectNext(true).verifyComplete(); assertThat(getSession().execute("SELECT * FROM user").one()).isNull(); } @Test // DATACASS-335 - public void queryForObjectShouldReturnFirstColumn() throws Exception { + public void queryForObjectShouldReturnFirstColumn() { - String id = template.queryForObject("SELECT id FROM user;", String.class).block(); - - assertThat(id).isEqualTo("WHITE"); + StepVerifier.create(template.queryForObject("SELECT id FROM user;", String.class)) // + .expectNext("WHITE") // + .verifyComplete(); } @Test // DATACASS-335 - public void queryForObjectShouldReturnMap() throws Exception { + public void queryForObjectShouldReturnMap() { - Map map = template.queryForMap("SELECT * FROM user;").block(); + StepVerifier.create(template.queryForMap("SELECT * FROM user;")) // + .consumeNextWith(actual -> { - assertThat(map).containsEntry("id", "WHITE").containsEntry("username", "Walter"); + assertThat(actual).containsEntry("id", "WHITE").containsEntry("username", "Walter"); + }).verifyComplete(); } @Test // DATACASS-335 - public void executeStatementShouldRemoveRecords() throws Exception { + public void executeStatementShouldRemoveRecords() { - template.execute(QueryBuilder.delete().from("user").where(QueryBuilder.eq("id", "WHITE"))).block(); + StepVerifier + .create(template.execute(QueryBuilder.delete() // + .from("user") // + .where(QueryBuilder.eq("id", "WHITE")))) // + .expectNext(true) // + .verifyComplete(); assertThat(getSession().execute("SELECT * FROM user").one()).isNull(); } @Test // DATACASS-335 - public void queryForObjectStatementShouldReturnFirstColumn() throws Exception { + public void queryForObjectStatementShouldReturnFirstColumn() { - String id = template.queryForObject(QueryBuilder.select("id").from("user"), String.class).block(); - - assertThat(id).isEqualTo("WHITE"); + StepVerifier + .create(template.queryForObject(QueryBuilder // + .select("id") // + .from("user"), String.class)) // + .expectNext("WHITE") // + .verifyComplete(); } @Test // DATACASS-335 - public void queryForObjectStatementShouldReturnMap() throws Exception { + public void queryForObjectStatementShouldReturnMap() { - Map map = template.queryForMap(QueryBuilder.select().from("user")).block(); + StepVerifier.create(template.queryForMap(QueryBuilder.select().from("user"))) // + .consumeNextWith(actual -> { - assertThat(map).containsEntry("id", "WHITE").containsEntry("username", "Walter"); + assertThat(actual).containsEntry("id", "WHITE").containsEntry("username", "Walter"); + }).verifyComplete(); } @Test // DATACASS-335 - public void executeWithArgsShouldRemoveRecords() throws Exception { + public void executeWithArgsShouldRemoveRecords() { - template.execute("DELETE FROM user WHERE id = ?", "WHITE").block(); + StepVerifier.create(template.execute("DELETE FROM user WHERE id = ?", "WHITE")).expectNext(true).verifyComplete(); assertThat(getSession().execute("SELECT * FROM user").one()).isNull(); } @Test // DATACASS-335 - public void queryForObjectWithArgsShouldReturnFirstColumn() throws Exception { + public void queryForObjectWithArgsShouldReturnFirstColumn() { - String id = template.queryForObject("SELECT id FROM user WHERE id = ?;", String.class, "WHITE").block(); - - assertThat(id).isEqualTo("WHITE"); + StepVerifier.create(template.queryForObject("SELECT id FROM user WHERE id = ?;", String.class, "WHITE")) // + .expectNext("WHITE") // + .verifyComplete(); } @Test // DATACASS-335 - public void queryForObjectWithArgsShouldReturnMap() throws Exception { + public void queryForObjectWithArgsShouldReturnMap() { - Map map = template.queryForMap("SELECT * FROM user WHERE id = ?;", "WHITE").block(); + StepVerifier.create(template.queryForMap("SELECT * FROM user WHERE id = ?;", "WHITE")) // + .consumeNextWith(actual -> { - assertThat(map).containsEntry("id", "WHITE").containsEntry("username", "Walter"); + assertThat(actual).containsEntry("id", "WHITE").containsEntry("username", "Walter"); + }).verifyComplete(); } } diff --git a/spring-cql/src/test/java/org/springframework/cassandra/core/ReactiveCqlTemplateUnitTests.java b/spring-cql/src/test/java/org/springframework/cassandra/core/ReactiveCqlTemplateUnitTests.java index c7ced72ec..035f5fdc3 100644 --- a/spring-cql/src/test/java/org/springframework/cassandra/core/ReactiveCqlTemplateUnitTests.java +++ b/spring-cql/src/test/java/org/springframework/cassandra/core/ReactiveCqlTemplateUnitTests.java @@ -22,9 +22,9 @@ import static org.mockito.Mockito.*; import reactor.core.publisher.Flux; import reactor.core.publisher.Mono; +import reactor.test.StepVerifier; import java.util.Collections; -import java.util.List; import java.util.function.Consumer; import org.junit.Before; @@ -90,7 +90,8 @@ public class ReactiveCqlTemplateUnitTests { }); verify(session, never()).close(); - assertThat(flux.blockLast()).isEqualTo("OK"); + + StepVerifier.create(flux).expectNext("OK").verifyComplete(); verify(session).close(); } @@ -101,13 +102,7 @@ public class ReactiveCqlTemplateUnitTests { throw new InvalidQueryException("wrong query"); }); - try { - flux.blockLast(); - - fail("Missing CassandraInvalidQueryException"); - } catch (CassandraInvalidQueryException e) { - assertThat(e).hasMessageContaining("wrong query"); - } + StepVerifier.create(flux).expectError(CassandraInvalidQueryException.class).verify(); } @Test // DATACASS-335 @@ -118,7 +113,9 @@ public class ReactiveCqlTemplateUnitTests { Mono mono = template.execute("UPDATE user SET a = 'b';"); verifyZeroInteractions(session); - assertThat(mono.block()).isFalse(); + + StepVerifier.create(mono).expectNext(false).verifyComplete(); + verify(session).execute(any(Statement.class)); } @@ -129,13 +126,7 @@ public class ReactiveCqlTemplateUnitTests { Mono mono = template.execute("UPDATE user SET a = 'b';"); - try { - mono.block(); - - fail("Missing CassandraConnectionFailureException"); - } catch (CassandraConnectionFailureException e) { - assertThat(e).hasMessageContaining("tried for query failed"); - } + StepVerifier.create(mono).expectError(CassandraConnectionFailureException.class).verify(); } // ------------------------------------------------------------------------- @@ -147,7 +138,7 @@ public class ReactiveCqlTemplateUnitTests { doTestStrings(null, null, null, reactiveCqlTemplate -> { - reactiveCqlTemplate.execute("SELECT * from USERS").block(); + StepVerifier.create(reactiveCqlTemplate.execute("SELECT * from USERS")).expectNextCount(1).verifyComplete(); verify(session).execute(any(Statement.class)); }); @@ -158,7 +149,9 @@ public class ReactiveCqlTemplateUnitTests { doTestStrings(5, ConsistencyLevel.ONE, DowngradingConsistencyRetryPolicy.INSTANCE, reactiveCqlTemplate -> { - reactiveCqlTemplate.execute("SELECT * from USERS").block(); + StepVerifier.create(reactiveCqlTemplate.execute("SELECT * from USERS")) // + .expectNextCount(1) // + .verifyComplete(); verify(session).execute(any(Statement.class)); }); @@ -171,9 +164,8 @@ public class ReactiveCqlTemplateUnitTests { Mono mono = reactiveCqlTemplate.queryForResultSet("SELECT * from USERS"); - List rows = mono.block().rows().collectList().block(); + StepVerifier.create(mono.flatMap(ReactiveResultSet::rows)).expectNextCount(3).verifyComplete(); - assertThat(rows).hasSize(3); verify(session).execute(any(Statement.class)); }); } @@ -185,9 +177,8 @@ public class ReactiveCqlTemplateUnitTests { Flux flux = reactiveCqlTemplate.query("SELECT * from USERS", (row, index) -> row.getString(0)); - List rows = flux.collectList().block(); + StepVerifier.create(flux).expectNext("Walter", "Hank", " Jesse").verifyComplete(); - assertThat(rows).hasSize(3).contains("Walter", "Hank", " Jesse"); verify(session).execute(any(Statement.class)); }); } @@ -199,9 +190,8 @@ public class ReactiveCqlTemplateUnitTests { Flux flux = reactiveCqlTemplate.query("SELECT * from USERS", (row, index) -> row.getString(0)); - List rows = flux.collectList().block(); + StepVerifier.create(flux).expectNext("Walter", "Hank", " Jesse").verifyComplete(); - assertThat(rows).hasSize(3).contains("Walter", "Hank", " Jesse"); verify(session).execute(any(Statement.class)); }); } @@ -215,7 +205,9 @@ public class ReactiveCqlTemplateUnitTests { Flux flux = template.query("UPDATE user SET a = 'b';", resultSet -> Mono.just(resultSet.wasApplied())); verifyZeroInteractions(session); - assertThat(flux.collectList().block()).hasSize(1).contains(true); + + StepVerifier.create(flux).expectNext(true).verifyComplete(); + verify(session).execute(any(Statement.class)); } @@ -226,13 +218,7 @@ public class ReactiveCqlTemplateUnitTests { Flux flux = template.query("UPDATE user SET a = 'b';", resultSet -> Mono.just(resultSet.wasApplied())); - try { - flux.blockLast(); - - fail("Missing CassandraConnectionFailureException"); - } catch (CassandraConnectionFailureException e) { - assertThat(e).hasMessageContaining("tried for query failed"); - } + StepVerifier.create(flux).expectError(CassandraConnectionFailureException.class).verify(); } @Test // DATACASS-335 @@ -242,7 +228,8 @@ public class ReactiveCqlTemplateUnitTests { when(reactiveResultSet.rows()).thenReturn(Flux.empty()); Mono mono = template.queryForObject("SELECT * FROM user", (row, rowNum) -> "OK"); - assertThat(mono.hasElement().block()).isFalse(); + + StepVerifier.create(mono).verifyComplete(); } @Test // DATACASS-335 @@ -252,7 +239,8 @@ public class ReactiveCqlTemplateUnitTests { when(reactiveResultSet.rows()).thenReturn(Flux.just(row)); Mono mono = template.queryForObject("SELECT * FROM user", (row, rowNum) -> "OK"); - assertThat(mono.block()).isEqualTo("OK"); + + StepVerifier.create(mono).expectNext("OK").verifyComplete(); } @Test // DATACASS-335 @@ -262,7 +250,8 @@ public class ReactiveCqlTemplateUnitTests { when(reactiveResultSet.rows()).thenReturn(Flux.just(row)); Mono mono = template.queryForObject("SELECT * FROM user", (row, rowNum) -> null); - assertThat(mono.hasElement().block()).isFalse(); + + StepVerifier.create(mono).verifyComplete(); } @Test // DATACASS-335 @@ -273,13 +262,7 @@ public class ReactiveCqlTemplateUnitTests { Mono mono = template.queryForObject("SELECT * FROM user", (row, rowNum) -> "OK"); - try { - mono.block(); - - fail("Missing IncorrectResultSizeDataAccessException"); - } catch (IncorrectResultSizeDataAccessException e) { - assertThat(e).hasMessageContaining("expected 1, actual 2"); - } + StepVerifier.create(mono).expectError(IncorrectResultSizeDataAccessException.class).verify(); } @Test // DATACASS-335 @@ -293,7 +276,7 @@ public class ReactiveCqlTemplateUnitTests { Mono mono = template.queryForObject("SELECT * FROM user", String.class); - assertThat(mono.block()).isEqualTo("OK"); + StepVerifier.create(mono).expectNext("OK").verifyComplete(); } @Test // DATACASS-335 @@ -307,7 +290,7 @@ public class ReactiveCqlTemplateUnitTests { Flux flux = template.queryForFlux("SELECT * FROM user", String.class); - assertThat(flux.collectList().block()).contains("OK", "NOT OK"); + StepVerifier.create(flux).expectNext("OK", "NOT OK").verifyComplete(); } @Test // DATACASS-335 @@ -318,7 +301,7 @@ public class ReactiveCqlTemplateUnitTests { Flux flux = template.queryForRows("SELECT * FROM user"); - assertThat(flux.collectList().block()).hasSize(2).contains(row); + StepVerifier.create(flux).expectNext(row, row).verifyComplete(); } @Test // DATACASS-335 @@ -329,7 +312,7 @@ public class ReactiveCqlTemplateUnitTests { Mono mono = template.execute("UPDATE user SET a = 'b';"); - assertThat(mono.block()).isTrue(); + StepVerifier.create(mono).expectNext(true).verifyComplete(); } @Test // DATACASS-335 @@ -341,7 +324,9 @@ public class ReactiveCqlTemplateUnitTests { Flux flux = template.execute(Flux.just("UPDATE user SET a = 'b';", "UPDATE user SET x = 'y';")); verifyZeroInteractions(session); - assertThat(flux.collectList().block()).hasSize(2).contains(true, false); + + StepVerifier.create(flux).expectNext(true).expectNext(false).verifyComplete(); + verify(session, times(2)).execute(any(Statement.class)); } @@ -354,7 +339,9 @@ public class ReactiveCqlTemplateUnitTests { doTestStrings(null, null, null, reactiveCqlTemplate -> { - reactiveCqlTemplate.execute(new SimpleStatement("SELECT * from USERS")).block(); + StepVerifier.create(reactiveCqlTemplate.execute(new SimpleStatement("SELECT * from USERS"))) // + .expectNextCount(1) // + .verifyComplete(); verify(session).execute(any(Statement.class)); }); @@ -365,7 +352,9 @@ public class ReactiveCqlTemplateUnitTests { doTestStrings(5, ConsistencyLevel.ONE, DowngradingConsistencyRetryPolicy.INSTANCE, reactiveCqlTemplate -> { - reactiveCqlTemplate.execute(new SimpleStatement("SELECT * from USERS")).block(); + StepVerifier.create(reactiveCqlTemplate.execute(new SimpleStatement("SELECT * from USERS"))) // + .expectNextCount(1) // + .verifyComplete(); verify(session).execute(any(Statement.class)); }); @@ -376,11 +365,12 @@ public class ReactiveCqlTemplateUnitTests { doTestStrings(null, null, null, reactiveCqlTemplate -> { - Mono mono = reactiveCqlTemplate.queryForResultSet(new SimpleStatement("SELECT * from USERS")); + StepVerifier + .create(reactiveCqlTemplate.queryForResultSet(new SimpleStatement("SELECT * from USERS")) + .flatMap(ReactiveResultSet::rows)) // + .expectNextCount(3) // + .verifyComplete(); - List rows = mono.block().rows().collectList().block(); - - assertThat(rows).hasSize(3); verify(session).execute(any(Statement.class)); }); } @@ -393,9 +383,8 @@ public class ReactiveCqlTemplateUnitTests { Flux flux = reactiveCqlTemplate.query(new SimpleStatement("SELECT * from USERS"), (row, index) -> row.getString(0)); - List rows = flux.collectList().block(); + StepVerifier.create(flux).expectNext("Walter", "Hank", " Jesse").verifyComplete(); - assertThat(rows).hasSize(3).contains("Walter", "Hank", " Jesse"); verify(session).execute(any(Statement.class)); }); } @@ -408,9 +397,11 @@ public class ReactiveCqlTemplateUnitTests { Flux flux = reactiveCqlTemplate.query(new SimpleStatement("SELECT * from USERS"), (row, index) -> row.getString(0)); - List rows = flux.collectList().block(); + StepVerifier.create(flux.collectList()).consumeNextWith(rows -> { + + assertThat(rows).hasSize(3).contains("Walter", "Hank", " Jesse"); + }).verifyComplete(); - assertThat(rows).hasSize(3).contains("Walter", "Hank", " Jesse"); verify(session).execute(any(Statement.class)); }); } @@ -425,7 +416,7 @@ public class ReactiveCqlTemplateUnitTests { resultSet -> Mono.just(resultSet.wasApplied())); verifyZeroInteractions(session); - assertThat(flux.collectList().block()).hasSize(1).contains(true); + StepVerifier.create(flux).expectNext(true).verifyComplete(); verify(session).execute(any(Statement.class)); } @@ -437,13 +428,7 @@ public class ReactiveCqlTemplateUnitTests { Flux flux = template.query(new SimpleStatement("UPDATE user SET a = 'b';"), resultSet -> Mono.just(resultSet.wasApplied())); - try { - flux.blockLast(); - - fail("Missing CassandraConnectionFailureException"); - } catch (CassandraConnectionFailureException e) { - assertThat(e).hasMessageContaining("tried for query failed"); - } + StepVerifier.create(flux).expectError(CassandraConnectionFailureException.class).verify(); } @Test // DATACASS-335 @@ -453,7 +438,8 @@ public class ReactiveCqlTemplateUnitTests { when(reactiveResultSet.rows()).thenReturn(Flux.empty()); Mono mono = template.queryForObject(new SimpleStatement("SELECT * FROM user"), (row, rowNum) -> "OK"); - assertThat(mono.hasElement().block()).isFalse(); + + StepVerifier.create(mono).verifyComplete(); } @Test // DATACASS-335 @@ -463,7 +449,8 @@ public class ReactiveCqlTemplateUnitTests { when(reactiveResultSet.rows()).thenReturn(Flux.just(row)); Mono mono = template.queryForObject(new SimpleStatement("SELECT * FROM user"), (row, rowNum) -> "OK"); - assertThat(mono.block()).isEqualTo("OK"); + + StepVerifier.create(mono).expectNext("OK").verifyComplete(); } @Test // DATACASS-335 @@ -473,7 +460,8 @@ public class ReactiveCqlTemplateUnitTests { when(reactiveResultSet.rows()).thenReturn(Flux.just(row)); Mono mono = template.queryForObject(new SimpleStatement("SELECT * FROM user"), (row, rowNum) -> null); - assertThat(mono.hasElement().block()).isFalse(); + + StepVerifier.create(mono).verifyComplete(); } @Test // DATACASS-335 @@ -484,13 +472,7 @@ public class ReactiveCqlTemplateUnitTests { Mono mono = template.queryForObject(new SimpleStatement("SELECT * FROM user"), (row, rowNum) -> "OK"); - try { - mono.block(); - - fail("Missing IncorrectResultSizeDataAccessException"); - } catch (IncorrectResultSizeDataAccessException e) { - assertThat(e).hasMessageContaining("expected 1, actual 2"); - } + StepVerifier.create(mono).expectError(IncorrectResultSizeDataAccessException.class).verify(); } @Test // DATACASS-335 @@ -504,7 +486,7 @@ public class ReactiveCqlTemplateUnitTests { Mono mono = template.queryForObject(new SimpleStatement("SELECT * FROM user"), String.class); - assertThat(mono.block()).isEqualTo("OK"); + StepVerifier.create(mono).expectNext("OK").verifyComplete(); } @Test // DATACASS-335 @@ -518,7 +500,7 @@ public class ReactiveCqlTemplateUnitTests { Flux flux = template.queryForFlux(new SimpleStatement("SELECT * FROM user"), String.class); - assertThat(flux.collectList().block()).contains("OK", "NOT OK"); + StepVerifier.create(flux).expectNext("OK", "NOT OK").verifyComplete(); } @Test // DATACASS-335 @@ -529,7 +511,7 @@ public class ReactiveCqlTemplateUnitTests { Flux flux = template.queryForRows(new SimpleStatement("SELECT * FROM user")); - assertThat(flux.collectList().block()).hasSize(2).contains(row); + StepVerifier.create(flux).expectNext(row, row).verifyComplete(); } @Test // DATACASS-335 @@ -538,9 +520,8 @@ public class ReactiveCqlTemplateUnitTests { when(session.execute(any(Statement.class))).thenReturn(Mono.just(reactiveResultSet)); when(reactiveResultSet.wasApplied()).thenReturn(true); - Mono mono = template.execute(new SimpleStatement("UPDATE user SET a = 'b';")); - - assertThat(mono.block()).isTrue(); + StepVerifier.create(template.execute(new SimpleStatement("UPDATE user SET a = 'b';"))).expectNext(true) + .verifyComplete(); } // ------------------------------------------------------------------------- @@ -557,9 +538,7 @@ public class ReactiveCqlTemplateUnitTests { return session.execute(ps.bind("A")).flatMap(ReactiveResultSet::rows); }); - List rows = flux.collectList().block(); - - assertThat(rows).hasSize(3); + StepVerifier.create(flux).expectNextCount(3).verifyComplete(); }); } @@ -572,7 +551,7 @@ public class ReactiveCqlTemplateUnitTests { when(this.preparedStatement.bind("White")).thenReturn(this.boundStatement); when(this.reactiveResultSet.wasApplied()).thenReturn(true); - assertThat(applied.block()).isTrue(); + StepVerifier.create(applied).expectNext(true).verifyComplete(); }); } @@ -587,7 +566,9 @@ public class ReactiveCqlTemplateUnitTests { (session, ps) -> session.execute(ps.bind())); verifyZeroInteractions(session); - assertThat(flux.collectList().block()).hasSize(1).contains(reactiveResultSet); + + StepVerifier.create(flux).expectNext(reactiveResultSet).verifyComplete(); + verify(session).prepare(anyString()); verify(session).execute(boundStatement); } @@ -601,7 +582,9 @@ public class ReactiveCqlTemplateUnitTests { (session, ps) -> session.execute(boundStatement)); verifyZeroInteractions(session); - assertThat(flux.collectList().block()).hasSize(1).contains(reactiveResultSet); + + StepVerifier.create(flux).expectNext(reactiveResultSet).verifyComplete(); + verify(session).execute(boundStatement); } @@ -612,13 +595,7 @@ public class ReactiveCqlTemplateUnitTests { throw new NoHostAvailableException(Collections.emptyMap()); }, (session, ps) -> session.execute(boundStatement)); - try { - flux.blockLast(); - - fail("Missing CassandraConnectionFailureException"); - } catch (CassandraConnectionFailureException e) { - assertThat(e).hasMessageContaining("tried for query"); - } + StepVerifier.create(flux).expectError(CassandraConnectionFailureException.class).verify(); } @Test // DATACASS-335 @@ -628,13 +605,7 @@ public class ReactiveCqlTemplateUnitTests { throw new NoHostAvailableException(Collections.emptyMap()); }); - try { - flux.blockLast(); - - fail("Missing CassandraConnectionFailureException"); - } catch (CassandraConnectionFailureException e) { - assertThat(e).hasMessageContaining("tried for query"); - } + StepVerifier.create(flux).expectError(CassandraConnectionFailureException.class).verify(); } @Test // DATACASS-335 @@ -647,7 +618,8 @@ public class ReactiveCqlTemplateUnitTests { Flux flux = template.query(session -> Mono.just(preparedStatement), ReactiveResultSet::rows); verifyZeroInteractions(session); - assertThat(flux.collectList().block()).hasSize(1).contains(row); + + StepVerifier.create(flux).expectNext(row).verifyComplete(); verify(preparedStatement).bind(); } @@ -663,7 +635,9 @@ public class ReactiveCqlTemplateUnitTests { }, ReactiveResultSet::rows); verifyZeroInteractions(session); - assertThat(flux.collectList().block()).hasSize(1).contains(row); + + StepVerifier.create(flux).expectNext(row).verifyComplete(); + verify(preparedStatement).bind("a", "b"); } @@ -679,7 +653,9 @@ public class ReactiveCqlTemplateUnitTests { }, (row, rowNum) -> row); verifyZeroInteractions(session); - assertThat(flux.collectList().block()).hasSize(1).contains(row); + + StepVerifier.create(flux).expectNext(row).verifyComplete(); + verify(preparedStatement).bind("a", "b"); } @@ -693,7 +669,8 @@ public class ReactiveCqlTemplateUnitTests { Mono mono = template.queryForObject("SELECT * FROM user WHERE username = ?", (row, rowNum) -> "OK", "Walter"); - assertThat(mono.hasElement().block()).isFalse(); + + StepVerifier.create(mono).verifyComplete(); } @Test // DATACASS-335 @@ -706,7 +683,8 @@ public class ReactiveCqlTemplateUnitTests { Mono mono = template.queryForObject("SELECT * FROM user WHERE username = ?", (row, rowNum) -> "OK", "Walter"); - assertThat(mono.block()).isEqualTo("OK"); + + StepVerifier.create(mono).expectNext("OK").verifyComplete(); } @Test // DATACASS-335 @@ -719,13 +697,8 @@ public class ReactiveCqlTemplateUnitTests { Mono mono = template.queryForObject("SELECT * FROM user WHERE username = ?", (row, rowNum) -> "OK", "Walter"); - try { - mono.block(); - fail("Missing IncorrectResultSizeDataAccessException"); - } catch (IncorrectResultSizeDataAccessException e) { - assertThat(e).hasMessageContaining("expected 1, actual 2"); - } + StepVerifier.create(mono).expectError(IncorrectResultSizeDataAccessException.class).verify(); } @Test // DATACASS-335 @@ -741,7 +714,7 @@ public class ReactiveCqlTemplateUnitTests { Mono mono = template.queryForObject("SELECT * FROM user WHERE username = ?", String.class, "Walter"); - assertThat(mono.block()).isEqualTo("OK"); + StepVerifier.create(mono).expectNext("OK").verifyComplete(); } @Test // DATACASS-335 @@ -757,7 +730,7 @@ public class ReactiveCqlTemplateUnitTests { Flux flux = template.queryForFlux("SELECT * FROM user WHERE username = ?", String.class, "Walter"); - assertThat(flux.collectList().block()).contains("OK", "NOT OK"); + StepVerifier.create(flux).expectNext("OK", "NOT OK").verifyComplete(); } @Test // DATACASS-335 @@ -770,7 +743,7 @@ public class ReactiveCqlTemplateUnitTests { Flux flux = template.queryForRows("SELECT * FROM user WHERE username = ?", "Walter"); - assertThat(flux.collectList().block()).hasSize(2).contains(row); + StepVerifier.create(flux).expectNextCount(2).verifyComplete(); } @Test // DATACASS-335 @@ -783,7 +756,7 @@ public class ReactiveCqlTemplateUnitTests { Mono mono = template.execute("UPDATE user SET username = ?", "Walter"); - assertThat(mono.block()).isTrue(); + StepVerifier.create(mono).expectNext(true).verifyComplete(); } @Test // DATACASS-335 @@ -798,7 +771,8 @@ public class ReactiveCqlTemplateUnitTests { Flux flux = template.execute("UPDATE user SET username = ?", Flux.just(new Object[] { "Walter" }, new Object[] { "Hank" })); - assertThat(flux.collectList().block()).hasSize(2).contains(true); + StepVerifier.create(flux).expectNext(true, true).verifyComplete(); + verify(session, atMost(1)).prepare("UPDATE user SET username = ?"); verify(session, times(2)).execute(boundStatement); } diff --git a/spring-data-cassandra/pom.xml b/spring-data-cassandra/pom.xml index 3b8b7182a..11ff6906a 100644 --- a/spring-data-cassandra/pom.xml +++ b/spring-data-cassandra/pom.xml @@ -79,6 +79,13 @@ true + + io.projectreactor.addons + reactor-test + ${reactor} + test + + io.reactivex rxjava diff --git a/spring-data-cassandra/src/test/java/org/springframework/data/cassandra/core/ReactiveCassandraTemplateIntegrationTests.java b/spring-data-cassandra/src/test/java/org/springframework/data/cassandra/core/ReactiveCassandraTemplateIntegrationTests.java index d91f5d78e..5591c53ee 100644 --- a/spring-data-cassandra/src/test/java/org/springframework/data/cassandra/core/ReactiveCassandraTemplateIntegrationTests.java +++ b/spring-data-cassandra/src/test/java/org/springframework/data/cassandra/core/ReactiveCassandraTemplateIntegrationTests.java @@ -15,10 +15,9 @@ */ package org.springframework.data.cassandra.core; -import static org.assertj.core.api.Assertions.*; - import reactor.core.publisher.Mono; import reactor.core.scheduler.Schedulers; +import reactor.test.StepVerifier; import org.junit.Before; import org.junit.Test; @@ -36,7 +35,7 @@ import org.springframework.data.cassandra.test.integration.support.SchemaTestUti */ public class ReactiveCassandraTemplateIntegrationTests extends AbstractKeyspaceCreatingIntegrationTest { - private ReactiveCassandraTemplate template; + ReactiveCassandraTemplate template; @Before public void setUp() throws Exception { @@ -57,14 +56,11 @@ public class ReactiveCassandraTemplateIntegrationTests extends AbstractKeyspaceC Person person = new Person("heisenberg", "Walter", "White"); Mono insert = template.insert(person); - Mono oneById = template.selectOneById(person.getId(), Person.class); + StepVerifier.create(template.selectOneById(person.getId(), Person.class)).verifyComplete(); - assertThat(oneById.hasElement().block()).isFalse(); + StepVerifier.create(insert).expectNext(person).verifyComplete(); - Person saved = insert.block(); - - assertThat(saved).isNotNull().isEqualTo(person); - assertThat(oneById.block()).isNotNull().isEqualTo(saved); + StepVerifier.create(template.selectOneById(person.getId(), Person.class)).expectNext(person).verifyComplete(); } @Test // DATACASS-335 @@ -72,11 +68,9 @@ public class ReactiveCassandraTemplateIntegrationTests extends AbstractKeyspaceC Person person = new Person("heisenberg", "Walter", "White"); - template.insert(person).block(); + StepVerifier.create(template.insert(person)).expectNextCount(1).verifyComplete(); - Mono count = template.count(Person.class); - - assertThat(count.block()).isEqualTo(1L); + StepVerifier.create(template.count(Person.class)).expectNext(1L).verifyComplete(); } @Test // DATACASS-335 @@ -84,17 +78,13 @@ public class ReactiveCassandraTemplateIntegrationTests extends AbstractKeyspaceC Person person = new Person("heisenberg", "Walter", "White"); - template.insert(person).block(); + StepVerifier.create(template.insert(person)).expectNextCount(1).verifyComplete(); person.setFirstname("Walter Hartwell"); - Person updated = template.update(person).block(); + StepVerifier.create(template.insert(person)).expectNextCount(1).verifyComplete(); - assertThat(updated).isNotNull(); - - Mono oneById = template.selectOneById(person.getId(), Person.class); - - assertThat(oneById.block()).isEqualTo(person); + StepVerifier.create(template.selectOneById(person.getId(), Person.class)).expectNext(person).verifyComplete(); } @Test // DATACASS-335 @@ -102,15 +92,11 @@ public class ReactiveCassandraTemplateIntegrationTests extends AbstractKeyspaceC Person person = new Person("heisenberg", "Walter", "White"); - template.insert(person).block(); + StepVerifier.create(template.insert(person)).expectNextCount(1).verifyComplete(); - Person deleted = template.delete(person).block(); + StepVerifier.create(template.delete(person)).expectNext(person).verifyComplete(); - assertThat(deleted).isNotNull(); - - Mono oneById = template.selectOneById(person.getId(), Person.class); - - assertThat(oneById.block()).isNull(); + StepVerifier.create(template.selectOneById(person.getId(), Person.class)).verifyComplete(); } @Test // DATACASS-335 @@ -118,14 +104,10 @@ public class ReactiveCassandraTemplateIntegrationTests extends AbstractKeyspaceC Person person = new Person("heisenberg", "Walter", "White"); - template.insert(person).block(); + StepVerifier.create(template.insert(person)).expectNextCount(1).verifyComplete(); - Boolean deleted = template.deleteById(person.getId(), Person.class).block(); + StepVerifier.create(template.deleteById(person.getId(), Person.class)).expectNext(true).verifyComplete(); - assertThat(deleted).isTrue(); - - Mono oneById = template.selectOneById(person.getId(), Person.class); - - assertThat(oneById.block()).isNull(); + StepVerifier.create(template.selectOneById(person.getId(), Person.class)).verifyComplete(); } } diff --git a/spring-data-cassandra/src/test/java/org/springframework/data/cassandra/core/ReactiveCassandraTemplateUnitTests.java b/spring-data-cassandra/src/test/java/org/springframework/data/cassandra/core/ReactiveCassandraTemplateUnitTests.java index b05eceb52..86021502e 100644 --- a/spring-data-cassandra/src/test/java/org/springframework/data/cassandra/core/ReactiveCassandraTemplateUnitTests.java +++ b/spring-data-cassandra/src/test/java/org/springframework/data/cassandra/core/ReactiveCassandraTemplateUnitTests.java @@ -23,6 +23,7 @@ import static org.mockito.Mockito.*; import reactor.core.publisher.Flux; import reactor.core.publisher.Mono; +import reactor.test.StepVerifier; import java.util.Collections; @@ -35,7 +36,6 @@ import org.mockito.Mock; import org.mockito.junit.MockitoJUnitRunner; import org.springframework.cassandra.core.session.ReactiveResultSet; import org.springframework.cassandra.core.session.ReactiveSession; -import org.springframework.cassandra.support.exception.CassandraConnectionFailureException; import org.springframework.data.cassandra.domain.Person; import com.datastax.driver.core.ColumnDefinitions; @@ -85,9 +85,10 @@ public class ReactiveCassandraTemplateUnitTests { when(row.getObject(1)).thenReturn("Walter"); when(row.getObject(2)).thenReturn("White"); - Flux flux = template.select("SELECT * FROM person", Person.class); + StepVerifier.create(template.select("SELECT * FROM person", Person.class)) // + .expectNext(new Person("myid", "Walter", "White")) // + .verifyComplete(); - assertThat(flux.collectList().block()).hasSize(1).contains(new Person("myid", "Walter", "White")); verify(session).execute(statementCaptor.capture()); assertThat(statementCaptor.getValue().toString()).isEqualTo("SELECT * FROM person"); } @@ -97,15 +98,10 @@ public class ReactiveCassandraTemplateUnitTests { when(reactiveResultSet.rows()).thenThrow(new NoHostAvailableException(Collections.emptyMap())); - Flux flux = template.select("SELECT * FROM person", Person.class); - - try { - flux.last().block(); - - fail("Missing CassandraConnectionFailureException"); - } catch (CassandraConnectionFailureException e) { - assertThat(e).hasRootCauseInstanceOf(NoHostAvailableException.class); - } + StepVerifier.create(template.select("SELECT * FROM person", Person.class)) // + .consumeErrorWith(e -> { + assertThat(e).hasRootCauseInstanceOf(NoHostAvailableException.class); + }).verify(); } @Test // DATACASS-335 @@ -123,9 +119,10 @@ public class ReactiveCassandraTemplateUnitTests { when(row.getObject(1)).thenReturn("Walter"); when(row.getObject(2)).thenReturn("White"); - Mono mono = template.selectOneById("myid", Person.class); + StepVerifier.create(template.selectOneById("myid", Person.class)) // + .expectNext(new Person("myid", "Walter", "White")) // + .verifyComplete(); - assertThat(mono.block()).isEqualTo(new Person("myid", "Walter", "White")); verify(session).execute(statementCaptor.capture()); assertThat(statementCaptor.getValue().toString()).isEqualTo("SELECT * FROM person WHERE id='myid';"); } @@ -135,9 +132,8 @@ public class ReactiveCassandraTemplateUnitTests { when(reactiveResultSet.rows()).thenReturn(Flux.just(row)); - Mono mono = template.exists("myid", Person.class); + StepVerifier.create(template.exists("myid", Person.class)).expectNext(true).verifyComplete(); - assertThat(mono.block()).isTrue(); verify(session).execute(statementCaptor.capture()); assertThat(statementCaptor.getValue().toString()).isEqualTo("SELECT * FROM person WHERE id='myid';"); } @@ -147,9 +143,8 @@ public class ReactiveCassandraTemplateUnitTests { when(reactiveResultSet.rows()).thenReturn(Flux.empty()); - Mono mono = template.exists("myid", Person.class); + StepVerifier.create(template.exists("myid", Person.class)).expectNext(false).verifyComplete(); - assertThat(mono.block()).isFalse(); verify(session).execute(statementCaptor.capture()); assertThat(statementCaptor.getValue().toString()).isEqualTo("SELECT * FROM person WHERE id='myid';"); } @@ -161,9 +156,8 @@ public class ReactiveCassandraTemplateUnitTests { when(row.getLong(0)).thenReturn(42L); when(columnDefinitions.size()).thenReturn(1); - Mono mono = template.count(Person.class); + StepVerifier.create(template.count(Person.class)).expectNext(42L).verifyComplete(); - assertThat(mono.block()).isEqualTo(42L); verify(session).execute(statementCaptor.capture()); assertThat(statementCaptor.getValue().toString()).isEqualTo("SELECT count(*) FROM person;"); } @@ -174,9 +168,8 @@ public class ReactiveCassandraTemplateUnitTests { when(reactiveResultSet.wasApplied()).thenReturn(true); Person person = new Person("heisenberg", "Walter", "White"); - Mono mono = template.insert(person); + StepVerifier.create(template.insert(person)).expectNext(person).verifyComplete(); - assertThat(mono.block()).isEqualTo(person); verify(session).execute(statementCaptor.capture()); assertThat(statementCaptor.getValue().toString()) .isEqualTo("INSERT INTO person (firstname,id,lastname) VALUES ('Walter','heisenberg','White');"); @@ -189,15 +182,11 @@ public class ReactiveCassandraTemplateUnitTests { when(session.execute(any(Statement.class))) .thenReturn(Mono.error(new NoHostAvailableException(Collections.emptyMap()))); - Mono mono = template.insert(new Person("heisenberg", "Walter", "White")); + StepVerifier.create(template.insert(new Person("heisenberg", "Walter", "White"))) // + .consumeErrorWith(e -> { - try { - mono.block(); - - fail("Missing CassandraConnectionFailureException"); - } catch (CassandraConnectionFailureException e) { - assertThat(e).hasRootCauseInstanceOf(NoHostAvailableException.class); - } + assertThat(e).hasRootCauseInstanceOf(NoHostAvailableException.class); + }).verify(); } @Test // DATACASS-335 @@ -206,9 +195,8 @@ public class ReactiveCassandraTemplateUnitTests { when(reactiveResultSet.wasApplied()).thenReturn(false); Person person = new Person("heisenberg", "Walter", "White"); - Mono mono = template.insert(person); - assertThat(mono.block()).isNull(); + StepVerifier.create(template.insert(person)).verifyComplete(); } @Test // DATACASS-335 @@ -217,41 +205,22 @@ public class ReactiveCassandraTemplateUnitTests { when(reactiveResultSet.wasApplied()).thenReturn(true); Person person = new Person("heisenberg", "Walter", "White"); - Mono mono = template.update(person); - assertThat(mono.block()).isEqualTo(person); + StepVerifier.create(template.update(person)).expectNext(person).verifyComplete(); + verify(session).execute(statementCaptor.capture()); assertThat(statementCaptor.getValue().toString()) .isEqualTo("UPDATE person SET firstname='Walter',lastname='White' WHERE id='heisenberg';"); } - @Test // DATACASS-335 - public void updateShouldTranslateException() { - - reset(session); - when(session.execute(any(Statement.class))) - .thenReturn(Mono.error(new NoHostAvailableException(Collections.emptyMap()))); - - Mono mono = template.update(new Person("heisenberg", "Walter", "White")); - - try { - mono.block(); - - fail("Missing CassandraConnectionFailureException"); - } catch (CassandraConnectionFailureException e) { - assertThat(e).hasRootCauseInstanceOf(NoHostAvailableException.class); - } - } - @Test // DATACASS-335 public void updateShouldNotApplyUpdate() { when(reactiveResultSet.wasApplied()).thenReturn(false); Person person = new Person("heisenberg", "Walter", "White"); - Mono mono = template.update(person); - assertThat(mono.block()).isNull(); + StepVerifier.create(template.update(person)).verifyComplete(); } @Test // DATACASS-335 @@ -261,46 +230,26 @@ public class ReactiveCassandraTemplateUnitTests { Person person = new Person("heisenberg", "Walter", "White"); - Mono mono = template.delete(person); + StepVerifier.create(template.delete(person)).expectNext(person).verifyComplete(); - assertThat(mono.block()).isEqualTo(person); verify(session).execute(statementCaptor.capture()); assertThat(statementCaptor.getValue().toString()).isEqualTo("DELETE FROM person WHERE id='heisenberg';"); } - @Test // DATACASS-335 - public void deleteShouldTranslateException() { - - reset(session); - when(session.execute(any(Statement.class))) - .thenReturn(Mono.error(new NoHostAvailableException(Collections.emptyMap()))); - - Mono mono = template.delete(new Person("heisenberg", "Walter", "White")); - - try { - mono.block(); - - fail("Missing CassandraConnectionFailureException"); - } catch (CassandraConnectionFailureException e) { - assertThat(e).hasRootCauseInstanceOf(NoHostAvailableException.class); - } - } - @Test // DATACASS-335 public void deleteShouldNotApplyRemoval() { when(reactiveResultSet.wasApplied()).thenReturn(false); Person person = new Person("heisenberg", "Walter", "White"); - Mono mono = template.delete(person); - assertThat(mono.block()).isNull(); + StepVerifier.create(template.delete(person)).verifyComplete(); } @Test // DATACASS-335 public void truncateShouldRemoveEntities() { - template.truncate(Person.class).block(); + StepVerifier.create(template.truncate(Person.class)).verifyComplete(); verify(session).execute(statementCaptor.capture()); assertThat(statementCaptor.getValue().toString()).isEqualTo("TRUNCATE person;"); diff --git a/spring-data-cassandra/src/test/java/org/springframework/data/cassandra/repository/ConvertingReactiveCassandraRepositoryTests.java b/spring-data-cassandra/src/test/java/org/springframework/data/cassandra/repository/ConvertingReactiveCassandraRepositoryTests.java index f1bf52c22..48a8399d4 100644 --- a/spring-data-cassandra/src/test/java/org/springframework/data/cassandra/repository/ConvertingReactiveCassandraRepositoryTests.java +++ b/spring-data-cassandra/src/test/java/org/springframework/data/cassandra/repository/ConvertingReactiveCassandraRepositoryTests.java @@ -19,11 +19,12 @@ import static org.assertj.core.api.Assertions.*; import reactor.core.publisher.Flux; import reactor.core.publisher.Mono; -import reactor.test.TestSubscriber; +import reactor.test.StepVerifier; import rx.Observable; import rx.Single; import java.util.Arrays; +import java.util.List; import org.junit.Before; import org.junit.Test; @@ -33,7 +34,6 @@ import org.springframework.beans.factory.annotation.Autowired; import org.springframework.cassandra.test.integration.AbstractKeyspaceCreatingIntegrationTest; import org.springframework.context.annotation.ComponentScan.Filter; import org.springframework.context.annotation.Configuration; -import org.springframework.data.cassandra.core.ReactiveCassandraTemplate; import org.springframework.data.cassandra.domain.Person; import org.springframework.data.cassandra.repository.config.EnableReactiveCassandraRepositories; import org.springframework.data.cassandra.test.integration.support.IntegrationTestConfig; @@ -68,7 +68,6 @@ public class ConvertingReactiveCassandraRepositoryTests extends AbstractKeyspace } @Autowired Session session; - @Autowired ReactiveCassandraTemplate template; @Autowired MixedPersonRepostitory reactiveRepository; @Autowired PersonRepostitory reactivePersonRepostitory; @Autowired RxJava1PersonRepostitory rxJava1PersonRepostitory; @@ -86,130 +85,114 @@ public class ConvertingReactiveCassandraRepositoryTests extends AbstractKeyspace Thread.sleep(500); } - reactiveRepository.deleteAll().block(); + StepVerifier.create(reactiveRepository.deleteAll()).verifyComplete(); dave = new Person("42", "Dave", "Matthews"); oliver = new Person("4", "Oliver August", "Matthews"); carter = new Person("49", "Carter", "Beauford"); boyd = new Person("45", "Boyd", "Tinsley"); - TestSubscriber subscriber = TestSubscriber.create(); - - reactiveRepository.save(Arrays.asList(oliver, dave, carter, boyd)).subscribe(subscriber); - - subscriber.await().assertComplete().assertNoError(); + StepVerifier.create(reactiveRepository.save(Arrays.asList(oliver, dave, carter, boyd))).expectNextCount(4) + .verifyComplete(); } @Test // DATACASS-335 - public void reactiveStreamsMethodsShouldWork() throws InterruptedException { - - TestSubscriber subscriber = TestSubscriber.subscribe(reactivePersonRepostitory.exists(dave.getId())); - - subscriber.awaitAndAssertNextValueCount(1).assertNoError().assertValues(true); + public void reactiveStreamsMethodsShouldWork() { + StepVerifier.create(reactivePersonRepostitory.exists(dave.getId())).expectNext(true).verifyComplete(); } @Test // DATACASS-335 public void reactiveStreamsQueryMethodsShouldWork() { - - TestSubscriber subscriber = TestSubscriber - .subscribe(reactivePersonRepostitory.findByLastname(boyd.getLastname())); - - subscriber.awaitAndAssertNextValueCount(1).assertValues(boyd); + StepVerifier.create(reactivePersonRepostitory.findByLastname(boyd.getLastname())).expectNext(boyd).verifyComplete(); } @Test // DATACASS-360 public void dtoProjectionShouldWork() { - TestSubscriber subscriber = TestSubscriber - .subscribe(reactivePersonRepostitory.findProjectedByLastname(boyd.getLastname())); + StepVerifier.create(reactivePersonRepostitory.findProjectedByLastname(boyd.getLastname())) + .consumeNextWith(actual -> { - subscriber.awaitAndAssertNextValueCount(1).assertValuesWith(personDto -> { - assertThat(personDto.firstname).isEqualTo(boyd.getFirstname()); - assertThat(personDto.lastname).isEqualTo(boyd.getLastname()); - }); + assertThat(actual.firstname).isEqualTo(boyd.getFirstname()); + assertThat(actual.lastname).isEqualTo(boyd.getLastname()); + }).verifyComplete(); } @Test // DATACASS-335 public void simpleRxJavaMethodsShouldWork() { - - rx.observers.TestSubscriber subscriber = new rx.observers.TestSubscriber<>(); - - rxJava1PersonRepostitory.exists(dave.getId()).subscribe(subscriber); - - subscriber.awaitTerminalEvent(); - subscriber.assertCompleted(); - subscriber.assertNoErrors(); - subscriber.assertValue(true); + rxJava1PersonRepostitory.exists(dave.getId()) // + .test() // + .awaitTerminalEvent() // + .assertResult(true) // + .assertCompleted() // + .assertNoErrors(); } @Test // DATACASS-335 public void existsWithSingleRxJavaIdMethodsShouldWork() { - rx.observers.TestSubscriber subscriber = new rx.observers.TestSubscriber<>(); - - rxJava1PersonRepostitory.exists(Single.just(dave.getId())).subscribe(subscriber); - - subscriber.awaitTerminalEvent(); - subscriber.assertCompleted(); - subscriber.assertNoErrors(); - subscriber.assertValue(true); + rxJava1PersonRepostitory.exists(Single.just(dave.getId())) // + .test() // + .awaitTerminalEvent() // + .assertResult(true) // + .assertCompleted() // + .assertNoErrors(); } @Test // DATACASS-335 public void singleRxJavaQueryMethodShouldWork() { - rx.observers.TestSubscriber subscriber = new rx.observers.TestSubscriber<>(); - - rxJava1PersonRepostitory.findManyByLastname(dave.getLastname()).subscribe(subscriber); - - subscriber.awaitTerminalEvent(); - subscriber.assertNoErrors(); - subscriber.assertCompleted(); - subscriber.assertValueCount(2); + rxJava1PersonRepostitory.findManyByLastname(dave.getLastname()) // + .test() // + .awaitTerminalEvent() // + .assertValueCount(2) // + .assertNoErrors() // + .assertCompleted(); } @Test // DATACASS-335 public void singleProjectedRxJavaQueryMethodShouldWork() { - rx.observers.TestSubscriber subscriber = new rx.observers.TestSubscriber<>(); + List values = rxJava1PersonRepostitory.findProjectedByLastname(carter.getLastname()) // + .test() // + .awaitTerminalEvent() // + .assertValueCount(1) // + .assertCompleted() // + .assertNoErrors() // + .getOnNextEvents(); - rxJava1PersonRepostitory.findProjectedByLastname(carter.getLastname()).subscribe(subscriber); - - subscriber.awaitTerminalEvent(); - subscriber.assertCompleted(); - subscriber.assertNoErrors(); - - ProjectedPerson projectedPerson = subscriber.getOnNextEvents().get(0); + ProjectedPerson projectedPerson = values.get(0); assertThat(projectedPerson.getFirstname()).isEqualTo(carter.getFirstname()); } @Test // DATACASS-335 public void observableRxJavaQueryMethodShouldWork() { - rx.observers.TestSubscriber subscriber = new rx.observers.TestSubscriber<>(); - - rxJava1PersonRepostitory.findByLastname(boyd.getLastname()).subscribe(subscriber); - - subscriber.awaitTerminalEvent(); - subscriber.assertCompleted(); - subscriber.assertNoErrors(); - subscriber.assertValue(boyd); + rxJava1PersonRepostitory.findByLastname(boyd.getLastname()) // + .test() // + .awaitTerminalEvent() // + .assertValue(boyd) // + .assertNoErrors() // + .assertCompleted(); } @Test // DATACASS-335 public void mixedRepositoryShouldWork() { - Person value = reactiveRepository.findByLastname(boyd.getLastname()).toBlocking().value(); - - assertThat(value).isEqualTo(boyd); + reactiveRepository.findByLastname(boyd.getLastname()) // + .test() // + .awaitTerminalEvent() // + .assertValue(boyd) // + .assertCompleted() // + .assertNoErrors(); } @Test // DATACASS-335 public void shouldFindOneByPublisherOfLastName() { - Person carter = reactiveRepository.findByLastname(Single.just(this.carter.getLastname())).block(); + StepVerifier.create(reactiveRepository.findByLastname(Single.just(this.carter.getLastname()))) // + .expectNext(carter) // + .verifyComplete(); - assertThat(carter.getFirstname()).isEqualTo(this.carter.getFirstname()); } @Repository diff --git a/spring-data-cassandra/src/test/java/org/springframework/data/cassandra/repository/ReactiveCassandraRepositoryIntegrationTests.java b/spring-data-cassandra/src/test/java/org/springframework/data/cassandra/repository/ReactiveCassandraRepositoryIntegrationTests.java index f26f432d4..ca4bf6404 100644 --- a/spring-data-cassandra/src/test/java/org/springframework/data/cassandra/repository/ReactiveCassandraRepositoryIntegrationTests.java +++ b/spring-data-cassandra/src/test/java/org/springframework/data/cassandra/repository/ReactiveCassandraRepositoryIntegrationTests.java @@ -15,13 +15,11 @@ */ package org.springframework.data.cassandra.repository; -import static org.assertj.core.api.Assertions.*; - import reactor.core.publisher.Flux; import reactor.core.publisher.Mono; +import reactor.test.StepVerifier; import java.util.Arrays; -import java.util.List; import org.junit.Before; import org.junit.Test; @@ -111,47 +109,34 @@ public class ReactiveCassandraRepositoryIntegrationTests extends AbstractKeyspac repository = factory.getRepository(PersonRepository.class); groupRepostitory = factory.getRepository(GroupRepository.class); - repository.deleteAll().block(); - groupRepostitory.deleteAll().block(); + StepVerifier.create(repository.deleteAll().concatWith(groupRepostitory.deleteAll())).verifyComplete(); dave = new Person("42", "Dave", "Matthews"); oliver = new Person("4", "Oliver August", "Matthews"); carter = new Person("49", "Carter", "Beauford"); boyd = new Person("45", "Boyd", "Tinsley"); - repository.save(Arrays.asList(oliver, dave, carter, boyd)).last().block(); + StepVerifier.create(repository.save(Arrays.asList(oliver, dave, carter, boyd))).expectNextCount(4).verifyComplete(); } @Test // DATACASS-335 public void shouldFindByLastName() { - - List list = repository.findByLastname("Matthews").collectList().block(); - - assertThat(list).hasSize(2).contains(dave, oliver); + StepVerifier.create(repository.findByLastname(dave.getLastname())).expectNextCount(2).verifyComplete(); } @Test // DATACASS-335 public void shouldFindOneByLastName() { - - Person carter = repository.findOneByLastname("Beauford").block(); - - assertThat(carter.getFirstname()).isEqualTo("Carter"); + StepVerifier.create(repository.findOneByLastname(carter.getLastname())).expectNext(carter).verifyComplete(); } @Test // DATACASS-335 public void shouldFindOneByPublisherOfLastName() { - - Person carter = repository.findByLastname(Mono.just("Beauford")).block(); - - assertThat(carter.getFirstname()).isEqualTo("Carter"); + StepVerifier.create(repository.findByLastname(Mono.just(carter.getLastname()))).expectNext(carter).verifyComplete(); } @Test // DATACASS-335 public void shouldFindUsingPublishersInStringQuery() { - - List persons = repository.findStringQuery(Mono.just("Matthews")).collectList().block(); - - assertThat(persons).contains(dave); + StepVerifier.create(repository.findStringQuery(Mono.just(dave.getLastname()))).expectNextCount(2).verifyComplete(); } @Test // DATACASS-335 @@ -160,17 +145,20 @@ public class ReactiveCassandraRepositoryIntegrationTests extends AbstractKeyspac GroupKey key1 = new GroupKey("Simpsons", "hash", "Bart"); GroupKey key2 = new GroupKey("Simpsons", "hash", "Homer"); - groupRepostitory.save(Flux.just(new Group(key1), new Group(key2))).blockLast(); + StepVerifier.create(groupRepostitory.save(Flux.just(new Group(key1), new Group(key2)))).expectNextCount(2) + .verifyComplete(); - List persons = groupRepostitory - .findByIdGroupnameAndIdHashPrefix("Simpsons", "hash", new Sort(Direction.ASC, "id.username")).collectList() - .block(); - assertThat(persons).containsSequence(new Group(key1), new Group(key2)); + StepVerifier + .create(groupRepostitory.findByIdGroupnameAndIdHashPrefix("Simpsons", "hash", + new Sort(Direction.ASC, "id.username"))) // + .expectNext(new Group(key1), new Group(key2)) // + .verifyComplete(); - List reversed = groupRepostitory - .findByIdGroupnameAndIdHashPrefix("Simpsons", "hash", new Sort(Direction.DESC, "id.username")).collectList() - .block(); - assertThat(reversed).containsSequence(new Group(key2), new Group(key1)); + StepVerifier + .create(groupRepostitory.findByIdGroupnameAndIdHashPrefix("Simpsons", "hash", + new Sort(Direction.DESC, "id.username"))) // + .expectNext(new Group(key2), new Group(key1)) // + .verifyComplete(); } interface PersonRepository extends ReactiveCassandraRepository { diff --git a/spring-data-cassandra/src/test/java/org/springframework/data/cassandra/repository/support/SimpleReactiveCassandraRepositoryIntegrationTests.java b/spring-data-cassandra/src/test/java/org/springframework/data/cassandra/repository/support/SimpleReactiveCassandraRepositoryIntegrationTests.java index d5c408ea7..1b979f9c3 100644 --- a/spring-data-cassandra/src/test/java/org/springframework/data/cassandra/repository/support/SimpleReactiveCassandraRepositoryIntegrationTests.java +++ b/spring-data-cassandra/src/test/java/org/springframework/data/cassandra/repository/support/SimpleReactiveCassandraRepositoryIntegrationTests.java @@ -19,10 +19,9 @@ import static org.assertj.core.api.Assertions.*; import reactor.core.publisher.Flux; import reactor.core.publisher.Mono; -import reactor.test.TestSubscriber; +import reactor.test.StepVerifier; import java.util.Arrays; -import java.util.List; import org.junit.Before; import org.junit.Test; @@ -91,166 +90,149 @@ public class SimpleReactiveCassandraRepositoryIntegrationTests extends AbstractK repository = factory.getRepository(PersonRepostitory.class); - repository.deleteAll().block(); + deleteAll(); dave = new Person("42", "Dave", "Matthews"); oliver = new Person("4", "Oliver August", "Matthews"); carter = new Person("49", "Carter", "Beauford"); boyd = new Person("45", "Boyd", "Tinsley"); + } - repository.save(Arrays.asList(oliver, dave, carter, boyd)).last().block(); + private void insertTestData() { + StepVerifier.create(repository.save(Arrays.asList(oliver, dave, carter, boyd))).expectNextCount(4).verifyComplete(); + } + + private void deleteAll() { + StepVerifier.create(repository.deleteAll()).verifyComplete(); } @Test // DATACASS-335 public void existsByIdShouldReturnTrueForExistingObject() { - Boolean exists = repository.exists(dave.getId()).block(); + insertTestData(); - assertThat(exists).isTrue(); + StepVerifier.create(repository.exists(dave.getId())).expectNext(true).verifyComplete(); } @Test // DATACASS-335 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 // DATACASS-335 public void existsByMonoOfIdShouldReturnTrueForExistingObject() { - Boolean exists = repository.exists(Mono.just(dave.getId())).block(); - assertThat(exists).isTrue(); + insertTestData(); + + StepVerifier.create(repository.exists(Mono.just(dave.getId()))).expectNext(true).verifyComplete(); } @Test // DATACASS-335 public void existsByEmptyMonoOfIdShouldReturnEmptyMono() { - - TestSubscriber testSubscriber = TestSubscriber.subscribe(repository.exists(Mono.empty())); - - testSubscriber.await().assertComplete().assertNoValues().assertNoError(); + StepVerifier.create(repository.exists(Mono.empty())).verifyComplete(); } @Test // DATACASS-335 public void findOneShouldReturnObject() { - Person person = repository.findOne(dave.getId()).block(); + insertTestData(); - assertThat(person).isEqualTo(dave); + StepVerifier.create(repository.findOne(dave.getId())).expectNext(dave).verifyComplete(); } @Test // DATACASS-335 public void findOneShouldCompleteWithoutValueForAbsentObject() { - - TestSubscriber testSubscriber = TestSubscriber.subscribe(repository.findOne("unknown")); - - testSubscriber.await().assertComplete().assertNoValues().assertNoError(); + StepVerifier.create(repository.findOne("unknown")).verifyComplete(); } @Test // DATACASS-335 public void findOneByMonoOfIdShouldReturnTrueForExistingObject() { - Person person = repository.findOne(Mono.just(dave.getId())).block(); + insertTestData(); - assertThat(person).isEqualTo(dave); + StepVerifier.create(repository.findOne(Mono.just(dave.getId()))).expectNext(dave).verifyComplete(); } @Test // DATACASS-335 public void findOneByEmptyMonoOfIdShouldReturnEmptyMono() { - - TestSubscriber testSubscriber = TestSubscriber.subscribe(repository.findOne(Mono.empty())); - - testSubscriber.await().assertComplete().assertNoValues().assertNoError(); + StepVerifier.create(repository.findOne(Mono.empty())).verifyComplete(); } @Test // DATACASS-335 public void findAllShouldReturnAllResults() { - List persons = repository.findAll().collectList().block(); + insertTestData(); - assertThat(persons).hasSize(4); + StepVerifier.create(repository.findAll()).expectNextCount(4).verifyComplete(); } @Test // DATACASS-335 public void findAllByIterableOfIdShouldReturnResults() { - List persons = repository.findAll(Arrays.asList(dave.getId(), boyd.getId())).collectList().block(); + insertTestData(); - assertThat(persons).hasSize(2); + StepVerifier.create(repository.findAll(Arrays.asList(dave.getId(), boyd.getId()))) // + .expectNextCount(2) // + .verifyComplete(); } @Test // DATACASS-335 public void findAllByPublisherOfIdShouldReturnResults() { - List persons = repository.findAll(Flux.just(dave.getId(), boyd.getId())).collectList().block(); + insertTestData(); - assertThat(persons).hasSize(2); + StepVerifier.create(repository.findAll(Flux.just(dave.getId(), boyd.getId()))) // + .expectNextCount(2) // + .verifyComplete(); } @Test // DATACASS-335 public void findAllByEmptyPublisherOfIdShouldReturnResults() { - - TestSubscriber testSubscriber = TestSubscriber.subscribe(repository.findAll(Flux.empty())); - - testSubscriber.await().assertComplete().assertNoValues().assertNoError(); + StepVerifier.create(repository.findAll(Flux.empty())).verifyComplete(); } @Test // DATACASS-335 public void countShouldReturnNumberOfRecords() { - TestSubscriber testSubscriber = TestSubscriber.subscribe(repository.count()); + insertTestData(); - testSubscriber.await().assertComplete().assertValueCount(1).assertValues(4L).assertNoError(); + StepVerifier.create(repository.count()).expectNext(4L).verifyComplete(); } @Test // DATACASS-335 public void insertEntityShouldInsertEntity() { - repository.deleteAll().block(); - Person person = new Person("36", "Homer", "Simpson"); - TestSubscriber testSubscriber = TestSubscriber.subscribe(repository.insert(person)); + StepVerifier.create(repository.insert(person)).expectNext(person).verifyComplete(); - testSubscriber.await().assertComplete().assertValueCount(1).assertValues(person); - repository.findAll().count().subscribeWith(TestSubscriber.create()).awaitAndAssertNextValues(1L); + StepVerifier.create(repository.findAll()).expectNextCount(1L).verifyComplete(); } @Test // DATACASS-335 public void insertShouldDeferredWrite() { - repository.deleteAll().block(); - Person person = new Person("36", "Homer", "Simpson"); repository.insert(person); - repository.findAll().count().subscribeWith(TestSubscriber.create()).awaitAndAssertNextValues(0L); + StepVerifier.create(repository.findAll()).expectNextCount(0L).verifyComplete(); } @Test // DATACASS-335 public void insertIterableOfEntitiesShouldInsertEntity() { - repository.deleteAll().block(); + StepVerifier.create(repository.insert(Arrays.asList(dave, oliver, boyd))).expectNextCount(3L).verifyComplete(); - TestSubscriber testSubscriber = TestSubscriber - .subscribe(repository.insert(Arrays.asList(dave, oliver, boyd))); - - testSubscriber.await().assertComplete().assertValueCount(3); - - repository.findAll().count().subscribeWith(TestSubscriber.create()).awaitAndAssertNextValues(3L); + StepVerifier.create(repository.findAll()).expectNextCount(3L).verifyComplete(); } @Test // DATACASS-335 public void insertPublisherOfEntitiesShouldInsertEntity() { - repository.deleteAll().block(); + StepVerifier.create(repository.insert(Flux.just(dave, oliver, boyd))).expectNextCount(3L).verifyComplete(); - TestSubscriber testSubscriber = TestSubscriber.subscribe(repository.insert(Flux.just(dave, oliver, boyd))); - - testSubscriber.await().assertComplete().assertValueCount(3); - repository.findAll().count().subscribeWith(TestSubscriber.create()).awaitAndAssertNextValues(3L); + StepVerifier.create(repository.findAll()).expectNextCount(3L).verifyComplete(); } @Test // DATACASS-335 @@ -259,14 +241,13 @@ public class SimpleReactiveCassandraRepositoryIntegrationTests extends AbstractK dave.setFirstname("Hello, Dave"); dave.setLastname("Bowman"); - TestSubscriber testSubscriber = TestSubscriber.subscribe(repository.save(dave)); + StepVerifier.create(repository.save(dave)).expectNextCount(1).verifyComplete(); - testSubscriber.await().assertComplete().assertValueCount(1).assertValues(dave); + StepVerifier.create(repository.findOne(dave.getId())).consumeNextWith(actual -> { - Person loaded = repository.findOne(dave.getId()).block(); - - assertThat(loaded.getFirstname()).isEqualTo(dave.getFirstname()); - assertThat(loaded.getLastname()).isEqualTo(dave.getLastname()); + assertThat(actual.getFirstname()).isEqualTo(dave.getFirstname()); + assertThat(actual.getLastname()).isEqualTo(dave.getLastname()); + }).verifyComplete(); } @Test // DATACASS-335 @@ -274,26 +255,17 @@ public class SimpleReactiveCassandraRepositoryIntegrationTests extends AbstractK Person person = new Person("36", "Homer", "Simpson"); - TestSubscriber testSubscriber = TestSubscriber.subscribe(repository.save(person)); + StepVerifier.create(repository.save(person)).expectNextCount(1).verifyComplete(); - testSubscriber.await().assertComplete().assertValueCount(1).assertValues(person); - - Person loaded = repository.findOne(person.getId()).block(); - - assertThat(loaded).isEqualTo(person); + StepVerifier.create(repository.findOne(person.getId())).expectNext(person).verifyComplete(); } @Test // DATACASS-335 public void saveIterableOfNewEntitiesShouldInsertEntity() { - repository.deleteAll().block(); + StepVerifier.create(repository.save(Arrays.asList(dave, oliver, boyd))).expectNextCount(3).verifyComplete(); - TestSubscriber testSubscriber = TestSubscriber - .subscribe(repository.save(Arrays.asList(dave, oliver, boyd))); - - testSubscriber.await().assertComplete().assertValueCount(3); - - repository.findAll().count().subscribeWith(TestSubscriber.create()).awaitAndAssertNextValues(3L); + StepVerifier.create(repository.findAll()).expectNextCount(3L).verifyComplete(); } @Test // DATACASS-335 @@ -304,82 +276,61 @@ public class SimpleReactiveCassandraRepositoryIntegrationTests extends AbstractK 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); + StepVerifier.create(repository.findOne(dave.getId())).expectNext(dave).verifyComplete(); - Person persistentDave = repository.findOne(dave.getId()).block(); - assertThat(persistentDave).isEqualTo(dave); - - Person persistentHomer = repository.findOne(person.getId()).block(); - assertThat(persistentHomer).isEqualTo(person); + StepVerifier.create(repository.findOne(person.getId())).expectNext(person).verifyComplete(); } @Test // DATACASS-335 public void savePublisherOfEntitiesShouldInsertEntity() { - repository.deleteAll().block(); + StepVerifier.create(repository.save(Flux.just(dave, oliver, boyd))).expectNextCount(3).verifyComplete(); - TestSubscriber testSubscriber = TestSubscriber.subscribe(repository.save(Flux.just(dave, oliver, boyd))); - - testSubscriber.await().assertComplete().assertValueCount(3); - repository.findAll().count().subscribeWith(TestSubscriber.create()).awaitAndAssertNextValues(3L); + StepVerifier.create(repository.findAll()).expectNextCount(3L).verifyComplete(); } @Test // DATACASS-335 public void deleteAllShouldRemoveEntities() { - repository.deleteAll().block(); + insertTestData(); - TestSubscriber testSubscriber = TestSubscriber.subscribe(repository.findAll()); + StepVerifier.create(repository.deleteAll()).verifyComplete(); - testSubscriber.await().assertComplete().assertValueCount(0); + StepVerifier.create(repository.findAll()).verifyComplete(); } @Test // DATACASS-335 public void deleteByIdShouldRemoveEntity() { - TestSubscriber testSubscriber = TestSubscriber.subscribe(repository.delete(dave.getId())); + StepVerifier.create(repository.delete(dave.getId())).verifyComplete(); - testSubscriber.await().assertComplete().assertNoValues(); - - TestSubscriber verificationSubscriber = TestSubscriber.subscribe(repository.findOne(dave.getId())); - - verificationSubscriber.await().assertComplete().assertNoValues(); + StepVerifier.create(repository.findOne(dave.getId())).expectNextCount(0).verifyComplete(); } @Test // DATACASS-335 public void deleteShouldRemoveEntity() { - TestSubscriber testSubscriber = TestSubscriber.subscribe(repository.delete(dave)); + StepVerifier.create(repository.delete(dave)).verifyComplete(); - testSubscriber.await().assertComplete().assertNoValues(); - - TestSubscriber verificationSubscriber = TestSubscriber.subscribe(repository.findOne(dave.getId())); - - verificationSubscriber.await().assertComplete().assertNoValues(); + StepVerifier.create(repository.findOne(dave.getId())).expectNextCount(0).verifyComplete(); } @Test // DATACASS-335 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.getId())); - verificationSubscriber.await().assertComplete().assertNoValues(); + StepVerifier.create(repository.findOne(boyd.getId())).expectNextCount(0).verifyComplete(); } @Test // DATACASS-335 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.getId())); - verificationSubscriber.await().assertComplete().assertNoValues(); + StepVerifier.create(repository.findOne(boyd.getId())).expectNextCount(0).verifyComplete(); } interface PersonRepostitory extends ReactiveCassandraRepository {} diff --git a/spring-data-cassandra/src/test/java/reactor/test/TestSubscriber.java b/spring-data-cassandra/src/test/java/reactor/test/TestSubscriber.java deleted file mode 100644 index d7ba663dd..000000000 --- a/spring-data-cassandra/src/test/java/reactor/test/TestSubscriber.java +++ /dev/null @@ -1,1129 +0,0 @@ -/* - * Copyright 2016-2017 the original author or authors. - * - * Licensed under the Apache License, Version 2.0 (the "License"); - * you may not use this file except in compliance with the License. - * 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 reactor.core.Fuseable; -import reactor.core.Receiver; -import reactor.core.Trackable; -import reactor.core.publisher.Operators; - -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; - -/** - * 
- *  ###############################################################
- *  ###############################################################
- *  ###############################################################
- *
- *  	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() + ")"; - } - -}