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

DATACASS-292 - Polish.

Browse Source
This commit is contained in:
John Blum
2016-11-22 21:54:32 -08:00
parent 8af23880e9
commit 542ce24173
30 changed files with 2302 additions and 2384 deletions
Download Patch File Download Diff File Expand all files Collapse all files

803
spring-cql/src/main/java/org/springframework/cassandra/core/AsyncCqlOperations.java
View File

File diff suppressed because it is too large Load Diff

1018
spring-cql/src/main/java/org/springframework/cassandra/core/AsyncCqlTemplate.java
View File

File diff suppressed because it is too large Load Diff

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

@@ -15,12 +15,12 @@
*/
package org.springframework.cassandra.core;
import org.springframework.util.concurrent.ListenableFuture;
import com.datastax.driver.core.PreparedStatement;
import com.datastax.driver.core.Session;
import com.datastax.driver.core.exceptions.DriverException;
import org.springframework.util.concurrent.ListenableFuture;
/**
* One of the two central callback interfaces used by the {@link AsyncCqlTemplate} class. This interface prepares a CQL
* statement returning a {@link org.springframework.util.concurrent.ListenableFuture} given a {@link Session}, provided
@@ -43,11 +43,12 @@ public interface AsyncPreparedStatementCreator {
* Create a statement in this session. Allows implementations to use {@link PreparedStatement}s. The
* {@link CqlTemplate} will attempt to cache the {@link PreparedStatement}s for future use without the overhead of
* re-preparing on the entire cluster.
*
*
* @param session Session to use to create statement, must not be {@literal null}.
* @return a prepared statement
* @throws DriverException there is no need to catch DriverException that may be thrown in the implementation of this
* method. The {@link CqlTemplate} class will handle them.
*/
ListenableFuture<PreparedStatement> createPreparedStatement(Session session) throws DriverException;
}

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

