DATACASS-335 - Add support for reactive data access.

We now support reactive data access with Spring Data Cassandra by adopting Datastax' asynchronous driver.

ReactiveCqlTemplate and ReactiveCassandraTemplate use Project Reactor wrapper types Mono and Flux to implement Template API and repository support. Reactive template supports common operations such as:

* Query/Execution methods for static CQL and prepared statements
* Insert/Save/Update/Delete methods
* Exists and Count projections
* Reactive Callback methods

Person person = new Person("Dave", 25);

template.insert(person) //
    .flatMap(p -> template.update(new Person("Sven", 25))) //
    .flatMap(p -> template.selectOneById(person.getId(), Person.class)) //
    .subscribeWith(TestSubscriber.create()) //
    .await() //
    .assertValuesWith(result -> {
        assertThat(result.getFirstName(), is(equalTo("Sven")));
    });

Reactive Repository support is built on top of ReactiveCassandraTemplate using ReactiveCassandraRepository as the store-specific base repository. Reactive repositories are enabled by using @EnableReactiveCassandraRepositories on a @Configuration class to opt-in for reactive support. Reactive repositories can be composed of a reactive base interface such as

* ReactiveCrudRepository
* ReactiveSortingRepository
* RxJava1CrudRepository
* RxJava1SortingRepository

and are identified as reactive repository if one method uses a reactive wrapper type (such as Flux or Observable). If a reactive repository is discovered, it's not implemented by the blocking repository support but with the reactive repository factory. Blocking methods are not (yet) synchronized when using a reactive repository so each repository method must use a reactive wrapper result type. Reactive repository support with Spring Data allows using RxJava1 and Project Reactor types to declare repository methods. Reactive wrapper types are internally converted so the composition library choice on repository level is left up to the user.

There's feature parity between Reactive Cassandra repository support and blocking repository support.

Feature overview:

* Query Methods using String queries and Query Derivation
* Projections

@Configuration
@EnableReactiveCassandraRepositories
class ApplicationConfig extends AbstractReactiveCassandraConfiguration {

  @Override
  protected String getKeyspaceName() {
    return "mykeyspace";
  }

  @Override
  protected String getEntityBasePackages() {
    return new String[] {"com.springdata.cassandra"};
  }
}

public interface PersonRepository extends ReactiveSortingRepository<Person, String> {

  Flux<Person> findByFirstname(String firstname);

  Flux<Person> findByFirstname(Publisher<String> firstname);

  Mono<Person> findByFirstnameAndLastname(String firstname, String lastname);
}

public interface PersonRepository extends RxJava1SortingRepository<Person, String> {

  Observable<Person> findByFirstname(String firstname);

  Observable<Person> findByFirstname(Single<String> firstname);

  Single<Person> findByFirstnameAndLastname(String firstname, String lastname);
}
This commit is contained in:
Mark Paluch
2016-09-16 14:56:07 +02:00
committed by John Blum
parent dcc34a284a
commit 52f4f570a9
78 changed files with 11579 additions and 487 deletions

View File

@@ -0,0 +1,45 @@
/*
* Copyright 2016 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.cassandra.core;
import com.datastax.driver.core.BoundStatement;
import com.datastax.driver.core.PreparedStatement;
import com.datastax.driver.core.exceptions.DriverException;
/**
* Simple adapter for {@link PreparedStatementBinder} that applies a given array of arguments.
*
* @author Mark Paluch
* @since 2.0
*/
public class ArgumentPreparedStatementBinder implements PreparedStatementBinder {
private final Object[] args;
/**
* Create a new {@link ArgumentPreparedStatementBinder} for the given arguments.
*
* @param args the arguments to set. May be empty or {@link null} if no arguments are provided.
*/
public ArgumentPreparedStatementBinder(Object[] args) {
this.args = args;
}
@Override
public BoundStatement bindValues(PreparedStatement ps) throws DriverException {
return args != null ? ps.bind(args) : ps.bind();
}
}

View File

@@ -0,0 +1,95 @@
/*
* Copyright 2016 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.cassandra.core;
import java.util.Map;
import org.springframework.util.LinkedCaseInsensitiveMap;
import com.datastax.driver.core.ColumnDefinitions;
import com.datastax.driver.core.Row;
/**
* {@link RowMapper} implementation that creates a {@code java.util.Map} for each row, representing all columns as
* key-value pairs: one entry for each column, with the column name as key.
* <p>
* The Map implementation to use and the key to use for each column in the column Map can be customized through
* overriding {@link #createColumnMap} and {@link #getColumnKey}, respectively.
* <p>
* <b>Note:</b> By default, ColumnMapRowMapper will try to build a linked Map with case-insensitive keys, to preserve
* column order as well as allow any casing to be used for column names. This requires Commons Collections on the
* classpath (which will be autodetected). Else, the fallback is a standard linked HashMap, which will still preserve
* column order but requires the application to specify the column names in the same casing as exposed by the driver.
*
* @author Mark Paluch
* @since 2.0
* @see ReactiveCqlTemplate#queryForFlux(String)
* @see ReactiveCqlTemplate#queryForMap(String)
*/
public class ColumnMapRowMapper implements RowMapper<Map<String, Object>> {
@Override
public Map<String, Object> mapRow(Row rs, int rowNum) {
ColumnDefinitions columnDefinitions = rs.getColumnDefinitions();
int columnCount = columnDefinitions.size();
Map<String, Object> mapOfColValues = createColumnMap(columnCount);
for (int i = 0; i < columnCount; i++) {
String key = getColumnKey(columnDefinitions.getName(i));
Object obj = getColumnValue(rs, i);
mapOfColValues.put(key, obj);
}
return mapOfColValues;
}
/**
* Create a {@link Map} instance to be used as column map.
* <p>
* By default, a linked case-insensitive Map will be created.
*
* @param columnCount the column count, to be used as initial capacity for the {@link Map}, must not be {@literal null}.
* @return the new Map instance.
* @see org.springframework.util.LinkedCaseInsensitiveMap
*/
protected Map<String, Object> createColumnMap(int columnCount) {
return new LinkedCaseInsensitiveMap<>(columnCount);
}
/**
* Determine the key to use for the given column in the column Map.
*
* @param columnName the column name as returned by the {@link Row}, must not be {@literal null}.
* @return the column key to use.
* @see ColumnDefinitions#getName(int)
*/
protected String getColumnKey(String columnName) {
return columnName;
}
/**
* Retrieve a CQL object value for the specified column.
* <p>
* The default implementation uses the {@code getObject} method.
*
* @param row is the {@link Row} holding the data, must not be {@literal null}.
* @param index is the column index.
* @return the Object returned
*/
protected Object getColumnValue(Row row, int index) {
return row.getObject(index);
}
}

View File

@@ -0,0 +1,38 @@
/*
* Copyright 2016 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.cassandra.core;
/**
* Interface to be implemented by objects that can provide CQL strings.
* <p>
* Typically implemented by {@link PreparedStatementCreator}s and statement callbacks that want to expose the CQL they
* use to create their statements, to allow for better contextual information in case of exceptions.
*
* @author Mark Paluch
* @since 2.0
* @see PreparedStatementCreator
* @see ReactivePreparedStatementCreator
* @see ReactiveStatementCallback
*/
public interface CqlProvider {
/**
* Return the CQL string for this object, i.e. typically the CQL used for creating statements.
*
* @return the CQL string, or {@literal null}.
*/
String getCql();
}

View File

@@ -55,19 +55,7 @@ import org.springframework.dao.QueryTimeoutException;
import org.springframework.dao.support.PersistenceExceptionTranslator;
import org.springframework.util.Assert;
import com.datastax.driver.core.BoundStatement;
import com.datastax.driver.core.CodecRegistry;
import com.datastax.driver.core.ColumnDefinitions;
import com.datastax.driver.core.Host;
import com.datastax.driver.core.PreparedStatement;
import com.datastax.driver.core.ProtocolVersion;
import com.datastax.driver.core.ResultSet;
import com.datastax.driver.core.ResultSetFuture;
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.datastax.driver.core.TypeCodec;
import com.datastax.driver.core.*;
import com.datastax.driver.core.ColumnDefinitions.Definition;
import com.datastax.driver.core.exceptions.DriverException;
import com.datastax.driver.core.querybuilder.Batch;

View File

@@ -0,0 +1,278 @@
/*
* Copyright 2016 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.cassandra.core;
import java.util.List;
import java.util.Map;
import java.util.concurrent.CompletableFuture;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.util.Assert;
import com.datastax.driver.core.*;
import com.google.common.util.concurrent.ListenableFuture;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import reactor.core.scheduler.Scheduler;
import reactor.core.scheduler.Schedulers;
/**
* Default implementation of a {@link ReactiveSession}. This implementation bridges asynchronous {@link Session} methods
* to reactive execution patterns.
* <p>
* Calls are deferred until a subscriber subscribes to the resulting {@link org.reactivestreams.Publisher}. The calls
* are executed by subscribing to {@link ListenableFuture} and returning the result as calls complete.
* <p>
* {@link ResultSet} implements transparent paging that invokes in the middle of result streaming blocking calls to
* Cassandra. {@link DefaultBridgedReactiveSession} uses therefore {@link ReactiveResultSet} to avoid client thread
* blocking. Elements are emitted on netty EventLoop threads and transported by the provided {@link Scheduler}. However,
* this is an intermediate solution until Datastax can provide a fully reactive driver.
* <p>
* All CQL operations performed by this class are logged at debug level, using
* "org.springframework.cassandra.core.DefaultBridgedReactiveSession" as log category.
* <p>
*
* @author Mark Paluch
* @since 2.0
* @see Mono
* @see ReactiveResultSet
* @see Scheduler
* @see ReactiveSession
*/
public class DefaultBridgedReactiveSession implements ReactiveSession {
private final Logger logger = LoggerFactory.getLogger(getClass());
private final Session session;
private final Scheduler scheduler;
/**
* Creates a new {@link DefaultBridgedReactiveSession} for a {@link Session} and {@link Scheduler}.
*
* @param session must not be {@literal null}.
* @param scheduler must not be {@literal null}.
*/
public DefaultBridgedReactiveSession(Session session, Scheduler scheduler) {
Assert.notNull(session, "Session must not be null");
Assert.notNull(scheduler, "Scheduler must not be null");
this.session = session;
this.scheduler = scheduler;
}
/* (non-Javadoc)
* @see org.springframework.cassandra.core.ReactiveSession#execute(java.lang.String)
*/
@Override
public Mono<ReactiveResultSet> execute(String query) {
Assert.hasText(query, "Query must not be empty");
return execute(new SimpleStatement(query));
}
/* (non-Javadoc)
* @see org.springframework.cassandra.core.ReactiveSession#execute(java.lang.String, java.lang.Object[])
*/
@Override
public Mono<ReactiveResultSet> execute(String query, Object... values) {
Assert.hasText(query, "Query must not be empty");
return execute(new SimpleStatement(query, values));
}
/* (non-Javadoc)
* @see org.springframework.cassandra.core.ReactiveSession#execute(java.lang.String, java.util.Map)
*/
@Override
public Mono<ReactiveResultSet> execute(String query, Map<String, Object> values) {
Assert.hasText(query, "Query must not be empty");
return execute(new SimpleStatement(query, values));
}
/* (non-Javadoc)
* @see org.springframework.cassandra.core.ReactiveSession#execute(com.datastax.driver.core.Statement)
*/
@Override
public Mono<ReactiveResultSet> execute(Statement statement) {
Assert.notNull(statement, "Statement must not be null");
return Mono.defer(() -> {
try {
if (logger.isDebugEnabled()) {
logger.debug("Executing Statement [{}]", statement);
}
CompletableFuture<ReactiveResultSet> future = new CompletableFuture<>();
ResultSetFuture resultSetFuture = session.executeAsync(statement);
resultSetFuture.addListener(() -> {
if (resultSetFuture.isDone()) {
try {
future.complete(new DefaultReactiveResultSet(resultSetFuture.getUninterruptibly(), scheduler));
} catch (Exception e) {
future.completeExceptionally(e);
}
}
}, Runnable::run);
return Mono.fromFuture(future);
} catch (Exception e) {
return Mono.error(e);
}
}).subscribeOn(scheduler);
}
/* (non-Javadoc)
* @see org.springframework.cassandra.core.ReactiveSession#prepare(java.lang.String)
*/
@Override
public Mono<PreparedStatement> prepare(String query) {
Assert.hasText(query, "Query must not be empty");
return prepare(new SimpleStatement(query));
}
/* (non-Javadoc)
* @see org.springframework.cassandra.core.ReactiveSession#prepare(com.datastax.driver.core.RegularStatement)
*/
@Override
public Mono<PreparedStatement> prepare(RegularStatement statement) {
Assert.notNull(statement, "Statement must not be null");
return Mono.defer(() -> {
try {
if (logger.isDebugEnabled()) {
logger.debug("Preparing Statement [{}]", statement);
}
CompletableFuture<PreparedStatement> future = new CompletableFuture<>();
ListenableFuture<PreparedStatement> resultSetFuture = session.prepareAsync(statement);
resultSetFuture.addListener(() -> {
if (resultSetFuture.isDone()) {
try {
future.complete(resultSetFuture.get());
} catch (Exception e) {
future.completeExceptionally(e);
}
}
}, Runnable::run);
return Mono.fromFuture(future);
} catch (Exception e) {
return Mono.error(e);
}
}).subscribeOn(scheduler);
}
/* (non-Javadoc)
* @see org.springframework.cassandra.core.ReactiveSession#close()
*/
@Override
public void close() {
session.close();
}
/* (non-Javadoc)
* @see org.springframework.cassandra.core.ReactiveSession#isClosed()
*/
@Override
public boolean isClosed() {
return session.isClosed();
}
/* (non-Javadoc)
* @see org.springframework.cassandra.core.ReactiveSession#getCluster()
*/
@Override
public Cluster getCluster() {
return session.getCluster();
}
private static class DefaultReactiveResultSet implements ReactiveResultSet {
private final ResultSet resultSet;
private final Scheduler scheduler;
DefaultReactiveResultSet(ResultSet resultSet, Scheduler scheduler) {
this.resultSet = resultSet;
this.scheduler = scheduler;
}
/* (non-Javadoc)
* @see org.springframework.cassandra.core.ReactiveResultSet#rows()
*/
@Override
public Flux<Row> rows() {
int prefetch = Math.max(1, resultSet.getAvailableWithoutFetching());
return Flux.fromIterable(resultSet) //
.subscribeOn(scheduler) //
.publishOn(Schedulers.immediate(), prefetch); // limit prefetching to available size
}
/* (non-Javadoc)
* @see org.springframework.cassandra.core.ReactiveResultSet#getColumnDefinitions()
*/
@Override
public ColumnDefinitions getColumnDefinitions() {
return resultSet.getColumnDefinitions();
}
/* (non-Javadoc)
* @see org.springframework.cassandra.core.ReactiveResultSet#wasApplied()
*/
@Override
public boolean wasApplied() {
return resultSet.wasApplied();
}
/* (non-Javadoc)
* @see org.springframework.cassandra.core.ReactiveResultSet#getExecutionInfo()
*/
@Override
public ExecutionInfo getExecutionInfo() {
return resultSet.getExecutionInfo();
}
/* (non-Javadoc)
* @see org.springframework.cassandra.core.ReactiveResultSet#getAllExecutionInfo()
*/
@Override
public List<ExecutionInfo> getAllExecutionInfo() {
return resultSet.getAllExecutionInfo();
}
}
}

View File

@@ -0,0 +1,43 @@
/*
* Copyright 2016 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.cassandra.core;
/**
* Default implementation of {@link ReactiveSessionFactory}.
* <p>
* This implementation returns always the same {@link ReactiveSession}.
*
* @author Mark Paluch
* @since 2.0
*/
public class DefaultReactiveSessionFactory implements ReactiveSessionFactory {
private final ReactiveSession session;
/**
* Create a new {@link ReactiveRowMapperResultSetExtractor}.
*
* @param session the {@link ReactiveSession} provides connections to Cassandra, must not be {@literal null}.
*/
public DefaultReactiveSessionFactory(ReactiveSession session) {
this.session = session;
}
@Override
public ReactiveSession getSession() {
return session;
}
}

View File

@@ -20,10 +20,31 @@ import com.datastax.driver.core.PreparedStatement;
import com.datastax.driver.core.exceptions.DriverException;
/**
* General callback interface used by the {@link CqlTemplate} and {@link ReactiveCqlTemplate} classes.
* <p>
* This interface binds values on a {@link PreparedStatement} provided by the {@link CqlTemplate} class, for each of a
* number of updates in a batch using the same CQL. Implementations are responsible for setting any necessary
* parameters. CQL with placeholders will already have been supplied.
* <p>
* It's easier to use this interface than {@link PreparedStatementCreator}: The {@link CqlTemplate} will create the
* {@link PreparedStatement}, with the callback only being responsible for setting parameter values.
* <p>
* Implementations <i>do not</i> need to concern themselves with {@link DriverException}s that may be thrown from
* operations they attempt. The {@link CqlTemplate} class will catch and handle {@link DriverException} appropriately.
*
* @author David Webb
* @author Mark Paluch
* @see CqlTemplate#query(String, PreparedStatementBinder, ResultSetExtractor)
* @see ReactiveCqlTemplate#query(String, PreparedStatementBinder, ReactiveResultSetExtractor)
*/
public interface PreparedStatementBinder {
/**
* Bind parameter values on the given {@link PreparedStatement}.
*
* @param ps the PreparedStatement to invoke setter methods on
* @throws DriverException if a {@link DriverException} is encountered (i.e. there is no need to catch
* {@link DriverException})
*/
BoundStatement bindValues(PreparedStatement ps) throws DriverException;
}

View File

