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

DATACASS-168: Enable cancellation of asynchronous CqlOperations methods

Browse Source
This commit is contained in:
Matthew Adams
2014-09-22 11:44:20 -05:00
parent dfa4b34a10
commit d1edce4f7e
4 changed files with 243 additions and 172 deletions
Download Patch File Download Diff File Expand all files Collapse all files

37
spring-cql/src/main/java/org/springframework/cassandra/core/BasicQueryCancellor.java Normal file
View File

@@ -0,0 +1,37 @@
/*
* Copyright 2013-2014 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* 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;
/**
* Convenient default implementation of a {@link QueryCancellor}.
*
* @author Matthew T. Adams
*/
class BasicQueryCancellor implements QueryCancellor {
ResultSetFuture rsf;
public BasicQueryCancellor(ResultSetFuture rsf) {
this.rsf = rsf;
}
@Override
public void cancelQuery(boolean mayInterruptIfRunning) {
rsf.cancel(mayInterruptIfRunning);
}
}

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

@@ -122,25 +122,28 @@ public interface CqlOperations {
* Executes the supplied Query Asynchronously and returns nothing.
*
* @param cql The CQL String to execute
* @return A {@link ResultSetFuture} that can be used to cancel the query.
*/
void executeAsynchronously(String cql) throws DataAccessException;
ResultSetFuture executeAsynchronously(String cql) throws DataAccessException;
/**
* Executes the supplied Query Asynchronously and returns nothing.
*
* @param cql The CQL String to execute
* @param options The {@link QueryOptions} to use. Only applies to cql statements that can use QueryOptions.
* @return A {@link ResultSetFuture} that can be used to cancel the query.
*/
void executeAsynchronously(String cql, QueryOptions options) throws DataAccessException;
ResultSetFuture executeAsynchronously(String cql, QueryOptions options) throws DataAccessException;
/**
* Executes the supplied Query Asynchronously and returns nothing.
*
* @param cql The CQL String to execute
* @param listener The {@link Runnable} to register with the {@link ResultSetFuture}
* @return A {@link QueryCancellor} that can be used to cancel the query.
* @see queryAsyncronously for Reads
*/
void executeAsynchronously(String cql, Runnable listener) throws DataAccessException;
QueryCancellor executeAsynchronously(String cql, Runnable listener) throws DataAccessException;
/**
* Executes the supplied Query Asynchronously and returns nothing.
@@ -148,18 +151,20 @@ public interface CqlOperations {
* @param cql The CQL String to execute
* @param listener The {@link Runnable} to register with the {@link ResultSetFuture}
* @param executor The {@link Executor} to regsiter with the {@link ResultSetFuture}
* @return A {@link QueryCancellor} that can be used to cancel the query.
* @see queryAsyncronously for Reads
*/
void executeAsynchronously(String cql, Runnable listener, Executor executor) throws DataAccessException;
QueryCancellor executeAsynchronously(String cql, Runnable listener, Executor executor) throws DataAccessException;
/**
* Executes the supplied Query Asynchronously and returns nothing.
*
* @param cql The CQL String to execute
* @param listener The {@link AsynchronousQueryListener} to register with the {@link ResultSetFuture}
* @return A {@link QueryCancellor} that can be used to cancel the query.
* @see queryAsyncronously for Reads
*/
void executeAsynchronously(String cql, AsynchronousQueryListener listener) throws DataAccessException;
QueryCancellor executeAsynchronously(String cql, AsynchronousQueryListener listener) throws DataAccessException;
/**
* Executes the supplied Query Asynchronously and returns nothing.
@@ -167,80 +172,92 @@ public interface CqlOperations {
* @param cql The CQL String to execute
* @param listener The {@link AsynchronousQueryListener} to register with the {@link ResultSetFuture}
* @param executor The {@link Executor} to regsiter with the {@link ResultSetFuture}
* @return A {@link QueryCancellor} that can be used to cancel the query.
* @see queryAsyncronously for Reads
*/
void executeAsynchronously(String cql, AsynchronousQueryListener listener, Executor executor)
QueryCancellor executeAsynchronously(String cql, AsynchronousQueryListener listener, Executor executor)
throws DataAccessException;
/**
* Executes the supplied CQL Truncate Asynchronously and returns nothing.
*
* @param query The {@link Truncate} to execute
* @return A {@link ResultSetFuture} that can be used to cancel the query.
*/
void executeAsynchronously(Truncate truncate) throws DataAccessException;
ResultSetFuture executeAsynchronously(Truncate truncate) throws DataAccessException;
/**
* Executes the supplied CQL Delete Asynchronously and returns nothing.
*
* @param query The {@link Delete} to execute
* @return A {@link ResultSetFuture} that can be used to cancel the query.
*/
void executeAsynchronously(Delete delete) throws DataAccessException;
ResultSetFuture executeAsynchronously(Delete delete) throws DataAccessException;
/**
* Executes the supplied CQL Insert Asynchronously and returns nothing.
*
* @param query The {@link Insert} to execute
* @return A {@link ResultSetFuture} that can be used to cancel the query.
*/
void executeAsynchronously(Insert insert) throws DataAccessException;
ResultSetFuture executeAsynchronously(Insert insert) throws DataAccessException;
/**
* Executes the supplied CQL Update Asynchronously and returns nothing.
*
* @param query The {@link Update} to execute
* @return A {@link ResultSetFuture} that can be used to cancel the query.
*/
void executeAsynchronously(Update update) throws DataAccessException;
ResultSetFuture executeAsynchronously(Update update) throws DataAccessException;
/**
* Executes the supplied CQL Batch Asynchronously and returns nothing.
*
* @param query The {@link Batch} to execute
* @return A {@link ResultSetFuture} that can be used to cancel the query.
*/
void executeAsynchronously(Batch batch) throws DataAccessException;
ResultSetFuture executeAsynchronously(Batch batch) throws DataAccessException;
/**
* Executes the supplied CQL Query Asynchronously and returns nothing.
*
* @param query The {@link Statement} to execute
* @return A {@link ResultSetFuture} that can be used to cancel the query.
*/
void executeAsynchronously(Statement query) throws DataAccessException;
ResultSetFuture executeAsynchronously(Statement query) throws DataAccessException;
/**
* Executes the supplied CQL Query Asynchronously and returns nothing.
*
* @param query The {@link Statement} to execute
* @return A {@link QueryCancellor} that can be used to cancel the query.
*/
void executeAsynchronously(Statement query, Runnable runnable) throws DataAccessException;
QueryCancellor executeAsynchronously(Statement query, Runnable runnable) throws DataAccessException;
/**
* Executes the supplied CQL Query Asynchronously and returns nothing.
*
* @param query The {@link Statement} to execute
* @return A {@link QueryCancellor} that can be used to cancel the query.
*/
void executeAsynchronously(Statement query, AsynchronousQueryListener listener) throws DataAccessException;
QueryCancellor executeAsynchronously(Statement query, AsynchronousQueryListener listener) throws DataAccessException;
/**
* Executes the supplied CQL Query Asynchronously and returns nothing.
*
* @param query The {@link Statement} to execute
* @return A {@link QueryCancellor} that can be used to cancel the query.
*/
void executeAsynchronously(Statement query, Runnable runnable, Executor executor) throws DataAccessException;
QueryCancellor executeAsynchronously(Statement query, Runnable runnable, Executor executor)
throws DataAccessException;
/**
* Executes the supplied CQL Query Asynchronously and returns nothing.
*
* @param query The {@link Statement} to execute
* @return A {@link QueryCancellor} that can be used to cancel the query.
*/
void executeAsynchronously(Statement query, AsynchronousQueryListener listener, Executor executor)
QueryCancellor executeAsynchronously(Statement query, AsynchronousQueryListener listener, Executor executor)
throws DataAccessException;
/**
@@ -300,9 +317,10 @@ public interface CqlOperations {
*
* @param cql The Query
* @param listener {@link Runnable} listener for handling the query in a separate thread
* @return A {@link QueryCancellor} that can be used to cancel the query.
* @see #queryAsynchronously(String, AsynchronousQueryListener)
*/
void queryAsynchronously(String cql, Runnable listener);
QueryCancellor queryAsynchronously(String cql, Runnable listener);
/**
* Executes the provided CQL Select with the provided {@link Runnable}, which is started after the query has
@@ -313,9 +331,10 @@ public interface CqlOperations {
*
* @param select The Select Query
* @param listener {@link Runnable} listener for handling the query in a separate thread
* @return A {@link QueryCancellor} that can be used to cancel the query.
* @see #queryAsynchronously(Select, AsynchronousQueryListener)
*/
void queryAsynchronously(Select select, Runnable listener);
QueryCancellor queryAsynchronously(Select select, Runnable listener);
/**
* Executes the provided CQL Query with the provided listener. This is preferred over the same method that takes a
@@ -325,8 +344,9 @@ public interface CqlOperations {
* @param cql The Query
* @param listener {@link AsynchronousQueryListener} for handling the query's {@link ResultSetFuture} in a separate
* thread
* @return A {@link QueryCancellor} that can be used to cancel the query.
*/
void queryAsynchronously(String cql, AsynchronousQueryListener listener);
QueryCancellor queryAsynchronously(String cql, AsynchronousQueryListener listener);
/**
* Executes the provided CQL Select with the provided listener. This is preferred over the same method that takes a
@@ -336,8 +356,9 @@ public interface CqlOperations {
* @param select The Select
* @param listener {@link AsynchronousQueryListener} for handling the query's {@link ResultSetFuture} in a separate
* thread
* @return A {@link QueryCancellor} that can be used to cancel the query.
*/
void queryAsynchronously(Select select, AsynchronousQueryListener listener);
QueryCancellor queryAsynchronously(Select select, AsynchronousQueryListener listener);
/**
* Executes the provided CQL Query with the Runnable implementations using the Query Options.
@@ -345,8 +366,9 @@ public interface CqlOperations {
* @param cql The Query
* @param options Query Option
* @param listener Runnable Listener for handling the query in a separate thread
* @return A {@link QueryCancellor} that can be used to cancel the query.
*/
void queryAsynchronously(String cql, Runnable listener, QueryOptions options);
QueryCancellor 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
@@ -356,8 +378,9 @@ public interface CqlOperations {
* @param cql The Query
* @param options Query Option
* @param listener Runnable Listener for handling the query in a separate thread
* @return A {@link QueryCancellor} that can be used to cancel the query.
*/
void queryAsynchronously(String cql, AsynchronousQueryListener listener, QueryOptions options);
QueryCancellor queryAsynchronously(String cql, AsynchronousQueryListener listener, QueryOptions options);
/**
* Executes the provided CQL Query with the provided Executor and Runnable implementations.
@@ -365,8 +388,9 @@ public interface CqlOperations {
* @param cql The Query
* @param listener Runnable Listener for handling the query in a separate thread
* @param executor To execute the Runnable Listener
* @return A {@link QueryCancellor} that can be used to cancel the query.
*/
void queryAsynchronously(String cql, Runnable listener, Executor executor);
QueryCancellor queryAsynchronously(String cql, Runnable listener, Executor executor);
/**
* Executes the provided CQL Select with the provided Executor and Runnable implementations.
@@ -374,8 +398,9 @@ public interface CqlOperations {
* @param select The Select Query
* @param listener Runnable Listener for handling the query in a separate thread
* @param executor To execute the Runnable Listener
* @return A {@link QueryCancellor} that can be used to cancel the query.
*/
void queryAsynchronously(Select select, Runnable listener, Executor executor);
QueryCancellor queryAsynchronously(Select select, Runnable listener, Executor executor);
/**
* Executes the provided CQL Query with the provided listener and executor. This is preferred over the same method
@@ -386,8 +411,9 @@ public interface CqlOperations {
* @param options Query Option
* @param listener Runnable Listener for handling the query in a separate thread
* @param executor To execute the Runnable Listener
* @return A {@link QueryCancellor} that can be used to cancel the query.
*/
void queryAsynchronously(String cql, AsynchronousQueryListener listener, Executor executor);
QueryCancellor queryAsynchronously(String cql, AsynchronousQueryListener listener, Executor executor);
/**
* Executes the provided Select Query with the provided listener and executor. This is preferred over the same method
@@ -397,8 +423,9 @@ public interface CqlOperations {
* @param select The Select Query
* @param listener Runnable Listener for handling the query in a separate thread
* @param executor To execute the Runnable Listener
* @return A {@link QueryCancellor} that can be used to cancel the query.
*/
void queryAsynchronously(Select select, AsynchronousQueryListener listener, Executor executor);
QueryCancellor queryAsynchronously(Select select, AsynchronousQueryListener listener, Executor executor);
/**
* Executes the provided CQL Query with the provided Executor and Runnable implementations.
@@ -407,8 +434,9 @@ public interface CqlOperations {
* @param options Query Option
* @param listener Runnable Listener for handling the query in a separate thread
* @param executor To execute the Runnable Listener
* @return A {@link QueryCancellor} that can be used to cancel the query.
*/
void queryAsynchronously(String cql, Runnable listener, QueryOptions options, Executor executor);
QueryCancellor 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
@@ -419,8 +447,10 @@ public interface CqlOperations {
* @param listener
* @param options
* @param executor
* @return A {@link QueryCancellor} that can be used to cancel the query.
*/
void queryAsynchronously(String cql, AsynchronousQueryListener listener, QueryOptions options, Executor executor);
QueryCancellor queryAsynchronously(String cql, AsynchronousQueryListener listener, QueryOptions options,
Executor executor);
/**
* Executes the provided CQL query and returns the {@link ResultSet}.

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

@@ -227,7 +227,6 @@ public class CqlTemplate extends CassandraAccessor implements CqlOperations {
@Override
public ResultSetFuture queryAsynchronously(final String cql) {
return execute(new SessionCallback<ResultSetFuture>() {
@Override
public ResultSetFuture doInSession(Session s) throws DataAccessException {
return s.executeAsync(cql);
@@ -280,89 +279,76 @@ public class CqlTemplate extends CassandraAccessor implements CqlOperations {
}
@Override
public void queryAsynchronously(String cql, Runnable listener) {
queryAsynchronously(cql, listener, new Executor() {
public QueryCancellor queryAsynchronously(String cql, Runnable listener) {
return 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() {
public QueryCancellor queryAsynchronously(String cql, AsynchronousQueryListener listener) {
return 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() {
public QueryCancellor queryAsynchronously(String cql, Runnable listener, QueryOptions options) {
return 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() {
public QueryCancellor queryAsynchronously(String cql, AsynchronousQueryListener listener, QueryOptions options) {
return 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);
public QueryCancellor queryAsynchronously(String cql, Runnable listener, Executor executor) {
return queryAsynchronously(cql, listener, null, executor);
}
@Override
public void queryAsynchronously(String cql, AsynchronousQueryListener listener, Executor executor) {
queryAsynchronously(cql, listener, null, executor);
public QueryCancellor queryAsynchronously(String cql, AsynchronousQueryListener listener, Executor executor) {
return queryAsynchronously(cql, listener, null, executor);
}
@Override
public void queryAsynchronously(final String cql, final Runnable listener, final QueryOptions options,
public QueryCancellor queryAsynchronously(final String cql, final Runnable listener, final QueryOptions options,
final Executor executor) {
execute(new SessionCallback<Object>() {
return execute(new SessionCallback<QueryCancellor>() {
@Override
public Object doInSession(Session s) throws DataAccessException {
public QueryCancellor doInSession(Session s) throws DataAccessException {
Statement statement = new SimpleStatement(cql);
addQueryOptions(statement, options);
ResultSetFuture rsf = s.executeAsync(statement);
rsf.addListener(listener, executor);
return null;
return new BasicQueryCancellor(rsf);
}
});
}
@Override
public void queryAsynchronously(final String cql, final AsynchronousQueryListener listener,
public QueryCancellor queryAsynchronously(final String cql, final AsynchronousQueryListener listener,
final QueryOptions options, final Executor executor) {
execute(new SessionCallback<Object>() {
return execute(new SessionCallback<QueryCancellor>() {
@Override
public Object doInSession(Session s) throws DataAccessException {
public QueryCancellor doInSession(Session s) throws DataAccessException {
Statement statement = new SimpleStatement(cql);
addQueryOptions(statement, options);
final ResultSetFuture rsf = s.executeAsync(statement);
@@ -373,7 +359,7 @@ public class CqlTemplate extends CassandraAccessor implements CqlOperations {
}
};
rsf.addListener(wrapper, executor);
return null;
return new BasicQueryCancellor(rsf);
}
});
}
@@ -592,87 +578,45 @@ public class CqlTemplate extends CassandraAccessor implements CqlOperations {
}
@Override
public void executeAsynchronously(String cql) throws DataAccessException {
executeAsynchronously(cql, (QueryOptions) null);
public ResultSetFuture executeAsynchronously(String cql) throws DataAccessException {
return executeAsynchronously(cql, (QueryOptions) null);
}
@Override
public void executeAsynchronously(final String cql, QueryOptions options) throws DataAccessException {
doExecuteAsync(addQueryOptions(new SimpleStatement(cql), options));
public ResultSetFuture executeAsynchronously(final String cql, QueryOptions options) throws DataAccessException {
return doExecuteAsync(addQueryOptions(new SimpleStatement(cql), options));
}
@Override
public void executeAsynchronously(String cql, Runnable listener) throws DataAccessException {
executeAsynchronously(cql, listener, new Executor() {
public QueryCancellor executeAsynchronously(String cql, Runnable listener) throws DataAccessException {
return executeAsynchronously(cql, listener, new Executor() {
@Override
public void execute(Runnable command) {
command.run();
}
});
}
@Override
public void executeAsynchronously(final String cql, final Runnable listener, final Executor executor)
public QueryCancellor executeAsynchronously(final String cql, final Runnable listener, final Executor executor)
throws DataAccessException {
execute(new SessionCallback<Object>() {
return execute(new SessionCallback<QueryCancellor>() {
@Override
public Object doInSession(Session s) throws DataAccessException {
public QueryCancellor doInSession(Session s) throws DataAccessException {
Statement statement = new SimpleStatement(cql);
final ResultSetFuture rsf = s.executeAsync(statement);
rsf.addListener(listener, executor);
return null;
return new BasicQueryCancellor(rsf);
}
});
}
@Override
public void executeAsynchronously(String cql, AsynchronousQueryListener listener) throws DataAccessException {
executeAsynchronously(cql, listener, new Executor() {
@Override
public void execute(Runnable command) {
command.run();
}
});
}
@Override
public void executeAsynchronously(final String cql, final AsynchronousQueryListener listener, final Executor executor)
public QueryCancellor executeAsynchronously(String cql, AsynchronousQueryListener listener)
throws DataAccessException {
execute(new SessionCallback<Object>() {
@Override
public Object doInSession(Session s) throws DataAccessException {
Statement statement = new SimpleStatement(cql);
final ResultSetFuture rsf = s.executeAsync(statement);
Runnable wrapper = new Runnable() {
@Override
public void run() {
listener.onQueryComplete(rsf);
}
};
rsf.addListener(wrapper, executor);
return null;
}
});
}
@Override
public void executeAsynchronously(Statement query) throws DataAccessException {
doExecuteAsync(query);
}
@Override
public void executeAsynchronously(Statement query, Runnable listener) throws DataAccessException {
executeAsynchronously(query, listener, new Executor() {
return executeAsynchronously(cql, listener, new Executor() {
@Override
public void execute(Runnable command) {
command.run();
@@ -681,36 +625,72 @@ public class CqlTemplate extends CassandraAccessor implements CqlOperations {
}
@Override
public void executeAsynchronously(Statement query, AsynchronousQueryListener listener) throws DataAccessException {
executeAsynchronously(query, listener, new Executor() {
@Override
public void execute(Runnable command) {
command.run();
}
});
}
@Override
public void executeAsynchronously(final Statement query, final Runnable listener, final Executor executor)
throws DataAccessException {
execute(new SessionCallback<Object>() {
@Override
public Object doInSession(Session s) throws DataAccessException {
final ResultSetFuture rsf = s.executeAsync(query);
rsf.addListener(listener, executor);
return null;
}
});
}
@Override
public void executeAsynchronously(final Statement query, final AsynchronousQueryListener listener,
public QueryCancellor executeAsynchronously(final String cql, final AsynchronousQueryListener listener,
final Executor executor) throws DataAccessException {
execute(new SessionCallback<Object>() {
return execute(new SessionCallback<QueryCancellor>() {
@Override
public Object doInSession(Session s) throws DataAccessException {
public QueryCancellor doInSession(Session s) throws DataAccessException {
Statement statement = new SimpleStatement(cql);
final ResultSetFuture rsf = s.executeAsync(statement);
Runnable wrapper = new Runnable() {
@Override
public void run() {
listener.onQueryComplete(rsf);
}
};
rsf.addListener(wrapper, executor);
return new BasicQueryCancellor(rsf);
}
});
}
@Override
public ResultSetFuture executeAsynchronously(Statement query) throws DataAccessException {
return doExecuteAsync(query);
}
@Override
public QueryCancellor executeAsynchronously(Statement query, Runnable listener) throws DataAccessException {
return executeAsynchronously(query, listener, new Executor() {
@Override
public void execute(Runnable command) {
command.run();
}
});
}
@Override
public QueryCancellor executeAsynchronously(Statement query, AsynchronousQueryListener listener)
throws DataAccessException {
return executeAsynchronously(query, listener, new Executor() {
@Override
public void execute(Runnable command) {
command.run();
}
});
}
@Override
public QueryCancellor executeAsynchronously(final Statement query, final Runnable listener, final Executor executor)
throws DataAccessException {
return execute(new SessionCallback<QueryCancellor>() {
@Override
public QueryCancellor doInSession(Session s) throws DataAccessException {
final ResultSetFuture rsf = s.executeAsync(query);
rsf.addListener(listener, executor);
return new BasicQueryCancellor(rsf);
}
});
}
@Override
public QueryCancellor executeAsynchronously(final Statement query, final AsynchronousQueryListener listener,
final Executor executor) throws DataAccessException {
return execute(new SessionCallback<QueryCancellor>() {
@Override
public QueryCancellor doInSession(Session s) throws DataAccessException {
final ResultSetFuture rsf = s.executeAsync(query);
Runnable wrapper = new Runnable() {
@Override
@@ -719,10 +699,9 @@ public class CqlTemplate extends CassandraAccessor implements CqlOperations {
}
};
rsf.addListener(wrapper, executor);
return null;
return new BasicQueryCancellor(rsf);
}
});
}
@Override
@@ -915,8 +894,9 @@ public class CqlTemplate extends CassandraAccessor implements CqlOperations {
PreparedStatement preparedStatement = cpsc.createPreparedStatement(getSession());
addPreparedStatementOptions(preparedStatement, options);
Session s = getSession();
while (rowIterator.hasNext()) {
getSession().executeAsync(preparedStatement.bind(rowIterator.next()));
s.executeAsync(preparedStatement.bind(rowIterator.next()));
}
}
@@ -1227,34 +1207,33 @@ public class CqlTemplate extends CassandraAccessor implements CqlOperations {
}
@Override
public void executeAsynchronously(Truncate truncate) throws DataAccessException {
doExecuteAsync(truncate);
public ResultSetFuture executeAsynchronously(Truncate truncate) throws DataAccessException {
return doExecuteAsync(truncate);
}
@Override
public void executeAsynchronously(Delete delete) throws DataAccessException {
doExecuteAsync(delete);
public ResultSetFuture executeAsynchronously(Delete delete) throws DataAccessException {
return doExecuteAsync(delete);
}
@Override
public void executeAsynchronously(Insert insert) throws DataAccessException {
doExecuteAsync(insert);
public ResultSetFuture executeAsynchronously(Insert insert) throws DataAccessException {
return doExecuteAsync(insert);
}
@Override
public void executeAsynchronously(Update update) throws DataAccessException {
doExecuteAsync(update);
public ResultSetFuture executeAsynchronously(Update update) throws DataAccessException {
return doExecuteAsync(update);
}
@Override
public void executeAsynchronously(Batch batch) throws DataAccessException {
doExecuteAsync(batch);
public ResultSetFuture executeAsynchronously(Batch batch) throws DataAccessException {
return doExecuteAsync(batch);
}
@Override
public ResultSetFuture queryAsynchronously(final Select select) {
return execute(new SessionCallback<ResultSetFuture>() {
@Override
public ResultSetFuture doInSession(Session s) throws DataAccessException {
return s.executeAsync(select);
@@ -1263,9 +1242,8 @@ public class CqlTemplate extends CassandraAccessor implements CqlOperations {
}
@Override
public void queryAsynchronously(Select select, Runnable listener) {
queryAsynchronously(select, listener, new Executor() {
public QueryCancellor queryAsynchronously(Select select, Runnable listener) {
return queryAsynchronously(select, listener, new Executor() {
@Override
public void execute(Runnable command) {
command.run();
@@ -1274,9 +1252,8 @@ public class CqlTemplate extends CassandraAccessor implements CqlOperations {
}
@Override
public void queryAsynchronously(Select select, AsynchronousQueryListener listener) {
queryAsynchronously(select, listener, new Executor() {
public QueryCancellor queryAsynchronously(Select select, AsynchronousQueryListener listener) {
return queryAsynchronously(select, listener, new Executor() {
@Override
public void execute(Runnable command) {
command.run();
@@ -1285,10 +1262,12 @@ public class CqlTemplate extends CassandraAccessor implements CqlOperations {
}
@Override
public void queryAsynchronously(final Select select, final AsynchronousQueryListener listener, final Executor executor) {
execute(new SessionCallback<Object>() {
public QueryCancellor queryAsynchronously(final Select select, final AsynchronousQueryListener listener,
final Executor executor) {
return execute(new SessionCallback<QueryCancellor>() {
@Override
public Object doInSession(Session s) throws DataAccessException {
public QueryCancellor doInSession(Session s) throws DataAccessException {
final ResultSetFuture rsf = s.executeAsync(select);
Runnable wrapper = new Runnable() {
@Override
@@ -1297,19 +1276,19 @@ public class CqlTemplate extends CassandraAccessor implements CqlOperations {
}
};
rsf.addListener(wrapper, executor);
return null;
return new BasicQueryCancellor(rsf);
}
});
}
@Override
public void queryAsynchronously(final Select select, final Runnable listener, final Executor executor) {
execute(new SessionCallback<Object>() {
public QueryCancellor queryAsynchronously(final Select select, final Runnable listener, final Executor executor) {
return execute(new SessionCallback<QueryCancellor>() {
@Override
public Object doInSession(Session s) throws DataAccessException {
public QueryCancellor doInSession(Session s) throws DataAccessException {
ResultSetFuture rsf = s.executeAsync(select);
rsf.addListener(listener, executor);
return null;
return new BasicQueryCancellor(rsf);
}
});
}
@@ -1317,7 +1296,6 @@ public class CqlTemplate extends CassandraAccessor implements CqlOperations {
@Override
public ResultSet query(Select select) {
return query(select, new ResultSetExtractor<ResultSet>() {
@Override
public ResultSet extractData(ResultSet rs) throws DriverException, DataAccessException {
return rs;

26
spring-cql/src/main/java/org/springframework/cassandra/core/QueryCancellor.java Normal file
View File

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