diff --git a/src/main/java/org/springframework/cassandra/core/CassandraOperations.java b/src/main/java/org/springframework/cassandra/core/CassandraOperations.java index 730f0b078..fbd3653d6 100644 --- a/src/main/java/org/springframework/cassandra/core/CassandraOperations.java +++ b/src/main/java/org/springframework/cassandra/core/CassandraOperations.java @@ -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 execute(SessionCallback sessionCallback); - - void execute(final String cql); - - RuntimeException potentiallyConvertRuntimeException(RuntimeException ex); - - T query(String cql, ResultSetExtractor rse) throws DataAccessException; - - void query(String cql, RowCallbackHandler rch) throws DataAccessException; - - List query(String cql, RowMapper rowMapper) throws DataAccessException; - - T queryForObject(String cql, RowMapper rowMapper) throws DataAccessException; - - T queryForObject(String cql, Class requiredType) throws DataAccessException; - - Map queryForMap(String cql) throws DataAccessException; - - List queryForList(String cql, Class elementType) throws DataAccessException; - - List> 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 execute() or executeAsync() the operation. * - * @param cql must not be {@literal null}. + * @param sessionCallback * @return */ - ResultSet executeQuery(final String cql); + T execute(SessionCallback 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 query(final String cql, ResultSetExtractor 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 queryAsynchronously(final String cql, ResultSetFutureExtractor rse) throws DataAccessException; + + void query(final String cql, RowCallbackHandler rch) throws DataAccessException; + + void process(ResultSet resultSet, RowCallbackHandler rch) throws DataAccessException; + + List query(final String cql, RowMapper rowMapper) throws DataAccessException; + + List process(ResultSet resultSet, RowMapper rowMapper) throws DataAccessException; + + T queryForObject(final String cql, RowMapper rowMapper) throws DataAccessException; + + T processOne(ResultSet resultSet, RowMapper rowMapper) throws DataAccessException; + + T queryForObject(final String cql, Class requiredType) throws DataAccessException; + + T processOne(ResultSet resultSet, Class requiredType) throws DataAccessException; + + Map queryForMap(final String cql) throws DataAccessException; + + Map processMap(ResultSet resultSet) throws DataAccessException; + + List queryForList(final String cql, Class elementType) throws DataAccessException; + + List processList(ResultSet resultSet, Class elementType) throws DataAccessException; + + List> queryForListOfMap(final String cql) throws DataAccessException; + + List> processListOfMap(ResultSet resultSet) throws DataAccessException; + + T execute(String cql, PreparedStatementCallback action) throws DataAccessException; + + T execute(PreparedStatementCreator psc, PreparedStatementCallback action) throws DataAccessException; + + T query(final String cql, PreparedStatementBinder pss, ResultSetExtractor rse) throws DataAccessException; + + void query(final String cql, PreparedStatementBinder pss, RowCallbackHandler rch) throws DataAccessException; + + List query(final String cql, PreparedStatementBinder pss, RowMapper rowMapper) throws DataAccessException; + + T query(PreparedStatementCreator psc, ResultSetExtractor rse) throws DataAccessException; + + void query(PreparedStatementCreator psc, RowCallbackHandler rch) throws DataAccessException; + + List query(PreparedStatementCreator psc, RowMapper rowMapper) throws DataAccessException; + + T query(PreparedStatementCreator psc, final PreparedStatementBinder pss, final ResultSetExtractor rse) + throws DataAccessException; + + void query(PreparedStatementCreator psc, final PreparedStatementBinder pss, final RowCallbackHandler rch) + throws DataAccessException; + + List query(PreparedStatementCreator psc, final PreparedStatementBinder pss, final RowMapper rowMapper) + throws DataAccessException; + + /** + * Describe the current Ring + * + * @return The list of ring tokens that are active in the cluster + */ + List describeRing() throws DataAccessException; + + Collection describeRing(HostMapper hostMapper) throws DataAccessException; } diff --git a/src/main/java/org/springframework/cassandra/core/CassandraTemplate.java b/src/main/java/org/springframework/cassandra/core/CassandraTemplate.java index 36122f9f8..5ed7e8aba 100644 --- a/src/main/java/org/springframework/cassandra/core/CassandraTemplate.java +++ b/src/main/java/org/springframework/cassandra/core/CassandraTemplate.java @@ -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 execute(SessionCallback sessionCallback) { + public T execute(SessionCallback 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() { - - @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() { - - @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() { - + public T queryAsynchronously(final String cql, ResultSetFutureExtractor rse) throws DataAccessException { + return rse.extractData(execute(new SessionCallback() { @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 query(String cql, ResultSetExtractor 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 List query(String cql, RowMapper rowMapper) throws DataAccessException { + return process(doExecute(cql), rowMapper); + } + + /* (non-Javadoc) + * @see org.springframework.cassandra.core.CassandraOperations#queryForList(java.lang.String) + */ + public List> 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 List queryForList(String cql, Class elementType) throws DataAccessException { + return processList(doExecute(cql), elementType); + } + + /* (non-Javadoc) + * @see org.springframework.cassandra.core.CassandraOperations#queryForMap(java.lang.String) + */ + public Map 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 queryForObject(String cql, Class 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 queryForObject(String cql, RowMapper 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 query(String cql, ResultSetExtractor 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() { + + @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 List query(String cql, RowMapper rowMapper) throws DataAccessException { - try { - ResultSet rs = getSession().execute(cql); - int i = 0; - List mappedRows = new ArrayList(); - for (Row row : rs.all()) { - mappedRows.add(rowMapper.mapRow(row, i++)); + protected ResultSet doExecute(final BoundStatement bs) { + + return doExecute(new SessionCallback() { + + @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 queryForObject(String cql, RowMapper rowMapper) throws DataAccessException { - try { - ResultSet rs = getSession().execute(cql); - List 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 queryForObject(String cql, Class 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 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 describeRing() throws DataAccessException { + return new ArrayList(describeRing(new RingMemberHostMapper())); + } + + /** + * Pulls the list of Hosts for the current Session + * + * @return + */ + private Set getHosts() { + + /* + * Get the cluster metadata for this session + */ + Metadata clusterMetadata = doExecute(new SessionCallback() { + + @Override + public Metadata doInSession(Session s) throws DataAccessException { + return s.getCluster().getMetadata(); + } + + }); + + /* + * Get all hosts in the cluster + */ + Set hosts = clusterMetadata.getAllHosts(); + + return hosts; + + } + + /* (non-Javadoc) + * @see org.springframework.cassandra.core.CassandraOperations#describeRing(org.springframework.cassandra.core.HostMapper) + */ + @Override + public Collection describeRing(HostMapper hostMapper) throws DataAccessException { + Set 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() { + @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 List process(ResultSet resultSet, RowMapper rowMapper) throws DataAccessException { + List mappedRows = new ArrayList(); + 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 processOne(ResultSet resultSet, RowMapper rowMapper) throws DataAccessException { + T row = null; + Assert.notNull(resultSet, "ResultSet cannot be null"); + try { + List 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 List queryForList(String cql, Class elementType) throws DataAccessException { - ResultSet rs = getSession().execute(cql); - List rows = rs.all(); + @Override + public T processOne(ResultSet resultSet, Class 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 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 List processList(ResultSet resultSet, Class elementType) throws DataAccessException { + List rows = resultSet.all(); List list = new ArrayList(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> queryForList(String cql) throws DataAccessException { - ResultSet rs = getSession().execute(cql); - List rows = rs.all(); + @Override + public List> processListOfMap(ResultSet resultSet) throws DataAccessException { + List rows = resultSet.all(); List> list = new ArrayList>(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 execute(PreparedStatementCreator psc, PreparedStatementCallback 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 execute(String cql, PreparedStatementCallback 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 query(PreparedStatementCreator psc, ResultSetExtractor 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 List query(PreparedStatementCreator psc, RowMapper 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 query(PreparedStatementCreator psc, final PreparedStatementBinder pss, final ResultSetExtractor rse) + throws DataAccessException { + + Assert.notNull(rse, "ResultSetExtractor must not be null"); + logger.debug("Executing prepared CQL query"); + + return execute(psc, new PreparedStatementCallback() { + 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 query(String cql, PreparedStatementBinder pss, ResultSetExtractor 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 List query(String cql, PreparedStatementBinder pss, RowMapper 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() { + 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 List query(PreparedStatementCreator psc, final PreparedStatementBinder pss, final RowMapper rowMapper) + throws DataAccessException { + Assert.notNull(rowMapper, "RowMapper must not be null"); + logger.debug("Executing prepared CQL query"); + + return execute(psc, new PreparedStatementCallback>() { + public List 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); + } + }); + } +} \ No newline at end of file diff --git a/src/main/java/org/springframework/cassandra/core/CqlProvider.java b/src/main/java/org/springframework/cassandra/core/CqlProvider.java new file mode 100644 index 000000000..7b0ddd59d --- /dev/null +++ b/src/main/java/org/springframework/cassandra/core/CqlProvider.java @@ -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(); + +} diff --git a/src/main/java/org/springframework/cassandra/core/HostMapper.java b/src/main/java/org/springframework/cassandra/core/HostMapper.java new file mode 100644 index 000000000..66b81f3da --- /dev/null +++ b/src/main/java/org/springframework/cassandra/core/HostMapper.java @@ -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 { + + Collection mapHosts(Set host) throws DriverException; + +} diff --git a/src/main/java/org/springframework/cassandra/core/PreparedStatementBinder.java b/src/main/java/org/springframework/cassandra/core/PreparedStatementBinder.java new file mode 100644 index 000000000..f4bd1cca2 --- /dev/null +++ b/src/main/java/org/springframework/cassandra/core/PreparedStatementBinder.java @@ -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; + +} diff --git a/src/main/java/org/springframework/cassandra/core/PreparedStatementCallback.java b/src/main/java/org/springframework/cassandra/core/PreparedStatementCallback.java new file mode 100644 index 000000000..1b2ba5fda --- /dev/null +++ b/src/main/java/org/springframework/cassandra/core/PreparedStatementCallback.java @@ -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 doInPreparedStatement(PreparedStatement ps) throws DriverException, DataAccessException; + +} diff --git a/src/main/java/org/springframework/cassandra/core/PreparedStatementCreator.java b/src/main/java/org/springframework/cassandra/core/PreparedStatementCreator.java new file mode 100644 index 000000000..d95e92862 --- /dev/null +++ b/src/main/java/org/springframework/cassandra/core/PreparedStatementCreator.java @@ -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; + +} diff --git a/src/main/java/org/springframework/cassandra/core/PreparedStatementCreatorImpl.java b/src/main/java/org/springframework/cassandra/core/PreparedStatementCreatorImpl.java new file mode 100644 index 000000000..2bedb7f67 --- /dev/null +++ b/src/main/java/org/springframework/cassandra/core/PreparedStatementCreatorImpl.java @@ -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 values; + + public PreparedStatementCreatorImpl(String cql) { + this.cql = cql; + } + + public PreparedStatementCreatorImpl(String cql, List 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); + } + +} diff --git a/src/main/java/org/springframework/cassandra/core/ResultSetFutureExtractor.java b/src/main/java/org/springframework/cassandra/core/ResultSetFutureExtractor.java new file mode 100644 index 000000000..7c52af2eb --- /dev/null +++ b/src/main/java/org/springframework/cassandra/core/ResultSetFutureExtractor.java @@ -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 extractData(ResultSetFuture rs) throws DriverException, DataAccessException; +} diff --git a/src/main/java/org/springframework/data/cassandra/core/RingMember.java b/src/main/java/org/springframework/cassandra/core/RingMember.java similarity index 77% rename from src/main/java/org/springframework/data/cassandra/core/RingMember.java rename to src/main/java/org/springframework/cassandra/core/RingMember.java index 829a8d1ab..705d1b6b7 100644 --- a/src/main/java/org/springframework/data/cassandra/core/RingMember.java +++ b/src/main/java/org/springframework/cassandra/core/RingMember.java @@ -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(); diff --git a/src/main/java/org/springframework/cassandra/core/RingMemberHostMapper.java b/src/main/java/org/springframework/cassandra/core/RingMemberHostMapper.java new file mode 100644 index 000000000..d4a0e44ed --- /dev/null +++ b/src/main/java/org/springframework/cassandra/core/RingMemberHostMapper.java @@ -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 + * + */ +public class RingMemberHostMapper implements HostMapper { + + /* (non-Javadoc) + * @see org.springframework.cassandra.core.HostMapper#mapHosts(java.util.Set) + */ + @Override + public List mapHosts(Set hosts) throws DriverException { + + List members = new ArrayList(); + + Assert.notNull(hosts); + Assert.notEmpty(hosts); + + RingMember r = null; + for (Host host : hosts) { + r = new RingMember(host); + members.add(r); + } + + return members; + + } +} diff --git a/src/main/java/org/springframework/cassandra/core/SessionFactoryBean.java b/src/main/java/org/springframework/cassandra/core/SessionFactoryBean.java index c930c9f2d..179f6a64b 100644 --- a/src/main/java/org/springframework/cassandra/core/SessionFactoryBean.java +++ b/src/main/java/org/springframework/cassandra/core/SessionFactoryBean.java @@ -64,7 +64,7 @@ public class SessionFactoryBean implements FactoryBean, InitializingBea * @see org.springframework.beans.factory.FactoryBean#getObject() */ @Override - public Session getObject() throws Exception { + public Session getObject() { return keyspace.getSession(); } diff --git a/src/main/java/org/springframework/cassandra/core/SimplePreparedStatementCreator.java b/src/main/java/org/springframework/cassandra/core/SimplePreparedStatementCreator.java new file mode 100644 index 000000000..f2ae91a5c --- /dev/null +++ b/src/main/java/org/springframework/cassandra/core/SimplePreparedStatementCreator.java @@ -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); + } + +} diff --git a/src/main/java/org/springframework/data/cassandra/core/CassandraDataOperations.java b/src/main/java/org/springframework/data/cassandra/core/CassandraDataOperations.java index 780db138d..39a0722a8 100644 --- a/src/main/java/org/springframework/data/cassandra/core/CassandraDataOperations.java +++ b/src/main/java/org/springframework/data/cassandra/core/CassandraDataOperations.java @@ -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 describeRing(); - /** * The table name used for the specified class by this template. * diff --git a/src/main/java/org/springframework/data/cassandra/core/CassandraDataTemplate.java b/src/main/java/org/springframework/data/cassandra/core/CassandraDataTemplate.java index 122865606..32e577a9b 100644 --- a/src/main/java/org/springframework/data/cassandra/core/CassandraDataTemplate.java +++ b/src/main/java/org/springframework/data/cassandra/core/CassandraDataTemplate.java @@ -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 describeRing() { - - /* - * Initialize the return variable - */ - List ring = new ArrayList(); - - /* - * Get the cluster metadata for this session - */ - Metadata clusterMetadata = doExecute(new SessionCallback() { - - @Override - public Metadata doInSession(Session s) throws DataAccessException { - return s.getCluster().getMetadata(); - } - - }); - - /* - * Get all hosts in the cluster - */ - Set 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); } } diff --git a/src/test/java/org/springframework/data/cassandra/config/TestConfig.java b/src/test/java/org/springframework/data/cassandra/config/TestConfig.java index b72f71d4e..12ee8f2c7 100644 --- a/src/test/java/org/springframework/data/cassandra/config/TestConfig.java +++ b/src/test/java/org/springframework/data/cassandra/config/TestConfig.java @@ -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() { diff --git a/src/test/java/org/springframework/data/cassandra/template/CassandraDataOperationsTest.java b/src/test/java/org/springframework/data/cassandra/template/CassandraDataOperationsTest.java index 30c216ba8..6e0467ffe 100644 --- a/src/test/java/org/springframework/data/cassandra/template/CassandraDataOperationsTest.java +++ b/src/test/java/org/springframework/data/cassandra/template/CassandraDataOperationsTest.java @@ -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 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() { diff --git a/src/test/java/org/springframework/data/cassandra/template/CassandraOperationsTest.java b/src/test/java/org/springframework/data/cassandra/template/CassandraOperationsTest.java new file mode 100644 index 000000000..c71168c1b --- /dev/null +++ b/src/test/java/org/springframework/data/cassandra/template/CassandraOperationsTest.java @@ -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 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 ring = (List) cassandraTemplate.describeRing(new HostMapper() { + + @Override + public Collection mapHosts(Set host) throws DriverException { + + List list = new LinkedList(); + + 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(); + } +}