@@ -0,0 +1,709 @@
/*
* Copyright 2016 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.cassandra.core;
import java.util.Map;
import org.reactivestreams.Publisher;
import org.springframework.dao.DataAccessException;
import org.springframework.dao.IncorrectResultSizeDataAccessException;
import com.datastax.driver.core.PreparedStatement;
import com.datastax.driver.core.Row;
import com.datastax.driver.core.Statement;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
/**
* Interface specifying a basic set of CQL operations executed in a reactive fashion. Implemented by
* {@link ReactiveCqlTemplate}. Not often used directly, but a useful option to enhance testability, as it can easily be
* mocked or stubbed.
*
* @author Mark Paluch
* @since 2.0
* @see ReactiveCqlTemplate
* @see Mono
* @see Flux
*/
public interface ReactiveCqlOperations {
// -------------------------------------------------------------------------
// Methods dealing with a plain ReactiveSession
// -------------------------------------------------------------------------
/**
* Execute a CQL data access operation, implemented as callback action working on a {@link ReactiveSession}. This
* allows for implementing arbitrary data access operations, within Spring's managed CQL environment: that is,
* converting CQL {@link com.datastax.driver.core.exceptions.DriverException}s into Spring's
* {@link DataAccessException} hierarchy.
* <p>
* The callback action can return a result object, for example a domain object or a collection of domain objects.
*
* @param action the callback object that specifies the action.
* @return a result object returned by the action, or {@literal null}.
* @throws DataAccessException if there is any problem executing the query.
*/
<T> Flux<T> execute(ReactiveSessionCallback<T> action) throws DataAccessException;
// -------------------------------------------------------------------------
// Methods dealing with static CQL
// -------------------------------------------------------------------------
/**
* Issue a single CQL execute, typically a DDL statement, insert, update or delete statement.
*
* @param cql static CQL to execute, must not be empty or {@literal null}.
* @return boolean value whether the statement was applied.
* @throws DataAccessException if there is any problem executing the query.
*/
Mono<Boolean> execute(String cql) throws DataAccessException;
/**
* Execute a query given static CQL, reading the {@link ReactiveResultSet} with a {@link ReactiveResultSetExtractor}.
* <p>
* Uses a CQL Statement, not a {@link PreparedStatement}. If you want to execute a static query with a
* {@link PreparedStatement}, use the overloaded {@code query} method with {@literal null} as argument array.
*
* @param cql static CQL to execute, must not be empty or {@literal null}.
* @param rse object that will extract all rows of results, must not be {@literal null}.
* @return an arbitrary result object, as returned by the ReactiveResultSetExtractor.
* @throws DataAccessException if there is any problem executing the query.
* @see #query(String, ReactiveResultSetExtractor, Object...)
*/
<T> Flux<T> query(String cql, ReactiveResultSetExtractor<T> rse) throws DataAccessException;
/**
* Execute a query given static CQL, mapping each row to a Java object via a {@link RowMapper}.
* <p>
* Uses a CQL Statement, not a {@link PreparedStatement}. If you want to execute a static query with a
* {@link PreparedStatement}, use the overloaded {@code query} method with {@literal null} as argument array.
*
* @param cql static CQL to execute, must not be empty or {@literal null}.
* @param rowMapper object that will map one object per row, must not be {@literal null}.
* @return the result {@link Flux}, containing mapped objects.
* @throws DataAccessException if there is any problem executing the query
* @see #query(String, RowMapper, Object[])
*/
<T> Flux<T> query(String cql, RowMapper<T> rowMapper) throws DataAccessException;
/**
* Execute a query given static CQL, mapping a single result row to a Java object via a {@link RowMapper}.
* <p>
* Uses a CQL Statement, not a {@link PreparedStatement}. If you want to execute a static query with a
* {@link PreparedStatement}, use the overloaded {@link #queryForObject(String, RowMapper, Object...)} method with
* {@literal null} as argument array.
*
* @param cql static CQL to execute, must not be empty or {@literal null}.
* @param rowMapper object that will map one object per row, must not be {@literal null}.
* @return the single mapped object.
* @throws IncorrectResultSizeDataAccessException if the query does not return exactly one row.
* @throws DataAccessException if there is any problem executing the query.
* @see #queryForObject(String, RowMapper, Object[])
*/
<T> Mono<T> queryForObject(String cql, RowMapper<T> rowMapper) throws DataAccessException;
/**
* Execute a query for a result object, given static CQL.
* <p>
* Uses a CQL Statement, not a {@link PreparedStatement}. If you want to execute a static query with a
* {@link PreparedStatement}, use the overloaded {@link #queryForObject(String, Class, Object...)} method with
* {@literal null} as argument array.
* <p>
* This method is useful for running static CQL with a known outcome. The query is expected to be a single row/single
* column query; the returned result will be directly mapped to the corresponding object type.
*
* @param cql static CQL to execute, must not be empty or {@literal null}.
* @param requiredType the type that the result object is expected to match, must not be {@literal null}.
* @return the result object of the required type, or {@link Mono#empty()} in case of CQL NULL.
* @throws IncorrectResultSizeDataAccessException if the query does not return exactly one row, or does not return
* exactly one column in that row.
* @throws DataAccessException if there is any problem executing the query.
* @see #queryForObject(String, Class, Object[])
*/
<T> Mono<T> queryForObject(String cql, Class<T> requiredType) throws DataAccessException;
/**
* Execute a query for a result Map, given static CQL.
* <p>
* Uses a CQL Statement, not a {@link PreparedStatement}. If you want to execute a static query with a
* {@link PreparedStatement}, use the overloaded {@link #queryForMap(String, Object...)} method with {@literal null}
* as argument array.
* <p>
* The query is expected to be a single row query; the result row will be mapped to a Map (one entry for each column,
* using the column name as the key).
*
* @param cql static CQL to execute, must not be empty or {@literal null}.
* @return the result Map (one entry for each column, using the column name as the key), must not be {@literal null}.
* @throws IncorrectResultSizeDataAccessException if the query does not return exactly one row.
* @throws DataAccessException if there is any problem executing the query.
* @see #queryForMap(String, Object[])
* @see ColumnMapRowMapper
*/
Mono<Map<String, Object>> queryForMap(String cql) throws DataAccessException;
/**
* Execute a query for a result {@link Flux}, given static CQL.
* <p>
* Uses a CQL Statement, not a {@link PreparedStatement}. If you want to execute a static query with a
* {@link PreparedStatement}, use the overloaded {@code queryForFlux} method with {@literal null} as argument array.
* <p>
* The results will be mapped to a {@link Flux} (one item for each row) of result objects, each of them matching the
* specified element type.
*
* @param cql static CQL to execute, must not be empty or {@literal null}.
* @param elementType the required type of element in the result {@link Flux} (for example, {@code Integer.class}),
* must not be {@literal null}.
* @return a {@link Flux} of objects that match the specified element type.
* @throws DataAccessException if there is any problem executing the query.
* @see #queryForFlux(String, Class, Object[])
* @see SingleColumnRowMapper
*/
<T> Flux<T> queryForFlux(String cql, Class<T> elementType) throws DataAccessException;
/**
* Execute a query for a result {@link Flux}, given static CQL.
* <p>
* Uses a CQL Statement, not a {@link PreparedStatement}. If you want to execute a static query with a
* {@link PreparedStatement}, use the overloaded {@code queryForFlux} method with {@literal null} as argument array.
* <p>
* The results will be mapped to a {@link Flux} (one item for each row) of {@link Map}s (one entry for each column
* using the column name as the key). Each item in the {@link Flux} will be of the form returned by this interface's
* queryForMap() methods.
*
* @param cql static CQL to execute, must not be empty or {@literal null}.
* @return a {@link Flux} that contains a {@link Map} per row.
* @throws DataAccessException if there is any problem executing the query.
* @see #queryForFlux(String, Object[])
*/
Flux<Map<String, Object>> queryForFlux(String cql) throws DataAccessException;
/**
* Execute a query for a ResultSet, given static CQL.
* <p>
* Uses a CQL Statement, not a {@link PreparedStatement}. If you want to execute a static query with a
* {@link PreparedStatement}, use the overloaded {@code queryForResultSet} method with {@literal null} as argument
* array.
* <p>
* The results will be mapped to an {@link ReactiveResultSet}.
*
* @param cql static CQL to execute, must not be empty or {@literal null}.
* @return a {@link ReactiveResultSet} representation.
* @throws DataAccessException if there is any problem executing the query.
* @see #queryForResultSet(String, Object[])
*/
Mono<ReactiveResultSet> queryForResultSet(String cql) throws DataAccessException;
/**
* Execute a query for Rows, given static CQL.
* <p>
* Uses a CQL Statement, not a {@link PreparedStatement}. If you want to execute a static query with a
* {@link PreparedStatement}, use the overloaded {@code queryForResultSet} method with {@literal null} as argument
* array.
* <p>
* The results will be mapped to {@link Row}s.
*
* @param cql static CQL to execute, must not be empty or {@literal null}.
* @return a Row representation.
* @throws DataAccessException if there is any problem executing the query.
* @see #queryForResultSet(String, Object[])
*/
Flux<Row> queryForRows(String cql) throws DataAccessException;
/**
* Issue multiple CQL statements from a CQL statement {@link Publisher}.
*
* @param statementPublisher defining a {@link Publisher} of CQL statements that will be executed.
* @return an array of the number of rows affected by each statement
* @throws DataAccessException if there is any problem executing the batch
*/
Flux<Boolean> execute(Publisher<String> statementPublisher) throws DataAccessException;
// -------------------------------------------------------------------------
// Methods dealing with com.datastax.driver.core.Statement
// -------------------------------------------------------------------------
/**
* Issue a single CQL execute, typically a DDL statement, insert, update or delete statement.
*
* @param statement static CQL {@link Statement}, must not be {@literal null}.
* @return boolean value whether the statement was applied.
* @throws DataAccessException if there is any problem executing the query.
*/
Mono<Boolean> execute(Statement statement) throws DataAccessException;
/**
* Execute a query given static CQL, reading the {@link ReactiveResultSet} with a {@link ReactiveResultSetExtractor}.
* <p>
* Uses a CQL Statement, not a {@link PreparedStatement}. If you want to execute a static query with a
* {@link PreparedStatement}, use the overloaded {@code query} method with {@literal null} as argument array.
*
* @param statement static CQL {@link Statement}, must not be {@literal null}.
* @param rse object that will extract all rows of results, must not be {@literal null}.
* @return an arbitrary result object, as returned by the ReactiveResultSetExtractor.
* @throws DataAccessException if there is any problem executing the query.
* @see #query(String, ReactiveResultSetExtractor, Object...)
*/
<T> Flux<T> query(Statement statement, ReactiveResultSetExtractor<T> rse) throws DataAccessException;
/**
* Execute a query given static CQL, mapping each row to a Java object via a {@link RowMapper}.
* <p>
* Uses a CQL Statement, not a {@link PreparedStatement}. If you want to execute a static query with a
* {@link PreparedStatement}, use the overloaded {@code query} method with {@literal null} as argument array.
*
* @param statement static CQL {@link Statement}, must not be {@literal null}.
* @param rowMapper object that will map one object per row, must not be {@literal null}.
* @return the result {@link Flux}, containing mapped objects.
* @throws DataAccessException if there is any problem executing the query
* @see #query(String, RowMapper, Object[])
*/
<T> Flux<T> query(Statement statement, RowMapper<T> rowMapper) throws DataAccessException;
/**
* Execute a query given static CQL, mapping a single result row to a Java object via a {@link RowMapper}.
* <p>
* Uses a CQL Statement, not a {@link PreparedStatement}. If you want to execute a static query with a
* {@link PreparedStatement}, use the overloaded {@link #queryForObject(String, RowMapper, Object...)} method with
* {@literal null} as argument array.
*
* @param statement static CQL {@link Statement}, must not be {@literal null}.
* @param rowMapper object that will map one object per row, must not be {@literal null}.
* @return the single mapped object.
* @throws IncorrectResultSizeDataAccessException if the query does not return exactly one row.
* @throws DataAccessException if there is any problem executing the query.
* @see #queryForObject(String, RowMapper, Object[])
*/
<T> Mono<T> queryForObject(Statement statement, RowMapper<T> rowMapper) throws DataAccessException;
/**
* Execute a query for a result object, given static CQL.
* <p>
* Uses a CQL Statement, not a {@link PreparedStatement}. If you want to execute a static query with a
* {@link PreparedStatement}, use the overloaded {@link #queryForObject(String, Class, Object...)} method with
* {@literal null} as argument array.
* <p>
* This method is useful for running static CQL with a known outcome. The query is expected to be a single row/single
* column query; the returned result will be directly mapped to the corresponding object type.
*
* @param statement static CQL {@link Statement}, must not be {@literal null}.
* @param requiredType the type that the result object is expected to match, must not be {@literal null}.
* @return the result object of the required type, or {@link Mono#empty()} in case of CQL NULL.
* @throws IncorrectResultSizeDataAccessException if the query does not return exactly one row, or does not return
* exactly one column in that row.
* @throws DataAccessException if there is any problem executing the query.
* @see #queryForObject(String, Class, Object[])
*/
<T> Mono<T> queryForObject(Statement statement, Class<T> requiredType) throws DataAccessException;
/**
* Execute a query for a result Map, given static CQL.
* <p>
* Uses a CQL Statement, not a {@link PreparedStatement}. If you want to execute a static query with a
* {@link PreparedStatement}, use the overloaded {@link #queryForMap(String, Object...)} method with {@literal null}
* as argument array.
* <p>
* The query is expected to be a single row query; the result row will be mapped to a Map (one entry for each column,
* using the column name as the key).
*
* @param statement static CQL {@link Statement}, must not be {@literal null}.
* @return the result Map (one entry for each column, using the column name as the key), must not be {@literal null}.
* @throws IncorrectResultSizeDataAccessException if the query does not return exactly one row.
* @throws DataAccessException if there is any problem executing the query.
* @see #queryForMap(String, Object[])
* @see ColumnMapRowMapper
*/
Mono<Map<String, Object>> queryForMap(Statement statement) throws DataAccessException;
/**
* Execute a query for a result {@link Flux}, given static CQL.
* <p>
* Uses a CQL Statement, not a {@link PreparedStatement}. If you want to execute a static query with a
* {@link PreparedStatement}, use the overloaded {@code queryForFlux} method with {@literal null} as argument array.
* <p>
* The results will be mapped to a {@link Flux} (one item for each row) of result objects, each of them matching the
* specified element type.
*
* @param statement static CQL {@link Statement}, must not be {@literal null}.
* @param elementType the required type of element in the result {@link Flux} (for example, {@code Integer.class}),
* must not be {@literal null}.
* @return a {@link Flux} of objects that match the specified element type.
* @throws DataAccessException if there is any problem executing the query.
* @see #queryForFlux(String, Class, Object[])
* @see SingleColumnRowMapper
*/
<T> Flux<T> queryForFlux(Statement statement, Class<T> elementType) throws DataAccessException;
/**
* Execute a query for a result {@link Flux}, given static CQL.
* <p>
* Uses a CQL Statement, not a {@link PreparedStatement}. If you want to execute a static query with a
* {@link PreparedStatement}, use the overloaded {@code queryForFlux} method with {@literal null} as argument array.
* <p>
* The results will be mapped to a {@link Flux} (one item for each row) of {@link Map}s (one entry for each column
* using the column name as the key). Each item in the {@link Flux} will be of the form returned by this interface's
* queryForMap() methods.
*
* @param statement static CQL {@link Statement}, must not be {@literal null}.
* @return a {@link Flux} that contains a {@link Map} per row.
* @throws DataAccessException if there is any problem executing the query.
* @see #queryForFlux(String, Object[])
*/
Flux<Map<String, Object>> queryForFlux(Statement statement) throws DataAccessException;
/**
* Execute a query for a ResultSet, given static CQL.
* <p>
* Uses a CQL Statement, not a {@link PreparedStatement}. If you want to execute a static query with a
* {@link PreparedStatement}, use the overloaded {@code queryForResultSet} method with {@literal null} as argument
* array.
* <p>
* The results will be mapped to an {@link ReactiveResultSet}.
*
* @param statement static CQL {@link Statement}, must not be {@literal null}.
* @return a {@link ReactiveResultSet} representation.
* @throws DataAccessException if there is any problem executing the query.
* @see #queryForResultSet(String, Object[])
*/
Mono<ReactiveResultSet> queryForResultSet(Statement statement) throws DataAccessException;
/**
* Execute a query for Rows, given static CQL.
* <p>
* Uses a CQL Statement, not a {@link PreparedStatement}. If you want to execute a static query with a
* {@link PreparedStatement}, use the overloaded {@code queryForResultSet} method with {@literal null} as argument
* array.
* <p>
* The results will be mapped to {@link Row}s.
*
* @param statement static CQL {@link Statement}, must not be {@literal null}.
* @return a Row representation.
* @throws DataAccessException if there is any problem executing the query.
* @see #queryForResultSet(String, Object[])
*/
Flux<Row> queryForRows(Statement statement) throws DataAccessException;
// -------------------------------------------------------------------------
// Methods dealing with prepared statements
// -------------------------------------------------------------------------
/**
* Execute a CQL data access operation, implemented as callback action working on a CQL {@link PreparedStatement}.
* This allows for implementing arbitrary data access operations on a single {@link PreparedStatement}, within
* Spring's managed CQL environment: that is, participating in Spring-managed transactions and converting CQL
* {@link com.datastax.driver.core.exceptions.DriverException}s into Spring's {@link DataAccessException} hierarchy.
* <p>
* The callback action can return a result object, for example a domain object or a collection of domain objects.
*
* @param psc object that can create a {@link PreparedStatement} given a {@link ReactiveSession}, must not be
* {@literal null}.
* @param action callback object that specifies the action, must not be {@literal null}.
* @return a result object returned by the action, or {@literal null}.
* @throws DataAccessException if there is any problem
*/
<T> Flux<T> execute(ReactivePreparedStatementCreator psc, ReactivePreparedStatementCallback<T> action)
throws DataAccessException;
/**
* Execute a CQL data access operation, implemented as callback action working on a CQL {@link PreparedStatement}.
* This allows for implementing arbitrary data access operations on a single Statement, within Spring's managed CQL
* environment: that is, participating in Spring-managed transactions and converting CQL
* {@link com.datastax.driver.core.exceptions.DriverException}s into Spring's {@link DataAccessException} hierarchy.
* <p>
* The callback action can return a result object, for example a domain object or a collection of domain objects.
*
* @param cql static CQL to execute, must not be empty or {@literal null}.
* @param action callback object that specifies the action, must not be {@literal null}.
* @return a result object returned by the action, or {@literal null}
* @throws DataAccessException if there is any problem
*/
<T> Flux<T> execute(String cql, ReactivePreparedStatementCallback<T> action) throws DataAccessException;
/**
* Query using a prepared statement, reading the {@link ReactiveResultSet} with a {@link ReactiveResultSetExtractor}.
*
* @param psc object that can create a {@link PreparedStatement} given a {@link ReactiveSession}, must not be
* {@literal null}.
* @param rse object that will extract results, must not be {@literal null}.
* @return an arbitrary result object, as returned by the {@link ReactiveResultSetExtractor}
* @throws DataAccessException if there is any problem
*/
<T> Flux<T> query(ReactivePreparedStatementCreator psc, ReactiveResultSetExtractor<T> rse) throws DataAccessException;
/**
* Query using a prepared statement, reading the {@link ReactiveResultSet} with a {@link ReactiveResultSetExtractor}.
*
* @param cql static CQL to execute, must not be empty or {@literal null}.
* @param psb object that knows how to set values on the prepared statement. If this is {@literal null}, the CQL will
* be assumed to contain no bind parameters. Even if there are no bind parameters, this object may be used to
* set fetch size and other performance options.
* @param rse object that will extract results, must not be {@literal null}.
* @return an arbitrary result object, as returned by the {@link ReactiveResultSetExtractor}.
* @throws DataAccessException if there is any problem
*/
<T> Flux<T> query(String cql, PreparedStatementBinder psb, ReactiveResultSetExtractor<T> rse)
throws DataAccessException;
/**
* Query using a prepared statement and a {@link PreparedStatementBinder} implementation that knows how to bind values
* to the query, reading the {@link ReactiveResultSet} with a {@link ResultSetExtractor}.
*
* @param psc object that can create a {@link PreparedStatement} given a {@link com.datastax.driver.core.Session},
* must not be {@literal null}.
* @param psb object that knows how to set values on the prepared statement. If this is {@literal null}, the CQL will
* be assumed to contain no bind parameters. Even if there are no bind parameters, this object may be used to
* set fetch size and other performance options.
* @param rse object that will extract results, must not be {@literal null}.
* @return an arbitrary result object, as returned by the {@link ResultSetExtractor}.
* @throws DataAccessException if there is any problem
*/
<T> Flux<T> query(ReactivePreparedStatementCreator psc, PreparedStatementBinder psb,
ReactiveResultSetExtractor<T> rse) throws DataAccessException;
/**
* Query given CQL to create a prepared statement from CQL and a list of arguments to bind to the query, reading the
* {@link ReactiveResultSet} with a {@link ReactiveResultSetExtractor}.
*
* @param cql static CQL to execute, must not be empty or {@literal null}.
* @param rse object that will extract results, must not be {@literal null}.
* @param args arguments to bind to the query (leaving it to the {@link PreparedStatement} to guess the corresponding
* CQL type).
* @return an arbitrary result object, as returned by the {@link ReactiveResultSetExtractor}
* @throws DataAccessException if there is any problem executing the query.
*/
<T> Flux<T> query(String cql, ReactiveResultSetExtractor<T> rse, Object... args) throws DataAccessException;
/**
* Query using a prepared statement, mapping each row to a Java object via a {@link RowMapper}.
*
* @param psc object that can create a {@link PreparedStatement} given a {@link ReactiveSession}, must not be
* {@literal null}.
* @param rowMapper object that will map one object per row, must not be {@literal null}.
* @return the result {@link Flux}, containing mapped objects.
* @throws DataAccessException if there is any problem executing the query.
*/
<T> Flux<T> query(ReactivePreparedStatementCreator psc, RowMapper<T> rowMapper) throws DataAccessException;
/**
* Query given CQL to create a prepared statement from CQL and a {@link PreparedStatement}Binder implementation that
* knows how to bind values to the query, mapping each row to a Java object via a {@link RowMapper}.
*
* @param cql static CQL to execute, must not be empty or {@literal null}.
* @param psb object that knows how to set values on the prepared statement. If this is {@literal null}, the CQL will
* be assumed to contain no bind parameters. Even if there are no bind parameters, this object may be used to
* set fetch size and other performance options.
* @param rowMapper object that will map one object per row, must not be {@literal null}.
* @return the result {@link Flux}, containing mapped objects.
* @throws DataAccessException if there is any problem executing the query.
*/
<T> Flux<T> query(String cql, PreparedStatementBinder psb, RowMapper<T> rowMapper) throws DataAccessException;
/**
* Query using a prepared statement and a {@link PreparedStatementBinder} implementation that knows how to bind values
* to the query, mapping each row to a Java object via a {@link RowMapper}.
*
* @param psc object that can create a {@link PreparedStatement} given a {@link com.datastax.driver.core.Session},
* must not be {@literal null}.
* @param psb object that knows how to set values on the prepared statement. If this is {@literal null}, the CQL will
* be assumed to contain no bind parameters. Even if there are no bind parameters, this object may be used to
* set fetch size and other performance options.
* @param rowMapper object that will map one object per row, must not be {@literal null}.
* @return the result {@link Flux}, containing mapped objects.
* @throws DataAccessException if there is any problem executing the query.
*/
<T> Flux<T> query(ReactivePreparedStatementCreator psc, PreparedStatementBinder psb, RowMapper<T> rowMapper)
throws DataAccessException;
/**
* Query given CQL to create a prepared statement from CQL and a list of arguments to bind to the query, mapping each
* row to a Java object via a {@link RowMapper}.
*
* @param cql static CQL to execute, must not be empty or {@literal null}.
* @param rowMapper object that will map one object per row
* @param args arguments to bind to the query (leaving it to the {@link PreparedStatement} to guess the corresponding
* CQL type)
* @return the result {@link Flux}, containing mapped objects
* @throws DataAccessException if there is any problem executing the query.
*/
<T> Flux<T> query(String cql, RowMapper<T> rowMapper, Object... args) throws DataAccessException;
/**
* Query given CQL to create a prepared statement from CQL and a list of arguments to bind to the query, mapping a
* single result row to a Java object via a {@link RowMapper}.
*
* @param cql static CQL to execute, must not be empty or {@literal null}.
* @param rowMapper object that will map one object per row, must not be {@literal null}.
* @param args arguments to bind to the query (leaving it to the {@link PreparedStatement} to guess the corresponding
* CQL type)
* @return the single mapped object
* @throws IncorrectResultSizeDataAccessException if the query does not return exactly one row.
* @throws DataAccessException if there is any problem executing the query.
*/
<T> Mono<T> queryForObject(String cql, RowMapper<T> rowMapper, Object... args) throws DataAccessException;
/**
* Query given CQL to create a prepared statement from CQL and a list of arguments to bind to the query, expecting a
* result object.
* <p>
* The query is expected to be a single row/single column query; the returned result will be directly mapped to the
* corresponding object type.
*
* @param cql static CQL to execute, must not be empty or {@literal null}.
* @param requiredType the type that the result object is expected to match, must not be {@literal null}.
* @param args arguments to bind to the query (leaving it to the PreparedStatement to guess the corresponding CQL
* type)
* @return the result object of the required type, or {@link Mono#empty()} in case of CQL NULL.
* @throws IncorrectResultSizeDataAccessException if the query does not return exactly one row, or does not return
* exactly one column in that row.
* @throws DataAccessException if there is any problem executing the query.
* @see #queryForObject(String, Class)
*/
<T> Mono<T> queryForObject(String cql, Class<T> requiredType, Object... args) throws DataAccessException;
/**
* Query given CQL to create a prepared statement from CQL and a list of arguments to bind to the query, expecting a
* result Map. The queryForMap() methods defined by this interface are appropriate when you don't have a domain model.
* Otherwise, consider using one of the queryForObject() methods.
* <p>
* The query is expected to be a single row query; the result row will be mapped to a Map (one entry for each column,
* using the column name as the key).
*
* @param cql static CQL to execute, must not be empty or {@literal null}.
* @param args arguments to bind to the query (leaving it to the {@link PreparedStatement} to guess the corresponding
* CQL type).
* @return the result Map (one entry for each column, using the column name as the key).
* @throws IncorrectResultSizeDataAccessException if the query does not return exactly one row
* @throws DataAccessException if there is any problem executing the query.
* @see #queryForMap(String)
* @see ColumnMapRowMapper
*/
Mono<Map<String, Object>> queryForMap(String cql, Object... args) throws DataAccessException;
/**
* Query given CQL to create a prepared statement from CQL and a list of arguments to bind to the query, expecting a
* result {@link Flux}.
* <p>
* The results will be mapped to a {@link Flux} (one item for each row) of result objects, each of them matching the
* specified element type.
*
* @param cql static CQL to execute, must not be empty or {@literal null}.
* @param elementType the required type of element in the result {@link Flux} (for example, {@code Integer.class}),
* must not be {@literal null}.
* @param args arguments to bind to the query (leaving it to the {@link PreparedStatement} to guess the corresponding
* CQL type).
* @return a {@link Flux} of objects that match the specified element type.
* @throws DataAccessException if there is any problem executing the query.
* @see #queryForFlux(String, Class)
* @see SingleColumnRowMapper
*/
<T> Flux<T> queryForFlux(String cql, Class<T> elementType, Object... args) throws DataAccessException;
/**
* Query given CQL to create a prepared statement from CQL and a list of arguments to bind to the query, expecting a
* result {@link Flux}.
* <p>
* The results will be mapped to a {@link Flux} (one item for each row) of {@link Map}s (one entry for each column,
* using the column name as the key). Each item in the {@link Flux} will be of the form returned by this interface's
* queryForMap() methods.
*
* @param cql static CQL to execute, must not be empty or {@literal null}.
* @param args arguments to bind to the query (leaving it to the {@link PreparedStatement} to guess the corresponding
* CQL type).
* @return a {@link Flux} that contains a {@link Map} per row
* @throws DataAccessException if there is any problem executing the query.
* @see #queryForFlux(String)
*/
Flux<Map<String, Object>> queryForFlux(String cql, Object... args) throws DataAccessException;
/**
* Query given CQL to create a prepared statement from CQL and a list of arguments to bind to the query, expecting a
* ResultSet.
* <p>
* The results will be mapped to an {@link ReactiveResultSet}.
*
* @param cql static CQL to execute, must not be empty or {@literal null}.
* @param args arguments to bind to the query (leaving it to the {@link PreparedStatement} to guess the corresponding
* CQL type).
* @return a {@link ReactiveResultSet} representation.
* @throws DataAccessException if there is any problem executing the query.
* @see #queryForResultSet(String)
*/
Mono<ReactiveResultSet> queryForResultSet(String cql, Object... args) throws DataAccessException;
/**
* Query given CQL to create a prepared statement from CQL and a list of arguments to bind to the query, expecting
* Rows.
* <p>
* The results will be mapped to {@link Row}s.
*
* @param cql static CQL to execute, must not be empty or {@literal null}.
* @param args arguments to bind to the query (leaving it to the {@link PreparedStatement} to guess the corresponding
* CQL type).
* @return a {@link Row} representation.
* @throws DataAccessException if there is any problem executing the query.
* @see #queryForResultSet(String)
*/
Flux<Row> queryForRows(String cql, Object... args) throws DataAccessException;
/**
* Issue a single CQL execute operation (such as an insert, update or delete statement) using a
* {@link ReactivePreparedStatementCreator} to provide CQL and any required parameters.
*
* @param psc object that provides CQL and any necessary parameters, must not be {@literal null}.
* @return boolean value whether the statement was applied.
* @throws DataAccessException if there is any problem issuing the execution.
*/
// TODO: Interferes with execute(session callback lambda)
Mono<Boolean> execute(ReactivePreparedStatementCreator psc) throws DataAccessException;
/**
* Issue an statement using a {@link PreparedStatementBinder} to set bind parameters, with given CQL. Simpler than
* using a {@link ReactivePreparedStatementCreator} as this method will create the {@link PreparedStatement}: The
* {@link PreparedStatementBinder} just needs to set parameters.
*
* @param cql static CQL to execute, must not be empty or {@literal null}.
* @param psb object that knows how to set values on the prepared statement. If this is {@literal null}, the CQL will
* be assumed to contain no bind parameters. Even if there are no bind parameters, this object may be used to
* set fetch size and other performance options.
* @return boolean value whether the statement was applied.
* @throws DataAccessException if there is any problem issuing the execution.
*/
Mono<Boolean> execute(String cql, PreparedStatementBinder psb) throws DataAccessException;
/**
* Issue a single CQL operation (such as an insert, update or delete statement) via a prepared statement, binding the
* given arguments.
*
* @param cql static CQL to execute, must not be empty or {@literal null}.
* @param args arguments to bind to the query (leaving it to the {@link PreparedStatement} to guess the corresponding
* CQL type).
* @return boolean value whether the statement was applied.
* @throws DataAccessException if there is any problem issuing the execution.
*/
Mono<Boolean> execute(String cql, Object... args) throws DataAccessException;
/**
* Issue a single CQL operation (such as an insert, update or delete statement) via a prepared statement, binding the
* given arguments.
*
* @param cql static CQL to execute containing bind parameters, must not be empty or {@literal null}.
* @param args arguments to bind to the query (leaving it to the {@link PreparedStatement} to guess the corresponding
* CQL type).
* @return boolean value whether the statement was applied.
* @throws DataAccessException if there is any problem issuing the execution.
*/
Flux<Boolean> execute(String cql, Publisher<Object[]> args) throws DataAccessException;
}

