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

Merge branch 'DATACASS-32' of

Browse Source 
https://github.com/shvid/spring-data-cassandra into DATACASS-32
This commit is contained in:
David Webb
2013-11-20 20:25:27 +00:00
parent 37de816c2d 8ed4f65874
commit ad7db43d5a
18 changed files with 1014 additions and 263 deletions
Download Patch File Download Diff File Expand all files Collapse all files

133
src/main/java/org/springframework/cassandra/core/CassandraOperations.java
View File

@@ -15,61 +15,126 @@
*/
package org.springframework.cassandra.core;
import java.util.Collection;
import java.util.List;
import java.util.Map;
import org.springframework.dao.DataAccessException;
import com.datastax.driver.core.ResultSet;
import com.datastax.driver.core.ResultSetFuture;
/**
* Operations for interacting with Cassandra. These operations are used by the Repository implementation, but can also
* be used directly when that is desired by the developer.
* Operations for interacting with Cassandra at the lowest level. This interface provides Exception Translation.
*
* @author Alex Shvid
* @author David Webb
* @author Matthew Adams
*
*/
public interface CassandraOperations {
<T> T execute(SessionCallback<T> sessionCallback);
void execute(final String cql);
RuntimeException potentiallyConvertRuntimeException(RuntimeException ex);
<T> T query(String cql, ResultSetExtractor<T> rse) throws DataAccessException;
void query(String cql, RowCallbackHandler rch) throws DataAccessException;
<T> List<T> query(String cql, RowMapper<T> rowMapper) throws DataAccessException;
<T> T queryForObject(String cql, RowMapper<T> rowMapper) throws DataAccessException;
<T> T queryForObject(String cql, Class<T> requiredType) throws DataAccessException;
Map<String, Object> queryForMap(String cql) throws DataAccessException;
<T> List<T> queryForList(String cql, Class<T> elementType) throws DataAccessException;
List<Map<String, Object>> queryForList(String cql) throws DataAccessException;
/**
* Execute query and return Cassandra ResultSet
* Executes the supplied {@link SessionCallback} in the current Template Session. The implementation of
* SessionCallback can decide whether or not to <code>execute()</code> or <code>executeAsync()</code> the operation.
*
* @param cql must not be {@literal null}.
* @param sessionCallback
* @return
*/
ResultSet executeQuery(final String cql);
<T> T execute(SessionCallback<T> sessionCallback) throws DataAccessException;
/**
* Execute async query and return Cassandra ResultSetFuture
* Executes the supplied CQL Query and returns nothing.
*
* @param cql must not be {@literal null}.
* @return
* @param cql
*/
ResultSetFuture executeQueryAsynchronously(final String cql);
void execute(final String cql) throws DataAccessException;
/**
* Executes the supplied CQL Query Asynchrously and returns nothing.
*
* @param cql
*/
void executeAsynchronously(final String cql) throws DataAccessException;
/**
* Executes the provided CQL Query, and extracts the results with the ResultSetExtractor
*
* @param cql The Query
* @param rse The implementation for extracting the results
*
* @return
* @throws DataAccessException
*/
<T> T query(final String cql, ResultSetExtractor<T> rse) throws DataAccessException;
/**
* Executes the provided CQL Query asynchronously, and extracts the results with the ResultSetFutureExtractor
*
* @param cql The Query
* @param rse The implementation for extracting the results
* @return
* @throws DataAccessException
*/
<T> T queryAsynchronously(final String cql, ResultSetFutureExtractor<T> rse) throws DataAccessException;
void query(final String cql, RowCallbackHandler rch) throws DataAccessException;
void process(ResultSet resultSet, RowCallbackHandler rch) throws DataAccessException;
<T> List<T> query(final String cql, RowMapper<T> rowMapper) throws DataAccessException;
<T> List<T> process(ResultSet resultSet, RowMapper<T> rowMapper) throws DataAccessException;
<T> T queryForObject(final String cql, RowMapper<T> rowMapper) throws DataAccessException;
<T> T processOne(ResultSet resultSet, RowMapper<T> rowMapper) throws DataAccessException;
<T> T queryForObject(final String cql, Class<T> requiredType) throws DataAccessException;
<T> T processOne(ResultSet resultSet, Class<T> requiredType) throws DataAccessException;
Map<String, Object> queryForMap(final String cql) throws DataAccessException;
Map<String, Object> processMap(ResultSet resultSet) throws DataAccessException;
<T> List<T> queryForList(final String cql, Class<T> elementType) throws DataAccessException;
<T> List<T> processList(ResultSet resultSet, Class<T> elementType) throws DataAccessException;
List<Map<String, Object>> queryForListOfMap(final String cql) throws DataAccessException;
List<Map<String, Object>> processListOfMap(ResultSet resultSet) throws DataAccessException;
<T> T execute(String cql, PreparedStatementCallback<T> action) throws DataAccessException;
<T> T execute(PreparedStatementCreator psc, PreparedStatementCallback<T> action) throws DataAccessException;
<T> T query(final String cql, PreparedStatementBinder pss, ResultSetExtractor<T> rse) throws DataAccessException;
void query(final String cql, PreparedStatementBinder pss, RowCallbackHandler rch) throws DataAccessException;
<T> List<T> query(final String cql, PreparedStatementBinder pss, RowMapper<T> rowMapper) throws DataAccessException;
<T> T query(PreparedStatementCreator psc, ResultSetExtractor<T> rse) throws DataAccessException;
void query(PreparedStatementCreator psc, RowCallbackHandler rch) throws DataAccessException;
<T> List<T> query(PreparedStatementCreator psc, RowMapper<T> rowMapper) throws DataAccessException;
<T> T query(PreparedStatementCreator psc, final PreparedStatementBinder pss, final ResultSetExtractor<T> rse)
throws DataAccessException;
void query(PreparedStatementCreator psc, final PreparedStatementBinder pss, final RowCallbackHandler rch)
throws DataAccessException;
<T> List<T> query(PreparedStatementCreator psc, final PreparedStatementBinder pss, final RowMapper<T> rowMapper)
throws DataAccessException;
/**
* Describe the current Ring
*
* @return The list of ring tokens that are active in the cluster
*/
List<RingMember> describeRing() throws DataAccessException;
<T> Collection<T> describeRing(HostMapper<T> hostMapper) throws DataAccessException;
}

