diff --git a/spring-cql/src/main/java/org/springframework/cassandra/core/BasicQueryCancellor.java b/spring-cql/src/main/java/org/springframework/cassandra/core/BasicQueryCancellor.java new file mode 100644 index 000000000..d099b16b9 --- /dev/null +++ b/spring-cql/src/main/java/org/springframework/cassandra/core/BasicQueryCancellor.java @@ -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); + } +} diff --git a/spring-cql/src/main/java/org/springframework/cassandra/core/CqlOperations.java b/spring-cql/src/main/java/org/springframework/cassandra/core/CqlOperations.java index 704950028..5f9263564 100644 --- a/spring-cql/src/main/java/org/springframework/cassandra/core/CqlOperations.java +++ b/spring-cql/src/main/java/org/springframework/cassandra/core/CqlOperations.java @@ -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}. diff --git a/spring-cql/src/main/java/org/springframework/cassandra/core/CqlTemplate.java b/spring-cql/src/main/java/org/springframework/cassandra/core/CqlTemplate.java index f4d3106a4..c6b73bf91 100644 --- a/spring-cql/src/main/java/org/springframework/cassandra/core/CqlTemplate.java +++ b/spring-cql/src/main/java/org/springframework/cassandra/core/CqlTemplate.java @@ -227,7 +227,6 @@ public class CqlTemplate extends CassandraAccessor implements CqlOperations { @Override public ResultSetFuture queryAsynchronously(final String cql) { return execute(new SessionCallback() { - @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() { + return execute(new SessionCallback() { @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() { + return execute(new SessionCallback() { @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() { + return execute(new SessionCallback() { @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() { - @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() { - @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() { + return execute(new SessionCallback() { @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() { + @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() { + @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() { - @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() { + public QueryCancellor queryAsynchronously(final Select select, final AsynchronousQueryListener listener, + final Executor executor) { + + return execute(new SessionCallback() { @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() { + public QueryCancellor queryAsynchronously(final Select select, final Runnable listener, final Executor executor) { + return execute(new SessionCallback() { @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() { - @Override public ResultSet extractData(ResultSet rs) throws DriverException, DataAccessException { return rs; diff --git a/spring-cql/src/main/java/org/springframework/cassandra/core/QueryCancellor.java b/spring-cql/src/main/java/org/springframework/cassandra/core/QueryCancellor.java new file mode 100644 index 000000000..a4ec1b975 --- /dev/null +++ b/spring-cql/src/main/java/org/springframework/cassandra/core/QueryCancellor.java @@ -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); +}