View File

@@ -0,0 +1,869 @@
/*
* Copyright 2016 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.cassandra.core;
import java.util.Map;
import java.util.function.Function;
import org.reactivestreams.Publisher;
import org.springframework.cassandra.support.ReactiveCassandraAccessor;
import org.springframework.dao.DataAccessException;
import org.springframework.dao.support.DataAccessUtils;
import org.springframework.util.Assert;
import com.datastax.driver.core.BoundStatement;
import com.datastax.driver.core.ConsistencyLevel;
import com.datastax.driver.core.PreparedStatement;
import com.datastax.driver.core.Row;
import com.datastax.driver.core.SimpleStatement;
import com.datastax.driver.core.Statement;
import com.datastax.driver.core.exceptions.DriverException;
import com.datastax.driver.core.policies.RetryPolicy;
import com.datastax.driver.core.querybuilder.QueryBuilder;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
/**
* <b>This is the central class in the CQL core package for reactive Cassandra data access.</b> It simplifies the use of
* CQL and helps to avoid common errors. It executes core CQL workflow, leaving application code to provide CQL and
* extract results. This class executes CQL queries or updates, initiating iteration over {@link ReactiveResultSet}s and
* catching {@link DriverException} exceptions and translating them to the generic, more informative exception hierarchy
* defined in the {@code org.springframework.dao} package.
* <p>
* Code using this class need only implement callback interfaces, giving them a clearly defined contract. The
* {@link PreparedStatementCreator} callback interface creates a prepared statement given a Connection, providing CQL
* and any necessary parameters. The {@link ResultSetExtractor} interface extracts values from a
* {@link ReactiveResultSet}. See also {@link PreparedStatementBinder} and {@link RowMapper} for two popular alternative
* callback interfaces.
* <p>
* Can be used within a service implementation via direct instantiation with a {@link ReactiveSessionFactory} reference,
* or get prepared in an application context and given to services as bean reference. Note: The
* {@link ReactiveSessionFactory} should always be configured as a bean in the application context, in the first case
* given to the service directly, in the second case to the prepared template.
* <p>
* Because this class is parameterizable by the callback interfaces and the
* {@link org.springframework.dao.support.PersistenceExceptionTranslator} interface, there should be no need to subclass
* it.
* <p>
* All CQL operations performed by this class are logged at debug level, using
* "org.springframework.cassandra.core.ReactiveCqlTemplate" as log category.
* <p>
* <b>NOTE: An instance of this class is thread-safe once configured.</b>
*
* @author Mark Paluch
* @since 2.0
* @see PreparedStatementCreator
* @see PreparedStatementBinder
* @see PreparedStatementCallback
* @see ResultSetExtractor
* @see RowCallbackHandler
* @see RowMapper
* @see org.springframework.dao.support.PersistenceExceptionTranslator
*/
@SuppressWarnings("WeakerAccess")
public class ReactiveCqlTemplate extends ReactiveCassandraAccessor implements ReactiveCqlOperations {
/**
* Placeholder for default values.
*/
private final static Statement DEFAULTS = QueryBuilder.select().from("DEFAULT");
/**
* If this variable is set to a non-negative value, it will be used for setting the {@code fetchSize} property on
* statements used for query processing.
*/
private int fetchSize = -1;
/**
* If this variable is set to a value, it will be used for setting the {@code retryPolicy} property on statements used
* for query processing.
*/
private RetryPolicy retryPolicy;
/**
* If this variable is set to a value, it will be used for setting the {@code consistencyLevel} property on statements
* used for query processing.
*/
private com.datastax.driver.core.ConsistencyLevel consistencyLevel;
/**
* Construct a new {@link ReactiveCqlTemplate Note: The {@link ReactiveSessionFactory} has to be set before using the
* instance.
*
* @see #setSessionFactory
*/
public ReactiveCqlTemplate() {}
/**
* Construct a new {@link ReactiveCqlTemplate}, given a {@link ReactiveSession}.
*
* @param reactiveSession the {@link ReactiveSession}, must not be {@literal null}.
*/
public ReactiveCqlTemplate(ReactiveSession reactiveSession) {
Assert.notNull(reactiveSession, "ReactiveSession must not be null");
setSessionFactory(new DefaultReactiveSessionFactory(reactiveSession));
afterPropertiesSet();
}
/**
* Construct a new {@link ReactiveCqlTemplate}, given a {@link ReactiveSessionFactory} to obtain
* {@link ReactiveSession}s from.
*
* @param reactiveSessionFactory the {@link ReactiveSessionFactory} to obtain {@link ReactiveSession}s from, must not
* be {@literal null}.
*/
public ReactiveCqlTemplate(ReactiveSessionFactory reactiveSessionFactory) {
setSessionFactory(reactiveSessionFactory);
afterPropertiesSet();
}
/**
* Set the fetch size for this {@link ReactiveCqlTemplate}. This is important for processing large result sets:
* Setting this higher than the default value will increase processing speed at the cost of memory consumption;
* setting this lower can avoid transferring row data that will never be read by the application. Default is -1,
* indicating to use the CQL driver's default configuration (i.e. to not pass a specific fetch size setting on to the
* driver).
*
* @see Statement#setFetchSize(int)
*/
public void setFetchSize(int fetchSize) {
this.fetchSize = fetchSize;
}
/**
* @return the fetch size specified for this {@link ReactiveCqlTemplate}.
*/
public int getFetchSize() {
return this.fetchSize;
}
/**
* Set the retry policy for this {@link ReactiveCqlTemplate}. This is important for defining behavior when a request
* fails.
*
* @see Statement#setRetryPolicy(RetryPolicy)
* @see RetryPolicy
*/
public void setRetryPolicy(RetryPolicy retryPolicy) {
this.retryPolicy = retryPolicy;
}
/**
* @return the {@link RetryPolicy} specified for this {@link ReactiveCqlTemplate}.
*/
public RetryPolicy getRetryPolicy() {
return retryPolicy;
}
/**
* Set the consistency level for this {@link ReactiveCqlTemplate}. Consistency level defines the number of nodes
* involved into query processing. Relaxed consistency level settings use fewer nodes but eventual consistency is more
* likely to occur while a higher consistency level involves more nodes to obtain results with a higher consistency
* guarantee.
*
* @see Statement#setConsistencyLevel(ConsistencyLevel)
* @see RetryPolicy
*/
public void setConsistencyLevel(ConsistencyLevel consistencyLevel) {
this.consistencyLevel = consistencyLevel;
}
/**
* @return the {@link ConsistencyLevel} specified for this {@link ReactiveCqlTemplate}.
*/
public ConsistencyLevel getConsistencyLevel() {
return consistencyLevel;
}
// -------------------------------------------------------------------------
// Methods dealing with a plain org.springframework.cassandra.core.ReactiveSession
// -------------------------------------------------------------------------
/* (non-Javadoc)
* @see org.springframework.cassandra.core.ReactiveCqlOperations#execute(org.springframework.cassandra.core.ReactiveSessionCallback)
*/
@Override
public <T> Flux<T> execute(ReactiveSessionCallback<T> action) throws DataAccessException {
Assert.notNull(action, "Callback object must not be null");
return createFlux(action).onErrorResumeWith(translateException("ReactiveSessionCallback", getCql(action)));
}
// -------------------------------------------------------------------------
// Methods dealing with static CQL
// -------------------------------------------------------------------------
/* (non-Javadoc)
* @see org.springframework.cassandra.core.ReactiveCqlOperations#execute(java.lang.String)
*/
@Override
public Mono<Boolean> execute(String cql) throws DataAccessException {
Assert.hasText(cql, "CQL must not be empty");
return queryForResultSet(cql).map(ReactiveResultSet::wasApplied);
}
/* (non-Javadoc)
* @see org.springframework.cassandra.core.ReactiveCqlOperations#query(java.lang.String, org.springframework.cassandra.core.ReactiveResultSetExtractor)
*/
@Override
public <T> Flux<T> query(String cql, ReactiveResultSetExtractor<T> rse) throws DataAccessException {
Assert.hasText(cql, "CQL must not be empty");
Assert.notNull(rse, "ReactiveResultSetExtractor must not be null");
return createFlux(new SimpleStatement(cql), (session, stmt) -> {
if (logger.isDebugEnabled()) {
logger.debug("Executing CQL Statement [{}]", cql);
}
return session.execute(stmt).flatMap(rse::extractData);
}).onErrorResumeWith(translateException("Query", cql));
}
/* (non-Javadoc)
* @see org.springframework.cassandra.core.ReactiveCqlOperations#query(java.lang.String, org.springframework.cassandra.core.RowMapper)
*/
@Override
public <T> Flux<T> query(String cql, RowMapper<T> rowMapper) throws DataAccessException {
return query(cql, new ReactiveRowMapperResultSetExtractor<>(rowMapper));
}
/* (non-Javadoc)
* @see org.springframework.cassandra.core.ReactiveCqlOperations#queryForObject(java.lang.String, org.springframework.cassandra.core.RowMapper)
*/
@Override
public <T> Mono<T> queryForObject(String cql, RowMapper<T> rowMapper) throws DataAccessException {
return query(cql, rowMapper).buffer(2).flatMap(list -> Mono.just(DataAccessUtils.requiredSingleResult(list)))
.next();
}
/* (non-Javadoc)
* @see org.springframework.cassandra.core.ReactiveCqlOperations#queryForObject(java.lang.String, java.lang.Class)
*/
@Override
public <T> Mono<T> queryForObject(String cql, Class<T> requiredType) throws DataAccessException {
return queryForObject(cql, getSingleColumnRowMapper(requiredType));
}
/* (non-Javadoc)
* @see org.springframework.cassandra.core.ReactiveCqlOperations#queryForMap(java.lang.String)
*/
@Override
public Mono<Map<String, Object>> queryForMap(String cql) throws DataAccessException {
return queryForObject(cql, getColumnMapRowMapper());
}
/* (non-Javadoc)
* @see org.springframework.cassandra.core.ReactiveCqlOperations#queryForFlux(java.lang.String, java.lang.Class)
*/
@Override
public <T> Flux<T> queryForFlux(String cql, Class<T> elementType) throws DataAccessException {
return query(cql, getSingleColumnRowMapper(elementType));
}
/* (non-Javadoc)
* @see org.springframework.cassandra.core.ReactiveCqlOperations#queryForFlux(java.lang.String)
*/
@Override
public Flux<Map<String, Object>> queryForFlux(String cql) throws DataAccessException {
return query(cql, getColumnMapRowMapper());
}
/* (non-Javadoc)
* @see org.springframework.cassandra.core.ReactiveCqlOperations#queryForResultSet(java.lang.String)
*/
@Override
public Mono<ReactiveResultSet> queryForResultSet(String cql) throws DataAccessException {
Assert.hasText(cql, "CQL must not be empty");
return createMono(new SimpleStatement(cql), (session, statement) -> {
if (logger.isDebugEnabled()) {
logger.debug("Executing CQL [{}]", cql);
}
return session.execute(statement);
}).otherwise(translateException("QueryForResultSet", cql));
}
/* (non-Javadoc)
* @see org.springframework.cassandra.core.ReactiveCqlOperations#queryForRows(java.lang.String)
*/
@Override
public Flux<Row> queryForRows(String cql) throws DataAccessException {
return queryForResultSet(cql).flatMap(ReactiveResultSet::rows)
.onErrorResumeWith(translateException("QueryForRows", cql));
}
/* (non-Javadoc)
* @see org.springframework.cassandra.core.ReactiveCqlOperations#execute(org.reactivestreams.Publisher)
*/
@Override
public Flux<Boolean> execute(Publisher<String> statementPublisher) throws DataAccessException {
Assert.notNull(statementPublisher, "CQL Publisher must not be null");
return Flux.from(statementPublisher).flatMap(this::execute);
}
// -------------------------------------------------------------------------
// Methods dealing with com.datastax.driver.core.Statement
// -------------------------------------------------------------------------
/* (non-Javadoc)
* @see org.springframework.cassandra.core.ReactiveCqlOperations#execute(com.datastax.driver.core.Statement)
*/
@Override
public Mono<Boolean> execute(Statement statement) throws DataAccessException {
Assert.notNull(statement, "CQL Statement must not be null");
return queryForResultSet(statement).map(ReactiveResultSet::wasApplied);
}
/* (non-Javadoc)
* @see org.springframework.cassandra.core.ReactiveCqlOperations#query(com.datastax.driver.core.Statement, org.springframework.cassandra.core.ReactiveResultSetExtractor)
*/
@Override
public <T> Flux<T> query(Statement statement, ReactiveResultSetExtractor<T> rse) throws DataAccessException {
Assert.notNull(statement, "CQL Statement must not be null");
Assert.notNull(rse, "ReactiveResultSetExtractor must not be null");
return createFlux(statement, (session, stmt) -> {
if (logger.isDebugEnabled()) {
logger.debug("Executing CQL Statement [{}]", statement);
}
return session.execute(stmt).flatMap(rse::extractData);
}).onErrorResumeWith(translateException("Query", statement.toString()));
}
/* (non-Javadoc)
* @see org.springframework.cassandra.core.ReactiveCqlOperations#query(com.datastax.driver.core.Statement, org.springframework.cassandra.core.RowMapper)
*/
@Override
public <T> Flux<T> query(Statement statement, RowMapper<T> rowMapper) throws DataAccessException {
return query(statement, new ReactiveRowMapperResultSetExtractor<>(rowMapper));
}
/* (non-Javadoc)
* @see org.springframework.cassandra.core.ReactiveCqlOperations#queryForObject(com.datastax.driver.core.Statement, org.springframework.cassandra.core.RowMapper)
*/
@Override
public <T> Mono<T> queryForObject(Statement statement, RowMapper<T> rowMapper) throws DataAccessException {
return query(statement, rowMapper).buffer(2).flatMap(list -> Mono.just(DataAccessUtils.requiredSingleResult(list)))
.next();
}
/* (non-Javadoc)
* @see org.springframework.cassandra.core.ReactiveCqlOperations#queryForObject(com.datastax.driver.core.Statement, java.lang.Class)
*/
@Override
public <T> Mono<T> queryForObject(Statement statement, Class<T> requiredType) throws DataAccessException {
return queryForObject(statement, getSingleColumnRowMapper(requiredType));
}
/* (non-Javadoc)
* @see org.springframework.cassandra.core.ReactiveCqlOperations#queryForMap(com.datastax.driver.core.Statement)
*/
@Override
public Mono<Map<String, Object>> queryForMap(Statement statement) throws DataAccessException {
return queryForObject(statement, getColumnMapRowMapper());
}
/* (non-Javadoc)
* @see org.springframework.cassandra.core.ReactiveCqlOperations#queryForFlux(com.datastax.driver.core.Statement, java.lang.Class)
*/
@Override
public <T> Flux<T> queryForFlux(Statement statement, Class<T> elementType) throws DataAccessException {
return query(statement, getSingleColumnRowMapper(elementType));
}
/* (non-Javadoc)
* @see org.springframework.cassandra.core.ReactiveCqlOperations#queryForFlux(com.datastax.driver.core.Statement)
*/
@Override
public Flux<Map<String, Object>> queryForFlux(Statement statement) throws DataAccessException {
return query(statement, getColumnMapRowMapper());
}
/* (non-Javadoc)
* @see org.springframework.cassandra.core.ReactiveCqlOperations#queryForResultSet(com.datastax.driver.core.Statement)
*/
@Override
public Mono<ReactiveResultSet> queryForResultSet(Statement statement) throws DataAccessException {
Assert.notNull(statement, "CQL Statement must not be null");
return createMono(statement, (session, executedStatement) -> {
if (logger.isDebugEnabled()) {
logger.debug("Executing CQL [{}]", executedStatement);
}
return session.execute(executedStatement);
}).otherwise(translateException("QueryForResultSet", statement.toString()));
}
@Override
public Flux<Row> queryForRows(Statement statement) throws DataAccessException {
return queryForResultSet(statement).flatMap(ReactiveResultSet::rows)
.onErrorResumeWith(translateException("QueryForRows", statement.toString()));
}
// -------------------------------------------------------------------------
// Methods dealing with prepared statements
// -------------------------------------------------------------------------
/* (non-Javadoc)
* @see org.springframework.cassandra.core.ReactiveCqlOperations#execute(org.springframework.cassandra.core.ReactivePreparedStatementCreator, org.springframework.cassandra.core.ReactivePreparedStatementCallback)
*/
@Override
public <T> Flux<T> execute(ReactivePreparedStatementCreator psc, ReactivePreparedStatementCallback<T> action)
throws DataAccessException {
Assert.notNull(psc, "ReactivePreparedStatementCreator must not be null");
Assert.notNull(action, "ReactivePreparedStatementCallback object must not be null");
return createFlux(session -> {
logger.debug("Preparing statement [{}] using {}", getCql(psc), psc);
return psc.createPreparedStatement(session).doOnNext(this::applyStatementSettings)
.flatMap(ps -> action.doInPreparedStatement(session, ps));
}).onErrorResumeWith(translateException("ReactivePreparedStatementCallback", getCql(psc)));
}
/* (non-Javadoc)
* @see org.springframework.cassandra.core.ReactiveCqlOperations#execute(java.lang.String, org.springframework.cassandra.core.ReactivePreparedStatementCallback)
*/
@Override
public <T> Flux<T> execute(String cql, ReactivePreparedStatementCallback<T> action) throws DataAccessException {
return execute(new SimpleReactivePreparedStatementCreator(cql), action);
}
/**
* Query using a prepared statement, reading the {@link ReactiveResultSet} with a {@link ReactiveResultSetExtractor}.
*
* @param psc object that can create a {@link PreparedStatement} given a {@link ReactiveSession}
* @param psb object that knows how to set values on the prepared statement. If this is {@literal null}, the CQL will
* be assumed to contain no bind parameters.
* @param rse object that will extract results
* @return an arbitrary result object, as returned by the {@link ReactiveResultSetExtractor}
* @throws DataAccessException if there is any problem
*/
public <T> Flux<T> query(ReactivePreparedStatementCreator psc, PreparedStatementBinder psb,
ReactiveResultSetExtractor<T> rse) throws DataAccessException {
Assert.notNull(psc, "ReactivePreparedStatementCreator must not be null");
Assert.notNull(rse, "ReactiveResultSetExtractor object must not be null");
return execute(psc, (session, ps) -> Mono.just(ps).flatMap(pps -> {
if (logger.isDebugEnabled()) {
logger.debug("Executing Prepared CQL Statement [{}]", ps.getQueryString());
}
BoundStatement boundStatement = psb != null ? psb.bindValues(ps) : ps.bind();
applyStatementSettings(boundStatement);
return session.execute(boundStatement);
}).flatMap(rse::extractData)).onErrorResumeWith(translateException("Query", getCql(psc)));
}
/* (non-Javadoc)
* @see org.springframework.cassandra.core.ReactiveCqlOperations#query(org.springframework.cassandra.core.ReactivePreparedStatementCreator, org.springframework.cassandra.core.ReactiveResultSetExtractor)
*/
@Override
public <T> Flux<T> query(ReactivePreparedStatementCreator psc, ReactiveResultSetExtractor<T> rse)
throws DataAccessException {
return query(psc, null, rse);
}
/* (non-Javadoc)
* @see org.springframework.cassandra.core.ReactiveCqlOperations#query(java.lang.String, org.springframework.cassandra.core.PreparedStatementBinder, org.springframework.cassandra.core.ReactiveResultSetExtractor)
*/
@Override
public <T> Flux<T> query(String cql, PreparedStatementBinder psb, ReactiveResultSetExtractor<T> rse)
throws DataAccessException {
return query(new SimpleReactivePreparedStatementCreator(cql), psb, rse);
}
/* (non-Javadoc)
* @see org.springframework.cassandra.core.ReactiveCqlOperations#query(java.lang.String, org.springframework.cassandra.core.ReactiveResultSetExtractor, java.lang.Object[])
*/
@Override
public <T> Flux<T> query(String cql, ReactiveResultSetExtractor<T> rse, Object... args) throws DataAccessException {
return query(new SimpleReactivePreparedStatementCreator(cql), newArgPreparedStatementBinder(args), rse);
}
/* (non-Javadoc)
* @see org.springframework.cassandra.core.ReactiveCqlOperations#query(org.springframework.cassandra.core.ReactivePreparedStatementCreator, org.springframework.cassandra.core.RowMapper)
*/
@Override
public <T> Flux<T> query(ReactivePreparedStatementCreator psc, RowMapper<T> rowMapper) throws DataAccessException {
return query(psc, null, new ReactiveRowMapperResultSetExtractor<>(rowMapper));
}
/* (non-Javadoc)
* @see org.springframework.cassandra.core.ReactiveCqlOperations#query(java.lang.String, org.springframework.cassandra.core.PreparedStatementBinder, org.springframework.cassandra.core.RowMapper)
*/
@Override
public <T> Flux<T> query(String cql, PreparedStatementBinder psb, RowMapper<T> rowMapper) throws DataAccessException {
return query(cql, psb, new ReactiveRowMapperResultSetExtractor<>(rowMapper));
}
/* (non-Javadoc)
* @see org.springframework.cassandra.core.ReactiveCqlOperations#query(org.springframework.cassandra.core.ReactivePreparedStatementCreator, org.springframework.cassandra.core.PreparedStatementBinder, org.springframework.cassandra.core.RowMapper)
*/
@Override
public <T> Flux<T> query(ReactivePreparedStatementCreator psc, PreparedStatementBinder psb, RowMapper<T> rowMapper)
throws DataAccessException {
return query(psc, psb, new ReactiveRowMapperResultSetExtractor<>(rowMapper));
}
/* (non-Javadoc)
* @see org.springframework.cassandra.core.ReactiveCqlOperations#query(java.lang.String, org.springframework.cassandra.core.RowMapper, java.lang.Object[])
*/
@Override
public <T> Flux<T> query(String cql, RowMapper<T> rowMapper, Object... args) throws DataAccessException {
return query(cql, newArgPreparedStatementBinder(args), rowMapper);
}
/* (non-Javadoc)
* @see org.springframework.cassandra.core.ReactiveCqlOperations#queryForObject(java.lang.String, org.springframework.cassandra.core.RowMapper, java.lang.Object[])
*/
@Override
public <T> Mono<T> queryForObject(String cql, RowMapper<T> rowMapper, Object... args) throws DataAccessException {
return query(cql, rowMapper, args).buffer(2).flatMap(list -> Mono.just(DataAccessUtils.requiredSingleResult(list)))
.next();
}
/* (non-Javadoc)
* @see org.springframework.cassandra.core.ReactiveCqlOperations#queryForObject(java.lang.String, java.lang.Class, java.lang.Object[])
*/
@Override
public <T> Mono<T> queryForObject(String cql, Class<T> requiredType, Object... args) throws DataAccessException {
return queryForObject(cql, getSingleColumnRowMapper(requiredType), args);
}
/* (non-Javadoc)
* @see org.springframework.cassandra.core.ReactiveCqlOperations#queryForMap(java.lang.String, java.lang.Object[])
*/
@Override
public Mono<Map<String, Object>> queryForMap(String cql, Object... args) throws DataAccessException {
return queryForObject(cql, getColumnMapRowMapper(), args);
}
/* (non-Javadoc)
* @see org.springframework.cassandra.core.ReactiveCqlOperations#queryForFlux(java.lang.String, java.lang.Class, java.lang.Object[])
*/
@Override
public <T> Flux<T> queryForFlux(String cql, Class<T> elementType, Object... args) throws DataAccessException {
return query(cql, getSingleColumnRowMapper(elementType), args);
}
/* (non-Javadoc)
* @see org.springframework.cassandra.core.ReactiveCqlOperations#queryForFlux(java.lang.String, java.lang.Object[])
*/
@Override
public Flux<Map<String, Object>> queryForFlux(String cql, Object... args) throws DataAccessException {
return query(cql, getColumnMapRowMapper(), args);
}
/* (non-Javadoc)
* @see org.springframework.cassandra.core.ReactiveCqlOperations#queryForResultSet(java.lang.String, java.lang.Object[])
*/
@Override
public Mono<ReactiveResultSet> queryForResultSet(String cql, Object... args) throws DataAccessException {
Assert.hasText(cql, "CQL must not be empty");
return query(new SimpleReactivePreparedStatementCreator(cql), newArgPreparedStatementBinder(args), Mono::just)
.next();
}
/* (non-Javadoc)
* @see org.springframework.cassandra.core.ReactiveCqlOperations#queryForRows(java.lang.String, java.lang.Object[])
*/
@Override
public Flux<Row> queryForRows(String cql, Object... args) throws DataAccessException {
return queryForResultSet(cql, args).flatMap(ReactiveResultSet::rows)
.onErrorResumeWith(translateException("QueryForRows", cql));
}
/* (non-Javadoc)
* @see org.springframework.cassandra.core.ReactiveCqlOperations#execute(org.springframework.cassandra.core.ReactivePreparedStatementCreator)
*/
@Override
public Mono<Boolean> execute(ReactivePreparedStatementCreator psc) throws DataAccessException {
return query(psc, resultSet -> Mono.just(resultSet.wasApplied())).last();
}
/* (non-Javadoc)
* @see org.springframework.cassandra.core.ReactiveCqlOperations#execute(java.lang.String, org.springframework.cassandra.core.PreparedStatementBinder)
*/
@Override
public Mono<Boolean> execute(String cql, PreparedStatementBinder psb) throws DataAccessException {
return query(new SimpleReactivePreparedStatementCreator(cql), psb, resultSet -> Mono.just(resultSet.wasApplied()))
.next();
}
/* (non-Javadoc)
* @see org.springframework.cassandra.core.ReactiveCqlOperations#execute(java.lang.String, java.lang.Object[])
*/
@Override
public Mono<Boolean> execute(String cql, Object... args) throws DataAccessException {
return execute(cql, newArgPreparedStatementBinder(args));
}
/* (non-Javadoc)
* @see org.springframework.cassandra.core.ReactiveCqlOperations#execute(java.lang.String, org.reactivestreams.Publisher)
*/
@Override
public Flux<Boolean> execute(String cql, Publisher<Object[]> args) throws DataAccessException {
Assert.notNull(args, "Args Publisher must not be null");
SimpleReactivePreparedStatementCreator psc = new SimpleReactivePreparedStatementCreator(cql);
return execute(psc, (session, ps) -> Flux.from(args).flatMap(objects -> {
if (logger.isDebugEnabled()) {
logger.debug("Executing Prepared CQL Statement [{}]", cql);
}
BoundStatement boundStatement = newArgPreparedStatementBinder(objects).bindValues(ps);
applyStatementSettings(boundStatement);
return session.execute(boundStatement);
}).map(ReactiveResultSet::wasApplied));
}
// -------------------------------------------------------------------------
// Implementation hooks and helper methods
// -------------------------------------------------------------------------
/**
* Create a reusable {@link Flux} given a {@link ReactiveStatementCallback} without exception translation.
*
* @param callback must not be {@literal null}.
* @return a reusable {@link Flux} wrapping the {@link ReactiveStatementCallback}.
*/
protected <T> Flux<T> createFlux(Statement statement, ReactiveStatementCallback<T> callback) {
Assert.notNull(callback);
applyStatementSettings(statement);
ReactiveSession session = getSession();
return Flux.defer(() -> callback.doInStatement(session, statement));
}
/**
* Create a reusable {@link Mono} given a {@link ReactiveStatementCallback} without exception translation.
*
* @param callback must not be {@literal null}.
* @return a reusable {@link Mono} wrapping the {@link ReactiveStatementCallback }.
*/
protected <T> Mono<T> createMono(Statement statement, ReactiveStatementCallback<T> callback) {
Assert.notNull(callback);
applyStatementSettings(statement);
ReactiveSession session = getSession();
return Mono.defer(() -> Mono.from(callback.doInStatement(session, statement)));
}
/**
* Create a reusable {@link Flux} given a {@link ReactiveSessionCallback} without exception translation.
*
* @param callback must not be {@literal null}.
* @return a reusable {@link Flux} wrapping the {@link ReactiveSessionCallback}.
*/
protected <T> Flux<T> createFlux(ReactiveSessionCallback<T> callback) {
Assert.notNull(callback);
ReactiveSession session = getSession();
return Flux.defer(() -> callback.doInSession(session));
}
/**
* Exception translation {@link Function} intended for {@link Mono#otherwise(Function)} usage.
*
* @return the exception translation {@link Function}
*/
@SuppressWarnings("ThrowableResultOfMethodCallIgnored")
protected <T> Function<Throwable, Mono<? extends T>> translateException() {
return throwable -> Mono.error(
throwable instanceof DriverException ? translateExceptionIfPossible((DriverException) throwable) : throwable);
}
/**
* Exception translation {@link Function} intended for {@link Mono#otherwise(Function)} usage.
*
* @param task readable text describing the task being attempted
* @param cql CQL query or update that caused the problem (may be {@code null})
* @return the exception translation {@link Function}
* @see CqlProvider
*/
@SuppressWarnings("ThrowableResultOfMethodCallIgnored")
protected <T> Function<Throwable, Mono<? extends T>> translateException(String task, String cql) {
return throwable -> Mono.error(throwable instanceof DriverException
? ReactiveCqlTemplate.this.translate(task, cql, (DriverException) throwable) : throwable);
}
/**
* Create a new RowMapper for reading columns as key-value pairs.
*
* @return the RowMapper to use
* @see ColumnMapRowMapper
*/
protected RowMapper<Map<String, Object>> getColumnMapRowMapper() {
return new ColumnMapRowMapper();
}
/**
* Create a new RowMapper for reading result objects from a single column.
*
* @param requiredType the type that each result object is expected to match
* @return the RowMapper to use
* @see SingleColumnRowMapper
*/
protected <T> RowMapper<T> getSingleColumnRowMapper(Class<T> requiredType) {
return SingleColumnRowMapper.newInstance(requiredType);
}
/**
* Prepare the given CQL Statement (or {@link com.datastax.driver.core.PreparedStatement}), applying statement
* settings such as fetch size, retry policy, and consistency level.
*
* @param stmt the CQL Statement to prepare
* @see #setFetchSize(int)
* @see #setRetryPolicy(RetryPolicy)
* @see #setConsistencyLevel(ConsistencyLevel)
*/
protected void applyStatementSettings(Statement stmt) {
int fetchSize = getFetchSize();
if (fetchSize != -1 && stmt.getFetchSize() == DEFAULTS.getFetchSize()) {
stmt.setFetchSize(fetchSize);
}
RetryPolicy retryPolicy = getRetryPolicy();
if (retryPolicy != null && stmt.getRetryPolicy() == DEFAULTS.getRetryPolicy()) {
stmt.setRetryPolicy(retryPolicy);
}
ConsistencyLevel consistencyLevel = getConsistencyLevel();
if (consistencyLevel != null && stmt.getConsistencyLevel() == DEFAULTS.getConsistencyLevel()) {
stmt.setConsistencyLevel(consistencyLevel);
}
}
/**
* Prepare the given CQL Statement (or {@link com.datastax.driver.core.PreparedStatement}), applying statement
* settings such as retry policy and consistency level.
*
* @param stmt the CQL Statement to prepare
* @see #setRetryPolicy(RetryPolicy)
* @see #setConsistencyLevel(ConsistencyLevel)
*/
protected void applyStatementSettings(PreparedStatement stmt) {
RetryPolicy retryPolicy = getRetryPolicy();
if (retryPolicy != null) {
stmt.setRetryPolicy(retryPolicy);
}
ConsistencyLevel consistencyLevel = getConsistencyLevel();
if (consistencyLevel != null) {
stmt.setConsistencyLevel(consistencyLevel);
}
}
/**
* Create a new arg-based PreparedStatementSetter using the args passed in.
* <p>
* By default, we'll create an {@link ArgumentPreparedStatementBinder}. This method allows for the creation to be
* overridden by subclasses.
*
* @param args object array with arguments
* @return the new {@link PreparedStatementBinder} to use
*/
protected PreparedStatementBinder newArgPreparedStatementBinder(Object[] args) {
return new ArgumentPreparedStatementBinder(args);
}
private ReactiveSession getSession() {
return getSessionFactory().getSession();
}
/**
* Determine CQL from potential provider object.
*
* @param cqlProvider object that's potentially a {@link CqlProvider}
* @return the CQL string, or {@code null}
* @see CqlProvider
*/
private static String getCql(Object cqlProvider) {
if (cqlProvider instanceof CqlProvider) {
return ((CqlProvider) cqlProvider).getCql();
} else {
return null;
}
}
private class SimpleReactivePreparedStatementCreator implements ReactivePreparedStatementCreator, CqlProvider {
private final String cql;
SimpleReactivePreparedStatementCreator(String cql) {
Assert.notNull(cql, "CQL must not be null");
this.cql = cql;
}
@Override
public Mono<PreparedStatement> createPreparedStatement(ReactiveSession session) throws DriverException {
return session.prepare(cql);
}
@Override
public String getCql() {
return cql;
}
}
}

