Spring/spring-data-cassandra
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

DATACASS-509 - Review and polish.

Browse Source
This commit is contained in:
John Blum
2017-11-11 23:07:55 -08:00
parent 65b4cd3c8c
commit ea957d73fa
6 changed files with 169 additions and 124 deletions
Download Patch File Download Diff File Expand all files Collapse all files

5
spring-data-cassandra/src/main/java/org/springframework/data/cassandra/ReactiveResultSet.java
View File

@@ -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<ExecutionInfo> getAllExecutionInfo();
}

37
spring-data-cassandra/src/main/java/org/springframework/data/cassandra/ReactiveSession.java
View File

@@ -52,6 +52,25 @@ import com.datastax.driver.core.exceptions.UnsupportedFeatureException;
*/
public interface ReactiveSession extends Closeable {
/**
* Whether this Session instance has been closed.
* <p/>
* 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.
* <p/>
@@ -179,22 +198,4 @@ public interface ReactiveSession extends Closeable {
@Override
void close();
/**
* Whether this Session instance has been closed.
* <p/>
* 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();
}

26
spring-data-cassandra/src/main/java/org/springframework/data/cassandra/config/AbstractCassandraConfiguration.java
View File

@@ -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

120
spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/cql/session/DefaultBridgedReactiveSession.java
View File

@@ -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<ResultSet> future = session.executeAsync(statement);
ListenableFuture<ReactiveResultSet> resultSetFuture = Futures.transform(future, DefaultReactiveResultSet::new);
ListenableFuture<ResultSet> future = this.session.executeAsync(statement);
ListenableFuture<ReactiveResultSet> 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<PreparedStatement> resultSetFuture = session.prepareAsync(statement);
ListenableFuture<PreparedStatement> 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<Row> rows() {
return getRows(Mono.just(resultSet));
return getRows(Mono.just(this.resultSet));
}
Flux<Row> getRows(Mono<ResultSet> nextResults) {
@@ -255,15 +271,17 @@ public class DefaultBridgedReactiveSession implements ReactiveSession {
}
MonoProcessor<ResultSet> 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<Row> 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<ExecutionInfo> getAllExecutionInfo() {
return resultSet.getAllExecutionInfo();
return this.resultSet.getAllExecutionInfo();
}
}
}

51
spring-data-cassandra/src/test/java/org/springframework/data/cassandra/core/cql/DefaultBridgedReactiveSessionIntegrationTests.java
View File

@@ -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<String> 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<ReactiveResultSet> execution = reactiveSession.execute(statement);
Collection<String> 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);
}

54
spring-data-cassandra/src/test/java/org/springframework/data/cassandra/core/cql/DefaultBridgedReactiveSessionUnitTests.java
View File

@@ -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<Row> 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<Row> 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<Row> rows = mockIterator();
Queue<Runnable> 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<Row> mockIterator() {
Row row = mock(Row.class);
Iterator<Row> rows = mock(Iterator.class);
when(rows.hasNext()).thenReturn(true);
when(rows.next()).thenReturn(row);
return rows;
}
@SuppressWarnings("all")
private static <T extends Statement> 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));
}
}