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

DATACASS-144: done; rebased all branch commits

Browse Source
This commit is contained in:
Matthew Adams
2014-09-23 10:50:57 -05:00
parent 88d5aa2100
commit ff41674809
152 changed files with 2624 additions and 762 deletions
Download Patch File Download Diff File Expand all files Collapse all files

8
spring-cql/src/main/java/org/springframework/cassandra/config/CassandraCqlClusterFactoryBean.java
View File

@@ -336,10 +336,10 @@ public class CassandraCqlClusterFactoryBean implements FactoryBean<Cluster>, Ini
private static Compression convertCompressionType(CompressionType type) {
switch (type) {
case NONE:
return Compression.NONE;
case SNAPPY:
return Compression.SNAPPY;
case NONE:
return Compression.NONE;
case SNAPPY:
return Compression.SNAPPY;
}
throw new IllegalArgumentException("unknown compression type " + type);
}

22
spring-cql/src/main/java/org/springframework/cassandra/config/KeyspaceActionSpecificationFactoryBean.java
View File

@@ -35,13 +35,11 @@ import org.springframework.cassandra.core.keyspace.Option;
import org.springframework.util.Assert;
/**
* A single keyspace XML Element can result in multiple actions. Example: {@literal CREATE_DROP}.
*
* This FactoryBean inspects the action required to satisfy the keyspace element, and then returns a Set of atomic
* A single keyspace XML Element can result in multiple actions. Example: {@literal CREATE_DROP}. This FactoryBean
* inspects the action required to satisfy the keyspace element, and then returns a Set of atomic
* {@link KeyspaceActionSpecification} required to satisfy the configuration action.
*
* @author David Webb
*
*/
public class KeyspaceActionSpecificationFactoryBean implements FactoryBean<Set<KeyspaceActionSpecification<?>>>,
InitializingBean, DisposableBean {
@@ -74,14 +72,14 @@ public class KeyspaceActionSpecificationFactoryBean implements FactoryBean<Set<K
Assert.notNull(action, "Keyspace Action is required for a Keyspace Action");
switch (action) {
case CREATE_DROP:
specs.add(generateDropKeyspaceSpecification());
case CREATE:
// Assert.notNull(replicationStrategy, "Replication Strategy is required to create a Keyspace");
specs.add(generateCreateKeyspaceSpecification());
break;
case ALTER:
break;
case CREATE_DROP:
specs.add(generateDropKeyspaceSpecification());
case CREATE:
// Assert.notNull(replicationStrategy, "Replication Strategy is required to create a Keyspace");
specs.add(generateCreateKeyspaceSpecification());
break;
case ALTER:
break;
}
}

1
spring-cql/src/main/java/org/springframework/cassandra/config/MultiLevelListFlattenerFactoryBean.java
View File