View File

@@ -0,0 +1,62 @@
/*
* Copyright 2016 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.cassandra.core;
import org.reactivestreams.Publisher;
import org.springframework.dao.DataAccessException;
import com.datastax.driver.core.PreparedStatement;
import com.datastax.driver.core.Statement;
import com.datastax.driver.core.exceptions.DriverException;
/**
* Generic callback interface for code that operates on a {@link PreparedStatement}. Allows to execute any number of
* operations on a single {@link PreparedStatement}, for example a single {@link ReactiveSession#execute(Statement).
* <p>
* Used internally by {@link ReactiveCqlTemplate}, but also useful for application code. Note that the passed-in
* {@link PreparedStatement} can have been created by the framework or by a custom
* {@link ReactivePreparedStatementCreator}. However, the latter is hardly ever necessary, as most custom callback
* actions will perform updates in which case a standard {@link PreparedStatement is fine. Custom actions will always
* set parameter values themselves, so that {@link ReactivePreparedStatementCreator} capability is not needed either.
*
* @author Mark Paluch
* @since 2.0
* @see ReactiveCqlTemplate#execute(ReactivePreparedStatementCreator, ReactivePreparedStatementCallback)
* @see ReactiveCqlTemplate#execute(String, ReactivePreparedStatementCallback)
*/
public interface ReactivePreparedStatementCallback<T> {
/**
* Gets called by {@link ReactiveCqlTemplate#execute(String, ReactivePreparedStatementCallback)} with an active CQL
* session and {@link PreparedStatement}. Does not need to care about closing the session: this will all be handled by
* Spring's {@link ReactiveCqlTemplate}.
* <p>
* Allows for returning a result object created within the callback, i.e. a domain object or a collection of domain
* objects. Note that there's special support for single step actions: see
* {@link ReactiveCqlTemplate#queryForObject(String, Class, Object...)} etc. A thrown RuntimeException is treated as
* application exception, it gets propagated to the caller of the template.
*
* @param session active Cassandra session, must not be {@literal null}.
* @param ps the {@link PreparedStatement}, must not be {@literal null}.
* @return a result object publisher.
* @throws DriverException if thrown by a session method, to be auto-converted to a DataAccessException.
* @throws DataAccessException in case of custom exceptions.
* @see ReactiveCqlTemplate#queryForObject(String, Class, Object...)
* @see ReactiveCqlTemplate#queryForFlux(String, Object...)
*/
Publisher<T> doInPreparedStatement(ReactiveSession session, PreparedStatement ps)
throws DriverException, DataAccessException;
}

