From 180dec546422094b31f541870710dee633ff526b Mon Sep 17 00:00:00 2001 From: John Blum Date: Sat, 11 Nov 2017 23:07:55 -0800 Subject: [PATCH] DATACASS-509 - Review and polish. --- .../data/cassandra/ReactiveResultSet.java | 5 +- .../data/cassandra/ReactiveSession.java | 37 +++--- .../AbstractCassandraConfiguration.java | 26 ++-- .../DefaultBridgedReactiveSession.java | 120 ++++++++++-------- ...ridgedReactiveSessionIntegrationTests.java | 51 ++++---- ...efaultBridgedReactiveSessionUnitTests.java | 54 +++++--- 6 files changed, 169 insertions(+), 124 deletions(-) diff --git a/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/ReactiveResultSet.java b/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/ReactiveResultSet.java index e8a719cee..43c00c599 100644 --- a/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/ReactiveResultSet.java +++ b/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/ReactiveResultSet.java @@ -15,10 +15,10 @@ */ package org.springframework.data.cassandra; -import reactor.core.publisher.Flux; - import java.util.List; +import reactor.core.publisher.Flux; + import com.datastax.driver.core.ColumnDefinitions; import com.datastax.driver.core.ExecutionInfo; import com.datastax.driver.core.Row; @@ -105,4 +105,5 @@ public interface ReactiveResultSet { * @return a list of the execution info for all the queries made for this result set. */ List getAllExecutionInfo(); + } diff --git a/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/ReactiveSession.java b/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/ReactiveSession.java index 9d69d15f8..0fd42ae78 100644 --- a/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/ReactiveSession.java +++ b/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/ReactiveSession.java @@ -52,6 +52,25 @@ import com.datastax.driver.core.exceptions.UnsupportedFeatureException; */ public interface ReactiveSession extends Closeable { + /** + * Whether this Session instance has been closed. + *

+ * Note that this method returns true as soon as the closing of this Session has started but it does not guarantee + * that the closing is done. If you want to guarantee that the closing is done, you can call {@code close()} and wait + * until it returns (or call the get method on {@code closeAsync()} with a very short timeout and check this doesn't + * timeout). + * + * @return {@code true} if this Session instance has been closed, {@code false} otherwise. + */ + boolean isClosed(); + + /** + * Returns the {@code Cluster} object this session is part of. + * + * @return the {@code Cluster} object this session is part of. + */ + Cluster getCluster(); + /** * Executes the provided query. *

@@ -179,22 +198,4 @@ public interface ReactiveSession extends Closeable { @Override void close(); - /** - * Whether this Session instance has been closed. - *

- * Note that this method returns true as soon as the closing of this Session has started but it does not guarantee - * that the closing is done. If you want to guarantee that the closing is done, you can call {@code close()} and wait - * until it returns (or call the get method on {@code closeAsync()} with a very short timeout and check this doesn't - * timeout). - * - * @return {@code true} if this Session instance has been closed, {@code false} otherwise. - */ - boolean isClosed(); - - /** - * Returns the {@code Cluster} object this session is part of. - * - * @return the {@code Cluster} object this session is part of. - */ - Cluster getCluster(); } diff --git a/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/config/AbstractCassandraConfiguration.java b/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/config/AbstractCassandraConfiguration.java index c25a8c5fd..af7d14e8a 100644 --- a/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/config/AbstractCassandraConfiguration.java +++ b/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/config/AbstractCassandraConfiguration.java @@ -129,19 +129,6 @@ public abstract class AbstractCassandraConfiguration extends AbstractClusterConf } } - /** - * Register custom {@link Converter}s in a {@link CustomConversions} object if required. These - * {@link CustomConversions} will be registered with the {@link #cassandraConverter()} and {@link #cassandraMapping()} - * . Returns an empty {@link CustomConversions} instance by default. - * - * @return must not be {@literal null}. - * @since 1.5 - */ - @Bean - public CustomConversions customConversions() { - return new CassandraCustomConversions(Collections.emptyList()); - } - /** * Return the {@link MappingContext} instance to map Entities to properties. * @@ -169,6 +156,19 @@ public abstract class AbstractCassandraConfiguration extends AbstractClusterConf return mappingContext; } + /** + * Register custom {@link Converter}s in a {@link CustomConversions} object if required. These + * {@link CustomConversions} will be registered with the {@link #cassandraConverter()} and {@link #cassandraMapping()} + * . Returns an empty {@link CustomConversions} instance by default. + * + * @return must not be {@literal null}. + * @since 1.5 + */ + @Bean + public CustomConversions customConversions() { + return new CassandraCustomConversions(Collections.emptyList()); + } + /** * Return the {@link Set} of initial entity classes. Scans by default the class path using * {@link #getEntityBasePackages()}. Can be overriden by subclasses to skip class path scanning and return a fixed set diff --git a/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/cql/session/DefaultBridgedReactiveSession.java b/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/cql/session/DefaultBridgedReactiveSession.java index 08f4984a9..a2c0d9f49 100644 --- a/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/cql/session/DefaultBridgedReactiveSession.java +++ b/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/cql/session/DefaultBridgedReactiveSession.java @@ -15,23 +15,33 @@ */ package org.springframework.data.cassandra.core.cql.session; +import java.util.List; +import java.util.Map; +import java.util.concurrent.ExecutionException; + import reactor.core.publisher.Flux; import reactor.core.publisher.Mono; import reactor.core.publisher.MonoProcessor; import reactor.core.publisher.MonoSink; import reactor.core.scheduler.Scheduler; -import java.util.List; -import java.util.Map; -import java.util.concurrent.ExecutionException; - -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; import org.springframework.data.cassandra.ReactiveResultSet; import org.springframework.data.cassandra.ReactiveSession; import org.springframework.util.Assert; -import com.datastax.driver.core.*; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import com.datastax.driver.core.Cluster; +import com.datastax.driver.core.ColumnDefinitions; +import com.datastax.driver.core.ExecutionInfo; +import com.datastax.driver.core.PreparedStatement; +import com.datastax.driver.core.RegularStatement; +import com.datastax.driver.core.ResultSet; +import com.datastax.driver.core.Row; +import com.datastax.driver.core.Session; +import com.datastax.driver.core.SimpleStatement; +import com.datastax.driver.core.Statement; import com.google.common.util.concurrent.Futures; import com.google.common.util.concurrent.ListenableFuture; @@ -89,6 +99,22 @@ public class DefaultBridgedReactiveSession implements ReactiveSession { this(session); } + /* (non-Javadoc) + * @see org.springframework.data.cassandra.ReactiveSession#isClosed() + */ + @Override + public boolean isClosed() { + return this.session.isClosed(); + } + + /* (non-Javadoc) + * @see org.springframework.data.cassandra.ReactiveSession#getCluster() + */ + @Override + public Cluster getCluster() { + return this.session.getCluster(); + } + /* (non-Javadoc) * @see org.springframework.data.cassandra.ReactiveSession#execute(java.lang.String) */ @@ -137,12 +163,15 @@ public class DefaultBridgedReactiveSession implements ReactiveSession { logger.debug("Executing Statement [{}]", statement); } - ListenableFuture future = session.executeAsync(statement); - ListenableFuture resultSetFuture = Futures.transform(future, DefaultReactiveResultSet::new); + ListenableFuture future = this.session.executeAsync(statement); + + ListenableFuture resultSetFuture = + Futures.transform(future, DefaultReactiveResultSet::new); adaptFuture(resultSetFuture, sink); - } catch (Exception e) { - sink.error(e); + } + catch (Exception cause) { + sink.error(cause); } }); } @@ -173,11 +202,12 @@ public class DefaultBridgedReactiveSession implements ReactiveSession { logger.debug("Preparing Statement [{}]", statement); } - ListenableFuture resultSetFuture = session.prepareAsync(statement); + ListenableFuture resultSetFuture = this.session.prepareAsync(statement); adaptFuture(resultSetFuture, sink); - } catch (Exception e) { - sink.error(e); + } + catch (Exception cause) { + sink.error(cause); } }); } @@ -187,23 +217,7 @@ public class DefaultBridgedReactiveSession implements ReactiveSession { */ @Override public void close() { - session.close(); - } - - /* (non-Javadoc) - * @see org.springframework.data.cassandra.ReactiveSession#isClosed() - */ - @Override - public boolean isClosed() { - return session.isClosed(); - } - - /* (non-Javadoc) - * @see org.springframework.data.cassandra.ReactiveSession#getCluster() - */ - @Override - public Cluster getCluster() { - return session.getCluster(); + this.session.close(); } /** @@ -219,10 +233,12 @@ public class DefaultBridgedReactiveSession implements ReactiveSession { if (future.isDone()) { try { sink.success(future.get()); - } catch (ExecutionException e) { - sink.error(e.getCause()); - } catch (Exception e) { - sink.error(e); + } + catch (ExecutionException cause) { + sink.error(cause.getCause()); + } + catch (Exception cause) { + sink.error(cause); } } }, Runnable::run); @@ -241,7 +257,7 @@ public class DefaultBridgedReactiveSession implements ReactiveSession { */ @Override public Flux rows() { - return getRows(Mono.just(resultSet)); + return getRows(Mono.just(this.resultSet)); } Flux getRows(Mono nextResults) { @@ -255,15 +271,17 @@ public class DefaultBridgedReactiveSession implements ReactiveSession { } MonoProcessor processor = MonoProcessor.create(); - return rows // - .doOnComplete(() -> fetchMore(it.fetchMoreResults(), processor)) // - .concatWith(getRows(processor)); + + return rows + .doOnComplete(() -> fetchMore(it.fetchMoreResults(), processor)) + .concatWith(getRows(processor)); }); } static Flux toRows(ResultSet resultSet) { int prefetch = Math.max(1, resultSet.getAvailableWithoutFetching()); + return Flux.fromIterable(resultSet).take(prefetch); } @@ -274,18 +292,20 @@ public class DefaultBridgedReactiveSession implements ReactiveSession { future.addListener(() -> { try { - sink.onNext(future.get()); sink.onComplete(); - } catch (ExecutionException e) { - sink.onError(e.getCause()); - } catch (Exception e) { - sink.onError(e); + } + catch (ExecutionException cause) { + sink.onError(cause.getCause()); + } + catch (Exception cause) { + sink.onError(cause); } }, Runnable::run); - } catch (Exception e) { - sink.onError(e); + } + catch (Exception cause) { + sink.onError(cause); } } @@ -294,7 +314,7 @@ public class DefaultBridgedReactiveSession implements ReactiveSession { */ @Override public ColumnDefinitions getColumnDefinitions() { - return resultSet.getColumnDefinitions(); + return this.resultSet.getColumnDefinitions(); } /* (non-Javadoc) @@ -302,7 +322,7 @@ public class DefaultBridgedReactiveSession implements ReactiveSession { */ @Override public boolean wasApplied() { - return resultSet.wasApplied(); + return this.resultSet.wasApplied(); } /* (non-Javadoc) @@ -310,7 +330,7 @@ public class DefaultBridgedReactiveSession implements ReactiveSession { */ @Override public ExecutionInfo getExecutionInfo() { - return resultSet.getExecutionInfo(); + return this.resultSet.getExecutionInfo(); } /* (non-Javadoc) @@ -318,7 +338,7 @@ public class DefaultBridgedReactiveSession implements ReactiveSession { */ @Override public List getAllExecutionInfo() { - return resultSet.getAllExecutionInfo(); + return this.resultSet.getAllExecutionInfo(); } } } diff --git a/spring-data-cassandra/src/test/java/org/springframework/data/cassandra/core/cql/DefaultBridgedReactiveSessionIntegrationTests.java b/spring-data-cassandra/src/test/java/org/springframework/data/cassandra/core/cql/DefaultBridgedReactiveSessionIntegrationTests.java index 059c3711c..98ff206e0 100644 --- a/spring-data-cassandra/src/test/java/org/springframework/data/cassandra/core/cql/DefaultBridgedReactiveSessionIntegrationTests.java +++ b/spring-data-cassandra/src/test/java/org/springframework/data/cassandra/core/cql/DefaultBridgedReactiveSessionIntegrationTests.java @@ -15,19 +15,20 @@ */ package org.springframework.data.cassandra.core.cql; -import static org.assertj.core.api.Assertions.*; - -import reactor.core.publisher.Mono; -import reactor.core.scheduler.Schedulers; -import reactor.test.StepVerifier; +import static org.assertj.core.api.Assertions.assertThat; import java.util.ArrayList; import java.util.Collection; import java.util.List; import java.util.concurrent.ConcurrentLinkedQueue; +import reactor.core.publisher.Mono; +import reactor.core.scheduler.Schedulers; +import reactor.test.StepVerifier; + import org.junit.Before; import org.junit.Test; + import org.springframework.data.cassandra.ReactiveResultSet; import org.springframework.data.cassandra.core.cql.session.DefaultBridgedReactiveSession; import org.springframework.data.cassandra.test.util.AbstractKeyspaceCreatingIntegrationTest; @@ -65,10 +66,9 @@ public class DefaultBridgedReactiveSessionIntegrationTests extends AbstractKeysp assertThat(keyspace.getTable("users")).isNull(); - StepVerifier.create(execution).consumeNextWith(actual -> { - - assertThat(actual.wasApplied()).isTrue(); - }).verifyComplete(); + StepVerifier.create(execution) + .consumeNextWith(actual -> assertThat(actual.wasApplied()).isTrue()) + .verifyComplete(); assertThat(keyspace.getTable("users")).isNotNull(); } @@ -84,13 +84,9 @@ 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');"); - StepVerifier.create(reactiveSession.execute("SELECT * FROM users;")).consumeNextWith(actual -> { - - StepVerifier.create(actual.rows()).consumeNextWith(row -> { - - assertThat(row.getString("userid")).isEqualTo("White"); - }).verifyComplete(); - }).verifyComplete(); + StepVerifier.create(reactiveSession.execute("SELECT * FROM users;")).consumeNextWith(actual -> + StepVerifier.create(actual.rows()).consumeNextWith(row -> + assertThat(row.getString("userid")).isEqualTo("White")).verifyComplete()).verifyComplete(); } @Test // DATACASS-335 @@ -99,39 +95,44 @@ public class DefaultBridgedReactiveSessionIntegrationTests extends AbstractKeysp session.execute("CREATE TABLE users (\n" + " userid text PRIMARY KEY,\n" + " first_name text\n" + ");"); StepVerifier.create(reactiveSession.prepare("INSERT INTO users (userid, first_name) VALUES (?, ?);")) - .consumeNextWith(actual -> { - - assertThat(actual.getQueryString()).isEqualTo("INSERT INTO users (userid, first_name) VALUES (?, ?);"); - }).verifyComplete(); + .consumeNextWith(actual -> + assertThat(actual.getQueryString()).isEqualTo("INSERT INTO users (userid, first_name) VALUES (?, ?);")) + .verifyComplete(); } @Test // DATACASS-509 public void shouldFetchBatches() { String createTable = "CREATE TABLE users (\n" + " userid text PRIMARY KEY,\n" + " first_name text\n" + ");"; + this.session.execute(createTable); List keys = new ArrayList<>(); + for (int i = 0; i < 100; i++) { String key = String.format("u-03%d", i); String value = "v-" + i; + keys.add(key); - this.session.execute(String.format("INSERT INTO users (userid,first_name) VALUES ('%s', '%s');", key, value)); + this.session.execute(String.format("INSERT INTO users (userid, first_name) VALUES ('%s', '%s');", key, value)); } - session.getCluster().register(QueryLogger.builder().build()); + this.session.getCluster().register(QueryLogger.builder().build()); SimpleStatement statement = new SimpleStatement("SELECT * FROM users;"); + statement.setFetchSize(10); Mono execution = reactiveSession.execute(statement); Collection received = new ConcurrentLinkedQueue<>(); - StepVerifier.create(execution.flatMapMany(ReactiveResultSet::rows).map(row -> row.getString(0))) // - .recordWith(() -> received) // - .expectNextCount(100).verifyComplete(); + + StepVerifier.create(execution.flatMapMany(ReactiveResultSet::rows).map(row -> row.getString(0))) + .recordWith(() -> received) + .expectNextCount(100) + .verifyComplete(); assertThat(received).containsAll(keys).hasSize(100); } diff --git a/spring-data-cassandra/src/test/java/org/springframework/data/cassandra/core/cql/DefaultBridgedReactiveSessionUnitTests.java b/spring-data-cassandra/src/test/java/org/springframework/data/cassandra/core/cql/DefaultBridgedReactiveSessionUnitTests.java index 8c7b9a9e5..8912a69dc 100644 --- a/spring-data-cassandra/src/test/java/org/springframework/data/cassandra/core/cql/DefaultBridgedReactiveSessionUnitTests.java +++ b/spring-data-cassandra/src/test/java/org/springframework/data/cassandra/core/cql/DefaultBridgedReactiveSessionUnitTests.java @@ -15,23 +15,30 @@ */ package org.springframework.data.cassandra.core.cql; -import static org.assertj.core.api.Assertions.*; -import static org.mockito.Mockito.*; - -import reactor.core.publisher.Flux; -import reactor.test.StepVerifier; +import static org.assertj.core.api.Assertions.assertThat; +import static org.mockito.Mockito.any; +import static org.mockito.Mockito.doAnswer; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.never; +import static org.mockito.Mockito.times; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; import java.util.ArrayDeque; import java.util.Collections; import java.util.Iterator; import java.util.Queue; +import reactor.core.publisher.Flux; +import reactor.test.StepVerifier; + import org.junit.Before; import org.junit.Test; import org.junit.runner.RunWith; import org.mockito.ArgumentMatchers; import org.mockito.Mock; import org.mockito.junit.MockitoJUnitRunner; + import org.springframework.data.cassandra.ReactiveResultSet; import org.springframework.data.cassandra.core.cql.session.DefaultBridgedReactiveSession; @@ -153,16 +160,19 @@ public class DefaultBridgedReactiveSessionUnitTests { } @Test // DATACASS-509 - public void shouldReadNotMoreThanAvailable() throws Exception { + public void shouldNotReadMoreThanAvailable() throws Exception { Iterator rows = mockIterator(); ResultSet resultSet = mock(ResultSet.class); + when(resultSet.getAvailableWithoutFetching()).thenReturn(10); when(resultSet.iterator()).thenReturn(rows); + doAnswer(invocation -> { Runnable listener = invocation.getArgument(0); + listener.run(); return null; @@ -171,7 +181,10 @@ public class DefaultBridgedReactiveSessionUnitTests { when(future.get()).thenReturn(resultSet); when(resultSet.isFullyFetched()).thenReturn(true); - reactiveSession.execute(new SimpleStatement("")).flatMapMany(ReactiveResultSet::rows).collectList().subscribe(); + reactiveSession.execute(new SimpleStatement("")) + .flatMapMany(ReactiveResultSet::rows) + .collectList() + .subscribe(); verify(rows, times(10)).next(); verify(resultSet, never()).fetchMoreResults(); @@ -183,16 +196,19 @@ public class DefaultBridgedReactiveSessionUnitTests { Iterator rows = mockIterator(); ResultSet resultSet = mock(ResultSet.class); + when(resultSet.getAvailableWithoutFetching()).thenReturn(10); when(resultSet.iterator()).thenReturn(rows); ResultSet emptyResultSet = mock(ResultSet.class); + when(emptyResultSet.iterator()).thenReturn(Collections.emptyIterator()); when(emptyResultSet.isFullyFetched()).thenReturn(true); doAnswer(invocation -> { Runnable listener = invocation.getArgument(0); + listener.run(); return null; @@ -212,12 +228,14 @@ public class DefaultBridgedReactiveSessionUnitTests { } @Test // DATACASS-509 - public void shouldFetchDependingOfCompletion() throws Exception { + public void shouldFetchDependingOnCompletion() throws Exception { Iterator rows = mockIterator(); Queue runnables = new ArrayDeque<>(); + ResultSet resultSet = mock(ResultSet.class); + when(resultSet.getAvailableWithoutFetching()).thenReturn(10); when(resultSet.iterator()).thenReturn(rows); @@ -234,46 +252,50 @@ public class DefaultBridgedReactiveSessionUnitTests { StepVerifier.create(flux, 0) // .then(() -> runnables.poll().run()) // complete the first future from executeAsync() - .thenRequest(9).expectNextCount(9) // + .thenRequest(9).expectNextCount(9) .then(() -> { // feed the 9 elements from the initial ResultSet verify(resultSet, never()).fetchMoreResults(); - }).thenRequest(1).expectNextCount(1) // + }).thenRequest(1).expectNextCount(1) .then(() -> { // initial ResultSet exhausted, fetch next chunk verify(resultSet).fetchMoreResults(); runnables.poll().run(); - }).thenRequest(1).expectNextCount(1) // + }).thenRequest(1).expectNextCount(1) .then(() -> { // first element from the second ResultSet received, no subsequent fetch assertThat(runnables).isEmpty(); - }).thenRequest(19).expectNextCount(9) // + }).thenRequest(19).expectNextCount(9) .then(() -> { // second ResultSet exhausted assertThat(runnables).hasSize(1); runnables.poll().run(); - }) // - .thenRequest(10).expectNextCount(10) // + }) + .thenRequest(10).expectNextCount(10) .verifyComplete(); } + @SuppressWarnings("unchecked") private static Iterator mockIterator() { Row row = mock(Row.class); + Iterator rows = mock(Iterator.class); + when(rows.hasNext()).thenReturn(true); when(rows.next()).thenReturn(row); return rows; } + @SuppressWarnings("all") private static T eq(T value) { - return ArgumentMatchers.argThat(argument -> argument instanceof Statement // - ? value.toString().equals(argument.toString()) // + return ArgumentMatchers.argThat(argument -> argument instanceof Statement + ? value.toString().equals(argument.toString()) : value.equals(argument)); } }