@@ -15,10 +15,10 @@
*/
package org.springframework.cassandra.core;
import org.springframework.dao.DataAccessException;
import com.datastax.driver.core.Session;
import com.datastax.driver.core.exceptions.DriverException;
import org.springframework.dao.DataAccessException;
import org.springframework.util.concurrent.ListenableFuture;
/**
@@ -44,7 +44,7 @@ public interface AsyncSessionCallback<T> {
* objects. Note that there's special support for single step actions: see {@link CqlTemplate#queryForObject} etc. A
* thrown {@link RuntimeException} is treated as application exception: it gets propagated to the caller of the
* template.
*
*
* @param session active Cassandra Session, must not be {@literal null}.
* @return a result object, or {@code null} if none.
* @throws DriverException if thrown by a Session method, to be auto-converted to a {@link DataAccessException}.

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

File diff suppressed because it is too large Load Diff

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

@@ -15,25 +15,13 @@
*/
package org.springframework.cassandra.core;
import static org.springframework.cassandra.core.cql.CqlIdentifier.cqlId;
import java.util.ArrayList;
import java.util.Collection;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.stream.StreamSupport;
import com.datastax.driver.core.querybuilder.Insert;
import com.datastax.driver.core.querybuilder.Update;
import org.springframework.cassandra.support.CassandraAccessor;
import org.springframework.dao.DataAccessException;
import org.springframework.dao.support.DataAccessUtils;
import org.springframework.util.Assert;
import java.util.function.Function;
import com.datastax.driver.core.BoundStatement;
import com.datastax.driver.core.ConsistencyLevel;
import com.datastax.driver.core.Host;
import com.datastax.driver.core.PreparedStatement;
import com.datastax.driver.core.ResultSet;
@@ -42,8 +30,11 @@ import com.datastax.driver.core.Session;
import com.datastax.driver.core.SimpleStatement;
import com.datastax.driver.core.Statement;
import com.datastax.driver.core.exceptions.DriverException;
import com.datastax.driver.core.policies.RetryPolicy;
import com.datastax.driver.core.querybuilder.QueryBuilder;
import org.springframework.cassandra.support.CassandraAccessor;
import org.springframework.dao.DataAccessException;
import org.springframework.dao.support.DataAccessUtils;
import org.springframework.util.Assert;
/**
* <b>This is the central class in the CQL core package.</b> It simplifies the use of CQL and helps to avoid common
@@ -88,39 +79,20 @@ import com.datastax.driver.core.querybuilder.QueryBuilder;
public class CqlTemplate extends CassandraAccessor implements CqlOperations {
/**
* Placeholder for default values.
*/
private final static Statement DEFAULTS = QueryBuilder.select().from("DEFAULT");
/**
* If this variable is set to a non-negative value, it will be used for setting the {@code fetchSize} property on
* statements used for query processing.
*/
private int fetchSize = -1;
/**
* If this variable is set to a value, it will be used for setting the {@code retryPolicy} property on statements used
* for query processing.
*/
private RetryPolicy retryPolicy;
/**
* If this variable is set to a value, it will be used for setting the {@code consistencyLevel} property on statements
* used for query processing.
*/
private com.datastax.driver.core.ConsistencyLevel consistencyLevel;
/**
* Construct a new {@link CqlTemplate}. Note: The {@link Session} has to be set before using the instance.
* Constructs a new, uninitialized {@link CqlTemplate}.
*
* Note: The {@link Session} has to be set before using the instance.
*
* @see #setSession(Session)
*/
public CqlTemplate() {}
/**
* Construct a new {@link CqlTemplate}, given a {@link Session}.
* Constructs a new {@link CqlTemplate} initialized with the given {@link Session}.
*
* @param session the active Cassandra {@link Session}.
* @throws IllegalStateException if {@link Session} is {@literal null}.
* @see com.datastax.driver.core.Session
*/
public CqlTemplate(Session session) {
@@ -129,63 +101,6 @@ public class CqlTemplate extends CassandraAccessor implements CqlOperations {
setSession(session);
}
/**
* Set the fetch size for this {@link CqlTemplate}. This is important for processing large result sets: Setting this
* higher than the default value will increase processing speed at the cost of memory consumption; setting this lower
* can avoid transferring row data that will never be read by the application. Default is -1, indicating to use the
* CQL driver's default configuration (i.e. to not pass a specific fetch size setting on to the driver).
*
* @see Statement#setFetchSize(int)
*/
public void setFetchSize(int fetchSize) {
this.fetchSize = fetchSize;
}
/**
* @return the fetch size specified for this {@link CqlTemplate}.
*/
public int getFetchSize() {
return this.fetchSize;
}
/**
* Set the retry policy for this {@link CqlTemplate}. This is important for defining behavior when a request
* fails.
*
* @see Statement#setRetryPolicy(RetryPolicy)
* @see RetryPolicy
*/
public void setRetryPolicy(RetryPolicy retryPolicy) {
this.retryPolicy = retryPolicy;
}
/**
* @return the {@link RetryPolicy} specified for this {@link CqlTemplate}.
*/
public RetryPolicy getRetryPolicy() {
return retryPolicy;
}
/**
* Set the consistency level for this {@link CqlTemplate}. Consistency level defines the number of nodes
* involved into query processing. Relaxed consistency level settings use fewer nodes but eventual consistency is more
* likely to occur while a higher consistency level involves more nodes to obtain results with a higher consistency
* guarantee.
*
* @see Statement#setConsistencyLevel(ConsistencyLevel)
* @see RetryPolicy
*/
public void setConsistencyLevel(ConsistencyLevel consistencyLevel) {
this.consistencyLevel = consistencyLevel;
}
/**
* @return the {@link ConsistencyLevel} specified for this {@link CqlTemplate}.
*/
public ConsistencyLevel getConsistencyLevel() {
return consistencyLevel;
}
// -------------------------------------------------------------------------
// Methods dealing with a plain com.datastax.driver.core.Session
// -------------------------------------------------------------------------
@@ -202,7 +117,7 @@ public class CqlTemplate extends CassandraAccessor implements CqlOperations {
try {
return action.doInSession(getSession());
} catch (DriverException e) {
throw translateException("SessionCallback", getCql(action), e);
throw translateException("SessionCallback", toCql(action), e);
}
}
@@ -227,22 +142,21 @@ public class CqlTemplate extends CassandraAccessor implements CqlOperations {
* @see org.springframework.cassandra.core.CqlOperationsNG#query(java.lang.String, org.springframework.cassandra.core.ResultSetExtractor)
*/
@Override
public <T> T query(String cql, ResultSetExtractor<T> rse) throws DataAccessException {
public <T> T query(String cql, ResultSetExtractor<T> resultSetExtractor) throws DataAccessException {
Assert.hasText(cql, "CQL must not be empty");
Assert.notNull(rse, "ResultSetExtractor must not be null");
Assert.notNull(resultSetExtractor, "ResultSetExtractor must not be null");
try {
if (logger.isDebugEnabled()) {
logger.debug("Executing CQL Statement [{}]", cql);
}
SimpleStatement simpleStatement = new SimpleStatement(cql);
SimpleStatement statement = applyStatementSettings(new SimpleStatement(cql));
applyStatementSettings(simpleStatement);
ResultSet results = getSession().execute(statement);
return rse.extractData(getSession().execute(simpleStatement));
return resultSetExtractor.extractData(results);
} catch (DriverException e) {
throw translateException("Query", cql, e);
}
@@ -253,8 +167,8 @@ public class CqlTemplate extends CassandraAccessor implements CqlOperations {
* @see org.springframework.cassandra.core.CqlOperationsNG#query(java.lang.String, org.springframework.cassandra.core.RowCallbackHandler)
*/
@Override
public void query(String cql, RowCallbackHandler rch) throws DataAccessException {
query(cql, new RowCallbackHandlerResultSetExtractor(rch));
public void query(String cql, RowCallbackHandler rowCallbackHandler) throws DataAccessException {
query(cql, newResultSetExtractor(rowCallbackHandler));
}
/*
@@ -263,45 +177,7 @@ public class CqlTemplate extends CassandraAccessor implements CqlOperations {
*/
@Override
public <T> List<T> query(String cql, RowMapper<T> rowMapper) throws DataAccessException {
return query(cql, new RowMapperResultSetExtractor<>(rowMapper));
}
/*
* (non-Javadoc)
* @see org.springframework.cassandra.core.CqlOperationsNG#queryForObject(java.lang.String, org.springframework.cassandra.core.RowMapper)
*/
@Override
public <T> T queryForObject(String cql, RowMapper<T> rowMapper) throws DataAccessException {
List<T> results = query(cql, rowMapper);
return DataAccessUtils.requiredSingleResult(results);
}
/*
* (non-Javadoc)
* @see org.springframework.cassandra.core.CqlOperationsNG#queryForObject(java.lang.String, java.lang.Class)
*/
@Override
public <T> T queryForObject(String cql, Class<T> requiredType) throws DataAccessException {
return queryForObject(cql, getSingleColumnRowMapper(requiredType));
}
/*
* (non-Javadoc)
* @see org.springframework.cassandra.core.CqlOperationsNG#queryForMap(java.lang.String)
*/
@Override
public Map<String, Object> queryForMap(String cql) throws DataAccessException {
return queryForObject(cql, getColumnMapRowMapper());
}
/*
* (non-Javadoc)
* @see org.springframework.cassandra.core.CqlOperationsNG#queryForList(java.lang.String, java.lang.Class)
*/
@Override
public <T> List<T> queryForList(String cql, Class<T> elementType) throws DataAccessException {
return query(cql, getSingleColumnRowMapper(elementType));
return query(cql, newResultSetExtractor(rowMapper));
}
/*
@@ -310,7 +186,43 @@ public class CqlTemplate extends CassandraAccessor implements CqlOperations {
*/
@Override
public List<Map<String, Object>> queryForList(String cql) throws DataAccessException {
return query(cql, getColumnMapRowMapper());
return query(cql, newResultSetExtractor(newColumnMapRowMapper()));
}
/*
* (non-Javadoc)
* @see org.springframework.cassandra.core.CqlOperationsNG#queryForList(java.lang.String, java.lang.Class)
*/
@Override
public <T> List<T> queryForList(String cql, Class<T> elementType) throws DataAccessException {
return query(cql, newResultSetExtractor(newSingleColumnRowMapper(elementType)));
}
/*
* (non-Javadoc)
* @see org.springframework.cassandra.core.CqlOperationsNG#queryForMap(java.lang.String)
*/
@Override
public Map<String, Object> queryForMap(String cql) throws DataAccessException {
return queryForObject(cql, newColumnMapRowMapper());
}
/*
* (non-Javadoc)
* @see org.springframework.cassandra.core.CqlOperationsNG#queryForObject(java.lang.String, java.lang.Class)
*/
@Override
public <T> T queryForObject(String cql, Class<T> requiredType) throws DataAccessException {
return queryForObject(cql, newSingleColumnRowMapper(requiredType));
}
/*
* (non-Javadoc)
* @see org.springframework.cassandra.core.CqlOperationsNG#queryForObject(java.lang.String, org.springframework.cassandra.core.RowMapper)
*/
@Override
public <T> T queryForObject(String cql, RowMapper<T> rowMapper) throws DataAccessException {
return DataAccessUtils.requiredSingleResult(query(cql, newResultSetExtractor(rowMapper)));
}
/*
@@ -327,8 +239,8 @@ public class CqlTemplate extends CassandraAccessor implements CqlOperations {
* @see org.springframework.cassandra.core.CqlOperationsNG#queryForRows(java.lang.String)
*/
@Override
public Iterator<Row> queryForRows(String cql) throws DataAccessException {
return queryForResultSet(cql).iterator();
public Iterable<Row> queryForRows(String cql) throws DataAccessException {
return () -> queryForResultSet(cql).iterator();
}
// -------------------------------------------------------------------------
@@ -352,20 +264,17 @@ public class CqlTemplate extends CassandraAccessor implements CqlOperations {
* @see org.springframework.cassandra.core.CqlOperationsNG#query(com.datastax.driver.core.Statement, org.springframework.cassandra.core.ResultSetExtractor)
*/
@Override
public <T> T query(Statement statement, ResultSetExtractor<T> rse) throws DataAccessException {
public <T> T query(Statement statement, ResultSetExtractor<T> resultSetExtractor) throws DataAccessException {
Assert.notNull(statement, "CQL Statement must not be null");
Assert.notNull(rse, "ResultSetExtractor must not be null");
Assert.notNull(resultSetExtractor, "ResultSetExtractor must not be null");
try {
if (logger.isDebugEnabled()) {
logger.debug("Executing CQL Statement [{}]", statement);
}
applyStatementSettings(statement);
return rse.extractData(getSession().execute(statement));
return resultSetExtractor.extractData(getSession().execute(applyStatementSettings(statement)));
} catch (DriverException e) {
throw translateException("Query", statement.toString(), e);
}
@@ -376,8 +285,8 @@ public class CqlTemplate extends CassandraAccessor implements CqlOperations {
* @see org.springframework.cassandra.core.CqlOperationsNG#query(com.datastax.driver.core.Statement, org.springframework.cassandra.core.RowCallbackHandler)
*/
@Override
public void query(Statement statement, RowCallbackHandler rch) throws DataAccessException {
query(statement, new RowCallbackHandlerResultSetExtractor(rch));
public void query(Statement statement, RowCallbackHandler rowCallbackHandler) throws DataAccessException {
query(statement, newResultSetExtractor(rowCallbackHandler));
}
/*
@@ -386,45 +295,7 @@ public class CqlTemplate extends CassandraAccessor implements CqlOperations {
*/
@Override
public <T> List<T> query(Statement statement, RowMapper<T> rowMapper) throws DataAccessException {
return query(statement, new RowMapperResultSetExtractor<>(rowMapper));
}
/*
* (non-Javadoc)
* @see org.springframework.cassandra.core.CqlOperationsNG#queryForObject(com.datastax.driver.core.Statement, org.springframework.cassandra.core.RowMapper)
*/
@Override
public <T> T queryForObject(Statement statement, RowMapper<T> rowMapper) throws DataAccessException {
List<T> results = query(statement, rowMapper);
return DataAccessUtils.requiredSingleResult(results);
}
/*
* (non-Javadoc)
* @see org.springframework.cassandra.core.CqlOperationsNG#queryForObject(com.datastax.driver.core.Statement, java.lang.Class)
*/
@Override
public <T> T queryForObject(Statement statement, Class<T> requiredType) throws DataAccessException {
return queryForObject(statement, getSingleColumnRowMapper(requiredType));
}
/*
* (non-Javadoc)
* @see org.springframework.cassandra.core.CqlOperationsNG#queryForMap(com.datastax.driver.core.Statement)
*/
@Override
public Map<String, Object> queryForMap(Statement statement) throws DataAccessException {
return queryForObject(statement, getColumnMapRowMapper());
}
/*
* (non-Javadoc)
* @see org.springframework.cassandra.core.CqlOperationsNG#queryForList(com.datastax.driver.core.Statement, java.lang.Class)
*/
@Override
public <T> List<T> queryForList(Statement statement, Class<T> elementType) throws DataAccessException {
return query(statement, getSingleColumnRowMapper(elementType));
return query(statement, newResultSetExtractor(rowMapper));
}
/*
@@ -433,7 +304,43 @@ public class CqlTemplate extends CassandraAccessor implements CqlOperations {
*/
@Override
public List<Map<String, Object>> queryForList(Statement statement) throws DataAccessException {
return query(statement, getColumnMapRowMapper());
return query(statement, newResultSetExtractor(newColumnMapRowMapper()));
}
/*
* (non-Javadoc)
* @see org.springframework.cassandra.core.CqlOperationsNG#queryForList(com.datastax.driver.core.Statement, java.lang.Class)
*/
@Override
public <T> List<T> queryForList(Statement statement, Class<T> elementType) throws DataAccessException {
return query(statement, newResultSetExtractor(newSingleColumnRowMapper(elementType)));
}
/*
* (non-Javadoc)
* @see org.springframework.cassandra.core.CqlOperationsNG#queryForMap(com.datastax.driver.core.Statement)
*/
@Override
public Map<String, Object> queryForMap(Statement statement) throws DataAccessException {
return queryForObject(statement, newColumnMapRowMapper());
}
/*
* (non-Javadoc)
* @see org.springframework.cassandra.core.CqlOperationsNG#queryForObject(com.datastax.driver.core.Statement, java.lang.Class)
*/
@Override
public <T> T queryForObject(Statement statement, Class<T> requiredType) throws DataAccessException {
return queryForObject(statement, newSingleColumnRowMapper(requiredType));
}
/*
* (non-Javadoc)
* @see org.springframework.cassandra.core.CqlOperationsNG#queryForObject(com.datastax.driver.core.Statement, org.springframework.cassandra.core.RowMapper)
*/
@Override
public <T> T queryForObject(Statement statement, RowMapper<T> rowMapper) throws DataAccessException {
return DataAccessUtils.requiredSingleResult(query(statement, newResultSetExtractor(rowMapper)));
}
/*
@@ -450,72 +357,30 @@ public class CqlTemplate extends CassandraAccessor implements CqlOperations {
* @see org.springframework.cassandra.core.CqlOperationsNG#queryForRows(com.datastax.driver.core.Statement)
*/
@Override
public Iterator<Row> queryForRows(Statement statement) throws DataAccessException {
return queryForResultSet(statement).iterator();
public Iterable<Row> queryForRows(Statement statement) throws DataAccessException {
return () -> queryForResultSet(statement).iterator();
}
// -------------------------------------------------------------------------
// Methods dealing with prepared statements
// Methods dealing with com.datastax.driver.core.PreparedStatement
// -------------------------------------------------------------------------
/*
* (non-Javadoc)
* @see org.springframework.cassandra.core.CqlOperationsNG#execute(org.springframework.cassandra.core.PreparedStatementCreator, org.springframework.cassandra.core.PreparedStatementCallback)
* @see org.springframework.cassandra.core.CqlOperationsNG#execute(java.lang.String, java.lang.Object[])
*/
@Override
public <T> T execute(PreparedStatementCreator psc, PreparedStatementCallback<T> action) throws DataAccessException {
Assert.notNull(psc, "PreparedStatementCreator must not be null");
Assert.notNull(action, "PreparedStatementCallback object must not be null");
try {
if (logger.isDebugEnabled()) {
logger.debug("Preparing statement [{}] using {}", getCql(psc), psc);
}
PreparedStatement preparedStatement = psc.createPreparedStatement(getSession());
applyStatementSettings(preparedStatement);
return action.doInPreparedStatement(preparedStatement);
} catch (DriverException e) {
throw translateException("PreparedStatementCallback", getCql(psc), e);
}
public boolean execute(String cql, Object... args) throws DataAccessException {
return execute(cql, newPreparedStatementBinder(args));
}
/*
* (non-Javadoc)
* @see org.springframework.cassandra.core.CqlOperationsNG#query(org.springframework.cassandra.core.PreparedStatementCreator, org.springframework.cassandra.core.PreparedStatementBinder, org.springframework.cassandra.core.ResultSetExtractor)
* @see org.springframework.cassandra.core.CqlOperationsNG#execute(java.lang.String, org.springframework.cassandra.core.PreparedStatementBinder)
*/
@Override
public <T> T query(PreparedStatementCreator psc, PreparedStatementBinder psb, ResultSetExtractor<T> rse)
throws DataAccessException {
Assert.notNull(psc, "PreparedStatementCreator must not be null");
Assert.notNull(rse, "ResultSetExtractor object must not be null");
try {
if (logger.isDebugEnabled()) {
logger.debug("Preparing statement [{}] using {}", getCql(psc), psc);
}
Session session = getSession();
PreparedStatement ps = psc.createPreparedStatement(session);
if (logger.isDebugEnabled()) {
logger.debug("Executing prepared statement [{}]", ps);
}
BoundStatement boundStatement = psb != null ? psb.bindValues(ps) : ps.bind();
applyStatementSettings(boundStatement);
return rse.extractData(session.execute(boundStatement));
} catch (DriverException e) {
throw translateException("Query", getCql(psc), e);
}
public boolean execute(String cql, PreparedStatementBinder preparedStatementBinder) throws DataAccessException {
return query(new SimplePreparedStatementCreator(cql), preparedStatementBinder, ResultSet::wasApplied);
}
/*
@@ -524,7 +389,40 @@ public class CqlTemplate extends CassandraAccessor implements CqlOperations {
*/
@Override
public <T> T execute(String cql, PreparedStatementCallback<T> action) throws DataAccessException {
return execute(new SimplePreparedStatementCreator(cql), action);
return execute(newPreparedStatementCreator(cql), action);
}
/*
* (non-Javadoc)
* @see org.springframework.cassandra.core.CqlOperationsNG#execute(org.springframework.cassandra.core.PreparedStatementCreator)
*/
@Override
public boolean execute(PreparedStatementCreator preparedStatementCreator) throws DataAccessException {
return query(preparedStatementCreator, ResultSet::wasApplied);
}
/*
* (non-Javadoc)
* @see org.springframework.cassandra.core.CqlOperationsNG#execute(org.springframework.cassandra.core.PreparedStatementCreator, org.springframework.cassandra.core.PreparedStatementCallback)
*/
@Override
public <T> T execute(PreparedStatementCreator preparedStatementCreator, PreparedStatementCallback<T> action)
throws DataAccessException {
Assert.notNull(preparedStatementCreator, "PreparedStatementCreator must not be null");
Assert.notNull(action, "PreparedStatementCallback object must not be null");
try {
if (logger.isDebugEnabled()) {
logger.debug("Preparing statement [{}] using {}", toCql(preparedStatementCreator), preparedStatementCreator);
}
return action.doInPreparedStatement(applyStatementSettings(
preparedStatementCreator.createPreparedStatement(getSession())));
} catch (DriverException e) {
throw translateException("PreparedStatementCallback", toCql(preparedStatementCreator), e);
}
}
/*
@@ -532,26 +430,8 @@ public class CqlTemplate extends CassandraAccessor implements CqlOperations {
* @see org.springframework.cassandra.core.CqlOperationsNG#query(org.springframework.cassandra.core.PreparedStatementCreator, org.springframework.cassandra.core.ResultSetExtractor)
*/
@Override
public <T> T query(PreparedStatementCreator psc, ResultSetExtractor<T> rse) throws DataAccessException {
return query(psc, null, rse);
}
/*
* (non-Javadoc)
* @see org.springframework.cassandra.core.CqlOperationsNG#query(java.lang.String, org.springframework.cassandra.core.PreparedStatementBinder, org.springframework.cassandra.core.ResultSetExtractor)
*/
@Override
public <T> T query(String cql, PreparedStatementBinder psb, ResultSetExtractor<T> rse) throws DataAccessException {
return query(new SimplePreparedStatementCreator(cql), psb, rse);
}
/*
* (non-Javadoc)
* @see org.springframework.cassandra.core.CqlOperationsNG#query(java.lang.String, org.springframework.cassandra.core.ResultSetExtractor, java.lang.Object[])
*/
@Override
public <T> T query(String cql, ResultSetExtractor<T> rse, Object... args) throws DataAccessException {
return query(cql, newArgPreparedStatementBinder(args), rse);
public <T> T query(PreparedStatementCreator preparedStatementCreator, ResultSetExtractor<T> resultSetExtractor) throws DataAccessException {
return query(preparedStatementCreator, null, resultSetExtractor);
}
/*
@@ -559,36 +439,8 @@ public class CqlTemplate extends CassandraAccessor implements CqlOperations {
* @see org.springframework.cassandra.core.CqlOperationsNG#query(org.springframework.cassandra.core.PreparedStatementCreator, org.springframework.cassandra.core.RowCallbackHandler)
*/
@Override
public void query(PreparedStatementCreator psc, RowCallbackHandler rch) throws DataAccessException {
query(psc, new RowCallbackHandlerResultSetExtractor(rch));
}
/*
* (non-Javadoc)
* @see org.springframework.cassandra.core.CqlOperationsNG#query(java.lang.String, org.springframework.cassandra.core.PreparedStatementBinder, org.springframework.cassandra.core.RowCallbackHandler)
*/
@Override
public void query(String cql, PreparedStatementBinder psb, RowCallbackHandler rch) throws DataAccessException {
query(cql, psb, new RowCallbackHandlerResultSetExtractor(rch));
}
/*
* (non-Javadoc)
* @see org.springframework.cassandra.core.CqlOperationsNG#query(org.springframework.cassandra.core.PreparedStatementCreator, org.springframework.cassandra.core.PreparedStatementBinder, org.springframework.cassandra.core.RowCallbackHandler)
*/
@Override
public void query(PreparedStatementCreator psc, PreparedStatementBinder psb, RowCallbackHandler rch)
throws DataAccessException {
query(psc, psb, new RowCallbackHandlerResultSetExtractor(rch));
}
/*
* (non-Javadoc)
* @see org.springframework.cassandra.core.CqlOperationsNG#query(java.lang.String, org.springframework.cassandra.core.RowCallbackHandler, java.lang.Object[])
*/
@Override
public void query(String cql, RowCallbackHandler rch, Object... args) throws DataAccessException {
query(cql, newArgPreparedStatementBinder(args), new RowCallbackHandlerResultSetExtractor(rch));
public void query(PreparedStatementCreator preparedStatementCreator, RowCallbackHandler rowCallbackHandler) throws DataAccessException {
query(preparedStatementCreator, null, newResultSetExtractor(rowCallbackHandler));
}
/*
@@ -596,17 +448,55 @@ public class CqlTemplate extends CassandraAccessor implements CqlOperations {
* @see org.springframework.cassandra.core.CqlOperationsNG#query(org.springframework.cassandra.core.PreparedStatementCreator, org.springframework.cassandra.core.RowMapper)
*/
@Override
public <T> List<T> query(PreparedStatementCreator psc, RowMapper<T> rowMapper) throws DataAccessException {
return query(psc, new RowMapperResultSetExtractor<>(rowMapper));
public <T> List<T> query(PreparedStatementCreator preparedStatementCreator, RowMapper<T> rowMapper) throws DataAccessException {
return query(preparedStatementCreator, null, newResultSetExtractor(rowMapper));
}
/*
* (non-Javadoc)
* @see org.springframework.cassandra.core.CqlOperationsNG#query(java.lang.String, org.springframework.cassandra.core.PreparedStatementBinder, org.springframework.cassandra.core.RowMapper)
* @see org.springframework.cassandra.core.CqlOperationsNG#query(org.springframework.cassandra.core.PreparedStatementCreator, org.springframework.cassandra.core.PreparedStatementBinder, org.springframework.cassandra.core.ResultSetExtractor)
*/
@Override
public <T> List<T> query(String cql, PreparedStatementBinder psb, RowMapper<T> rowMapper) throws DataAccessException {
return query(cql, psb, new RowMapperResultSetExtractor<>(rowMapper));
public <T> T query(PreparedStatementCreator preparedStatementCreator, PreparedStatementBinder preparedStatementBinder,
ResultSetExtractor<T> resultSetExtractor) throws DataAccessException {
Assert.notNull(preparedStatementCreator, "PreparedStatementCreator must not be null");
Assert.notNull(resultSetExtractor, "ResultSetExtractor object must not be null");
try {
if (logger.isDebugEnabled()) {
logger.debug("Preparing statement [{}] using {}", toCql(preparedStatementCreator), preparedStatementCreator);
}
Session session = getSession();
PreparedStatement preparedStatement = preparedStatementCreator.createPreparedStatement(session);
if (logger.isDebugEnabled()) {
logger.debug("Executing prepared statement [{}]", preparedStatement);
}
BoundStatement boundStatement = applyStatementSettings(preparedStatementBinder != null
? preparedStatementBinder.bindValues(preparedStatement) : preparedStatement.bind());
ResultSet results = session.execute(boundStatement);
return resultSetExtractor.extractData(results);
} catch (DriverException e) {
throw translateException("Query", toCql(preparedStatementCreator), e);
}
}
/*
* (non-Javadoc)
* @see org.springframework.cassandra.core.CqlOperationsNG#query(org.springframework.cassandra.core.PreparedStatementCreator, org.springframework.cassandra.core.PreparedStatementBinder, org.springframework.cassandra.core.RowCallbackHandler)
*/
@Override
public void query(PreparedStatementCreator preparedStatementCreator, PreparedStatementBinder preparedStatementBinder,
RowCallbackHandler rowCallbackHandler) throws DataAccessException {
query(preparedStatementCreator, preparedStatementBinder, newResultSetExtractor(rowCallbackHandler));
}
/*
@@ -614,9 +504,28 @@ public class CqlTemplate extends CassandraAccessor implements CqlOperations {
* @see org.springframework.cassandra.core.CqlOperationsNG#query(org.springframework.cassandra.core.PreparedStatementCreator, org.springframework.cassandra.core.PreparedStatementBinder, org.springframework.cassandra.core.RowMapper)
*/
@Override
public <T> List<T> query(PreparedStatementCreator psc, PreparedStatementBinder psb, RowMapper<T> rowMapper)
throws DataAccessException {
return query(psc, psb, new RowMapperResultSetExtractor<>(rowMapper));
public <T> List<T> query(PreparedStatementCreator preparedStatementCreator, PreparedStatementBinder preparedStatementBinder,
RowMapper<T> rowMapper) throws DataAccessException {
return query(preparedStatementCreator, preparedStatementBinder, newResultSetExtractor(rowMapper));
}
/*
* (non-Javadoc)
* @see org.springframework.cassandra.core.CqlOperationsNG#query(java.lang.String, org.springframework.cassandra.core.ResultSetExtractor, java.lang.Object[])
*/
@Override
public <T> T query(String cql, ResultSetExtractor<T> resultSetExtractor, Object... args) throws DataAccessException {
return query(newPreparedStatementCreator(cql), newPreparedStatementBinder(args), resultSetExtractor);
}
/*
* (non-Javadoc)
* @see org.springframework.cassandra.core.CqlOperationsNG#query(java.lang.String, org.springframework.cassandra.core.RowCallbackHandler, java.lang.Object[])
*/
@Override
public void query(String cql, RowCallbackHandler rowCallbackHandler, Object... args) throws DataAccessException {
query(newPreparedStatementCreator(cql), newPreparedStatementBinder(args), newResultSetExtractor(rowCallbackHandler));
}
/*
@@ -625,45 +534,40 @@ public class CqlTemplate extends CassandraAccessor implements CqlOperations {
*/
@Override
public <T> List<T> query(String cql, RowMapper<T> rowMapper, Object... args) throws DataAccessException {
return query(cql, newArgPreparedStatementBinder(args), new RowMapperResultSetExtractor<>(rowMapper));
return query(newPreparedStatementCreator(cql), newPreparedStatementBinder(args), newResultSetExtractor(rowMapper));
}
/*
* (non-Javadoc)
* @see org.springframework.cassandra.core.CqlOperationsNG#queryForObject(java.lang.String, org.springframework.cassandra.core.RowMapper, java.lang.Object[])
* @see org.springframework.cassandra.core.CqlOperationsNG#query(java.lang.String, org.springframework.cassandra.core.PreparedStatementBinder, org.springframework.cassandra.core.ResultSetExtractor)
*/
@Override
public <T> T queryForObject(String cql, RowMapper<T> rowMapper, Object... args) throws DataAccessException {
public <T> T query(String cql, PreparedStatementBinder preparedStatementBinder, ResultSetExtractor<T> resultSetExtractor)
throws DataAccessException {
List<T> results = query(cql, newArgPreparedStatementBinder(args), new RowMapperResultSetExtractor<>(rowMapper, 1));
return DataAccessUtils.requiredSingleResult(results);
return query(newPreparedStatementCreator(cql), preparedStatementBinder, resultSetExtractor);
}
/*
* (non-Javadoc)
* @see org.springframework.cassandra.core.CqlOperationsNG#queryForObject(java.lang.String, java.lang.Class, java.lang.Object[])
* @see org.springframework.cassandra.core.CqlOperationsNG#query(java.lang.String, org.springframework.cassandra.core.PreparedStatementBinder, org.springframework.cassandra.core.RowCallbackHandler)
*/
@Override
public <T> T queryForObject(String cql, Class<T> requiredType, Object... args) throws DataAccessException {
return queryForObject(cql, getSingleColumnRowMapper(requiredType), args);
public void query(String cql, PreparedStatementBinder preparedStatementBinder, RowCallbackHandler rowCallbackHandler)
throws DataAccessException {
query(newPreparedStatementCreator(cql), preparedStatementBinder, newResultSetExtractor(rowCallbackHandler));
}
/*
* (non-Javadoc)
* @see org.springframework.cassandra.core.CqlOperationsNG#queryForMap(java.lang.String, java.lang.Object[])
* @see org.springframework.cassandra.core.CqlOperationsNG#query(java.lang.String, org.springframework.cassandra.core.PreparedStatementBinder, org.springframework.cassandra.core.RowMapper)
*/
@Override
public Map<String, Object> queryForMap(String cql, Object... args) throws DataAccessException {
return queryForObject(cql, getColumnMapRowMapper(), args);
}
public <T> List<T> query(String cql, PreparedStatementBinder preparedStatementBinder, RowMapper<T> rowMapper)
throws DataAccessException {
/*
* (non-Javadoc)
* @see org.springframework.cassandra.core.CqlOperationsNG#queryForList(java.lang.String, java.lang.Class, java.lang.Object[])
*/
@Override
public <T> List<T> queryForList(String cql, Class<T> elementType, Object... args) throws DataAccessException {
return query(cql, getSingleColumnRowMapper(elementType), args);
return query(newPreparedStatementCreator(cql), preparedStatementBinder, newResultSetExtractor(rowMapper));
}
/*
@@ -672,7 +576,46 @@ public class CqlTemplate extends CassandraAccessor implements CqlOperations {
*/
@Override
public List<Map<String, Object>> queryForList(String cql, Object... args) throws DataAccessException {
return query(cql, getColumnMapRowMapper(), args);
return query(newPreparedStatementCreator(cql), newPreparedStatementBinder(args),
newResultSetExtractor(newColumnMapRowMapper()));
}
/*
* (non-Javadoc)
* @see org.springframework.cassandra.core.CqlOperationsNG#queryForList(java.lang.String, java.lang.Class, java.lang.Object[])
*/
@Override
public <T> List<T> queryForList(String cql, Class<T> elementType, Object... args) throws DataAccessException {
return query(newPreparedStatementCreator(cql), newPreparedStatementBinder(args),
newResultSetExtractor(newSingleColumnRowMapper(elementType)));
}
/*
* (non-Javadoc)
* @see org.springframework.cassandra.core.CqlOperationsNG#queryForMap(java.lang.String, java.lang.Object[])
*/
@Override
public Map<String, Object> queryForMap(String cql, Object... args) throws DataAccessException {
return queryForObject(cql, newColumnMapRowMapper(), args);
}
/*
* (non-Javadoc)
* @see org.springframework.cassandra.core.CqlOperationsNG#queryForObject(java.lang.String, java.lang.Class, java.lang.Object[])
*/
@Override
public <T> T queryForObject(String cql, Class<T> requiredType, Object... args) throws DataAccessException {
return queryForObject(cql, newSingleColumnRowMapper(requiredType), args);
}
/*
* (non-Javadoc)
* @see org.springframework.cassandra.core.CqlOperationsNG#queryForObject(java.lang.String, org.springframework.cassandra.core.RowMapper, java.lang.Object[])
*/
@Override
public <T> T queryForObject(String cql, RowMapper<T> rowMapper, Object... args) throws DataAccessException {
return DataAccessUtils.requiredSingleResult(query(newPreparedStatementCreator(cql),
newPreparedStatementBinder(args), newResultSetExtractor(rowMapper, 1)));
}
/*
@@ -681,7 +624,7 @@ public class CqlTemplate extends CassandraAccessor implements CqlOperations {
*/
@Override
public ResultSet queryForResultSet(String cql, Object... args) throws DataAccessException {
return query(cql, rs -> rs, args);
return query(newPreparedStatementCreator(cql), newPreparedStatementBinder(args), rs -> rs);
}
/*
@@ -689,35 +632,8 @@ public class CqlTemplate extends CassandraAccessor implements CqlOperations {
* @see org.springframework.cassandra.core.CqlOperationsNG#queryForRows(java.lang.String, java.lang.Object[])
*/
@Override
public Iterator<Row> queryForRows(String cql, Object... args) throws DataAccessException {
return queryForResultSet(cql, args).iterator();
}
/*
* (non-Javadoc)
* @see org.springframework.cassandra.core.CqlOperationsNG#execute(org.springframework.cassandra.core.PreparedStatementCreator)
*/
@Override
public boolean execute(PreparedStatementCreator psc) throws DataAccessException {
return query(psc, ResultSet::wasApplied);
}
/*
* (non-Javadoc)
* @see org.springframework.cassandra.core.CqlOperationsNG#execute(java.lang.String, org.springframework.cassandra.core.PreparedStatementBinder)
*/
@Override
public boolean execute(String cql, PreparedStatementBinder psb) throws DataAccessException {
return query(new SimplePreparedStatementCreator(cql), psb, ResultSet::wasApplied);
}
/*
* (non-Javadoc)
* @see org.springframework.cassandra.core.CqlOperationsNG#execute(java.lang.String, java.lang.Object[])
*/
@Override
public boolean execute(String cql, Object... args) throws DataAccessException {
return execute(cql, newArgPreparedStatementBinder(args));
public Iterable<Row> queryForRows(String cql, Object... args) throws DataAccessException {
return () -> queryForResultSet(cql, args).iterator();
}
/*
@@ -741,10 +657,16 @@ public class CqlTemplate extends CassandraAccessor implements CqlOperations {
return hostMapper.mapHosts(getHosts());
}
/* (non-Javadoc) */
private Set<Host> getHosts() {
return getSession().getCluster().getMetadata().getAllHosts();
}
/* (non-Javadoc) */
protected PreparedStatementCreator newPreparedStatementCreator(String cql) {
return new SimplePreparedStatementCreator(cql);
}
// -------------------------------------------------------------------------
// Implementation hooks and helper methods
// -------------------------------------------------------------------------
@@ -754,7 +676,7 @@ public class CqlTemplate extends CassandraAccessor implements CqlOperations {
*
* @param task readable text describing the task being attempted
* @param cql CQL query or update that caused the problem (may be {@code null})
* @param ex the offending {@code RuntimeException}.
* @param driverException the offending {@code RuntimeException}.
* @return the exception translation {@link Function}
* @see CqlProvider
*/
@@ -763,102 +685,6 @@ public class CqlTemplate extends CassandraAccessor implements CqlOperations {
return translate(task, cql, driverException);
}
/**
* Create a new RowMapper for reading columns as key-value pairs.
*
* @return the RowMapper to use
* @see ColumnMapRowMapper
*/
protected RowMapper<Map<String, Object>> getColumnMapRowMapper() {
return new ColumnMapRowMapper();
}
/**
* Create a new RowMapper for reading result objects from a single column.
*
* @param requiredType the type that each result object is expected to match
* @return the RowMapper to use
* @see SingleColumnRowMapper
*/
protected <T> RowMapper<T> getSingleColumnRowMapper(Class<T> requiredType) {
return SingleColumnRowMapper.newInstance(requiredType);
}
/**
* Prepare the given CQL Statement (or {@link com.datastax.driver.core.PreparedStatement}), applying statement
* settings such as fetch size, retry policy, and consistency level.
*
* @param stmt the CQL Statement to prepare
* @see #setFetchSize(int)
* @see #setRetryPolicy(RetryPolicy)
* @see #setConsistencyLevel(ConsistencyLevel)
*/
protected void applyStatementSettings(Statement stmt) {
int fetchSize = getFetchSize();
if (fetchSize != -1 && stmt.getFetchSize() == DEFAULTS.getFetchSize()) {
stmt.setFetchSize(fetchSize);
}
RetryPolicy retryPolicy = getRetryPolicy();
if (retryPolicy != null && stmt.getRetryPolicy() == DEFAULTS.getRetryPolicy()) {
stmt.setRetryPolicy(retryPolicy);
}
ConsistencyLevel consistencyLevel = getConsistencyLevel();
if (consistencyLevel != null && stmt.getConsistencyLevel() == DEFAULTS.getConsistencyLevel()) {
stmt.setConsistencyLevel(consistencyLevel);
}
}
/**
* Prepare the given CQL Statement (or {@link com.datastax.driver.core.PreparedStatement}), applying statement
* settings such as retry policy and consistency level.
*
* @param stmt the CQL Statement to prepare
* @see #setRetryPolicy(RetryPolicy)
* @see #setConsistencyLevel(ConsistencyLevel)
*/
protected void applyStatementSettings(PreparedStatement stmt) {
RetryPolicy retryPolicy = getRetryPolicy();
if (retryPolicy != null) {
stmt.setRetryPolicy(retryPolicy);
}
ConsistencyLevel consistencyLevel = getConsistencyLevel();
if (consistencyLevel != null) {
stmt.setConsistencyLevel(consistencyLevel);
}
}
/**
* Create a new arg-based PreparedStatementSetter using the args passed in. By default, we'll create an
* {@link ArgumentPreparedStatementBinder}. This method allows for the creation to be overridden by subclasses.
*
* @param args object array with arguments
* @return the new {@link PreparedStatementBinder} to use
*/
protected PreparedStatementBinder newArgPreparedStatementBinder(Object[] args) {
return new ArgumentPreparedStatementBinder(args);
}
/**
* Determine CQL from potential provider object.
*
* @param cqlProvider object that's potentially a {@link CqlProvider}
* @return the CQL string, or {@code null}
* @see CqlProvider
*/
private static String getCql(Object cqlProvider) {
if (cqlProvider instanceof CqlProvider) {
return ((CqlProvider) cqlProvider).getCql();
} else {
return null;
}
}
private class SimplePreparedStatementCreator implements PreparedStatementCreator, CqlProvider {
private final String cql;
@@ -880,23 +706,4 @@ public class CqlTemplate extends CassandraAccessor implements CqlOperations {
return cql;
}
}
/**
* Adapter to enable use of a {@link RowCallbackHandler} inside a {@link ResultSetExtractor}.
*/
private static class RowCallbackHandlerResultSetExtractor implements ResultSetExtractor<Object> {
private final RowCallbackHandler rch;
public RowCallbackHandlerResultSetExtractor(RowCallbackHandler rch) {
this.rch = rch;
}
@Override
public Object extractData(ResultSet rs) {
StreamSupport.stream(rs.spliterator(), false).forEach(rch::processRow);
return null;
}
}
}

26
spring-cql/src/main/java/org/springframework/cassandra/core/ExceptionTranslatingListenableFutureAdapter.java
View File

@@ -31,7 +31,7 @@ import org.springframework.util.concurrent.SuccessCallback;
/**
* Adapter class to {@link ListenableFuture} {@link ExecutionException} by applying a
* {@link PersistenceExceptionTranslator}.
*
*
* @author Mark Paluch
* @since 2.0
*/
@@ -57,10 +57,10 @@ class ExceptionTranslatingListenableFutureAdapter<T> implements ListenableFuture
this.future = adaptListenableFuture(adaptee, persistenceExceptionTranslator);
}
private static <T> ListenableFuture adaptListenableFuture(ListenableFuture<T> listenableFuture,
private static <T> ListenableFuture<T> adaptListenableFuture(ListenableFuture<T> listenableFuture,
PersistenceExceptionTranslator exceptionTranslator) {
SettableListenableFuture<T> settableFuture = new SettableListenableFuture<T>();
SettableListenableFuture<T> settableFuture = new SettableListenableFuture<>();
listenableFuture.addCallback(new ListenableFutureCallback<T>() {
@@ -71,11 +71,10 @@ class ExceptionTranslatingListenableFutureAdapter<T> implements ListenableFuture
@Override
public void onFailure(Throwable ex) {
if (ex instanceof RuntimeException) {
DataAccessException dataAccessException =
exceptionTranslator.translateExceptionIfPossible((RuntimeException) ex);
DataAccessException dataAccessException = exceptionTranslator
.translateExceptionIfPossible((RuntimeException) ex);
if (dataAccessException != null) {
settableFuture.setException(dataAccessException);
return;
@@ -87,10 +86,9 @@ class ExceptionTranslatingListenableFutureAdapter<T> implements ListenableFuture
});
return settableFuture;
}
/*
/*
* (non-Javadoc)
* @see org.springframework.util.concurrent.ListenableFuture#addCallback(org.springframework.util.concurrent.ListenableFutureCallback)
*/
@@ -99,7 +97,7 @@ class ExceptionTranslatingListenableFutureAdapter<T> implements ListenableFuture
future.addCallback(callback);
}
/*
/*
* (non-Javadoc)
* @see org.springframework.util.concurrent.ListenableFuture#addCallback(org.springframework.util.concurrent.SuccessCallback, org.springframework.util.concurrent.FailureCallback)
*/
@@ -108,7 +106,7 @@ class ExceptionTranslatingListenableFutureAdapter<T> implements ListenableFuture
future.addCallback(successCallback, failureCallback);
}
/*
/*
* (non-Javadoc)
* @see java.util.concurrent.Future#cancel(boolean)
*/
@@ -117,7 +115,7 @@ class ExceptionTranslatingListenableFutureAdapter<T> implements ListenableFuture
return adaptee.cancel(mayInterruptIfRunning);
}
/*
/*
* (non-Javadoc)
* @see java.util.concurrent.Future#isCancelled()
*/
@@ -126,7 +124,7 @@ class ExceptionTranslatingListenableFutureAdapter<T> implements ListenableFuture
return adaptee.isCancelled();
}
/*
/*
* (non-Javadoc)
* @see java.util.concurrent.Future#isDone()
*/
@@ -135,7 +133,7 @@ class ExceptionTranslatingListenableFutureAdapter<T> implements ListenableFuture
return future.isDone();
}
/*
/*
* (non-Javadoc)
* @see java.util.concurrent.Future#get()
*/
@@ -144,7 +142,7 @@ class ExceptionTranslatingListenableFutureAdapter<T> implements ListenableFuture
return future.get();
}
/*
/*
* (non-Javadoc)
* @see java.util.concurrent.Future#get(long, java.util.concurrent.TimeUnit)
*/

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

@@ -19,6 +19,9 @@ import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import com.google.common.util.concurrent.FutureCallback;
import com.google.common.util.concurrent.Futures;
import org.springframework.dao.DataAccessException;
import org.springframework.dao.support.PersistenceExceptionTranslator;
import org.springframework.util.Assert;
@@ -28,13 +31,10 @@ import org.springframework.util.concurrent.ListenableFutureCallback;
import org.springframework.util.concurrent.SettableListenableFuture;
import org.springframework.util.concurrent.SuccessCallback;
import com.google.common.util.concurrent.FutureCallback;
import com.google.common.util.concurrent.Futures;
/**
* Adapter class to adapt Guava's {@link com.google.common.util.concurrent.ListenableFuture} into a Spring
* {@link ListenableFuture}.
*
*
* @author Mark Paluch
* @since 2.0
*/
@@ -46,7 +46,7 @@ public class GuavaListenableFutureAdapter<T> implements ListenableFuture<T> {
/**
* Create a new {@link GuavaListenableFutureAdapter} given a Guava
* {@link com.google.common.util.concurrent.ListenableFuture} and a {@link PersistenceExceptionTranslator}.
*
*
* @param adaptee must not be {@literal null}.
* @param persistenceExceptionTranslator must not be {@literal null}.
*/
@@ -60,11 +60,11 @@ public class GuavaListenableFutureAdapter<T> implements ListenableFuture<T> {
this.future = adaptListenableFuture(adaptee, persistenceExceptionTranslator);
}
private static <T> ListenableFuture adaptListenableFuture(
private static <T> ListenableFuture<T> adaptListenableFuture(
com.google.common.util.concurrent.ListenableFuture<T> guavaFuture,
PersistenceExceptionTranslator exceptionTranslator) {
SettableListenableFuture<T> settableFuture = new SettableListenableFuture<T>();
SettableListenableFuture<T> settableFuture = new SettableListenableFuture<>();
Futures.addCallback(guavaFuture, new FutureCallback<T>() {
@Override
@@ -74,11 +74,10 @@ public class GuavaListenableFutureAdapter<T> implements ListenableFuture<T> {
@Override
public void onFailure(Throwable t) {
if (t instanceof RuntimeException) {
DataAccessException dataAccessException =
exceptionTranslator.translateExceptionIfPossible((RuntimeException) t);
DataAccessException dataAccessException = exceptionTranslator
.translateExceptionIfPossible((RuntimeException) t);
if (dataAccessException != null) {
settableFuture.setException(dataAccessException);
return;
@@ -90,10 +89,9 @@ public class GuavaListenableFutureAdapter<T> implements ListenableFuture<T> {
});
return settableFuture;
}
/*
/*
* (non-Javadoc)
* @see org.springframework.util.concurrent.ListenableFuture#addCallback(org.springframework.util.concurrent.ListenableFutureCallback)
*/
@@ -102,7 +100,7 @@ public class GuavaListenableFutureAdapter<T> implements ListenableFuture<T> {
future.addCallback(callback);
}
/*
/*
* (non-Javadoc)
* @see org.springframework.util.concurrent.ListenableFuture#addCallback(org.springframework.util.concurrent.SuccessCallback, org.springframework.util.concurrent.FailureCallback)
*/
@@ -111,7 +109,7 @@ public class GuavaListenableFutureAdapter<T> implements ListenableFuture<T> {
future.addCallback(successCallback, failureCallback);
}
/*
/*
* (non-Javadoc)
* @see java.util.concurrent.Future#cancel(boolean)
*/
@@ -120,7 +118,7 @@ public class GuavaListenableFutureAdapter<T> implements ListenableFuture<T> {
return adaptee.cancel(mayInterruptIfRunning);
}
/*
/*
* (non-Javadoc)
* @see java.util.concurrent.Future#isCancelled()
*/
@@ -129,7 +127,7 @@ public class GuavaListenableFutureAdapter<T> implements ListenableFuture<T> {
return adaptee.isCancelled();
}
/*
/*
* (non-Javadoc)
* @see java.util.concurrent.Future#isDone()
*/
@@ -138,7 +136,7 @@ public class GuavaListenableFutureAdapter<T> implements ListenableFuture<T> {
return future.isDone();
}
/*
/*
* (non-Javadoc)
* @see java.util.concurrent.Future#get()
*/
@@ -147,7 +145,7 @@ public class GuavaListenableFutureAdapter<T> implements ListenableFuture<T> {
return future.get();
}
/*
/*
* (non-Javadoc)
* @see java.util.concurrent.Future#get(long, java.util.concurrent.TimeUnit)
*/

9
spring-cql/src/main/java/org/springframework/cassandra/core/HostMapper.java
View File

@@ -1,12 +1,12 @@
/*
* 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.
@@ -35,11 +35,12 @@ public interface HostMapper<T> {
/**
* Implementations must implement this method to map each {@link Host} in the
* {@link com.datastax.driver.core.Metadata}.
*
*
* @param hosts the {@link Iterable} of {@link Host}s to map, must not be {@literal null}.
* @return the result objects for the given hosts.
* @throws DriverException if a {@link DriverException} is encountered mapping values (that is, there's no need to
* catch {@link DriverException}).
*/
Collection<T> mapHosts(Iterable<Host> hosts) throws DriverException;
}

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

@@ -1,12 +1,12 @@
/*
* 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.
@@ -15,13 +15,13 @@
*/
package org.springframework.cassandra.core;
import org.springframework.dao.DataAccessException;
import com.datastax.driver.core.PreparedStatement;
import com.datastax.driver.core.Session;
import com.datastax.driver.core.Statement;
import com.datastax.driver.core.exceptions.DriverException;
import org.springframework.dao.DataAccessException;
/**
* Generic callback interface for code that operates on a {@link PreparedStatement}. Allows to execute any number of
* operations on a single {@link PreparedStatement}, for example a single {@link Session#execute(Statement).
@@ -46,14 +46,14 @@ public interface PreparedStatementCallback<T> {
* objects. Note that there's special support for single step actions: see
* {@link CqlTemplate#queryForObject(String, Class, Object...)} etc. A thrown RuntimeException is treated as
* application exception, it gets propagated to the caller of the template.
*
* @param ps the {@link PreparedStatement}, must not be {@literal null}.
*
* @param preparedStatement the {@link PreparedStatement}, must not be {@literal null}.
* @return a result object publisher.
* @throws DriverException if thrown by a session method, to be auto-converted to a DataAccessException.
* @throws DataAccessException in case of custom exceptions.
* @see CqlTemplate#queryForObject(String, Class, Object...)
* @see CqlTemplate#queryForList(String, Object...)
*/
T doInPreparedStatement(PreparedStatement ps) throws DriverException, DataAccessException;
T doInPreparedStatement(PreparedStatement preparedStatement) throws DriverException, DataAccessException;
}

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

@@ -1,12 +1,12 @@
/*
* 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.
@@ -36,11 +36,12 @@ public interface PreparedStatementCreator {
/**
* Create a statement in this session. Allows implementations to use {@link PreparedStatement}.
*
*
* @param session {@link Session} to use to create statement
* @return a prepared statement
* @throws DriverException there is no need to catch {@link DriverException} that may be thrown in the implementation
* of this method. The {@link CqlTemplate} class will handle them.
*/
PreparedStatement createPreparedStatement(Session session) throws DriverException;
}

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

@@ -15,17 +15,17 @@
*/
package org.springframework.cassandra.core;
import org.springframework.util.Assert;
import com.datastax.driver.core.PreparedStatement;
import com.datastax.driver.core.Statement;
import com.datastax.driver.core.querybuilder.Insert;
import com.datastax.driver.core.querybuilder.QueryBuilder;
import com.datastax.driver.core.querybuilder.Update;
import org.springframework.util.Assert;
/**
* Utility class to associate {@link QueryOptions} and {@link WriteOptions} with QueryBuilder {@link Statement}s.
*
*
* @author Mark Paluch
* @since 2.0
*/
@@ -115,7 +115,6 @@ public abstract class QueryOptionsUtil {
Assert.notNull(insert, "Insert must not be null");
if (writeOptions != null) {
addQueryOptions(insert, writeOptions);
if (writeOptions.getTtl() != null) {
@@ -138,7 +137,6 @@ public abstract class QueryOptionsUtil {
Assert.notNull(update, "Update must not be null");
if (writeOptions != null) {
addQueryOptions(update, writeOptions);
if (writeOptions.getTtl() != null) {

17
spring-cql/src/main/java/org/springframework/cassandra/core/ResultSetExtractor.java
View File

@@ -1,12 +1,12 @@
/*
* 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.
@@ -15,11 +15,11 @@
*/
package org.springframework.cassandra.core;
import org.springframework.dao.DataAccessException;
import com.datastax.driver.core.ResultSet;
import com.datastax.driver.core.exceptions.DriverException;
import org.springframework.dao.DataAccessException;
/**
* Callback interface used by {@link CqlTemplate}'s query methods. Implementations of this interface perform the actual
* work of extracting results from a {@link ResultSet}, but don't need to worry about exception handling.
@@ -43,13 +43,14 @@ public interface ResultSetExtractor<T> {
/**
* Implementations must implement this method to process the entire {@link ResultSet}.
*
* @param rs {@link ResultSet} to extract data from.
*
* @param resultSet {@link ResultSet} to extract data from.
* @return an arbitrary result object, or {@code null} if none (the extractor will typically be stateful in the latter
* case).
* @throws DriverException if a {@link DriverException} is encountered getting column values or navigating (that is,
* there's no need to catch {@link DriverException})
* @throws DataAccessException in case of custom exceptions
*/
T extractData(ResultSet rs) throws DriverException, DataAccessException;
T extractData(ResultSet resultSet) throws DriverException, DataAccessException;
}

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

@@ -1,12 +1,12 @@
/*
* 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.
@@ -19,14 +19,14 @@ import java.util.Collection;
import java.util.stream.Collectors;
import java.util.stream.StreamSupport;
import org.springframework.util.Assert;
import com.datastax.driver.core.Host;
import com.datastax.driver.core.exceptions.DriverException;
import org.springframework.util.Assert;
/**
* {@link HostMapper} to to map hosts into {@link RingMember} objects.
*
*
* @author David Webb
* @author Mark Paluch
* @param <T>
@@ -43,8 +43,6 @@ public enum RingMemberHostMapper implements HostMapper<RingMember> {
Assert.notNull(hosts, "Hosts must not be null");
return StreamSupport.stream(hosts.spliterator(), false) //
.map(RingMember::new) //
.collect(Collectors.toList());
return StreamSupport.stream(hosts.spliterator(), false).map(RingMember::new).collect(Collectors.toList());
}
}

9
spring-cql/src/main/java/org/springframework/cassandra/core/RowCallbackHandler.java
View File

@@ -1,12 +1,12 @@
/*
* 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.
@@ -44,10 +44,11 @@ public interface RowCallbackHandler {
* <p>
* Exactly what the implementation chooses to do is up to it: A trivial implementation might simply count rows, while
* another implementation might build an XML document.
*
*
* @param row the {@link Row} to process (pre-initialized for the current row)
* @throws DriverException if a {@link DriverException} is encountered getting column values (that is, there's no need
* to catch {@link DriverException})
*/
void processRow(Row row) throws DriverException;
}

9
spring-cql/src/main/java/org/springframework/cassandra/core/RowMapper.java
View File

@@ -1,12 +1,12 @@
/*
* 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.
@@ -38,7 +38,7 @@ public interface RowMapper<T> {
/**
* Implementations must implement this method to map each row of data in the
* {@link com.datastax.driver.core.ResultSet}.
*
*
* @param row the {@link Row} to map, must not be {@literal null}.
* @param rowNum the number of the current row.
* @return the result object for the current row.
@@ -46,4 +46,5 @@ public interface RowMapper<T> {
* to catch {@link DriverException})
*/
T mapRow(Row row, int rowNum) throws DriverException;
}

21
spring-cql/src/main/java/org/springframework/cassandra/core/RowMapperResultSetExtractor.java
View File

@@ -18,13 +18,13 @@ package org.springframework.cassandra.core;
import java.util.ArrayList;
import java.util.List;
import org.springframework.dao.DataAccessException;
import org.springframework.util.Assert;
import com.datastax.driver.core.ResultSet;
import com.datastax.driver.core.Row;
import com.datastax.driver.core.exceptions.DriverException;
import org.springframework.dao.DataAccessException;
import org.springframework.util.Assert;
/**
* Adapter implementation of the {@link ResultSetExtractor} interface that delegates to a {@link RowMapper} which is
* supposed to create an object for each row. Each object is added to the results List of this
@@ -42,13 +42,13 @@ import com.datastax.driver.core.exceptions.DriverException;
*/
public class RowMapperResultSetExtractor<T> implements ResultSetExtractor<List<T>> {
private final RowMapper<T> rowMapper;
private final int rowsExpected;
private final RowMapper<T> rowMapper;
/**
* Create a new {@link RowMapperResultSetExtractor}.
*
*
* @param rowMapper the {@link RowMapper} which creates an object for each row, must not be {@literal null}.
*/
public RowMapperResultSetExtractor(RowMapper<T> rowMapper) {
@@ -57,7 +57,7 @@ public class RowMapperResultSetExtractor<T> implements ResultSetExtractor<List<T
/**
* Create a new {@link RowMapperResultSetExtractor}.
*
*
* @param rowMapper the {@link RowMapper} which creates an object for each row, must not be {@literal null}.
* @param rowsExpected the number of expected rows (just used for optimized collection handling).
*/
@@ -75,11 +75,12 @@ public class RowMapperResultSetExtractor<T> implements ResultSetExtractor<List<T
@Override
public List<T> extractData(ResultSet resultSet) throws DriverException, DataAccessException {
List<T> results = (this.rowsExpected > 0 ? new ArrayList<>(this.rowsExpected) : new ArrayList<T>());
List<T> results = (this.rowsExpected > 0 ? new ArrayList<>(this.rowsExpected) : new ArrayList<>());
int rowNumber = 0;
int rowNum = 0;
for (Row row : resultSet) {
results.add(this.rowMapper.mapRow(row, rowNum++));
results.add(this.rowMapper.mapRow(row, rowNumber++));
}
return results;

273
spring-cql/src/main/java/org/springframework/cassandra/support/CassandraAccessor.java
View File

@@ -15,15 +15,33 @@
*/
package org.springframework.cassandra.support;
import java.util.Map;
import java.util.stream.StreamSupport;
import com.datastax.driver.core.ConsistencyLevel;
import com.datastax.driver.core.PreparedStatement;
import com.datastax.driver.core.ResultSet;
import com.datastax.driver.core.Session;
import com.datastax.driver.core.Statement;
import com.datastax.driver.core.exceptions.DriverException;
import com.datastax.driver.core.policies.RetryPolicy;
import com.datastax.driver.core.querybuilder.QueryBuilder;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.cassandra.core.ArgumentPreparedStatementBinder;
import org.springframework.cassandra.core.ColumnMapRowMapper;
import org.springframework.cassandra.core.CqlProvider;
import org.springframework.cassandra.core.PreparedStatementBinder;
import org.springframework.cassandra.core.ResultSetExtractor;
import org.springframework.cassandra.core.RowCallbackHandler;
import org.springframework.cassandra.core.RowMapper;
import org.springframework.cassandra.core.RowMapperResultSetExtractor;
import org.springframework.cassandra.core.SingleColumnRowMapper;
import org.springframework.dao.DataAccessException;
import org.springframework.util.Assert;
import com.datastax.driver.core.Session;
/**
* {@link CassandraAccessor} provides access to a Cassandra {@link Session} and the {@link CassandraExceptionTranslator}
* .
@@ -39,10 +57,33 @@ import com.datastax.driver.core.Session;
*/
public class CassandraAccessor implements InitializingBean {
/**
* Placeholder for default values.
*/
private final static Statement DEFAULTS = QueryBuilder.select().from("DEFAULT");
/**
* If this variable is set to a non-negative value, it will be used for setting the {@code fetchSize} property on
* statements used for query processing.
*/
private int fetchSize = -1;
protected CassandraExceptionTranslator exceptionTranslator = new CassandraExceptionTranslator();
/**
* If this variable is set to a value, it will be used for setting the {@code consistencyLevel} property on statements
* used for query processing.
*/
private com.datastax.driver.core.ConsistencyLevel consistencyLevel;
protected final Logger logger = LoggerFactory.getLogger(getClass());
/**
* If this variable is set to a value, it will be used for setting the {@code retryPolicy} property on statements used
* for query processing.
*/
private com.datastax.driver.core.policies.RetryPolicy retryPolicy;
private Session session;
/**
@@ -54,12 +95,33 @@ public class CassandraAccessor implements InitializingBean {
}
/* (non-Javadoc) */
@SuppressWarnings("unused")
protected void logDebug(String logMessage, Object... array) {
if (logger.isDebugEnabled()) {
logger.debug(logMessage, array);
}
}
/**
* Set the consistency level for this template. Consistency level defines the number of nodes
* involved into query processing. Relaxed consistency level settings use fewer nodes but eventual consistency is more
* likely to occur while a higher consistency level involves more nodes to obtain results with a higher consistency
* guarantee.
*
* @see Statement#setConsistencyLevel(ConsistencyLevel)
* @see RetryPolicy
*/
public void setConsistencyLevel(ConsistencyLevel consistencyLevel) {
this.consistencyLevel = consistencyLevel;
}
/**
* @return the {@link ConsistencyLevel} specified for this template.
*/
public ConsistencyLevel getConsistencyLevel() {
return this.consistencyLevel;
}
/**
* Sets the exception translator used by this template to translate Cassandra specific Exceptions into Spring DAO's
* Exception Hierarchy.
@@ -84,6 +146,43 @@ public class CassandraAccessor implements InitializingBean {
return this.exceptionTranslator;
}
/**
* Set the fetch size for this template. This is important for processing large result sets: Setting this
* higher than the default value will increase processing speed at the cost of memory consumption; setting this lower
* can avoid transferring row data that will never be read by the application. Default is -1, indicating to use the
* CQL driver's default configuration (i.e. to not pass a specific fetch size setting on to the driver).
*
* @see Statement#setFetchSize(int)
*/
public void setFetchSize(int fetchSize) {
this.fetchSize = fetchSize;
}
/**
* @return the fetch size specified for this template.
*/
public int getFetchSize() {
return this.fetchSize;
}
/**
* Set the retry policy for this template. This is important for defining behavior when a request
* fails.
*
* @see Statement#setRetryPolicy(RetryPolicy)
* @see RetryPolicy
*/
public void setRetryPolicy(RetryPolicy retryPolicy) {
this.retryPolicy = retryPolicy;
}
/**
* @return the {@link RetryPolicy} specified for this template.
*/
public RetryPolicy getRetryPolicy() {
return this.retryPolicy;
}
/**
* Sets the Cassandra {@link Session} used by this template to perform Cassandra data access operations.
*
@@ -105,8 +204,151 @@ public class CassandraAccessor implements InitializingBean {
Assert.state(this.session != null, "Session was not properly initialized");
return this.session;
}
/**
/**
* Prepare the given CQL Statement (or {@link com.datastax.driver.core.PreparedStatement}), applying statement
* settings such as retry policy and consistency level.
*
* @param statement the CQL Statement to prepare
* @see #setRetryPolicy(RetryPolicy)
* @see #setConsistencyLevel(ConsistencyLevel)
*/
protected <T extends PreparedStatement> T applyStatementSettings(T statement) {
ConsistencyLevel consistencyLevel = getConsistencyLevel();
if (consistencyLevel != null) {
statement.setConsistencyLevel(consistencyLevel);
}
RetryPolicy retryPolicy = getRetryPolicy();
if (retryPolicy != null) {
statement.setRetryPolicy(retryPolicy);
}
return statement;
}
/**
* Prepare the given CQL Statement (or {@link com.datastax.driver.core.PreparedStatement}), applying statement
* settings such as fetch size, retry policy, and consistency level.
*
* @param statement the CQL Statement to prepare
* @see #setFetchSize(int)
* @see #setRetryPolicy(RetryPolicy)
* @see #setConsistencyLevel(ConsistencyLevel)
*/
protected <T extends Statement> T applyStatementSettings(T statement) {
ConsistencyLevel consistencyLevel = getConsistencyLevel();
if (consistencyLevel != null && statement.getConsistencyLevel() == DEFAULTS.getConsistencyLevel()) {
statement.setConsistencyLevel(consistencyLevel);
}
int fetchSize = getFetchSize();
if (fetchSize != -1 && statement.getFetchSize() == DEFAULTS.getFetchSize()) {
statement.setFetchSize(fetchSize);
}
RetryPolicy retryPolicy = getRetryPolicy();
if (retryPolicy != null && statement.getRetryPolicy() == DEFAULTS.getRetryPolicy()) {
statement.setRetryPolicy(retryPolicy);
}
return statement;
}
/**
* Create a new arg-based PreparedStatementSetter using the args passed in. By default, we'll create an
* {@link ArgumentPreparedStatementBinder}. This method allows for the creation to be overridden by subclasses.
*
* @param args object array with arguments
* @return the new {@link PreparedStatementBinder} to use
*/
protected PreparedStatementBinder newPreparedStatementBinder(Object[] args) {
return new ArgumentPreparedStatementBinder(args);
}
/**
* Constructs a new instance of the {@link ResultSetExtractor} initialized with and adapting
* the given {@link RowCallbackHandler}.
*
* @param rowCallbackHandler {@link RowCallbackHandler} to adapt as a {@link ResultSetExtractor}.
* @return a {@link ResultSetExtractor} implementation adapting an instance of the {@link RowCallbackHandler}.
* @see org.springframework.cassandra.core.AsyncCqlTemplate.RowCallbackHandlerResultSetExtractor
* @see org.springframework.cassandra.core.ResultSetExtractor
* @see org.springframework.cassandra.core.RowCallbackHandler
*/
protected RowCallbackHandlerResultSetExtractor newResultSetExtractor(RowCallbackHandler rowCallbackHandler) {
return new RowCallbackHandlerResultSetExtractor(rowCallbackHandler);
}
/**
* Constructs a new instance of the {@link ResultSetExtractor} initialized with and adapting
* the given {@link RowMapper}.
*
* @param rowMapper {@link RowMapper} to adapt as a {@link ResultSetExtractor}.
* @return a {@link ResultSetExtractor} implementation adapting an instance of the {@link RowMapper}.
* @see org.springframework.cassandra.core.ResultSetExtractor
* @see org.springframework.cassandra.core.RowMapper
* @see org.springframework.cassandra.core.RowMapperResultSetExtractor
*/
protected <T> RowMapperResultSetExtractor<T> newResultSetExtractor(RowMapper<T> rowMapper) {
return new RowMapperResultSetExtractor<>(rowMapper);
}
/**
* Constructs a new instance of the {@link ResultSetExtractor} initialized with and adapting
* the given {@link RowMapper}.
*
* @param rowMapper {@link RowMapper} to adapt as a {@link ResultSetExtractor}.
* @param rowsExpected number of expected rows in the {@link ResultSet}.
* @return a {@link ResultSetExtractor} implementation adapting an instance of the {@link RowMapper}.
* @see org.springframework.cassandra.core.ResultSetExtractor
* @see org.springframework.cassandra.core.RowMapper
* @see org.springframework.cassandra.core.RowMapperResultSetExtractor
*/
protected <T> RowMapperResultSetExtractor<T> newResultSetExtractor(RowMapper<T> rowMapper, int rowsExpected) {
return new RowMapperResultSetExtractor<>(rowMapper, rowsExpected);
}
/**
* Create a new RowMapper for reading columns as key-value pairs.
*
* @return the RowMapper to use
* @see ColumnMapRowMapper
*/
protected RowMapper<Map<String, Object>> newColumnMapRowMapper() {
return new ColumnMapRowMapper();
}
/**
* Create a new RowMapper for reading result objects from a single column.
*
* @param requiredType the type that each result object is expected to match
* @return the RowMapper to use
* @see SingleColumnRowMapper
*/
protected <T> RowMapper<T> newSingleColumnRowMapper(Class<T> requiredType) {
return SingleColumnRowMapper.newInstance(requiredType);
}
/**
* Determine CQL from potential provider object.
*
* @param cqlProvider object that's potentially a {@link CqlProvider}
* @return the CQL string, or {@code null}
* @see CqlProvider
*/
protected static String toCql(Object cqlProvider) {
return (cqlProvider instanceof CqlProvider ? ((CqlProvider) cqlProvider).getCql() : null);
}
/**
* Translate the given {@link DriverException} into a generic {@link DataAccessException}.
* <p>
* The returned {@link DataAccessException} is supposed to contain the original {@code DriverException} as root cause.
@@ -151,4 +393,27 @@ public class CassandraAccessor implements InitializingBean {
return getExceptionTranslator().translate(task, cql, ex);
}
/**
* Adapter to enable use of a {@link RowCallbackHandler} inside a {@link ResultSetExtractor}.
*/
protected static class RowCallbackHandlerResultSetExtractor implements ResultSetExtractor<Object> {
private final RowCallbackHandler rowCallbackHandler;
protected RowCallbackHandlerResultSetExtractor(RowCallbackHandler rowCallbackHandler) {
this.rowCallbackHandler = rowCallbackHandler;
}
/**
* @inheritDoc
*/
@Override
public Object extractData(ResultSet resultSet) {
StreamSupport.stream(resultSet.spliterator(), false).forEach(rowCallbackHandler::processRow);
return null;
}
}
}

71
spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/AsyncCassandraOperations.java
View File

@@ -32,11 +32,28 @@ import com.datastax.driver.core.Statement;
* Not often used directly, but a useful option to enhance testability, as it can easily be mocked or stubbed.
*
* @author Mark Paluch
* @author John Blum
* @since 2.0
* @see AsyncCassandraTemplate
* @see CassandraOperations
*/
public interface AsyncCassandraOperations {
/**
* Expose the underlying {@link AsyncCqlOperationsOperations} to allow asynchronous CQL operations.
*
* @return the underlying {@link AsyncCqlOperations}.
* @see AsyncCqlOperations
*/
AsyncCqlOperations getAsyncCqlOperations();
/**
* Returns the underlying {@link CassandraConverter}.
*
* @return the underlying {@link CassandraConverter}.
*/
CassandraConverter getConverter();
// -------------------------------------------------------------------------
// Methods dealing with static CQL
// -------------------------------------------------------------------------
@@ -116,14 +133,13 @@ public interface AsyncCassandraOperations {
// -------------------------------------------------------------------------
/**
* Execute the Select by {@code id} for the given {@code entityClass}.
* Returns the number of rows for the given entity class.
*
* @param id must not be {@literal null}.
* @param entityClass The entity type must not be {@literal null}.
* @return the converted object or {@literal null}.
* @param entityClass must not be {@literal null}.
* @return the number of existing entities.
* @throws DataAccessException if there is any problem executing the query.
*/
<T> ListenableFuture<T> selectOneById(Object id, Class<T> entityClass) throws DataAccessException;
ListenableFuture<Long> count(Class<?> entityClass) throws DataAccessException;
/**
* Determine whether the row {@code entityClass} with the given {@code id} exists.
@@ -136,13 +152,14 @@ public interface AsyncCassandraOperations {
ListenableFuture<Boolean> exists(Object id, Class<?> entityClass) throws DataAccessException;
/**
* Returns the number of rows for the given entity class.
* Execute the Select by {@code id} for the given {@code entityClass}.
*
* @param entityClass must not be {@literal null}.
* @return the number of existing entities.
* @param id must not be {@literal null}.
* @param entityClass The entity type must not be {@literal null}.
* @return the converted object or {@literal null}.
* @throws DataAccessException if there is any problem executing the query.
*/
ListenableFuture<Long> count(Class<?> entityClass) throws DataAccessException;
<T> ListenableFuture<T> selectOneById(Object id, Class<T> entityClass) throws DataAccessException;
/**
* Insert the given entity and return the entity if the insert was applied.
@@ -182,16 +199,6 @@ public interface AsyncCassandraOperations {
*/
<T> ListenableFuture<T> update(T entity, WriteOptions options) throws DataAccessException;
/**
* Remove the given object from the table by id.
*
* @param id must not be {@literal null}.
* @param entityClass The entity type must not be {@literal null}.
* @return {@literal true} if the deletion was applied.
* @throws DataAccessException if there is any problem executing the query.
*/
ListenableFuture<Boolean> deleteById(Object id, Class<?> entityClass) throws DataAccessException;
/**
* Delete the given entity and return the entity if the delete was applied.
*
@@ -211,26 +218,22 @@ public interface AsyncCassandraOperations {
*/
<T> ListenableFuture<T> delete(T entity, QueryOptions options) throws DataAccessException;
/**
* Remove the given object from the table by id.
*
* @param id must not be {@literal null}.
* @param entityClass The entity type must not be {@literal null}.
* @return {@literal true} if the deletion was applied.
* @throws DataAccessException if there is any problem executing the query.
*/
ListenableFuture<Boolean> deleteById(Object id, Class<?> entityClass) throws DataAccessException;
/**
* Execute a {@code TRUNCATE} query to remove all entities of a given class.
*
*
* @param entityClass The entity type must not be {@literal null}.
* @throws DataAccessException if there is any problem executing the query.
*/
ListenableFuture<Void> truncate(Class<?> entityClass) throws DataAccessException;
/**
* Returns the underlying {@link CassandraConverter}.
*
* @return the underlying {@link CassandraConverter}.
*/
CassandraConverter getConverter();
/**
* Expose the underlying {@link AsyncCqlOperationsOperations} to allow asynchronous CQL operations.
*
* @return the underlying {@link AsyncCqlOperations}.
* @see AsyncCqlOperations
*/
AsyncCqlOperations getAsyncCqlOperations();
}

201
spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/AsyncCassandraTemplate.java
View File

@@ -64,15 +64,19 @@ import com.datastax.driver.core.querybuilder.Update;
* to the service directly, in the second case to the prepared template.
*
* @author Mark Paluch
* @author John Blum
* @since 2.0
*/
public class AsyncCassandraTemplate implements AsyncCassandraOperations {
private final CQLExceptionTranslator exceptionTranslator;
private final CassandraConverter converter;
private final CassandraMappingContext mappingContext;
private final AsyncCqlOperations cqlOperations;
private final CassandraConverter converter;
private final CassandraMappingContext mappingContext;
private final CQLExceptionTranslator exceptionTranslator;
/**
* Creates an instance of {@link AsyncCassandraTemplate} initialized with the given {@link Session} and a default
* {@link MappingCassandraConverter}.
@@ -104,8 +108,9 @@ public class AsyncCassandraTemplate implements AsyncCassandraOperations {
this.mappingContext = converter.getMappingContext();
AsyncCqlTemplate asyncCqlTemplate = new AsyncCqlTemplate(session);
this.exceptionTranslator = asyncCqlTemplate.getExceptionTranslator();
this.cqlOperations = asyncCqlTemplate;
this.exceptionTranslator = asyncCqlTemplate.getExceptionTranslator();
}
/**
@@ -129,9 +134,28 @@ public class AsyncCassandraTemplate implements AsyncCassandraOperations {
this.exceptionTranslator = asyncCqlTemplate.getExceptionTranslator();
}
/*
* (non-Javadoc)
* @see org.springframework.data.cassandra.core.AsyncCassandraOperations#getAsyncCqlOperations()
*/
@Override
public AsyncCqlOperations getAsyncCqlOperations() {
return cqlOperations;
}
/*
* (non-Javadoc)
* @see org.springframework.data.cassandra.core.AsyncCassandraOperations#getConverter()
*/
@Override
public CassandraConverter getConverter() {
return converter;
}
private static MappingCassandraConverter newConverter() {
MappingCassandraConverter converter = new MappingCassandraConverter();
converter.afterPropertiesSet();
return converter;
@@ -202,9 +226,7 @@ public class AsyncCassandraTemplate implements AsyncCassandraOperations {
Assert.notNull(entityConsumer, "Entity Consumer must not be empty");
Assert.notNull(entityClass, "Entity type must not be null");
return cqlOperations.query(statement, (row) -> {
entityConsumer.accept(converter.read(entityClass, row));
});
return cqlOperations.query(statement, (row) -> { entityConsumer.accept(converter.read(entityClass, row)); });
}
/*
@@ -214,56 +236,14 @@ public class AsyncCassandraTemplate implements AsyncCassandraOperations {
@Override
public <T> ListenableFuture<T> selectOne(Statement statement, Class<T> entityClass) {
return new MappingListenableFutureAdapter<>(select(statement, entityClass), list -> {
if (list.isEmpty()) {
return null;
}
return list.get(0);
});
return new MappingListenableFutureAdapter<>(select(statement, entityClass),
list -> list.isEmpty() ? null : list.get(0));
}
// -------------------------------------------------------------------------
// Methods dealing with entities
// -------------------------------------------------------------------------
/*
* (non-Javadoc)
* @see org.springframework.data.cassandra.core.AsyncCassandraOperations#selectOneById(java.lang.Object, java.lang.Class)
*/
@Override
public <T> ListenableFuture<T> selectOneById(Object id, Class<T> entityClass) {
Assert.notNull(id, "Id must not be null");
Assert.notNull(entityClass, "Entity type must not be null");
CassandraPersistentEntity<?> entity = getPersistentEntity(entityClass);
Select select = QueryBuilder.select().all().from(entity.getTableName().toCql());
converter.write(id, select.where(), entity);
return selectOne(select, entityClass);
}
/*
* (non-Javadoc)
* @see org.springframework.data.cassandra.core.AsyncCassandraOperations#exists(java.lang.Object, java.lang.Class)
*/
@Override
public ListenableFuture<Boolean> exists(Object id, Class<?> entityClass) {
Assert.notNull(id, "Id must not be null");
Assert.notNull(entityClass, "Entity type must not be null");
CassandraPersistentEntity<?> entity = getPersistentEntity(entityClass);
Select select = QueryBuilder.select().from(entity.getTableName().toCql());
converter.write(id, select.where(), entity);
return new MappingListenableFutureAdapter<>(cqlOperations.queryForResultSet(select),
resultSet -> resultSet.iterator().hasNext());
}
/*
* (non-Javadoc)
* @see org.springframework.data.cassandra.core.AsyncCassandraOperations#count(java.lang.Class)
@@ -278,6 +258,45 @@ public class AsyncCassandraTemplate implements AsyncCassandraOperations {
return cqlOperations.queryForObject(select, Long.class);
}
/*
* (non-Javadoc)
* @see org.springframework.data.cassandra.core.AsyncCassandraOperations#exists(java.lang.Object, java.lang.Class)
*/
@Override
public ListenableFuture<Boolean> exists(Object id, Class<?> entityClass) {
Assert.notNull(id, "Id must not be null");
Assert.notNull(entityClass, "Entity type must not be null");
CassandraPersistentEntity<?> entity = getPersistentEntity(entityClass);
Select select = QueryBuilder.select().from(entity.getTableName().toCql());
converter.write(id, select.where(), entity);
return new MappingListenableFutureAdapter<>(cqlOperations.queryForResultSet(select),
resultSet -> resultSet.iterator().hasNext());
}
/*
* (non-Javadoc)
* @see org.springframework.data.cassandra.core.AsyncCassandraOperations#selectOneById(java.lang.Object, java.lang.Class)
*/
@Override
public <T> ListenableFuture<T> selectOneById(Object id, Class<T> entityClass) {
Assert.notNull(id, "Id must not be null");
Assert.notNull(entityClass, "Entity type must not be null");
CassandraPersistentEntity<?> entity = getPersistentEntity(entityClass);
Select select = QueryBuilder.select().all().from(entity.getTableName().toCql());
converter.write(id, select.where(), entity);
return selectOne(select, entityClass);
}
/*
* (non-Javadoc)
* @see org.springframework.data.cassandra.core.AsyncCassandraOperations#insert(java.lang.Object)
@@ -296,9 +315,7 @@ public class AsyncCassandraTemplate implements AsyncCassandraOperations {
Assert.notNull(entity, "Entity must not be null");
CqlIdentifier tableName = getTableName(entity);
Insert insert = QueryUtils.createInsertQuery(tableName.toCql(), entity, options, converter);
Insert insert = QueryUtils.createInsertQuery(getTableName(entity).toCql(), entity, options, converter);
return new MappingListenableFutureAdapter<>(cqlOperations.execute(new AsyncStatementCallback(insert)),
resultSet -> resultSet.wasApplied() ? entity : null);
@@ -322,32 +339,12 @@ public class AsyncCassandraTemplate implements AsyncCassandraOperations {
Assert.notNull(entity, "Entity must not be null");
CqlIdentifier tableName = getTableName(entity);
Update update = QueryUtils.createUpdateQuery(tableName.toCql(), entity, options, converter);
Update update = QueryUtils.createUpdateQuery(getTableName(entity).toCql(), entity, options, converter);
return new MappingListenableFutureAdapter<>(cqlOperations.execute(new AsyncStatementCallback(update)),
resultSet -> resultSet.wasApplied() ? entity : null);
}
/*
* (non-Javadoc)
* @see org.springframework.data.cassandra.core.AsyncCassandraOperations#deleteById(java.lang.Object, java.lang.Class)
*/
@Override
public ListenableFuture<Boolean> deleteById(Object id, Class<?> entityClass) {
Assert.notNull(id, "Id must not be null");
Assert.notNull(entityClass, "Entity type must not be null");
CassandraPersistentEntity<?> entity = getPersistentEntity(entityClass);
Delete delete = QueryBuilder.delete().from(entity.getTableName().toCql());
converter.write(id, delete.where(), entity);
return cqlOperations.execute(delete);
}
/*
* (non-Javadoc)
* @see org.springframework.data.cassandra.core.AsyncCassandraOperations#delete(java.lang.Object)
@@ -366,14 +363,31 @@ public class AsyncCassandraTemplate implements AsyncCassandraOperations {
Assert.notNull(entity, "Entity must not be null");
CqlIdentifier tableName = getTableName(entity);
Delete delete = QueryUtils.createDeleteQuery(tableName.toCql(), entity, options, converter);
Delete delete = QueryUtils.createDeleteQuery(getTableName(entity).toCql(), entity, options, converter);
return new MappingListenableFutureAdapter<>(cqlOperations.execute(new AsyncStatementCallback(delete)),
resultSet -> resultSet.wasApplied() ? entity : null);
}
/*
* (non-Javadoc)
* @see org.springframework.data.cassandra.core.AsyncCassandraOperations#deleteById(java.lang.Object, java.lang.Class)
*/
@Override
public ListenableFuture<Boolean> deleteById(Object id, Class<?> entityClass) {
Assert.notNull(id, "Id must not be null");
Assert.notNull(entityClass, "Entity type must not be null");
CassandraPersistentEntity<?> entity = getPersistentEntity(entityClass);
Delete delete = QueryBuilder.delete().from(entity.getTableName().toCql());
converter.write(id, delete.where(), entity);
return cqlOperations.execute(delete);
}
/*
* (non-Javadoc)
* @see org.springframework.data.cassandra.core.AsyncCassandraOperations#truncate(java.lang.Class)
@@ -382,29 +396,12 @@ public class AsyncCassandraTemplate implements AsyncCassandraOperations {
public ListenableFuture<Void> truncate(Class<?> entityClass) {
Assert.notNull(entityClass, "Entity type must not be null");
Truncate truncate = QueryBuilder.truncate(getPersistentEntity(entityClass).getTableName().toCql());
return new MappingListenableFutureAdapter<>(cqlOperations.execute(truncate), aBoolean -> null);
}
/*
* (non-Javadoc)
* @see org.springframework.data.cassandra.core.AsyncCassandraOperations#getConverter()
*/
@Override
public CassandraConverter getConverter() {
return converter;
}
/*
* (non-Javadoc)
* @see org.springframework.data.cassandra.core.AsyncCassandraOperations#getAsyncCqlOperations()
*/
@Override
public AsyncCqlOperations getAsyncCqlOperations() {
return cqlOperations;
}
private <T> CassandraPersistentEntity<?> getPersistentEntity(Class<T> entityClass) {
Assert.notNull(entityClass, "Entity type must not be null");
@@ -449,14 +446,10 @@ public class AsyncCassandraTemplate implements AsyncCassandraOperations {
@Override
public ListenableFuture<ResultSet> doInSession(Session session) throws DriverException, DataAccessException {
return new GuavaListenableFutureAdapter<>(session.executeAsync(statement), e -> {
if (e instanceof DriverException) {
return exceptionTranslator.translate("AsyncStatementCallback", getCql(), (DriverException) e);
}
return exceptionTranslator.translateExceptionIfPossible(e);
});
return new GuavaListenableFutureAdapter<>(session.executeAsync(statement), e -> (e instanceof DriverException
? exceptionTranslator.translate("AsyncStatementCallback", getCql(), (DriverException) e)
: exceptionTranslator.translateExceptionIfPossible(e))
);
}
@Override

11
spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/CassandraAdminOperations.java
View File

@@ -17,14 +17,14 @@ package org.springframework.data.cassandra.core;
import java.util.Map;
import org.springframework.cassandra.core.cql.CqlIdentifier;
import com.datastax.driver.core.KeyspaceMetadata;
import com.datastax.driver.core.TableMetadata;
import org.springframework.cassandra.core.cql.CqlIdentifier;
/**
* Operations for managing a Cassandra keyspace.
*
*
* @author David Webb
* @author Matthew T. Adams
* @author Mark Paluch
@@ -37,7 +37,7 @@ public interface CassandraAdminOperations extends CassandraOperations {
* parameter <code>ifNotExists</code> is {@literal true}, this is a no-op and {@literal false} is returned. If the
* table doesn't exist, parameter <code>ifNotExists</code> is ignored, the table is created and {@literal true} is
* returned.
*
*
* @param ifNotExists If true, will only create the table if it doesn't exist, else the create operation will be
* ignored and the method will return {@literal false}.
* @param tableName The name of the table.
@@ -49,7 +49,7 @@ public interface CassandraAdminOperations extends CassandraOperations {
/**
* Drops the named table.
*
*
* @param tableName The name of the table.
*/
void dropTable(CqlIdentifier tableName);
@@ -78,4 +78,5 @@ public interface CassandraAdminOperations extends CassandraOperations {
* @since 1.5
*/
void dropUserType(CqlIdentifier typeName);
}

24
spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/CassandraAdminTemplate.java
View File

@@ -17,6 +17,10 @@ package org.springframework.data.cassandra.core;
import java.util.Map;
import com.datastax.driver.core.KeyspaceMetadata;
import com.datastax.driver.core.Session;
import com.datastax.driver.core.TableMetadata;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.cassandra.core.SessionCallback;
@@ -32,15 +36,12 @@ import org.springframework.data.cassandra.convert.CassandraConverter;
import org.springframework.data.cassandra.mapping.CassandraPersistentEntity;
import org.springframework.util.Assert;
import com.datastax.driver.core.KeyspaceMetadata;
import com.datastax.driver.core.Session;
import com.datastax.driver.core.TableMetadata;
/**
* Default implementation of {@link CassandraAdminOperations}.
*
* @author Mark Paluch
* @author Fabio J. Mendes
* @author John Blum
*/
public class CassandraAdminTemplate extends CassandraTemplate implements CassandraAdminOperations {
@@ -61,10 +62,11 @@ public class CassandraAdminTemplate extends CassandraTemplate implements Cassand
* @see org.springframework.data.cassandra.core.CassandraAdminOperations#createTable(boolean, org.springframework.cassandra.core.cql.CqlIdentifier, java.lang.Class, java.util.Map)
*/
@Override
public void createTable(final boolean ifNotExists, final CqlIdentifier tableName, Class<?> entityClass,
public void createTable(boolean ifNotExists, CqlIdentifier tableName, Class<?> entityClass,
Map<String, Object> optionsByName) {
CassandraPersistentEntity<?> entity = getPersistentEntity(entityClass);
CreateTableSpecification createTableSpecification = getConverter().getMappingContext()
.getCreateTableSpecificationFor(entity).ifNotExists(ifNotExists);
@@ -92,6 +94,7 @@ public class CassandraAdminTemplate extends CassandraTemplate implements Cassand
public void dropUserType(CqlIdentifier typeName) {
Assert.notNull(typeName, "Type name must not be null");
getCqlOperations().execute(DropUserTypeCqlGenerator.toCql(DropUserTypeSpecification.dropType(typeName)));
}
@@ -119,11 +122,14 @@ public class CassandraAdminTemplate extends CassandraTemplate implements Cassand
return getCqlOperations().execute(new SessionCallback<KeyspaceMetadata>() {
@Override
public KeyspaceMetadata doInSession(Session s) throws DataAccessException {
public KeyspaceMetadata doInSession(Session session) throws DataAccessException {
KeyspaceMetadata keyspaceMetadata = session.getCluster().getMetadata()
.getKeyspace(session.getLoggedKeyspace());
Assert.state(keyspaceMetadata != null, String.format("Metadata for keyspace [%s] not available",
session.getLoggedKeyspace()));
KeyspaceMetadata keyspaceMetadata = s.getCluster().getMetadata().getKeyspace(s.getLoggedKeyspace());
Assert.state(keyspaceMetadata != null,
String.format("Metadata for keyspace [%s] not available", s.getLoggedKeyspace()));
return keyspaceMetadata;
}
});

22
spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/CassandraBatchTemplate.java
View File

@@ -39,9 +39,20 @@ class CassandraBatchTemplate implements CassandraBatchOperations {
private final Batch batch;
private final CassandraOperations operations;
/* (non-Javadoc) */
@SafeVarargs
private static <T> Iterable<T> nullSafeIterable(T... array) {
return (array == null ? Collections.emptyList() : Arrays.asList(array));
}
/* (non-Javadoc) */
private static <T> Iterable<T> nullSafeIterable(Iterable<T> iterable) {
return (iterable != null ? iterable : Collections::emptyIterator);
}
/**
* Creates a new {@link CassandraBatchTemplate} given {@link CassandraOperations}.
*
*
* @param operations must not be {@literal null}.
*/
public CassandraBatchTemplate(CassandraOperations operations) {
@@ -166,14 +177,7 @@ class CassandraBatchTemplate implements CassandraBatchOperations {
private String getTableName(Object entity) {
Assert.notNull(entity, "Entity must not be null");
return operations.getTableName(entity.getClass()).toCql();
}
private <T> Iterable<T> nullSafeIterable(T... array) {
return (array == null ? Collections.<T> emptyList() : Arrays.asList(array));
}
private <T> Iterable<T> nullSafeIterable(Iterable<T> iterable) {
return (iterable != null ? iterable : Collections.<T> emptyList());
}
}

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

@@ -19,6 +19,8 @@ import java.util.Iterator;
import java.util.List;
import java.util.stream.Stream;
import com.datastax.driver.core.Statement;
import org.springframework.cassandra.core.CqlOperations;
import org.springframework.cassandra.core.QueryOptions;
import org.springframework.cassandra.core.WriteOptions;
@@ -26,8 +28,6 @@ import org.springframework.cassandra.core.cql.CqlIdentifier;
import org.springframework.dao.DataAccessException;
import org.springframework.data.cassandra.convert.CassandraConverter;
import com.datastax.driver.core.Statement;
/**
* Interface specifying a basic set of Cassandra operations. Implemented by {@link CassandraTemplate}. Not often used
* directly, but a useful option to enhance testability, as it can easily be mocked or stubbed.
@@ -42,6 +42,29 @@ import com.datastax.driver.core.Statement;
*/
public interface CassandraOperations {
/**
* Returns a new {@link CassandraBatchOperations}. Each {@link CassandraBatchOperations} instance can be executed only
* once so you might want to obtain new {@link CassandraBatchOperations} instances for each batch.
*
* @return a new {@link CassandraBatchOperations} associated with the given entity class.
*/
CassandraBatchOperations batchOps();
/**
* Returns the underlying {@link CassandraConverter}.
*
* @return the underlying {@link CassandraConverter}.
*/
CassandraConverter getConverter();
/**
* Expose the underlying {@link CqlOperations} to allow CQL operations.
*
* @return the underlying {@link CqlOperations}.
* @see CqlOperations
*/
CqlOperations getCqlOperations();
/**
* The table name used for the specified class by this template.
*
@@ -130,6 +153,25 @@ public interface CassandraOperations {
// Methods dealing with entities
// -------------------------------------------------------------------------
/**
* Returns the number of rows for the given entity class.
*
* @param entityClass must not be {@literal null}.
* @return the number of existing entities.
* @throws DataAccessException if there is any problem executing the query.
*/
long count(Class<?> entityClass) throws DataAccessException;
/**
* Determine whether the row {@code entityClass} with the given {@code id} exists.
*
* @param id must not be {@literal null}.
* @param entityClass The entity type must not be {@literal null}.
* @return true, if the object exists.
* @throws DataAccessException if there is any problem executing the query.
*/
boolean exists(Object id, Class<?> entityClass) throws DataAccessException;
/**
* Execute the Select by {@code id} for the given {@code entityClass}.
*
@@ -150,25 +192,6 @@ public interface CassandraOperations {
*/
<T> List<T> selectBySimpleIds(Iterable<?> ids, Class<T> entityClass) throws DataAccessException;
/**
* Determine whether the row {@code entityClass} with the given {@code id} exists.
*
* @param id must not be {@literal null}.
* @param entityClass The entity type must not be {@literal null}.
* @return true, if the object exists.
* @throws DataAccessException if there is any problem executing the query.
*/
boolean exists(Object id, Class<?> entityClass) throws DataAccessException;
/**
* Returns the number of rows for the given entity class.
*
* @param entityClass must not be {@literal null}.
* @return the number of existing entities.
* @throws DataAccessException if there is any problem executing the query.
*/
long count(Class<?> entityClass) throws DataAccessException;
/**
* Insert the given entity and return the entity if the insert was applied.
*
@@ -207,15 +230,6 @@ public interface CassandraOperations {
*/
<T> T update(T entity, WriteOptions options) throws DataAccessException;
/**
* Remove the given object from the table by id.
*
* @param id must not be {@literal null}.
* @param entityClass The entity type must not be {@literal null}.
* @throws DataAccessException if there is any problem executing the query.
*/
boolean deleteById(Object id, Class<?> entityClass) throws DataAccessException;
/**
* Delete the given entity and return the entity if the delete was applied.
*
@@ -235,34 +249,21 @@ public interface CassandraOperations {
*/
<T> T delete(T entity, QueryOptions options) throws DataAccessException;
/**
* Remove the given object from the table by id.
*
* @param id must not be {@literal null}.
* @param entityClass The entity type must not be {@literal null}.
* @throws DataAccessException if there is any problem executing the query.
*/
boolean deleteById(Object id, Class<?> entityClass) throws DataAccessException;
/**
* Execute a {@code TRUNCATE} query to remove all entities of a given class.
*
*
* @param entityClass The entity type must not be {@literal null}.
* @throws DataAccessException if there is any problem executing the query.
*/
void truncate(Class<?> entityClass) throws DataAccessException;
/**
* Returns a new {@link CassandraBatchOperations}. Each {@link CassandraBatchOperations} instance can be executed only
* once so you might want to obtain new {@link CassandraBatchOperations} instances for each batch.
*
* @return a new {@link CassandraBatchOperations} associated with the given entity class.
*/
CassandraBatchOperations batchOps();
/**
* Returns the underlying {@link CassandraConverter}.
*
* @return the underlying {@link CassandraConverter}.
*/
CassandraConverter getConverter();
/**
* Expose the underlying {@link CqlOperations} to allow CQL operations.
*
* @return the underlying {@link CqlOperations}.
* @see CqlOperations
*/
CqlOperations getCqlOperations();
}

70
spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/CassandraPersistentEntitySchemaCreator.java
View File

@@ -72,6 +72,25 @@ public class CassandraPersistentEntitySchemaCreator {
this.cassandraAdminOperations = cassandraAdminOperations;
}
/**
* Create user types. Can drop types and drop unused types.
*
* @param dropTables {@literal true} to drop tables before creation.
* @param dropUnused {@literal true} to drop unused tables before creation. Table usage is determined by existing
* table mappings.
* @param ifNotExists {@literal true} to create tables using {@code IF NOT EXISTS}.
*/
public void createTables(boolean dropTables, boolean dropUnused, boolean ifNotExists) {
if (dropTables) {
dropTables(dropUnused);
}
for (CreateTableSpecification specification : createTableSpecifications(ifNotExists)) {
cassandraAdminOperations.getCqlOperations().execute(CreateTableCqlGenerator.toCql(specification));
}
}
/**
* Create user types. Can drop types and drop unused types.
*
@@ -87,74 +106,52 @@ public class CassandraPersistentEntitySchemaCreator {
dropUserTypes(dropUnused);
}
List<CreateUserTypeSpecification> specifications = createUserTypeSpecifications(ifNotExists);
for (CreateUserTypeSpecification specification : specifications) {
for (CreateUserTypeSpecification specification : createUserTypeSpecifications(ifNotExists)) {
cassandraAdminOperations.getCqlOperations().execute(CreateUserTypeCqlGenerator.toCql(specification));
}
}
/**
* Create user types. Can drop types and drop unused types.
*
* @param dropTables {@literal true} to drop tables before creation.
* @param dropUnused {@literal true} to drop unused tables before creation. Table usage is determined by existing
* table mappings.
* @param ifNotExists {@literal true} to create tables using {@code IF NOT EXISTS}.
*/
public void createTables(boolean dropTables, boolean dropUnused, boolean ifNotExists) {
if (dropTables) {
dropTables(dropUnused);
}
List<CreateTableSpecification> specifications = createTableSpecifications(ifNotExists);
for (CreateTableSpecification specification : specifications) {
cassandraAdminOperations.getCqlOperations().execute(CreateTableCqlGenerator.toCql(specification));
}
}
protected List<CreateUserTypeSpecification> createUserTypeSpecifications(boolean ifNotExists) {
Collection<? extends CassandraPersistentEntity<?>> entities = new ArrayList<CassandraPersistentEntity<?>>(
Collection<? extends CassandraPersistentEntity<?>> entities = new ArrayList<>(
mappingContext.getUserDefinedTypeEntities());
Map<CqlIdentifier, CassandraPersistentEntity<?>> byName = new HashMap<CqlIdentifier, CassandraPersistentEntity<?>>();
Map<CqlIdentifier, CassandraPersistentEntity<?>> byName = new HashMap<>();
for (CassandraPersistentEntity<?> entity : entities) {
byName.put(entity.getTableName(), entity);
}
List<CreateUserTypeSpecification> specifications = new ArrayList<CreateUserTypeSpecification>();
List<CreateUserTypeSpecification> specifications = new ArrayList<>();
Set<CqlIdentifier> created = new HashSet<CqlIdentifier>();
Set<CqlIdentifier> created = new HashSet<>();
for (CassandraPersistentEntity<?> entity : entities) {
Set<CqlIdentifier> seen = new LinkedHashSet<CqlIdentifier>();
Set<CqlIdentifier> seen = new LinkedHashSet<>();
seen.add(entity.getTableName());
visitUserTypes(entity, seen);
List<CqlIdentifier> ordered = new ArrayList<CqlIdentifier>(seen);
List<CqlIdentifier> ordered = new ArrayList<>(seen);
Collections.reverse(ordered);
for (CqlIdentifier identifier : ordered) {
if (created.add(identifier)) {
specifications.add(mappingContext.getCreateUserTypeSpecificationFor(
byName.get(identifier)).ifNotExists(ifNotExists));
}
}
}
return specifications;
}
protected List<CreateTableSpecification> createTableSpecifications(boolean ifNotExists) {
Collection<? extends CassandraPersistentEntity<?>> entities = new ArrayList<CassandraPersistentEntity<?>>(
Collection<? extends CassandraPersistentEntity<?>> entities = new ArrayList<>(
mappingContext.getNonPrimaryKeyEntities());
List<CreateTableSpecification> specifications = new ArrayList<CreateTableSpecification>();
List<CreateTableSpecification> specifications = new ArrayList<>();
for (CassandraPersistentEntity<?> entity : entities) {
specifications.add(mappingContext.getCreateTableSpecificationFor(entity).ifNotExists(ifNotExists));
@@ -171,6 +168,7 @@ public class CassandraPersistentEntitySchemaCreator {
public void doWithPersistentProperty(CassandraPersistentProperty persistentProperty) {
CassandraPersistentEntity<?> persistentEntity = mappingContext.getPersistentEntity(persistentProperty);
if (persistentEntity != null && persistentEntity.isUserDefinedType()) {
if (seen.add(persistentEntity.getTableName())) {
visitUserTypes(persistentEntity, seen);
@@ -182,15 +180,15 @@ public class CassandraPersistentEntitySchemaCreator {
private void dropUserTypes(boolean dropUnused) {
KeyspaceMetadata keyspaceMetadata = cassandraAdminOperations.getKeyspaceMetadata();
Collection<CassandraPersistentEntity<?>> userDefinedTypeEntities = mappingContext.getUserDefinedTypeEntities();
Set<CqlIdentifier> canRecreate = new HashSet<CqlIdentifier>();
Set<CqlIdentifier> canRecreate = new HashSet<>();
for (CassandraPersistentEntity<?> userDefinedTypeEntity : userDefinedTypeEntities) {
canRecreate.add(userDefinedTypeEntity.getTableName());
}
KeyspaceMetadata keyspaceMetadata = cassandraAdminOperations.getKeyspaceMetadata();
for (UserType userType : keyspaceMetadata.getUserTypes()) {
CqlIdentifier identifier = CqlIdentifier.cqlId(userType.getTypeName());

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

@@ -19,6 +19,18 @@ import java.util.List;
import java.util.stream.Stream;
import java.util.stream.StreamSupport;
import com.datastax.driver.core.ResultSet;
import com.datastax.driver.core.Session;
import com.datastax.driver.core.SimpleStatement;
import com.datastax.driver.core.Statement;
import com.datastax.driver.core.exceptions.DriverException;
import com.datastax.driver.core.querybuilder.Delete;
import com.datastax.driver.core.querybuilder.Insert;
import com.datastax.driver.core.querybuilder.QueryBuilder;
import com.datastax.driver.core.querybuilder.Select;
import com.datastax.driver.core.querybuilder.Truncate;
import com.datastax.driver.core.querybuilder.Update;
import org.springframework.cassandra.core.CqlOperations;
import org.springframework.cassandra.core.CqlProvider;
import org.springframework.cassandra.core.CqlTemplate;
@@ -36,18 +48,6 @@ import org.springframework.data.cassandra.mapping.CassandraPersistentEntity;
import org.springframework.util.Assert;
import org.springframework.util.ClassUtils;
import com.datastax.driver.core.ResultSet;
import com.datastax.driver.core.Session;
import com.datastax.driver.core.SimpleStatement;
import com.datastax.driver.core.Statement;
import com.datastax.driver.core.exceptions.DriverException;
import com.datastax.driver.core.querybuilder.Delete;
import com.datastax.driver.core.querybuilder.Insert;
import com.datastax.driver.core.querybuilder.QueryBuilder;
import com.datastax.driver.core.querybuilder.Select;
import com.datastax.driver.core.querybuilder.Truncate;
import com.datastax.driver.core.querybuilder.Update;
/**
* Primary implementation of {@link CassandraOperations}. It simplifies the use of Cassandra usage and helps to avoid
* common errors. It executes core Cassandra workflow. This class executes CQL queries or updates, initiating iteration
@@ -61,6 +61,7 @@ import com.datastax.driver.core.querybuilder.Update;
* to the service directly, in the second case to the prepared template.
*
* @author Mark Paluch
* @author John Blum
* @since 2.0
*/
public class CassandraTemplate implements CassandraOperations {
@@ -209,17 +210,46 @@ public class CassandraTemplate implements CassandraOperations {
List<T> result = select(statement, entityClass);
if (result.isEmpty()) {
return null;
}
return result.get(0);
return (result.isEmpty() ? null : result.get(0));
}
// -------------------------------------------------------------------------
// Methods dealing with entities
// -------------------------------------------------------------------------
/*
* (non-Javadoc)
* @see org.springframework.data.cassandra.core.CassandraOperations#count(java.lang.Class)
*/
@Override
public long count(Class<?> entityClass) {
Assert.notNull(entityClass, "Entity type must not be null");
Select select = QueryBuilder.select().countAll().from(getPersistentEntity(entityClass).getTableName().toCql());
return cqlOperations.queryForObject(select, Long.class);
}
/*
* (non-Javadoc)
* @see org.springframework.data.cassandra.core.CassandraOperations#exists(java.lang.Object, java.lang.Class)
*/
@Override
public boolean exists(Object id, Class<?> entityClass) {
Assert.notNull(id, "Id must not be null");
Assert.notNull(entityClass, "Entity type must not be null");
CassandraPersistentEntity<?> entity = getPersistentEntity(entityClass);
Select select = QueryBuilder.select().from(entity.getTableName().toCql());
converter.write(id, select.where(), entity);
return cqlOperations.queryForResultSet(select).iterator().hasNext();
}
/*
* (non-Javadoc)
* @see org.springframework.data.cassandra.core.CassandraOperations#selectOneById(java.lang.Object, java.lang.Class)
@@ -231,6 +261,7 @@ public class CassandraTemplate implements CassandraOperations {
Assert.notNull(entityClass, "Entity type must not be null");
CassandraPersistentEntity<?> entity = getPersistentEntity(entityClass);
Select select = QueryBuilder.select().all().from(entity.getTableName().toCql());
converter.write(id, select.where(), entity);
@@ -262,37 +293,6 @@ public class CassandraTemplate implements CassandraOperations {
return select(select, entityClass);
}
/*
* (non-Javadoc)
* @see org.springframework.data.cassandra.core.CassandraOperations#exists(java.lang.Object, java.lang.Class)
*/
@Override
public boolean exists(Object id, Class<?> entityClass) {
Assert.notNull(id, "Id must not be null");
Assert.notNull(entityClass, "Entity type must not be null");
CassandraPersistentEntity<?> entity = getPersistentEntity(entityClass);
Select select = QueryBuilder.select().from(entity.getTableName().toCql());
converter.write(id, select.where(), entity);
return cqlOperations.queryForResultSet(select).iterator().hasNext();
}
/*
* (non-Javadoc)
* @see org.springframework.data.cassandra.core.CassandraOperations#count(java.lang.Class)
*/
@Override
public long count(Class<?> entityClass) {
Assert.notNull(entityClass, "Entity type must not be null");
Select select = QueryBuilder.select().countAll().from(getPersistentEntity(entityClass).getTableName().toCql());
return cqlOperations.queryForObject(select, Long.class);
}
/*
* (non-Javadoc)
* @see org.springframework.data.cassandra.core.CassandraOperations#insert(java.lang.Object)
@@ -311,9 +311,8 @@ public class CassandraTemplate implements CassandraOperations {
Assert.notNull(entity, "Entity must not be null");
CqlIdentifier tableName = getTableName(entity.getClass());
Insert insert = QueryUtils.createInsertQuery(tableName.toCql(), entity, options, converter);
Insert insert = QueryUtils.createInsertQuery(getTableName(entity.getClass()).toCql(),
entity, options, converter);
return cqlOperations.execute(new StatementCallback<>(insert, entity));
}
@@ -336,31 +335,12 @@ public class CassandraTemplate implements CassandraOperations {
Assert.notNull(entity, "Entity must not be null");
CqlIdentifier tableName = getTableName(entity.getClass());
Update update = QueryUtils.createUpdateQuery(tableName.toCql(), entity, options, converter);
Update update = QueryUtils.createUpdateQuery(getTableName(entity.getClass()).toCql(),
entity, options, converter);
return cqlOperations.execute(new StatementCallback<>(update, entity));
}
/*
* (non-Javadoc)
* @see org.springframework.data.cassandra.core.CassandraOperations#deleteById(java.lang.Object, java.lang.Class)
*/
@Override
public boolean deleteById(Object id, Class<?> entityClass) {
Assert.notNull(id, "Id must not be null");
Assert.notNull(entityClass, "Entity type must not be null");
CassandraPersistentEntity<?> entity = getPersistentEntity(entityClass);
Delete delete = QueryBuilder.delete().from(entity.getTableName().toCql());
converter.write(id, delete.where(), entity);
return cqlOperations.execute(delete);
}
/*
* (non-Javadoc)
* @see org.springframework.data.cassandra.core.CassandraOperations#delete(java.lang.Object)
@@ -379,13 +359,31 @@ public class CassandraTemplate implements CassandraOperations {
Assert.notNull(entity, "Entity must not be null");
CqlIdentifier tableName = getTableName(entity.getClass());
Delete delete = QueryUtils.createDeleteQuery(tableName.toCql(), entity, options, converter);
Delete delete = QueryUtils.createDeleteQuery(getTableName(entity.getClass()).toCql(),
entity, options, converter);
return cqlOperations.execute(new StatementCallback<>(delete, entity));
}
/*
* (non-Javadoc)
* @see org.springframework.data.cassandra.core.CassandraOperations#deleteById(java.lang.Object, java.lang.Class)
*/
@Override
public boolean deleteById(Object id, Class<?> entityClass) {
Assert.notNull(id, "Id must not be null");
Assert.notNull(entityClass, "Entity type must not be null");
CassandraPersistentEntity<?> entity = getPersistentEntity(entityClass);
Delete delete = QueryBuilder.delete().from(entity.getTableName().toCql());
converter.write(id, delete.where(), entity);
return cqlOperations.execute(delete);
}
/*
* (non-Javadoc)
* @see org.springframework.data.cassandra.core.CassandraOperations#truncate(java.lang.Class)
@@ -394,11 +392,16 @@ public class CassandraTemplate implements CassandraOperations {
public void truncate(Class<?> entityClass) {
Assert.notNull(entityClass, "Entity type must not be null");
Truncate truncate = QueryBuilder.truncate(getPersistentEntity(entityClass).getTableName().toCql());
cqlOperations.execute(truncate);
}
// -------------------------------------------------------------------------
// Implementation hooks and helper methods
// -------------------------------------------------------------------------
/*
* (non-Javadoc)
* @see org.springframework.data.cassandra.core.CassandraOperations#getConverter()
@@ -417,26 +420,6 @@ public class CassandraTemplate implements CassandraOperations {
return cqlOperations;
}
// -------------------------------------------------------------------------
// Implementation hooks and helper methods
// -------------------------------------------------------------------------
/* (non-Javadoc)
* @see org.springframework.data.cassandra.core.CassandraOperationsNG#getTableName(java.lang.Class)
*/
@Override
public CqlIdentifier getTableName(Class<?> entityClass) {
return getPersistentEntity(ClassUtils.getUserClass(entityClass)).getTableName();
}
/* (non-Javadoc)
* @see org.springframework.data.cassandra.core.CassandraOperationsNG#batchOps()
*/
@Override
public CassandraBatchOperations batchOps() {
return new CassandraBatchTemplate(this);
}
protected <T> CassandraPersistentEntity<?> getPersistentEntity(Class<T> entityClass) {
Assert.notNull(entityClass, "Entity type must not be null");
@@ -445,12 +428,30 @@ public class CassandraTemplate implements CassandraOperations {
if (entity == null) {
throw new InvalidDataAccessApiUsageException(
String.format("No Persistent Entity information found for the class [%s]", entityClass.getName()));
String.format("No Persistent Entity information found for the class [%s]", entityClass.getName()));
}
return entity;
}
/*
* (non-Javadoc)
* @see org.springframework.data.cassandra.core.CassandraOperationsNG#getTableName(java.lang.Class)
*/
@Override
public CqlIdentifier getTableName(Class<?> entityClass) {
return getPersistentEntity(ClassUtils.getUserClass(entityClass)).getTableName();
}
/*
* (non-Javadoc)
* @see org.springframework.data.cassandra.core.CassandraOperationsNG#batchOps()
*/
@Override
public CassandraBatchOperations batchOps() {
return new CassandraBatchTemplate(this);
}
private static class StatementCallback<T> implements SessionCallback<T>, CqlProvider {
private final Statement statement;

39
spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/ReactiveCassandraTemplate.java
View File

@@ -15,6 +15,17 @@
*/
package org.springframework.data.cassandra.core;
import com.datastax.driver.core.Session;
import com.datastax.driver.core.SimpleStatement;
import com.datastax.driver.core.Statement;
import com.datastax.driver.core.exceptions.DriverException;
import com.datastax.driver.core.querybuilder.Delete;
import com.datastax.driver.core.querybuilder.Insert;
import com.datastax.driver.core.querybuilder.QueryBuilder;
import com.datastax.driver.core.querybuilder.Select;
import com.datastax.driver.core.querybuilder.Truncate;
import com.datastax.driver.core.querybuilder.Update;
import org.reactivestreams.Publisher;
import org.springframework.cassandra.core.CqlProvider;
import org.springframework.cassandra.core.DefaultReactiveSessionFactory;
@@ -36,17 +47,6 @@ import org.springframework.data.cassandra.mapping.CassandraPersistentEntity;
import org.springframework.util.Assert;
import org.springframework.util.ClassUtils;
import com.datastax.driver.core.Session;
import com.datastax.driver.core.SimpleStatement;
import com.datastax.driver.core.Statement;
import com.datastax.driver.core.exceptions.DriverException;
import com.datastax.driver.core.querybuilder.Delete;
import com.datastax.driver.core.querybuilder.Insert;
import com.datastax.driver.core.querybuilder.QueryBuilder;
import com.datastax.driver.core.querybuilder.Select;
import com.datastax.driver.core.querybuilder.Truncate;
import com.datastax.driver.core.querybuilder.Update;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
@@ -141,6 +141,7 @@ public class ReactiveCassandraTemplate implements ReactiveCassandraOperations {
private static MappingCassandraConverter newConverter() {
MappingCassandraConverter converter = new MappingCassandraConverter();
converter.afterPropertiesSet();
return converter;
@@ -212,6 +213,7 @@ public class ReactiveCassandraTemplate implements ReactiveCassandraOperations {
Assert.notNull(entityClass, "Entity type must not be null");
CassandraPersistentEntity<?> entity = getPersistentEntity(entityClass);
Select select = QueryBuilder.select().all().from(entity.getTableName().toCql());
converter.write(id, select.where(), entity);
@@ -230,7 +232,9 @@ public class ReactiveCassandraTemplate implements ReactiveCassandraOperations {
Assert.notNull(entityClass, "Entity type must not be null");
CassandraPersistentEntity<?> entity = getPersistentEntity(entityClass);
Select select = QueryBuilder.select().from(entity.getTableName().toCql());
converter.write(id, select.where(), entity);
return cqlOperations.queryForRows(select).hasElements();
@@ -268,9 +272,7 @@ public class ReactiveCassandraTemplate implements ReactiveCassandraOperations {
Assert.notNull(entity, "Entity must not be null");
CqlIdentifier tableName = getTableName(entity);
Insert insert = QueryUtils.createInsertQuery(tableName.toCql(), entity, options, converter);
Insert insert = QueryUtils.createInsertQuery(getTableName(entity).toCql(), entity, options, converter);
class InsertCallback implements ReactiveSessionCallback<T>, CqlProvider {
@@ -328,9 +330,7 @@ public class ReactiveCassandraTemplate implements ReactiveCassandraOperations {
Assert.notNull(entity, "Entity must not be null");
CqlIdentifier tableName = getTableName(entity);
Update update = QueryUtils.createUpdateQuery(tableName.toCql(), entity, options, converter);
Update update = QueryUtils.createUpdateQuery(getTableName(entity).toCql(), entity, options, converter);
class UpdateCallback implements ReactiveSessionCallback<T>, CqlProvider {
@@ -381,6 +381,7 @@ public class ReactiveCassandraTemplate implements ReactiveCassandraOperations {
Assert.notNull(entityClass, "Entity type must not be null");
CassandraPersistentEntity<?> entity = getPersistentEntity(entityClass);
Delete delete = QueryBuilder.delete().from(entity.getTableName().toCql());
converter.write(id, delete.where(), entity);
@@ -406,9 +407,7 @@ public class ReactiveCassandraTemplate implements ReactiveCassandraOperations {
Assert.notNull(entity, "Entity must not be null");
CqlIdentifier tableName = getTableName(entity);
Delete delete = QueryUtils.createDeleteQuery(tableName.toCql(), entity, options, converter);
Delete delete = QueryUtils.createDeleteQuery(getTableName(entity).toCql(), entity, options, converter);
class DeleteCallback implements ReactiveSessionCallback<T>, CqlProvider {

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

@@ -16,13 +16,10 @@
package org.springframework.data.cassandra.repository.query;
import java.util.function.Function;
import org.springframework.core.convert.converter.Converter;
import org.springframework.data.cassandra.core.CassandraOperations;
import org.springframework.data.repository.query.ResultProcessor;
import org.springframework.data.repository.query.ReturnedType;
import org.springframework.data.util.StreamUtils;
import org.springframework.util.ClassUtils;
import lombok.NonNull;

6
spring-data-cassandra/src/main/java/org/springframework/data/cassandra/repository/query/StringBasedCassandraQuery.java
View File

@@ -16,7 +16,9 @@
package org.springframework.data.cassandra.repository.query;
import com.datastax.driver.core.Cluster;
import com.datastax.driver.core.CodecRegistry;
import com.datastax.driver.core.Session;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.data.cassandra.core.CassandraOperations;
@@ -24,8 +26,6 @@ import org.springframework.data.repository.query.EvaluationContextProvider;
import org.springframework.data.repository.query.QueryCreationException;
import org.springframework.expression.spel.standard.SpelExpressionParser;
import com.datastax.driver.core.CodecRegistry;
/**
* String-based {@link AbstractCassandraQuery} implementation.
* <p>
@@ -78,6 +78,7 @@ public class StringBasedCassandraQuery extends AbstractCassandraQuery {
Cluster cluster = operations.getCqlOperations().execute(Session::getCluster);
CodecRegistry codecRegistry = cluster.getConfiguration().getCodecRegistry();
this.stringBasedQuery = new StringBasedQuery(query,
new ExpressionEvaluatingParameterBinder(expressionParser, evaluationContextProvider), codecRegistry);
}
@@ -87,7 +88,6 @@ public class StringBasedCassandraQuery extends AbstractCassandraQuery {
*/
@Override
public String createQuery(CassandraParameterAccessor parameterAccessor) {
try {
String boundQuery = stringBasedQuery.bindQuery(parameterAccessor, getQueryMethod());

16
spring-data-cassandra/src/main/java/org/springframework/data/cassandra/repository/support/SimpleCassandraRepository.java
View File

@@ -1,12 +1,12 @@
/*
* Copyright 2013-2016 the original author or authors
*
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
*
* http://www.apache.org/licenses/LICENSE-2.0
*
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
@@ -18,6 +18,9 @@ package org.springframework.data.cassandra.repository.support;
import java.io.Serializable;
import java.util.List;
import com.datastax.driver.core.querybuilder.QueryBuilder;
import com.datastax.driver.core.querybuilder.Select;
import org.springframework.cassandra.core.util.CollectionUtils;
import org.springframework.data.cassandra.core.CassandraOperations;
import org.springframework.data.cassandra.core.CassandraTemplate;
@@ -25,12 +28,9 @@ import org.springframework.data.cassandra.repository.TypedIdCassandraRepository;
import org.springframework.data.cassandra.repository.query.CassandraEntityInformation;
import org.springframework.util.Assert;
import com.datastax.driver.core.querybuilder.QueryBuilder;
import com.datastax.driver.core.querybuilder.Select;
/**
* Repository base implementation for Cassandra.
*
*
* @author Alex Shvid
* @author Matthew T. Adams
* @author Mark Paluch
@@ -43,7 +43,7 @@ public class SimpleCassandraRepository<T, ID extends Serializable> implements Ty
/**
* Creates a new {@link SimpleCassandraRepository} for the given {@link CassandraEntityInformation} and
* {@link CassandraTemplate}.
*
*
* @param metadata must not be {@literal null}.
* @param operations must not be {@literal null}.
*/