View File

@@ -0,0 +1,51 @@
/*
* Copyright 2016 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.cassandra.core;
import com.datastax.driver.core.PreparedStatement;
import com.datastax.driver.core.exceptions.DriverException;
import reactor.core.publisher.Mono;
/**
* One of the two central callback interfaces used by the {@link ReactiveCqlTemplate} class. This interface creates a
* {@link PreparedStatement} given a {@link ReactiveSession}, provided by the {@link ReactiveCqlTemplate} class.
* <p>
* Implementations may either create new prepared statements or reuse cached instances. Implementations do not need to
* concern themselves with {@link DriverException}s that may be thrown from operations they attempt. The
* {@link ReactiveCqlTemplate} class will catch and handle {@link DriverException}s appropriately.
* <p>
* A {@link ReactivePreparedStatementCreator} should also implement the {@link CqlProvider} interface if it is able to
* provide the CQL it uses for {@link PreparedStatement} creation. This allows for better contextual information in case
* of exceptions.
*
* @author Mark Paluch
* @since 2.0
*/
public interface ReactivePreparedStatementCreator {
/**
* Create a statement in this session. Allows implementations to use {@link PreparedStatement}s. The
* {@link ReactiveCqlTemplate} will attempt to cache the {@link PreparedStatement}s for future use without the
* overhead of re-preparing on the entire cluster.
*
* @param session Session to use to create statement, must not be {@literal null}.
* @return a prepared statement
* @throws DriverException there is no need to catch DriverException that may be thrown in the implementation of this
* method. The {@link ReactiveCqlTemplate} class will handle them.
*/
Mono<PreparedStatement> createPreparedStatement(ReactiveSession session) throws DriverException;
}

View File

@@ -0,0 +1,109 @@
/*
* Copyright 2016 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.cassandra.core;
import java.util.List;
import com.datastax.driver.core.ColumnDefinitions;
import com.datastax.driver.core.ExecutionInfo;
import com.datastax.driver.core.Row;
import com.datastax.driver.core.Statement;
import reactor.core.publisher.Flux;
/**
* The reactive result of a query.
* <p>
* The retrieval of the rows of a {@link ReactiveResultSet} is generally paged (a first page of result is fetched and
* the next one is only fetched once all the results of the first one has been consumed). The size of the pages can be
* configured either globally through {@link QueryOptions#setFetchSize} or per-statement with
* {@link Statement#setFetchSize}.
* <p>
* Please note however that this {@link ReactiveResultSet} paging is not available with the version 1 of the native
* protocol (i.e. with Cassandra 1.2 or if version 1 has been explicitly requested through
* {@link com.datastax.driver.core.Cluster.Builder#withProtocolVersion}). If the protocol version 1 is in use, a
* {@link ReactiveResultSet} is always fetched in it's entirely and it's up to the client to make sure that no query can
* yield {@link ReactiveResultSet} that won't hold in memory.
* <p>
* Note that this class is not thread-safe.
*
* @author Mark Paluch
* @since 2.0
* @see Flux
* @see ReactiveSession
* @see com.datastax.driver.core.ResultSet
*/
public interface ReactiveResultSet {
/**
* Returns a {@link Flux} over the rows contained in this result set.
* <p>
* The {@link Flux} will stream over all records that in this {@link ReactiveResultSet} according to the reactive
* demand.
* <p>
*
* @return a {@link Flux} of rows that will stream over all {@link Row rows} in this {@link ReactiveResultSet}.
*/
Flux<Row> rows();
/**
* Returns the columns returned in this ResultSet.
*
* @return the columns returned in this ResultSet.
*/
public ColumnDefinitions getColumnDefinitions();
/**
* If the query that produced this ResultSet was a conditional update, return whether it was successfully applied.
* <p>
* For consistency, this method always returns {@code true} for non-conditional queries (although there is no reason
* to call the method in that case). This is also the case for conditional DDL statements
* ({@code CREATE KEYSPACE... IF NOT EXISTS}, {@code CREATE TABLE... IF NOT EXISTS}), for which Cassandra doesn't
* return an {@code [applied]} column.
* <p>
* Note that, for versions of Cassandra strictly lower than 2.0.9 and 2.1.0-rc2, a server-side bug (CASSANDRA-7337)
* causes this method to always return {@code true} for batches containing conditional queries.
*
* @return if the query was a conditional update, whether it was applied. {@code true} for other types of queries.
* @see <a href="https://issues.apache.org/jira/browse/CASSANDRA-7337">CASSANDRA-7337</a>
*/
public boolean wasApplied();
/**
* Returns information on the execution of the last query made for this result set.
* <p>
* Note that in most cases, a result set is fetched with only one query, but large result sets can be paged and thus
* be retrieved by multiple queries. In that case this method return the {@link ExecutionInfo} for the last query
* performed. To retrieve the information for all queries, use {@link #getAllExecutionInfo}.
* <p>
* The returned object includes basic information such as the queried hosts, but also the Cassandra query trace if
* tracing was enabled for the query.
*
* @return the execution info for the last query made for this result set.
*/
ExecutionInfo getExecutionInfo();
/**
* Return the execution information for all queries made to retrieve this result set.
* <p>
* Unless the result set is large enough to get paged underneath, the returned list will be singleton. If paging has
* been used however, the returned list contains the {@link ExecutionInfo} objects for all the queries done to obtain
* this result set (at the time of the call) in the order those queries were made.
*
* @return a list of the execution info for all the queries made for this result set.
*/
List<ExecutionInfo> getAllExecutionInfo();
}

View File

