diff --git a/spring-cassandra/src/main/java/org/springframework/cassandra/core/CqlOperations.java b/spring-cassandra/src/main/java/org/springframework/cassandra/core/CqlOperations.java
index 75531cabb..895296609 100644
--- a/spring-cassandra/src/main/java/org/springframework/cassandra/core/CqlOperations.java
+++ b/spring-cassandra/src/main/java/org/springframework/cassandra/core/CqlOperations.java
@@ -60,6 +60,14 @@ public interface CqlOperations {
*/
void execute(String cql) throws DataAccessException;
+ /**
+ * Executes the supplied CQL Query and returns nothing.
+ *
+ * @param cql
+ * @param options may be null
+ */
+ void execute(String cql, QueryOptions options) throws DataAccessException;
+
/**
* Executes the supplied Query and returns nothing.
*
@@ -74,6 +82,13 @@ public interface CqlOperations {
*/
void executeAsynchronously(String cql) throws DataAccessException;
+ /**
+ * Executes the supplied Query Asynchronously and returns nothing.
+ *
+ * @param cql The {@link Query} to execute
+ */
+ void executeAsynchronously(String cql, QueryOptions options) throws DataAccessException;
+
/**
* Executes the supplied CQL Query Asynchronously and returns nothing.
*
@@ -123,21 +138,26 @@ public interface CqlOperations {
ResultSetFuture queryAsynchronously(String cql, QueryOptions options);
/**
- * Executes the provided CQL Query with the provided {@link Runnable} implementation.
+ * Executes the provided CQL Query with the provided {@link Runnable}, which is started after the query has completed.
+ *
+ * A more useful method than this one is {@link #queryAsynchronously(String, AsynchronousQueryListener)}, where you're
+ * given the {@link ResultSetFuture} after the query has been executed.
*
* @param cql The Query
* @param listener {@link Runnable} listener for handling the query in a separate thread
+ *
+ * @see #queryAsynchronously(String, AsynchronousQueryListener)
*/
void queryAsynchronously(String cql, Runnable listener);
/**
* Executes the provided CQL Query with the provided listener. This is preferred over the same method that takes a
- * plain Runnable. The {@link AsynchronousQueryListener} gives you access to the {@link ResultSetFuture} once the
+ * {@link Runnable}. The {@link AsynchronousQueryListener} gives you access to the {@link ResultSetFuture} once the
* query is completed for optimal flexibility.
*
* @param cql The Query
- * @param listener {@link AsynchronousQueryListener} Listener for handling the query's {@link ResultSetFuture} in a
- * separate thread
+ * @param listener {@link AsynchronousQueryListener} for handling the query's {@link ResultSetFuture} in a separate
+ * thread
*/
void queryAsynchronously(String cql, AsynchronousQueryListener listener);
@@ -771,9 +791,8 @@ public interface CqlOperations {
*
* @param cql The CQL
* @param rows Object array of Object array of values to bind to the CQL.
- * @param options The Query Options Object
*/
- void ingest(String cql, Object[][] rows, QueryOptions options);
+ void ingest(String cql, Object[][] rows);
/**
* This is an operation designed for high performance writes. The cql is used to create a PreparedStatement once, then
@@ -785,8 +804,9 @@ public interface CqlOperations {
*
* @param cql The CQL
* @param rows Object array of Object array of values to bind to the CQL.
+ * @param options The Query Options Object
*/
- void ingest(String cql, Object[][] rows);
+ void ingest(String cql, Object[][] rows, QueryOptions options);
/**
* Delete all rows in the table
diff --git a/spring-cassandra/src/main/java/org/springframework/cassandra/core/CqlTemplate.java b/spring-cassandra/src/main/java/org/springframework/cassandra/core/CqlTemplate.java
index c606e1e28..1cdb5928b 100644
--- a/spring-cassandra/src/main/java/org/springframework/cassandra/core/CqlTemplate.java
+++ b/spring-cassandra/src/main/java/org/springframework/cassandra/core/CqlTemplate.java
@@ -53,7 +53,6 @@ import com.datastax.driver.core.BoundStatement;
import com.datastax.driver.core.ColumnDefinitions;
import com.datastax.driver.core.ColumnDefinitions.Definition;
import com.datastax.driver.core.Host;
-import com.datastax.driver.core.Metadata;
import com.datastax.driver.core.PreparedStatement;
import com.datastax.driver.core.Query;
import com.datastax.driver.core.ResultSet;
@@ -83,8 +82,53 @@ import com.datastax.driver.core.querybuilder.Truncate;
public class CqlTemplate extends CassandraAccessor implements CqlOperations {
/**
- * Blank constructor. You must wire in the Session before use.
+ * Add common {@link Query} options for all types of queries.
*
+ * @param q
+ * @param options
+ * @return the {@link Query} given.
+ */
+ public static Query addQueryOptions(Query q, QueryOptions options) {
+
+ if (options == null) {
+ return q;
+ }
+
+ if (options.getConsistencyLevel() != null) {
+ q.setConsistencyLevel(ConsistencyLevelResolver.resolve(options.getConsistencyLevel()));
+ }
+ if (options.getRetryPolicy() != null) {
+ q.setRetryPolicy(RetryPolicyResolver.resolve(options.getRetryPolicy()));
+ }
+ return q;
+ }
+
+ /**
+ * Add common Query options for all types of queries.
+ *
+ * @param q
+ * @param optionsByName
+ */
+ public static void addPreparedStatementOptions(PreparedStatement s, QueryOptions options) {
+
+ if (options == null) {
+ return;
+ }
+
+ /*
+ * Add Query Options
+ */
+ if (options.getConsistencyLevel() != null) {
+ s.setConsistencyLevel(ConsistencyLevelResolver.resolve(options.getConsistencyLevel()));
+ }
+ if (options.getRetryPolicy() != null) {
+ s.setRetryPolicy(RetryPolicyResolver.resolve(options.getRetryPolicy()));
+ }
+
+ }
+
+ /**
+ * Blank constructor. You must wire in the Session before use.
*/
public CqlTemplate() {
}
@@ -105,13 +149,18 @@ public class CqlTemplate extends CassandraAccessor implements CqlOperations {
}
@Override
- public void execute(Query query) throws DataAccessException {
- doExecute(query, null);
+ public void execute(String cql) throws DataAccessException {
+ execute(cql, (QueryOptions) null);
}
@Override
- public void execute(final String cql) throws DataAccessException {
- doExecute(cql, null);
+ public void execute(String cql, QueryOptions options) throws DataAccessException {
+ doExecute(cql, options);
+ }
+
+ @Override
+ public void execute(Query query) throws DataAccessException {
+ doExecute(query);
}
@Override
@@ -376,41 +425,38 @@ public class CqlTemplate extends CassandraAccessor implements CqlOperations {
}
}
- /**
- * Execute a command at the Session Level
- *
- * @param callback
- * @return
- */
- protected ResultSet doExecute(final String cql, final QueryOptions options) {
+ protected ResultSet doExecute(String cql) {
+ return doExecute(cql, null);
+ }
- logger.info(cql);
+ protected ResultSet doExecute(String cql, QueryOptions options) {
+ return doExecute(addQueryOptions(new SimpleStatement(cql), options));
+ }
+
+ /**
+ * Execute a command at the Session Level with optional options
+ *
+ * @param q The query to execute.
+ * @param options The {@link QueryOptions}. May be null.
+ */
+ protected ResultSet doExecute(final Query q) {
return doExecute(new SessionCallback() {
@Override
public ResultSet doInSession(Session s) throws DataAccessException {
- SimpleStatement statement = new SimpleStatement(cql);
- addQueryOptions(statement, options);
- return s.execute(statement);
+ return s.execute(q);
}
});
}
- /**
- * Execute a command at the Session Level
- *
- * @param callback
- * @return
- */
- protected ResultSet doExecute(final Query q, final QueryOptions options) {
+ protected ResultSetFuture doExecuteAsync(final Query q) {
- return doExecute(new SessionCallback() {
+ return doExecute(new SessionCallback() {
@Override
- public ResultSet doInSession(Session s) throws DataAccessException {
- addQueryOptions(q, options);
- return s.execute(q);
+ public ResultSetFuture doInSession(Session s) throws DataAccessException {
+ return s.executeAsync(q);
}
});
}
@@ -459,25 +505,13 @@ public class CqlTemplate extends CassandraAccessor implements CqlOperations {
*/
protected Set getHosts() {
- /*
- * Get the cluster metadata for this session
- */
- Metadata clusterMetadata = doExecute(new SessionCallback() {
+ return doExecute(new SessionCallback>() {
@Override
- public Metadata doInSession(Session s) throws DataAccessException {
- return s.getCluster().getMetadata();
+ public Set doInSession(Session s) throws DataAccessException {
+ return s.getCluster().getMetadata().getAllHosts();
}
-
});
-
- /*
- * Get all hosts in the cluster
- */
- Set hosts = clusterMetadata.getAllHosts();
-
- return hosts;
-
}
@Override
@@ -487,23 +521,18 @@ public class CqlTemplate extends CassandraAccessor implements CqlOperations {
}
@Override
- public void executeAsynchronously(final String cql) throws DataAccessException {
- execute(new SessionCallback