diff --git a/spring-cql/src/main/java/org/springframework/cassandra/core/ArgumentPreparedStatementBinder.java b/spring-cql/src/main/java/org/springframework/cassandra/core/ArgumentPreparedStatementBinder.java
new file mode 100644
index 000000000..e4dda65db
--- /dev/null
+++ b/spring-cql/src/main/java/org/springframework/cassandra/core/ArgumentPreparedStatementBinder.java
@@ -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();
+ }
+}
diff --git a/spring-cql/src/main/java/org/springframework/cassandra/core/ColumnMapRowMapper.java b/spring-cql/src/main/java/org/springframework/cassandra/core/ColumnMapRowMapper.java
new file mode 100644
index 000000000..86b21e0cf
--- /dev/null
+++ b/spring-cql/src/main/java/org/springframework/cassandra/core/ColumnMapRowMapper.java
@@ -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.
+ *
+ * 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.
+ *
+ * Note: 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> {
+
+ @Override
+ public Map mapRow(Row rs, int rowNum) {
+
+ ColumnDefinitions columnDefinitions = rs.getColumnDefinitions();
+ int columnCount = columnDefinitions.size();
+ Map 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.
+ *
+ * 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 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.
+ *
+ * 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);
+ }
+}
diff --git a/spring-cql/src/main/java/org/springframework/cassandra/core/CqlProvider.java b/spring-cql/src/main/java/org/springframework/cassandra/core/CqlProvider.java
new file mode 100644
index 000000000..52af64a27
--- /dev/null
+++ b/spring-cql/src/main/java/org/springframework/cassandra/core/CqlProvider.java
@@ -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.
+ *
+ * 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();
+}
diff --git a/spring-cql/src/main/java/org/springframework/cassandra/core/CqlTemplate.java b/spring-cql/src/main/java/org/springframework/cassandra/core/CqlTemplate.java
index 2a5a53bac..bf97804aa 100644
--- a/spring-cql/src/main/java/org/springframework/cassandra/core/CqlTemplate.java
+++ b/spring-cql/src/main/java/org/springframework/cassandra/core/CqlTemplate.java
@@ -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;
diff --git a/spring-cql/src/main/java/org/springframework/cassandra/core/DefaultBridgedReactiveSession.java b/spring-cql/src/main/java/org/springframework/cassandra/core/DefaultBridgedReactiveSession.java
new file mode 100644
index 000000000..cd6a70130
--- /dev/null
+++ b/spring-cql/src/main/java/org/springframework/cassandra/core/DefaultBridgedReactiveSession.java
@@ -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.
+ *
+ * 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.
+ *
+ * {@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.
+ *
+ * All CQL operations performed by this class are logged at debug level, using
+ * "org.springframework.cassandra.core.DefaultBridgedReactiveSession" as log category.
+ *
+ *
+ * @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 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 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 execute(String query, Map 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 execute(Statement statement) {
+
+ Assert.notNull(statement, "Statement must not be null");
+
+ return Mono.defer(() -> {
+
+ try {
+
+ if (logger.isDebugEnabled()) {
+ logger.debug("Executing Statement [{}]", statement);
+ }
+
+ CompletableFuture 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 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 prepare(RegularStatement statement) {
+
+ Assert.notNull(statement, "Statement must not be null");
+
+ return Mono.defer(() -> {
+
+ try {
+
+ if (logger.isDebugEnabled()) {
+ logger.debug("Preparing Statement [{}]", statement);
+ }
+
+ CompletableFuture future = new CompletableFuture<>();
+ ListenableFuture 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 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 getAllExecutionInfo() {
+ return resultSet.getAllExecutionInfo();
+ }
+ }
+}
diff --git a/spring-cql/src/main/java/org/springframework/cassandra/core/DefaultReactiveSessionFactory.java b/spring-cql/src/main/java/org/springframework/cassandra/core/DefaultReactiveSessionFactory.java
new file mode 100644
index 000000000..7cfc9b62d
--- /dev/null
+++ b/spring-cql/src/main/java/org/springframework/cassandra/core/DefaultReactiveSessionFactory.java
@@ -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}.
+ *
+ * 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;
+ }
+}
diff --git a/spring-cql/src/main/java/org/springframework/cassandra/core/PreparedStatementBinder.java b/spring-cql/src/main/java/org/springframework/cassandra/core/PreparedStatementBinder.java
index da3a25b4b..8548c1e06 100644
--- a/spring-cql/src/main/java/org/springframework/cassandra/core/PreparedStatementBinder.java
+++ b/spring-cql/src/main/java/org/springframework/cassandra/core/PreparedStatementBinder.java
@@ -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.
+ *
+ * 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.
+ *
+ * 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.
+ *
+ * Implementations do not 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;
-
}
diff --git a/spring-cql/src/main/java/org/springframework/cassandra/core/ReactiveCqlOperations.java b/spring-cql/src/main/java/org/springframework/cassandra/core/ReactiveCqlOperations.java
new file mode 100644
index 000000000..1ad4b9cba
--- /dev/null
+++ b/spring-cql/src/main/java/org/springframework/cassandra/core/ReactiveCqlOperations.java
@@ -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.
+ *
+ * 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.
+ */
+ Flux execute(ReactiveSessionCallback 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 execute(String cql) throws DataAccessException;
+
+ /**
+ * Execute a query given static CQL, reading the {@link ReactiveResultSet} with a {@link ReactiveResultSetExtractor}.
+ *
+ * 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...)
+ */
+ Flux query(String cql, ReactiveResultSetExtractor rse) throws DataAccessException;
+
+ /**
+ * Execute a query given static CQL, mapping each row to a Java object via a {@link RowMapper}.
+ *
+ * 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[])
+ */
+ Flux query(String cql, RowMapper rowMapper) throws DataAccessException;
+
+ /**
+ * Execute a query given static CQL, mapping a single result row to a Java object via a {@link RowMapper}.
+ *
+ * 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[])
+ */
+ Mono queryForObject(String cql, RowMapper rowMapper) throws DataAccessException;
+
+ /**
+ * Execute a query for a result object, given static CQL.
+ *
+ * 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.
+ *
+ * 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[])
+ */
+ Mono queryForObject(String cql, Class requiredType) throws DataAccessException;
+
+ /**
+ * Execute a query for a result Map, given static CQL.
+ *
+ * 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.
+ *
+ * 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> queryForMap(String cql) throws DataAccessException;
+
+ /**
+ * Execute a query for a result {@link Flux}, given static CQL.
+ *
+ * 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.
+ *
+ * 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
+ */
+ Flux queryForFlux(String cql, Class elementType) throws DataAccessException;
+
+ /**
+ * Execute a query for a result {@link Flux}, given static CQL.
+ *
+ * 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.
+ *
+ * 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> queryForFlux(String cql) throws DataAccessException;
+
+ /**
+ * Execute a query for a ResultSet, given static CQL.
+ *
+ * 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.
+ *
+ * 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 queryForResultSet(String cql) throws DataAccessException;
+
+ /**
+ * Execute a query for Rows, given static CQL.
+ *
+ * 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.
+ *
+ * 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 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 execute(Publisher 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 execute(Statement statement) throws DataAccessException;
+
+ /**
+ * Execute a query given static CQL, reading the {@link ReactiveResultSet} with a {@link ReactiveResultSetExtractor}.
+ *
+ * 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...)
+ */
+ Flux query(Statement statement, ReactiveResultSetExtractor rse) throws DataAccessException;
+
+ /**
+ * Execute a query given static CQL, mapping each row to a Java object via a {@link RowMapper}.
+ *
+ * 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[])
+ */
+ Flux query(Statement statement, RowMapper rowMapper) throws DataAccessException;
+
+ /**
+ * Execute a query given static CQL, mapping a single result row to a Java object via a {@link RowMapper}.
+ *
+ * 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[])
+ */
+ Mono queryForObject(Statement statement, RowMapper rowMapper) throws DataAccessException;
+
+ /**
+ * Execute a query for a result object, given static CQL.
+ *
+ * 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.
+ *
+ * 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[])
+ */
+ Mono queryForObject(Statement statement, Class requiredType) throws DataAccessException;
+
+ /**
+ * Execute a query for a result Map, given static CQL.
+ *
+ * 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.
+ *
+ * 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> queryForMap(Statement statement) throws DataAccessException;
+
+ /**
+ * Execute a query for a result {@link Flux}, given static CQL.
+ *
+ * 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.
+ *
+ * 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
+ */
+ Flux queryForFlux(Statement statement, Class elementType) throws DataAccessException;
+
+ /**
+ * Execute a query for a result {@link Flux}, given static CQL.
+ *
+ * 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.
+ *
+ * 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> queryForFlux(Statement statement) throws DataAccessException;
+
+ /**
+ * Execute a query for a ResultSet, given static CQL.
+ *
+ * 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.
+ *
+ * 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 queryForResultSet(Statement statement) throws DataAccessException;
+
+ /**
+ * Execute a query for Rows, given static CQL.
+ *
+ * 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.
+ *
+ * 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 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.
+ *
+ * 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
+ */
+ Flux execute(ReactivePreparedStatementCreator psc, ReactivePreparedStatementCallback 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.
+ *
+ * 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
+ */
+ Flux execute(String cql, ReactivePreparedStatementCallback 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
+ */
+ Flux query(ReactivePreparedStatementCreator psc, ReactiveResultSetExtractor 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
+ */
+ Flux query(String cql, PreparedStatementBinder psb, ReactiveResultSetExtractor 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
+ */
+ Flux query(ReactivePreparedStatementCreator psc, PreparedStatementBinder psb,
+ ReactiveResultSetExtractor 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.
+ */
+ Flux query(String cql, ReactiveResultSetExtractor 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.
+ */
+ Flux query(ReactivePreparedStatementCreator psc, RowMapper 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.
+ */
+ Flux query(String cql, PreparedStatementBinder psb, RowMapper 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.
+ */
+ Flux query(ReactivePreparedStatementCreator psc, PreparedStatementBinder psb, RowMapper 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.
+ */
+ Flux query(String cql, RowMapper 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.
+ */
+ Mono queryForObject(String cql, RowMapper 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.
+ *
+ * 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)
+ */
+ Mono queryForObject(String cql, Class 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.
+ *
+ * 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> 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}.
+ *
+ * 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
+ */
+ Flux queryForFlux(String cql, Class 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}.
+ *
+ * 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> 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.
+ *
+ * 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 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.
+ *
+ * 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 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 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 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 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 execute(String cql, Publisher args) throws DataAccessException;
+}
diff --git a/spring-cql/src/main/java/org/springframework/cassandra/core/ReactiveCqlTemplate.java b/spring-cql/src/main/java/org/springframework/cassandra/core/ReactiveCqlTemplate.java
new file mode 100644
index 000000000..5c92b5ac9
--- /dev/null
+++ b/spring-cql/src/main/java/org/springframework/cassandra/core/ReactiveCqlTemplate.java
@@ -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;
+
+/**
+ * This is the central class in the CQL core package for reactive Cassandra data access. 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.
+ *
+ * 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.
+ *
+ * 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.
+ *
+ * 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.
+ *
+ * All CQL operations performed by this class are logged at debug level, using
+ * "org.springframework.cassandra.core.ReactiveCqlTemplate" as log category.
+ *
+ * NOTE: An instance of this class is thread-safe once configured.
+ *
+ * @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 Flux execute(ReactiveSessionCallback 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 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 Flux query(String cql, ReactiveResultSetExtractor 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 Flux query(String cql, RowMapper 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 Mono queryForObject(String cql, RowMapper 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 Mono queryForObject(String cql, Class requiredType) throws DataAccessException {
+ return queryForObject(cql, getSingleColumnRowMapper(requiredType));
+ }
+
+ /* (non-Javadoc)
+ * @see org.springframework.cassandra.core.ReactiveCqlOperations#queryForMap(java.lang.String)
+ */
+ @Override
+ public Mono> 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 Flux queryForFlux(String cql, Class elementType) throws DataAccessException {
+ return query(cql, getSingleColumnRowMapper(elementType));
+ }
+
+ /* (non-Javadoc)
+ * @see org.springframework.cassandra.core.ReactiveCqlOperations#queryForFlux(java.lang.String)
+ */
+ @Override
+ public Flux> queryForFlux(String cql) throws DataAccessException {
+ return query(cql, getColumnMapRowMapper());
+ }
+
+ /* (non-Javadoc)
+ * @see org.springframework.cassandra.core.ReactiveCqlOperations#queryForResultSet(java.lang.String)
+ */
+ @Override
+ public Mono 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 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 execute(Publisher 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 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 Flux query(Statement statement, ReactiveResultSetExtractor 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 Flux query(Statement statement, RowMapper 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 Mono queryForObject(Statement statement, RowMapper 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 Mono queryForObject(Statement statement, Class 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> 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 Flux queryForFlux(Statement statement, Class 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> 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 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 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 Flux execute(ReactivePreparedStatementCreator psc, ReactivePreparedStatementCallback 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 Flux execute(String cql, ReactivePreparedStatementCallback 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 Flux query(ReactivePreparedStatementCreator psc, PreparedStatementBinder psb,
+ ReactiveResultSetExtractor 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 Flux query(ReactivePreparedStatementCreator psc, ReactiveResultSetExtractor 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 Flux query(String cql, PreparedStatementBinder psb, ReactiveResultSetExtractor 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 Flux query(String cql, ReactiveResultSetExtractor 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 Flux query(ReactivePreparedStatementCreator psc, RowMapper 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 Flux query(String cql, PreparedStatementBinder psb, RowMapper 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 Flux query(ReactivePreparedStatementCreator psc, PreparedStatementBinder psb, RowMapper 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 Flux query(String cql, RowMapper 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 Mono queryForObject(String cql, RowMapper 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 Mono queryForObject(String cql, Class 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> 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 Flux queryForFlux(String cql, Class 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> 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 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 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 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 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 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 execute(String cql, Publisher 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 Flux createFlux(Statement statement, ReactiveStatementCallback 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 Mono createMono(Statement statement, ReactiveStatementCallback 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 Flux createFlux(ReactiveSessionCallback 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 Function> 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 Function> 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> 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 RowMapper getSingleColumnRowMapper(Class 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.
+ *
+ * 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 createPreparedStatement(ReactiveSession session) throws DriverException {
+ return session.prepare(cql);
+ }
+
+ @Override
+ public String getCql() {
+ return cql;
+ }
+ }
+}
diff --git a/spring-cql/src/main/java/org/springframework/cassandra/core/ReactivePreparedStatementCallback.java b/spring-cql/src/main/java/org/springframework/cassandra/core/ReactivePreparedStatementCallback.java
new file mode 100644
index 000000000..799367dfb
--- /dev/null
+++ b/spring-cql/src/main/java/org/springframework/cassandra/core/ReactivePreparedStatementCallback.java
@@ -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).
+ *
+ * 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