@@ -0,0 +1,54 @@
/*
* Copyright 2016 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.cassandra.core;
import org.reactivestreams.Publisher;
import org.springframework.dao.DataAccessException;
import com.datastax.driver.core.exceptions.DriverException;
/**
* Callback interface used by {@link ReactiveCqlTemplate}'s query methods. Implementations of this interface perform the
* actual work of extracting results from a {@link ReactiveResultSet}, but don't need to worry about exception handling.
* {@link DriverException}s will be caught and handled by the calling {@link ReactiveCqlTemplate}.
* <p>
* This interface is mainly used within the CQL framework itself. A {@link RowMapper} is usually a simpler choice for
* {@link ReactiveResultSet} processing, mapping one result object per row instead of one result object for the entire
* {@link ReactiveResultSet}.
* <p>
* Note: {@link ReactiveResultSetExtractor} object is typically stateless and thus reusable, as long as it doesn't
* access stateful resources or keep result state within the object.
*
* @param <T>
* @author Mark Paluch
* @since 2.0
* @see ReactiveCqlTemplate
* @see RowCallbackHandler
* @see RowMapper
*/
public interface ReactiveResultSetExtractor<T> {
/**
* Implementations must implement this method to process the entire {@link ReactiveResultSet}.
*
* @param resultSet {@link ReactiveResultSet} to extract data from, must not be {@literal null}.
* @return an arbitrary result object {@link Publisher}.
* @throws DriverException if a {@link DriverException} is encountered getting column values or navigating (that is,
* there's no need to catch {@link DriverException}).
* @throws DataAccessException in case of custom exceptions.
*/
Publisher<T> extractData(ReactiveResultSet resultSet) throws DriverException, DataAccessException;
}

View File

@@ -0,0 +1,73 @@
/*
* Copyright 2016 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.cassandra.core;
import org.reactivestreams.Publisher;
import org.springframework.dao.DataAccessException;
import org.springframework.util.Assert;
import com.datastax.driver.core.exceptions.DriverException;
import reactor.core.publisher.Mono;
/**
* Adapter implementation of the {@link ReactiveResultSetExtractor} interface that delegates to a {@link RowMapper}
* which is supposed to create an object for each row. Each object is emitted through the {@link Publisher} of this
* {@link ReactiveResultSetExtractor}.
* <p>
* Useful for the typical case of one object per row in the database table. The number of entries in the results will
* match the number of rows.
* <p>
* Note that a {@link RowMapper} object is typically stateless and thus reusable.
*
* @author Mark Paluch
* @since 2.0
* @see RowMapper
* @see ReactiveCqlTemplate
*/
public class ReactiveRowMapperResultSetExtractor<T> implements ReactiveResultSetExtractor<T> {
private final RowMapper<T> rowMapper;
/**
* Create a new {@link ReactiveRowMapperResultSetExtractor}.
*
* @param rowMapper the {@link RowMapper} which creates an object for each row, must not be {@literal null}.
*/
public ReactiveRowMapperResultSetExtractor(RowMapper<T> rowMapper) {
Assert.notNull(rowMapper, "RowMapper is must not be null");
this.rowMapper = rowMapper;
}
/* (non-Javadoc)
* @see org.springframework.cassandra.core.ReactiveResultSetExtractor#extractData(org.springframework.cassandra.core.ReactiveResultSet)
*/
@Override
public Publisher<T> extractData(ReactiveResultSet resultSet) throws DriverException, DataAccessException {
return resultSet.rows().flatMap(row -> {
T value = this.rowMapper.mapRow(row, 0);
if (value == null) {
return Mono.empty();
}
return Mono.just(value);
});
}
}

View File

@@ -0,0 +1,200 @@
/*
* Copyright 2016 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.cassandra.core;
import java.io.Closeable;
import java.util.Map;
import com.datastax.driver.core.Cluster;
import com.datastax.driver.core.PreparedStatement;
import com.datastax.driver.core.RegularStatement;
import com.datastax.driver.core.Statement;
import com.datastax.driver.core.exceptions.NoHostAvailableException;
import com.datastax.driver.core.exceptions.QueryExecutionException;
import com.datastax.driver.core.exceptions.QueryValidationException;
import com.datastax.driver.core.exceptions.UnsupportedFeatureException;
import reactor.core.publisher.Mono;
/**
* A session holds connections to a Cassandra cluster, allowing it to be queried. {@link ReactiveSession} executes
* queries and prepares statements in a reactive style returning results wrapped in {@link Mono} and
* {@link reactor.core.publisher.Flux}.
* <p/>
* Each session maintains multiple connections to the cluster nodes, provides policies to choose which node to use for
* each query (round-robin on all nodes of the cluster by default), and handles retries for failed queries (when it
* makes sense).
* <p/>
* Session instances are thread-safe and usually a single instance is enough per application. As a given session can
* only be "logged" into one keyspace at a time (where the "logged" keyspace is the one used by queries that don't
* explicitly use a fully qualified table name), it can make sense to create one session per keyspace used. This is
* however not necessary when querying multiple keyspaces since it is always possible to use a single session with fully
* qualified table names in queries.
*
* @author Mark Paluch
* @since 2.0
* @see org.reactivestreams.Publisher
* @see Mono
* @see ReactiveResultSet
*/
public interface ReactiveSession extends Closeable {
/**
* Executes the provided query.
* <p/>
* This is a convenience method for {@code execute(new SimpleStatement(query))}.
*
* @param query the CQL query to execute.
* @return the result of the query. That result will never be null but can be empty (and will be for any non SELECT
* query).
* @throws NoHostAvailableException if no host in the cluster can be contacted successfully to execute this query.
* @throws QueryExecutionException if the query triggered an execution exception, i.e. an exception thrown by
* Cassandra when it cannot execute the query with the requested consistency level successfully.
* @throws QueryValidationException if the query if invalid (syntax error, unauthorized or any other validation
* problem).
*/
Mono<ReactiveResultSet> execute(String query);
/**
* Executes the provided query using the provided values.
* <p/>
* This is a convenience method for {@code execute(new SimpleStatement(query, values))}.
*
* @param query the CQL query to execute.
* @param values values required for the execution of {@code query}. See
* {@link SimpleStatement#SimpleStatement(String, Object...)} for more details.
* @return the result of the query. That result will never be null but can be empty (and will be for any non SELECT
* query).
* @throws NoHostAvailableException if no host in the cluster can be contacted successfully to execute this query.
* @throws QueryExecutionException if the query triggered an execution exception, i.e. an exception thrown by
* Cassandra when it cannot execute the query with the requested consistency level successfully.
* @throws QueryValidationException if the query if invalid (syntax error, unauthorized or any other validation
* problem).
* @throws UnsupportedFeatureException if version 1 of the protocol is in use (i.e. if you've forced version 1 through
* {@link Cluster.Builder#withProtocolVersion} or you use Cassandra 1.2).
*/
Mono<ReactiveResultSet> execute(String query, Object... values);
/**
* Executes the provided query using the provided named values.
* <p/>
* This is a convenience method for {@code execute(new SimpleStatement(query, values))}.
*
* @param query the CQL query to execute.
* @param values values required for the execution of {@code query}. See
* {@link SimpleStatement#SimpleStatement(String, Map)} for more details.
* @return the result of the query. That result will never be null but can be empty (and will be for any non SELECT
* query).
* @throws NoHostAvailableException if no host in the cluster can be contacted successfully to execute this query.
* @throws QueryExecutionException if the query triggered an execution exception, i.e. an exception thrown by
* Cassandra when it cannot execute the query with the requested consistency level successfully.
* @throws QueryValidationException if the query if invalid (syntax error, unauthorized or any other validation
* problem).
* @throws UnsupportedFeatureException if version 1 or 2 of the protocol is in use (i.e. if you've forced it through
* {@link Cluster.Builder#withProtocolVersion} or you use Cassandra 1.2 or 2.0).
*/
Mono<ReactiveResultSet> execute(String query, Map<String, Object> values);
/**
* Executes the provided query.
* <p/>
* This method blocks until at least some result has been received from the database. However, for SELECT queries, it
* does not guarantee that the result has been received in full. But it does guarantee that some response has been
* received from the database, and in particular guarantees that if the request is invalid, an exception will be
* thrown by this method.
*
* @param statement the CQL query to execute (that can be any {@link Statement}).
* @return the result of the query. That result will never be null but can be empty (and will be for any non SELECT
* query).
* @throws NoHostAvailableException if no host in the cluster can be contacted successfully to execute this query.
* @throws QueryExecutionException if the query triggered an execution exception, i.e. an exception thrown by
* Cassandra when it cannot execute the query with the requested consistency level successfully.
* @throws QueryValidationException if the query if invalid (syntax error, unauthorized or any other validation
* problem).
* @throws UnsupportedFeatureException if the protocol version 1 is in use and a feature not supported has been used.
* Features that are not supported by the version protocol 1 include: BatchStatement, ReactiveResultSet
* paging and binary values in RegularStatement.
*/
Mono<ReactiveResultSet> execute(Statement statement);
/**
* Prepares the provided query string.
*
* @param query the CQL query string to prepare
* @return the prepared statement corresponding to {@code query}.
* @throws NoHostAvailableException if no host in the cluster can be contacted successfully to prepare this query.
*/
Mono<PreparedStatement> prepare(String query);
/**
* Prepares the provided query.
* <p/>
* This method behaves like {@link #prepare(String)}, but note that the resulting {@code PreparedStatement} will
* inherit the query properties set on {@code statement}. Concretely, this means that in the following code:
*
* <pre>
* RegularStatement toPrepare = new SimpleStatement("SELECT * FROM test WHERE k=?")
* .setConsistencyLevel(ConsistencyLevel.QUORUM);
* PreparedStatement prepared = session.prepare(toPrepare);
* session.execute(prepared.bind("someValue"));
* </pre>
*
* the final execution will be performed with Quorum consistency.
* <p/>
* Please note that if the same CQL statement is prepared more than once, all calls to this method will return the
* same {@code PreparedStatement} object but the method will still apply the properties of the prepared
* {@code Statement} to this object.
*
* @param statement the statement to prepare
* @return the prepared statement corresponding to {@code statement}.
* @throws NoHostAvailableException if no host in the cluster can be contacted successfully to prepare this statement.
* @throws IllegalArgumentException if {@code statement.getValues() != null} (values for executing a prepared
* statement should be provided after preparation though the {@link PreparedStatement#bind} method or
* through a corresponding {@link BoundStatement}).
*/
Mono<PreparedStatement> prepare(RegularStatement statement);
/**
* Initiates a shutdown of this session instance and blocks until that shutdown completes.
* <p/>
* This method is a shortcut for {@code closeAsync().get()}.
* <p/>
* Note that this method does not close the corresponding {@code Cluster} instance (which holds additional resources,
* in particular internal executors that must be shut down in order for the client program to terminate). If you want
* to do so, use {@link Cluster#close}, but note that it will close all sessions created from that cluster.
*/
@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();
}

View File

@@ -0,0 +1,55 @@
/*
* Copyright 2016 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.cassandra.core;
import org.reactivestreams.Publisher;
import org.springframework.dao.DataAccessException;
import com.datastax.driver.core.Statement;
import com.datastax.driver.core.exceptions.DriverException;
/**
* Generic callback interface for code that operates on a CQL {@link ReactiveSession}. Allows to execute any number of
* operations on a single {@link ReactiveSession}, using any type and number of Statements.
* <p>
* This is particularly useful for delegating to existing data access code that expects a {@link ReactiveSession} to
* work on and throws {@link DriverException}. For newly written code, it is strongly recommended to use
* {@link CqlTemplate}'s more specific operations, for example a query or update variant.
*
* @param <T>
* @author Mark Paluch
* @since 2.0
* @see ReactiveCqlTemplate#execute(ReactiveSessionCallback)
*/
@FunctionalInterface
public interface ReactiveSessionCallback<T> {
/**
* Gets called by {@link ReactiveCqlTemplate#execute(ReactiveSessionCallback)} with an active Cassandra session. Does not
* need to care about activating or closing the {@link ReactiveSession}.
* <p>
* Allows for returning a result object created within the callback, i.e. a domain object or a collection of domain
* objects. Note that there's special support for single step actions: see
* {@link ReactiveCqlTemplate#queryForObject(Statement, Class)} etc. A thrown {@link RuntimeException} is treated as
* application exception: it gets propagated to the caller of the template.
*
* @param session active Cassandra session.
* @return a result object publisher
* @throws DriverException if thrown by a session method, to be auto-converted to a DataAccessException
* @throws DataAccessException in case of custom exceptions
*/
Publisher<T> doInSession(ReactiveSession session) throws DriverException, DataAccessException;
}

View File

@@ -0,0 +1,39 @@
/*
* Copyright 2016 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.cassandra.core;
/**
* Strategy interface to produce {@link ReactiveSession} instances.
* <p>
* Spring provides a {@link DefaultReactiveSessionFactory} implementation that just returns the same
* {@link ReactiveSession} instance. Implementations are free to return the same session or route calls to different
* sessions.
*
* @author Mark Paluch
* @see 2.0
* @see ReactiveSession
* @see DefaultReactiveSessionFactory
* @see ReactiveCqlTemplate
*/
public interface ReactiveSessionFactory {
/**
* Return a {@link ReactiveSession} to be used directly or inside a callback inside {@link ReactiveCqlTemplate}.
*
* @return a {@link ReactiveSession}.
*/
ReactiveSession getSession();
}

View File

@@ -0,0 +1,55 @@
/*
* Copyright 2016 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.cassandra.core;
import org.reactivestreams.Publisher;
import org.springframework.dao.DataAccessException;
import com.datastax.driver.core.Statement;
import com.datastax.driver.core.exceptions.DriverException;
/**
* Generic callback interface for code that operates on a CQL {@link Statement}. Allows to execute any number of
* operations on a single {@link Statement}, for example a single {@link ReactiveSession#execute(Statement)}.
* <p>
* Used internally by {@link ReactiveCqlTemplate}, but also useful for application code.
*
* @param <T>
* @author Mark Paluch
* @since 2.0
*/
@FunctionalInterface
public interface ReactiveStatementCallback<T> {
/**
* Gets called by {@link ReactiveCqlTemplate#execute(String)} with an active Cassandra session. Does not need to care about
* closing the the session: this will all be handled by Spring's {@link ReactiveCqlTemplate}.
* <p>
* Allows for returning a result object created within the callback, i.e. a domain object or a collection of domain
* objects. Note that there's special support for single step actions: see
* {@link ReactiveCqlTemplate#queryForObject(String, Class, Object...)} etc. A thrown RuntimeException is treated as
* application exception, it gets propagated to the caller of the template.
*
* @param session active Cassandra session.
* @param stmt CQL Statement
* @return a result object publisher
* @throws DriverException if thrown by a session method, to be auto-converted to a DataAccessException
* @throws DataAccessException in case of custom exceptions
* @see ReactiveCqlTemplate#queryForObject(String, Class)
* @see ReactiveCqlTemplate#queryForResultSet(String)
*/
Publisher<T> doInStatement(ReactiveSession session, Statement stmt) throws DriverException;
}

View File

@@ -0,0 +1,192 @@
/*
* Copyright 2016 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.cassandra.core;
import org.springframework.cassandra.support.RowUtils;
import org.springframework.cassandra.support.exception.IncorrectResultSetColumnCountException;
import org.springframework.dao.TypeMismatchDataAccessException;
import org.springframework.util.ClassUtils;
import org.springframework.util.NumberUtils;
import com.datastax.driver.core.ColumnDefinitions;
import com.datastax.driver.core.Row;
import com.datastax.driver.core.exceptions.DriverException;
/**
* {@link RowMapper} implementation that converts a single column into a single result value per row. Expects to operate
* on a {@link com.datastax.driver.core.Row} that just contains a single column.
* <p>
* The type of the result value for each row can be specified. The value for the single column will be extracted from a
* {@link Row} and converted into the specified target type.
*
* @author Mark Paluch
* @since 2.0
* @see ReactiveCqlTemplate#queryForFlux(String, Class)
* @see ReactiveCqlTemplate#queryForObject(String, Class)
*/
public class SingleColumnRowMapper<T> implements RowMapper<T> {
private Class<?> requiredType;
/**
* Create a new {@link SingleColumnRowMapper} for bean-style configuration.
*
* @see #setRequiredType
*/
public SingleColumnRowMapper() {}
/**
* Create a new {@code SingleColumnRowMapper}.
* <p>
* Consider using the {@link #newInstance} factory method instead, which allows for specifying the required type once
* only.
*
* @param requiredType the type that each result object is expected to match
*/
public SingleColumnRowMapper(Class<T> requiredType) {
setRequiredType(requiredType);
}
/**
* Set the type that each result object is expected to match.
* <p>
* If not specified, the column value will be exposed as returned by the {@link Row}.
*/
public void setRequiredType(Class<T> requiredType) {
this.requiredType = ClassUtils.resolvePrimitiveIfNecessary(requiredType);
}
/**
* Extract a value for the single column in the current row.
* <p>
* Validates that there is only one column selected, then delegates to {@code getColumnValue()} and also
* {@code convertValueToRequiredType}, if necessary.
*
* @see ColumnDefinitions#size()
* @see #getColumnValue(Row, int, Class)
* @see #convertValueToRequiredType(Object, Class)
*/
@SuppressWarnings("unchecked")
@Override
public T mapRow(Row row, int rowNum) throws DriverException {
// Validate column count.
ColumnDefinitions definitions = row.getColumnDefinitions();
int nrOfColumns = definitions.size();
if (nrOfColumns != 1) {
throw new IncorrectResultSetColumnCountException(1, nrOfColumns);
}
// Extract column value from CQL ResultSet.
Object result = getColumnValue(row, 0, this.requiredType);
if (result != null && this.requiredType != null && !this.requiredType.isInstance(result)) {
// Extracted value does not match already: try to convert it.
try {
return (T) convertValueToRequiredType(result, this.requiredType);
} catch (IllegalArgumentException ex) {
throw new TypeMismatchDataAccessException(
String.format("Type mismatch affecting row number %d and column type '%s': %s", rowNum,
definitions.getType(0), ex.getMessage()));
}
}
return (T) result;
}
/**
* Retrieve a CQL object value for the specified column.
* <p>
* The default implementation calls {@link RowUtils#getRowValue(Row, int, Class)}. If no required type has been
* specified, this method delegates to {@code getColumnValue(rs, index)}, which basically calls
* {@link Row#getObject(int)} but applies some additional default conversion to appropriate value types.
*
* @param row is the {@link Row} holding the data, must not be {@literal null}.
* @param index is the column index
* @param requiredType the type that each result object is expected to match (or {@code null} if none specified).
* @return the Object value.
* @throws DriverException in case of extraction failure
* @see RowUtils#getRowValue(Row, int, Class)
* @see #getColumnValue(Row, int)
*/
protected Object getColumnValue(Row row, int index, Class<?> requiredType) throws DriverException {
if (requiredType != null) {
return RowUtils.getRowValue(row, index, requiredType);
} else {
// No required type specified -> perform default extraction.
return getColumnValue(row, index);
}
}
/**
* Retrieve a object value for the specified column, using the most appropriate value type. Called if no required type
* has been specified.
* <p>
* The default implementation delegates to {@link RowUtils#getRowValue(Row, int, Class)}, which uses the
* {@link Row#getObject(int)} method.
*
* @param row is the {@link Row} holding the data, must not be {@literal null}.
* @param index is the column index
* @return the Object value.
* @throws DriverException in case of extraction failure.
* @see RowUtils#getRowValue(Row, int, Class)
*/
protected Object getColumnValue(Row row, int index) {
return RowUtils.getRowValue(row, index, null);
}
/**
* Convert the given column value to the specified required type. Only called if the extracted column value does not
* match already.
* <p>
* If the required type is String, the value will simply get stringified via {@code toString()}. In case of a Number,
* the value will be converted into a Number, either through number conversion or through String parsing (depending on
* the value type).
*
* @param value the column value as extracted from {@code getColumnValue()} (never {@code null})
* @param requiredType the type that each result object is expected to match (never {@code null})
* @return the converted value
* @see #getColumnValue(Row, int, Class)
*/
@SuppressWarnings("unchecked")
protected Object convertValueToRequiredType(Object value, Class<?> requiredType) {
if (String.class == requiredType) {
return value.toString();
} else if (Number.class.isAssignableFrom(requiredType)) {
if (value instanceof Number) {
// Convert original Number to target Number class.
return NumberUtils.convertNumberToTargetClass(((Number) value), (Class<Number>) requiredType);
} else {
// Convert stringified value to target Number class.
return NumberUtils.parseNumber(value.toString(), (Class<Number>) requiredType);
}
} else {
throw new IllegalArgumentException(
String.format("Value [%s] is of type [%s] and cannot be converted to required type [%s]", value,
value.getClass().getName(), requiredType.getName()));
}
}
/**
* Static factory method to create a new {@code SingleColumnRowMapper} (with the required type specified only once).
*
* @param requiredType the type that each result object is expected to match
*/
public static <T> SingleColumnRowMapper<T> newInstance(Class<T> requiredType) {
return new SingleColumnRowMapper<>(requiredType);
}
}

