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

Merge remote-tracking branch 'origin/DATACASS-69'

Browse Source
This commit is contained in:
David Webb
2014-01-23 20:19:04 -05:00
parent 002bea2775 d7075ec49e
commit ef876c445b
8 changed files with 639 additions and 303 deletions
Download Patch File Download Diff File Expand all files Collapse all files

33
spring-cassandra/src/main/java/org/springframework/cassandra/core/AsynchronousQueryListener.java Normal file
View File

@@ -0,0 +1,33 @@
/*
* Copyright 2011-2014 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.cassandra.core;
import com.datastax.driver.core.ResultSetFuture;
/**
* @author David Webb
*
*/
public interface AsynchronousQueryListener {
/**
* Called upon Query Completion.
*
* @param rsf The given ResultSetFuture's get methods should return immediately.
*/
public void onQueryComplete(ResultSetFuture rsf);
}

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

@@ -18,10 +18,13 @@ package org.springframework.cassandra.core;
import java.util.Collection;
import java.util.List;
import java.util.Map;
import java.util.concurrent.Executor;
import java.util.concurrent.TimeUnit;
import org.springframework.dao.DataAccessException;
import com.datastax.driver.core.ResultSet;
import com.datastax.driver.core.ResultSetFuture;
import com.datastax.driver.core.Session;
/**
@@ -46,14 +49,137 @@ public interface CassandraOperations {
*
* @param cql
*/
void execute(final String cql) throws DataAccessException;
void execute(String cql) throws DataAccessException;
/**
* Executes the supplied CQL Query Asynchronously and returns nothing.
*
* @param cql The CQL Statement to execute
*/
void executeAsynchronously(final String cql) throws DataAccessException;
void executeAsynchronously(String cql) throws DataAccessException;
/**
* Executes the provided CQL Query, and extracts the results with the ResultSetExtractor. This uses default Query
* Options when extracting the ResultSet.
*
* @param cql The Query
* @param rse The implementation for extracting the ResultSet
* @param timeout Time to wait for results
* @param timeUnit Time unit to wait for results
* @return
*/
<T> T queryAsynchronously(String cql, ResultSetExtractor<T> rse, Long timeout, TimeUnit timeUnit);
/**
* Executes the provided CQL Query, and extracts the results with the ResultSetExtractor.
*
* @param cql The Query
* @param rse The implementation for extracting the ResultSet
* @param timeout Time to wait for results
* @param timeUnit Time unit to wait for results
* @param options Query Options
* @return
*/
<T> T queryAsynchronously(String cql, ResultSetExtractor<T> rse, Long timeout, TimeUnit timeUnit, QueryOptions options);
/**
* Executes the provided CQL Query and returns the ResultSetFuture for user processing.
*
* @param cql The Query
* @return
*/
ResultSetFuture queryAsynchronously(String cql);
/**
* Executes the provided CQL Query and returns the ResultSetFuture for user processing.
*
* @param cql The Query
* @param options Query Options
* @return
*/
ResultSetFuture queryAsynchronously(String cql, QueryOptions options);
/**
* Executes the provided CQL Query with the provided Runnable implementations.
*
* @param cql The Query
* @param listener Runnable Listener for handling the query in a separate thread
*/
void queryAsynchronously(String cql, Runnable listener);
/**
* Executes the provided CQL Query with the provided listener. This is preferred over the same method that takes a
* plain Runnable. The {@link AsynchronousQueryListener} gives you access to the {@link ResultSetFuture} once the
* query is completed for optimal flexibility.
*
* @param cql The Query
* @param listener Runnable Listener for handling the query in a separate thread
*/
void queryAsynchronously(String cql, AsynchronousQueryListener listener);
/**
* Executes the provided CQL Query with the Runnable implementations using the Query Options.
*
* @param cql The Query
* @param options Query Option
* @param listener Runnable Listener for handling the query in a separate thread
*/
void queryAsynchronously(String cql, Runnable listener, QueryOptions options);
/**
* Executes the provided CQL Query with the provided Listener and Query Options. This is preferred over the same
* method that takes a plain Runnable. The {@link AsynchronousQueryListener} gives you access to the
* {@link ResultSetFuture} once the query is completed for optimal flexibility.
*
* @param cql The Query
* @param options Query Option
* @param listener Runnable Listener for handling the query in a separate thread
*/
void queryAsynchronously(String cql, AsynchronousQueryListener listener, QueryOptions options);
/**
* Executes the provided CQL Query with the provided Executor and Runnable implementations.
*
* @param cql The Query
* @param options Query Option
* @param listener Runnable Listener for handling the query in a separate thread
* @param executor To execute the Runnable Listener
*/
void queryAsynchronously(String cql, Runnable listener, Executor executor);
/**
* Executes the provided CQL Query with the provided listener and executor. This is preferred over the same method
* that takes a plain Runnable. The {@link AsynchronousQueryListener} gives you access to the {@link ResultSetFuture}
* once the query is completed for optimal flexibility.
*
* @param cql The Query
* @param options Query Option
* @param listener Runnable Listener for handling the query in a separate thread
* @param executor To execute the Runnable Listener
*/
void queryAsynchronously(String cql, AsynchronousQueryListener listener, Executor executor);
/**
* Executes the provided CQL Query with the provided Executor and Runnable implementations.
*
* @param cql The Query
* @param options Query Option
* @param listener Runnable Listener for handling the query in a separate thread
* @param executor To execute the Runnable Listener
*/
void queryAsynchronously(String cql, Runnable listener, QueryOptions options, Executor executor);
/**
* Executes the provided CQL Query with the provided Listener, Executor and Query Options. This is preferred over the
* same method that takes a plain Runnable. The {@link AsynchronousQueryListener} gives you access to the
* {@link ResultSetFuture} once the query is completed for optimal flexibility.
*
* @param cql
* @param listener
* @param options
* @param executor
*/
void queryAsynchronously(String cql, AsynchronousQueryListener listener, QueryOptions options, Executor executor);
/**
* Executes the provided CQL Query, and extracts the results with the ResultSetExtractor.
@@ -64,41 +190,19 @@ public interface CassandraOperations {
* @return Type <T> specified in the ResultSetExtractor
* @throws DataAccessException
*/
<T> T query(final String cql, ResultSetExtractor<T> rse) throws DataAccessException;
<T> T query(String cql, ResultSetExtractor<T> rse) 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 ResultSet
* @param options Query Options Object
* @param options Query Options
*
* @return
* @throws DataAccessException
*/
<T> T query(final String cql, ResultSetExtractor<T> rse, final QueryOptions options) 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 future results
* @return
* @throws DataAccessException
*/
<T> T queryAsynchronously(final String cql, ResultSetFutureExtractor<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 future results
* @param options Query Options Object
* @return
* @throws DataAccessException
*/
<T> T queryAsynchronously(final String cql, ResultSetFutureExtractor<T> rse, final QueryOptions options)
throws DataAccessException;
<T> T query(String cql, ResultSetExtractor<T> rse, QueryOptions options) throws DataAccessException;
/**
* Executes the provided CQL Query, and then processes the results with the <code>RowCallbackHandler</code>.
@@ -107,7 +211,7 @@ public interface CassandraOperations {
* @param rch The implementation for processing the rows returned.
* @throws DataAccessException
*/
void query(final String cql, RowCallbackHandler rch) throws DataAccessException;
void query(String cql, RowCallbackHandler rch) throws DataAccessException;
/**
* Executes the provided CQL Query, and then processes the results with the <code>RowCallbackHandler</code>.
@@ -117,7 +221,7 @@ public interface CassandraOperations {
* @param options Query Options Object
* @throws DataAccessException
*/
void query(final String cql, RowCallbackHandler rch, final QueryOptions options) throws DataAccessException;
void query(String cql, RowCallbackHandler rch, QueryOptions options) throws DataAccessException;
/**
* Processes the ResultSet through the RowCallbackHandler and return nothing. This is used internal to the Template
@@ -138,7 +242,7 @@ public interface CassandraOperations {
* @return List of <T> processed by the RowMapper
* @throws DataAccessException
*/
<T> List<T> query(final String cql, RowMapper<T> rowMapper) throws DataAccessException;
<T> List<T> query(String cql, RowMapper<T> rowMapper) throws DataAccessException;
/**
* Executes the provided CQL Query, and maps all Rows returned with the supplied RowMapper.
@@ -149,7 +253,7 @@ public interface CassandraOperations {
* @return List of <T> processed by the RowMapper
* @throws DataAccessException
*/
<T> List<T> query(final String cql, RowMapper<T> rowMapper, final QueryOptions options) throws DataAccessException;
<T> List<T> query(String cql, RowMapper<T> rowMapper, QueryOptions options) throws DataAccessException;
/**
* Processes the ResultSet through the RowMapper and returns the List of mapped Rows. This is used internal to the
@@ -175,7 +279,7 @@ public interface CassandraOperations {
* @return Object<T>
* @throws DataAccessException
*/
<T> T queryForObject(final String cql, RowMapper<T> rowMapper) throws DataAccessException;
<T> T queryForObject(String cql, RowMapper<T> rowMapper) throws DataAccessException;
/**
* Process a ResultSet through a RowMapper. This is used internal to the Template for core operations, but is made
@@ -197,7 +301,7 @@ public interface CassandraOperations {
* @return The Object<T> - item [0,0] in the result table of the query.
* @throws DataAccessException
*/
<T> T queryForObject(final String cql, Class<T> requiredType) throws DataAccessException;
<T> T queryForObject(String cql, Class<T> requiredType) throws DataAccessException;
/**
* Process a ResultSet, trying to convert the first columns of the first Row to Class<T>. This is used internal to the
@@ -219,7 +323,7 @@ public interface CassandraOperations {
* @return Map representing the results of the Query
* @throws DataAccessException
*/
Map<String, Object> queryForMap(final String cql) throws DataAccessException;
Map<String, Object> queryForMap(String cql) throws DataAccessException;
/**
* Process a ResultSet with <b>ONE</b> Row and convert to a Map. This is used internal to the Template for core
@@ -241,7 +345,7 @@ public interface CassandraOperations {
* @return List of elementType
* @throws DataAccessException
*/
<T> List<T> queryForList(final String cql, Class<T> elementType) throws DataAccessException;
<T> List<T> queryForList(String cql, Class<T> elementType) throws DataAccessException;
/**
* Process a ResultSet and convert the first column of the results to a List. This is used internal to the Template
@@ -263,7 +367,7 @@ public interface CassandraOperations {
* @return List of Maps with the query results
* @throws DataAccessException
*/
List<Map<String, Object>> queryForListOfMap(final String cql) throws DataAccessException;
List<Map<String, Object>> queryForListOfMap(String cql) throws DataAccessException;
/**
* Process a ResultSet and convert it to a List of Maps with column/value. This is used internal to the Template for
@@ -313,9 +417,9 @@ public interface CassandraOperations {
* @return Type<T> generated by the ResultSetExtractor
* @throws DataAccessException
*/
<T> T query(final String cql, PreparedStatementBinder psb, ResultSetExtractor<T> rse) throws DataAccessException;
<T> T query(String cql, PreparedStatementBinder psb, ResultSetExtractor<T> rse) throws DataAccessException;
<T> T query(final String cql, PreparedStatementBinder psb, ResultSetExtractor<T> rse, final QueryOptions options)
<T> T query(String cql, PreparedStatementBinder psb, ResultSetExtractor<T> rse, QueryOptions options)
throws DataAccessException;
/**
@@ -328,7 +432,7 @@ public interface CassandraOperations {
* @param rch The RowCallbackHandler for processing the ResultSet
* @throws DataAccessException
*/
void query(final String cql, PreparedStatementBinder psb, RowCallbackHandler rch) throws DataAccessException;
void query(String cql, PreparedStatementBinder psb, RowCallbackHandler rch) throws DataAccessException;
/**
* Converts the CQL provided into a {@link SimplePreparedStatementCreator}. Then, the PreparedStatementBinder will
@@ -341,7 +445,7 @@ public interface CassandraOperations {
* @param options The Query Options Object
* @throws DataAccessException
*/
void query(final String cql, PreparedStatementBinder psb, RowCallbackHandler rch, final QueryOptions options)
void query(String cql, PreparedStatementBinder psb, RowCallbackHandler rch, QueryOptions options)
throws DataAccessException;
/**
@@ -356,7 +460,7 @@ public interface CassandraOperations {
* @return List of <T> for each Row returned from the Query.
* @throws DataAccessException
*/
<T> List<T> query(final String cql, PreparedStatementBinder psb, RowMapper<T> rowMapper) throws DataAccessException;
<T> List<T> query(String cql, PreparedStatementBinder psb, RowMapper<T> rowMapper) throws DataAccessException;
/**
* Converts the CQL provided into a {@link SimplePreparedStatementCreator}. Then, the PreparedStatementBinder will
@@ -371,7 +475,7 @@ public interface CassandraOperations {
* @return List of <T> for each Row returned from the Query.
* @throws DataAccessException
*/
<T> List<T> query(final String cql, PreparedStatementBinder psb, RowMapper<T> rowMapper, final QueryOptions options)
<T> List<T> query(String cql, PreparedStatementBinder psb, RowMapper<T> rowMapper, QueryOptions options)
throws DataAccessException;
/**
@@ -397,8 +501,7 @@ public interface CassandraOperations {
* @return Type <T> which is the output of the ResultSetExtractor
* @throws DataAccessException
*/
<T> T query(PreparedStatementCreator psc, ResultSetExtractor<T> rse, final QueryOptions options)
throws DataAccessException;
<T> T query(PreparedStatementCreator psc, ResultSetExtractor<T> rse, QueryOptions options) throws DataAccessException;
/**
* Uses the provided PreparedStatementCreator to prepare a new Session call. <b>This can only be used for CQL
@@ -421,8 +524,7 @@ public interface CassandraOperations {
* @param options The Query Options Object
* @throws DataAccessException
*/
void query(PreparedStatementCreator psc, RowCallbackHandler rch, final QueryOptions options)
throws DataAccessException;
void query(PreparedStatementCreator psc, RowCallbackHandler rch, QueryOptions options) throws DataAccessException;
/**
* Uses the provided PreparedStatementCreator to prepare a new Session call. <b>This can only be used for CQL
@@ -447,7 +549,7 @@ public interface CassandraOperations {
* @return List of Type <T> mapped from each Row in the Results
* @throws DataAccessException
*/
<T> List<T> query(PreparedStatementCreator psc, RowMapper<T> rowMapper, final QueryOptions options)
<T> List<T> query(PreparedStatementCreator psc, RowMapper<T> rowMapper, QueryOptions options)
throws DataAccessException;
/**
@@ -462,8 +564,8 @@ public interface CassandraOperations {
* @return Type <T> which is the output of the ResultSetExtractor
* @throws DataAccessException
*/
<T> T query(PreparedStatementCreator psc, final PreparedStatementBinder psb, final ResultSetExtractor<T> rse,
final QueryOptions options) throws DataAccessException;
<T> T query(PreparedStatementCreator psc, PreparedStatementBinder psb, ResultSetExtractor<T> rse, QueryOptions options)
throws DataAccessException;
/**
* Uses the provided PreparedStatementCreator to prepare a new Session call. Binds the values from the
@@ -476,7 +578,7 @@ public interface CassandraOperations {
* @return Type <T> which is the output of the ResultSetExtractor
* @throws DataAccessException
*/
<T> T query(PreparedStatementCreator psc, final PreparedStatementBinder psb, final ResultSetExtractor<T> rse)
<T> T query(PreparedStatementCreator psc, PreparedStatementBinder psb, ResultSetExtractor<T> rse)
throws DataAccessException;
/**
@@ -491,8 +593,8 @@ public interface CassandraOperations {
* @return Type <T> which is the output of the ResultSetExtractor
* @throws DataAccessException
*/
void query(PreparedStatementCreator psc, final PreparedStatementBinder psb, final RowCallbackHandler rch,
final QueryOptions options) throws DataAccessException;
void query(PreparedStatementCreator psc, PreparedStatementBinder psb, RowCallbackHandler rch, QueryOptions options)
throws DataAccessException;
/**
* Uses the provided PreparedStatementCreator to prepare a new Session call. Binds the values from the
@@ -505,7 +607,7 @@ public interface CassandraOperations {
* @return Type <T> which is the output of the ResultSetExtractor
* @throws DataAccessException
*/
void query(PreparedStatementCreator psc, final PreparedStatementBinder psb, final RowCallbackHandler rch)
void query(PreparedStatementCreator psc, PreparedStatementBinder psb, RowCallbackHandler rch)
throws DataAccessException;
/**
@@ -520,8 +622,8 @@ public interface CassandraOperations {
* @return Type <T> which is the output of the ResultSetExtractor
* @throws DataAccessException
*/
<T> List<T> query(PreparedStatementCreator psc, final PreparedStatementBinder psb, final RowMapper<T> rowMapper,
final QueryOptions options) throws DataAccessException;
<T> List<T> query(PreparedStatementCreator psc, PreparedStatementBinder psb, RowMapper<T> rowMapper,
QueryOptions options) throws DataAccessException;
/**
* Uses the provided PreparedStatementCreator to prepare a new Session call. Binds the values from the
@@ -534,7 +636,7 @@ public interface CassandraOperations {
* @return Type <T> which is the output of the ResultSetExtractor
* @throws DataAccessException
*/
<T> List<T> query(PreparedStatementCreator psc, final PreparedStatementBinder psb, final RowMapper<T> rowMapper)
<T> List<T> query(PreparedStatementCreator psc, PreparedStatementBinder psb, RowMapper<T> rowMapper)
throws DataAccessException;
/**

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

@@ -22,9 +22,14 @@ import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Executor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import org.springframework.cassandra.support.CassandraAccessor;
import org.springframework.dao.DataAccessException;
import org.springframework.dao.QueryTimeoutException;
import org.springframework.util.Assert;
import com.datastax.driver.core.BoundStatement;
@@ -77,26 +82,169 @@ public class CassandraTemplate extends CassandraAccessor implements CassandraOpe
afterPropertiesSet();
}
/* (non-Javadoc)
* @see org.springframework.data.cassandra.core.CassandraOperations#execute(org.springframework.data.cassandra.core.SessionCallback)
*/
@Override
public <T> T execute(SessionCallback<T> sessionCallback) throws DataAccessException {
return doExecute(sessionCallback);
}
/* (non-Javadoc)
* @see org.springframework.data.cassandra.core.CassandraOperations#execute(java.lang.String)
*/
@Override
public void execute(final String cql) throws DataAccessException {
doExecute(cql, null);
}
/* (non-Javadoc)
* @see org.springframework.cassandra.core.CassandraOperations#queryAsynchronously(java.lang.String, org.springframework.cassandra.core.ResultSetFutureExtractor, org.springframework.cassandra.core.QueryOptions)
*/
@Override
public ResultSetFuture queryAsynchronously(final String cql) {
return execute(new SessionCallback<ResultSetFuture>() {
@Override
public ResultSetFuture doInSession(Session s) throws DataAccessException {
return s.executeAsync(cql);
}
});
}
@Override
public <T> T queryAsynchronously(String cql, ResultSetExtractor<T> rse, Long timeout, TimeUnit timeUnit) {
return queryAsynchronously(cql, rse, timeout, timeUnit, null);
}
@Override
public <T> T queryAsynchronously(final String cql, final ResultSetExtractor<T> rse, final Long timeout,
final TimeUnit timeUnit, final QueryOptions options) {
return rse.extractData(execute(new SessionCallback<ResultSet>() {
@Override
public ResultSet doInSession(Session s) throws DataAccessException {
Statement statement = new SimpleStatement(cql);
addQueryOptions(statement, options);
ResultSetFuture rsf = s.executeAsync(statement);
ResultSet rs = null;
try {
rs = rsf.get(timeout, timeUnit);
} catch (TimeoutException e) {
throw new QueryTimeoutException("Asyncronous Query Timed Out.", e);
} catch (InterruptedException e) {
throw translateExceptionIfPossible(e);
} catch (ExecutionException e) {
if (e.getCause() instanceof Exception) {
throw translateExceptionIfPossible((Exception) e.getCause());
}
throw new CassandraUncategorizedDataAccessException("Unknown Throwable", e.getCause());
}
return rs;
}
}));
}
@Override
public ResultSetFuture queryAsynchronously(final String cql, final QueryOptions options) {
return execute(new SessionCallback<ResultSetFuture>() {
@Override
public ResultSetFuture doInSession(Session s) throws DataAccessException {
Statement statement = new SimpleStatement(cql);
addQueryOptions(statement, options);
return s.executeAsync(statement);
}
});
}
@Override
public void queryAsynchronously(String cql, Runnable listener) {
queryAsynchronously(cql, listener, new Executor() {
@Override
public void execute(Runnable command) {
command.run();
}
});
}
@Override
public void queryAsynchronously(String cql, AsynchronousQueryListener listener) {
queryAsynchronously(cql, listener, new Executor() {
@Override
public void execute(Runnable command) {
command.run();
}
});
}
@Override
public void queryAsynchronously(String cql, Runnable listener, QueryOptions options) {
queryAsynchronously(cql, listener, options, new Executor() {
@Override
public void execute(Runnable command) {
command.run();
}
});
}
@Override
public void queryAsynchronously(String cql, AsynchronousQueryListener listener, QueryOptions options) {
queryAsynchronously(cql, listener, options, new Executor() {
@Override
public void execute(Runnable command) {
command.run();
}
});
}
@Override
public void queryAsynchronously(String cql, Runnable listener, Executor executor) {
queryAsynchronously(cql, listener, null, executor);
}
@Override
public void queryAsynchronously(String cql, AsynchronousQueryListener listener, Executor executor) {
queryAsynchronously(cql, listener, null, executor);
}
@Override
public void queryAsynchronously(final String cql, final Runnable listener, final QueryOptions options,
final Executor executor) {
execute(new SessionCallback<Object>() {
@Override
public Object doInSession(Session s) throws DataAccessException {
Statement statement = new SimpleStatement(cql);
addQueryOptions(statement, options);
ResultSetFuture rsf = s.executeAsync(statement);
rsf.addListener(listener, executor);
return null;
}
});
}
@Override
public void queryAsynchronously(final String cql, final AsynchronousQueryListener listener,
final QueryOptions options, final Executor executor) {
execute(new SessionCallback<Object>() {
@Override
public Object doInSession(Session s) throws DataAccessException {
Statement statement = new SimpleStatement(cql);
addQueryOptions(statement, options);
final ResultSetFuture rsf = s.executeAsync(statement);
Runnable wrapper = new Runnable() {
@Override
public void run() {
listener.onQueryComplete(rsf);
}
};
rsf.addListener(wrapper, executor);
return null;
}
});
}
public <T> T queryAsynchronously(final String cql, ResultSetFutureExtractor<T> rse, final QueryOptions options)
throws DataAccessException {
return rse.extractData(execute(new SessionCallback<ResultSetFuture>() {
@@ -109,25 +257,15 @@ public class CassandraTemplate extends CassandraAccessor implements CassandraOpe
}));
}
/* (non-Javadoc)
* @see org.springframework.cassandra.core.CassandraOperations#query(java.lang.String, org.springframework.cassandra.core.FutureResultSetExtractor)
*/
@Override
public <T> T queryAsynchronously(final String cql, ResultSetFutureExtractor<T> rse) throws DataAccessException {
return queryAsynchronously(cql, rse, null);
}
/* (non-Javadoc)
* @see org.springframework.cassandra.core.CassandraOperations#query(java.lang.String, org.springframework.cassandra.core.ResultSetExtractor, java.util.Map)
*/
@Override
public <T> T query(String cql, ResultSetExtractor<T> rse) throws DataAccessException {
return query(cql, rse, null);
}
/* (non-Javadoc)
* @see org.springframework.cassandra.core.CassandraOperations#query(java.lang.String, org.springframework.cassandra.core.ResultSetExtractor, org.springframework.cassandra.core.QueryOptions)
*/
@Override
public <T> T query(String cql, ResultSetExtractor<T> rse, QueryOptions options) throws DataAccessException {
Assert.notNull(cql);
@@ -135,67 +273,40 @@ public class CassandraTemplate extends CassandraAccessor implements CassandraOpe
return rse.extractData(rs);
}
/* (non-Javadoc)
* @see org.springframework.cassandra.core.CassandraOperations#query(java.lang.String, org.springframework.cassandra.core.RowCallbackHandler, org.springframework.cassandra.core.QueryOptions)
*/
@Override
public void query(String cql, RowCallbackHandler rch, QueryOptions options) throws DataAccessException {
process(doExecute(cql, options), rch);
}
/* (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 {
query(cql, rch, null);
}
/* (non-Javadoc)
* @see org.springframework.cassandra.core.CassandraOperations#query(java.lang.String, org.springframework.cassandra.core.RowMapper, org.springframework.cassandra.core.QueryOptions)
*/
@Override
public <T> List<T> query(String cql, RowMapper<T> rowMapper, QueryOptions options) throws DataAccessException {
return process(doExecute(cql, options), rowMapper);
}
/* (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 query(cql, rowMapper, null);
}
/* (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, null));
}
/* (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, null), 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, null));
}
/* (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, null), 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, null), rowMapper);
}
@@ -215,7 +326,7 @@ public class CassandraTemplate extends CassandraAccessor implements CassandraOpe
return callback.doInSession(getSession());
} catch (DataAccessException e) {
throw throwTranslated(e);
throw translateExceptionIfPossible(e);
}
}
@@ -290,9 +401,6 @@ public class CassandraTemplate extends CassandraAccessor implements CassandraOpe
return map;
}
/* (non-Javadoc)
* @see org.springframework.cassandra.core.CassandraOperations#describeRing()
*/
@Override
public List<RingMember> describeRing() throws DataAccessException {
return new ArrayList<RingMember>(describeRing(new RingMemberHostMapper()));
@@ -326,18 +434,12 @@ public class CassandraTemplate extends CassandraAccessor implements CassandraOpe
}
/* (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>() {
@@ -348,9 +450,6 @@ public class CassandraTemplate extends CassandraAccessor implements CassandraOpe
});
}
/* (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 {
@@ -358,13 +457,10 @@ public class CassandraTemplate extends CassandraAccessor implements CassandraOpe
rch.processRow(row);
}
} catch (DriverException dx) {
throwTranslated(dx);
translateExceptionIfPossible(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>();
@@ -374,14 +470,11 @@ public class CassandraTemplate extends CassandraAccessor implements CassandraOpe
mappedRows.add(rowMapper.mapRow(row, i++));
}
} catch (DriverException dx) {
throwTranslated(dx);
translateExceptionIfPossible(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;
@@ -392,14 +485,11 @@ public class CassandraTemplate extends CassandraAccessor implements CassandraOpe
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);
translateExceptionIfPossible(dx);
}
return row;
}
/* (non-Javadoc)
* @see org.springframework.cassandra.core.CassandraOperations#processOne(com.datastax.driver.core.ResultSet, java.lang.Class)
*/
@SuppressWarnings("unchecked")
@Override
public <T> T processOne(ResultSet resultSet, Class<T> requiredType) throws DataAccessException {
@@ -413,9 +503,6 @@ public class CassandraTemplate extends CassandraAccessor implements CassandraOpe
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) {
@@ -424,11 +511,8 @@ public class CassandraTemplate extends CassandraAccessor implements CassandraOpe
return toMap(resultSet.one());
}
/* (non-Javadoc)
* @see org.springframework.cassandra.core.CassandraOperations#processList(com.datastax.driver.core.ResultSet, java.lang.Class)
*/
@Override
@SuppressWarnings( "unchecked" )
@SuppressWarnings("unchecked")
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());
@@ -438,9 +522,6 @@ public class CassandraTemplate extends CassandraAccessor implements CassandraOpe
return list;
}
/* (non-Javadoc)
* @see org.springframework.cassandra.core.CassandraOperations#processListOfMap(com.datastax.driver.core.ResultSet)
*/
@Override
public List<Map<String, Object>> processListOfMap(ResultSet resultSet) throws DataAccessException {
List<Row> rows = resultSet.all();
@@ -457,14 +538,18 @@ public class CassandraTemplate extends CassandraAccessor implements CassandraOpe
* @param ex
* @return
*/
protected RuntimeException throwTranslated(RuntimeException ex) {
protected RuntimeException translateExceptionIfPossible(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)
*/
protected RuntimeException translateExceptionIfPossible(Exception ex) {
if (ex instanceof RuntimeException) {
return translateExceptionIfPossible((RuntimeException) ex);
}
return new CassandraUncategorizedDataAccessException("Caught Uncategorized Exception", ex);
}
@Override
public <T> T execute(PreparedStatementCreator psc, PreparedStatementCallback<T> action) {
@@ -472,125 +557,83 @@ public class CassandraTemplate extends CassandraAccessor implements CassandraOpe
PreparedStatement ps = psc.createPreparedStatement(getSession());
return action.doInPreparedStatement(ps);
} catch (DriverException dx) {
throwTranslated(dx);
translateExceptionIfPossible(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, org.springframework.cassandra.core.QueryOptions)
*/
@Override
public <T> T query(PreparedStatementCreator psc, ResultSetExtractor<T> rse, QueryOptions options)
throws DataAccessException {
return query(psc, null, rse, options);
}
/* (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, rse, null);
}
/* (non-Javadoc)
* @see org.springframework.cassandra.core.CassandraOperations#query(org.springframework.cassandra.core.PreparedStatementCreator, org.springframework.cassandra.core.RowCallbackHandler, org.springframework.cassandra.core.QueryOptions)
*/
@Override
public void query(PreparedStatementCreator psc, RowCallbackHandler rch, QueryOptions options)
throws DataAccessException {
query(psc, null, rch, options);
}
/* (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, rch, null);
}
/* (non-Javadoc)
* @see org.springframework.cassandra.core.CassandraOperations#query(org.springframework.cassandra.core.PreparedStatementCreator, org.springframework.cassandra.core.RowMapper, org.springframework.cassandra.core.QueryOptions)
*/
@Override
public <T> List<T> query(PreparedStatementCreator psc, RowMapper<T> rowMapper, QueryOptions options)
throws DataAccessException {
return query(psc, null, rowMapper, options);
}
/* (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, rowMapper, null);
}
/* (non-Javadoc)
* @see org.springframework.cassandra.core.CassandraOperations#query(java.lang.String, org.springframework.cassandra.core.PreparedStatementBinder, org.springframework.cassandra.core.ResultSetExtractor, org.springframework.cassandra.core.QueryOptions)
*/
@Override
public <T> T query(String cql, PreparedStatementBinder psb, ResultSetExtractor<T> rse, QueryOptions options)
throws DataAccessException {
return query(new SimplePreparedStatementCreator(cql), psb, rse, options);
}
/* (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 psb, ResultSetExtractor<T> rse) throws DataAccessException {
return query(cql, psb, rse, null);
}
/* (non-Javadoc)
* @see org.springframework.cassandra.core.CassandraOperations#query(java.lang.String, org.springframework.cassandra.core.PreparedStatementBinder, org.springframework.cassandra.core.RowCallbackHandler, org.springframework.cassandra.core.QueryOptions)
*/
@Override
public void query(String cql, PreparedStatementBinder psb, RowCallbackHandler rch, QueryOptions options)
throws DataAccessException {
query(new SimplePreparedStatementCreator(cql), psb, rch, options);
}
/* (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 psb, RowCallbackHandler rch) throws DataAccessException {
query(cql, psb, rch, null);
}
/* (non-Javadoc)
* @see org.springframework.cassandra.core.CassandraOperations#query(java.lang.String, org.springframework.cassandra.core.PreparedStatementBinder, org.springframework.cassandra.core.RowMapper, org.springframework.cassandra.core.QueryOptions)
*/
@Override
public <T> List<T> query(String cql, PreparedStatementBinder psb, RowMapper<T> rowMapper, QueryOptions options)
throws DataAccessException {
return query(new SimplePreparedStatementCreator(cql), psb, rowMapper, options);
}
/* (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 psb, RowMapper<T> rowMapper) throws DataAccessException {
return query(cql, psb, rowMapper, null);
}
/* (non-Javadoc)
* @see org.springframework.cassandra.core.CassandraOperations#ingest(java.lang.String, org.springframework.cassandra.core.RowIterator, org.springframework.cassandra.core.QueryOptions)
*/
@Override
public void ingest(String cql, RowIterator rowIterator, QueryOptions options) {
@@ -602,17 +645,11 @@ public class CassandraTemplate extends CassandraAccessor implements CassandraOpe
}
}
/* (non-Javadoc)
* @see org.springframework.cassandra.core.CassandraOperations#ingest(java.lang.String, org.springframework.cassandra.core.RowIterator)
*/
@Override
public void ingest(String cql, RowIterator rowIterator) {
ingest(cql, rowIterator, null);
}
/* (non-Javadoc)
* @see org.springframework.cassandra.core.CassandraOperations#ingest(java.lang.String, java.util.List, org.springframework.cassandra.core.QueryOptions)
*/
@Override
public void ingest(String cql, final List<List<?>> rows, QueryOptions options) {
@@ -637,17 +674,11 @@ public class CassandraTemplate extends CassandraAccessor implements CassandraOpe
}
/* (non-Javadoc)
* @see org.springframework.cassandra.core.CassandraOperations#ingest(java.lang.String, java.util.List)
*/
@Override
public void ingest(String cql, List<List<?>> rows) {
ingest(cql, rows, null);
}
/* (non-Javadoc)
* @see org.springframework.cassandra.core.CassandraOperations#ingest(java.lang.String, java.lang.Object[][], org.springframework.cassandra.core.QueryOptions)
*/
@Override
public void ingest(String cql, final Object[][] rows, QueryOptions options) {
@@ -668,17 +699,11 @@ public class CassandraTemplate extends CassandraAccessor implements CassandraOpe
}, options);
}
/* (non-Javadoc)
* @see org.springframework.cassandra.core.CassandraOperations#ingest(java.lang.String, java.lang.Object[][])
*/
@Override
public void ingest(String cql, final Object[][] rows) {
ingest(cql, rows, null);
}
/* (non-Javadoc)
* @see org.springframework.cassandra.core.CassandraOperations#truncate(java.lang.String)
*/
@Override
public void truncate(String tableName) throws DataAccessException {
Truncate truncate = QueryBuilder.truncate(tableName);
@@ -733,9 +758,6 @@ public class CassandraTemplate extends CassandraAccessor implements CassandraOpe
}
/* (non-Javadoc)
* @see org.springframework.cassandra.core.CassandraOperations#query(org.springframework.cassandra.core.PreparedStatementCreator, org.springframework.cassandra.core.PreparedStatementBinder, org.springframework.cassandra.core.ResultSetExtractor, org.springframework.cassandra.core.QueryOptions)
*/
@Override
public <T> T query(PreparedStatementCreator psc, final PreparedStatementBinder psb, final ResultSetExtractor<T> rse,
final QueryOptions options) throws DataAccessException {
@@ -758,18 +780,12 @@ public class CassandraTemplate extends CassandraAccessor implements CassandraOpe
});
}
/* (non-Javadoc)
* @see org.springframework.cassandra.core.CassandraOperations#query(org.springframework.cassandra.core.PreparedStatementCreator, org.springframework.cassandra.core.PreparedStatementBinder, org.springframework.cassandra.core.ResultSetExtractor)
*/
@Override
public <T> T query(PreparedStatementCreator psc, PreparedStatementBinder psb, ResultSetExtractor<T> rse)
throws DataAccessException {
return query(psc, psb, rse, null);
}
/* (non-Javadoc)
* @see org.springframework.cassandra.core.CassandraOperations#query(org.springframework.cassandra.core.PreparedStatementCreator, org.springframework.cassandra.core.PreparedStatementBinder, org.springframework.cassandra.core.RowCallbackHandler, org.springframework.cassandra.core.QueryOptions)
*/
@Override
public void query(PreparedStatementCreator psc, final PreparedStatementBinder psb, final RowCallbackHandler rch,
final QueryOptions options) throws DataAccessException {
@@ -793,18 +809,12 @@ public class CassandraTemplate extends CassandraAccessor implements CassandraOpe
});
}
/* (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, PreparedStatementBinder psb, RowCallbackHandler rch)
throws DataAccessException {
query(psc, psb, rch, 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, org.springframework.cassandra.core.QueryOptions)
*/
@Override
public <T> List<T> query(PreparedStatementCreator psc, final PreparedStatementBinder psb,
final RowMapper<T> rowMapper, final QueryOptions options) throws DataAccessException {
@@ -827,9 +837,6 @@ public class CassandraTemplate extends CassandraAccessor implements CassandraOpe
});
}
/* (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, PreparedStatementBinder psb, RowMapper<T> rowMapper)
throws DataAccessException {

38
spring-cassandra/src/main/java/org/springframework/cassandra/core/CassandraUncategorizedDataAccessException.java Normal file
View File

@@ -0,0 +1,38 @@
/*
* Copyright 2011-2014 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.cassandra.core;
import org.springframework.dao.UncategorizedDataAccessException;
/**
* Wrapper for all non RuntimeExceptions throws by the Cassandra Driver
*
* @author David Webb
*
*/
public class CassandraUncategorizedDataAccessException extends UncategorizedDataAccessException {
/**
* Create the Exception
*
* @param msg
* @param cause
*/
public CassandraUncategorizedDataAccessException(String msg, Throwable cause) {
super(msg, cause);
}
}

