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

DATACASS-330 - Add SessionFactory to CqlTemplate and AsyncCqlTemplate.

Browse Source 
We now support SessionFactory to obtain Cassandra Session's on a per-request basis. CassandraAccessor is configured primarily with a SessionFactory now, the existing initialization configures a DefaultSessionFactory that returns the initially given Session instance.

Sessions should not be acquired directly by getSession but inside a callback-block so it's guaranteed to operate on the same session within a particular operation. That's especially relevant when preparing and executing prepared statements.

PreparedStatementCallback was changed in a breaking way as it now accepts additionally a Session to retain the session context.

Related ticket: DATACASS-32
This commit is contained in:
Mark Paluch
2017-01-18 18:03:33 +01:00
committed by John Blum
parent 7e33bf3bcf
commit 68d61461cd
12 changed files with 448 additions and 248 deletions
Download Patch File Download Diff File Expand all files Collapse all files

240
spring-cql/src/main/java/org/springframework/cassandra/core/AsyncCqlTemplate.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2016 the original author or authors.
* Copyright 2016-2017 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.
@@ -20,6 +20,15 @@ import java.util.Map;
import java.util.concurrent.ExecutionException;
import java.util.function.Function;
import org.springframework.cassandra.core.session.SessionFactory;
import org.springframework.cassandra.support.CassandraAccessor;
import org.springframework.dao.DataAccessException;
import org.springframework.dao.support.DataAccessUtils;
import org.springframework.dao.support.PersistenceExceptionTranslator;
import org.springframework.util.Assert;
import org.springframework.util.concurrent.ListenableFuture;
import org.springframework.util.concurrent.SettableListenableFuture;
import com.datastax.driver.core.BoundStatement;
import com.datastax.driver.core.PreparedStatement;
import com.datastax.driver.core.ResultSet;
@@ -31,14 +40,6 @@ import com.datastax.driver.core.exceptions.DriverException;
import com.google.common.util.concurrent.FutureCallback;
import com.google.common.util.concurrent.Futures;
import org.springframework.cassandra.support.CassandraAccessor;
import org.springframework.dao.DataAccessException;
import org.springframework.dao.support.DataAccessUtils;
import org.springframework.dao.support.PersistenceExceptionTranslator;
import org.springframework.util.Assert;
import org.springframework.util.concurrent.ListenableFuture;
import org.springframework.util.concurrent.SettableListenableFuture;
/**
* <b>This is the central class in the CQL core package for asynchronous Cassandra data access.</b> It simplifies the
* use of CQL and helps to avoid common errors. It executes core CQL workflow, leaving application code to provide CQL
@@ -79,18 +80,17 @@ import org.springframework.util.concurrent.SettableListenableFuture;
public class AsyncCqlTemplate extends CassandraAccessor implements AsyncCqlOperations {
/**
* Constructs a new, uninitialized {@link AsyncCqlTemplate}.
* Constructs a new, uninitialized {@link AsyncCqlTemplate}. Note: The {@link Session} has to be set before using the
* instance.
*
* Note: The {@link Session} has to be set before using the instance.
*
* @see #setSession(Session)
* @see #setSessionFactory(SessionFactory)
*/
public AsyncCqlTemplate() {}
/**
* Constructs a new {@link AsyncCqlTemplate} with the given {@link Session}.
*
* @param session the active Cassandra {@link Session}.
* @param session the active Cassandra {@link Session}, must not be {@literal null}.
* @throws IllegalStateException if {@link Session} is {@literal null}.
* @see com.datastax.driver.core.Session
*/
@@ -101,6 +101,20 @@ public class AsyncCqlTemplate extends CassandraAccessor implements AsyncCqlOpera
setSession(session);
}
/**
* Constructs a new {@link AsyncCqlTemplate} with the given {@link SessionFactory}.
*
* @param sessionFactory the active Cassandra {@link SessionFactory}.
* @since 2.0
* @see SessionFactory
*/
public AsyncCqlTemplate(SessionFactory sessionFactory) {
Assert.notNull(sessionFactory, "SessionFactory must not be null");
setSessionFactory(sessionFactory);
}
// -------------------------------------------------------------------------
// Methods dealing with a plain com.datastax.driver.core.Session
// -------------------------------------------------------------------------
@@ -115,7 +129,7 @@ public class AsyncCqlTemplate extends CassandraAccessor implements AsyncCqlOpera
Assert.notNull(action, "Callback object must not be null");
try {
return action.doInSession(getSession());
return action.doInSession(getCurrentSession());
} catch (DriverException e) {
throw translateException("SessionCallback", toCql(action), e);
}
@@ -142,7 +156,8 @@ public class AsyncCqlTemplate extends CassandraAccessor implements AsyncCqlOpera
* @see org.springframework.cassandra.core.AsyncCqlOperations#query(java.lang.String, org.springframework.cassandra.core.ResultSetExtractor)
*/
@Override
public <T> ListenableFuture<T> query(String cql, ResultSetExtractor<T> resultSetExtractor) throws DataAccessException {
public <T> ListenableFuture<T> query(String cql, ResultSetExtractor<T> resultSetExtractor)
throws DataAccessException {
Assert.hasText(cql, "CQL must not be empty");
Assert.notNull(resultSetExtractor, "ResultSetExtractor must not be null");
@@ -154,7 +169,7 @@ public class AsyncCqlTemplate extends CassandraAccessor implements AsyncCqlOpera
SimpleStatement simpleStatement = applyStatementSettings(new SimpleStatement(cql));
ResultSetFuture results = getSession().executeAsync(simpleStatement);
ResultSetFuture results = getCurrentSession().executeAsync(simpleStatement);
return new ExceptionTranslatingListenableFutureAdapter<>(new MappingListenableFutureAdapter<>(
new GuavaListenableFutureAdapter<>(results, ex -> translateExceptionIfPossible("Query", cql, ex)),
@@ -173,8 +188,8 @@ public class AsyncCqlTemplate extends CassandraAccessor implements AsyncCqlOpera
ListenableFuture<?> results = query(cql, newResultSetExtractor(rowCallbackHandler));
return new ExceptionTranslatingListenableFutureAdapter<>(new MappingListenableFutureAdapter<>(
results, o -> null), getExceptionTranslator());
return new ExceptionTranslatingListenableFutureAdapter<>(new MappingListenableFutureAdapter<>(results, o -> null),
getExceptionTranslator());
}
/*
@@ -231,8 +246,8 @@ public class AsyncCqlTemplate extends CassandraAccessor implements AsyncCqlOpera
ListenableFuture<List<T>> results = query(cql, newResultSetExtractor(rowMapper));
return new ExceptionTranslatingListenableFutureAdapter<>(new MappingListenableFutureAdapter<>(
results, DataAccessUtils::requiredSingleResult), getExceptionTranslator());
return new ExceptionTranslatingListenableFutureAdapter<>(
new MappingListenableFutureAdapter<>(results, DataAccessUtils::requiredSingleResult), getExceptionTranslator());
}
/*
@@ -265,7 +280,8 @@ public class AsyncCqlTemplate extends CassandraAccessor implements AsyncCqlOpera
* @see org.springframework.cassandra.core.AsyncCqlOperations#query(com.datastax.driver.core.Statement, org.springframework.cassandra.core.ResultSetExtractor)
*/
@Override
public <T> ListenableFuture<T> query(Statement statement, ResultSetExtractor<T> resultSetExtractor) throws DataAccessException {
public <T> ListenableFuture<T> query(Statement statement, ResultSetExtractor<T> resultSetExtractor)
throws DataAccessException {
Assert.notNull(statement, "CQL Statement must not be null");
Assert.notNull(resultSetExtractor, "ResultSetExtractor must not be null");
@@ -275,12 +291,13 @@ public class AsyncCqlTemplate extends CassandraAccessor implements AsyncCqlOpera
logger.debug("Executing CQL Statement [{}]", statement);
}
ResultSetFuture results = getSession().executeAsync(applyStatementSettings(statement));
ResultSetFuture results = getCurrentSession().executeAsync(applyStatementSettings(statement));
return new ExceptionTranslatingListenableFutureAdapter<>(
new MappingListenableFutureAdapter<>(new GuavaListenableFutureAdapter<>(
results, ex -> translateExceptionIfPossible("Query", statement.toString(), ex)),
resultSetExtractor::extractData),
new MappingListenableFutureAdapter<>(
new GuavaListenableFutureAdapter<>(results,
ex -> translateExceptionIfPossible("Query", statement.toString(), ex)),
resultSetExtractor::extractData),
getExceptionTranslator());
} catch (DriverException e) {
throw translateException("Query", statement.toString(), e);
@@ -292,12 +309,13 @@ public class AsyncCqlTemplate extends CassandraAccessor implements AsyncCqlOpera
* @see org.springframework.cassandra.core.AsyncCqlOperations#query(com.datastax.driver.core.Statement, org.springframework.cassandra.core.RowCallbackHandler)
*/
@Override
public ListenableFuture<Void> query(Statement statement, RowCallbackHandler rowCallbackHandler) throws DataAccessException {
public ListenableFuture<Void> query(Statement statement, RowCallbackHandler rowCallbackHandler)
throws DataAccessException {
ListenableFuture<?> result = query(statement, newResultSetExtractor(rowCallbackHandler));
return new ExceptionTranslatingListenableFutureAdapter<>(new MappingListenableFutureAdapter<>(
result, o -> null), getExceptionTranslator());
return new ExceptionTranslatingListenableFutureAdapter<>(new MappingListenableFutureAdapter<>(result, o -> null),
getExceptionTranslator());
}
/*
@@ -323,7 +341,8 @@ public class AsyncCqlTemplate extends CassandraAccessor implements AsyncCqlOpera
* @see org.springframework.cassandra.core.AsyncCqlOperations#queryForList(com.datastax.driver.core.Statement, java.lang.Class)
*/
@Override
public <T> ListenableFuture<List<T>> queryForList(Statement statement, Class<T> elementType) throws DataAccessException {
public <T> ListenableFuture<List<T>> queryForList(Statement statement, Class<T> elementType)
throws DataAccessException {
return query(statement, newResultSetExtractor(newSingleColumnRowMapper(elementType)));
}
@@ -355,8 +374,8 @@ public class AsyncCqlTemplate extends CassandraAccessor implements AsyncCqlOpera
ListenableFuture<List<T>> results = query(statement, newResultSetExtractor(rowMapper));
return new ExceptionTranslatingListenableFutureAdapter<>(new MappingListenableFutureAdapter<>(
results, DataAccessUtils::requiredSingleResult), getExceptionTranslator());
return new ExceptionTranslatingListenableFutureAdapter<>(
new MappingListenableFutureAdapter<>(results, DataAccessUtils::requiredSingleResult), getExceptionTranslator());
}
/*
@@ -377,7 +396,8 @@ public class AsyncCqlTemplate extends CassandraAccessor implements AsyncCqlOpera
* @see org.springframework.cassandra.core.AsyncCqlOperations#execute(org.springframework.cassandra.core.AsyncPreparedStatementCreator)
*/
@Override
public ListenableFuture<Boolean> execute(AsyncPreparedStatementCreator preparedStatementCreator) throws DataAccessException {
public ListenableFuture<Boolean> execute(AsyncPreparedStatementCreator preparedStatementCreator)
throws DataAccessException {
return query(preparedStatementCreator, ResultSet::wasApplied);
}
@@ -395,7 +415,8 @@ public class AsyncCqlTemplate extends CassandraAccessor implements AsyncCqlOpera
* @see org.springframework.cassandra.core.AsyncCqlOperations#execute(java.lang.String, org.springframework.cassandra.core.PreparedStatementBinder)
*/
@Override
public ListenableFuture<Boolean> execute(String cql, PreparedStatementBinder preparedStatementBinder) throws DataAccessException {
public ListenableFuture<Boolean> execute(String cql, PreparedStatementBinder preparedStatementBinder)
throws DataAccessException {
return query(newAsyncPreparedStatementCreator(cql), preparedStatementBinder, ResultSet::wasApplied);
}
@@ -413,27 +434,26 @@ public class AsyncCqlTemplate extends CassandraAccessor implements AsyncCqlOpera
* @see org.springframework.cassandra.core.AsyncCqlOperations#execute(org.springframework.cassandra.core.AsyncPreparedStatementCreator, org.springframework.cassandra.core.PreparedStatementCallback)
*/
@Override
public <T> ListenableFuture<T> execute(AsyncPreparedStatementCreator preparedStatementCreator, PreparedStatementCallback<T> action)
throws DataAccessException {
public <T> ListenableFuture<T> execute(AsyncPreparedStatementCreator preparedStatementCreator,
PreparedStatementCallback<T> action) throws DataAccessException {
Assert.notNull(preparedStatementCreator, "PreparedStatementCreator must not be null");
Assert.notNull(action, "PreparedStatementCallback object must not be null");
try {
if (logger.isDebugEnabled()) {
logger.debug("Preparing statement [{}] using {}", toCql(preparedStatementCreator),
preparedStatementCreator);
logger.debug("Preparing statement [{}] using {}", toCql(preparedStatementCreator), preparedStatementCreator);
}
return new ExceptionTranslatingListenableFutureAdapter<>(
new MappingListenableFutureAdapter<>(preparedStatementCreator.createPreparedStatement(getSession()),
preparedStatement -> {
try {
return action.doInPreparedStatement(applyStatementSettings(preparedStatement));
} catch (DriverException e) {
throw translateException("PreparedStatementCallback", preparedStatement.toString(), e);
}
}), getExceptionTranslator());
Session currentSession = getCurrentSession();
return new ExceptionTranslatingListenableFutureAdapter<>(new MappingListenableFutureAdapter<>(
preparedStatementCreator.createPreparedStatement(currentSession), preparedStatement -> {
try {
return action.doInPreparedStatement(currentSession, applyStatementSettings(preparedStatement));
} catch (DriverException e) {
throw translateException("PreparedStatementCallback", preparedStatement.toString(), e);
}
}), getExceptionTranslator());
} catch (DriverException e) {
throw translateException("PreparedStatementCallback", toCql(preparedStatementCreator), e);
@@ -445,8 +465,8 @@ public class AsyncCqlTemplate extends CassandraAccessor implements AsyncCqlOpera
* @see org.springframework.cassandra.core.AsyncCqlOperations#query(org.springframework.cassandra.core.AsyncPreparedStatementCreator, org.springframework.cassandra.core.ResultSetExtractor)
*/
@Override
public <T> ListenableFuture<T> query(AsyncPreparedStatementCreator preparedStatementCreator, ResultSetExtractor<T> resultSetExtractor)
throws DataAccessException {
public <T> ListenableFuture<T> query(AsyncPreparedStatementCreator preparedStatementCreator,
ResultSetExtractor<T> resultSetExtractor) throws DataAccessException {
return query(preparedStatementCreator, null, resultSetExtractor);
}
@@ -456,13 +476,13 @@ public class AsyncCqlTemplate extends CassandraAccessor implements AsyncCqlOpera
* @see org.springframework.cassandra.core.AsyncCqlOperations#query(org.springframework.cassandra.core.AsyncPreparedStatementCreator, org.springframework.cassandra.core.RowCallbackHandler)
*/
@Override
public ListenableFuture<Void> query(AsyncPreparedStatementCreator preparedStatementCreator, RowCallbackHandler rowCallbackHandler)
throws DataAccessException {
public ListenableFuture<Void> query(AsyncPreparedStatementCreator preparedStatementCreator,
RowCallbackHandler rowCallbackHandler) throws DataAccessException {
ListenableFuture<?> results = query(preparedStatementCreator, null, newResultSetExtractor(rowCallbackHandler));
return new ExceptionTranslatingListenableFutureAdapter<>(new MappingListenableFutureAdapter<>(
results, o -> null), getExceptionTranslator());
return new ExceptionTranslatingListenableFutureAdapter<>(new MappingListenableFutureAdapter<>(results, o -> null),
getExceptionTranslator());
}
/*
@@ -470,8 +490,8 @@ public class AsyncCqlTemplate extends CassandraAccessor implements AsyncCqlOpera
* @see org.springframework.cassandra.core.AsyncCqlOperations#query(org.springframework.cassandra.core.AsyncPreparedStatementCreator, org.springframework.cassandra.core.RowMapper)
*/
@Override
public <T> ListenableFuture<List<T>> query(AsyncPreparedStatementCreator preparedStatementCreator, RowMapper<T> rowMapper)
throws DataAccessException {
public <T> ListenableFuture<List<T>> query(AsyncPreparedStatementCreator preparedStatementCreator,
RowMapper<T> rowMapper) throws DataAccessException {
return query(preparedStatementCreator, null, newResultSetExtractor(rowMapper));
}
@@ -482,21 +502,21 @@ public class AsyncCqlTemplate extends CassandraAccessor implements AsyncCqlOpera
*/
@Override
public <T> ListenableFuture<T> query(AsyncPreparedStatementCreator preparedStatementCreator,
PreparedStatementBinder preparedStatementBinder, ResultSetExtractor<T> resultSetExtractor) throws DataAccessException {
PreparedStatementBinder preparedStatementBinder, ResultSetExtractor<T> resultSetExtractor)
throws DataAccessException {
Assert.notNull(preparedStatementCreator, "AsyncPreparedStatementCreator must not be null");
Assert.notNull(resultSetExtractor, "ResultSetExtractor object must not be null");
try {
if (logger.isDebugEnabled()) {
logger.debug("Preparing statement [{}] using {}", toCql(preparedStatementCreator),
preparedStatementCreator);
logger.debug("Preparing statement [{}] using {}", toCql(preparedStatementCreator), preparedStatementCreator);
}
Session session = getSession();
Session session = getCurrentSession();
PersistenceExceptionTranslator exceptionTranslator = ex -> translateExceptionIfPossible(
"Query", toCql(preparedStatementCreator), ex);
PersistenceExceptionTranslator exceptionTranslator = ex -> translateExceptionIfPossible("Query",
toCql(preparedStatementCreator), ex);
ListenableFuture<BoundStatement> statementFuture = new MappingListenableFutureAdapter<>(
preparedStatementCreator.createPreparedStatement(session), preparedStatement -> {
@@ -505,39 +525,39 @@ public class AsyncCqlTemplate extends CassandraAccessor implements AsyncCqlOpera
}
return applyStatementSettings(preparedStatementBinder != null
? preparedStatementBinder.bindValues(preparedStatement) : preparedStatement.bind());
? preparedStatementBinder.bindValues(preparedStatement) : preparedStatement.bind());
});
SettableListenableFuture<T> settableListenableFuture = new SettableListenableFuture<>();
statementFuture.addCallback(boundStatement -> Futures.addCallback(session.executeAsync(boundStatement),
new FutureCallback<ResultSet>() {
@Override
public void onSuccess(ResultSet result) {
try {
settableListenableFuture.set(resultSetExtractor.extractData(result));
} catch (DriverException e) {
settableListenableFuture.setException(exceptionTranslator.translateExceptionIfPossible(e));
statementFuture.addCallback(
boundStatement -> Futures.addCallback(session.executeAsync(boundStatement), new FutureCallback<ResultSet>() {
@Override
public void onSuccess(ResultSet result) {
try {
settableListenableFuture.set(resultSetExtractor.extractData(result));
} catch (DriverException e) {
settableListenableFuture.setException(exceptionTranslator.translateExceptionIfPossible(e));
}
}
}
@Override
public void onFailure(Throwable ex) {
@Override
public void onFailure(Throwable ex) {
if (ex instanceof DriverException) {
settableListenableFuture
.setException(exceptionTranslator.translateExceptionIfPossible((DriverException) ex));
} else {
settableListenableFuture.setException(ex);
}
}
}), ex -> {
if (ex instanceof DriverException) {
settableListenableFuture.setException(
exceptionTranslator.translateExceptionIfPossible((DriverException) ex));
settableListenableFuture
.setException(exceptionTranslator.translateExceptionIfPossible((DriverException) ex));
} else {
settableListenableFuture.setException(ex);
}
}
}), ex -> {
if (ex instanceof DriverException) {
settableListenableFuture.setException(
exceptionTranslator.translateExceptionIfPossible((DriverException) ex));
} else {
settableListenableFuture.setException(ex);
}
});
});
return settableListenableFuture;
@@ -552,13 +572,14 @@ public class AsyncCqlTemplate extends CassandraAccessor implements AsyncCqlOpera
*/
@Override
public ListenableFuture<Void> query(AsyncPreparedStatementCreator preparedStatementCreator,
PreparedStatementBinder preparedStatementBinder, RowCallbackHandler rowCallbackHandler) throws DataAccessException {
PreparedStatementBinder preparedStatementBinder, RowCallbackHandler rowCallbackHandler)
throws DataAccessException {
ListenableFuture<?> results = query(preparedStatementCreator, preparedStatementBinder,
newResultSetExtractor(rowCallbackHandler));
newResultSetExtractor(rowCallbackHandler));
return new ExceptionTranslatingListenableFutureAdapter<>(new MappingListenableFutureAdapter<>(
results, o -> null), getExceptionTranslator());
return new ExceptionTranslatingListenableFutureAdapter<>(new MappingListenableFutureAdapter<>(results, o -> null),
getExceptionTranslator());
}
/*
@@ -592,10 +613,10 @@ public class AsyncCqlTemplate extends CassandraAccessor implements AsyncCqlOpera
throws DataAccessException {
ListenableFuture<?> results = query(newAsyncPreparedStatementCreator(cql), newPreparedStatementBinder(args),
newResultSetExtractor(rowCallbackHandler));
newResultSetExtractor(rowCallbackHandler));
return new ExceptionTranslatingListenableFutureAdapter<>(new MappingListenableFutureAdapter<>(
results, o -> null), getExceptionTranslator());
return new ExceptionTranslatingListenableFutureAdapter<>(new MappingListenableFutureAdapter<>(results, o -> null),
getExceptionTranslator());
}
/*
@@ -604,10 +625,10 @@ public class AsyncCqlTemplate extends CassandraAccessor implements AsyncCqlOpera
*/
@Override
public <T> ListenableFuture<List<T>> query(String cql, RowMapper<T> rowMapper, Object... args)
throws DataAccessException {
throws DataAccessException {
return query(newAsyncPreparedStatementCreator(cql), newPreparedStatementBinder(args),
newResultSetExtractor(rowMapper));
newResultSetExtractor(rowMapper));
}
/*
@@ -630,10 +651,10 @@ public class AsyncCqlTemplate extends CassandraAccessor implements AsyncCqlOpera
RowCallbackHandler rowCallbackHandler) throws DataAccessException {
ListenableFuture<?> results = query(newAsyncPreparedStatementCreator(cql), preparedStatementBinder,
newResultSetExtractor(rowCallbackHandler));
newResultSetExtractor(rowCallbackHandler));
return new ExceptionTranslatingListenableFutureAdapter<>(new MappingListenableFutureAdapter<>(
results, o -> null), getExceptionTranslator());
return new ExceptionTranslatingListenableFutureAdapter<>(new MappingListenableFutureAdapter<>(results, o -> null),
getExceptionTranslator());
}
/*
@@ -644,8 +665,7 @@ public class AsyncCqlTemplate extends CassandraAccessor implements AsyncCqlOpera
public <T> ListenableFuture<List<T>> query(String cql, PreparedStatementBinder preparedStatementBinder,
RowMapper<T> rowMapper) throws DataAccessException {
return query(newAsyncPreparedStatementCreator(cql), preparedStatementBinder,
newResultSetExtractor(rowMapper));
return query(newAsyncPreparedStatementCreator(cql), preparedStatementBinder, newResultSetExtractor(rowMapper));
}
/*
@@ -657,7 +677,7 @@ public class AsyncCqlTemplate extends CassandraAccessor implements AsyncCqlOpera
throws DataAccessException {
return query(newAsyncPreparedStatementCreator(cql), newPreparedStatementBinder(args),
newResultSetExtractor(newColumnMapRowMapper()));
newResultSetExtractor(newColumnMapRowMapper()));
}
/*
@@ -669,7 +689,7 @@ public class AsyncCqlTemplate extends CassandraAccessor implements AsyncCqlOpera
throws DataAccessException {
return query(newAsyncPreparedStatementCreator(cql), newPreparedStatementBinder(args),
newResultSetExtractor(newSingleColumnRowMapper(elementType)));
newResultSetExtractor(newSingleColumnRowMapper(elementType)));
}
/*
@@ -700,11 +720,11 @@ public class AsyncCqlTemplate extends CassandraAccessor implements AsyncCqlOpera
public <T> ListenableFuture<T> queryForObject(String cql, RowMapper<T> rowMapper, Object... args)
throws DataAccessException {
ListenableFuture<List<T>> results = query(newAsyncPreparedStatementCreator(cql),
newPreparedStatementBinder(args), newResultSetExtractor(rowMapper, 1));
ListenableFuture<List<T>> results = query(newAsyncPreparedStatementCreator(cql), newPreparedStatementBinder(args),
newResultSetExtractor(rowMapper, 1));
return new ExceptionTranslatingListenableFutureAdapter<>(new MappingListenableFutureAdapter<>(
results, DataAccessUtils::requiredSingleResult), getExceptionTranslator());
return new ExceptionTranslatingListenableFutureAdapter<>(
new MappingListenableFutureAdapter<>(results, DataAccessUtils::requiredSingleResult), getExceptionTranslator());
}
/*
@@ -729,7 +749,7 @@ public class AsyncCqlTemplate extends CassandraAccessor implements AsyncCqlOpera
*/
protected AsyncPreparedStatementCreator newAsyncPreparedStatementCreator(String cql) {
return new SimpleAsyncPreparedStatementCreator(cql,
ex -> translateExceptionIfPossible("PrepareStatement", cql, ex));
ex -> translateExceptionIfPossible("PrepareStatement", cql, ex));
}
/**
@@ -760,13 +780,18 @@ public class AsyncCqlTemplate extends CassandraAccessor implements AsyncCqlOpera
return (ex instanceof DriverException ? translate(task, cql, (DriverException) ex) : null);
}
private Session getCurrentSession() {
return getSessionFactory().getSession();
}
private static class SimpleAsyncPreparedStatementCreator implements AsyncPreparedStatementCreator, CqlProvider {
private final PersistenceExceptionTranslator persistenceExceptionTranslator;
private final String cql;
private SimpleAsyncPreparedStatementCreator(String cql, PersistenceExceptionTranslator persistenceExceptionTranslator) {
private SimpleAsyncPreparedStatementCreator(String cql,
PersistenceExceptionTranslator persistenceExceptionTranslator) {
Assert.hasText(cql, "CQL must not be empty");
@@ -782,8 +807,7 @@ public class AsyncCqlTemplate extends CassandraAccessor implements AsyncCqlOpera
@Override
public ListenableFuture<PreparedStatement> createPreparedStatement(Session session) throws DriverException {
return new GuavaListenableFutureAdapter<>(session.prepareAsync(getCql()),
this.persistenceExceptionTranslator);
return new GuavaListenableFutureAdapter<>(session.prepareAsync(getCql()), this.persistenceExceptionTranslator);
}
}

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

@@ -1,5 +1,5 @@
/*
* Copyright 2016 the original author or authors.
* Copyright 2016-2017 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.
@@ -21,6 +21,12 @@ import java.util.Map;
import java.util.Set;
import java.util.function.Function;
import org.springframework.cassandra.core.session.SessionFactory;
import org.springframework.cassandra.support.CassandraAccessor;
import org.springframework.dao.DataAccessException;
import org.springframework.dao.support.DataAccessUtils;
import org.springframework.util.Assert;
import com.datastax.driver.core.BoundStatement;
import com.datastax.driver.core.Host;
import com.datastax.driver.core.PreparedStatement;
@@ -31,11 +37,6 @@ import com.datastax.driver.core.SimpleStatement;
import com.datastax.driver.core.Statement;
import com.datastax.driver.core.exceptions.DriverException;
import org.springframework.cassandra.support.CassandraAccessor;
import org.springframework.dao.DataAccessException;
import org.springframework.dao.support.DataAccessUtils;
import org.springframework.util.Assert;
/**
* <b>This is the central class in the CQL core package.</b> It simplifies the use of CQL and helps to avoid common
* errors. It executes core CQL workflow, leaving application code to provide CQL and extract results. This class
@@ -79,11 +80,10 @@ import org.springframework.util.Assert;
public class CqlTemplate extends CassandraAccessor implements CqlOperations {
/**
* Constructs a new, uninitialized {@link CqlTemplate}.
* Constructs a new, uninitialized {@link CqlTemplate}. Note: The {@link Session} has to be set before using the
* instance.
*
* Note: The {@link Session} has to be set before using the instance.
*
* @see #setSession(Session)
* @see #setSessionFactory(SessionFactory)
*/
public CqlTemplate() {}
@@ -101,6 +101,19 @@ public class CqlTemplate extends CassandraAccessor implements CqlOperations {
setSession(session);
}
/**
* Constructs a new {@link CqlTemplate} with the given {@link SessionFactory}.
*
* @param sessionFactory the active Cassandra {@link SessionFactory}, must not be {@literal null}.
* @see SessionFactory
*/
public CqlTemplate(SessionFactory sessionFactory) {
Assert.notNull(sessionFactory, "SessionFactory must not be null");
setSessionFactory(sessionFactory);
}
// -------------------------------------------------------------------------
// Methods dealing with a plain com.datastax.driver.core.Session
// -------------------------------------------------------------------------
@@ -115,7 +128,7 @@ public class CqlTemplate extends CassandraAccessor implements CqlOperations {
Assert.notNull(action, "Callback object must not be null");
try {
return action.doInSession(getSession());
return action.doInSession(getCurrentSession());
} catch (DriverException e) {
throw translateException("SessionCallback", toCql(action), e);
}
@@ -154,7 +167,7 @@ public class CqlTemplate extends CassandraAccessor implements CqlOperations {
SimpleStatement statement = applyStatementSettings(new SimpleStatement(cql));
ResultSet results = getSession().execute(statement);
ResultSet results = getCurrentSession().execute(statement);
return resultSetExtractor.extractData(results);
} catch (DriverException e) {
@@ -274,7 +287,7 @@ public class CqlTemplate extends CassandraAccessor implements CqlOperations {
logger.debug("Executing CQL Statement [{}]", statement);
}
return resultSetExtractor.extractData(getSession().execute(applyStatementSettings(statement)));
return resultSetExtractor.extractData(getCurrentSession().execute(applyStatementSettings(statement)));
} catch (DriverException e) {
throw translateException("Query", statement.toString(), e);
}
@@ -417,8 +430,9 @@ public class CqlTemplate extends CassandraAccessor implements CqlOperations {
logger.debug("Preparing statement [{}] using {}", toCql(preparedStatementCreator), preparedStatementCreator);
}
return action.doInPreparedStatement(applyStatementSettings(
preparedStatementCreator.createPreparedStatement(getSession())));
Session session = getCurrentSession();
return action.doInPreparedStatement(session,
applyStatementSettings(preparedStatementCreator.createPreparedStatement(session)));
} catch (DriverException e) {
throw translateException("PreparedStatementCallback", toCql(preparedStatementCreator), e);
@@ -430,7 +444,8 @@ public class CqlTemplate extends CassandraAccessor implements CqlOperations {
* @see org.springframework.cassandra.core.CqlOperationsNG#query(org.springframework.cassandra.core.PreparedStatementCreator, org.springframework.cassandra.core.ResultSetExtractor)
*/
@Override
public <T> T query(PreparedStatementCreator preparedStatementCreator, ResultSetExtractor<T> resultSetExtractor) throws DataAccessException {
public <T> T query(PreparedStatementCreator preparedStatementCreator, ResultSetExtractor<T> resultSetExtractor)
throws DataAccessException {
return query(preparedStatementCreator, null, resultSetExtractor);
}
@@ -439,7 +454,8 @@ public class CqlTemplate extends CassandraAccessor implements CqlOperations {
* @see org.springframework.cassandra.core.CqlOperationsNG#query(org.springframework.cassandra.core.PreparedStatementCreator, org.springframework.cassandra.core.RowCallbackHandler)
*/
@Override
public void query(PreparedStatementCreator preparedStatementCreator, RowCallbackHandler rowCallbackHandler) throws DataAccessException {
public void query(PreparedStatementCreator preparedStatementCreator, RowCallbackHandler rowCallbackHandler)
throws DataAccessException {
query(preparedStatementCreator, null, newResultSetExtractor(rowCallbackHandler));
}
@@ -448,7 +464,8 @@ public class CqlTemplate extends CassandraAccessor implements CqlOperations {
* @see org.springframework.cassandra.core.CqlOperationsNG#query(org.springframework.cassandra.core.PreparedStatementCreator, org.springframework.cassandra.core.RowMapper)
*/
@Override
public <T> List<T> query(PreparedStatementCreator preparedStatementCreator, RowMapper<T> rowMapper) throws DataAccessException {
public <T> List<T> query(PreparedStatementCreator preparedStatementCreator, RowMapper<T> rowMapper)
throws DataAccessException {
return query(preparedStatementCreator, null, newResultSetExtractor(rowMapper));
}
@@ -468,7 +485,7 @@ public class CqlTemplate extends CassandraAccessor implements CqlOperations {
logger.debug("Preparing statement [{}] using {}", toCql(preparedStatementCreator), preparedStatementCreator);
}
Session session = getSession();
Session session = getCurrentSession();
PreparedStatement preparedStatement = preparedStatementCreator.createPreparedStatement(session);
@@ -477,7 +494,7 @@ public class CqlTemplate extends CassandraAccessor implements CqlOperations {
}
BoundStatement boundStatement = applyStatementSettings(preparedStatementBinder != null
? preparedStatementBinder.bindValues(preparedStatement) : preparedStatement.bind());
? preparedStatementBinder.bindValues(preparedStatement) : preparedStatement.bind());
ResultSet results = session.execute(boundStatement);
@@ -504,8 +521,8 @@ public class CqlTemplate extends CassandraAccessor implements CqlOperations {
* @see org.springframework.cassandra.core.CqlOperationsNG#query(org.springframework.cassandra.core.PreparedStatementCreator, org.springframework.cassandra.core.PreparedStatementBinder, org.springframework.cassandra.core.RowMapper)
*/
@Override
public <T> List<T> query(PreparedStatementCreator preparedStatementCreator, PreparedStatementBinder preparedStatementBinder,
RowMapper<T> rowMapper) throws DataAccessException {
public <T> List<T> query(PreparedStatementCreator preparedStatementCreator,
PreparedStatementBinder preparedStatementBinder, RowMapper<T> rowMapper) throws DataAccessException {
return query(preparedStatementCreator, preparedStatementBinder, newResultSetExtractor(rowMapper));
}
@@ -525,7 +542,8 @@ public class CqlTemplate extends CassandraAccessor implements CqlOperations {
*/
@Override
public void query(String cql, RowCallbackHandler rowCallbackHandler, Object... args) throws DataAccessException {
query(newPreparedStatementCreator(cql), newPreparedStatementBinder(args), newResultSetExtractor(rowCallbackHandler));
query(newPreparedStatementCreator(cql), newPreparedStatementBinder(args),
newResultSetExtractor(rowCallbackHandler));
}
/*
@@ -542,8 +560,8 @@ public class CqlTemplate extends CassandraAccessor implements CqlOperations {
* @see org.springframework.cassandra.core.CqlOperationsNG#query(java.lang.String, org.springframework.cassandra.core.PreparedStatementBinder, org.springframework.cassandra.core.ResultSetExtractor)
*/
@Override
public <T> T query(String cql, PreparedStatementBinder preparedStatementBinder, ResultSetExtractor<T> resultSetExtractor)
throws DataAccessException {
public <T> T query(String cql, PreparedStatementBinder preparedStatementBinder,
ResultSetExtractor<T> resultSetExtractor) throws DataAccessException {
return query(newPreparedStatementCreator(cql), preparedStatementBinder, resultSetExtractor);
}
@@ -577,7 +595,7 @@ public class CqlTemplate extends CassandraAccessor implements CqlOperations {
@Override
public List<Map<String, Object>> queryForList(String cql, Object... args) throws DataAccessException {
return query(newPreparedStatementCreator(cql), newPreparedStatementBinder(args),
newResultSetExtractor(newColumnMapRowMapper()));
newResultSetExtractor(newColumnMapRowMapper()));
}
/*
@@ -587,7 +605,7 @@ public class CqlTemplate extends CassandraAccessor implements CqlOperations {
@Override
public <T> List<T> queryForList(String cql, Class<T> elementType, Object... args) throws DataAccessException {
return query(newPreparedStatementCreator(cql), newPreparedStatementBinder(args),
newResultSetExtractor(newSingleColumnRowMapper(elementType)));
newResultSetExtractor(newSingleColumnRowMapper(elementType)));
}
/*
@@ -614,8 +632,8 @@ public class CqlTemplate extends CassandraAccessor implements CqlOperations {
*/
@Override
public <T> T queryForObject(String cql, RowMapper<T> rowMapper, Object... args) throws DataAccessException {
return DataAccessUtils.requiredSingleResult(query(newPreparedStatementCreator(cql),
newPreparedStatementBinder(args), newResultSetExtractor(rowMapper, 1)));
return DataAccessUtils.requiredSingleResult(
query(newPreparedStatementCreator(cql), newPreparedStatementBinder(args), newResultSetExtractor(rowMapper, 1)));
}
/*
@@ -659,7 +677,7 @@ public class CqlTemplate extends CassandraAccessor implements CqlOperations {
/* (non-Javadoc) */
private Set<Host> getHosts() {
return getSession().getCluster().getMetadata().getAllHosts();
return getCurrentSession().getCluster().getMetadata().getAllHosts();
}
/* (non-Javadoc) */
@@ -685,6 +703,10 @@ public class CqlTemplate extends CassandraAccessor implements CqlOperations {
return translate(task, cql, driverException);
}
private Session getCurrentSession() {
return getSessionFactory().getSession();
}
private class SimplePreparedStatementCreator implements PreparedStatementCreator, CqlProvider {
private final String cql;

14
spring-cql/src/main/java/org/springframework/cassandra/core/PreparedStatementCallback.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2013-2014 the original author or authors.
* Copyright 2013-2017 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.
@@ -15,13 +15,13 @@
*/
package org.springframework.cassandra.core;
import org.springframework.dao.DataAccessException;
import com.datastax.driver.core.PreparedStatement;
import com.datastax.driver.core.Session;
import com.datastax.driver.core.Statement;
import com.datastax.driver.core.exceptions.DriverException;
import org.springframework.dao.DataAccessException;
/**
* Generic callback interface for code that operates on a {@link PreparedStatement}. Allows to execute any number of
* operations on a single {@link PreparedStatement}, for example a single {@link Session#execute(Statement).
@@ -40,13 +40,16 @@ import org.springframework.dao.DataAccessException;
public interface PreparedStatementCallback<T> {
/**
* Gets called by {@link CqlTemplate#execute(String, PreparedStatementCallback)} with a {@link PreparedStatement}.
* Gets called by {@link CqlTemplate#execute(String, PreparedStatementCallback)} with an active CQL session and
* {@link PreparedStatement}. Does not need to care about closing the session: this will all be handled by Spring's
* {@link CqlTemplate}.
* <p>
* Allows for returning a result object created within the callback, i.e. a domain object or a collection of domain
* objects. Note that there's special support for single step actions: see
* {@link CqlTemplate#queryForObject(String, Class, Object...)} etc. A thrown RuntimeException is treated as
* application exception, it gets propagated to the caller of the template.
*
* @param session active Cassandra session, must not be {@literal null}.
* @param preparedStatement the {@link PreparedStatement}, must not be {@literal null}.
* @return a result object publisher.
* @throws DriverException if thrown by a session method, to be auto-converted to a DataAccessException.
@@ -54,6 +57,7 @@ public interface PreparedStatementCallback<T> {
* @see CqlTemplate#queryForObject(String, Class, Object...)
* @see CqlTemplate#queryForList(String, Object...)
*/
T doInPreparedStatement(PreparedStatement preparedStatement) throws DriverException, DataAccessException;
T doInPreparedStatement(Session session, PreparedStatement preparedStatement)
throws DriverException, DataAccessException;
}

52
spring-cql/src/main/java/org/springframework/cassandra/core/session/DefaultSessionFactory.java Normal file
View File

@@ -0,0 +1,52 @@
/*
* Copyright 2017 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.session;
import org.springframework.util.Assert;
import com.datastax.driver.core.Session;
/**
* Default {@link SessionFactory} implementation.
* <p>
* This class uses a singleton {@link Session} and returns the same instances.
*
* @author Mark Paluch
* @since 2.0
* @see #getSession()
* @see Session
*/
public class DefaultSessionFactory implements SessionFactory {
private final Session session;
/**
* Constructs a new {@link DefaultSessionFactory} given {@link Session}.
*
* @param session the {@link Session} to be used in {@link #getSession()}.
*/
public DefaultSessionFactory(Session session) {
Assert.notNull(session, "Session must not be null");
this.session = session;
}
@Override
public Session getSession() {
return session;
}
}

44
spring-cql/src/main/java/org/springframework/cassandra/core/session/SessionFactory.java Normal file
View File

@@ -0,0 +1,44 @@
/*
* Copyright 2017 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.session;
import com.datastax.driver.core.Session;
/**
* A factory for Apache Cassandra sessions.
* <p>
* An alternative to the {@link com.datastax.driver.core.Cluster} facility, a {@link SessionFactory} object is the
* preferred means of getting a connection. The {@link SessionFactory} interface is implemented by a {@link Session}
* provider.
* <p>
* A {@link SessionFactory} object can have properties that can be modified when necessary. For example, if the
* {@link Session} is moved to a different server, the property for the server can be changed. The benefit is that
* because the data source's properties can be changed, any code accessing that {@link SessionFactory} does not need to
* be changed.
*
* @author Mark Paluch
* @since 2.0
*/
public interface SessionFactory {
/**
* Attempts to establish a {@link Session} with the connection infrastructure that this {@link SessionFactory} object
* represents.
*
* @return a {@link Session} to Apache Cassandra.
*/
Session getSession();
}

7
spring-cql/src/main/java/org/springframework/cassandra/core/session/package-info.java Normal file
View File

@@ -0,0 +1,7 @@
/**
* Provides utility classes for simple {@link com.datastax.driver.core.Session} access and various simple DataSource
* implementations.
*
* @author Mark Paluch
*/
package org.springframework.cassandra.core.session;

123
spring-cql/src/main/java/org/springframework/cassandra/support/CassandraAccessor.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2013-2016 the original author or authors.
* Copyright 2013-2017 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.
@@ -18,15 +18,6 @@ package org.springframework.cassandra.support;
import java.util.Map;
import java.util.stream.StreamSupport;
import com.datastax.driver.core.ConsistencyLevel;
import com.datastax.driver.core.PreparedStatement;
import com.datastax.driver.core.ResultSet;
import com.datastax.driver.core.Session;
import com.datastax.driver.core.Statement;
import com.datastax.driver.core.exceptions.DriverException;
import com.datastax.driver.core.policies.RetryPolicy;
import com.datastax.driver.core.querybuilder.QueryBuilder;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.InitializingBean;
@@ -39,15 +30,26 @@ import org.springframework.cassandra.core.RowCallbackHandler;
import org.springframework.cassandra.core.RowMapper;
import org.springframework.cassandra.core.RowMapperResultSetExtractor;
import org.springframework.cassandra.core.SingleColumnRowMapper;
import org.springframework.cassandra.core.session.DefaultSessionFactory;
import org.springframework.cassandra.core.session.SessionFactory;
import org.springframework.dao.DataAccessException;
import org.springframework.util.Assert;
import com.datastax.driver.core.ConsistencyLevel;
import com.datastax.driver.core.PreparedStatement;
import com.datastax.driver.core.ResultSet;
import com.datastax.driver.core.Session;
import com.datastax.driver.core.Statement;
import com.datastax.driver.core.exceptions.DriverException;
import com.datastax.driver.core.policies.RetryPolicy;
import com.datastax.driver.core.querybuilder.QueryBuilder;
/**
* {@link CassandraAccessor} provides access to a Cassandra {@link Session} and the {@link CassandraExceptionTranslator}
* .
* {@link CassandraAccessor} provides access to a Cassandra {@link SessionFactory} and the
* {@link CassandraExceptionTranslator}.
* <p>
* Classes providing a higher abstraction level usually extend {@link CassandraAccessor} to provide a richer set of
* functionality on top of a {@link Session}.
* functionality on top of a {@link SessionFactory} using {@link Session}.
*
* @author David Webb
* @author Mark Paluch
@@ -84,29 +86,20 @@ public class CassandraAccessor implements InitializingBean {
*/
private com.datastax.driver.core.policies.RetryPolicy retryPolicy;
private Session session;
private SessionFactory sessionFactory;
/**
* Ensures the Cassandra {@link Session} and exception translator has been propertly set.
*/
@Override
public void afterPropertiesSet() {
Assert.state(session != null, "Session must not be null");
}
/* (non-Javadoc) */
@SuppressWarnings("unused")
protected void logDebug(String logMessage, Object... array) {
if (logger.isDebugEnabled()) {
logger.debug(logMessage, array);
}
Assert.state(sessionFactory != null, "SessionFactory must not be null");
}
/**
* Set the consistency level for this template. Consistency level defines the number of nodes
* involved into query processing. Relaxed consistency level settings use fewer nodes but eventual consistency is more
* likely to occur while a higher consistency level involves more nodes to obtain results with a higher consistency
* guarantee.
* Set the consistency level for this template. Consistency level defines the number of nodes involved into query
* processing. Relaxed consistency level settings use fewer nodes but eventual consistency is more likely to occur
* while a higher consistency level involves more nodes to obtain results with a higher consistency guarantee.
*
* @see Statement#setConsistencyLevel(ConsistencyLevel)
* @see RetryPolicy
@@ -147,10 +140,10 @@ public class CassandraAccessor implements InitializingBean {
}
/**
* Set the fetch size for this template. This is important for processing large result sets: Setting this
* higher than the default value will increase processing speed at the cost of memory consumption; setting this lower
* can avoid transferring row data that will never be read by the application. Default is -1, indicating to use the
* CQL driver's default configuration (i.e. to not pass a specific fetch size setting on to the driver).
* Set the fetch size for this template. This is important for processing large result sets: Setting this higher than
* the default value will increase processing speed at the cost of memory consumption; setting this lower can avoid
* transferring row data that will never be read by the application. Default is -1, indicating to use the CQL driver's
* default configuration (i.e. to not pass a specific fetch size setting on to the driver).
*
* @see Statement#setFetchSize(int)
*/
@@ -166,8 +159,7 @@ public class CassandraAccessor implements InitializingBean {
}
/**
* Set the retry policy for this template. This is important for defining behavior when a request
* fails.
* Set the retry policy for this template. This is important for defining behavior when a request fails.
*
* @see Statement#setRetryPolicy(RetryPolicy)
* @see RetryPolicy
@@ -184,25 +176,60 @@ public class CassandraAccessor implements InitializingBean {
}
/**
* Sets the Cassandra {@link Session} used by this template to perform Cassandra data access operations.
* Sets the Cassandra {@link Session} used by this template to perform Cassandra data access operations. The
* {@code session} will replace the current {@link #getSessionFactory()} with {@link DefaultSessionFactory}.
*
* @param session Cassandra {@link Session} used by this template. Must not be{@literal null}.
* @param session Cassandra {@link Session} used by this template, must not be{@literal null}.
* @see com.datastax.driver.core.Session
* @see DefaultSessionFactory
*/
public void setSession(Session session) {
Assert.notNull(session, "Session must not be null");
this.session = session;
setSessionFactory(new DefaultSessionFactory(session));
}
/**
* Returns the Cassandra {@link Session} used by this template to perform Cassandra data access operations.
* Returns the Cassandra {@link Session} from {@link SessionFactory} used by this template to perform Cassandra data
* access operations.
*
* @return the Cassandra {@link Session} used by this template.
* @see com.datastax.driver.core.Session
* @deprecated since 2.0. This class uses a {@link SessionFactory} to dispatch CQL calls amongst different
* {@link Session}s during its lifecycle.
*/
@Deprecated
public Session getSession() {
Assert.state(this.session != null, "Session was not properly initialized");
return this.session;
Assert.state(getSessionFactory() != null, "SessionFactory was not properly initialized");
return getSessionFactory().getSession();
}
/**
* Sets the Cassandra {@link SessionFactory} used by this template to perform Cassandra data access operations.
*
* @param sessionFactory Cassandra {@link Session} used by this template. Must not be{@literal null}.
* @since 2.0
* @see com.datastax.driver.core.Session
*/
public void setSessionFactory(SessionFactory sessionFactory) {
Assert.notNull(sessionFactory, "SessionFactory must not be null");
this.sessionFactory = sessionFactory;
}
/**
* Returns the Cassandra {@link SessionFactory} used by this template to perform Cassandra data access operations.
*
* @return the Cassandra {@link SessionFactory} used by this template.
* @since 2.0
* @see SessionFactory
*/
public SessionFactory getSessionFactory() {
return this.sessionFactory;
}
/**
@@ -274,8 +301,8 @@ public class CassandraAccessor implements InitializingBean {
}
/**
* Constructs a new instance of the {@link ResultSetExtractor} initialized with and adapting
* the given {@link RowCallbackHandler}.
* Constructs a new instance of the {@link ResultSetExtractor} initialized with and adapting the given
* {@link RowCallbackHandler}.
*
* @param rowCallbackHandler {@link RowCallbackHandler} to adapt as a {@link ResultSetExtractor}.
* @return a {@link ResultSetExtractor} implementation adapting an instance of the {@link RowCallbackHandler}.
@@ -288,8 +315,8 @@ public class CassandraAccessor implements InitializingBean {
}
/**
* Constructs a new instance of the {@link ResultSetExtractor} initialized with and adapting
* the given {@link RowMapper}.
* Constructs a new instance of the {@link ResultSetExtractor} initialized with and adapting the given
* {@link RowMapper}.
*
* @param rowMapper {@link RowMapper} to adapt as a {@link ResultSetExtractor}.
* @return a {@link ResultSetExtractor} implementation adapting an instance of the {@link RowMapper}.
@@ -302,8 +329,8 @@ public class CassandraAccessor implements InitializingBean {
}
/**
* Constructs a new instance of the {@link ResultSetExtractor} initialized with and adapting
* the given {@link RowMapper}.
* Constructs a new instance of the {@link ResultSetExtractor} initialized with and adapting the given
* {@link RowMapper}.
*
* @param rowMapper {@link RowMapper} to adapt as a {@link ResultSetExtractor}.
* @param rowsExpected number of expected rows in the {@link ResultSet}.
@@ -348,6 +375,12 @@ public class CassandraAccessor implements InitializingBean {
return (cqlProvider instanceof CqlProvider ? ((CqlProvider) cqlProvider).getCql() : null);
}
protected void logDebug(String logMessage, Object... array) {
if (logger.isDebugEnabled()) {
logger.debug(logMessage, array);
}
}
/**
* Translate the given {@link DriverException} into a generic {@link DataAccessException}.
* <p>

9
spring-cql/src/test/java/org/springframework/cassandra/core/AsyncCqlTemplateUnitTests.java
View File

@@ -481,8 +481,7 @@ public class AsyncCqlTemplateUnitTests {
doTestStrings(null, null, null, asyncCqlTemplate -> {
ListenableFuture<ResultSetFuture> futureOfFuture = asyncCqlTemplate.execute("SELECT * from USERS",
(PreparedStatementCallback<ResultSetFuture>) (ps) -> asyncCqlTemplate.getSession()
.executeAsync(ps.bind("A")));
(session, ps) -> session.executeAsync(ps.bind("A")));
try {
assertThat(getUninterruptibly(futureOfFuture).get()).hasSize(3);
@@ -515,7 +514,7 @@ public class AsyncCqlTemplateUnitTests {
try {
template.execute(session -> {
throw new NoHostAvailableException(Collections.emptyMap());
}, (ps) -> session.executeAsync(boundStatement));
}, (session, ps) -> session.executeAsync(boundStatement));
fail("Missing CassandraConnectionFailureException");
} catch (CassandraConnectionFailureException e) {
assertThat(e).hasMessageContaining("tried for query");
@@ -523,7 +522,7 @@ public class AsyncCqlTemplateUnitTests {
ListenableFuture<ResultSetFuture> future = template.execute(
session -> AsyncResult.forExecutionException(new NoHostAvailableException(Collections.emptyMap())),
(ps) -> session.executeAsync(boundStatement));
(session, ps) -> session.executeAsync(boundStatement));
try {
future.get();
@@ -542,7 +541,7 @@ public class AsyncCqlTemplateUnitTests {
when(resultSet.wasApplied()).thenReturn(true);
ListenableFuture<ResultSetFuture> future = template.execute(session -> new AsyncResult<>(preparedStatement),
(ps) -> {
(session, ps) -> {
throw new NoHostAvailableException(Collections.emptyMap());
});

6
spring-cql/src/test/java/org/springframework/cassandra/core/CqlTemplateUnitTests.java
View File

@@ -442,7 +442,7 @@ public class CqlTemplateUnitTests {
doTestStrings(null, null, null, cqlTemplate -> {
ResultSet resultSet = cqlTemplate.execute("SELECT * from USERS",
(PreparedStatementCallback<ResultSet>) (ps) -> cqlTemplate.getSession().execute(ps.bind("A")));
(session, ps) -> session.execute(ps.bind("A")));
try {
assertThat(resultSet).hasSize(3);
@@ -475,7 +475,7 @@ public class CqlTemplateUnitTests {
try {
template.execute(session -> {
throw new NoHostAvailableException(Collections.emptyMap());
}, (ps) -> session.execute(boundStatement));
}, (session, ps) -> session.execute(boundStatement));
fail("Missing CassandraConnectionFailureException");
} catch (CassandraConnectionFailureException e) {
@@ -490,7 +490,7 @@ public class CqlTemplateUnitTests {
when(resultSet.wasApplied()).thenReturn(true);
try {
template.execute(session -> preparedStatement, (ps) -> {
template.execute(session -> preparedStatement, (session, ps) -> {
throw new NoHostAvailableException(Collections.emptyMap());
});

8
spring-cql/src/test/java/org/springframework/cassandra/support/CassandraAccessorUnitTests.java
View File

@@ -49,14 +49,14 @@ public class CassandraAccessorUnitTests {
cassandraAccessor = new CassandraAccessor();
}
@Test // DATACASS-286
@Test // DATACASS-286, DATACASS-330
public void afterPropertiesSetWithUnitializedSessionThrowsIllegalStateException() {
try {
cassandraAccessor.afterPropertiesSet();
fail("Missing IllegalStateException");
} catch (IllegalStateException e) {
assertThat(e).hasMessageContaining("Session must not be null");
assertThat(e).hasMessageContaining("SessionFactory must not be null");
}
}
@@ -101,14 +101,14 @@ public class CassandraAccessorUnitTests {
}
}
@Test // DATACASS-286
@Test // DATACASS-286, DATACASS-330
public void getUninitializedSessionThrowsIllegalStateException() {
try {
cassandraAccessor.getSession();
fail("Missing IllegalStateException");
} catch (IllegalStateException e) {
assertThat(e).hasMessageContaining("Session was not properly initialized");
assertThat(e).hasMessageContaining("SessionFactory was not properly initialized");
}
}
}

42
spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/AsyncCassandraTemplate.java
View File

@@ -28,6 +28,8 @@ import org.springframework.cassandra.core.GuavaListenableFutureAdapter;
import org.springframework.cassandra.core.QueryOptions;
import org.springframework.cassandra.core.WriteOptions;
import org.springframework.cassandra.core.cql.CqlIdentifier;
import org.springframework.cassandra.core.session.DefaultSessionFactory;
import org.springframework.cassandra.core.session.SessionFactory;
import org.springframework.cassandra.core.support.CQLExceptionTranslator;
import org.springframework.dao.DataAccessException;
import org.springframework.dao.InvalidDataAccessApiUsageException;
@@ -100,17 +102,21 @@ public class AsyncCassandraTemplate implements AsyncCassandraOperations {
* @see Session
*/
public AsyncCassandraTemplate(Session session, CassandraConverter converter) {
this(new DefaultSessionFactory(session), converter);
}
Assert.notNull(session, "Session must not be null");
Assert.notNull(converter, "CassandraConverter must not be null");
this.converter = converter;
this.mappingContext = converter.getMappingContext();
AsyncCqlTemplate asyncCqlTemplate = new AsyncCqlTemplate(session);
this.cqlOperations = asyncCqlTemplate;
this.exceptionTranslator = asyncCqlTemplate.getExceptionTranslator();
/**
* Creates an instance of {@link AsyncCassandraTemplate} initialized with the given {@link SessionFactory} and
* {@link CassandraConverter}.
*
* @param sessionFactory {@link SessionFactory} used to interact with Cassandra; must not be {@literal null}.
* @param converter {@link CassandraConverter} used to convert between Java and Cassandra types; must not be
* {@literal null}.
* @see CassandraConverter
* @see Session
*/
public AsyncCassandraTemplate(SessionFactory sessionFactory, CassandraConverter converter) {
this(new AsyncCqlTemplate(sessionFactory), converter);
}
/**
@@ -226,7 +232,9 @@ public class AsyncCassandraTemplate implements AsyncCassandraOperations {
Assert.notNull(entityConsumer, "Entity Consumer must not be empty");
Assert.notNull(entityClass, "Entity type must not be null");
return cqlOperations.query(statement, (row) -> { entityConsumer.accept(converter.read(entityClass, row)); });
return cqlOperations.query(statement, (row) -> {
entityConsumer.accept(converter.read(entityClass, row));
});
}
/*
@@ -237,7 +245,7 @@ public class AsyncCassandraTemplate implements AsyncCassandraOperations {
public <T> ListenableFuture<T> selectOne(Statement statement, Class<T> entityClass) {
return new MappingListenableFutureAdapter<>(select(statement, entityClass),
list -> list.isEmpty() ? null : list.get(0));
list -> list.isEmpty() ? null : list.get(0));
}
// -------------------------------------------------------------------------
@@ -275,7 +283,7 @@ public class AsyncCassandraTemplate implements AsyncCassandraOperations {
converter.write(id, select.where(), entity);
return new MappingListenableFutureAdapter<>(cqlOperations.queryForResultSet(select),
resultSet -> resultSet.iterator().hasNext());
resultSet -> resultSet.iterator().hasNext());
}
/*
@@ -446,10 +454,10 @@ public class AsyncCassandraTemplate implements AsyncCassandraOperations {
@Override
public ListenableFuture<ResultSet> doInSession(Session session) throws DriverException, DataAccessException {
return new GuavaListenableFutureAdapter<>(session.executeAsync(statement), e -> (e instanceof DriverException
? exceptionTranslator.translate("AsyncStatementCallback", getCql(), (DriverException) e)
: exceptionTranslator.translateExceptionIfPossible(e))
);
return new GuavaListenableFutureAdapter<>(session.executeAsync(statement),
e -> (e instanceof DriverException
? exceptionTranslator.translate("AsyncStatementCallback", getCql(), (DriverException) e)
: exceptionTranslator.translateExceptionIfPossible(e)));
}
@Override

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

@@ -1,5 +1,5 @@
/*
* Copyright 2016 the original author or authors.
* Copyright 2016-2017 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.
@@ -19,6 +19,25 @@ import java.util.List;
import java.util.stream.Stream;
import java.util.stream.StreamSupport;
import org.springframework.cassandra.core.CqlOperations;
import org.springframework.cassandra.core.CqlProvider;
import org.springframework.cassandra.core.CqlTemplate;
import org.springframework.cassandra.core.QueryOptions;
import org.springframework.cassandra.core.SessionCallback;
import org.springframework.cassandra.core.WriteOptions;
import org.springframework.cassandra.core.cql.CqlIdentifier;
import org.springframework.cassandra.core.session.DefaultSessionFactory;
import org.springframework.cassandra.core.session.SessionFactory;
import org.springframework.cassandra.core.util.CollectionUtils;
import org.springframework.dao.DataAccessException;
import org.springframework.dao.InvalidDataAccessApiUsageException;
import org.springframework.data.cassandra.convert.CassandraConverter;
import org.springframework.data.cassandra.convert.MappingCassandraConverter;
import org.springframework.data.cassandra.mapping.CassandraMappingContext;
import org.springframework.data.cassandra.mapping.CassandraPersistentEntity;
import org.springframework.util.Assert;
import org.springframework.util.ClassUtils;
import com.datastax.driver.core.ResultSet;
import com.datastax.driver.core.Session;
import com.datastax.driver.core.SimpleStatement;
@@ -31,23 +50,6 @@ import com.datastax.driver.core.querybuilder.Select;
import com.datastax.driver.core.querybuilder.Truncate;
import com.datastax.driver.core.querybuilder.Update;
import org.springframework.cassandra.core.CqlOperations;
import org.springframework.cassandra.core.CqlProvider;
import org.springframework.cassandra.core.CqlTemplate;
import org.springframework.cassandra.core.QueryOptions;
import org.springframework.cassandra.core.SessionCallback;
import org.springframework.cassandra.core.WriteOptions;
import org.springframework.cassandra.core.cql.CqlIdentifier;
import org.springframework.cassandra.core.util.CollectionUtils;
import org.springframework.dao.DataAccessException;
import org.springframework.dao.InvalidDataAccessApiUsageException;
import org.springframework.data.cassandra.convert.CassandraConverter;
import org.springframework.data.cassandra.convert.MappingCassandraConverter;
import org.springframework.data.cassandra.mapping.CassandraMappingContext;
import org.springframework.data.cassandra.mapping.CassandraPersistentEntity;
import org.springframework.util.Assert;
import org.springframework.util.ClassUtils;
/**
* Primary implementation of {@link CassandraOperations}. It simplifies the use of Cassandra usage and helps to avoid
* common errors. It executes core Cassandra workflow. This class executes CQL queries or updates, initiating iteration
@@ -93,13 +95,21 @@ public class CassandraTemplate implements CassandraOperations {
* @see Session
*/
public CassandraTemplate(Session session, CassandraConverter converter) {
this(new DefaultSessionFactory(session), converter);
}
Assert.notNull(session, "Session must not be null");
Assert.notNull(converter, "CassandraConverter must not be null");
this.converter = converter;
this.mappingContext = converter.getMappingContext();
this.cqlOperations = new CqlTemplate(session);
/**
* Creates an instance of {@link CassandraTemplate} initialized with the given {@link SessionFactory} and
* {@link CassandraConverter}.
*
* @param sessionFactory {@link SessionFactory} used to interact with Cassandra; must not be {@literal null}.
* @param converter {@link CassandraConverter} used to convert between Java and Cassandra types; must not be
* {@literal null}.
* @see CassandraConverter
* @see SessionFactory
*/
public CassandraTemplate(SessionFactory sessionFactory, CassandraConverter converter) {
this(new CqlTemplate(sessionFactory), converter);
}
/**
@@ -311,8 +321,7 @@ public class CassandraTemplate implements CassandraOperations {
Assert.notNull(entity, "Entity must not be null");
Insert insert = QueryUtils.createInsertQuery(getTableName(entity.getClass()).toCql(),
entity, options, converter);
Insert insert = QueryUtils.createInsertQuery(getTableName(entity.getClass()).toCql(), entity, options, converter);
return cqlOperations.execute(new StatementCallback<>(insert, entity));
}
@@ -335,8 +344,7 @@ public class CassandraTemplate implements CassandraOperations {
Assert.notNull(entity, "Entity must not be null");
Update update = QueryUtils.createUpdateQuery(getTableName(entity.getClass()).toCql(),
entity, options, converter);
Update update = QueryUtils.createUpdateQuery(getTableName(entity.getClass()).toCql(), entity, options, converter);
return cqlOperations.execute(new StatementCallback<>(update, entity));
}
@@ -359,8 +367,7 @@ public class CassandraTemplate implements CassandraOperations {
Assert.notNull(entity, "Entity must not be null");
Delete delete = QueryUtils.createDeleteQuery(getTableName(entity.getClass()).toCql(),
entity, options, converter);
Delete delete = QueryUtils.createDeleteQuery(getTableName(entity.getClass()).toCql(), entity, options, converter);
return cqlOperations.execute(new StatementCallback<>(delete, entity));
}
@@ -428,7 +435,7 @@ public class CassandraTemplate implements CassandraOperations {
if (entity == null) {
throw new InvalidDataAccessApiUsageException(
String.format("No Persistent Entity information found for the class [%s]", entityClass.getName()));
String.format("No Persistent Entity information found for the class [%s]", entityClass.getName()));
}
return entity;