View File

@@ -0,0 +1,51 @@
/*
* Copyright 2016 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.cassandra.core.support;
import org.springframework.dao.DataAccessException;
import org.springframework.dao.support.PersistenceExceptionTranslator;
import com.datastax.driver.core.exceptions.DriverException;
/**
* Strategy interface for translating between {@link DriverException DriverExceptios} and Spring's data access
* strategy-agnostic {@link DataAccessException} hierarchy.
*
* @author Mark Paluch
* @see org.springframework.dao.DataAccessException
* @see 2.0
*/
@FunctionalInterface
public interface CQLExceptionTranslator extends PersistenceExceptionTranslator {
/**
* Translate the given {@link DriverException} into a generic {@link DataAccessException}.
* <p>
* The returned {@link DataAccessException} is supposed to contain the original {@code DriverException} as root cause.
* However, client code may not generally rely on this due to {@link DataAccessException}s possibly being caused by
* other resource APIs as well. That said, a {@code getRootCause() instanceof DataAccessException} check (and
* subsequent cast) is considered reliable when expecting Cassandra-based access to have happened.
*
* @param task readable text describing the task being attempted
* @param cql CQL query or update that caused the problem (may be {@code null})
* @param ex the offending {@link DriverException}
* @return the DataAccessException, wrapping the {@code DriverException}
* @see org.springframework.dao.DataAccessException#getRootCause()
*/
default DataAccessException translate(String task, String cql, DriverException ex) {
return translateExceptionIfPossible(ex);
}
}

View File