532
src/main/java/org/springframework/cassandra/core/CassandraTemplate.java
View File

@@ -16,17 +16,23 @@
package org.springframework.cassandra.core;
import java.util.ArrayList;
import java.util.Collection;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Set;
import org.springframework.cassandra.support.CassandraAccessor;
import org.springframework.dao.DataAccessException;
import org.springframework.data.cassandra.core.CassandraDataTemplate;
import org.springframework.util.Assert;
import com.datastax.driver.core.BoundStatement;
import com.datastax.driver.core.ColumnDefinitions;
import com.datastax.driver.core.ColumnDefinitions.Definition;
import com.datastax.driver.core.Host;
import com.datastax.driver.core.Metadata;
import com.datastax.driver.core.PreparedStatement;
import com.datastax.driver.core.ResultSet;
import com.datastax.driver.core.ResultSetFuture;
import com.datastax.driver.core.Row;
@@ -35,10 +41,10 @@ import com.datastax.driver.core.exceptions.DriverException;
/**
* The CassandraTemplate is a Spring convenience wrapper for low level and explicit operations on the Cassandra
* Database. For working iwth POJOs, use the {@link CassandraDataTemplate}
* Database. For working with POJOs, use the {@link CassandraDataTemplate}
*
* @author Alex Shvid
* @author David Webb
* @author Matthew Adams
*/
public class CassandraTemplate extends CassandraAccessor implements CassandraOperations {
@@ -63,7 +69,7 @@ public class CassandraTemplate extends CassandraAccessor implements CassandraOpe
* @see org.springframework.data.cassandra.core.CassandraOperations#execute(org.springframework.data.cassandra.core.SessionCallback)
*/
@Override
public <T> T execute(SessionCallback<T> sessionCallback) {
public <T> T execute(SessionCallback<T> sessionCallback) throws DataAccessException {
return doExecute(sessionCallback);
}
@@ -71,65 +77,78 @@ public class CassandraTemplate extends CassandraAccessor implements CassandraOpe
* @see org.springframework.data.cassandra.core.CassandraOperations#execute(java.lang.String)
*/
@Override
public void execute(final String cql) {
doExecute(new SessionCallback<Object>() {
@Override
public Object doInSession(Session s) throws DataAccessException {
return s.execute(cql);
}
});
public void execute(final String cql) throws DataAccessException {
doExecute(cql);
}
/* (non-Javadoc)
* @see org.springframework.data.cassandra.core.CassandraOperations#executeQuery(java.lang.String)
* @see org.springframework.cassandra.core.CassandraOperations#query(java.lang.String, org.springframework.cassandra.core.FutureResultSetExtractor)
*/
@Override
public ResultSet executeQuery(final String query) {
return doExecute(new SessionCallback<ResultSet>() {
@Override
public ResultSet doInSession(Session s) throws DataAccessException {
return s.execute(query);
}
});
}
/* (non-Javadoc)
* @see org.springframework.data.cassandra.core.CassandraOperations#executeQueryAsync(java.lang.String)
*/
@Override
public ResultSetFuture executeQueryAsynchronously(final String query) {
return doExecute(new SessionCallback<ResultSetFuture>() {
public <T> T queryAsynchronously(final String cql, ResultSetFutureExtractor<T> rse) throws DataAccessException {
return rse.extractData(execute(new SessionCallback<ResultSetFuture>() {
@Override
public ResultSetFuture doInSession(Session s) throws DataAccessException {
return s.executeAsync(query);
return s.executeAsync(cql);
}
});
}));
}
/**
* Attempt to translate a Runtime Exception to a Spring Data Exception
*
* @param ex
* @return
/* (non-Javadoc)
* @see org.springframework.cassandra.core.CassandraOperations#query(java.lang.String, org.springframework.cassandra.core.ResultSetExtractor)
*/
public RuntimeException potentiallyConvertRuntimeException(RuntimeException ex) {
RuntimeException resolved = getExceptionTranslator().translateExceptionIfPossible(ex);
return resolved == null ? ex : resolved;
public <T> T query(String cql, ResultSetExtractor<T> rse) throws DataAccessException {
ResultSet rs = doExecute(cql);
return rse.extractData(rs);
}
/* (non-Javadoc)
* @see org.springframework.cassandra.core.CassandraOperations#query(java.lang.String, org.springframework.cassandra.core.RowCallbackHandler)
*/
public void query(String cql, RowCallbackHandler rch) throws DataAccessException {
process(doExecute(cql), rch);
}
/* (non-Javadoc)
* @see org.springframework.cassandra.core.CassandraOperations#query(java.lang.String, org.springframework.cassandra.core.RowMapper)
*/
public <T> List<T> query(String cql, RowMapper<T> rowMapper) throws DataAccessException {
return process(doExecute(cql), rowMapper);
}
/* (non-Javadoc)
* @see org.springframework.cassandra.core.CassandraOperations#queryForList(java.lang.String)
*/
public List<Map<String, Object>> queryForListOfMap(String cql) throws DataAccessException {
return processListOfMap(doExecute(cql));
}
/* (non-Javadoc)
* @see org.springframework.cassandra.core.CassandraOperations#queryForList(java.lang.String, java.lang.Class)
*/
public <T> List<T> queryForList(String cql, Class<T> elementType) throws DataAccessException {
return processList(doExecute(cql), elementType);
}
/* (non-Javadoc)
* @see org.springframework.cassandra.core.CassandraOperations#queryForMap(java.lang.String)
*/
public Map<String, Object> queryForMap(String cql) throws DataAccessException {
return processMap(doExecute(cql));
}
/* (non-Javadoc)
* @see org.springframework.cassandra.core.CassandraOperations#queryForObject(java.lang.String, java.lang.Class)
*/
public <T> T queryForObject(String cql, Class<T> requiredType) throws DataAccessException {
return processOne(doExecute(cql), requiredType);
}
/* (non-Javadoc)
* @see org.springframework.cassandra.core.CassandraOperations#queryForObject(java.lang.String, org.springframework.cassandra.core.RowMapper)
*/
public <T> T queryForObject(String cql, RowMapper<T> rowMapper) throws DataAccessException {
return processOne(doExecute(cql), rowMapper);
}
/**
@@ -147,83 +166,42 @@ public class CassandraTemplate extends CassandraAccessor implements CassandraOpe
return callback.doInSession(getSession());
} catch (DataAccessException e) {
throw potentiallyConvertRuntimeException(e);
throw throwTranslated(e);
}
}
/* (non-Javadoc)
* @see org.springframework.cassandra.core.CassandraOperations#query(java.lang.String, org.springframework.cassandra.core.ResultSetExtractor)
/**
* Execute a command at the Session Level
*
* @param callback
* @return
*/
public <T> T query(String cql, ResultSetExtractor<T> rse) throws DataAccessException {
try {
ResultSet rs = getSession().execute(cql);
return rse.extractData(rs);
} catch (DriverException dx) {
throw getExceptionTranslator().translateExceptionIfPossible(dx);
}
}
protected ResultSet doExecute(final String cql) {
/* (non-Javadoc)
* @see org.springframework.cassandra.core.CassandraOperations#query(java.lang.String, org.springframework.cassandra.core.RowCallbackHandler)
*/
public void query(String cql, RowCallbackHandler rch) throws DataAccessException {
try {
ResultSet rs = getSession().execute(cql);
for (Row row : rs.all()) {
rch.processRow(row);
return doExecute(new SessionCallback<ResultSet>() {
@Override
public ResultSet doInSession(Session s) throws DataAccessException {
return s.execute(cql);
}
} catch (DriverException dx) {
throw getExceptionTranslator().translateExceptionIfPossible(dx);
}
});
}
/* (non-Javadoc)
* @see org.springframework.cassandra.core.CassandraOperations#query(java.lang.String, org.springframework.cassandra.core.RowMapper)
/**
* Execute a command at the Session Level
*
* @param callback
* @return
*/
public <T> List<T> query(String cql, RowMapper<T> rowMapper) throws DataAccessException {
try {
ResultSet rs = getSession().execute(cql);
int i = 0;
List<T> mappedRows = new ArrayList<T>();
for (Row row : rs.all()) {
mappedRows.add(rowMapper.mapRow(row, i++));
protected ResultSet doExecute(final BoundStatement bs) {
return doExecute(new SessionCallback<ResultSet>() {
@Override
public ResultSet doInSession(Session s) throws DataAccessException {
return s.execute(bs);
}
return mappedRows;
} catch (DriverException dx) {
throw getExceptionTranslator().translateExceptionIfPossible(dx);
}
}
/* (non-Javadoc)
* @see org.springframework.cassandra.core.CassandraOperations#queryForObject(java.lang.String, org.springframework.cassandra.core.RowMapper)
*/
public <T> T queryForObject(String cql, RowMapper<T> rowMapper) throws DataAccessException {
try {
ResultSet rs = getSession().execute(cql);
List<Row> rows = rs.all();
Assert.notNull(rows, "null row list returned from query");
Assert.isTrue(rows.size() == 1, "row list has " + rows.size() + " rows instead of one");
return rowMapper.mapRow(rows.get(0), 0);
} catch (DriverException dx) {
throw getExceptionTranslator().translateExceptionIfPossible(dx);
}
}
/* (non-Javadoc)
* @see org.springframework.cassandra.core.CassandraOperations#queryForObject(java.lang.String, java.lang.Class)
*/
@SuppressWarnings("unchecked")
public <T> T queryForObject(String cql, Class<T> requiredType) throws DataAccessException {
ResultSet rs = getSession().execute(cql);
if (rs == null) {
return null;
}
Row row = rs.one();
if (row == null) {
return null;
}
return (T) firstColumnToObject(row);
});
}
/**
@@ -238,17 +216,6 @@ public class CassandraTemplate extends CassandraAccessor implements CassandraOpe
return cols.getType(0).deserialize(row.getBytesUnsafe(0));
}
/* (non-Javadoc)
* @see org.springframework.cassandra.core.CassandraOperations#queryForMap(java.lang.String)
*/
public Map<String, Object> queryForMap(String cql) throws DataAccessException {
ResultSet rs = getSession().execute(cql);
if (rs == null) {
return null;
}
return toMap(rs.one());
}
/**
* @param row
* @return
@@ -270,12 +237,145 @@ public class CassandraTemplate extends CassandraAccessor implements CassandraOpe
}
/* (non-Javadoc)
* @see org.springframework.cassandra.core.CassandraOperations#queryForList(java.lang.String, java.lang.Class)
* @see org.springframework.cassandra.core.CassandraOperations#describeRing()
*/
@Override
public List<RingMember> describeRing() throws DataAccessException {
return new ArrayList<RingMember>(describeRing(new RingMemberHostMapper()));
}
/**
* Pulls the list of Hosts for the current Session
*
* @return
*/
private Set<Host> getHosts() {
/*
* Get the cluster metadata for this session
*/
Metadata clusterMetadata = doExecute(new SessionCallback<Metadata>() {
@Override
public Metadata doInSession(Session s) throws DataAccessException {
return s.getCluster().getMetadata();
}
});
/*
* Get all hosts in the cluster
*/
Set<Host> hosts = clusterMetadata.getAllHosts();
return hosts;
}
/* (non-Javadoc)
* @see org.springframework.cassandra.core.CassandraOperations#describeRing(org.springframework.cassandra.core.HostMapper)
*/
@Override
public <T> Collection<T> describeRing(HostMapper<T> hostMapper) throws DataAccessException {
Set<Host> hosts = getHosts();
return hostMapper.mapHosts(hosts);
}
/* (non-Javadoc)
* @see org.springframework.cassandra.core.CassandraOperations#executeAsynchronously(java.lang.String)
*/
@Override
public void executeAsynchronously(final String cql) throws DataAccessException {
execute(new SessionCallback<Object>() {
@Override
public Object doInSession(Session s) throws DataAccessException {
return s.executeAsync(cql);
}
});
}
/* (non-Javadoc)
* @see org.springframework.cassandra.core.CassandraOperations#process(com.datastax.driver.core.ResultSet, org.springframework.cassandra.core.RowCallbackHandler)
*/
@Override
public void process(ResultSet resultSet, RowCallbackHandler rch) throws DataAccessException {
try {
for (Row row : resultSet.all()) {
rch.processRow(row);
}
} catch (DriverException dx) {
throwTranslated(dx);
}
}
/* (non-Javadoc)
* @see org.springframework.cassandra.core.CassandraOperations#process(com.datastax.driver.core.ResultSet, org.springframework.cassandra.core.RowMapper)
*/
@Override
public <T> List<T> process(ResultSet resultSet, RowMapper<T> rowMapper) throws DataAccessException {
List<T> mappedRows = new ArrayList<T>();
try {
int i = 0;
for (Row row : resultSet.all()) {
mappedRows.add(rowMapper.mapRow(row, i++));
}
} catch (DriverException dx) {
throwTranslated(dx);
}
return mappedRows;
}
/* (non-Javadoc)
* @see org.springframework.cassandra.core.CassandraOperations#processOne(com.datastax.driver.core.ResultSet, org.springframework.cassandra.core.RowMapper)
*/
@Override
public <T> T processOne(ResultSet resultSet, RowMapper<T> rowMapper) throws DataAccessException {
T row = null;
Assert.notNull(resultSet, "ResultSet cannot be null");
try {
List<Row> rows = resultSet.all();
Assert.notNull(rows, "null row list returned from query");
Assert.isTrue(rows.size() == 1, "row list has " + rows.size() + " rows instead of one");
row = rowMapper.mapRow(rows.get(0), 0);
} catch (DriverException dx) {
throwTranslated(dx);
}
return row;
}
/* (non-Javadoc)
* @see org.springframework.cassandra.core.CassandraOperations#processOne(com.datastax.driver.core.ResultSet, java.lang.Class)
*/
@SuppressWarnings("unchecked")
public <T> List<T> queryForList(String cql, Class<T> elementType) throws DataAccessException {
ResultSet rs = getSession().execute(cql);
List<Row> rows = rs.all();
@Override
public <T> T processOne(ResultSet resultSet, Class<T> requiredType) throws DataAccessException {
if (resultSet == null) {
return null;
}
Row row = resultSet.one();
if (row == null) {
return null;
}
return (T) firstColumnToObject(row);
}
/* (non-Javadoc)
* @see org.springframework.cassandra.core.CassandraOperations#processMap(com.datastax.driver.core.ResultSet)
*/
@Override
public Map<String, Object> processMap(ResultSet resultSet) throws DataAccessException {
if (resultSet == null) {
return null;
}
return toMap(resultSet.one());
}
/* (non-Javadoc)
* @see org.springframework.cassandra.core.CassandraOperations#processList(com.datastax.driver.core.ResultSet, java.lang.Class)
*/
@Override
public <T> List<T> processList(ResultSet resultSet, Class<T> elementType) throws DataAccessException {
List<Row> rows = resultSet.all();
List<T> list = new ArrayList<T>(rows.size());
for (Row row : rows) {
list.add((T) firstColumnToObject(row));
@@ -284,11 +384,11 @@ public class CassandraTemplate extends CassandraAccessor implements CassandraOpe
}
/* (non-Javadoc)
* @see org.springframework.cassandra.core.CassandraOperations#queryForList(java.lang.String)
* @see org.springframework.cassandra.core.CassandraOperations#processListOfMap(com.datastax.driver.core.ResultSet)
*/
public List<Map<String, Object>> queryForList(String cql) throws DataAccessException {
ResultSet rs = getSession().execute(cql);
List<Row> rows = rs.all();
@Override
public List<Map<String, Object>> processListOfMap(ResultSet resultSet) throws DataAccessException {
List<Row> rows = resultSet.all();
List<Map<String, Object>> list = new ArrayList<Map<String, Object>>(rows.size());
for (Row row : rows) {
list.add(toMap(row));
@@ -296,4 +396,160 @@ public class CassandraTemplate extends CassandraAccessor implements CassandraOpe
return list;
}
}
/**
* Attempt to translate a Runtime Exception to a Spring Data Exception
*
* @param ex
* @return
*/
protected RuntimeException throwTranslated(RuntimeException ex) {
RuntimeException resolved = getExceptionTranslator().translateExceptionIfPossible(ex);
return resolved == null ? ex : resolved;
}
/* (non-Javadoc)
* @see org.springframework.cassandra.core.CassandraOperations#execute(org.springframework.cassandra.core.PreparedStatementCreator, org.springframework.cassandra.core.PreparedStatementCallback)
*/
@Override
public <T> T execute(PreparedStatementCreator psc, PreparedStatementCallback<T> action) {
try {
PreparedStatement ps = psc.createPreparedStatement(getSession());
return action.doInPreparedStatement(ps);
} catch (DriverException dx) {
throwTranslated(dx);
}
return null;
}
/* (non-Javadoc)
* @see org.springframework.cassandra.core.CassandraOperations#execute(java.lang.String, org.springframework.cassandra.core.PreparedStatementCallback)
*/
@Override
public <T> T execute(String cql, PreparedStatementCallback<T> action) {
return execute(new SimplePreparedStatementCreator(cql), action);
}
/* (non-Javadoc)
* @see org.springframework.cassandra.core.CassandraOperations#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.CassandraOperations#query(org.springframework.cassandra.core.PreparedStatementCreator, org.springframework.cassandra.core.RowCallbackHandler)
*/
@Override
public void query(PreparedStatementCreator psc, RowCallbackHandler rch) throws DataAccessException {
query(psc, null, rch);
}
/* (non-Javadoc)
* @see org.springframework.cassandra.core.CassandraOperations#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, null, rowMapper);
}
/* (non-Javadoc)
* @see org.springframework.cassandra.core.CassandraOperations#query(org.springframework.cassandra.core.PreparedStatementCreator, org.springframework.cassandra.core.PreparedStatementSetter, org.springframework.cassandra.core.ResultSetExtractor)
*/
public <T> T query(PreparedStatementCreator psc, final PreparedStatementBinder pss, final ResultSetExtractor<T> rse)
throws DataAccessException {
Assert.notNull(rse, "ResultSetExtractor must not be null");
logger.debug("Executing prepared CQL query");
return execute(psc, new PreparedStatementCallback<T>() {
public T doInPreparedStatement(PreparedStatement ps) throws DriverException {
ResultSet rs = null;
BoundStatement bs = null;
if (pss != null) {
bs = pss.bindValues(ps);
} else {
bs = ps.bind();
}
rs = doExecute(bs);
return rse.extractData(rs);
}
});
}
/* (non-Javadoc)
* @see org.springframework.cassandra.core.CassandraOperations#query(java.lang.String, org.springframework.cassandra.core.PreparedStatementSetter, org.springframework.cassandra.core.ResultSetExtractor)
*/
@Override
public <T> T query(String cql, PreparedStatementBinder pss, ResultSetExtractor<T> rse) throws DataAccessException {
return query(new SimplePreparedStatementCreator(cql), pss, rse);
}
/* (non-Javadoc)
* @see org.springframework.cassandra.core.CassandraOperations#query(java.lang.String, org.springframework.cassandra.core.PreparedStatementSetter, org.springframework.cassandra.core.RowCallbackHandler)
*/
@Override
public void query(String cql, PreparedStatementBinder pss, RowCallbackHandler rch) throws DataAccessException {
query(new SimplePreparedStatementCreator(cql), pss, rch);
}
/* (non-Javadoc)
* @see org.springframework.cassandra.core.CassandraOperations#query(java.lang.String, org.springframework.cassandra.core.PreparedStatementSetter, org.springframework.cassandra.core.RowMapper)
*/
@Override
public <T> List<T> query(String cql, PreparedStatementBinder pss, RowMapper<T> rowMapper) throws DataAccessException {
return query(new SimplePreparedStatementCreator(cql), pss, rowMapper);
}
/* (non-Javadoc)
* @see org.springframework.cassandra.core.CassandraOperations#query(org.springframework.cassandra.core.PreparedStatementCreator, org.springframework.cassandra.core.PreparedStatementBinder, org.springframework.cassandra.core.RowCallbackHandler)
*/
@Override
public void query(PreparedStatementCreator psc, final PreparedStatementBinder pss, final RowCallbackHandler rch)
throws DataAccessException {
Assert.notNull(rch, "RowCallbackHandler must not be null");
logger.debug("Executing prepared CQL query");
execute(psc, new PreparedStatementCallback<Object>() {
public Object doInPreparedStatement(PreparedStatement ps) throws DriverException {
ResultSet rs = null;
BoundStatement bs = null;
if (pss != null) {
bs = pss.bindValues(ps);
} else {
bs = ps.bind();
}
rs = doExecute(bs);
process(rs, rch);
return null;
}
});
}
/* (non-Javadoc)
* @see org.springframework.cassandra.core.CassandraOperations#query(org.springframework.cassandra.core.PreparedStatementCreator, org.springframework.cassandra.core.PreparedStatementBinder, org.springframework.cassandra.core.RowMapper)
*/
@Override
public <T> List<T> query(PreparedStatementCreator psc, final PreparedStatementBinder pss, final RowMapper<T> rowMapper)
throws DataAccessException {
Assert.notNull(rowMapper, "RowMapper must not be null");
logger.debug("Executing prepared CQL query");
return execute(psc, new PreparedStatementCallback<List<T>>() {
public List<T> doInPreparedStatement(PreparedStatement ps) throws DriverException {
ResultSet rs = null;
BoundStatement bs = null;
if (pss != null) {
bs = pss.bindValues(ps);
} else {
bs = ps.bind();
}
rs = doExecute(bs);
return process(rs, rowMapper);
}
});
}
}

26
src/main/java/org/springframework/cassandra/core/CqlProvider.java Normal file
View File

@@ -0,0 +1,26 @@
/*
* Copyright 2011-2013 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.cassandra.core;
/**
* @author David Webb
*
*/
public interface CqlProvider {
String getCql();
}

13
src/main/java/org/springframework/cassandra/core/HostMapper.java Normal file
View File

@@ -0,0 +1,13 @@
package org.springframework.cassandra.core;
import java.util.Collection;
import java.util.Set;
import com.datastax.driver.core.Host;
import com.datastax.driver.core.exceptions.DriverException;
public interface HostMapper<T> {
Collection<T> mapHosts(Set<Host> host) throws DriverException;
}

30
src/main/java/org/springframework/cassandra/core/PreparedStatementBinder.java Normal file
View File

@@ -0,0 +1,30 @@
/*
* Copyright 2011-2013 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.cassandra.core;
import com.datastax.driver.core.BoundStatement;
import com.datastax.driver.core.PreparedStatement;
import com.datastax.driver.core.exceptions.DriverException;
/**
* @author David Webb
*
*/
public interface PreparedStatementBinder {
BoundStatement bindValues(PreparedStatement ps) throws DriverException;
}

31
src/main/java/org/springframework/cassandra/core/PreparedStatementCallback.java Normal file
View File

@@ -0,0 +1,31 @@
/*
* Copyright 2011-2013 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.cassandra.core;
import org.springframework.dao.DataAccessException;
import com.datastax.driver.core.PreparedStatement;
import com.datastax.driver.core.exceptions.DriverException;
/**
* @author David Webb
*
*/
public interface PreparedStatementCallback<T> {
T doInPreparedStatement(PreparedStatement ps) throws DriverException, DataAccessException;
}

41
src/main/java/org/springframework/cassandra/core/PreparedStatementCreator.java Normal file
View File

@@ -0,0 +1,41 @@
/*
* Copyright 2011-2013 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.cassandra.core;
import com.datastax.driver.core.PreparedStatement;
import com.datastax.driver.core.Session;
import com.datastax.driver.core.exceptions.DriverException;
/**
* Creates a PreparedStatement for the usage with the DataStax Java Driver
*
* @author David Webb
*
*/
public interface PreparedStatementCreator {
/**
* Create a statement in this session. Allows implementations to use PreparedStatements. The CassandraTemlate will
* attempt to cache the PreparedStatement for future use without the overhead of re-preparing on the entire cluster.
*
* @param session Session to use to create statement
* @return a prepared statement
* @throws DriverException there is no need to catch DriverException that may be thrown in the implementation of this
* method. The CassandraTemlate class will handle them.
*/
PreparedStatement createPreparedStatement(Session session) throws DriverException;
}

73
src/main/java/org/springframework/cassandra/core/PreparedStatementCreatorImpl.java Normal file
View File

@@ -0,0 +1,73 @@
/*
* Copyright 2011-2013 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.cassandra.core;
import java.util.List;
import com.datastax.driver.core.BoundStatement;
import com.datastax.driver.core.PreparedStatement;
import com.datastax.driver.core.Session;
import com.datastax.driver.core.exceptions.DriverException;
/**
* @author David Webb
*
*/
public class PreparedStatementCreatorImpl implements PreparedStatementCreator, CqlProvider, PreparedStatementBinder {
private final String cql;
private List<Object> values;
public PreparedStatementCreatorImpl(String cql) {
this.cql = cql;
}
public PreparedStatementCreatorImpl(String cql, List<Object> values) {
this.cql = cql;
this.values = values;
}
/* (non-Javadoc)
* @see org.springframework.cassandra.core.PreparedStatementSetter#setValues(com.datastax.driver.core.PreparedStatement)
*/
@Override
public BoundStatement bindValues(PreparedStatement ps) throws DriverException {
// Nothing to set if there are no values
if (values == null) {
return null;
}
return ps.bind(values.toArray());
}
/* (non-Javadoc)
* @see org.springframework.cassandra.core.CqlProvider#getCql()
*/
@Override
public String getCql() {
return this.cql;
}
/* (non-Javadoc)
* @see org.springframework.cassandra.core.PreparedStatementCreator#createPreparedStatement(com.datastax.driver.core.Session)
*/
@Override
public PreparedStatement createPreparedStatement(Session session) throws DriverException {
return session.prepare(this.cql);
}
}

11
src/main/java/org/springframework/cassandra/core/ResultSetFutureExtractor.java Normal file
View File

@@ -0,0 +1,11 @@
package org.springframework.cassandra.core;
import org.springframework.dao.DataAccessException;
import com.datastax.driver.core.ResultSetFuture;
import com.datastax.driver.core.exceptions.DriverException;
public interface ResultSetFutureExtractor<T> {
T extractData(ResultSetFuture rs) throws DriverException, DataAccessException;
}

15
src/main/java/org/springframework/data/cassandra/core/RingMember.java → src/main/java/org/springframework/cassandra/core/RingMember.java
View File

@@ -13,7 +13,7 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.data.cassandra.core;
package org.springframework.cassandra.core;
import java.io.Serializable;
@@ -25,18 +25,13 @@ import com.datastax.driver.core.Host;
*/
public final class RingMember implements Serializable {
private static final long serialVersionUID = 1345346346L;
/*
* Ring attributes
*/
public final String hostName;
public final String address;
public final String DC;
public final String rack;
// public final String status;
// public final String state;
public String hostName;
public String address;
public String DC;
public String rack;
public RingMember(Host h) {
this.hostName = h.getAddress().getHostName();

54
src/main/java/org/springframework/cassandra/core/RingMemberHostMapper.java Normal file
View File

@@ -0,0 +1,54 @@
/*
* Copyright 2011-2013 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.cassandra.core;
import java.util.ArrayList;
import java.util.List;
import java.util.Set;
import org.springframework.util.Assert;
import com.datastax.driver.core.Host;
import com.datastax.driver.core.exceptions.DriverException;
/**
* @author David Webb
* @param <T>
*
*/
public class RingMemberHostMapper implements HostMapper<RingMember> {
/* (non-Javadoc)
* @see org.springframework.cassandra.core.HostMapper#mapHosts(java.util.Set)
*/
@Override
public List<RingMember> mapHosts(Set<Host> hosts) throws DriverException {
List<RingMember> members = new ArrayList<RingMember>();
Assert.notNull(hosts);
Assert.notEmpty(hosts);
RingMember r = null;
for (Host host : hosts) {
r = new RingMember(host);
members.add(r);
}
return members;
}
}

2
src/main/java/org/springframework/cassandra/core/SessionFactoryBean.java
View File

@@ -64,7 +64,7 @@ public class SessionFactoryBean implements FactoryBean<Session>, InitializingBea
* @see org.springframework.beans.factory.FactoryBean#getObject()
*/
@Override
public Session getObject() throws Exception {
public Session getObject() {
return keyspace.getSession();
}

58
src/main/java/org/springframework/cassandra/core/SimplePreparedStatementCreator.java Normal file
View File

@@ -0,0 +1,58 @@
/*
* Copyright 2011-2013 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.cassandra.core;
import org.springframework.util.Assert;
import com.datastax.driver.core.PreparedStatement;
import com.datastax.driver.core.Session;
import com.datastax.driver.core.exceptions.DriverException;
/**
* @author David Webb
*
*/
public class SimplePreparedStatementCreator implements PreparedStatementCreator, CqlProvider {
private final String cql;
/**
* Create a PreparedStatementCreator from the provided CQL.
*
* @param cql
*/
public SimplePreparedStatementCreator(String cql) {
Assert.notNull(cql, "CQL is required to create a PreparedStatement");
this.cql = cql;
}
/* (non-Javadoc)
* @see org.springframework.cassandra.core.CqlProvider#getCql()
*/
@Override
public String getCql() {
return this.cql;
}
/* (non-Javadoc)
* @see org.springframework.cassandra.core.PreparedStatementCreator#createPreparedStatement(com.datastax.driver.core.Session)
*/
@Override
public PreparedStatement createPreparedStatement(Session session) throws DriverException {
return session.prepare(this.cql);
}
}

7
src/main/java/org/springframework/data/cassandra/core/CassandraDataOperations.java
View File

@@ -33,13 +33,6 @@ import com.datastax.driver.core.querybuilder.Select;
*/
public interface CassandraDataOperations {
/**
* Describe the current Ring
*
* @return The list of ring tokens that are active in the cluster
*/
List<RingMember> describeRing();
/**
* The table name used for the specified class by this template.
*

53
src/main/java/org/springframework/data/cassandra/core/CassandraDataTemplate.java
View File

@@ -38,8 +38,6 @@ import org.springframework.data.cassandra.util.CqlUtils;
import org.springframework.data.mapping.context.MappingContext;
import org.springframework.util.Assert;
import com.datastax.driver.core.Host;
import com.datastax.driver.core.Metadata;
import com.datastax.driver.core.Query;
import com.datastax.driver.core.ResultSet;
import com.datastax.driver.core.Row;
@@ -362,55 +360,6 @@ public class CassandraDataTemplate extends CassandraTemplate implements Cassandr
deleteAsynchronously(entity, tableName, options.toMap());
}
/* (non-Javadoc)
* @see org.springframework.data.cassandra.core.CassandraOperations#describeRing()
*/
@Override
public List<RingMember> describeRing() {
/*
* Initialize the return variable
*/
List<RingMember> ring = new ArrayList<RingMember>();
/*
* Get the cluster metadata for this session
*/
Metadata clusterMetadata = doExecute(new SessionCallback<Metadata>() {
@Override
public Metadata doInSession(Session s) throws DataAccessException {
return s.getCluster().getMetadata();
}
});
/*
* Get all hosts in the cluster
*/
Set<Host> hosts = clusterMetadata.getAllHosts();
/*
* Loop variables
*/
RingMember member = null;
/*
* Populate Ring with Host Metadata
*/
for (Host h : hosts) {
member = new RingMember(h);
ring.add(member);
}
/*
* Return
*/
return ring;
}
/**
* @param entityClass
* @return
@@ -1218,7 +1167,7 @@ public class CassandraDataTemplate extends CassandraTemplate implements Cassandr
return callback.doInSession(getSession());
} catch (DataAccessException e) {
throw potentiallyConvertRuntimeException(e);
throw throwTranslated(e);
}
}

20
src/test/java/org/springframework/data/cassandra/config/TestConfig.java
View File

@@ -1,5 +1,8 @@
package org.springframework.data.cassandra.config;
import org.springframework.cassandra.core.CassandraOperations;
import org.springframework.cassandra.core.CassandraTemplate;
import org.springframework.cassandra.core.SessionFactoryBean;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.data.cassandra.core.CassandraDataOperations;
@@ -34,9 +37,7 @@ public class TestConfig extends AbstractCassandraConfiguration {
public Cluster cluster() {
Builder builder = Cluster.builder();
builder.addContactPoint("127.0.0.1");
return builder.build();
}
@@ -51,6 +52,21 @@ public class TestConfig extends AbstractCassandraConfiguration {
}
@Bean
public SessionFactoryBean sessionFactoryBean() {
SessionFactoryBean bean = new SessionFactoryBean(keyspaceFactoryBean().getObject());
return bean;
}
@Bean
public CassandraOperations cassandraTemplate() {
CassandraOperations template = new CassandraTemplate(sessionFactoryBean().getObject());
return template;
}
@Bean
public CassandraDataOperations cassandraDataTemplate() {

19
src/test/java/org/springframework/data/cassandra/template/CassandraDataOperationsTest.java
View File

@@ -15,8 +15,6 @@
*/
package org.springframework.data.cassandra.template;
import static org.junit.Assert.assertNotNull;
import java.io.IOException;
import java.util.ArrayList;
import java.util.HashMap;
@@ -47,7 +45,6 @@ import org.springframework.data.cassandra.core.CassandraDataOperations;
import org.springframework.data.cassandra.core.ConsistencyLevel;
import org.springframework.data.cassandra.core.QueryOptions;
import org.springframework.data.cassandra.core.RetryPolicy;
import org.springframework.data.cassandra.core.RingMember;
import org.springframework.data.cassandra.table.Book;
import org.springframework.test.context.ContextConfiguration;
import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
@@ -94,22 +91,6 @@ public class CassandraDataOperationsTest {
dataLoader.load(new ClassPathYamlDataSet("cassandra-keyspace.yaml"));
}
@Test
public void ringTest() {
List<RingMember> ring = cassandraDataTemplate.describeRing();
/*
* There must be 1 node in the cluster if the embedded server is
* running.
*/
assertNotNull(ring);
for (RingMember h : ring) {
log.info(h.address);
}
}
@Test
public void insertTest() {

159
src/test/java/org/springframework/data/cassandra/template/CassandraOperationsTest.java Normal file
View File

@@ -0,0 +1,159 @@
/*
* Copyright 2011-2013 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.data.cassandra.template;
import static org.junit.Assert.assertNotNull;
import java.io.IOException;
import java.util.Collection;
import java.util.LinkedList;
import java.util.List;
import java.util.Set;
import junit.framework.Assert;
import org.apache.cassandra.exceptions.ConfigurationException;
import org.apache.thrift.transport.TTransportException;
import org.cassandraunit.CassandraCQLUnit;
import org.cassandraunit.DataLoader;
import org.cassandraunit.dataset.cql.ClassPathCQLDataSet;
import org.cassandraunit.dataset.yaml.ClassPathYamlDataSet;
import org.cassandraunit.utils.EmbeddedCassandraServerHelper;
import org.junit.After;
import org.junit.AfterClass;
import org.junit.BeforeClass;
import org.junit.Rule;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.cassandra.core.CassandraOperations;
import org.springframework.cassandra.core.HostMapper;
import org.springframework.cassandra.core.RingMember;
import org.springframework.data.cassandra.config.TestConfig;
import org.springframework.test.context.ContextConfiguration;
import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
import org.springframework.test.context.support.AnnotationConfigContextLoader;
import com.datastax.driver.core.Host;
import com.datastax.driver.core.exceptions.DriverException;
/**
* Unit Tests for CassandraTemplate
*
* @author David Webb
*
*/
@RunWith(SpringJUnit4ClassRunner.class)
@ContextConfiguration(classes = { TestConfig.class }, loader = AnnotationConfigContextLoader.class)
public class CassandraOperationsTest {
/**
* @author David Webb
*
*/
public class MyHost {
public String someName;
}
@Autowired
private CassandraOperations cassandraTemplate;
private static Logger log = LoggerFactory.getLogger(CassandraOperationsTest.class);
private final static String CASSANDRA_CONFIG = "cassandra.yaml";
private final static String KEYSPACE_NAME = "test";
private final static String CASSANDRA_HOST = "localhost";
private final static int CASSANDRA_NATIVE_PORT = 9042;
private final static int CASSANDRA_THRIFT_PORT = 9160;
@Rule
public CassandraCQLUnit cassandraCQLUnit = new CassandraCQLUnit(new ClassPathCQLDataSet("cql-dataload.cql",
KEYSPACE_NAME), CASSANDRA_CONFIG, CASSANDRA_HOST, CASSANDRA_NATIVE_PORT);
@BeforeClass
public static void startCassandra() throws IOException, TTransportException, ConfigurationException,
InterruptedException {
EmbeddedCassandraServerHelper.startEmbeddedCassandra(CASSANDRA_CONFIG);
/*
* Load data file to creat the test keyspace before we init the template
*/
DataLoader dataLoader = new DataLoader("Test Cluster", CASSANDRA_HOST + ":" + CASSANDRA_THRIFT_PORT);
dataLoader.load(new ClassPathYamlDataSet("cassandra-keyspace.yaml"));
}
@Test
public void ringTest() {
List<RingMember> ring = cassandraTemplate.describeRing();
/*
* There must be 1 node in the cluster if the embedded server is
* running.
*/
assertNotNull(ring);
for (RingMember h : ring) {
log.info("ringTest Host -> " + h.address);
}
}
@Test
public void hostMapperTest() {
List<MyHost> ring = (List<MyHost>) cassandraTemplate.describeRing(new HostMapper<MyHost>() {
@Override
public Collection<MyHost> mapHosts(Set<Host> host) throws DriverException {
List<MyHost> list = new LinkedList<CassandraOperationsTest.MyHost>();
for (Host h : host) {
MyHost mh = new MyHost();
mh.someName = h.getAddress().getCanonicalHostName();
list.add(mh);
}
return list;
}
});
assertNotNull(ring);
Assert.assertTrue(ring.size() > 0);
for (MyHost h : ring) {
log.info("hostMapperTest Host -> " + h.someName);
}
}
@After
public void clearCassandra() {
EmbeddedCassandraServerHelper.cleanEmbeddedCassandra();
}
@AfterClass
public static void stopCassandra() {
EmbeddedCassandraServerHelper.stopEmbeddedCassandra();
}
}