7
spring-cassandra/src/main/java/org/springframework/cassandra/support/CassandraExceptionTranslator.java
View File

@@ -62,12 +62,7 @@ import com.datastax.driver.core.exceptions.WriteTimeoutException;
public class CassandraExceptionTranslator implements PersistenceExceptionTranslator {
/*
* (non-Javadoc)
*
* @see org.springframework.dao.support.PersistenceExceptionTranslator#
* translateExceptionIfPossible(java.lang.RuntimeException)
*/
@Override
public DataAccessException translateExceptionIfPossible(RuntimeException x) {
if (x instanceof DataAccessException) {

72
spring-cassandra/src/test/java/org/springframework/cassandra/test/integration/core/template/BookListener.java Normal file
View File

@@ -0,0 +1,72 @@
/*
* Copyright 2011-2014 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.cassandra.test.integration.core.template;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.cassandra.core.AsynchronousQueryListener;
import com.datastax.driver.core.ResultSetFuture;
import com.datastax.driver.core.Row;
/**
* Test Implementation of the {@link AsynchronousQueryListener}
*
* @author David Webb
*
*/
public class BookListener implements AsynchronousQueryListener {
private static Logger log = LoggerFactory.getLogger(BookListener.class);
private Book book;
private boolean done;
@Override
public void onQueryComplete(ResultSetFuture rsf) {
log.info("QueryCompleted");
Row row;
try {
row = rsf.get().one();
} catch (Exception e) {
throw new RuntimeException("Failed to get ResultSet from ResultSetFuture", e);
}
book = new Book();
book.setIsbn(row.getString("isbn"));
book.setTitle(row.getString("title"));
book.setAuthor(row.getString("author"));
book.setPages(row.getInt("pages"));
done = true;
log.info("DONE");
}
/**
* @return Returns the done.
*/
public boolean isDone() {
return done;
}
/**
* @return Returns the book.
*/
public Book getBook() {
return book;
}
}

241
spring-cassandra/src/test/java/org/springframework/cassandra/test/integration/core/template/CassandraOperationsTest.java
View File

@@ -25,6 +25,8 @@ import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.UUID;
import java.util.concurrent.Executor;
import java.util.concurrent.TimeUnit;
import org.cassandraunit.CassandraCQLUnit;
import org.cassandraunit.dataset.cql.ClassPathCQLDataSet;
@@ -35,12 +37,14 @@ import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.cassandra.core.CassandraOperations;
import org.springframework.cassandra.core.CassandraTemplate;
import org.springframework.cassandra.core.ConsistencyLevel;
import org.springframework.cassandra.core.HostMapper;
import org.springframework.cassandra.core.PreparedStatementBinder;
import org.springframework.cassandra.core.PreparedStatementCallback;
import org.springframework.cassandra.core.PreparedStatementCreator;
import org.springframework.cassandra.core.QueryOptions;
import org.springframework.cassandra.core.ResultSetExtractor;
import org.springframework.cassandra.core.ResultSetFutureExtractor;
import org.springframework.cassandra.core.RetryPolicy;
import org.springframework.cassandra.core.RingMember;
import org.springframework.cassandra.core.RowCallbackHandler;
import org.springframework.cassandra.core.RowIterator;
@@ -360,20 +364,19 @@ public class CassandraOperationsTest extends AbstractKeyspaceCreatingIntegration
final String isbn = "999999999";
Book b1 = cassandraTemplate.queryAsynchronously("select * from book where isbn='" + isbn + "'",
new ResultSetFutureExtractor<Book>() {
@Override
public Book extractData(ResultSetFuture rs) throws DriverException, DataAccessException {
new ResultSetExtractor<Book>() {
ResultSet frs = rs.getUninterruptibly();
Row r = frs.one();
assertNotNull(r);
@Override
public Book extractData(ResultSet rs) throws DriverException, DataAccessException {
Row r = rs.one();
assertNotNull(r);
Book b = rowToBook(r);
Book b = rowToBook(r);
return b;
}
});
return b;
}
}, 60l, TimeUnit.SECONDS);
Book b2 = getBook(isbn);
@@ -381,6 +384,130 @@ public class CassandraOperationsTest extends AbstractKeyspaceCreatingIntegration
}
@Test
public void queryAsynchronouslyTestCqlStringResultSetExtractorWithOptions() {
QueryOptions options = new QueryOptions();
options.setConsistencyLevel(ConsistencyLevel.ONE);
options.setRetryPolicy(RetryPolicy.DEFAULT);
final String isbn = "999999999";
Book b1 = cassandraTemplate.queryAsynchronously("select * from book where isbn='" + isbn + "'",
new ResultSetExtractor<Book>() {
@Override
public Book extractData(ResultSet rs) throws DriverException, DataAccessException {
Row r = rs.one();
assertNotNull(r);
Book b = rowToBook(r);
return b;
}
}, 60l, TimeUnit.SECONDS, options);
Book b2 = getBook(isbn);
assertBook(b1, b2);
}
@Test
public void queryAsynchronouslyWithListener() {
QueryOptions options = new QueryOptions();
options.setConsistencyLevel(ConsistencyLevel.ONE);
options.setRetryPolicy(RetryPolicy.DEFAULT);
final String isbn = "999999999";
BookListener listener = new BookListener();
cassandraTemplate.queryAsynchronously("select * from book where isbn='" + isbn + "'", listener);
// TODO Use better multi threading devices here.
while (!listener.isDone()) {
try {
Thread.sleep(100);
} catch (InterruptedException muted) {
}
}
Book book2 = getBook(isbn);
assertBook(listener.getBook(), book2);
}
@Test
public void queryAsynchronouslyWithListenerAndExecutor() {
QueryOptions options = new QueryOptions();
options.setConsistencyLevel(ConsistencyLevel.ONE);
options.setRetryPolicy(RetryPolicy.DEFAULT);
final String isbn = "999999999";
BookListener listener = new BookListener();
cassandraTemplate.queryAsynchronously("select * from book where isbn='" + isbn + "'", listener, new Executor() {
@Override
public void execute(Runnable command) {
command.run();
}
});
// TODO Use better multi threading devices here.
while (!listener.isDone()) {
try {
Thread.sleep(100);
} catch (InterruptedException muted) {
}
}
Book book2 = getBook(isbn);
assertBook(listener.getBook(), book2);
}
@Test
public void queryAsynchronouslyWithListenerAndExecutorAndOptions() {
QueryOptions options = new QueryOptions();
options.setConsistencyLevel(ConsistencyLevel.ONE);
options.setRetryPolicy(RetryPolicy.DEFAULT);
final String isbn = "999999999";
BookListener listener = new BookListener();
cassandraTemplate.queryAsynchronously("select * from book where isbn='" + isbn + "'", listener, options,
new Executor() {
@Override
public void execute(Runnable command) {
command.run();
}
});
// TODO Use better multi threading devices here.
while (!listener.isDone()) {
try {
Thread.sleep(100);
} catch (InterruptedException muted) {
}
}
Book book2 = getBook(isbn);
assertBook(listener.getBook(), book2);
}
@Test
public void queryTestCqlStringRowCallbackHandler() {
@@ -405,22 +532,20 @@ public class CassandraOperationsTest extends AbstractKeyspaceCreatingIntegration
}
@Test
public void processTestResultSetRowCallbackHandler() {
public void processTestResultSetRowCallbackHandlerWithAsyncOptions() {
QueryOptions options = new QueryOptions();
options.setConsistencyLevel(ConsistencyLevel.ONE);
options.setRetryPolicy(RetryPolicy.DEFAULT);
final String isbn = "999999999";
final Book b1 = getBook(isbn);
ResultSet rs = cassandraTemplate.queryAsynchronously("select * from book where isbn='" + isbn + "'",
new ResultSetFutureExtractor<ResultSet>() {
ResultSetFuture rsf = cassandraTemplate
.queryAsynchronously("select * from book where isbn='" + isbn + "'", options);
@Override
public ResultSet extractData(ResultSetFuture rs) throws DriverException, DataAccessException {
ResultSet frs = rs.getUninterruptibly();
return frs;
}
});
ResultSet rs = rsf.getUninterruptibly();
assertNotNull(rs);
@@ -470,16 +595,10 @@ public class CassandraOperationsTest extends AbstractKeyspaceCreatingIntegration
// Insert our 3 test books.
ingestionTestObjectArray();
ResultSet rs = cassandraTemplate.queryAsynchronously("select * from book where isbn in ('1234','2345','3456')",
new ResultSetFutureExtractor<ResultSet>() {
ResultSetFuture rsf = cassandraTemplate
.queryAsynchronously("select * from book where isbn in ('1234','2345','3456')");
@Override
public ResultSet extractData(ResultSetFuture rs) throws DriverException, DataAccessException {
ResultSet frs = rs.getUninterruptibly();
return frs;
}
});
ResultSet rs = rsf.getUninterruptibly();
assertNotNull(rs);
@@ -542,16 +661,10 @@ public class CassandraOperationsTest extends AbstractKeyspaceCreatingIntegration
// Insert our 3 test books.
ingestionTestObjectArray();
ResultSet rs = cassandraTemplate.queryAsynchronously("select * from book where isbn in ('" + ISBN_NINES + "')",
new ResultSetFutureExtractor<ResultSet>() {
ResultSetFuture rsf = cassandraTemplate.queryAsynchronously("select * from book where isbn in ('" + ISBN_NINES
+ "')");
@Override
public ResultSet extractData(ResultSetFuture rs) throws DriverException, DataAccessException {
ResultSet frs = rs.getUninterruptibly();
return frs;
}
});
ResultSet rs = rsf.getUninterruptibly();
assertNotNull(rs);
@@ -589,16 +702,10 @@ public class CassandraOperationsTest extends AbstractKeyspaceCreatingIntegration
@Test
public void processOneTestResultSetType() {
ResultSet rs = cassandraTemplate.queryAsynchronously("select title from book where isbn in ('" + ISBN_NINES + "')",
new ResultSetFutureExtractor<ResultSet>() {
ResultSetFuture rsf = cassandraTemplate.queryAsynchronously("select title from book where isbn in ('" + ISBN_NINES
+ "')");
@Override
public ResultSet extractData(ResultSetFuture rs) throws DriverException, DataAccessException {
ResultSet frs = rs.getUninterruptibly();
return frs;
}
});
ResultSet rs = rsf.getUninterruptibly();
assertNotNull(rs);
@@ -627,16 +734,10 @@ public class CassandraOperationsTest extends AbstractKeyspaceCreatingIntegration
@Test
public void processMapTestResultSet() {
ResultSet rs = cassandraTemplate.queryAsynchronously("select * from book where isbn in ('" + ISBN_NINES + "')",
new ResultSetFutureExtractor<ResultSet>() {
ResultSetFuture rsf = cassandraTemplate.queryAsynchronously("select * from book where isbn in ('" + ISBN_NINES
+ "')");
@Override
public ResultSet extractData(ResultSetFuture rs) throws DriverException, DataAccessException {
ResultSet frs = rs.getUninterruptibly();
return frs;
}
});
ResultSet rs = rsf.getUninterruptibly();
assertNotNull(rs);
@@ -674,16 +775,10 @@ public class CassandraOperationsTest extends AbstractKeyspaceCreatingIntegration
// Insert our 3 test books.
ingestionTestObjectArray();
ResultSet rs = cassandraTemplate.queryAsynchronously("select * from book where isbn in ('1234','2345','3456')",
new ResultSetFutureExtractor<ResultSet>() {
ResultSetFuture rsf = cassandraTemplate
.queryAsynchronously("select * from book where isbn in ('1234','2345','3456')");
@Override
public ResultSet extractData(ResultSetFuture rs) throws DriverException, DataAccessException {
ResultSet frs = rs.getUninterruptibly();
return frs;
}
});
ResultSet rs = rsf.getUninterruptibly();
assertNotNull(rs);
@@ -716,16 +811,10 @@ public class CassandraOperationsTest extends AbstractKeyspaceCreatingIntegration
// Insert our 3 test books.
ingestionTestObjectArray();
ResultSet rs = cassandraTemplate.queryAsynchronously("select * from book where isbn in ('1234','2345','3456')",
new ResultSetFutureExtractor<ResultSet>() {
ResultSetFuture rsf = cassandraTemplate
.queryAsynchronously("select * from book where isbn in ('1234','2345','3456')");
@Override
public ResultSet extractData(ResultSetFuture rs) throws DriverException, DataAccessException {
ResultSet frs = rs.getUninterruptibly();
return frs;
}
});
ResultSet rs = rsf.getUninterruptibly();
assertNotNull(rs);
@@ -1096,7 +1185,7 @@ public class CassandraOperationsTest extends AbstractKeyspaceCreatingIntegration
* @param b
* @param orderedElements
*/
private void assertBook(Book b1, Book b2) {
public static void assertBook(Book b1, Book b2) {
assertEquals(b1.getIsbn(), b2.getIsbn());
assertEquals(b1.getTitle(), b2.getTitle());

2
spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/CassandraDataTemplate.java
View File

@@ -946,7 +946,7 @@ public class CassandraDataTemplate extends CassandraTemplate implements Cassandr
return callback.doInSession(getSession());
} catch (DataAccessException e) {
throw throwTranslated(e);
throw translateExceptionIfPossible(e);
}
}