From 963158957a6274e1ab8df8a78a760c27c34e24e5 Mon Sep 17 00:00:00 2001 From: David Webb Date: Thu, 23 Jan 2014 15:50:39 -0500 Subject: [PATCH 1/2] DATACASS-69 - Add more queryAsynchronously methods to Operations Interface --- .../cassandra/support/CassandraExceptionTranslator.java | 7 +------ .../data/cassandra/core/CassandraDataTemplate.java | 2 +- 2 files changed, 2 insertions(+), 7 deletions(-) diff --git a/spring-cassandra/src/main/java/org/springframework/cassandra/support/CassandraExceptionTranslator.java b/spring-cassandra/src/main/java/org/springframework/cassandra/support/CassandraExceptionTranslator.java index b2fe93910..a2ab9a27d 100644 --- a/spring-cassandra/src/main/java/org/springframework/cassandra/support/CassandraExceptionTranslator.java +++ b/spring-cassandra/src/main/java/org/springframework/cassandra/support/CassandraExceptionTranslator.java @@ -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) { diff --git a/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/CassandraDataTemplate.java b/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/CassandraDataTemplate.java index 6e94b48d1..c4ebf9dfa 100644 --- a/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/CassandraDataTemplate.java +++ b/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/CassandraDataTemplate.java @@ -946,7 +946,7 @@ public class CassandraDataTemplate extends CassandraTemplate implements Cassandr return callback.doInSession(getSession()); } catch (DataAccessException e) { - throw throwTranslated(e); + throw translateExceptionIfPossible(e); } } From d7075ec49e3c6333f42a245e30ba14d50eb86618 Mon Sep 17 00:00:00 2001 From: David Webb Date: Thu, 23 Jan 2014 15:51:53 -0500 Subject: [PATCH 2/2] DATACASS-69 - Add more queryAsynchronously methods to Operations Interface --- .../core/AsynchronousQueryListener.java | 33 ++ .../cassandra/core/CassandraOperations.java | 214 ++++++++--- .../cassandra/core/CassandraTemplate.java | 335 +++++++++--------- ...andraUncategorizedDataAccessException.java | 38 ++ .../core/template/BookListener.java | 72 ++++ .../template/CassandraOperationsTest.java | 241 +++++++++---- 6 files changed, 637 insertions(+), 296 deletions(-) create mode 100644 spring-cassandra/src/main/java/org/springframework/cassandra/core/AsynchronousQueryListener.java create mode 100644 spring-cassandra/src/main/java/org/springframework/cassandra/core/CassandraUncategorizedDataAccessException.java create mode 100644 spring-cassandra/src/test/java/org/springframework/cassandra/test/integration/core/template/BookListener.java diff --git a/spring-cassandra/src/main/java/org/springframework/cassandra/core/AsynchronousQueryListener.java b/spring-cassandra/src/main/java/org/springframework/cassandra/core/AsynchronousQueryListener.java new file mode 100644 index 000000000..3c59206b4 --- /dev/null +++ b/spring-cassandra/src/main/java/org/springframework/cassandra/core/AsynchronousQueryListener.java @@ -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); + +} diff --git a/spring-cassandra/src/main/java/org/springframework/cassandra/core/CassandraOperations.java b/spring-cassandra/src/main/java/org/springframework/cassandra/core/CassandraOperations.java index 31bee51a9..577cc3de0 100644 --- a/spring-cassandra/src/main/java/org/springframework/cassandra/core/CassandraOperations.java +++ b/spring-cassandra/src/main/java/org/springframework/cassandra/core/CassandraOperations.java @@ -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 queryAsynchronously(String cql, ResultSetExtractor 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 queryAsynchronously(String cql, ResultSetExtractor 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 specified in the ResultSetExtractor * @throws DataAccessException */ - T query(final String cql, ResultSetExtractor rse) throws DataAccessException; + T query(String cql, ResultSetExtractor 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 query(final String cql, ResultSetExtractor 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 queryAsynchronously(final String cql, ResultSetFutureExtractor 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 queryAsynchronously(final String cql, ResultSetFutureExtractor rse, final QueryOptions options) - throws DataAccessException; + T query(String cql, ResultSetExtractor rse, QueryOptions options) throws DataAccessException; /** * Executes the provided CQL Query, and then processes the results with the RowCallbackHandler. @@ -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 RowCallbackHandler. @@ -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 processed by the RowMapper * @throws DataAccessException */ - List query(final String cql, RowMapper rowMapper) throws DataAccessException; + List query(String cql, RowMapper 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 processed by the RowMapper * @throws DataAccessException */ - List query(final String cql, RowMapper rowMapper, final QueryOptions options) throws DataAccessException; + List query(String cql, RowMapper 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 * @throws DataAccessException */ - T queryForObject(final String cql, RowMapper rowMapper) throws DataAccessException; + T queryForObject(String cql, RowMapper 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 - item [0,0] in the result table of the query. * @throws DataAccessException */ - T queryForObject(final String cql, Class requiredType) throws DataAccessException; + T queryForObject(String cql, Class requiredType) throws DataAccessException; /** * Process a ResultSet, trying to convert the first columns of the first Row to Class. This is used internal to the @@ -219,7 +323,7 @@ public interface CassandraOperations { * @return Map representing the results of the Query * @throws DataAccessException */ - Map queryForMap(final String cql) throws DataAccessException; + Map queryForMap(String cql) throws DataAccessException; /** * Process a ResultSet with ONE 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 */ - List queryForList(final String cql, Class elementType) throws DataAccessException; + List queryForList(String cql, Class 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> queryForListOfMap(final String cql) throws DataAccessException; + List> 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 generated by the ResultSetExtractor * @throws DataAccessException */ - T query(final String cql, PreparedStatementBinder psb, ResultSetExtractor rse) throws DataAccessException; + T query(String cql, PreparedStatementBinder psb, ResultSetExtractor rse) throws DataAccessException; - T query(final String cql, PreparedStatementBinder psb, ResultSetExtractor rse, final QueryOptions options) + T query(String cql, PreparedStatementBinder psb, ResultSetExtractor 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 for each Row returned from the Query. * @throws DataAccessException */ - List query(final String cql, PreparedStatementBinder psb, RowMapper rowMapper) throws DataAccessException; + List query(String cql, PreparedStatementBinder psb, RowMapper 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 for each Row returned from the Query. * @throws DataAccessException */ - List query(final String cql, PreparedStatementBinder psb, RowMapper rowMapper, final QueryOptions options) + List query(String cql, PreparedStatementBinder psb, RowMapper rowMapper, QueryOptions options) throws DataAccessException; /** @@ -397,8 +501,7 @@ public interface CassandraOperations { * @return Type which is the output of the ResultSetExtractor * @throws DataAccessException */ - T query(PreparedStatementCreator psc, ResultSetExtractor rse, final QueryOptions options) - throws DataAccessException; + T query(PreparedStatementCreator psc, ResultSetExtractor rse, QueryOptions options) throws DataAccessException; /** * Uses the provided PreparedStatementCreator to prepare a new Session call. 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. This can only be used for CQL @@ -447,7 +549,7 @@ public interface CassandraOperations { * @return List of Type mapped from each Row in the Results * @throws DataAccessException */ - List query(PreparedStatementCreator psc, RowMapper rowMapper, final QueryOptions options) + List query(PreparedStatementCreator psc, RowMapper rowMapper, QueryOptions options) throws DataAccessException; /** @@ -462,8 +564,8 @@ public interface CassandraOperations { * @return Type which is the output of the ResultSetExtractor * @throws DataAccessException */ - T query(PreparedStatementCreator psc, final PreparedStatementBinder psb, final ResultSetExtractor rse, - final QueryOptions options) throws DataAccessException; + T query(PreparedStatementCreator psc, PreparedStatementBinder psb, ResultSetExtractor 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 which is the output of the ResultSetExtractor * @throws DataAccessException */ - T query(PreparedStatementCreator psc, final PreparedStatementBinder psb, final ResultSetExtractor rse) + T query(PreparedStatementCreator psc, PreparedStatementBinder psb, ResultSetExtractor rse) throws DataAccessException; /** @@ -491,8 +593,8 @@ public interface CassandraOperations { * @return Type 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 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 which is the output of the ResultSetExtractor * @throws DataAccessException */ - List query(PreparedStatementCreator psc, final PreparedStatementBinder psb, final RowMapper rowMapper, - final QueryOptions options) throws DataAccessException; + List query(PreparedStatementCreator psc, PreparedStatementBinder psb, RowMapper 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 which is the output of the ResultSetExtractor * @throws DataAccessException */ - List query(PreparedStatementCreator psc, final PreparedStatementBinder psb, final RowMapper rowMapper) + List query(PreparedStatementCreator psc, PreparedStatementBinder psb, RowMapper rowMapper) throws DataAccessException; /** diff --git a/spring-cassandra/src/main/java/org/springframework/cassandra/core/CassandraTemplate.java b/spring-cassandra/src/main/java/org/springframework/cassandra/core/CassandraTemplate.java index 6537e2103..d98854105 100644 --- a/spring-cassandra/src/main/java/org/springframework/cassandra/core/CassandraTemplate.java +++ b/spring-cassandra/src/main/java/org/springframework/cassandra/core/CassandraTemplate.java @@ -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 execute(SessionCallback 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() { + + @Override + public ResultSetFuture doInSession(Session s) throws DataAccessException { + return s.executeAsync(cql); + } + }); + } + + @Override + public T queryAsynchronously(String cql, ResultSetExtractor rse, Long timeout, TimeUnit timeUnit) { + return queryAsynchronously(cql, rse, timeout, timeUnit, null); + } + + @Override + public T queryAsynchronously(final String cql, final ResultSetExtractor rse, final Long timeout, + final TimeUnit timeUnit, final QueryOptions options) { + return rse.extractData(execute(new SessionCallback() { + @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() { + @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() { + @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() { + @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 queryAsynchronously(final String cql, ResultSetFutureExtractor rse, final QueryOptions options) throws DataAccessException { return rse.extractData(execute(new SessionCallback() { @@ -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 queryAsynchronously(final String cql, ResultSetFutureExtractor 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 query(String cql, ResultSetExtractor 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 query(String cql, ResultSetExtractor 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 List query(String cql, RowMapper 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 List query(String cql, RowMapper rowMapper) throws DataAccessException { return query(cql, rowMapper, null); } - /* (non-Javadoc) - * @see org.springframework.cassandra.core.CassandraOperations#queryForList(java.lang.String) - */ public List> 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 List queryForList(String cql, Class elementType) throws DataAccessException { return processList(doExecute(cql, null), elementType); } - /* (non-Javadoc) - * @see org.springframework.cassandra.core.CassandraOperations#queryForMap(java.lang.String) - */ public Map 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 queryForObject(String cql, Class 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 queryForObject(String cql, RowMapper 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 describeRing() throws DataAccessException { return new ArrayList(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 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() { @@ -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 List process(ResultSet resultSet, RowMapper rowMapper) throws DataAccessException { List mappedRows = new ArrayList(); @@ -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 processOne(ResultSet resultSet, RowMapper 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 processOne(ResultSet resultSet, Class 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 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 List processList(ResultSet resultSet, Class elementType) throws DataAccessException { List rows = resultSet.all(); List list = new ArrayList(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> processListOfMap(ResultSet resultSet) throws DataAccessException { List 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 execute(PreparedStatementCreator psc, PreparedStatementCallback 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 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, org.springframework.cassandra.core.QueryOptions) - */ @Override public T query(PreparedStatementCreator psc, ResultSetExtractor 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 query(PreparedStatementCreator psc, ResultSetExtractor 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 List query(PreparedStatementCreator psc, RowMapper 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 List query(PreparedStatementCreator psc, RowMapper 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 query(String cql, PreparedStatementBinder psb, ResultSetExtractor 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 query(String cql, PreparedStatementBinder psb, ResultSetExtractor 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 List query(String cql, PreparedStatementBinder psb, RowMapper 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 List query(String cql, PreparedStatementBinder psb, RowMapper 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> 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> 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 query(PreparedStatementCreator psc, final PreparedStatementBinder psb, final ResultSetExtractor 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 query(PreparedStatementCreator psc, PreparedStatementBinder psb, ResultSetExtractor 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 List query(PreparedStatementCreator psc, final PreparedStatementBinder psb, final RowMapper 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 List query(PreparedStatementCreator psc, PreparedStatementBinder psb, RowMapper rowMapper) throws DataAccessException { diff --git a/spring-cassandra/src/main/java/org/springframework/cassandra/core/CassandraUncategorizedDataAccessException.java b/spring-cassandra/src/main/java/org/springframework/cassandra/core/CassandraUncategorizedDataAccessException.java new file mode 100644 index 000000000..2fd716638 --- /dev/null +++ b/spring-cassandra/src/main/java/org/springframework/cassandra/core/CassandraUncategorizedDataAccessException.java @@ -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); + } + +} diff --git a/spring-cassandra/src/test/java/org/springframework/cassandra/test/integration/core/template/BookListener.java b/spring-cassandra/src/test/java/org/springframework/cassandra/test/integration/core/template/BookListener.java new file mode 100644 index 000000000..b27e6aae8 --- /dev/null +++ b/spring-cassandra/src/test/java/org/springframework/cassandra/test/integration/core/template/BookListener.java @@ -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; + } + +} diff --git a/spring-cassandra/src/test/java/org/springframework/cassandra/test/integration/core/template/CassandraOperationsTest.java b/spring-cassandra/src/test/java/org/springframework/cassandra/test/integration/core/template/CassandraOperationsTest.java index 3b31ecffe..a95287abd 100644 --- a/spring-cassandra/src/test/java/org/springframework/cassandra/test/integration/core/template/CassandraOperationsTest.java +++ b/spring-cassandra/src/test/java/org/springframework/cassandra/test/integration/core/template/CassandraOperationsTest.java @@ -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() { - @Override - public Book extractData(ResultSetFuture rs) throws DriverException, DataAccessException { + new ResultSetExtractor() { - 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() { + + @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() { + 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() { + 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() { + 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() { + 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() { + 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() { + 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() { + 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); @@ -1112,7 +1201,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());