@@ -37,7 +37,7 @@ import com.datastax.driver.core.Session;
*/
public class CassandraAccessor implements InitializingBean {
CassandraExceptionTranslator exceptionTranslator = new CassandraExceptionTranslator();
protected CassandraExceptionTranslator exceptionTranslator = new CassandraExceptionTranslator();
protected final Logger logger = LoggerFactory.getLogger(getClass());

View File

@@ -15,6 +15,7 @@
*/
package org.springframework.cassandra.support;
import org.springframework.cassandra.core.support.CQLExceptionTranslator;
import org.springframework.cassandra.support.exception.CassandraAuthenticationException;
import org.springframework.cassandra.support.exception.CassandraConnectionFailureException;
import org.springframework.cassandra.support.exception.CassandraInsufficientReplicasAvailableException;
@@ -33,6 +34,7 @@ import org.springframework.cassandra.support.exception.CassandraUncategorizedExc
import org.springframework.cassandra.support.exception.CassandraWriteTimeoutException;
import org.springframework.dao.DataAccessException;
import org.springframework.dao.support.PersistenceExceptionTranslator;
import org.springframework.util.StringUtils;
import com.datastax.driver.core.WriteType;
import com.datastax.driver.core.exceptions.AlreadyExistsException;
@@ -52,80 +54,110 @@ import com.datastax.driver.core.exceptions.UnavailableException;
import com.datastax.driver.core.exceptions.WriteTimeoutException;
/**
* Simple {@link PersistenceExceptionTranslator} for Cassandra. Convert the given runtime exception to an appropriate
* exception from the {@code org.springframework.dao} hierarchy. Return {@literal null} if no translation is
* appropriate: any other exception may have resulted from user code, and should not be translated.
* Simple {@link PersistenceExceptionTranslator} for Cassandra.
* <p>
* Convert the given runtime exception to an appropriate exception from the {@code org.springframework.dao} hierarchy.
* Return {@literal null} if no translation is appropriate: any other exception may have resulted from user code, and
* should not be translated.
*
* @author Alex Shvid
* @author Matthew T. Adams
* @author Mark Paluch
*/
public class CassandraExceptionTranslator implements PersistenceExceptionTranslator {
public class CassandraExceptionTranslator implements CQLExceptionTranslator {
@Override
public DataAccessException translateExceptionIfPossible(RuntimeException x) {
public DataAccessException translateExceptionIfPossible(RuntimeException ex) {
if (x instanceof DataAccessException) {
return (DataAccessException) x;
if (ex instanceof DataAccessException) {
return (DataAccessException) ex;
}
if (!(x instanceof DriverException)) {
if (!(ex instanceof DriverException)) {
return null;
}
return translate(null, null, (DriverException) ex);
}
@Override
public DataAccessException translate(String task, String cql, DriverException ex) {
String message = buildMessage(task, cql, ex);
// Remember: subclasses must come before superclasses, otherwise the
// superclass would match before the subclass!
if (x instanceof AuthenticationException) {
return new CassandraAuthenticationException(((AuthenticationException) x).getHost(), x.getMessage(), x);
if (ex instanceof AuthenticationException) {
return new CassandraAuthenticationException(((AuthenticationException) ex).getHost(), message, ex);
}
if (x instanceof DriverInternalError) {
return new CassandraInternalException(x.getMessage(), x);
if (ex instanceof DriverInternalError) {
return new CassandraInternalException(message, ex);
}
if (x instanceof InvalidTypeException) {
return new CassandraTypeMismatchException(x.getMessage(), x);
if (ex instanceof InvalidTypeException) {
return new CassandraTypeMismatchException(message, ex);
}
if (x instanceof NoHostAvailableException) {
return new CassandraConnectionFailureException(((NoHostAvailableException) x).getErrors(), x.getMessage(), x);
if (ex instanceof NoHostAvailableException) {
return new CassandraConnectionFailureException(((NoHostAvailableException) ex).getErrors(), message, ex);
}
if (x instanceof ReadTimeoutException) {
return new CassandraReadTimeoutException(((ReadTimeoutException) x).wasDataRetrieved(), x.getMessage(), x);
if (ex instanceof ReadTimeoutException) {
return new CassandraReadTimeoutException(((ReadTimeoutException) ex).wasDataRetrieved(), message, ex);
}
if (x instanceof WriteTimeoutException) {
WriteType writeType = ((WriteTimeoutException) x).getWriteType();
return new CassandraWriteTimeoutException(writeType == null ? null : writeType.name(), x.getMessage(), x);
if (ex instanceof WriteTimeoutException) {
WriteType writeType = ((WriteTimeoutException) ex).getWriteType();
return new CassandraWriteTimeoutException(writeType == null ? null : writeType.name(), message, ex);
}
if (x instanceof TruncateException) {
return new CassandraTruncateException(x.getMessage(), x);
if (ex instanceof TruncateException) {
return new CassandraTruncateException(message, ex);
}
if (x instanceof UnavailableException) {
UnavailableException ux = (UnavailableException) x;
if (ex instanceof UnavailableException) {
UnavailableException ux = (UnavailableException) ex;
return new CassandraInsufficientReplicasAvailableException(ux.getRequiredReplicas(), ux.getAliveReplicas(),
x.getMessage(), x);
message, ex);
}
if (x instanceof AlreadyExistsException) {
AlreadyExistsException aex = (AlreadyExistsException) x;
if (ex instanceof AlreadyExistsException) {
AlreadyExistsException aex = (AlreadyExistsException) ex;
return aex.wasTableCreation() ? new CassandraTableExistsException(aex.getTable(), x.getMessage(), x)
: new CassandraKeyspaceExistsException(aex.getKeyspace(), x.getMessage(), x);
return aex.wasTableCreation() ? new CassandraTableExistsException(aex.getTable(), message, ex)
: new CassandraKeyspaceExistsException(aex.getKeyspace(), message, ex);
}
if (x instanceof InvalidConfigurationInQueryException) {
return new CassandraInvalidConfigurationInQueryException(x.getMessage(), x);
if (ex instanceof InvalidConfigurationInQueryException) {
return new CassandraInvalidConfigurationInQueryException(message, ex);
}
if (x instanceof InvalidQueryException) {
return new CassandraInvalidQueryException(x.getMessage(), x);
if (ex instanceof InvalidQueryException) {
return new CassandraInvalidQueryException(message, ex);
}
if (x instanceof SyntaxError) {
return new CassandraQuerySyntaxException(x.getMessage(), x);
if (ex instanceof SyntaxError) {
return new CassandraQuerySyntaxException(message, ex);
}
if (x instanceof UnauthorizedException) {
return new CassandraUnauthorizedException(x.getMessage(), x);
if (ex instanceof UnauthorizedException) {
return new CassandraUnauthorizedException(message, ex);
}
if (x instanceof TraceRetrievalException) {
return new CassandraTraceRetrievalException(x.getMessage(), x);
if (ex instanceof TraceRetrievalException) {
return new CassandraTraceRetrievalException(message, ex);
}
// unknown or unhandled exception
return new CassandraUncategorizedException(x.getMessage(), x);
return new CassandraUncategorizedException(message, ex);
}
/**
* Build a message {@code String} for the given {@link DriverException}.
* <p>
* To be called by translator subclasses when creating an instance of a generic
* {@link org.springframework.dao.DataAccessException} class.
*
* @param task readable text describing the task being attempted
* @param cql the CQL statement that caused the problem (may be {@code null})
* @param ex the offending {@code DriverException}
* @return the message {@code String} to use
*/
protected String buildMessage(String task, String cql, DriverException ex) {
if (StringUtils.hasText(task) || StringUtils.hasText(cql)) {
return task + "; CQL [" + cql + "]; " + ex.getMessage();
}
return ex.getMessage();
}
}

View File

@@ -0,0 +1,151 @@
/*
* Copyright 2016 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.cassandra.support;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.cassandra.core.ReactiveSessionFactory;
import org.springframework.cassandra.core.support.CQLExceptionTranslator;
import org.springframework.dao.DataAccessException;
import org.springframework.util.Assert;
import com.datastax.driver.core.exceptions.DriverException;
/**
* Base class for {@link org.springframework.cassandra.core.ReactiveCqlTemplate} and other CQL-accessing DAO helpers,
* defining common properties such as {@link org.springframework.cassandra.core.ReactiveSessionFactory} and exception
* translator.
* <p>
* Not intended to be used directly.
*
* @author Mark Paluch
* @since 2.0
* @see InitializingBean
* @see org.springframework.cassandra.core.ReactiveSession
* @see org.springframework.cassandra.core.ReactiveCqlTemplate
*/
public abstract class ReactiveCassandraAccessor implements InitializingBean {
/** Logger available to subclasses */
protected final Logger logger = LoggerFactory.getLogger(getClass());
private CQLExceptionTranslator exceptionTranslator = new CassandraExceptionTranslator();
private ReactiveSessionFactory sessionFactory;
/**
* Sets the {@link ReactiveSessionFactory} to use.
*
* @param sessionFactory must not be {@literal null}.
*/
public void setSessionFactory(ReactiveSessionFactory sessionFactory) {
Assert.notNull(sessionFactory, "ReactiveSessionFactory must not be null");
this.sessionFactory = sessionFactory;
}
/**
* Returns the configured {@link ReactiveSessionFactory}.
*
* @return the configured {@link ReactiveSessionFactory}.
*/
public ReactiveSessionFactory getSessionFactory() {
return sessionFactory;
}
/**
* Sets the exception translator used by this template to translate Cassandra specific exceptions into Spring DAO's
* Exception Hierarchy.
*
* @param exceptionTranslator exception translator to set; must not be {@literal null}.
* @see CassandraExceptionTranslator
* @see DataAccessException
*/
public void setExceptionTranslator(CQLExceptionTranslator exceptionTranslator) {
Assert.notNull(exceptionTranslator, "CQLExceptionTranslator must not be null");
this.exceptionTranslator = exceptionTranslator;
}
/**
* Returns the exception translator for this instance.
*
* @return the Cassandra exception translator.
* @see CassandraExceptionTranslator
*/
public CQLExceptionTranslator getExceptionTranslator() {
return this.exceptionTranslator;
}
/**
* Ensures the Cassandra {@link ReactiveSessionFactory} and exception translator has been properly set.
*/
@Override
public void afterPropertiesSet() {
Assert.notNull(sessionFactory != null, "ReactiveSessionFactory must not be null");
Assert.notNull(exceptionTranslator != null, "CassandraExceptionTranslator must not be null");
}
/**
* Translate the given {@link DriverException} into a generic {@link DataAccessException}.
* <p>
* The returned {@link DataAccessException} is supposed to contain the original {@code DriverException} as root cause.
* However, client code may not generally rely on this due to {@link DataAccessException}s possibly being caused by
* other resource APIs as well. That said, a {@code getRootCause() instanceof DataAccessException} check (and
* subsequent cast) is considered reliable when expecting Cassandra-based access to have happened.
*
* @param ex the offending {@link DriverException}
* @return the DataAccessException, wrapping the {@code DriverException}
* @see <a href=
* "http://docs.spring.io/spring/docs/current/spring-framework-reference/htmlsingle/#dao-exceptions">Consistent
* exception hierarchy</a>
* @see DataAccessException
*/
protected DataAccessException translateExceptionIfPossible(DriverException ex) {
Assert.notNull(ex, "DriverException must not be null");
return getExceptionTranslator().translateExceptionIfPossible(ex);
}
/**
* Translate the given {@link DriverException} into a generic {@link DataAccessException}.
* <p>
* The returned {@link DataAccessException} is supposed to contain the original {@code DriverException} as root cause.
* However, client code may not generally rely on this due to {@link DataAccessException}s possibly being caused by
* other resource APIs as well. That said, a {@code getRootCause() instanceof DataAccessException} check (and
* subsequent cast) is considered reliable when expecting Cassandra-based access to have happened.
*
* @param task readable text describing the task being attempted
* @param cql CQL query or update that caused the problem (may be {@code null})
* @param ex the offending {@link DriverException}
* @return the DataAccessException, wrapping the {@code DriverException}
* @see org.springframework.dao.DataAccessException#getRootCause()
* @see <a href=
* "http://docs.spring.io/spring/docs/current/spring-framework-reference/htmlsingle/#dao-exceptions">Consistent
* exception hierarchy</a>
*/
protected DataAccessException translate(String task, String cql, DriverException ex) {
Assert.notNull(ex, "DriverException must not be null");
return getExceptionTranslator().translate(task, cql, ex);
}
}

View File

@@ -0,0 +1,96 @@
/*
* Copyright 2016 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.cassandra.support;
import java.math.BigDecimal;
import java.nio.ByteBuffer;
import java.util.UUID;
import com.datastax.driver.core.LocalDate;
import com.datastax.driver.core.Row;
import com.datastax.driver.core.TupleValue;
import com.datastax.driver.core.UDTValue;
/**
* Generic utility methods for working with Cassandra. Mainly for internal use within the framework, but also useful for
* custom CQL access code.
*
* @author Mark Paluch
* @since 2.0
*/
public abstract class RowUtils {
/**
* Retrieve a CQL column value from a {@link Row}, using the specified value type.
* <p>
* Uses the specifically typed {@link Row} accessor methods, falling back to {@link Row#getObject(int)} for unknown
* types.
* <p>
* Note that the returned value may not be assignable to the specified required type, in case of an unknown type.
* Calling code needs to deal with this case appropriately, e.g. throwing a corresponding exception.
*
* @param row is the {@link Row} holding the data
* @param index is the column index
* @param requiredType the required value type (may be {@code null})
* @return the value object
*/
public static Object getRowValue(Row row, int index, Class<?> requiredType) {
if (requiredType == null) {
return row.getObject(index);
}
Object value;
// Explicitly extract typed value, as far as possible.
if (String.class == requiredType) {
return row.getString(index);
} else if (boolean.class == requiredType || Boolean.class == requiredType) {
value = row.getBool(index);
} else if (byte.class == requiredType || Byte.class == requiredType) {
value = row.getByte(index);
} else if (short.class == requiredType || Short.class == requiredType) {
value = row.getShort(index);
} else if (int.class == requiredType || Integer.class == requiredType) {
value = row.getInt(index);
} else if (long.class == requiredType || Long.class == requiredType) {
value = row.getLong(index);
} else if (float.class == requiredType || Float.class == requiredType) {
value = row.getFloat(index);
} else if (double.class == requiredType || Double.class == requiredType || Number.class == requiredType) {
value = row.getDouble(index);
} else if (BigDecimal.class == requiredType) {
return row.getDecimal(index);
} else if (LocalDate.class == requiredType) {
return row.getDate(index);
} else if (java.util.Date.class == requiredType) {
return row.getTimestamp(index);
} else if (ByteBuffer.class == requiredType) {
return row.getBytes(index);
} else if (TupleValue.class == requiredType) {
return row.getTupleValue(index);
} else if (UDTValue.class == requiredType) {
return row.getUDTValue(index);
} else if (UUID.class == requiredType) {
return row.getUUID(index);
} else {
// Some unknown type desired -> rely on getObject.
return row.getObject(index);
}
return (row.isNull(index) ? null : value);
}
}

View File

@@ -0,0 +1,63 @@
package org.springframework.cassandra.support.exception;
import org.springframework.dao.DataRetrievalFailureException;
/**
* Data access exception thrown when a result set did not have the correct column count, for example when expecting a
* single column but getting 0 or more than 1 columns.
*
* @author Mark Paluch
* @since 2.0
* @see org.springframework.dao.IncorrectResultSizeDataAccessException
*/
@SuppressWarnings("serial")
public class IncorrectResultSetColumnCountException extends DataRetrievalFailureException {
private int expectedCount;
private int actualCount;
/**
* Constructor for IncorrectResultSetColumnCountException.
*
* @param expectedCount the expected column count
* @param actualCount the actual column count
*/
public IncorrectResultSetColumnCountException(int expectedCount, int actualCount) {
super("Incorrect column count: expected " + expectedCount + ", actual " + actualCount);
this.expectedCount = expectedCount;
this.actualCount = actualCount;
}
/**
* Constructor for IncorrectResultCountDataAccessException.
*
* @param msg the detail message
* @param expectedCount the expected column count
* @param actualCount the actual column count
*/
public IncorrectResultSetColumnCountException(String msg, int expectedCount, int actualCount) {
super(msg);
this.expectedCount = expectedCount;
this.actualCount = actualCount;
}
/**
* Return the expected column count.
*/
public int getExpectedCount() {
return this.expectedCount;
}
/**
* Return the actual column count.
*/
public int getActualCount() {
return this.actualCount;
}
}

View File

@@ -0,0 +1,119 @@
/*
* Copyright 2016 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.cassandra.core;
import static org.assertj.core.api.Assertions.*;
import org.junit.Before;
import org.junit.Test;
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;
import reactor.core.publisher.Mono;
import reactor.core.scheduler.Schedulers;
/**
* Integration tests for {@link DefaultBridgedReactiveSession}.
*
* @author Mark Paluch
*/
public class DefaultBridgedReactiveSessionIntegrationTests extends AbstractKeyspaceCreatingIntegrationTest {
private DefaultBridgedReactiveSession reactiveSession;
@Before
public void before() throws Exception {
this.session.execute("DROP TABLE IF EXISTS users;");
this.reactiveSession = new DefaultBridgedReactiveSession(this.session, Schedulers.elastic());
}
/**
* @see DATACASS-335
*/
@Test
public void executeShouldExecuteDeferred() throws Exception {
Mono<ReactiveResultSet> execution = reactiveSession
.execute("CREATE TABLE users (\n" + " userid text PRIMARY KEY,\n" + " first_name text\n" + ");");
KeyspaceMetadata keyspace = getKeyspaceMetadata();
assertThat(keyspace.getTable("users")).isNull();
ReactiveResultSet resultSet = execution.block();
assertThat(resultSet.wasApplied()).isTrue();
assertThat(keyspace.getTable("users")).isNotNull();
}
/**
* @see DATACASS-335
*/
@Test
public void executeShouldTransportExceptionsInMono() throws Exception {
Mono<ReactiveResultSet> execution = reactiveSession.execute("INSERT INTO dummy;");
try {
execution.block();
fail("Missing SyntaxError");
} catch (SyntaxError e) {
assertThat(e).isInstanceOf(SyntaxError.class);
}
}
/**
* @see DATACASS-335
*/
@Test
public void executeShouldReturnRows() throws Exception {
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<ReactiveResultSet> execution = reactiveSession.execute("SELECT * FROM users;");
ReactiveResultSet resultSet = execution.block();
Row row = resultSet.rows().blockFirst();
assertThat(row).isNotNull();
assertThat(row.getString("userid")).isEqualTo("White");
}
/**
* @see DATACASS-335
*/
@Test
public void executeShouldPrepareStatement() throws Exception {
session.execute("CREATE TABLE users (\n" + " userid text PRIMARY KEY,\n" + " first_name text\n" + ");");
Mono<PreparedStatement> execution = reactiveSession
.prepare("INSERT INTO users (userid, first_name) VALUES (?, ?);");
PreparedStatement preparedStatement = execution.block();
assertThat(preparedStatement).isNotNull();
assertThat(preparedStatement.getQueryString()).isEqualTo("INSERT INTO users (userid, first_name) VALUES (?, ?);");
}
private KeyspaceMetadata getKeyspaceMetadata() {
return cluster.getMetadata().getKeyspace(this.session.getLoggedKeyspace());
}
}

View File

@@ -0,0 +1,178 @@
/*
* Copyright 2016 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.cassandra.core;
import static org.assertj.core.api.Assertions.*;
import static org.mockito.Mockito.*;
import java.util.Collections;
import org.hamcrest.core.IsEqual;
import org.junit.Before;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.mockito.Matchers;
import org.mockito.Mock;
import org.mockito.runners.MockitoJUnitRunner;
import com.datastax.driver.core.Cluster;
import com.datastax.driver.core.Session;
import com.datastax.driver.core.SimpleStatement;
import com.datastax.driver.core.Statement;
import reactor.core.scheduler.Schedulers;
/**
* Unit tests for {@link DefaultBridgedReactiveSession}.
*
* @author Mark Paluch
*/
@RunWith(MockitoJUnitRunner.class)
public class DefaultBridgedReactiveSessionUnitTests {
@Mock private Session sessionMock;
private DefaultBridgedReactiveSession reactiveSession;
@Before
public void before() throws Exception {
reactiveSession = new DefaultBridgedReactiveSession(sessionMock, Schedulers.immediate());
}
/**
* @see DATACASS-335
*/
@Test
public void executeStatementShouldForwardStatementToSession() throws Exception {
SimpleStatement statement = new SimpleStatement("SELECT *");
reactiveSession.execute(statement).subscribe();
verify(sessionMock).executeAsync(statement);
}
/**
* @see DATACASS-335
*/
@Test
public void executeShouldForwardStatementToSession() throws Exception {
reactiveSession.execute("SELECT *").subscribe();
verify(sessionMock).executeAsync(eq(new SimpleStatement("SELECT *")));
}
/**
* @see DATACASS-335
*/
@Test
public void executeWithValuesShouldForwardStatementToSession() throws Exception {
reactiveSession.execute("SELECT * WHERE a = ? and b = ?", "A", "B").subscribe();
verify(sessionMock).executeAsync(eq(new SimpleStatement("SELECT * WHERE a = ? and b = ?", "A", "B")));
}
/**
* @see DATACASS-335
*/
@Test
public void executeWithValueMapShouldForwardStatementToSession() throws Exception {
reactiveSession.execute("SELECT * WHERE a = ?", Collections.singletonMap("a", "value")).subscribe();
verify(sessionMock)
.executeAsync(eq(new SimpleStatement("SELECT * WHERE a = ?", Collections.singletonMap("a", "value"))));
}
/**
* @see DATACASS-335
*/
@Test
public void testPrepareQuery() throws Exception {
reactiveSession.prepare("SELECT *").subscribe();
verify(sessionMock).prepareAsync(eq(new SimpleStatement("SELECT *")));
}
/**
* @see DATACASS-335
*/
@Test
public void testPrepareStatement() throws Exception {
SimpleStatement statement = new SimpleStatement("SELECT *");
reactiveSession.prepare(statement).subscribe();
verify(sessionMock).prepareAsync(statement);
}
/**
* @see DATACASS-335
*/
@Test
public void testClose() throws Exception {
reactiveSession.close();
verify(sessionMock).close();
}
/**
* @see DATACASS-335
*/
@Test
public void testIsClosed() throws Exception {
when(reactiveSession.isClosed()).thenReturn(true);
boolean result = reactiveSession.isClosed();
assertThat(result).isTrue();
verify(sessionMock).isClosed();
}
/**
* @see DATACASS-335
*/
@Test
public void testGetCluster() throws Exception {
Cluster clusterMock = mock(Cluster.class);
when(sessionMock.getCluster()).thenReturn(clusterMock);
Cluster result = reactiveSession.getCluster();
assertThat(result).isSameAs(clusterMock);
}
private static <T extends Statement> T eq(T value) {
return Matchers.argThat(new IsEqual<T>(value) {
@Override
public boolean matches(Object actualValue) {
if (actualValue instanceof Statement) {
return value.toString().equals(actualValue.toString());
}
return super.matches(actualValue);
}
});
}
}

View File

@@ -0,0 +1,156 @@
/*
* Copyright 2016 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.cassandra.core;
import static org.assertj.core.api.Assertions.*;
import java.util.Map;
import java.util.concurrent.atomic.AtomicBoolean;
import org.junit.Before;
import org.junit.Test;
import org.springframework.cassandra.test.integration.AbstractKeyspaceCreatingIntegrationTest;
import com.datastax.driver.core.querybuilder.QueryBuilder;
import reactor.core.scheduler.Schedulers;
/**
* Integration tests for {@link ReactiveCqlTemplate}.
*
* @author Mark Paluch
*/
public class ReactiveCqlTemplateIntegrationTests extends AbstractKeyspaceCreatingIntegrationTest {
private static final AtomicBoolean initialized = new AtomicBoolean();
private ReactiveSession reactiveSession;
private ReactiveCqlTemplate template;
@Before
public void before() throws Exception {
reactiveSession = new DefaultBridgedReactiveSession(getSession(), Schedulers.elastic());
if (initialized.compareAndSet(false, true)) {
getSession().execute("CREATE TABLE IF NOT EXISTS user (id text PRIMARY KEY, username text);");
} else {
getSession().execute("TRUNCATE user;");
}
getSession().execute("INSERT INTO user (id, username) VALUES ('WHITE', 'Walter');");
template = new ReactiveCqlTemplate(new DefaultReactiveSessionFactory(reactiveSession));
}
/**
* @see DATACASS-335
*/
@Test
public void executeShouldRemoveRecords() throws Exception {
template.execute("DELETE FROM user WHERE id = 'WHITE'").block();
assertThat(getSession().execute("SELECT * FROM user").one()).isNull();
}
/**
* @see DATACASS-335
*/
@Test
public void queryForObjectShouldReturnFirstColumn() throws Exception {
String id = template.queryForObject("SELECT id FROM user;", String.class).block();
assertThat(id).isEqualTo("WHITE");
}
/**
* @see DATACASS-335
*/
@Test
public void queryForObjectShouldReturnMap() throws Exception {
Map<String, Object> map = template.queryForMap("SELECT * FROM user;").block();
assertThat(map).containsEntry("id", "WHITE").containsEntry("username", "Walter");
}
/**
* @see DATACASS-335
*/
@Test
public void executeStatementShouldRemoveRecords() throws Exception {
template.execute(QueryBuilder.delete().from("user").where(QueryBuilder.eq("id", "WHITE"))).block();
assertThat(getSession().execute("SELECT * FROM user").one()).isNull();
}
/**
* @see DATACASS-335
*/
@Test
public void queryForObjectStatementShouldReturnFirstColumn() throws Exception {
String id = template.queryForObject(QueryBuilder.select("id").from("user"), String.class).block();
assertThat(id).isEqualTo("WHITE");
}
/**
* @see DATACASS-335
*/
@Test
public void queryForObjectStatementShouldReturnMap() throws Exception {
Map<String, Object> map = template.queryForMap(QueryBuilder.select().from("user")).block();
assertThat(map).containsEntry("id", "WHITE").containsEntry("username", "Walter");
}
/**
* @see DATACASS-335
*/
@Test
public void executeWithArgsShouldRemoveRecords() throws Exception {
template.execute("DELETE FROM user WHERE id = ?", "WHITE").block();
assertThat(getSession().execute("SELECT * FROM user").one()).isNull();
}
/**
* @see DATACASS-335
*/
@Test
public void queryForObjectWithArgsShouldReturnFirstColumn() throws Exception {
String id = template.queryForObject("SELECT id FROM user WHERE id = ?;", String.class, "WHITE").block();
assertThat(id).isEqualTo("WHITE");
}
/**
* @see DATACASS-335
*/
@Test
public void queryForObjectWithArgsShouldReturnMap() throws Exception {
Map<String, Object> map = template.queryForMap("SELECT * FROM user WHERE id = ?;", "WHITE").block();
assertThat(map).containsEntry("id", "WHITE").containsEntry("username", "Walter");
}
}

View File

@@ -0,0 +1,167 @@
/*
* Copyright 2016 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.cassandra.core;
import static org.assertj.core.api.Assertions.*;
import static org.mockito.Mockito.*;
import org.junit.Before;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.mockito.Mock;
import org.mockito.runners.MockitoJUnitRunner;
import org.springframework.cassandra.support.exception.IncorrectResultSetColumnCountException;
import org.springframework.dao.TypeMismatchDataAccessException;
import com.datastax.driver.core.ColumnDefinitions;
import com.datastax.driver.core.DataType;
import com.datastax.driver.core.Row;
/**
* Unit tests for {@link SingleColumnRowMapper}.
*
* @author Mark Paluch
* @soundtrack Kos Vs Michael Buffer - Go For It All (Rubberboot Mix)
*/
@RunWith(MockitoJUnitRunner.class)
public class SingleColumnRowMapperUnitTests {
@Mock private Row row;
@Mock private ColumnDefinitions columnDefinitions;
private SingleColumnRowMapper rowMapper;
@Before
public void before() throws Exception {
when(row.getColumnDefinitions()).thenReturn(columnDefinitions);
}
/**
* @see DATACASS-335
*/
@Test
public void getColumnValueWithType() {
when(row.getDouble(2)).thenReturn(42d);
rowMapper = new SingleColumnRowMapper();
assertThat(rowMapper.getColumnValue(row, 2, Number.class)).isEqualTo(42d);
}
/**
* @see DATACASS-335
*/
@Test
public void getColumnValue() {
when(row.getObject(2)).thenReturn(42d);
rowMapper = new SingleColumnRowMapper();
assertThat(rowMapper.getColumnValue(row, 2)).isEqualTo(42d);
}
/**
* @see DATACASS-335
*/
@Test
public void convertValueToRequiredTypeForNumber() {
rowMapper = new SingleColumnRowMapper<Number>();
assertThat(rowMapper.convertValueToRequiredType(1234, Integer.class)).isEqualTo(1234);
assertThat(rowMapper.convertValueToRequiredType(1234.2, Integer.class)).isEqualTo(1234);
assertThat(rowMapper.convertValueToRequiredType(1234.2, Double.class)).isEqualTo(1234.2);
}
/**
* @see DATACASS-335
*/
@Test
public void convertValueToRequiredTypeForString() {
rowMapper = new SingleColumnRowMapper<Number>();
assertThat(rowMapper.convertValueToRequiredType("1234", Integer.class)).isEqualTo(1234);
assertThat(rowMapper.convertValueToRequiredType("1234.2", Double.class)).isEqualTo(1234.2);
}
/**
* @see DATACASS-335
*/
@Test(expected = IllegalArgumentException.class)
public void convertValueToRequiredTypeShouldFail() {
rowMapper = new SingleColumnRowMapper<>();
rowMapper.convertValueToRequiredType("1234", Object.class);
}
/**
* @see DATACASS-335
*/
@Test
public void mapRowSingleColumn() {
when(columnDefinitions.size()).thenReturn(1);
when(row.getInt(0)).thenReturn(42);
rowMapper = SingleColumnRowMapper.newInstance(Integer.class);
assertThat(rowMapper.mapRow(row, 2)).isEqualTo(42);
}
/**
* @see DATACASS-335
*/
@Test
public void mapRowSingleColumnNullValue() {
when(columnDefinitions.size()).thenReturn(1);
when(row.getObject(0)).thenReturn(null);
rowMapper = SingleColumnRowMapper.newInstance(Object.class);
assertThat(rowMapper.mapRow(row, 2)).isNull();
}
/**
* @see DATACASS-335
*/
@Test(expected = TypeMismatchDataAccessException.class)
public void mapRowSingleColumnWrongType() {
when(columnDefinitions.size()).thenReturn(1);
when(columnDefinitions.getType(0)).thenReturn(DataType.blob());
when(row.getObject(0)).thenReturn("hello");
rowMapper = SingleColumnRowMapper.newInstance(ColumnDefinitions.class);
rowMapper.mapRow(row, 2);
}
/**
* @see DATACASS-335
*/
@Test(expected = IncorrectResultSetColumnCountException.class)
public void tooManyColumns() {
when(columnDefinitions.size()).thenReturn(2);
rowMapper = SingleColumnRowMapper.newInstance(ColumnDefinitions.class);
rowMapper.mapRow(row, 1);
}
}

View File

@@ -15,7 +15,8 @@
*/
package org.springframework.cassandra.support;
import static org.assertj.core.api.Assertions.*;
import static org.assertj.core.api.Assertions.assertThat;
import static org.junit.Assert.*;
import org.junit.Test;
import org.springframework.cassandra.support.exception.CassandraInvalidConfigurationInQueryException;
@@ -33,8 +34,9 @@ import com.datastax.driver.core.exceptions.InvalidQueryException;
* Unit tests for {@link CassandraExceptionTranslator}
*
* @author Matthew T. Adams
* @author Mark Paluch
*/
public class CassandraExceptionTranslatorTest {
public class CassandraExceptionTranslatorUnitTests {
CassandraExceptionTranslator tx = new CassandraExceptionTranslator();
@@ -85,4 +87,17 @@ public class CassandraExceptionTranslatorTest {
assertThat(dax instanceof CassandraInvalidQueryException).isTrue();
assertThat(dax.getCause()).isEqualTo(cx);
}
/**
* @see DATACASS-335
*/
@Test
public void shouldTranslateWithCqlMessage() {
InvalidQueryException cx = new InvalidConfigurationInQueryException(null, "err");
DataAccessException dax = tx.translate("Query", "SELECT * FROM person", cx);
assertThat(dax).hasRootCauseInstanceOf(InvalidQueryException.class).hasMessage(
"Query; CQL [SELECT * FROM person]; err; nested exception is com.datastax.driver.core.exceptions.InvalidConfigurationInQueryException: err");
}
}