@@ -25,7 +25,6 @@ import org.springframework.beans.factory.FactoryBean;
*
* @author David Webb
* @param <T>
*
*/
public class MultiLevelListFlattenerFactoryBean<T> implements FactoryBean<List<T>> {

1
spring-cql/src/main/java/org/springframework/cassandra/config/MultiLevelSetFlattenerFactoryBean.java
View File

@@ -27,7 +27,6 @@ import org.springframework.beans.factory.FactoryBean;
*
* @author David Webb
* @param <T>
*
*/
public class MultiLevelSetFlattenerFactoryBean<T> implements FactoryBean<Set<T>> {

1
spring-cql/src/main/java/org/springframework/cassandra/config/xml/ParsingUtils.java
View File

@@ -244,7 +244,6 @@ public class ParsingUtils {
* null or empty, and <code>defaultValue</code> is null or empty, then no property is added and this method
* silently returns.
* @param reference If <code>true</code>, this method will add the property as a reference, else as a value.
*
* @see BeanDefinitionBuilder#addPropertyReference(String, String)
* @see BeanDefinitionBuilder#addPropertyValue(String, Object)
*/

6
spring-cql/src/main/java/org/springframework/cassandra/core/AsynchronousQueryListener.java
View File

@@ -21,14 +21,14 @@ import com.datastax.driver.core.ResultSetFuture;
* Interface used to give an implementation access to a {@link ResultSetFuture} after the query has completed.
*
* @author David Webb
* @author Matthew T. Adams
*/
public interface AsynchronousQueryListener {
/**
* Called upon Query Completion.
* Called upon query completion.
*
* @param rsf The {@link ResultSetFuture}. Since this isn't called until the asynchronous query completes, it can be
* immediately interrogated.
* @param rsf The {@link ResultSetFuture}. Called when the query operation is completed.
*/
public void onQueryComplete(ResultSetFuture rsf);
}

7
spring-cql/src/main/java/org/springframework/cassandra/core/CachedPreparedStatementCreator.java
View File

@@ -28,13 +28,10 @@ import com.datastax.driver.core.exceptions.DriverException;
/**
* This Prepared Statement Creator maintains a cache of all prepared statements for the duration of this life of the
* container.
*
* When preparing statements with Cassandra, each Statement should be prepared once and only once due to the overhead of
* preparing the statement.
* container. When preparing statements with Cassandra, each Statement should be prepared once and only once due to the
* overhead of preparing the statement.
*
* @author David Webb
*
*/
public class CachedPreparedStatementCreator implements PreparedStatementCreator {

7
spring-cql/src/main/java/org/springframework/cassandra/core/QueryCancellor.java → spring-cql/src/main/java/org/springframework/cassandra/core/Cancellable.java
View File

@@ -20,7 +20,10 @@ package org.springframework.cassandra.core;
*
* @author Matthew T. Adams
*/
public interface QueryCancellor {
public interface Cancellable {
void cancelQuery(boolean mayInterruptIfRunning);
/**
* Cancels the query operation that this cancellable came from.
*/
void cancel();
}

1
spring-cql/src/main/java/org/springframework/cassandra/core/CassandraUncategorizedDataAccessException.java
View File

@@ -21,7 +21,6 @@ import org.springframework.dao.UncategorizedDataAccessException;
* Wrapper for all non RuntimeExceptions throws by the Cassandra Driver
*
* @author David Webb
*
*/
public class CassandraUncategorizedDataAccessException extends UncategorizedDataAccessException {

1
spring-cql/src/main/java/org/springframework/cassandra/core/ConsistencyLevel.java
View File

@@ -19,7 +19,6 @@ package org.springframework.cassandra.core;
* Generic Consistency Levels associated with Cassandra.
*
* @author David Webb
*
*/
public enum ConsistencyLevel {

74
spring-cql/src/main/java/org/springframework/cassandra/core/ConsistencyLevelResolver.java
View File

@@ -19,15 +19,13 @@ package org.springframework.cassandra.core;
* Determine driver consistency level based on ConsistencyLevel
*
* @author David Webb
*
*/
public final class ConsistencyLevelResolver {
/**
* No instances allowed
*/
private ConsistencyLevelResolver() {
}
private ConsistencyLevelResolver() {}
/**
* Decode the generic spring data cassandra enum to the type required by the DataStax Driver.
@@ -43,41 +41,41 @@ public final class ConsistencyLevelResolver {
* Determine the driver level based on our enum
*/
switch (level) {
case ONE:
resolvedLevel = com.datastax.driver.core.ConsistencyLevel.ONE;
break;
case LOCAL_ONE:
resolvedLevel = com.datastax.driver.core.ConsistencyLevel.LOCAL_ONE;
break;
case ALL:
resolvedLevel = com.datastax.driver.core.ConsistencyLevel.ALL;
break;
case ANY:
resolvedLevel = com.datastax.driver.core.ConsistencyLevel.ANY;
break;
case EACH_QUOROM:
resolvedLevel = com.datastax.driver.core.ConsistencyLevel.EACH_QUORUM;
break;
case LOCAL_QUOROM:
resolvedLevel = com.datastax.driver.core.ConsistencyLevel.LOCAL_QUORUM;
break;
case QUOROM:
resolvedLevel = com.datastax.driver.core.ConsistencyLevel.QUORUM;
break;
case THREE:
resolvedLevel = com.datastax.driver.core.ConsistencyLevel.THREE;
break;
case TWO:
resolvedLevel = com.datastax.driver.core.ConsistencyLevel.TWO;
break;
case SERIAL:
resolvedLevel = com.datastax.driver.core.ConsistencyLevel.SERIAL;
break;
case LOCAL_SERIAL:
resolvedLevel = com.datastax.driver.core.ConsistencyLevel.LOCAL_SERIAL;
break;
default:
break;
case ONE:
resolvedLevel = com.datastax.driver.core.ConsistencyLevel.ONE;
break;
case LOCAL_ONE:
resolvedLevel = com.datastax.driver.core.ConsistencyLevel.LOCAL_ONE;
break;
case ALL:
resolvedLevel = com.datastax.driver.core.ConsistencyLevel.ALL;
break;
case ANY:
resolvedLevel = com.datastax.driver.core.ConsistencyLevel.ANY;
break;
case EACH_QUOROM:
resolvedLevel = com.datastax.driver.core.ConsistencyLevel.EACH_QUORUM;
break;
case LOCAL_QUOROM:
resolvedLevel = com.datastax.driver.core.ConsistencyLevel.LOCAL_QUORUM;
break;
case QUOROM:
resolvedLevel = com.datastax.driver.core.ConsistencyLevel.QUORUM;
break;
case THREE:
resolvedLevel = com.datastax.driver.core.ConsistencyLevel.THREE;
break;
case TWO:
resolvedLevel = com.datastax.driver.core.ConsistencyLevel.TWO;
break;
case SERIAL:
resolvedLevel = com.datastax.driver.core.ConsistencyLevel.SERIAL;
break;
case LOCAL_SERIAL:
resolvedLevel = com.datastax.driver.core.ConsistencyLevel.LOCAL_SERIAL;
break;
default:
break;
}
return resolvedLevel;

299
spring-cql/src/main/java/org/springframework/cassandra/core/CqlOperations.java
View File

@@ -52,6 +52,36 @@ import com.datastax.driver.core.querybuilder.Update;
*/
public interface CqlOperations {
/**
* Convenient method that delegates to {@link ResultSetFuture#getUninterruptibly()} but translates exceptions if any
* are thrown.
*
* @param rsf The {@link ResultSetFuture} from which to get the {@link ResultSet}.
* @return The {@link ResultSet}
*/
ResultSet getResultSetUninterruptibly(ResultSetFuture rsf);
/**
* Convenient method that delegates to {@link ResultSetFuture#getUninterruptibly()} but translates exceptions if any
* are thrown.
*
* @param rsf The {@link ResultSetFuture} from which to get the {@link ResultSet}.
* @param timeout The timeout to wait in milliseconds. A nonpositive value means wait indefinitely.
* @return The {@link ResultSet}
*/
ResultSet getResultSetUninterruptibly(ResultSetFuture rsf, long millis);
/**
* Convenient method that delegates to {@link ResultSetFuture#getUninterruptibly()} but translates exceptions if any
* are thrown.
*
* @param rsf The {@link ResultSetFuture} from which to get the {@link ResultSet}.
* @param timeout The timeout to wait. A nonpositive value means wait indefinitely.
* @param unit The {@link TimeUnit} of the timeout.
* @return The {@link ResultSet}
*/
ResultSet getResultSetUninterruptibly(ResultSetFuture rsf, long timeout, TimeUnit unit);
/**
* Executes the supplied {@link SessionCallback} in the current Template Session. The implementation of
* SessionCallback can decide whether or not to <code>execute()</code> or <code>executeAsync()</code> the operation.
@@ -140,10 +170,10 @@ public interface CqlOperations {
*
* @param cql The CQL String to execute
* @param listener The {@link Runnable} to register with the {@link ResultSetFuture}
* @return A {@link QueryCancellor} that can be used to cancel the query.
* @return A {@link Cancellable} that can be used to cancel the query.
* @see queryAsyncronously for Reads
*/
QueryCancellor executeAsynchronously(String cql, Runnable listener) throws DataAccessException;
Cancellable executeAsynchronously(String cql, Runnable listener) throws DataAccessException;
/**
* Executes the supplied Query Asynchronously and returns nothing.
@@ -151,20 +181,20 @@ public interface CqlOperations {
* @param cql The CQL String to execute
* @param listener The {@link Runnable} to register with the {@link ResultSetFuture}
* @param executor The {@link Executor} to regsiter with the {@link ResultSetFuture}
* @return A {@link QueryCancellor} that can be used to cancel the query.
* @return A {@link Cancellable} that can be used to cancel the query.
* @see queryAsyncronously for Reads
*/
QueryCancellor executeAsynchronously(String cql, Runnable listener, Executor executor) throws DataAccessException;
Cancellable executeAsynchronously(String cql, Runnable listener, Executor executor) throws DataAccessException;
/**
* Executes the supplied Query Asynchronously and returns nothing.
*
* @param cql The CQL String to execute
* @param listener The {@link AsynchronousQueryListener} to register with the {@link ResultSetFuture}
* @return A {@link QueryCancellor} that can be used to cancel the query.
* @return A {@link Cancellable} that can be used to cancel the query.
* @see queryAsyncronously for Reads
*/
QueryCancellor executeAsynchronously(String cql, AsynchronousQueryListener listener) throws DataAccessException;
Cancellable executeAsynchronously(String cql, AsynchronousQueryListener listener) throws DataAccessException;
/**
* Executes the supplied Query Asynchronously and returns nothing.
@@ -172,10 +202,10 @@ public interface CqlOperations {
* @param cql The CQL String to execute
* @param listener The {@link AsynchronousQueryListener} to register with the {@link ResultSetFuture}
* @param executor The {@link Executor} to regsiter with the {@link ResultSetFuture}
* @return A {@link QueryCancellor} that can be used to cancel the query.
* @return A {@link Cancellable} that can be used to cancel the query.
* @see queryAsyncronously for Reads
*/
QueryCancellor executeAsynchronously(String cql, AsynchronousQueryListener listener, Executor executor)
Cancellable executeAsynchronously(String cql, AsynchronousQueryListener listener, Executor executor)
throws DataAccessException;
/**
@@ -218,6 +248,16 @@ public interface CqlOperations {
*/
ResultSetFuture executeAsynchronously(Batch batch) throws DataAccessException;
Cancellable executeAsynchronously(Truncate truncate, AsynchronousQueryListener listener) throws DataAccessException;
Cancellable executeAsynchronously(Delete delete, AsynchronousQueryListener listener) throws DataAccessException;
Cancellable executeAsynchronously(Insert insert, AsynchronousQueryListener listener) throws DataAccessException;
Cancellable executeAsynchronously(Update update, AsynchronousQueryListener listener) throws DataAccessException;
Cancellable executeAsynchronously(Batch batch, AsynchronousQueryListener listener) throws DataAccessException;
/**
* Executes the supplied CQL Query Asynchronously and returns nothing.
*
@@ -230,34 +270,33 @@ public interface CqlOperations {
* Executes the supplied CQL Query Asynchronously and returns nothing.
*
* @param query The {@link Statement} to execute
* @return A {@link QueryCancellor} that can be used to cancel the query.
* @return A {@link Cancellable} that can be used to cancel the query.
*/
QueryCancellor executeAsynchronously(Statement query, Runnable runnable) throws DataAccessException;
Cancellable executeAsynchronously(Statement query, Runnable runnable) throws DataAccessException;
/**
* Executes the supplied CQL Query Asynchronously and returns nothing.
*
* @param query The {@link Statement} to execute
* @return A {@link QueryCancellor} that can be used to cancel the query.
* @return A {@link Cancellable} that can be used to cancel the query.
*/
QueryCancellor executeAsynchronously(Statement query, AsynchronousQueryListener listener) throws DataAccessException;
Cancellable executeAsynchronously(Statement query, AsynchronousQueryListener listener) throws DataAccessException;
/**
* Executes the supplied CQL Query Asynchronously and returns nothing.
*
* @param query The {@link Statement} to execute
* @return A {@link QueryCancellor} that can be used to cancel the query.
* @return A {@link Cancellable} that can be used to cancel the query.
*/
QueryCancellor executeAsynchronously(Statement query, Runnable runnable, Executor executor)
throws DataAccessException;
Cancellable executeAsynchronously(Statement query, Runnable runnable, Executor executor) throws DataAccessException;
/**
* Executes the supplied CQL Query Asynchronously and returns nothing.
*
* @param query The {@link Statement} to execute
* @return A {@link QueryCancellor} that can be used to cancel the query.
* @return A {@link Cancellable} that can be used to cancel the query.
*/
QueryCancellor executeAsynchronously(Statement query, AsynchronousQueryListener listener, Executor executor)
Cancellable executeAsynchronously(Statement query, AsynchronousQueryListener listener, Executor executor)
throws DataAccessException;
/**
@@ -317,10 +356,10 @@ public interface CqlOperations {
*
* @param cql The Query
* @param listener {@link Runnable} listener for handling the query in a separate thread
* @return A {@link QueryCancellor} that can be used to cancel the query.
* @return A {@link Cancellable} that can be used to cancel the query.
* @see #queryAsynchronously(String, AsynchronousQueryListener)
*/
QueryCancellor queryAsynchronously(String cql, Runnable listener);
Cancellable queryAsynchronously(String cql, Runnable listener);
/**
* Executes the provided CQL Select with the provided {@link Runnable}, which is started after the query has
@@ -331,10 +370,10 @@ public interface CqlOperations {
*
* @param select The Select Query
* @param listener {@link Runnable} listener for handling the query in a separate thread
* @return A {@link QueryCancellor} that can be used to cancel the query.
* @return A {@link Cancellable} that can be used to cancel the query.
* @see #queryAsynchronously(Select, AsynchronousQueryListener)
*/
QueryCancellor queryAsynchronously(Select select, Runnable listener);
Cancellable queryAsynchronously(Select select, Runnable listener);
/**
* Executes the provided CQL Query with the provided listener. This is preferred over the same method that takes a
@@ -344,9 +383,9 @@ public interface CqlOperations {
* @param cql The Query
* @param listener {@link AsynchronousQueryListener} for handling the query's {@link ResultSetFuture} in a separate
* thread
* @return A {@link QueryCancellor} that can be used to cancel the query.
* @return A {@link Cancellable} that can be used to cancel the query.
*/
QueryCancellor queryAsynchronously(String cql, AsynchronousQueryListener listener);
Cancellable queryAsynchronously(String cql, AsynchronousQueryListener listener);
/**
* Executes the provided CQL Select with the provided listener. This is preferred over the same method that takes a
@@ -356,9 +395,9 @@ public interface CqlOperations {
* @param select The Select
* @param listener {@link AsynchronousQueryListener} for handling the query's {@link ResultSetFuture} in a separate
* thread
* @return A {@link QueryCancellor} that can be used to cancel the query.
* @return A {@link Cancellable} that can be used to cancel the query.
*/
QueryCancellor queryAsynchronously(Select select, AsynchronousQueryListener listener);
Cancellable queryAsynchronously(Select select, AsynchronousQueryListener listener);
/**
* Executes the provided CQL Query with the Runnable implementations using the Query Options.
@@ -366,9 +405,9 @@ public interface CqlOperations {
* @param cql The Query
* @param options Query Option
* @param listener Runnable Listener for handling the query in a separate thread
* @return A {@link QueryCancellor} that can be used to cancel the query.
* @return A {@link Cancellable} that can be used to cancel the query.
*/
QueryCancellor queryAsynchronously(String cql, Runnable listener, QueryOptions options);
Cancellable queryAsynchronously(String cql, Runnable listener, QueryOptions options);
/**
* Executes the provided CQL Query with the provided Listener and Query Options. This is preferred over the same
@@ -378,9 +417,9 @@ public interface CqlOperations {
* @param cql The Query
* @param options Query Option
* @param listener Runnable Listener for handling the query in a separate thread
* @return A {@link QueryCancellor} that can be used to cancel the query.
* @return A {@link Cancellable} that can be used to cancel the query.
*/
QueryCancellor queryAsynchronously(String cql, AsynchronousQueryListener listener, QueryOptions options);
Cancellable queryAsynchronously(String cql, AsynchronousQueryListener listener, QueryOptions options);
/**
* Executes the provided CQL Query with the provided Executor and Runnable implementations.
@@ -388,9 +427,9 @@ public interface CqlOperations {
* @param cql The Query
* @param listener Runnable Listener for handling the query in a separate thread
* @param executor To execute the Runnable Listener
* @return A {@link QueryCancellor} that can be used to cancel the query.
* @return A {@link Cancellable} that can be used to cancel the query.
*/
QueryCancellor queryAsynchronously(String cql, Runnable listener, Executor executor);
Cancellable queryAsynchronously(String cql, Runnable listener, Executor executor);
/**
* Executes the provided CQL Select with the provided Executor and Runnable implementations.
@@ -398,9 +437,9 @@ public interface CqlOperations {
* @param select The Select Query
* @param listener Runnable Listener for handling the query in a separate thread
* @param executor To execute the Runnable Listener
* @return A {@link QueryCancellor} that can be used to cancel the query.
* @return A {@link Cancellable} that can be used to cancel the query.
*/
QueryCancellor queryAsynchronously(Select select, Runnable listener, Executor executor);
Cancellable queryAsynchronously(Select select, Runnable listener, Executor executor);
/**
* Executes the provided CQL Query with the provided listener and executor. This is preferred over the same method
@@ -411,9 +450,9 @@ public interface CqlOperations {
* @param options Query Option
* @param listener Runnable Listener for handling the query in a separate thread
* @param executor To execute the Runnable Listener
* @return A {@link QueryCancellor} that can be used to cancel the query.
* @return A {@link Cancellable} that can be used to cancel the query.
*/
QueryCancellor queryAsynchronously(String cql, AsynchronousQueryListener listener, Executor executor);
Cancellable queryAsynchronously(String cql, AsynchronousQueryListener listener, Executor executor);
/**
* Executes the provided Select Query with the provided listener and executor. This is preferred over the same method
@@ -423,9 +462,9 @@ public interface CqlOperations {
* @param select The Select Query
* @param listener Runnable Listener for handling the query in a separate thread
* @param executor To execute the Runnable Listener
* @return A {@link QueryCancellor} that can be used to cancel the query.
* @return A {@link Cancellable} that can be used to cancel the query.
*/
QueryCancellor queryAsynchronously(Select select, AsynchronousQueryListener listener, Executor executor);
Cancellable queryAsynchronously(Select select, AsynchronousQueryListener listener, Executor executor);
/**
* Executes the provided CQL Query with the provided Executor and Runnable implementations.
@@ -434,9 +473,9 @@ public interface CqlOperations {
* @param options Query Option
* @param listener Runnable Listener for handling the query in a separate thread
* @param executor To execute the Runnable Listener
* @return A {@link QueryCancellor} that can be used to cancel the query.
* @return A {@link Cancellable} that can be used to cancel the query.
*/
QueryCancellor queryAsynchronously(String cql, Runnable listener, QueryOptions options, Executor executor);
Cancellable queryAsynchronously(String cql, Runnable listener, QueryOptions options, Executor executor);
/**
* Executes the provided CQL Query with the provided Listener, Executor and Query Options. This is preferred over the
@@ -447,9 +486,9 @@ public interface CqlOperations {
* @param listener
* @param options
* @param executor
* @return A {@link QueryCancellor} that can be used to cancel the query.
* @return A {@link Cancellable} that can be used to cancel the query.
*/
QueryCancellor queryAsynchronously(String cql, AsynchronousQueryListener listener, QueryOptions options,
Cancellable queryAsynchronously(String cql, AsynchronousQueryListener listener, QueryOptions options,
Executor executor);
/**
@@ -590,6 +629,43 @@ public interface CqlOperations {
*/
<T> List<T> process(ResultSet resultSet, RowMapper<T> rowMapper) throws DataAccessException;
/**
* Executes the provided string CQL query, and maps the first row returned with the supplied {@link RowMapper}.
*
* @param cql The string query CQL.
* @param rowMapper The {@link RowMapper} to convert the row into an object of type <code>T</code>.
* @param listener The listener that receives the results upon completion.
* @return A {@link Cancellable} that can be used to cancel the query.
* @throws DataAccessException
*/
<T> Cancellable queryForObjectAsynchronously(String cql, RowMapper<T> rowMapper, QueryForObjectListener<T> listener)
throws DataAccessException;
/**
* Executes the provided string CQL query, and maps the first row returned with the supplied {@link RowMapper}.
*
* @param cql The string query CQL.
* @param rowMapper The {@link RowMapper} to convert the row into an object of type <code>T</code>.
* @param listener The listener that receives the results upon completion.
* @param options The {@link QueryOptions} to use. May be <code>null</code>.
* @return A {@link Cancellable} that can be used to cancel the query.
* @throws DataAccessException
*/
<T> Cancellable queryForObjectAsynchronously(String cql, RowMapper<T> rowMapper, QueryForObjectListener<T> listener,
QueryOptions options) throws DataAccessException;
/**
* Executes the provided {@link Select} query, and maps the first row returned with the supplied {@link RowMapper}.
*
* @param select The {@link Select} query to execute.
* @param rowMapper The {@link RowMapper} to convert the row into an object of type <code>T</code>.
* @param listener The listener that receives the results upon completion.
* @return A {@link Cancellable} that can be used to cancel the query.
* @throws DataAccessException
*/
<T> Cancellable queryForObjectAsynchronously(Select select, RowMapper<T> rowMapper, QueryForObjectListener<T> listener)
throws DataAccessException;
/**
* Executes the provided CQL Query, and maps <b>ONE</b> Row returned with the supplied RowMapper.
* <p>
@@ -638,6 +714,43 @@ public interface CqlOperations {
*/
<T> T queryForObject(String cql, Class<T> requiredType) throws DataAccessException;
/**
* Executes the provided {@link Select} query and returns the first column of the first Row as an object of type
* <code>T</code>.
*
* @param select The {@link Select} query
* @param requiredType Type that Cassandra data types can be converted to.
* @return A {@link Cancellable} that can be used to cancel the query if necessary.
* @throws DataAccessException
*/
<T> Cancellable queryForObjectAsynchronously(Select select, Class<T> requiredType, QueryForObjectListener<T> listener)
throws DataAccessException;
/**
* Executes the provided select CQL query and returns the first column of the first Row as an object of type
* <code>T</code>.
*
* @param cql The select query CQL. Must not be <code>null</code> or blank.
* @param requiredType The type to convert the first column of the first row to. Must not be <code>null</code>.
* @return A {@link Cancellable} that can be used to cancel the query if necessary. Must not be <code>null</code>.
* @throws DataAccessException
*/
<T> Cancellable queryForObjectAsynchronously(String cql, Class<T> requiredType, QueryForObjectListener<T> listener)
throws DataAccessException;
/**
* Executes the provided select CQL query and returns the first column of the first Row as an object of type
* <code>T</code>.
*
* @param cql The select query CQL. Must not be <code>null</code> or blank.
* @param requiredType The type to convert the first column of the first row to. Must not be <code>null</code>.
* @param options The {@link QueryOptions} to use. May be <code>null</code>.
* @return A {@link Cancellable} that can be used to cancel the query if necessary. Must not be <code>null</code>.
* @throws DataAccessException
*/
<T> Cancellable queryForObjectAsynchronously(String cql, Class<T> requiredType, QueryForObjectListener<T> listener,
QueryOptions options) throws DataAccessException;
/**
* Executes the provided Select query and tries to return the first column of the first Row as a Class<T>.
*
@@ -680,6 +793,44 @@ public interface CqlOperations {
*/
Map<String, Object> queryForMap(Select select) throws DataAccessException;
/**
* Executes the provided CQL query asynchronously and maps the first row to a {@link Map}&lt;String,Object&gt;.
* Additional rows are ignored.
*
* @param cql The select query CQL. Must not be <code>null</code> or blank.
* @param listener The {@link QueryForMapListener} that will recieve the results upon query completion. Must not be
* <code>null</code>.
* @return A {@link Cancellable} that can be used to cancel the query if necessary. Must not be <code>null</code>.
* @throws DataAccessException
*/
Cancellable queryForMapAsynchronously(String cql, QueryForMapListener listener) throws DataAccessException;
/**
* Executes the provided CQL query asynchronously and maps the first row to a {@link Map}&lt;String,Object&gt;.
* Additional rows are ignored.
*
* @param cql The select query CQL. Must not be <code>null</code> or blank.
* @param listener The {@link QueryForMapListener} that will recieve the results upon query completion. Must not be
* <code>null</code>.
* @param options The {@link QueryOptions} to use. May be <code>null</code>.
* @return A {@link Cancellable} that can be used to cancel the query if necessary. Must not be <code>null</code>.
* @throws DataAccessException
*/
Cancellable queryForMapAsynchronously(String cql, QueryForMapListener listener, QueryOptions options)
throws DataAccessException;
/**
* Executes the provided {@link Select} query asynchronously and maps the first row to a {@link Map}
* &lt;String,Object&gt;. Additional rows are ignored.
*
* @param cql The select query CQL. Must not be <code>null</code> or blank.
* @param listener The {@link QueryForMapListener} that will recieve the results upon query completion. Must not be
* <code>null</code>.
* @return A {@link Cancellable} that can be used to cancel the query if necessary. Must not be <code>null</code>.
* @throws DataAccessException
*/
Cancellable queryForMapAsynchronously(Select select, QueryForMapListener listener) throws DataAccessException;
/**
* Process a ResultSet with <b>ONE</b> Row and convert to a Map. This is used internal to the Template for core
* operations, but is made available through Operations in the event you have a ResultSet to process. The ResultsSet
@@ -713,6 +864,32 @@ public interface CqlOperations {
*/
<T> List<T> queryForList(Select select, Class<T> elementType) throws DataAccessException;
/**
* Executes the provided {@link Select} query asynchronously and returns all values in the first column of the results
* as a {@link List} of the type in the second argument.
*
* @param select The {@link Select} query
* @param elementType The type to cast the data values to
* @param listener The listener to receive the results asynchronously. Must not be <code>null</code>.
* @return {@link Cancellable} to cancel the query if necessary
* @throws DataAccessException
*/
<T> Cancellable queryForListAsynchronously(Select select, Class<T> elementType, QueryForListListener<T> listener)
throws DataAccessException;
/**
* Executes the provided {@link Select} query asynchronously and returns all values in the first column of the results
* as a {@link List} of the type in the second argument.
*
* @param select The select query CQL
* @param elementType The type to cast the data values to
* @param listener The listener to receive the results asynchronously. Must not be <code>null</code>.
* @return {@link Cancellable} to cancel the query if necessary
* @throws DataAccessException
*/
<T> Cancellable queryForListAsynchronously(String select, Class<T> elementType, QueryForListListener<T> listener)
throws DataAccessException;
/**
* Process a ResultSet and convert the first column of the results to a List. This is used internal to the Template
* for core operations, but is made available through Operations in the event you have a ResultSet to process. The
@@ -735,6 +912,46 @@ public interface CqlOperations {
*/
List<Map<String, Object>> queryForListOfMap(String cql) throws DataAccessException;
/**
* Executes the provided {@link Select} query and converts the results to a {@link List} of {@link Map}s. Each element
* in the {@link List} represents a row returned from the query. Each row's column(s) are put into a {@link Map} as
* values keyed by column name.
*
* @param select The {@link Select} query. Must not be <code>null</code>.
* @param listener The listener that will receive the results upon query completion. Must not be <code>null</code>.
* @return A {@link Cancellable} that can be used to cancel the query. Must not be <code>null</code>.
* @throws DataAccessException
*/
Cancellable queryForListOfMapAsynchronously(Select select, QueryForListListener<Map<String, Object>> listener)
throws DataAccessException;
/**
* Executes the provided select CQL query and converts the results to a {@link List} of {@link Map}s. Each element in
* the {@link List} represents a row returned from the query. Each row's column(s) are put into a {@link Map} as
* values keyed by column name.
*
* @param select The select query CQL. Must not be <code>null</code> or blank.
* @param listener The listener that will receive the results upon query completion. Must not be <code>null</code>.
* @return A {@link Cancellable} that can be used to cancel the query. Must not be <code>null</code>.
* @throws DataAccessException
*/
Cancellable queryForListOfMapAsynchronously(String cql, QueryForListListener<Map<String, Object>> listener)
throws DataAccessException;
/**
* Executes the provided select CQL query and converts the results to a {@link List} of {@link Map}s. Each element in
* the {@link List} represents a row returned from the query. Each row's column(s) are put into a {@link Map} as
* values keyed by column name.
*
* @param select The select query CQL. Must not be <code>null</code> or blank.
* @param listener The listener that will receive the results upon query completion. Must not be <code>null</code>.
* @param options The {@link QueryOptions} to use. May be <code>null</code>.
* @return A {@link Cancellable} that can be used to cancel the query. Must not be <code>null</code>.
* @throws DataAccessException
*/
Cancellable queryForListOfMapAsynchronously(String cql, QueryForListListener<Map<String, Object>> listener,
QueryOptions options) throws DataAccessException;
/**
* Executes the provided Select Query and converts the results to a basic List of Maps. Each element in the List
* represents a Row returned from the Query. Each Row's columns are put into the map as column/value.

396
spring-cql/src/main/java/org/springframework/cassandra/core/CqlTemplate.java
View File

@@ -279,7 +279,7 @@ public class CqlTemplate extends CassandraAccessor implements CqlOperations {
}
@Override
public QueryCancellor queryAsynchronously(String cql, Runnable listener) {
public Cancellable queryAsynchronously(String cql, Runnable listener) {
return queryAsynchronously(cql, listener, new Executor() {
@Override
public void execute(Runnable command) {
@@ -289,7 +289,7 @@ public class CqlTemplate extends CassandraAccessor implements CqlOperations {
}
@Override
public QueryCancellor queryAsynchronously(String cql, AsynchronousQueryListener listener) {
public Cancellable queryAsynchronously(String cql, AsynchronousQueryListener listener) {
return queryAsynchronously(cql, listener, new Executor() {
@Override
public void execute(Runnable command) {
@@ -299,7 +299,7 @@ public class CqlTemplate extends CassandraAccessor implements CqlOperations {
}
@Override
public QueryCancellor queryAsynchronously(String cql, Runnable listener, QueryOptions options) {
public Cancellable queryAsynchronously(String cql, Runnable listener, QueryOptions options) {
return queryAsynchronously(cql, listener, options, new Executor() {
@Override
public void execute(Runnable command) {
@@ -309,7 +309,7 @@ public class CqlTemplate extends CassandraAccessor implements CqlOperations {
}
@Override
public QueryCancellor queryAsynchronously(String cql, AsynchronousQueryListener listener, QueryOptions options) {
public Cancellable queryAsynchronously(String cql, AsynchronousQueryListener listener, QueryOptions options) {
return queryAsynchronously(cql, listener, options, new Executor() {
@Override
public void execute(Runnable command) {
@@ -319,36 +319,36 @@ public class CqlTemplate extends CassandraAccessor implements CqlOperations {
}
@Override
public QueryCancellor queryAsynchronously(String cql, Runnable listener, Executor executor) {
public Cancellable queryAsynchronously(String cql, Runnable listener, Executor executor) {
return queryAsynchronously(cql, listener, null, executor);
}
@Override
public QueryCancellor queryAsynchronously(String cql, AsynchronousQueryListener listener, Executor executor) {
public Cancellable queryAsynchronously(String cql, AsynchronousQueryListener listener, Executor executor) {
return queryAsynchronously(cql, listener, null, executor);
}
@Override
public QueryCancellor queryAsynchronously(final String cql, final Runnable listener, final QueryOptions options,
public Cancellable queryAsynchronously(final String cql, final Runnable listener, final QueryOptions options,
final Executor executor) {
return execute(new SessionCallback<QueryCancellor>() {
return execute(new SessionCallback<Cancellable>() {
@Override
public QueryCancellor doInSession(Session s) throws DataAccessException {
public Cancellable doInSession(Session s) throws DataAccessException {
Statement statement = new SimpleStatement(cql);
addQueryOptions(statement, options);
ResultSetFuture rsf = s.executeAsync(statement);
rsf.addListener(listener, executor);
return new BasicQueryCancellor(rsf);
return new ResultSetFutureCancellable(rsf);
}
});
}
@Override
public QueryCancellor queryAsynchronously(final String cql, final AsynchronousQueryListener listener,
public Cancellable queryAsynchronously(final String cql, final AsynchronousQueryListener listener,
final QueryOptions options, final Executor executor) {
return execute(new SessionCallback<QueryCancellor>() {
return execute(new SessionCallback<Cancellable>() {
@Override
public QueryCancellor doInSession(Session s) throws DataAccessException {
public Cancellable doInSession(Session s) throws DataAccessException {
Statement statement = new SimpleStatement(cql);
addQueryOptions(statement, options);
final ResultSetFuture rsf = s.executeAsync(statement);
@@ -359,7 +359,7 @@ public class CqlTemplate extends CassandraAccessor implements CqlOperations {
}
};
rsf.addListener(wrapper, executor);
return new BasicQueryCancellor(rsf);
return new ResultSetFutureCancellable(rsf);
}
});
}
@@ -518,6 +518,48 @@ public class CqlTemplate extends CassandraAccessor implements CqlOperations {
});
}
protected Cancellable doExecuteAsync(final Statement q, final AsynchronousQueryListener listener) {
return doExecuteAsync(q, listener, null);
}
protected Cancellable doExecuteAsync(final Statement q, final AsynchronousQueryListener listener,
final QueryOptions options) {
return doExecute(new SessionCallback<Cancellable>() {
@Override
public Cancellable doInSession(Session s) throws DataAccessException {
if (log.isDebugEnabled()) {
log.debug("asynchronously executing [{}]", q.toString());
}
if (options != null) {
addQueryOptions(q, options);
}
final ResultSetFuture rsf = s.executeAsync(q);
if (listener != null) {
rsf.addListener(new Runnable() {
@Override
public void run() {
listener.onQueryComplete(rsf);
}
}, new Executor() {
@Override
public void execute(Runnable command) {
command.run();
}
});
}
return new ResultSetFutureCancellable(rsf);
}
});
}
/**
* @param row
* @return
@@ -588,7 +630,7 @@ public class CqlTemplate extends CassandraAccessor implements CqlOperations {
}
@Override
public QueryCancellor executeAsynchronously(String cql, Runnable listener) throws DataAccessException {
public Cancellable executeAsynchronously(String cql, Runnable listener) throws DataAccessException {
return executeAsynchronously(cql, listener, new Executor() {
@Override
public void execute(Runnable command) {
@@ -598,23 +640,22 @@ public class CqlTemplate extends CassandraAccessor implements CqlOperations {
}
@Override
public QueryCancellor executeAsynchronously(final String cql, final Runnable listener, final Executor executor)
public Cancellable executeAsynchronously(final String cql, final Runnable listener, final Executor executor)
throws DataAccessException {
return execute(new SessionCallback<QueryCancellor>() {
return execute(new SessionCallback<Cancellable>() {
@Override
public QueryCancellor doInSession(Session s) throws DataAccessException {
public Cancellable doInSession(Session s) throws DataAccessException {
Statement statement = new SimpleStatement(cql);
final ResultSetFuture rsf = s.executeAsync(statement);
rsf.addListener(listener, executor);
return new BasicQueryCancellor(rsf);
return new ResultSetFutureCancellable(rsf);
}
});
}
@Override
public QueryCancellor executeAsynchronously(String cql, AsynchronousQueryListener listener)
throws DataAccessException {
public Cancellable executeAsynchronously(String cql, AsynchronousQueryListener listener) throws DataAccessException {
return executeAsynchronously(cql, listener, new Executor() {
@Override
@@ -625,12 +666,12 @@ public class CqlTemplate extends CassandraAccessor implements CqlOperations {
}
@Override
public QueryCancellor executeAsynchronously(final String cql, final AsynchronousQueryListener listener,
public Cancellable executeAsynchronously(final String cql, final AsynchronousQueryListener listener,
final Executor executor) throws DataAccessException {
return execute(new SessionCallback<QueryCancellor>() {
return execute(new SessionCallback<Cancellable>() {
@Override
public QueryCancellor doInSession(Session s) throws DataAccessException {
public Cancellable doInSession(Session s) throws DataAccessException {
Statement statement = new SimpleStatement(cql);
final ResultSetFuture rsf = s.executeAsync(statement);
Runnable wrapper = new Runnable() {
@@ -640,7 +681,7 @@ public class CqlTemplate extends CassandraAccessor implements CqlOperations {
}
};
rsf.addListener(wrapper, executor);
return new BasicQueryCancellor(rsf);
return new ResultSetFutureCancellable(rsf);
}
});
}
@@ -651,7 +692,7 @@ public class CqlTemplate extends CassandraAccessor implements CqlOperations {
}
@Override
public QueryCancellor executeAsynchronously(Statement query, Runnable listener) throws DataAccessException {
public Cancellable executeAsynchronously(Statement query, Runnable listener) throws DataAccessException {
return executeAsynchronously(query, listener, new Executor() {
@Override
public void execute(Runnable command) {
@@ -661,7 +702,7 @@ public class CqlTemplate extends CassandraAccessor implements CqlOperations {
}
@Override
public QueryCancellor executeAsynchronously(Statement query, AsynchronousQueryListener listener)
public Cancellable executeAsynchronously(Statement query, AsynchronousQueryListener listener)
throws DataAccessException {
return executeAsynchronously(query, listener, new Executor() {
@Override
@@ -672,34 +713,36 @@ public class CqlTemplate extends CassandraAccessor implements CqlOperations {
}
@Override
public QueryCancellor executeAsynchronously(final Statement query, final Runnable listener, final Executor executor)
public Cancellable executeAsynchronously(final Statement query, final Runnable listener, final Executor executor)
throws DataAccessException {
return execute(new SessionCallback<QueryCancellor>() {
return execute(new SessionCallback<Cancellable>() {
@Override
public QueryCancellor doInSession(Session s) throws DataAccessException {
public Cancellable doInSession(Session s) throws DataAccessException {
final ResultSetFuture rsf = s.executeAsync(query);
rsf.addListener(listener, executor);
return new BasicQueryCancellor(rsf);
return new ResultSetFutureCancellable(rsf);
}
});
}
@Override
public QueryCancellor executeAsynchronously(final Statement query, final AsynchronousQueryListener listener,
public Cancellable executeAsynchronously(final Statement query, final AsynchronousQueryListener listener,
final Executor executor) throws DataAccessException {
return execute(new SessionCallback<QueryCancellor>() {
return execute(new SessionCallback<Cancellable>() {
@Override
public QueryCancellor doInSession(Session s) throws DataAccessException {
public Cancellable doInSession(Session s) throws DataAccessException {
final ResultSetFuture rsf = s.executeAsync(query);
Runnable wrapper = new Runnable() {
@Override
public void run() {
listener.onQueryComplete(rsf);
}
};
rsf.addListener(wrapper, executor);
return new BasicQueryCancellor(rsf);
if (listener != null) {
Runnable wrapper = new Runnable() {
@Override
public void run() {
listener.onQueryComplete(rsf);
}
};
rsf.addListener(wrapper, executor);
}
return new ResultSetFutureCancellable(rsf);
}
});
}
@@ -1231,6 +1274,35 @@ public class CqlTemplate extends CassandraAccessor implements CqlOperations {
return doExecuteAsync(batch);
}
@Override
public Cancellable executeAsynchronously(Truncate truncate, AsynchronousQueryListener listener)
throws DataAccessException {
return doExecuteAsync(truncate, listener);
}
@Override
public Cancellable executeAsynchronously(Delete delete, AsynchronousQueryListener listener)
throws DataAccessException {
return doExecuteAsync(delete, listener);
}
@Override
public Cancellable executeAsynchronously(Insert insert, AsynchronousQueryListener listener)
throws DataAccessException {
return doExecuteAsync(insert, listener);
}
@Override
public Cancellable executeAsynchronously(Update update, AsynchronousQueryListener listener)
throws DataAccessException {
return doExecuteAsync(update, listener);
}
@Override
public Cancellable executeAsynchronously(Batch batch, AsynchronousQueryListener listener) throws DataAccessException {
return doExecuteAsync(batch, listener);
}
@Override
public ResultSetFuture queryAsynchronously(final Select select) {
return execute(new SessionCallback<ResultSetFuture>() {
@@ -1242,7 +1314,7 @@ public class CqlTemplate extends CassandraAccessor implements CqlOperations {
}
@Override
public QueryCancellor queryAsynchronously(Select select, Runnable listener) {
public Cancellable queryAsynchronously(Select select, Runnable listener) {
return queryAsynchronously(select, listener, new Executor() {
@Override
public void execute(Runnable command) {
@@ -1252,7 +1324,7 @@ public class CqlTemplate extends CassandraAccessor implements CqlOperations {
}
@Override
public QueryCancellor queryAsynchronously(Select select, AsynchronousQueryListener listener) {
public Cancellable queryAsynchronously(Select select, AsynchronousQueryListener listener) {
return queryAsynchronously(select, listener, new Executor() {
@Override
public void execute(Runnable command) {
@@ -1262,12 +1334,12 @@ public class CqlTemplate extends CassandraAccessor implements CqlOperations {
}
@Override
public QueryCancellor queryAsynchronously(final Select select, final AsynchronousQueryListener listener,
public Cancellable queryAsynchronously(final Select select, final AsynchronousQueryListener listener,
final Executor executor) {
return execute(new SessionCallback<QueryCancellor>() {
return execute(new SessionCallback<Cancellable>() {
@Override
public QueryCancellor doInSession(Session s) throws DataAccessException {
public Cancellable doInSession(Session s) throws DataAccessException {
final ResultSetFuture rsf = s.executeAsync(select);
Runnable wrapper = new Runnable() {
@Override
@@ -1276,19 +1348,19 @@ public class CqlTemplate extends CassandraAccessor implements CqlOperations {
}
};
rsf.addListener(wrapper, executor);
return new BasicQueryCancellor(rsf);
return new ResultSetFutureCancellable(rsf);
}
});
}
@Override
public QueryCancellor queryAsynchronously(final Select select, final Runnable listener, final Executor executor) {
return execute(new SessionCallback<QueryCancellor>() {
public Cancellable queryAsynchronously(final Select select, final Runnable listener, final Executor executor) {
return execute(new SessionCallback<Cancellable>() {
@Override
public QueryCancellor doInSession(Session s) throws DataAccessException {
public Cancellable doInSession(Session s) throws DataAccessException {
ResultSetFuture rsf = s.executeAsync(select);
rsf.addListener(listener, executor);
return new BasicQueryCancellor(rsf);
return new ResultSetFutureCancellable(rsf);
}
});
}
@@ -1345,4 +1417,224 @@ public class CqlTemplate extends CassandraAccessor implements CqlOperations {
return processListOfMap(doExecute(select));
}
@Override
public <T> Cancellable queryForListAsynchronously(Select select, final Class<T> elementType,
final QueryForListListener<T> listener) throws DataAccessException {
Assert.notNull(select);
Assert.notNull(elementType);
Assert.notNull(listener);
return doExecuteAsync(select, new AsynchronousQueryListener() {
@Override
public void onQueryComplete(ResultSetFuture rsf) {
try {
listener.onQueryComplete(processList(rsf.getUninterruptibly(), elementType));
} catch (Exception e) {
listener.onException(translateExceptionIfPossible(e));
}
}
});
}
@Override
public <T> Cancellable queryForListAsynchronously(String select, final Class<T> elementType,
final QueryForListListener<T> listener) throws DataAccessException {
Assert.hasText(select);
Assert.notNull(elementType);
Assert.notNull(listener);
return doExecuteAsync(new SimpleStatement(select), new AsynchronousQueryListener() {
@Override
public void onQueryComplete(ResultSetFuture rsf) {
try {
listener.onQueryComplete(processList(rsf.getUninterruptibly(), elementType));
} catch (Exception e) {
listener.onException(translateExceptionIfPossible(e));
}
}
});
}
@Override
public Cancellable queryForListOfMapAsynchronously(Select select,
final QueryForListListener<Map<String, Object>> listener) throws DataAccessException {
return doExecuteAsync(select, new AsynchronousQueryListener() {
@Override
public void onQueryComplete(ResultSetFuture rsf) {
try {
listener.onQueryComplete(processListOfMap(rsf.getUninterruptibly()));
} catch (Exception e) {
listener.onException(translateExceptionIfPossible(e));
}
}
});
}
@Override
public Cancellable queryForListOfMapAsynchronously(String cql,
final QueryForListListener<Map<String, Object>> listener) throws DataAccessException {
return queryForListOfMapAsynchronously(cql, listener, null);
}
@Override
public Cancellable queryForListOfMapAsynchronously(String cql,
final QueryForListListener<Map<String, Object>> listener, QueryOptions options) throws DataAccessException {
return doExecuteAsync(new SimpleStatement(cql), new AsynchronousQueryListener() {
@Override
public void onQueryComplete(ResultSetFuture rsf) {
try {
listener.onQueryComplete(processListOfMap(rsf.getUninterruptibly()));
} catch (Exception e) {
listener.onException(translateExceptionIfPossible(e));
}
}
}, options);
}
@Override
public Cancellable queryForMapAsynchronously(String cql, QueryForMapListener listener) throws DataAccessException {
return queryForMapAsynchronously(cql, listener, null);
}
@Override
public Cancellable queryForMapAsynchronously(String cql, final QueryForMapListener listener,
final QueryOptions options) throws DataAccessException {
return doExecuteAsync(new SimpleStatement(cql), new AsynchronousQueryListener() {
@Override
public void onQueryComplete(ResultSetFuture rsf) {
try {
listener.onQueryComplete(processMap(rsf.getUninterruptibly()));
} catch (Exception e) {
listener.onException(translateExceptionIfPossible(e));
}
}
}, options);
}
@Override
public Cancellable queryForMapAsynchronously(Select select, final QueryForMapListener listener)
throws DataAccessException {
return doExecuteAsync(select, new AsynchronousQueryListener() {
@Override
public void onQueryComplete(ResultSetFuture rsf) {
try {
listener.onQueryComplete(processMap(rsf.getUninterruptibly()));
} catch (Exception e) {
listener.onException(translateExceptionIfPossible(e));
}
}
});
}
@Override
public <T> Cancellable queryForObjectAsynchronously(Select select, final Class<T> requiredType,
final QueryForObjectListener<T> listener) throws DataAccessException {
return doExecuteAsync(select, new AsynchronousQueryListener() {
@Override
public void onQueryComplete(ResultSetFuture rsf) {
try {
listener.onQueryComplete(processOne(rsf.getUninterruptibly(), requiredType));
} catch (Exception e) {
listener.onException(translateExceptionIfPossible(e));
}
}
});
}
@Override
public <T> Cancellable queryForObjectAsynchronously(String cql, Class<T> requiredType,
QueryForObjectListener<T> listener) throws DataAccessException {
return queryForObjectAsynchronously(cql, requiredType, listener, null);
}
@Override
public <T> Cancellable queryForObjectAsynchronously(String cql, final Class<T> requiredType,
final QueryForObjectListener<T> listener, QueryOptions options) throws DataAccessException {
return doExecuteAsync(new SimpleStatement(cql), new AsynchronousQueryListener() {
@Override
public void onQueryComplete(ResultSetFuture rsf) {
try {
listener.onQueryComplete(processOne(rsf.getUninterruptibly(), requiredType));
} catch (Exception e) {
listener.onException(translateExceptionIfPossible(e));
}
}
}, options);
}
@Override
public <T> Cancellable queryForObjectAsynchronously(String cql, RowMapper<T> rowMapper,
QueryForObjectListener<T> listener) throws DataAccessException {
return queryForObjectAsynchronously(cql, rowMapper, listener, null);
}
@Override
public <T> Cancellable queryForObjectAsynchronously(String cql, final RowMapper<T> rowMapper,
final QueryForObjectListener<T> listener, QueryOptions options) throws DataAccessException {
return doExecuteAsync(new SimpleStatement(cql), new AsynchronousQueryListener() {
@Override
public void onQueryComplete(ResultSetFuture rsf) {
try {
listener.onQueryComplete(processOne(rsf.getUninterruptibly(), rowMapper));
} catch (Exception e) {
listener.onException(translateExceptionIfPossible(e));
}
}
}, options);
}
@Override
public <T> Cancellable queryForObjectAsynchronously(Select select, final RowMapper<T> rowMapper,
final QueryForObjectListener<T> listener) throws DataAccessException {
return doExecuteAsync(select, new AsynchronousQueryListener() {
@Override
public void onQueryComplete(ResultSetFuture rsf) {
try {
listener.onQueryComplete(processOne(rsf.getUninterruptibly(), rowMapper));
} catch (Exception e) {
listener.onException(translateExceptionIfPossible(e));
}
}
});
}
@Override
public ResultSet getResultSetUninterruptibly(ResultSetFuture rsf) {
return getResultSetUninterruptibly(rsf, 0, null);
}
@Override
public ResultSet getResultSetUninterruptibly(ResultSetFuture rsf, long millis) {
return getResultSetUninterruptibly(rsf, millis, TimeUnit.MILLISECONDS);
}
@Override
public ResultSet getResultSetUninterruptibly(ResultSetFuture rsf, long timeout, TimeUnit unit) {
try {
return timeout <= 0 ? rsf.getUninterruptibly() : rsf.getUninterruptibly(timeout,
unit == null ? TimeUnit.MILLISECONDS : unit);
} catch (Exception x) {
throw translateExceptionIfPossible(x);
}
}
}

1
spring-cql/src/main/java/org/springframework/cassandra/core/PreparedStatementBinder.java
View File

@@ -21,7 +21,6 @@ import com.datastax.driver.core.exceptions.DriverException;
/**
* @author David Webb
*
*/
public interface PreparedStatementBinder {

1
spring-cql/src/main/java/org/springframework/cassandra/core/PreparedStatementCallback.java
View File

@@ -22,7 +22,6 @@ import com.datastax.driver.core.exceptions.DriverException;
/**
* @author David Webb
*
*/
public interface PreparedStatementCallback<T> {

1
spring-cql/src/main/java/org/springframework/cassandra/core/PreparedStatementCreator.java
View File

@@ -23,7 +23,6 @@ import com.datastax.driver.core.exceptions.DriverException;
* Creates a PreparedStatement for the usage with the DataStax Java Driver
*
* @author David Webb
*
*/
public interface PreparedStatementCreator {

24
spring-cql/src/main/java/org/springframework/cassandra/core/QueryForListListener.java Normal file
View File

@@ -0,0 +1,24 @@
package org.springframework.cassandra.core;
import java.util.List;
import com.datastax.driver.core.ResultSet;
/**
* Listener used to receive asynchronous results expected as a <code>List&lt;T&gt;</code>.
*
* @author Matthew T. Adams
* @param <T>
*/
public interface QueryForListListener<T> {
/**
* Called upon query completion.
*/
void onQueryComplete(List<T> results);
/**
* Called if an exception is raised while getting or converting the {@link ResultSet}.
*/
void onException(Exception x);
}

10
spring-cql/src/main/java/org/springframework/cassandra/core/QueryForListOfMapListener.java Normal file
View File

@@ -0,0 +1,10 @@
package org.springframework.cassandra.core;
import java.util.Map;
/**
* Listener used to receive asynchronous results expected as a <code>List&lt;T&gt;</code>.
*
* @author Matthew T. Adams
*/
public interface QueryForListOfMapListener extends QueryForListListener<Map<String, Object>> {}

24
spring-cql/src/main/java/org/springframework/cassandra/core/QueryForMapListener.java Normal file
View File

@@ -0,0 +1,24 @@
package org.springframework.cassandra.core;
import java.util.Map;
import com.datastax.driver.core.ResultSet;
/**
* Listener used to receive asynchronous results expected as a <code>List&lt;Map&lt;String,Object&gt;&gt;</code>.
*
* @author Matthew T. Adams
* @param <T>
*/
public interface QueryForMapListener {
/**
* Called upon query completion.
*/
void onQueryComplete(Map<String, Object> results);
/**
* Called if an exception is raised while getting or converting the {@link ResultSet}.
*/
void onException(Exception x);
}

22
spring-cql/src/main/java/org/springframework/cassandra/core/QueryForObjectListener.java Normal file
View File

@@ -0,0 +1,22 @@
package org.springframework.cassandra.core;
import com.datastax.driver.core.ResultSet;
/**
* Listener used to receive asynchronous results expected as an object of type <code>T</code>.
*
* @author Matthew T. Adams
* @param <T>
*/
public interface QueryForObjectListener<T> {
/**
* Called upon query completion.
*/
void onQueryComplete(T result);
/**
* Called if an exception is raised while getting or converting the {@link ResultSet}.
*/
void onException(Exception x);
}

8
spring-cql/src/main/java/org/springframework/cassandra/core/QueryOptions.java
View File

@@ -19,13 +19,19 @@ package org.springframework.cassandra.core;
* Contains Query Options for Cassandra queries. This controls the Consistency Tuning and Retry Policy for a Query.
*
* @author David Webb
*
*/
public class QueryOptions {
private ConsistencyLevel consistencyLevel;
private RetryPolicy retryPolicy;
public QueryOptions() {}
public QueryOptions(ConsistencyLevel consistencyLevel, RetryPolicy retryPolicy) {
setConsistencyLevel(consistencyLevel);
setRetryPolicy(retryPolicy);
}
/**
* @return Returns the consistencyLevel.
*/

50
spring-cql/src/main/java/org/springframework/cassandra/core/ReservedKeyword.java
View File

@@ -8,11 +8,57 @@ import org.springframework.util.StringUtils;
*
* @see <a
* href="http://cassandra.apache.org/doc/cql3/CQL.html#appendixA">http://cassandra.apache.org/doc/cql3/CQL.html#appendixA</a>
*
* @author Matthew T. Adams
*/
public enum ReservedKeyword {
ADD, ALTER, AND, ANY, APPLY, ASC, AUTHORIZE, BATCH, BEGIN, BY, COLUMNFAMILY, CREATE, DELETE, DESC, DROP, EACH_QUORUM, FROM, GRANT, IN, INDEX, INSERT, INTO, KEYSPACE, LIMIT, LOCAL_ONE, LOCAL_QUORUM, MODIFY, NORECURSIVE, OF, ON, ONE, ORDER, PRIMARY, QUORUM, REVOKE, SCHEMA, SELECT, SET, TABLE, THREE, TOKEN, TRUNCATE, TWO, UPDATE, USE, USING, WHERE, WITH;
ADD,
ALTER,
AND,
ANY,
APPLY,
ASC,
AUTHORIZE,
BATCH,
BEGIN,
BY,
COLUMNFAMILY,
CREATE,
DELETE,
DESC,
DROP,
EACH_QUORUM,
FROM,
GRANT,
IN,
INDEX,
INSERT,
INTO,
KEYSPACE,
LIMIT,
LOCAL_ONE,
LOCAL_QUORUM,
MODIFY,
NORECURSIVE,
OF,
ON,
ONE,
ORDER,
PRIMARY,
QUORUM,
REVOKE,
SCHEMA,
SELECT,
SET,
TABLE,
THREE,
TOKEN,
TRUNCATE,
TWO,
UPDATE,
USE,
USING,
WHERE,
WITH;
/**
* @see ReservedKeyword#isReserved(String)

10
spring-cql/src/main/java/org/springframework/cassandra/core/BasicQueryCancellor.java → spring-cql/src/main/java/org/springframework/cassandra/core/ResultSetFutureCancellable.java
View File

@@ -18,20 +18,20 @@ package org.springframework.cassandra.core;
import com.datastax.driver.core.ResultSetFuture;
/**
* Convenient default implementation of a {@link QueryCancellor}.
* Convenient default implementation of a {@link Cancellable}.
*
* @author Matthew T. Adams
*/
public class BasicQueryCancellor implements QueryCancellor {
public class ResultSetFutureCancellable implements Cancellable {
ResultSetFuture rsf;
public BasicQueryCancellor(ResultSetFuture rsf) {
public ResultSetFutureCancellable(ResultSetFuture rsf) {
this.rsf = rsf;
}
@Override
public void cancelQuery(boolean mayInterruptIfRunning) {
rsf.cancel(mayInterruptIfRunning);
public void cancel() {
rsf.cancel(true);
}
}

1
spring-cql/src/main/java/org/springframework/cassandra/core/RetryPolicy.java
View File

@@ -19,7 +19,6 @@ package org.springframework.cassandra.core;
* Retry Policies associated with Cassandra.
*
* @author David Webb
*
*/
public enum RetryPolicy {

28
spring-cql/src/main/java/org/springframework/cassandra/core/RetryPolicyResolver.java
View File

@@ -23,15 +23,13 @@ import com.datastax.driver.core.policies.FallthroughRetryPolicy;
* Determine driver query retry policy
*
* @author David Webb
*
*/
public final class RetryPolicyResolver {
/**
* No instances allowed
*/
private RetryPolicyResolver() {
}
private RetryPolicyResolver() {}
/**
* Decode the generic spring data cassandra enum to the type required by the DataStax Driver.
@@ -47,18 +45,18 @@ public final class RetryPolicyResolver {
* Determine the driver level based on our enum
*/
switch (policy) {
case DEFAULT:
resolvedPolicy = DefaultRetryPolicy.INSTANCE;
break;
case DOWNGRADING_CONSISTENCY:
resolvedPolicy = DowngradingConsistencyRetryPolicy.INSTANCE;
break;
case FALLTHROUGH:
resolvedPolicy = FallthroughRetryPolicy.INSTANCE;
break;
default:
resolvedPolicy = DefaultRetryPolicy.INSTANCE;
break;
case DEFAULT:
resolvedPolicy = DefaultRetryPolicy.INSTANCE;
break;
case DOWNGRADING_CONSISTENCY:
resolvedPolicy = DowngradingConsistencyRetryPolicy.INSTANCE;
break;
case FALLTHROUGH:
resolvedPolicy = FallthroughRetryPolicy.INSTANCE;
break;
default:
resolvedPolicy = DefaultRetryPolicy.INSTANCE;
break;
}
return resolvedPolicy;

1
spring-cql/src/main/java/org/springframework/cassandra/core/RingMember.java
View File

@@ -21,7 +21,6 @@ import com.datastax.driver.core.Host;
/**
* @author David Webb
*
*/
public final class RingMember implements Serializable {

1
spring-cql/src/main/java/org/springframework/cassandra/core/RingMemberHostMapper.java
View File

@@ -27,7 +27,6 @@ import com.datastax.driver.core.exceptions.DriverException;
/**
* @author David Webb
* @param <T>
*
*/
public class RingMemberHostMapper implements HostMapper<RingMember> {

2
spring-cql/src/main/java/org/springframework/cassandra/core/RowIterator.java
View File

@@ -15,10 +15,8 @@
*/
package org.springframework.cassandra.core;
/**
* @author David Webb
*
*/
public interface RowIterator {

1
spring-cql/src/main/java/org/springframework/cassandra/core/SessionCallback.java
View File

@@ -23,7 +23,6 @@ import com.datastax.driver.core.Session;
* Interface for operations on a Cassandra Session.
*
* @author David Webb
*
* @param <T>
*/
public interface SessionCallback<T> {

8
spring-cql/src/main/java/org/springframework/cassandra/core/SimplePreparedStatementCreator.java
View File

@@ -23,13 +23,11 @@ import com.datastax.driver.core.exceptions.DriverException;
/**
* This Prepared Statement Creator simply prepares a statement from the CQL string. This should not be used in
* Production systems with high volume reads and writes. Use {@link CachedPreparedStatementCreator}
*
* When preparing statements with Cassandra, each Statement should be prepared once and only once due to the overhead of
* preparing the statement.
* Production systems with high volume reads and writes. Use {@link CachedPreparedStatementCreator} When preparing
* statements with Cassandra, each Statement should be prepared once and only once due to the overhead of preparing the
* statement.
*
* @author David Webb
*
*/
public class SimplePreparedStatementCreator implements PreparedStatementCreator {

12
spring-cql/src/main/java/org/springframework/cassandra/core/WriteOptions.java
View File

@@ -17,12 +17,22 @@ package org.springframework.cassandra.core;
/**
* @author David Webb
*
*/
public class WriteOptions extends QueryOptions {
private Integer ttl;
public WriteOptions() {}
public WriteOptions(ConsistencyLevel consistencyLevel, RetryPolicy retryPolicy) {
this(consistencyLevel, retryPolicy, null);
}
public WriteOptions(ConsistencyLevel consistencyLevel, RetryPolicy retryPolicy, Integer ttl) {
super(consistencyLevel, retryPolicy);
setTtl(ttl);
}
/**
* @return Returns the ttl.
*/

1
spring-cql/src/main/java/org/springframework/cassandra/core/converter/AbstractResultSetConverter.java
View File

@@ -24,7 +24,6 @@ import com.datastax.driver.core.ResultSet;
* {@link IllegalArgumentException}.
*
* @author Matthew T. Adams
*
* @param <T>
*/
public abstract class AbstractResultSetConverter<T> implements Converter<ResultSet, T> {

1
spring-cql/src/main/java/org/springframework/cassandra/core/converter/AbstractResultSetToBasicFixedTypeConverter.java
View File

@@ -7,7 +7,6 @@ import org.springframework.core.convert.support.DefaultConversionService;
* Thin wrapper that allows subclasses to delegate conversion of the given value to a {@link DefaultConversionService}.
*
* @author Matthew T. Adams
*
* @param <T>
*/
public abstract class AbstractResultSetToBasicFixedTypeConverter<T> extends AbstractResultSetConverter<T> {

3
spring-cql/src/main/java/org/springframework/cassandra/core/converter/ResultSetToListConverter.java
View File

@@ -15,8 +15,7 @@ public class ResultSetToListConverter implements Converter<ResultSet, List<Map<S
protected Converter<Row, Map<String, Object>> rowConverter = new RowToMapConverter();
public ResultSetToListConverter() {
}
public ResultSetToListConverter() {}
public ResultSetToListConverter(Converter<Row, Map<String, Object>> rowConverter) {
setRowConverter(rowConverter);

10
spring-cql/src/main/java/org/springframework/cassandra/core/cql/CqlConstantType.java
View File

@@ -5,8 +5,12 @@ import java.util.regex.Pattern;
public enum CqlConstantType {
STRING(STRING_PATTERN), INTEGER(INTEGER_PATTERN), FLOAT(FLOAT_PATTERN), BOOLEAN(BOOLEAN_PATTERN), UUID(UUID_PATTERN), BLOB(
BLOB_PATTERN);
STRING(STRING_PATTERN),
INTEGER(INTEGER_PATTERN),
FLOAT(FLOAT_PATTERN),
BOOLEAN(BOOLEAN_PATTERN),
UUID(UUID_PATTERN),
BLOB(BLOB_PATTERN);
private Pattern pattern;
@@ -38,4 +42,4 @@ public enum CqlConstantType {
public static final String BLOB_REGEX = "(?i)0[x](0-9a-f)+";
public static final Pattern BLOB_PATTERN = Pattern.compile(BLOB_REGEX);
}
}
}

1
spring-cql/src/main/java/org/springframework/cassandra/core/cql/CqlIdentifier.java
View File

@@ -21,7 +21,6 @@ import com.datastax.driver.core.TableMetadata;
* @see #toCql()
* @see #toCql(StringBuilder)
* @see #toString()
*
* @author John McPeek
* @author Matthew T. Adams
*/

1
spring-cql/src/main/java/org/springframework/cassandra/core/cql/KeyspaceIdentifier.java
View File

@@ -15,7 +15,6 @@ import org.springframework.util.Assert;
* @see #toCql()
* @see #toCql(StringBuilder)
* @see #toString()
*
* @author Matthew T. Adams
*/
public final class KeyspaceIdentifier implements Comparable<KeyspaceIdentifier> {

4
spring-cql/src/main/java/org/springframework/cassandra/core/cql/generator/CreateIndexCqlGenerator.java
View File

@@ -43,8 +43,8 @@ public class CreateIndexCqlGenerator extends IndexNameCqlGenerator<CreateIndexSp
cql.append("CREATE").append(spec().isCustom() ? " CUSTOM" : "").append(" INDEX ")
.append(spec().getIfNotExists() ? "IF NOT EXISTS " : "")
.append(spec().getName() == null ? "" : spec().getName()).append(" ON ")
.append(spec().getTableName()).append(" (").append(spec().getColumnName()).append(")");
.append(spec().getName() == null ? "" : spec().getName()).append(" ON ").append(spec().getTableName())
.append(" (").append(spec().getColumnName()).append(")");
if (spec().isCustom()) {
cql.append(" USING ").append(spec().getUsing());

3
spring-cql/src/main/java/org/springframework/cassandra/core/keyspace/AlterKeyspaceSpecification.java
View File

@@ -35,8 +35,7 @@ public class AlterKeyspaceSpecification extends KeyspaceOptionsSpecification<Alt
return new AlterKeyspaceSpecification(name);
}
public AlterKeyspaceSpecification() {
}
public AlterKeyspaceSpecification() {}
public AlterKeyspaceSpecification(String name) {
name(name);

3
spring-cql/src/main/java/org/springframework/cassandra/core/keyspace/CreateKeyspaceSpecification.java
View File

@@ -49,8 +49,7 @@ public class CreateKeyspaceSpecification extends KeyspaceSpecification<CreateKey
private boolean ifNotExists = false;
public CreateKeyspaceSpecification() {
}
public CreateKeyspaceSpecification() {}
public CreateKeyspaceSpecification(String name) {
name(name);

8
spring-cql/src/main/java/org/springframework/cassandra/core/keyspace/DefaultOption.java
View File

@@ -96,13 +96,7 @@ public class DefaultOption implements Option {
}
ctor.newInstance(value.toString());
return true;
} catch (InstantiationException e) {
} catch (IllegalAccessException e) {
} catch (IllegalArgumentException e) {
} catch (InvocationTargetException e) {
} catch (NoSuchMethodException e) {
} catch (SecurityException e) {
}
} catch (InstantiationException e) {} catch (IllegalAccessException e) {} catch (IllegalArgumentException e) {} catch (InvocationTargetException e) {} catch (NoSuchMethodException e) {} catch (SecurityException e) {}
return false;
}

3
spring-cql/src/main/java/org/springframework/cassandra/core/keyspace/DropKeyspaceSpecification.java
View File

@@ -45,8 +45,7 @@ public class DropKeyspaceSpecification extends KeyspaceActionSpecification<DropK
private boolean ifExists;
public DropKeyspaceSpecification() {
}
public DropKeyspaceSpecification() {}
public DropKeyspaceSpecification(String name) {
name(name);

1
spring-cql/src/main/java/org/springframework/cassandra/core/keyspace/IndexNameSpecification.java
View File

@@ -24,7 +24,6 @@ import org.springframework.util.Assert;
* Abstract builder class to support the construction of an index.
*
* @param <T> The subtype of the {@link IndexNameSpecification}
*
* @author David Webb
* @author Matthew T. Adams
*/

1
spring-cql/src/main/java/org/springframework/cassandra/core/keyspace/KeyspaceOption.java
View File

@@ -72,7 +72,6 @@ public enum KeyspaceOption implements Option {
* Known Replication Strategy options.
*
* @author John McPeek
*
*/
public enum ReplicationStrategy {
SIMPLE_STRATEGY("SimpleStrategy"), NETWORK_TOPOLOGY_STRATEGY("NetworkTopologyStrategy");

3
spring-cql/src/main/java/org/springframework/cassandra/core/keyspace/KeyspaceSpecification.java
View File

@@ -23,5 +23,4 @@ package org.springframework.cassandra.core.keyspace;
* @author Matthew T. Adams
*/
public class KeyspaceSpecification<T> extends KeyspaceOptionsSpecification<KeyspaceSpecification<T>> implements
KeyspaceDescriptor {
}
KeyspaceDescriptor {}

4
spring-cql/src/main/java/org/springframework/cassandra/core/keyspace/TableOption.java
View File

@@ -20,9 +20,7 @@ import java.util.Map;
/**
* Enumeration that represents all known table options. If a table option is not listed here, but is supported by
* Cassandra, use the method {@link CreateTableSpecification#with(String, Object, boolean, boolean)} to write the raw
* value.
*
* Implements {@link Option} via delegation, since {@link Enum}s can't extend anything.
* value. Implements {@link Option} via delegation, since {@link Enum}s can't extend anything.
*
* @author Matthew T. Adams
* @see CompactionOption

34
spring-cql/src/main/java/org/springframework/cassandra/core/util/CollectionUtils.java
View File

@@ -39,4 +39,38 @@ public class CollectionUtils {
return list;
}
public static List<?> toList(Object thing) {
List<Object> list = new ArrayList<Object>();
list.add(thing);
return list;
}
public static List<?> toList(Object thing1, Object thing2) {
List<Object> list = new ArrayList<Object>();
list.add(thing1);
list.add(thing2);
return list;
}
public static List<?> toList(Object thing1, Object thing2, Object thing3) {
List<Object> list = new ArrayList<Object>();
list.add(thing1);
list.add(thing2);
list.add(thing3);
return list;
}
public static List<?> toList(Object thing1, Object thing2, Object thing3, Object... rest) {
List<Object> list = new ArrayList<Object>();
list.add(thing1);
list.add(thing2);
list.add(thing3);
if (rest != null) {
for (Object thing : rest) {
list.add(thing);
}
}
return list;
}
}

20
spring-cql/src/test/java/org/springframework/cassandra/test/integration/AbstractCqlTemplateIntegrationTest.java Normal file
View File

@@ -0,0 +1,20 @@
package org.springframework.cassandra.test.integration;
import org.springframework.cassandra.core.CqlOperations;
import org.springframework.cassandra.core.CqlTemplate;
public class AbstractCqlTemplateIntegrationTest extends AbstractKeyspaceCreatingIntegrationTest {
protected CqlOperations t;
{
t = new CqlTemplate(SESSION);
}
public AbstractCqlTemplateIntegrationTest() {}
public AbstractCqlTemplateIntegrationTest(String keyspace) {
super(keyspace);
}
}

1
spring-cql/src/test/java/org/springframework/cassandra/test/integration/config/xml/TestHostStateListener.java
View File

@@ -23,7 +23,6 @@ import com.datastax.driver.core.Host.StateListener;
/**
* @author David Webb
*
*/
public class TestHostStateListener implements StateListener {

1
spring-cql/src/test/java/org/springframework/cassandra/test/integration/config/xml/TestLatencyTracker.java
View File

@@ -23,7 +23,6 @@ import com.datastax.driver.core.LatencyTracker;
/**
* @author David Webb
*
*/
public class TestLatencyTracker implements LatencyTracker {

3
spring-cql/src/test/java/org/springframework/cassandra/test/integration/config/xml/XmlConfigTest.java
View File

@@ -27,7 +27,8 @@ import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
import com.datastax.driver.core.Session;
@RunWith(SpringJUnit4ClassRunner.class)
@ContextConfiguration(locations = "classpath:/org/springframework/cassandra/test/integration/config/xml/XmlConfigTest-context.xml")
@ContextConfiguration(
locations = "classpath:/org/springframework/cassandra/test/integration/config/xml/XmlConfigTest-context.xml")
public class XmlConfigTest extends AbstractKeyspaceCreatingIntegrationTest {
public static final String KEYSPACE = "xmlconfigtest";

68
spring-cql/src/test/java/org/springframework/cassandra/test/integration/core/cql/generator/CqlTableSpecificationAssertions.java
View File

@@ -93,23 +93,23 @@ public class CqlTableSpecificationAssertions {
switch (tableOption) {
case BLOOM_FILTER_FP_CHANCE:
case READ_REPAIR_CHANCE:
case DCLOCAL_READ_REPAIR_CHANCE:
assertEquals((Double) expected, (Double) actual, DELTA);
return;
case BLOOM_FILTER_FP_CHANCE:
case READ_REPAIR_CHANCE:
case DCLOCAL_READ_REPAIR_CHANCE:
assertEquals((Double) expected, (Double) actual, DELTA);
return;
case CACHING:
assertEquals(((String) expected).toUpperCase(), ((String) actual).toUpperCase());
return;
case CACHING:
assertEquals(((String) expected).toUpperCase(), ((String) actual).toUpperCase());
return;
case COMPACTION:
assertCompaction((Map<String, Object>) expected, (Map<String, String>) actual);
return;
case COMPACTION:
assertCompaction((Map<String, Object>) expected, (Map<String, String>) actual);
return;
case COMPRESSION:
assertCompression((Map<String, Object>) expected, (Map<String, String>) actual);
return;
case COMPRESSION:
assertCompression((Map<String, Object>) expected, (Map<String, String>) actual);
return;
}
log.info(actual.getClass().getName());
@@ -137,26 +137,26 @@ public class CqlTableSpecificationAssertions {
@SuppressWarnings("unchecked")
public static <T> T getOptionFor(TableOption option, Class<?> type, Options options) {
switch (option) {
case BLOOM_FILTER_FP_CHANCE:
return (T) (Double) options.getBloomFilterFalsePositiveChance();
case CACHING:
return (T) CqlStringUtils.singleQuote(options.getCaching());
case COMMENT:
return (T) CqlStringUtils.singleQuote(options.getComment());
case COMPACTION:
return (T) options.getCompaction();
case COMPACT_STORAGE:
throw new Error(); // TODO: figure out
case COMPRESSION:
return (T) options.getCompression();
case DCLOCAL_READ_REPAIR_CHANCE:
return (T) (Double) options.getLocalReadRepairChance();
case GC_GRACE_SECONDS:
return (T) new Long(options.getGcGraceInSeconds());
case READ_REPAIR_CHANCE:
return (T) (Double) options.getReadRepairChance();
case REPLICATE_ON_WRITE:
return (T) (Boolean) options.getReplicateOnWrite();
case BLOOM_FILTER_FP_CHANCE:
return (T) (Double) options.getBloomFilterFalsePositiveChance();
case CACHING:
return (T) CqlStringUtils.singleQuote(options.getCaching());
case COMMENT:
return (T) CqlStringUtils.singleQuote(options.getComment());
case COMPACTION:
return (T) options.getCompaction();
case COMPACT_STORAGE:
throw new Error(); // TODO: figure out
case COMPRESSION:
return (T) options.getCompression();
case DCLOCAL_READ_REPAIR_CHANCE:
return (T) (Double) options.getLocalReadRepairChance();
case GC_GRACE_SECONDS:
return (T) new Long(options.getGcGraceInSeconds());
case READ_REPAIR_CHANCE:
return (T) (Double) options.getReadRepairChance();
case REPLICATE_ON_WRITE:
return (T) (Boolean) options.getReplicateOnWrite();
}
return null;
}

1
spring-cql/src/test/java/org/springframework/cassandra/test/integration/core/cql/generator/CreateIndexCqlGeneratorIntegrationTests.java
View File

@@ -36,7 +36,6 @@ public class CreateIndexCqlGeneratorIntegrationTests {
* Integration test base class that knows how to do everything except instantiate the concrete unit test type T.
*
* @author Matthew T. Adams
*
* @param <T> The concrete unit test class to which this integration test corresponds.
*/
public static abstract class Base<T extends CreateIndexTest> extends AbstractKeyspaceCreatingIntegrationTest {

1
spring-cql/src/test/java/org/springframework/cassandra/test/integration/core/cql/generator/CreateTableCqlGeneratorIntegrationTests.java
View File

@@ -34,7 +34,6 @@ public class CreateTableCqlGeneratorIntegrationTests {
* Integration test base class that knows how to do everything except instantiate the concrete unit test type T.
*
* @author Matthew T. Adams
*
* @param <T> The concrete unit test class to which this integration test corresponds.
*/
public static abstract class Base<T extends CreateTableTest> extends AbstractKeyspaceCreatingIntegrationTest {

2
spring-cql/src/test/java/org/springframework/cassandra/test/integration/core/cql/generator/FunkyIdentifierIntegrationTest.java
View File

@@ -39,4 +39,4 @@ public class FunkyIdentifierIntegrationTest extends AbstractKeyspaceCreatingInte
.partitionKeyColumn(name, DataType.text())).toCql());
}
}
}
}

1
spring-cql/src/test/java/org/springframework/cassandra/test/integration/core/template/Book.java
View File

@@ -19,7 +19,6 @@ package org.springframework.cassandra.test.integration.core.template;
* Test POJO
*
* @author David Webb
*
*/
public class Book {

14
spring-cql/src/test/java/org/springframework/cassandra/test/integration/core/template/BookListener.java
View File

@@ -15,9 +15,8 @@
*/
package org.springframework.cassandra.test.integration.core.template;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.cassandra.core.AsynchronousQueryListener;
import org.springframework.cassandra.test.unit.support.TestListener;
import com.datastax.driver.core.ResultSetFuture;
import com.datastax.driver.core.Row;
@@ -26,18 +25,17 @@ import com.datastax.driver.core.Row;
* Test Implementation of the {@link AsynchronousQueryListener}
*
* @author David Webb
*
* @author Matthew T. Adams
*/
public class BookListener implements AsynchronousQueryListener {
private static Logger log = LoggerFactory.getLogger(BookListener.class);
public class BookListener extends TestListener implements AsynchronousQueryListener {
private Book book;
private boolean done;
@Override
public void onQueryComplete(ResultSetFuture rsf) {
log.info("QueryCompleted");
countDown();
Row row;
try {
row = rsf.get().one();
@@ -51,8 +49,6 @@ public class BookListener implements AsynchronousQueryListener {
book.setPages(row.getInt("pages"));
done = true;
log.info("DONE");
}
/**

76
spring-cql/src/test/java/org/springframework/cassandra/test/integration/core/template/CQLOperationsTest.java
View File

@@ -73,7 +73,6 @@ import static org.junit.Assert.assertTrue;
* Unit Tests for CqlTemplate
*
* @author David Webb
*
*/
public class CQLOperationsTest extends AbstractKeyspaceCreatingIntegrationTest {
@@ -106,12 +105,8 @@ public class CQLOperationsTest extends AbstractKeyspaceCreatingIntegrationTest {
@Before
public void setupTemplate() {
log.info("Running setupTemplate()");
if (cqlTemplate == null) {
log.info("null Template ... Initialzing DB test CQL");
// CassandraCQLUnit cassandraCQLUnit = new CassandraCQLUnit(new ClassPathCQLDataSet(
// "cassandraOperationsTest-cql-dataload.cql", keyspace), CASSANDRA_CONFIG, CASSANDRA_HOST,
// CASSANDRA_NATIVE_PORT);
@@ -130,10 +125,6 @@ public class CQLOperationsTest extends AbstractKeyspaceCreatingIntegrationTest {
* running.
*/
assertNotNull(ring);
for (RingMember h : ring) {
log.info("ringTest Host -> " + h.address);
}
}
@Test
@@ -424,7 +415,7 @@ public class CQLOperationsTest extends AbstractKeyspaceCreatingIntegrationTest {
}
@Test
public void queryAsynchronouslyWithListener() {
public void queryAsynchronouslyWithListener() throws InterruptedException {
QueryOptions options = new QueryOptions();
options.setConsistencyLevel(ConsistencyLevel.ONE);
@@ -433,25 +424,15 @@ public class CQLOperationsTest extends AbstractKeyspaceCreatingIntegrationTest {
final String isbn = "999999999";
BookListener listener = new BookListener();
cqlTemplate.queryAsynchronously("select * from book where isbn='" + isbn + "'", listener);
// TODO Use better multi threading devices here.
while (!listener.isDone()) {
try {
Thread.sleep(100);
} catch (InterruptedException muted) {
}
}
listener.await();
Book book2 = getBook(isbn);
assertBook(listener.getBook(), book2);
}
@Test
public void queryAsynchronouslyWithListenerAndExecutor() {
public void queryAsynchronouslyWithListenerAndExecutor() throws InterruptedException {
QueryOptions options = new QueryOptions();
options.setConsistencyLevel(ConsistencyLevel.ONE);
@@ -468,23 +449,14 @@ public class CQLOperationsTest extends AbstractKeyspaceCreatingIntegrationTest {
command.run();
}
});
// TODO Use better multi threading devices here.
while (!listener.isDone()) {
try {
Thread.sleep(100);
} catch (InterruptedException muted) {
}
}
listener.await();
Book book2 = getBook(isbn);
assertBook(listener.getBook(), book2);
}
@Test
public void queryAsynchronouslyWithListenerAndExecutorAndOptions() {
public void queryAsynchronouslyWithListenerAndExecutorAndOptions() throws InterruptedException {
QueryOptions options = new QueryOptions();
options.setConsistencyLevel(ConsistencyLevel.ONE);
@@ -501,19 +473,10 @@ public class CQLOperationsTest extends AbstractKeyspaceCreatingIntegrationTest {
command.run();
}
});
// TODO Use better multi threading devices here.
while (!listener.isDone()) {
try {
Thread.sleep(100);
} catch (InterruptedException muted) {
}
}
listener.await();
Book book2 = getBook(isbn);
assertBook(listener.getBook(), book2);
}
@Test
@@ -589,7 +552,6 @@ public class CQLOperationsTest extends AbstractKeyspaceCreatingIntegrationTest {
}
});
log.debug("Size of Book List -> " + books.size());
assertEquals(books.size(), 3);
assertBook(books.get(0), getBook(books.get(0).getIsbn()));
assertBook(books.get(1), getBook(books.get(1).getIsbn()));
@@ -617,7 +579,6 @@ public class CQLOperationsTest extends AbstractKeyspaceCreatingIntegrationTest {
}
});
log.debug("Size of Book List -> " + books.size());
assertEquals(books.size(), 3);
assertBook(books.get(0), getBook(books.get(0).getIsbn()));
assertBook(books.get(1), getBook(books.get(1).getIsbn()));
@@ -725,8 +686,6 @@ public class CQLOperationsTest extends AbstractKeyspaceCreatingIntegrationTest {
Map<String, Object> rsMap = cqlTemplate.queryForMap("select * from book where isbn in ('" + ISBN_NINES + "')");
log.debug(rsMap.toString());
Book b1 = objectToBook(rsMap.get("isbn"), rsMap.get("title"), rsMap.get("author"), rsMap.get("pages"));
Book b2 = getBook(ISBN_NINES);
@@ -746,8 +705,6 @@ public class CQLOperationsTest extends AbstractKeyspaceCreatingIntegrationTest {
Map<String, Object> rsMap = cqlTemplate.processMap(rs);
log.debug("Size of Book List -> " + rsMap.size());
Book b1 = objectToBook(rsMap.get("isbn"), rsMap.get("title"), rsMap.get("author"), rsMap.get("pages"));
Book b2 = getBook(ISBN_NINES);
@@ -765,8 +722,6 @@ public class CQLOperationsTest extends AbstractKeyspaceCreatingIntegrationTest {
List<String> titles = cqlTemplate.queryForList("select title from book where isbn in ('1234','2345','3456')",
String.class);
log.debug(titles.toString());
assertNotNull(titles);
assertEquals(titles.size(), 3);
@@ -786,8 +741,6 @@ public class CQLOperationsTest extends AbstractKeyspaceCreatingIntegrationTest {
List<String> titles = cqlTemplate.processList(rs, String.class);
log.debug(titles.toString());
assertNotNull(titles);
assertEquals(titles.size(), 3);
}
@@ -801,8 +754,6 @@ public class CQLOperationsTest extends AbstractKeyspaceCreatingIntegrationTest {
List<Map<String, Object>> results = cqlTemplate
.queryForListOfMap("select * from book where isbn in ('1234','2345','3456')");
log.debug(results.toString());
assertEquals(results.size(), 3);
}
@@ -821,8 +772,6 @@ public class CQLOperationsTest extends AbstractKeyspaceCreatingIntegrationTest {
List<Map<String, Object>> results = cqlTemplate.processListOfMap(rs);
log.debug(results.toString());
assertEquals(results.size(), 3);
}
@@ -981,8 +930,6 @@ public class CQLOperationsTest extends AbstractKeyspaceCreatingIntegrationTest {
}
});
log.debug("Size of all Books -> " + books.size());
assertTrue(books.size() > 0);
}
@@ -1004,10 +951,7 @@ public class CQLOperationsTest extends AbstractKeyspaceCreatingIntegrationTest {
@Override
public void processRow(Row row) throws DriverException {
Book b = rowToBook(row);
log.debug("Title -> " + b.getTitle());
rowToBook(row);
}
});
@@ -1034,8 +978,6 @@ public class CQLOperationsTest extends AbstractKeyspaceCreatingIntegrationTest {
}
});
log.debug("Size of all Books -> " + books.size());
assertTrue(books.size() > 0);
}
@@ -1073,8 +1015,6 @@ public class CQLOperationsTest extends AbstractKeyspaceCreatingIntegrationTest {
Book b2 = getBook(isbn);
log.debug("Book list Size -> " + books.size());
assertEquals(books.size(), 1);
assertBook(books.get(0), b2);
}
@@ -1144,8 +1084,6 @@ public class CQLOperationsTest extends AbstractKeyspaceCreatingIntegrationTest {
@Test
public void insertAndTruncateQueryObjectTest() {
log.info("Starting Insert and Truncate Query Object Test");
String tableName = "truncate_test";
CreateTableSpecification createTableSpec = new CreateTableSpecification();

5
spring-cql/src/test/java/org/springframework/cassandra/test/integration/core/template/async/AbstractAsynchronousTest.java Normal file
View File

@@ -0,0 +1,5 @@
package org.springframework.cassandra.test.integration.core.template.async;
import org.springframework.cassandra.test.integration.AbstractCqlTemplateIntegrationTest;
public abstract class AbstractAsynchronousTest extends AbstractCqlTemplateIntegrationTest {}

388
spring-cql/src/test/java/org/springframework/cassandra/test/integration/core/template/async/AsynchronousTest.java Normal file
View File

@@ -0,0 +1,388 @@
package org.springframework.cassandra.test.integration.core.template.async;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertTrue;
import static org.springframework.cassandra.core.keyspace.CreateTableSpecification.createTable;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Comparator;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.CancellationException;
import org.junit.Before;
import org.junit.Test;
import org.springframework.cassandra.core.AsynchronousQueryListener;
import org.springframework.cassandra.core.ConsistencyLevel;
import org.springframework.cassandra.core.Cancellable;
import org.springframework.cassandra.core.QueryForListOfMapListener;
import org.springframework.cassandra.core.QueryForMapListener;
import org.springframework.cassandra.core.QueryForObjectListener;
import org.springframework.cassandra.core.QueryOptions;
import org.springframework.cassandra.core.RetryPolicy;
import org.springframework.cassandra.support.exception.CassandraInsufficientReplicasAvailableException;
import org.springframework.util.StringUtils;
import com.datastax.driver.core.DataType;
import com.datastax.driver.core.Row;
import com.datastax.driver.core.querybuilder.QueryBuilder;
import com.datastax.driver.core.querybuilder.Select;
public class AsynchronousTest extends AbstractAsynchronousTest {
public static final String TABLE = "book";
public static String cql(Book book, String... columns) {
if (columns == null || columns.length == 0) {
columns = new String[] { "title", "isbn" };
}
return String.format("select %s from %s where title = '%s' and isbn = '%s'",
StringUtils.arrayToCommaDelimitedString(columns), TABLE, book.title, book.isbn);
}
public static String cql(String[] titles) {
String[] quoted = new String[titles.length];
System.arraycopy(titles, 0, quoted, 0, titles.length);
for (int i = 0; i < quoted.length; i++) {
quoted[i] = "'" + quoted[i] + "'";
}
return String
.format("select * from %s where title in (%s)", TABLE, StringUtils.arrayToCommaDelimitedString(quoted));
}
public static Select select(String isbn) {
Select select = QueryBuilder.select("isbn", "title").from(TABLE);
select.where(QueryBuilder.eq("isbn", isbn));
return select;
}
public static final Comparator<Book> BOOK_COMPARATOR = new Comparator<Book>() {
@Override
public int compare(Book l, Book r) {
return l.isbn.compareTo(r.isbn);
}
};
public static void assertMapEquals(Map<?, ?> expected, Map<?, ?> actual) {
for (Object key : expected.keySet()) {
assertTrue(actual.containsKey(key));
assertEquals(expected.get(key), actual.get(key));
}
}
void ensureTableExists() {
t.execute(createTable(TABLE).ifNotExists().partitionKeyColumn("title", DataType.ascii())
.clusteredKeyColumn("isbn", DataType.ascii()));
}
Book[] insert(int n) {
Book[] books = new Book[n];
for (int i = 0; i < n; i++) {
Book b = books[i] = Book.random();
t.execute(String.format("insert into %s (isbn, title) values ('%s', '%s')", TABLE, b.isbn, b.title));
}
return books;
}
@Before
public void beforeEach() {
ensureTableExists();
t.truncate(TABLE);
}
void assertBook(Book expected, Book actual) {
assertEquals(expected.isbn, actual.isbn);
assertEquals(expected.title, actual.title);
}
/**
* Tests that test {@link AsynchronousQueryListener} should create an anonymous subclass of this class then call
* either {@link #test()} or {@link #test(int)}
*/
abstract class AsynchronousQueryListenerTestTemplate {
/**
* Subclass must perform the asynchronous query using the given data and listener and set <code>this.expected</code>
* to the appropriate value before returning.
*/
abstract void doAsyncQuery(Book b, BasicListener listener);
void test() throws InterruptedException {
Book expected = insert(1)[0];
BasicListener listener = new BasicListener();
doAsyncQuery(expected, listener);
listener.await();
Row r = t.getResultSetUninterruptibly(listener.rsf).one();
Book actual = new Book(r.getString(0), r.getString(1));
assertBook(expected, actual);
}
}
/**
* Tests that test {@link QueryForObjectListener} should create an anonymous subclass of this class then call either
* {@link #test()} or {@link #test(int)}
*/
abstract class QueryForObjectListenerTestTemplate<T> {
/**
* Subclass must perform the asynchronous query using the given data and listener and set <code>this.expected</code>
* to the appropriate value before returning.
*/
abstract void doAsyncQuery(Book b, QueryForObjectListener<T> listener);
T expected; // subclass should set this value in doAsyncQuery
void test() throws Exception {
Book book = insert(1)[0];
ObjectListener<T> listener = new ObjectListener<T>();
doAsyncQuery(book, listener);
listener.await();
if (listener.exception != null) {
throw listener.exception;
}
assertEquals(expected, listener.result);
}
}
/**
* Tests that test {@link QueryForMapListener} should create an anonymous subclass of this class then call either
* {@link #test()} or {@link #test(int)}
*/
abstract class QueryForMapListenerTestTemplate {
/**
* Subclass must perform the asynchronous query using the given data and listener and set <code>this.expected</code>
* to the appropriate value before returning.
*/
abstract void doAsyncQuery(Book b, QueryForMapListener listener);
Map<String, Object> expected; // subclass should set this value in doAsyncQuery
void test() throws Exception {
Book book = insert(1)[0];
MapListener listener = new MapListener();
doAsyncQuery(book, listener);
listener.await();
if (listener.exception != null) {
throw listener.exception;
}
assertMapEquals(expected, listener.result);
}
}
/**
* Tests that test {@link QueryForMapListener} should create an anonymous subclass of this class then call either
* {@link #test()} or {@link #test(int)}
*/
abstract class QueryForListListenerTestTemplate {
/**
* Subclass must perform the asynchronous query using the given data and listener and set <code>this.expected</code>
* to the appropriate value before returning.
*/
abstract void doAsyncQuery(Book[] books, QueryForListOfMapListener listener);
List<Map<String, Object>> expected; // subclass should set this value in doAsyncQuery
void test(int n) throws Exception {
Book[] books = insert(n);
ListOfMapListener listener = new ListOfMapListener();
Arrays.sort(books, BOOK_COMPARATOR);
doAsyncQuery(books, listener);
listener.await();
if (listener.exception != null) {
throw listener.exception;
}
for (int i = 0; i < expected.size(); i++) {
assertMapEquals(expected.get(i), listener.result.get(i));
}
}
}
@Test(expected = CancellationException.class)
public void testString_AsynchronousQueryListener_Cancelled() throws InterruptedException {
new AsynchronousQueryListenerTestTemplate() {
@Override
void doAsyncQuery(Book b, BasicListener listener) {
Cancellable qc = t.queryAsynchronously(cql(b), listener);
qc.cancel();
}
}.test();
}
@Test
public void testString_AsynchronousQueryListener() throws InterruptedException {
new AsynchronousQueryListenerTestTemplate() {
@Override
void doAsyncQuery(Book b, BasicListener listener) {
t.queryAsynchronously(cql(b), listener);
}
}.test();
}
public void testString_AsynchronousQueryListener_QueryOptions(final ConsistencyLevel cl) throws InterruptedException {
new AsynchronousQueryListenerTestTemplate() {
@Override
void doAsyncQuery(Book b, BasicListener listener) {
t.queryAsynchronously(cql(b), listener, new QueryOptions(cl, RetryPolicy.LOGGING));
}
}.test();
}
@Test
public void testString_AsynchronousQueryListener_QueryOptionsWithConsistencyLevel1() throws InterruptedException {
testString_AsynchronousQueryListener_QueryOptions(ConsistencyLevel.ONE);
}
@Test(expected = CassandraInsufficientReplicasAvailableException.class)
public void testString_AsynchronousQueryListener_QueryOptionsWithConsistencyLevel2() throws InterruptedException {
testString_AsynchronousQueryListener_QueryOptions(ConsistencyLevel.TWO);
}
@Test
public void testSelect_AsynchronousQueryListener() throws InterruptedException {
new AsynchronousQueryListenerTestTemplate() {
@Override
void doAsyncQuery(Book b, BasicListener listener) {
t.queryAsynchronously(cql(b), listener);
}
}.test();
}
@Test
public void testString_QueryForObjectListener() throws Exception {
new QueryForObjectListenerTestTemplate<String>() {
@Override
void doAsyncQuery(Book b, QueryForObjectListener<String> listener) {
t.queryForObjectAsynchronously(cql(b, "title"), String.class, listener);
expected = b.title;
}
}.test();
}
public void testString_QueryForObjectListener_QueryOptions(final ConsistencyLevel cl) throws Exception {
new QueryForObjectListenerTestTemplate<String>() {
@Override
void doAsyncQuery(Book b, QueryForObjectListener<String> listener) {
QueryOptions opts = new QueryOptions(cl, RetryPolicy.LOGGING);
t.queryForObjectAsynchronously(cql(b, "title"), String.class, listener, opts);
expected = b.title;
}
}.test();
}
@Test
public void testString_QueryForObjectListener_QueryOptionsWithConsistencyLevel() throws Exception {
testString_QueryForObjectListener_QueryOptions(ConsistencyLevel.ONE);
}
@Test(expected = CassandraInsufficientReplicasAvailableException.class)
public void testString_QueryForObjectListener_QueryOptionsWithConsistencyLevel2() throws Exception {
testString_QueryForObjectListener_QueryOptions(ConsistencyLevel.TWO);
}
@Test
public void testString_QueryForMapListener() throws Exception {
new QueryForMapListenerTestTemplate() {
@Override
void doAsyncQuery(Book b, QueryForMapListener listener) {
t.queryForMapAsynchronously(cql(b), listener);
expected = new HashMap<String, Object>();
expected.put("isbn", b.isbn);
expected.put("title", b.title);
}
}.test();
}
public void testString_QueryForMapListener_QueryOptions(final ConsistencyLevel cl) throws Exception {
new QueryForMapListenerTestTemplate() {
@Override
void doAsyncQuery(Book b, QueryForMapListener listener) {
QueryOptions opts = new QueryOptions(cl, RetryPolicy.LOGGING);
t.queryForMapAsynchronously(cql(b), listener, opts);
expected = new HashMap<String, Object>();
expected.put("isbn", b.isbn);
expected.put("title", b.title);
}
}.test();
}
@Test
public void testString_QueryForMapListener_QueryOptionsWithConsistencyLevel1() throws Exception {
testString_QueryForMapListener_QueryOptions(ConsistencyLevel.ONE);
}
@Test(expected = CassandraInsufficientReplicasAvailableException.class)
public void testString_QueryForMapListener_QueryOptionsWithConsistencyLevel2() throws Exception {
testString_QueryForMapListener_QueryOptions(ConsistencyLevel.TWO);
}
@Test
public void testString_QueryForListListener() throws Exception {
new QueryForListListenerTestTemplate() {
@Override
void doAsyncQuery(Book[] books, QueryForListOfMapListener listener) {
String[] titles = new String[books.length];
expected = new ArrayList<Map<String, Object>>(books.length);
for (int i = 0; i < books.length; i++) {
Book b = books[i];
titles[i] = b.title;
HashMap<String, Object> row = new HashMap<String, Object>(2);
row.put("title", b.title);
row.put("isbn", b.isbn);
expected.add(row);
}
t.queryForListOfMapAsynchronously(cql(titles), listener);
}
}.test(2);
}
public void testString_QueryForListListener_QueryOptions(final ConsistencyLevel cl) throws Exception {
new QueryForListListenerTestTemplate() {
@Override
void doAsyncQuery(Book[] books, QueryForListOfMapListener listener) {
String[] titles = new String[books.length];
expected = new ArrayList<Map<String, Object>>(books.length);
for (int i = 0; i < books.length; i++) {
Book b = books[i];
titles[i] = b.title;
HashMap<String, Object> row = new HashMap<String, Object>(2);
row.put("title", b.title);
row.put("isbn", b.isbn);
expected.add(row);
}
t.queryForListOfMapAsynchronously(cql(titles), listener, new QueryOptions(cl, RetryPolicy.LOGGING));
}
}.test(2);
}
@Test
public void testString_QueryForListListener_QueryOptionsWithConsistencyLevel1() throws Exception {
testString_QueryForListListener_QueryOptions(ConsistencyLevel.ONE);
}
@Test(expected = CassandraInsufficientReplicasAvailableException.class)
public void testString_QueryForListListener_QueryOptionsWithConsistencyLevel2() throws Exception {
testString_QueryForListListener_QueryOptions(ConsistencyLevel.TWO);
}
}

17
spring-cql/src/test/java/org/springframework/cassandra/test/integration/core/template/async/BasicListener.java Normal file
View File

@@ -0,0 +1,17 @@
package org.springframework.cassandra.test.integration.core.template.async;
import org.springframework.cassandra.core.AsynchronousQueryListener;
import org.springframework.cassandra.test.unit.support.TestListener;
import com.datastax.driver.core.ResultSetFuture;
class BasicListener extends TestListener implements AsynchronousQueryListener {
ResultSetFuture rsf;
@Override
public void onQueryComplete(ResultSetFuture rsf) {
countDown();
this.rsf = rsf;
}
}

24
spring-cql/src/test/java/org/springframework/cassandra/test/integration/core/template/async/Book.java Normal file
View File

@@ -0,0 +1,24 @@
package org.springframework.cassandra.test.integration.core.template.async;
import java.util.UUID;
public class Book {
public static final String uuid() {
return UUID.randomUUID().toString();
}
public static Book random() {
return new Book(uuid(), uuid());
}
public Book() {}
public Book(String title, String isbn) {
this.isbn = isbn;
this.title = title;
}
public String isbn;
public String title;
}

24
spring-cql/src/test/java/org/springframework/cassandra/test/integration/core/template/async/ListListener.java Normal file
View File

@@ -0,0 +1,24 @@
package org.springframework.cassandra.test.integration.core.template.async;
import java.util.List;
import org.springframework.cassandra.core.QueryForListListener;
import org.springframework.cassandra.test.unit.support.TestListener;
public class ListListener<T> extends TestListener implements QueryForListListener<T> {
Exception exception;
List<T> result;
@Override
public void onQueryComplete(List<T> results) {
countDown();
this.result = results;
}
@Override
public void onException(Exception x) {
countDown();
this.exception = x;
}
}

7
spring-cql/src/test/java/org/springframework/cassandra/test/integration/core/template/async/ListOfMapListener.java Normal file
View File

@@ -0,0 +1,7 @@
package org.springframework.cassandra.test.integration.core.template.async;
import java.util.Map;
import org.springframework.cassandra.core.QueryForListOfMapListener;
public class ListOfMapListener extends ListListener<Map<String, Object>> implements QueryForListOfMapListener {}

24
spring-cql/src/test/java/org/springframework/cassandra/test/integration/core/template/async/MapListener.java Normal file
View File

@@ -0,0 +1,24 @@
package org.springframework.cassandra.test.integration.core.template.async;
import java.util.Map;
import org.springframework.cassandra.core.QueryForMapListener;
import org.springframework.cassandra.test.unit.support.TestListener;
public class MapListener extends TestListener implements QueryForMapListener {
Map<String, Object> result;
Exception exception;
@Override
public void onQueryComplete(Map<String, Object> results) {
countDown();
this.result = results;
}
@Override
public void onException(Exception x) {
countDown();
this.exception = x;
}
}

22
spring-cql/src/test/java/org/springframework/cassandra/test/integration/core/template/async/ObjectListener.java Normal file
View File

@@ -0,0 +1,22 @@
package org.springframework.cassandra.test.integration.core.template.async;
import org.springframework.cassandra.core.QueryForObjectListener;
import org.springframework.cassandra.test.unit.support.TestListener;
class ObjectListener<T> extends TestListener implements QueryForObjectListener<T> {
T result;
Exception exception;
@Override
public void onQueryComplete(T result) {
countDown();
this.result = result;
}
@Override
public void onException(Exception x) {
countDown();
this.exception = x;
}
}

4
spring-cql/src/test/java/org/springframework/cassandra/test/unit/core/cql/CqlStringUtilsTest.java
View File

@@ -15,6 +15,4 @@
*/
package org.springframework.cassandra.test.unit.core.cql;
public class CqlStringUtilsTest {
}
public class CqlStringUtilsTest {}

3
spring-cql/src/test/java/org/springframework/cassandra/test/unit/core/cql/generator/AlterKeyspaceCqlGeneratorTests.java
View File

@@ -54,8 +54,7 @@ public class AlterKeyspaceCqlGeneratorTests {
* Convenient base class that other test classes can use so as not to repeat the generics declarations.
*/
public static abstract class AlterKeyspaceTest extends
KeyspaceOperationCqlGeneratorTest<AlterKeyspaceSpecification, AlterKeyspaceCqlGenerator> {
}
KeyspaceOperationCqlGeneratorTest<AlterKeyspaceSpecification, AlterKeyspaceCqlGenerator> {}
public static class CompleteTest extends AlterKeyspaceTest {

4
spring-cql/src/test/java/org/springframework/cassandra/test/unit/core/cql/generator/AlterTableCqlGeneratorTests.java
View File

@@ -57,8 +57,7 @@ public class AlterTableCqlGeneratorTests {
* Convenient base class that other test classes can use so as not to repeat the generics declarations.
*/
public static abstract class AlterTableTest extends
TableOperationCqlGeneratorTest<AlterTableSpecification, AlterTableCqlGenerator> {
}
TableOperationCqlGeneratorTest<AlterTableSpecification, AlterTableCqlGenerator> {}
public static class BasicTest extends AlterTableTest {
@@ -95,7 +94,6 @@ public class AlterTableCqlGeneratorTests {
* Fully test all available create table options
*
* @author David Webb
*
*/
public static class MultipleOptionsTest extends AlterTableTest {

2
spring-cql/src/test/java/org/springframework/cassandra/test/unit/core/cql/generator/CreateTableCqlGeneratorTests.java
View File

@@ -167,7 +167,6 @@ public class CreateTableCqlGeneratorTests {
* Test just the Read Repair Chance
*
* @author David Webb
*
*/
public static class ReadRepairChanceTest extends CreateTableTest {
@@ -203,7 +202,6 @@ public class CreateTableCqlGeneratorTests {
* Fully test all available create table options
*
* @author David Webb
*
*/
public static class MultipleOptionsTest extends CreateTableTest {

3
spring-cql/src/test/java/org/springframework/cassandra/test/unit/core/cql/generator/DropIndexCqlGeneratorTests.java
View File

@@ -34,8 +34,7 @@ public class DropIndexCqlGeneratorTests {
* Convenient base class that other test classes can use so as not to repeat the generics declarations.
*/
public static abstract class DropIndexTest extends
IndexOperationCqlGeneratorTest<DropIndexSpecification, DropIndexCqlGenerator> {
}
IndexOperationCqlGeneratorTest<DropIndexSpecification, DropIndexCqlGenerator> {}
public static class BasicTest extends DropIndexTest {

3
spring-cql/src/test/java/org/springframework/cassandra/test/unit/core/cql/generator/DropKeyspaceCqlGeneratorTests.java
View File

@@ -35,8 +35,7 @@ public class DropKeyspaceCqlGeneratorTests {
* Convenient base class that other test classes can use so as not to repeat the generics declarations.
*/
public static abstract class DropTableTest extends
KeyspaceOperationCqlGeneratorTest<DropKeyspaceSpecification, DropKeyspaceCqlGenerator> {
}
KeyspaceOperationCqlGeneratorTest<DropKeyspaceSpecification, DropKeyspaceCqlGenerator> {}
public static class BasicTest extends DropTableTest {

3
spring-cql/src/test/java/org/springframework/cassandra/test/unit/core/cql/generator/DropTableCqlGeneratorTests.java
View File

@@ -43,8 +43,7 @@ public class DropTableCqlGeneratorTests {
* Convenient base class that other test classes can use so as not to repeat the generics declarations.
*/
public static abstract class DropTableTest extends
TableOperationCqlGeneratorTest<DropTableSpecification, DropTableCqlGenerator> {
}
TableOperationCqlGeneratorTest<DropTableSpecification, DropTableCqlGenerator> {}
public static class BasicTest extends DropTableTest {

1
spring-cql/src/test/java/org/springframework/cassandra/test/unit/core/cql/generator/IndexOperationCqlGeneratorTest.java
View File

@@ -26,7 +26,6 @@ import org.springframework.cassandra.core.keyspace.IndexNameSpecification;
*
* @author Matthew T. Adams
* @author David Webb
*
* @param <S> The type of the {@link IndexNameSpecification}
* @param <G> The type of the {@link IndexNameCqlGenerator}
*/

1
spring-cql/src/test/java/org/springframework/cassandra/test/unit/core/cql/generator/KeyspaceOperationCqlGeneratorTest.java
View File

@@ -30,7 +30,6 @@ import org.springframework.cassandra.core.keyspace.TableNameSpecification;
* need for encapsulation, and it makes for easier reuse in other tests like integration tests (hint hint).
*
* @author Matthew T. Adams
*
* @param <S> The type of the {@link TableNameSpecification}
* @param <G> The type of the {@link TableNameCqlGenerator}
*/

1
spring-cql/src/test/java/org/springframework/cassandra/test/unit/core/cql/generator/TableOperationCqlGeneratorTest.java
View File

@@ -24,7 +24,6 @@ import org.springframework.cassandra.core.keyspace.TableNameSpecification;
* need for encapsulation, and it makes for easier reuse in other tests like integration tests (hint hint).
*
* @author Matthew T. Adams
*
* @param <S> The type of the {@link TableNameSpecification}
* @param <G> The type of the {@link TableNameCqlGenerator}
*/

34
spring-cql/src/test/java/org/springframework/cassandra/test/unit/support/TestListener.java Normal file
View File

@@ -0,0 +1,34 @@
package org.springframework.cassandra.test.unit.support;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
/**
* Convenient listener base class that includes a {@link CountDownLatch} in order to test asynchronous behavior.
*
* @author Matthew T. Adams
*/
public class TestListener {
protected CountDownLatch latch;
public TestListener() {
this(1);
}
public TestListener(int latchCount) {
latch = new CountDownLatch(latchCount);
}
public void await() throws InterruptedException {
latch.await();
}
public void await(long ms) throws InterruptedException {
latch.await(ms, TimeUnit.MILLISECONDS);
}
public void countDown() {
latch.countDown();
}
}

1
spring-data-cassandra/.java-version Normal file
View File

@@ -0,0 +1 @@
oracle64-1.8.0.11

1
spring-data-cassandra/src/main/java/org/springframework/data/cassandra/Constants.java
View File

@@ -17,7 +17,6 @@ package org.springframework.data.cassandra;
/**
* @author David Webb
*
*/
public interface Constants {

7
spring-data-cassandra/src/main/java/org/springframework/data/cassandra/config/CassandraClusterFactoryBean.java
View File

@@ -18,10 +18,9 @@ package org.springframework.data.cassandra.config;
import org.springframework.cassandra.config.CassandraCqlClusterFactoryBean;
/**
* Spring Data Cassandra extension of CassandraCqlClusterFactoryBean. This class exists only in the name of symmetry, based
* on the other CassandraData*FactoryBean classes.
* Spring Data Cassandra extension of CassandraCqlClusterFactoryBean. This class exists only in the name of symmetry,
* based on the other CassandraData*FactoryBean classes.
*
* @author Matthew T. Adams
*/
public class CassandraClusterFactoryBean extends CassandraCqlClusterFactoryBean {
}
public class CassandraClusterFactoryBean extends CassandraCqlClusterFactoryBean {}

3
spring-data-cassandra/src/main/java/org/springframework/data/cassandra/config/CassandraEntityClassScanner.java
View File

@@ -59,8 +59,7 @@ public class CassandraEntityClassScanner {
protected Set<Class<?>> entityBasePackageClasses = new HashSet<Class<?>>();
protected ClassLoader beanClassLoader;
public CassandraEntityClassScanner() {
}
public CassandraEntityClassScanner() {}
public CassandraEntityClassScanner(Class<?>... entityBasePackageClasses) {
this(null, Arrays.asList(entityBasePackageClasses));

14
spring-data-cassandra/src/main/java/org/springframework/data/cassandra/config/CassandraMappingBeanFactoryPostProcessor.java
View File

@@ -51,13 +51,12 @@ import com.datastax.driver.core.Session;
* when creating a default definition for the {@link CassandraConverter}.
* <p/>
* If a single definition of a required type is present, then it is used. For example, if there is already a
* {@link CassandraMappingContext} definition present, then it will be used in the
* {@link BasicCassandraMappingContext} bean definition.
* {@link CassandraMappingContext} definition present, then it will be used in the {@link BasicCassandraMappingContext}
* bean definition.
* <p/>
* It requires that a single {@link Session} or {@link CassandraSessionFactoryBean} definition be present. As
* described above, multiple {@link Session} definitions, multiple {@link CassandraSessionFactoryBean} definitions,
* or both a {@link Session} and {@link CassandraSessionFactoryBean} will cause an {@link IllegalStateException} to
* be thrown.
* It requires that a single {@link Session} or {@link CassandraSessionFactoryBean} definition be present. As described
* above, multiple {@link Session} definitions, multiple {@link CassandraSessionFactoryBean} definitions, or both a
* {@link Session} and {@link CassandraSessionFactoryBean} will cause an {@link IllegalStateException} to be thrown.
*
* @author Matthew T. Adams
*/
@@ -67,8 +66,7 @@ public class CassandraMappingBeanFactoryPostProcessor implements BeanDefinitionR
* Does nothing.
*/
@Override
public void postProcessBeanFactory(ConfigurableListableBeanFactory beanFactory) throws BeansException {
}
public void postProcessBeanFactory(ConfigurableListableBeanFactory beanFactory) throws BeansException {}
/**
* Ensures that {@link BeanDefinition}s for a {@link CassandraMappingContext} and a {@link CassandraConverter} exist.

20
spring-data-cassandra/src/main/java/org/springframework/data/cassandra/config/CassandraSessionFactoryBean.java
View File

@@ -56,17 +56,17 @@ public class CassandraSessionFactoryBean extends CassandraCqlSessionFactoryBean
switch (schemaAction) {
case NONE:
return;
case NONE:
return;
case RECREATE_DROP_UNUSED:
dropUnused = true;
// don't break!
case RECREATE:
dropTables = true;
// don't break!
case CREATE:
createTables(dropTables, dropUnused);
case RECREATE_DROP_UNUSED:
dropUnused = true;
// don't break!
case RECREATE:
dropTables = true;
// don't break!
case CREATE:
createTables(dropTables, dropUnused);
}
}

3
spring-data-cassandra/src/main/java/org/springframework/data/cassandra/convert/AbstractCassandraConverter.java
View File

@@ -54,6 +54,5 @@ public abstract class AbstractCassandraConverter implements CassandraConverter,
}
@Override
public void afterPropertiesSet() {
}
public void afterPropertiesSet() {}
}

336
spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/CassandraOperations.java
View File

@@ -18,6 +18,8 @@ package org.springframework.data.cassandra.core;
import java.util.List;
import org.springframework.cassandra.core.CqlOperations;
import org.springframework.cassandra.core.Cancellable;
import org.springframework.cassandra.core.QueryForObjectListener;
import org.springframework.cassandra.core.QueryOptions;
import org.springframework.cassandra.core.WriteOptions;
import org.springframework.cassandra.core.cql.CqlIdentifier;
@@ -72,6 +74,46 @@ public interface CassandraOperations extends CqlOperations {
*/
<T> T selectOne(String cql, Class<T> type);
/**
* Executes the {@link Select} query asynchronously.
*
* @param select The {@link Select} query to execute.
* @param type The type of entity to retrieve.
* @return A {@link Cancellable} that can be used to cancel the query.
*/
<T> Cancellable selectOneAsynchronously(Select select, Class<T> type, QueryForObjectListener<T> listener);
/**
* Executes the string CQL query asynchronously.
*
* @param select The string query CQL to execute.
* @param type The type of entity to retrieve.
* @return A {@link Cancellable} that can be used to cancel the query.
*/
<T> Cancellable selectOneAsynchronously(String cql, Class<T> type, QueryForObjectListener<T> listener);
/**
* Executes the {@link Select} query asynchronously.
*
* @param select The {@link Select} query to execute.
* @param type The type of entity to retrieve.
* @param options The {@link QueryOptions} to use.
* @return A {@link Cancellable} that can be used to cancel the query.
*/
<T> Cancellable selectOneAsynchronously(Select select, Class<T> type, QueryForObjectListener<T> listener,
QueryOptions options);
/**
* Executes the string CQL query asynchronously.
*
* @param select The string query CQL to execute.
* @param type The type of entity to retrieve.
* @param options The {@link QueryOptions} to use.
* @return A {@link Cancellable} that can be used to cancel the query.
*/
<T> Cancellable selectOneAsynchronously(String cql, Class<T> type, QueryForObjectListener<T> listener,
QueryOptions options);
/**
* Execute Select query and convert ResultSet to the entity
*
@@ -86,126 +128,245 @@ public interface CassandraOperations extends CqlOperations {
long count(Class<?> type);
/**
* Insert the given object to the table by id.
* Insert the given entity.
*
* @param entity
* @param entity The entity to insert
* @return The entity given
*/
<T> T insert(T entity);
/**
* @param entity
* @param tableName
* @param options
* @return
* Insert the given entity.
*
* @param entity The entity to insert
* @param options The {@link WriteOptions} to use.
* @return The entity given
*/
<T> T insert(T entity, WriteOptions options);
/**
* Insert the given list of objects to the table by annotation table name.
* Insert the given list of entities.
*
* @param entities
* @return
* @param entities The entities to insert.
* @return The entities given.
*/
<T> List<T> insert(List<T> entities);
/**
* @param entities
* @param tableName
* @param options
* @return
* Insert the given list of entities.
*
* @param entities The entities to insert.
* @param options The {@link WriteOptions} to use.
* @return The entities given.
*/
<T> List<T> insert(List<T> entities, WriteOptions options);
/**
* Insert the given object to the table by id.
* Inserts the given entity asynchronously.
*
* @param object
* @param entity The entity to insert
* @return The entity given
* @see #insertAsynchronously(Object, WriteListener)
* @deprecated This method does not allow for query cancellation or notification of completion. Favor
* {@link #insertAsynchronously(Object, WriteListener)}.
*/
@Deprecated
<T> T insertAsynchronously(T entity);
/**
* @param entity
* @param tableName
* @param options
* @return
* Inserts the given entity asynchronously.
*
* @param entity The entity to insert
* @return The entity given
* @see #insertAsynchronously(Object, WriteOptions)
* @deprecated This method does not allow for query cancellation or notification of completion. Favor
* {@link #insertAsynchronously(Object, WriteListener, WriteOptions)}.
*/
@Deprecated
<T> T insertAsynchronously(T entity, WriteOptions options);
/**
* Insert the given object to the table by id.
* Inserts the given entity asynchronously.
*
* @param object
* @param entity The entity to insert
* @param listener The listener to receive notification of completion
* @return A {@link Cancellable} enabling the cancellation of the operation
*/
<T> Cancellable insertAsynchronously(T entity, WriteListener<T> listener);
/**
* Inserts the given entity asynchronously.
*
* @param entity The entity to insert
* @param listener The listener to receive notification of completion
* @param options The {@link WriteOptions} to use
* @return A {@link Cancellable} enabling the cancellation of the operation
*/
<T> Cancellable insertAsynchronously(T entity, WriteListener<T> listener, WriteOptions options);
/**
* Inserts the given entities asynchronously in a batch.
*
* @param entity The entities to insert
* @return The entities given
* @see #insertAsynchronously(List, WriteListener)
* @deprecated This method does not allow for query cancellation or notification of completion. Favor
* {@link #insertAsynchronously(List, WriteListener)}.
*/
@Deprecated
<T> List<T> insertAsynchronously(List<T> entities);
/**
* @param entities
* @param tableName
* @param options
* @return
* Inserts the given entities asynchronously in a batch.
*
* @param entity The entities to insert
* @return The entities given
* @see #insertAsynchronously(List, WriteListener, WriteOptions)
* @deprecated This method does not allow for query cancellation or notification of completion. Favor
* {@link #insertAsynchronously(List, WriteListener, WriteOptions)}.
*/
@Deprecated
<T> List<T> insertAsynchronously(List<T> entities, WriteOptions options);
/**
* Insert the given object to the table by id.
* Inserts the given entities asynchronously in a batch.
*
* @param object
* @param entity The entities to insert
* @param listener The listener to receive notification of completion
* @return A {@link Cancellable} enabling the cancellation of the operation
*/
<T> Cancellable insertAsynchronously(List<T> entities, WriteListener<T> listener);
/**
* Inserts the given entities asynchronously in a batch.
*
* @param entity The entities to insert
* @param listener The listener to receive notification of completion
* @param options The {@link WriteOptions} to use
* @return A {@link Cancellable} enabling the cancellation of the operation
*/
<T> Cancellable insertAsynchronously(List<T> entities, WriteListener<T> listener, WriteOptions options);
/**
* Update the given entity.
*
* @param entity The entity to update
* @return The entity given
*/
<T> T update(T entity);
/**
* @param entity
* @param tableName
* @param options
* @return
* Update the given entity.
*
* @param entity The entity to update
* @param options The {@link WriteOptions} to use.
* @return The entity given
*/
<T> T update(T entity, WriteOptions options);
/**
* Insert the given object to the table by id.
* Update the given list of entities.
*
* @param object
* @param entities The entities to update.
* @return The entities given.
*/
<T> List<T> update(List<T> entities);
/**
* @param entities
* @param tableName
* @param options
* @return
* Update the given list of entities.
*
* @param entities The entities to update.
* @param options The {@link WriteOptions} to use.
* @return The entities given.
*/
<T> List<T> update(List<T> entities, WriteOptions options);
/**
* Insert the given object to the table by id.
* Updates the given entity asynchronously.
*
* @param object
* @param entity The entity to update
* @return The entity given
* @see #updateAsynchronously(Object, WriteListener)
* @deprecated This method does not allow for query cancellation or notification of completion. Favor
* {@link #updateAsynchronously(Object, WriteListener)}.
*/
@Deprecated
<T> T updateAsynchronously(T entity);
/**
* @param entity
* @param tableName
* @param options
* @return
* Updates the given entity asynchronously.
*
* @param entity The entity to update
* @return The entity given
* @see #updateAsynchronously(Object, WriteOptions)
* @deprecated This method does not allow for query cancellation or notification of completion. Favor
* {@link #updateAsynchronously(Object, WriteListener, WriteOptions)}.
*/
@Deprecated
<T> T updateAsynchronously(T entity, WriteOptions options);
/**
* Insert the given object to the table by id.
* Updates the given entity asynchronously.
*
* @param object
* @param entity The entity to update
* @param listener The listener to receive notification of completion
* @return A {@link Cancellable} enabling the cancellation of the operation
*/
<T> Cancellable updateAsynchronously(T entity, WriteListener<T> listener);
/**
* Updates the given entity asynchronously.
*
* @param entity The entity to update
* @param listener The listener to receive notification of completion
* @param options The {@link WriteOptions} to use
* @return A {@link Cancellable} enabling the cancellation of the operation
*/
<T> Cancellable updateAsynchronously(T entity, WriteListener<T> listener, WriteOptions options);
/**
* Updates the given entities asynchronously in a batch.
*
* @param entity The entities to update
* @return The entities given
* @see #updateAsynchronously(List, WriteListener)
* @deprecated This method does not allow for query cancellation or notification of completion. Favor
* {@link #updateAsynchronously(List, WriteListener)}.
*/
@Deprecated
<T> List<T> updateAsynchronously(List<T> entities);
/**
* @param entities
* @param tableName
* @param options
* @return
* Updates the given entities asynchronously in a batch.
*
* @param entity The entities to update
* @return The entities given
* @see #updateAsynchronously(List, WriteListener, WriteOptions)
* @deprecated This method does not allow for query cancellation or notification of completion. Favor
* {@link #updateAsynchronously(List, WriteListener, WriteOptions)}.
*/
@Deprecated
<T> List<T> updateAsynchronously(List<T> entities, WriteOptions options);
/**
* Updates the given entities asynchronously in a batch.
*
* @param entity The entities to update
* @param listener The listener to receive notification of completion
* @return A {@link Cancellable} enabling the cancellation of the operation
*/
<T> Cancellable updateAsynchronously(List<T> entities, WriteListener<T> listener);
/**
* Updates the given entities asynchronously in a batch.
*
* @param entity The entities to update
* @param listener The listener to receive notification of completion
* @param options The {@link WriteOptions} to use
* @return A {@link Cancellable} enabling the cancellation of the operation
*/
<T> Cancellable updateAsynchronously(List<T> entities, WriteListener<T> listener, WriteOptions options);
/**
* Remove the given object from the table by id.
*
@@ -227,11 +388,6 @@ public interface CassandraOperations extends CqlOperations {
*/
<T> void delete(List<T> entities);
/**
* @param entities
* @param tableName
* @param options
*/
<T> void delete(List<T> entities, QueryOptions options);
/**
@@ -242,30 +398,66 @@ public interface CassandraOperations extends CqlOperations {
/**
* Remove the given object from the table by id.
*
* @param object
* @param entity The object to delete
*/
<T> void deleteAsynchronously(T entity);
/**
* @param entity
* @param tableName
* @param options
*/
<T> void deleteAsynchronously(T entity, QueryOptions options);
<T> Cancellable deleteAsynchronously(T entity);
/**
* Remove the given object from the table by id.
*
* @param object
* @param entity The object to delete
* @param options The {@link QueryOptions} to use
*/
<T> void deleteAsynchronously(List<T> entities);
<T> Cancellable deleteAsynchronously(T entity, QueryOptions options);
/**
* @param entities
* @param tableName
* @param options
* Remove the given object from the table by id.
*
* @param entity The object to delete
* @param listener The {@link DeletionListener} to receive notification upon completion
*/
<T> void deleteAsynchronously(List<T> entities, QueryOptions options);
<T> Cancellable deleteAsynchronously(T entity, DeletionListener<T> listener);
/**
* Remove the given object from the table by id.
*
* @param entity The object to delete
* @param listener The {@link DeletionListener} to receive notification upon completion
* @param options The {@link QueryOptions} to use
*/
<T> Cancellable deleteAsynchronously(T entity, DeletionListener<T> listener, QueryOptions options);
/**
* Remove the given objects from the table by id.
*
* @param entities The objects to delete
*/
<T> Cancellable deleteAsynchronously(List<T> entities);
/**
* Remove the given objects from the table by id.
*
* @param entities The objects to delete
* @param listener The {@link DeletionListener} to receive notification upon completion
*/
<T> Cancellable deleteAsynchronously(List<T> entities, DeletionListener<T> listener);
/**
* Remove the given objects from the table by id.
*
* @param entities The objects to delete
* @param options The {@link QueryOptions} to use
*/
<T> Cancellable deleteAsynchronously(List<T> entities, QueryOptions options);
/**
* Remove the given objects from the table by id.
*
* @param entities The objects to delete
* @param listener The {@link DeletionListener} to receive notification upon completion
* @param options The {@link QueryOptions} to use
*/
<T> Cancellable deleteAsynchronously(List<T> entities, DeletionListener<T> listener, QueryOptions options);
/**
* Returns the underlying {@link CassandraConverter}.
@@ -278,5 +470,11 @@ public interface CassandraOperations extends CqlOperations {
<T> List<T> selectBySimpleIds(Class<T> type, Iterable<?> ids);
/**
* @deprecated Calling this method could result in {@link OutOfMemoryError}, as this is a brute force selection.
* @param type The type of entity to select.
* @return A list of all entities of type <code>T</code>.
*/
@Deprecated
<T> List<T> selectAll(Class<T> type);
}

521
spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/CassandraTemplate.java
View File

@@ -16,11 +16,16 @@
package org.springframework.data.cassandra.core;
import java.util.ArrayList;
import java.util.Collection;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import org.springframework.cassandra.core.AsynchronousQueryListener;
import org.springframework.cassandra.core.CqlOperations;
import org.springframework.cassandra.core.CqlTemplate;
import org.springframework.cassandra.core.Cancellable;
import org.springframework.cassandra.core.QueryForObjectListener;
import org.springframework.cassandra.core.QueryOptions;
import org.springframework.cassandra.core.SessionCallback;
import org.springframework.cassandra.core.WriteOptions;
@@ -39,8 +44,11 @@ import org.springframework.data.mapping.model.BeanWrapper;
import org.springframework.util.Assert;
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.querybuilder.Batch;
import com.datastax.driver.core.querybuilder.Clause;
import com.datastax.driver.core.querybuilder.Delete;
@@ -138,7 +146,7 @@ public class CassandraTemplate extends CqlTemplate implements CassandraOperation
@Override
public <T> void delete(List<T> entities, QueryOptions options) {
batchDelete(entities, options, false);
doBatchDelete(entities, options);
}
@Override
@@ -162,27 +170,47 @@ public class CassandraTemplate extends CqlTemplate implements CassandraOperation
@Override
public <T> void delete(T entity, QueryOptions options) {
delete(entity, options, false);
doDelete(entity, options);
}
@Override
public <T> void deleteAsynchronously(List<T> entities) {
deleteAsynchronously(entities, null);
public <T> Cancellable deleteAsynchronously(List<T> entities) {
return doBatchDeleteAsync(entities, null, null);
}
@Override
public <T> void deleteAsynchronously(List<T> entities, QueryOptions options) {
batchDelete(entities, options, true);
public <T> Cancellable deleteAsynchronously(List<T> entities, QueryOptions options) {
return doBatchDeleteAsync(entities, null, options);
}
@Override
public <T> void deleteAsynchronously(T entity) {
deleteAsynchronously(entity, null);
public <T> Cancellable deleteAsynchronously(List<T> entities, DeletionListener<T> listener) {
return doBatchDeleteAsync(entities, listener, null);
}
@Override
public <T> void deleteAsynchronously(T entity, QueryOptions options) {
delete(entity, options, true);
public <T> Cancellable deleteAsynchronously(List<T> entities, DeletionListener<T> listener, QueryOptions options) {
return doBatchDeleteAsync(entities, listener, options);
}
@Override
public <T> Cancellable deleteAsynchronously(T entity) {
return doDeleteAsync(entity, null, null);
}
@Override
public <T> Cancellable deleteAsynchronously(T entity, QueryOptions options) {
return doDeleteAsync(entity, null, options);
}
@Override
public <T> Cancellable deleteAsynchronously(T entity, DeletionListener<T> listener) {
return doDeleteAsync(entity, listener, null);
}
@Override
public <T> Cancellable deleteAsynchronously(T entity, DeletionListener<T> listener, QueryOptions options) {
return doDeleteAsync(entity, listener, options);
}
@Override
@@ -197,7 +225,7 @@ public class CassandraTemplate extends CqlTemplate implements CassandraOperation
@Override
public <T> List<T> insert(List<T> entities, WriteOptions options) {
return batchInsert(entities, options, false);
return doBatchInsert(entities, options);
}
@Override
@@ -207,27 +235,76 @@ public class CassandraTemplate extends CqlTemplate implements CassandraOperation
@Override
public <T> T insert(T entity, WriteOptions options) {
return insert(entity, options, false);
return doInsert(entity, options);
}
/**
* @deprecated See {@link CassandraTemplate#insertAsynchronously(List)}
*/
@Deprecated
@Override
public <T> List<T> insertAsynchronously(List<T> entities) {
return insertAsynchronously(entities, null);
insertAsynchronously(entities, (WriteOptions) null);
return entities;
}
/**
* @deprecated See {@link CassandraTemplate#insertAsynchronously(List, WriteOptions)}
*/
@Deprecated
@Override
public <T> List<T> insertAsynchronously(List<T> entities, WriteOptions options) {
return batchInsert(entities, options, true);
doInsertAsynchronously(entities, null, options);
return entities;
}
@Override
public <T> Cancellable insertAsynchronously(List<T> entities, WriteListener<T> listener) {
return doInsertAsynchronously(entities, listener, null);
}
@Override
public <T> Cancellable insertAsynchronously(List<T> entities, WriteListener<T> listener, WriteOptions options) {
return doInsertAsynchronously(entities, listener, options);
}
/**
* This method resolves ambiguity the compiler sees as a result of type erasure between
* {@link #insertAsynchronously(Object, WriteListener, WriteOptions)} and {@link #insertAsynchronously(List<T>,
* WriteListener<T>, WriteOptions)}.
*/
protected <T> Cancellable doInsertAsynchronously(List<T> entities, WriteListener<T> listener, WriteOptions options) {
return doBatchInsertAsync(entities, listener, options);
}
/**
* @deprecated See {@link CqlOperations#insert(Object)}
*/
@Deprecated
@Override
public <T> T insertAsynchronously(T entity) {
return insertAsynchronously(entity, null);
insertAsynchronously(entity, null, null);
return entity;
}
/**
* @deprecated See {@link CqlOperations#insert(Object,WriteOptions)}
*/
@Deprecated
@Override
public <T> T insertAsynchronously(T entity, WriteOptions options) {
insertAsynchronously(entity, null, options);
return entity;
}
@Override
public <T> T insertAsynchronously(T entity, WriteOptions options) {
return insert(entity, options, true);
public <T> Cancellable insertAsynchronously(T entity, WriteListener<T> listener) {
return insertAsynchronously(entity, listener, null);
}
@Override
public <T> Cancellable insertAsynchronously(T entity, WriteListener<T> listener, WriteOptions options) {
return doInsertAsync(entity, listener, options);
}
@Override
@@ -372,7 +449,7 @@ public class CassandraTemplate extends CqlTemplate implements CassandraOperation
@Override
public <T> List<T> update(List<T> entities, WriteOptions options) {
return batchUpdate(entities, options, false);
return doBatchUpdate(entities, options);
}
@Override
@@ -382,27 +459,60 @@ public class CassandraTemplate extends CqlTemplate implements CassandraOperation
@Override
public <T> T update(T entity, WriteOptions options) {
return update(entity, options, false);
return doUpdate(entity, options);
}
@Override
public <T> List<T> updateAsynchronously(List<T> entities) {
return updateAsynchronously(entities, null);
doUpdateAsynchronously(entities, null, null);
return entities;
}
@Override
public <T> List<T> updateAsynchronously(List<T> entities, WriteOptions options) {
return batchUpdate(entities, options, true);
doUpdateAsynchronously(entities, null, options);
return entities;
}
@Override
public <T> Cancellable updateAsynchronously(List<T> entities, WriteListener<T> listener) {
return doUpdateAsynchronously(entities, listener, null);
}
@Override
public <T> Cancellable updateAsynchronously(List<T> entities, WriteListener<T> listener, WriteOptions options) {
return doUpdateAsynchronously(entities, listener, options);
}
/**
* This method resolves ambiguity the compiler sees as a result of type erasure between
* {@link #updateAsynchronously(Object, WriteListener, WriteOptions)} and {@link #updateAsynchronously(List<T>,
* WriteListener<T>, WriteOptions)}.
*/
protected <T> Cancellable doUpdateAsynchronously(List<T> entities, WriteListener<T> listener, WriteOptions options) {
return doBatchUpdateAsync(entities, listener, options);
}
@Override
public <T> T updateAsynchronously(T entity) {
return updateAsynchronously(entity, null);
updateAsynchronously(entity, null, null);
return entity;
}
@Override
public <T> T updateAsynchronously(T entity, WriteOptions options) {
return update(entity, options, true);
updateAsynchronously(entity, null, options);
return entity;
}
@Override
public <T> Cancellable updateAsynchronously(T entity, WriteListener<T> listener) {
return updateAsynchronously(entity, listener, null);
}
@Override
public <T> Cancellable updateAsynchronously(T entity, WriteListener<T> listener, WriteOptions options) {
return doUpdateAsync(entity, listener, options);
}
protected <T> CqlIdentifier determineTableName(T obj) {
@@ -498,42 +608,72 @@ public class CassandraTemplate extends CqlTemplate implements CassandraOperation
return null;
}
/**
* Perform the deletion on a list of objects
*
* @param tableName
* @param objectToRemove
*/
protected <T> void batchDelete(List<T> entities, QueryOptions options, boolean asynchronously) {
Assert.notEmpty(entities);
Batch b = createDeleteBatchQuery(getTableName(entities.get(0).getClass()).toCql(), entities, options,
cassandraConverter);
if (asynchronously) {
executeAsynchronously(b);
} else {
execute(b);
}
protected <T> void doBatchDelete(List<T> entities, QueryOptions options) {
execute(createDeleteBatchQuery(getTableName(entities.get(0).getClass()).toCql(), entities, options,
cassandraConverter));
}
protected <T> T insert(T entity, WriteOptions options, boolean asynchronously) {
protected <T> Cancellable doBatchDeleteAsync(final List<T> entities, final DeletionListener listener,
QueryOptions options) {
AsynchronousQueryListener aql = listener == null ? null : new AsynchronousQueryListener() {
@Override
public void onQueryComplete(ResultSetFuture rsf) {
try {
rsf.getUninterruptibly();
listener.onDeletionComplete(entities);
} catch (Exception e) {
listener.onException(translateExceptionIfPossible(e));
}
}
};
return executeAsynchronously(
createDeleteBatchQuery(getTableName(entities.get(0).getClass()).toCql(), entities, options, cassandraConverter),
aql);
}
protected <T> T doInsert(T entity, WriteOptions options) {
Assert.notNull(entity);
Insert insert = createInsertQuery(getTableName(entity.getClass()).toCql(), entity, options, cassandraConverter);
execute(insert);
return entity;
}
protected <T> Cancellable doInsertAsync(final T entity, final WriteListener<T> listener, WriteOptions options) {
Assert.notNull(entity);
Insert insert = createInsertQuery(getTableName(entity.getClass()).toCql(), entity, options, cassandraConverter);
if (asynchronously) {
executeAsynchronously(insert);
} else {
execute(insert);
}
AsynchronousQueryListener aql = listener == null ? null : new AsynchronousQueryListener() {
return entity;
@SuppressWarnings("unchecked")
@Override
public void onQueryComplete(ResultSetFuture rsf) {
try {
rsf.getUninterruptibly();
listener.onWriteComplete((Collection<T>) CollectionUtils.toList(entity));
} catch (Exception x) {
listener.onException(translateExceptionIfPossible(x));
}
}
};
return executeAsynchronously(insert, aql);
}
protected <T> List<T> batchInsert(List<T> entities, WriteOptions options, boolean asychronously) {
protected <T> List<T> doBatchInsert(List<T> entities, WriteOptions options) {
return doBatchWrite(entities, options, true);
}
protected <T> List<T> doBatchUpdate(List<T> entities, WriteOptions options) {
return doBatchWrite(entities, options, false);
}
protected <T> List<T> doBatchWrite(List<T> entities, WriteOptions options, boolean insert) {
if (entities == null || entities.size() == 0) {
if (logger.isWarnEnabled()) {
@@ -542,84 +682,147 @@ public class CassandraTemplate extends CqlTemplate implements CassandraOperation
return entities;
}
Batch b = createInsertBatchQuery(getTableName(entities.get(0).getClass()).toCql(), entities, options,
cassandraConverter);
if (asychronously) {
executeAsynchronously(b);
} else {
execute(b);
}
String tableName = getTableName(entities.get(0).getClass()).toCql();
Batch b = insert ? createInsertBatchQuery(tableName, entities, options, cassandraConverter)
: createUpdateBatchQuery(tableName, entities, options, cassandraConverter);
execute(b);
return entities;
}
/**
* Update a Batch of rows in a Cassandra CQL Table
* Asynchronously performs a batch insert or update.
*
* @param tableName
* @param entities
* @param optionsByName
* @param updateAsychronously
* @return
* @param entities The entities to insert or update.
* @param listener The listener that will receive notification of the completion of the batch insert or update. May be
* <code>null</code>.
* @param options The {@link WriteOptions} to use. May be <code>null</code>.
* @return A {@link Cancellable} that can be used to cancel the query if necessary.
*/
protected <T> List<T> batchUpdate(List<T> entities, WriteOptions options, boolean asychronously) {
Assert.notEmpty(entities);
Batch b = toUpdateBatchQuery(getTableName(entities.get(0).getClass()).toCql(), entities, options,
cassandraConverter);
if (asychronously) {
executeAsynchronously(b);
} else {
execute(b);
}
return entities;
protected <T> Cancellable doBatchInsertAsync(final List<T> entities, final WriteListener<T> listener,
WriteOptions options) {
return doBatchWriteAsync(entities, listener, options, true);
}
/**
* Perform the removal of a Row.
* Asynchronously performs a batch insert or update.
*
* @param tableName
* @param entity
* @param entities The entities to insert or update.
* @param listener The listener that will receive notification of the completion of the batch insert or update. May be
* <code>null</code>.
* @param options The {@link WriteOptions} to use. May be <code>null</code>.
* @return A {@link Cancellable} that can be used to cancel the query if necessary.
*/
protected <T> void delete(T entity, QueryOptions options, boolean asynchronously) {
protected <T> Cancellable doBatchUpdateAsync(final List<T> entities, final WriteListener<T> listener,
WriteOptions options) {
return doBatchWriteAsync(entities, listener, options, false);
}
/**
* Asynchronously performs a batch insert or update.
*
* @param entities The entities to insert or update.
* @param listener The listener that will receive notification of the completion of the batch insert or update. May be
* <code>null</code>.
* @param options The {@link WriteOptions} to use. May be <code>null</code>.
* @param insert If <code>true</code>, then an insert is performed, else an update is performed.
* @return A {@link Cancellable} that can be used to cancel the query if necessary.
*/
protected <T> Cancellable doBatchWriteAsync(final List<T> entities, final WriteListener<T> listener,
WriteOptions options, boolean insert) {
if (entities == null || entities.size() == 0) {
if (logger.isWarnEnabled()) {
logger.warn("no-op due to given null or empty list");
}
return new Cancellable() {
@Override
public void cancel() {
if (logger.isWarnEnabled()) {
logger.warn("no-op query cancellation due to given null or empty list");
}
}
};
}
String tableName = getTableName(entities.get(0).getClass()).toCql();
Batch b = insert ? createInsertBatchQuery(tableName, entities, options, cassandraConverter)
: createUpdateBatchQuery(tableName, entities, options, cassandraConverter);
AsynchronousQueryListener aql = listener == null ? null : new AsynchronousQueryListener() {
@Override
public void onQueryComplete(ResultSetFuture rsf) {
try {
rsf.getUninterruptibly();
listener.onWriteComplete(entities);
} catch (Exception e) {
listener.onException(translateExceptionIfPossible(e));
}
}
};
return executeAsynchronously(b, aql);
}
protected <T> void doDelete(T entity, QueryOptions options) {
Assert.notNull(entity);
Delete delete = createDeleteQuery(getTableName(entity.getClass()).toCql(), entity, options, cassandraConverter);
execute(delete);
}
protected <T> Cancellable doDeleteAsync(final T entity, final DeletionListener listener, QueryOptions options) {
Assert.notNull(entity);
Delete delete = createDeleteQuery(getTableName(entity.getClass()).toCql(), entity, options, cassandraConverter);
if (asynchronously) {
executeAsynchronously(delete);
} else {
execute(delete);
}
AsynchronousQueryListener aql = listener == null ? null : new AsynchronousQueryListener() {
@Override
public void onQueryComplete(ResultSetFuture rsf) {
try {
rsf.getUninterruptibly();
listener.onDeletionComplete(CollectionUtils.toList(entity));
} catch (Exception x) {
listener.onException(translateExceptionIfPossible(x));
}
}
};
return executeAsynchronously(delete, aql);
}
/**
* Update a row into a Cassandra CQL Table
*
* @param tableName
* @param entity
* @param optionsByName
* @param updateAsychronously
* @return
*/
protected <T> T update(T entity, WriteOptions options, boolean asychronously) {
protected <T> T doUpdate(T entity, WriteOptions options) {
Assert.notNull(entity);
Update update = createUpdateQuery(getTableName(entity.getClass()).toCql(), entity, options, cassandraConverter);
execute(update);
return entity;
}
protected <T> Cancellable doUpdateAsync(final T entity, final WriteListener<T> listener, WriteOptions options) {
Assert.notNull(entity);
Update update = toUpdateQuery(getTableName(entity.getClass()).toCql(), entity, options, cassandraConverter);
Update update = createUpdateQuery(getTableName(entity.getClass()).toCql(), entity, options, cassandraConverter);
if (asychronously) {
executeAsynchronously(update);
} else {
execute(update);
}
AsynchronousQueryListener aql = listener == null ? null : new AsynchronousQueryListener() {
return entity;
@SuppressWarnings("unchecked")
@Override
public void onQueryComplete(ResultSetFuture rsf) {
try {
rsf.getUninterruptibly();
listener.onWriteComplete((Collection<T>) CollectionUtils.toList(entity));
} catch (Exception x) {
listener.onException(translateExceptionIfPossible(x));
}
}
};
return executeAsynchronously(update, aql);
}
/**
@@ -635,20 +838,21 @@ public class CassandraTemplate extends CqlTemplate implements CassandraOperation
EntityWriter<Object, Object> entityWriter) {
Insert insert = QueryBuilder.insertInto(tableName);
/*
* Write properties
*/
entityWriter.write(objectToSave, insert);
/*
* Add Query Options
*/
CqlTemplate.addWriteOptions(insert, options);
return insert;
}
/**
* @deprecated Method renamed. Use {@link #createUpdateQuery(String, Object, WriteOptions, EntityWriter)}
* @see #createUpdateQuery(String, Object, WriteOptions, EntityWriter)
*/
@Deprecated
public static Update toUpdateQueryX(String tableName, Object objectToSave, WriteOptions options,
EntityWriter<Object, Object> entityWriter) {
return createUpdateQuery(tableName, objectToSave, options, entityWriter);
}
/**
* Generates a Query Object for an Update
*
@@ -658,24 +862,25 @@ public class CassandraTemplate extends CqlTemplate implements CassandraOperation
* @param optionsByName
* @return The Query object to run with session.execute();
*/
public static Update toUpdateQuery(String tableName, Object objectToSave, WriteOptions options,
public static Update createUpdateQuery(String tableName, Object objectToSave, WriteOptions options,
EntityWriter<Object, Object> entityWriter) {
Update update = QueryBuilder.update(tableName);
/*
* Write properties
*/
entityWriter.write(objectToSave, update);
/*
* Add Query Options
*/
CqlTemplate.addWriteOptions(update, options);
return update;
}
/**
* @deprecated Method renamed. Use {@link #createUpdateBatchQuery(String, List, WriteOptions, EntityWriter)}
* @see #createUpdateBatchQuery(String, List, WriteOptions, EntityWriter)
*/
@Deprecated
public static <T> Batch toUpdateBatchQuery(String tableName, List<T> objectsToSave, WriteOptions options,
EntityWriter<Object, Object> entityWriter) {
return createUpdateBatchQuery(tableName, objectsToSave, options, entityWriter);
}
/**
* Generates a Batch Object for multiple Updates
*
@@ -685,27 +890,18 @@ public class CassandraTemplate extends CqlTemplate implements CassandraOperation
* @param optionsByName
* @return The Query object to run with session.execute();
*/
public static <T> Batch toUpdateBatchQuery(String tableName, List<T> objectsToSave, WriteOptions options,
public static <T> Batch createUpdateBatchQuery(String tableName, List<T> objectsToSave, WriteOptions options,
EntityWriter<Object, Object> entityWriter) {
/*
* Return variable is a Batch statement
*/
Batch b = QueryBuilder.batch();
for (T objectToSave : objectsToSave) {
b.add(toUpdateQuery(tableName, objectToSave, options, entityWriter));
b.add(createUpdateQuery(tableName, objectToSave, options, entityWriter));
}
/*
* Add Query Options
*/
CqlTemplate.addQueryOptions(b, options);
return b;
}
/**
@@ -745,13 +941,9 @@ public class CassandraTemplate extends CqlTemplate implements CassandraOperation
Delete.Selection ds = QueryBuilder.delete();
Delete delete = ds.from(tableName);
Where w = delete.where();
entityWriter.write(object, w);
CqlTemplate.addQueryOptions(delete, options);
return delete;
}
@@ -790,4 +982,59 @@ public class CassandraTemplate extends CqlTemplate implements CassandraOperation
truncate(mappingContext.getPersistentEntity(clazz).getTableName());
}
@Override
public <T> Cancellable selectOneAsynchronously(Select select, Class<T> type, QueryForObjectListener<T> listener) {
return selectOneAsynchronously(select, type, listener, null);
}
@Override
public <T> Cancellable selectOneAsynchronously(String cql, Class<T> type, QueryForObjectListener<T> listener) {
return selectOneAsynchronously(cql, type, listener, null);
}
@Override
public <T> Cancellable selectOneAsynchronously(Select select, Class<T> type, QueryForObjectListener<T> listener,
QueryOptions options) {
return doSelectOneAsync(select, type, listener, options);
}
@Override
public <T> Cancellable selectOneAsynchronously(String cql, Class<T> type, QueryForObjectListener<T> listener,
QueryOptions options) {
return doSelectOneAsync(cql, type, listener, options);
}
protected <T> Cancellable doSelectOneAsync(final Object query, final Class<T> type,
final QueryForObjectListener<T> listener, QueryOptions options) {
AsynchronousQueryListener aql = new AsynchronousQueryListener() {
@Override
public void onQueryComplete(ResultSetFuture rsf) {
try {
ResultSet rs = rsf.getUninterruptibly();
Iterator<Row> iterator = rs.iterator();
if (iterator.hasNext()) {
Row row = iterator.next();
T result = new CassandraConverterRowCallback<T>(cassandraConverter, type).doWith(row);
if (iterator.hasNext()) {
throw new DuplicateKeyException("found two or more results in query " + query);
}
listener.onQueryComplete(result);
}
} catch (Exception e) {
listener.onException(translateExceptionIfPossible(e));
}
}
};
if (query instanceof String) {
return queryAsynchronously((String) query, aql, options);
}
if (query instanceof Select) {
return queryAsynchronously((Select) query, aql);
}
throw new IllegalArgumentException(String.format("Expected type String or Select; got type [%s] with value [%s]",
query.getClass(), query));
}
}

15
spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/DeletionListener.java Normal file
View File

@@ -0,0 +1,15 @@
package org.springframework.data.cassandra.core;
import java.util.Collection;
import com.datastax.driver.core.ResultSet;
public interface DeletionListener<T> {
void onDeletionComplete(Collection<T> entities);
/**
* Called if an exception is raised while getting or converting the {@link ResultSet}.
*/
void onException(Exception x);
}

40
spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/WriteListener.java Normal file
View File

@@ -0,0 +1,40 @@
/*
* Copyright 2013-2014 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.data.cassandra.core;
import java.util.Collection;
import com.datastax.driver.core.ResultSet;
/**
* Listener for asynchronous repository insert or update methods.
*
* @author Matthew T. Adams
*/
public interface WriteListener<T> {
/**
* Called upon completion of the asynchronous insert or update.
*
* @param entities The entities inserted or updated.
*/
void onWriteComplete(Collection<T> entities);
/**
* Called if an exception is raised while getting or converting the {@link ResultSet}.
*/
void onException(Exception x);
}

28
spring-data-cassandra/src/main/java/org/springframework/data/cassandra/mapping/BasicCassandraPersistentProperty.java
View File

@@ -192,23 +192,23 @@ public class BasicCassandraPersistentProperty extends AnnotationBasedPersistentP
if (type.isCollection()) {
switch (type) {
case MAP:
ensureTypeArguments(annotation.typeArguments().length, 2);
return DataType.map(getDataTypeFor(annotation.typeArguments()[0]),
getDataTypeFor(annotation.typeArguments()[1]));
case MAP:
ensureTypeArguments(annotation.typeArguments().length, 2);
return DataType.map(getDataTypeFor(annotation.typeArguments()[0]),
getDataTypeFor(annotation.typeArguments()[1]));
case LIST:
ensureTypeArguments(annotation.typeArguments().length, 1);
return DataType.list(getDataTypeFor(annotation.typeArguments()[0]));
case LIST:
ensureTypeArguments(annotation.typeArguments().length, 1);
return DataType.list(getDataTypeFor(annotation.typeArguments()[0]));
case SET:
ensureTypeArguments(annotation.typeArguments().length, 1);
return DataType.set(getDataTypeFor(annotation.typeArguments()[0]));
case SET:
ensureTypeArguments(annotation.typeArguments().length, 1);
return DataType.set(getDataTypeFor(annotation.typeArguments()[0]));
default:
throw new InvalidDataAccessApiUsageException(
String.format("unknown multivalued DataType [%s] for property [%s] in entity [%s]", type, getType(),
getOwner().getName()));
default:
throw new InvalidDataAccessApiUsageException(String.format(
"unknown multivalued DataType [%s] for property [%s] in entity [%s]", type, getType(), getOwner()
.getName()));
}
} else {

1
spring-data-cassandra/src/main/java/org/springframework/data/cassandra/mapping/CassandraPrimaryKeyColumnAnnotationComparator.java
View File

@@ -31,7 +31,6 @@ import org.springframework.cassandra.core.PrimaryKeyType;
*
* @see PrimaryKeyType#compareTo(PrimaryKeyType)
* @see Ordering#compareTo(Ordering)
*
* @author Matthew T. Adams
*/
public enum CassandraPrimaryKeyColumnAnnotationComparator implements Comparator<PrimaryKeyColumn> {

1
spring-data-cassandra/src/main/java/org/springframework/data/cassandra/mapping/VerifierMappingExceptions.java
View File

@@ -25,7 +25,6 @@ import org.springframework.data.mapping.model.MappingException;
* framework to communicate all verification errors to the user of the framework, rather than one at a time.
*
* @author David Webb
*
*/
@SuppressWarnings("serial")
public class VerifierMappingExceptions extends MappingException {

1
spring-data-cassandra/src/main/java/org/springframework/data/cassandra/repository/config/CassandraRepositoryConfigurationExtension.java
View File

@@ -31,7 +31,6 @@ import org.w3c.dom.Element;
* {@link RepositoryConfigurationExtension} for Cassandra.
*
* @author Alex Shvid
*
*/
public class CassandraRepositoryConfigurationExtension extends RepositoryConfigurationExtensionSupport {

1
spring-data-cassandra/src/main/java/org/springframework/data/cassandra/repository/query/CassandraEntityInformation.java
View File

@@ -23,7 +23,6 @@ import org.springframework.data.repository.core.EntityInformation;
* Cassandra specific {@link EntityInformation}.
*
* @author Alex Shvid
*
*/
public interface CassandraEntityInformation<T, ID extends Serializable> extends EntityInformation<T, ID>,
CassandraEntityMetadata<T> {

1
spring-data-cassandra/src/main/java/org/springframework/data/cassandra/repository/query/CassandraEntityMetadata.java
View File

@@ -23,7 +23,6 @@ import org.springframework.data.repository.core.EntityMetadata;
*
* @author Alex Shvid
* @author Matthew T. Adams
*
* @param <T>
*/
public interface CassandraEntityMetadata<T> extends EntityMetadata<T> {

3
spring-data-cassandra/src/main/java/org/springframework/data/cassandra/repository/query/CassandraParameterAccessor.java
View File

@@ -2,5 +2,4 @@ package org.springframework.data.cassandra.repository.query;
import org.springframework.data.repository.query.ParameterAccessor;
public interface CassandraParameterAccessor extends ParameterAccessor {
}
public interface CassandraParameterAccessor extends ParameterAccessor {}

20
spring-data-cassandra/src/main/java/org/springframework/data/cassandra/repository/query/CassandraQueryCreator.java
View File

@@ -167,22 +167,22 @@ class CassandraQueryCreator extends AbstractQueryCreator<Select, Clause> {
//
// return isSimpleComparisionPossible(part) ? criteria.ne(parameters.nextConverted(property))
// : createLikeRegexCriteriaOrThrow(part, property, criteria, parameters, true);
default:
throw new UnsupportedCassandraQueryOperatorException(String.format(""));
default:
throw new UnsupportedCassandraQueryOperatorException(String.format(""));
}
}
private boolean isSimpleComparisionPossible(Part part) {
switch (part.shouldIgnoreCase()) {
case NEVER:
return true;
case WHEN_POSSIBLE:
return part.getProperty().getType() != String.class;
case ALWAYS:
return false;
default:
return true;
case NEVER:
return true;
case WHEN_POSSIBLE:
return part.getProperty().getType() != String.class;
case ALWAYS:
return false;
default:
return true;
}
}

2
spring-data-cassandra/src/main/java/org/springframework/data/cassandra/repository/query/CassandraQueryMethod.java
View File

@@ -182,4 +182,4 @@ public class CassandraQueryMethod extends QueryMethod {
public boolean isDateParameter(int parameterIndex) {
return dateParameterIndexes.contains(parameterIndex);
}
}
}

1
spring-data-cassandra/src/main/java/org/springframework/data/cassandra/repository/query/PartTreeCassandraQuery.java
View File

@@ -24,7 +24,6 @@ import org.springframework.data.repository.query.parser.PartTree;
/**
* {@link RepositoryQuery} implementation for Cassandra.
*
*/
public class PartTreeCassandraQuery extends AbstractCassandraQuery {

3
spring-data-cassandra/src/main/java/org/springframework/data/cassandra/repository/support/BasicMapId.java
View File

@@ -52,8 +52,7 @@ public class BasicMapId implements MapId {
protected Map<String, Serializable> map = new HashMap<String, Serializable>();
public BasicMapId() {
}
public BasicMapId() {}
public BasicMapId(Map<String, Serializable> map) {

1
spring-data-cassandra/src/main/java/org/springframework/data/cassandra/repository/support/CassandraRepositoryFactoryBean.java
View File

@@ -28,7 +28,6 @@ import org.springframework.util.Assert;
* {@link org.springframework.beans.factory.FactoryBean} to create {@link TypedIdCassandraRepository} instances.
*
* @author Alex Shvid
*
*/
public class CassandraRepositoryFactoryBean<T extends Repository<S, ID>, S, ID extends Serializable> extends
RepositoryFactoryBeanSupport<T, S, ID> {

Some files were not shown because too many files have changed in this diff Show More