From 68d61461cdc9d41134cc53e4dfa48613e649162d Mon Sep 17 00:00:00 2001 From: Mark Paluch Date: Wed, 18 Jan 2017 18:03:33 +0100 Subject: [PATCH] DATACASS-330 - Add SessionFactory to CqlTemplate and AsyncCqlTemplate. 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 --- .../cassandra/core/AsyncCqlTemplate.java | 240 ++++++++++-------- .../cassandra/core/CqlTemplate.java | 82 +++--- .../core/PreparedStatementCallback.java | 14 +- .../core/session/DefaultSessionFactory.java | 52 ++++ .../core/session/SessionFactory.java | 44 ++++ .../cassandra/core/session/package-info.java | 7 + .../cassandra/support/CassandraAccessor.java | 123 +++++---- .../core/AsyncCqlTemplateUnitTests.java | 9 +- .../cassandra/core/CqlTemplateUnitTests.java | 6 +- .../support/CassandraAccessorUnitTests.java | 8 +- .../core/AsyncCassandraTemplate.java | 42 +-- .../cassandra/core/CassandraTemplate.java | 69 ++--- 12 files changed, 448 insertions(+), 248 deletions(-) create mode 100644 spring-cql/src/main/java/org/springframework/cassandra/core/session/DefaultSessionFactory.java create mode 100644 spring-cql/src/main/java/org/springframework/cassandra/core/session/SessionFactory.java create mode 100644 spring-cql/src/main/java/org/springframework/cassandra/core/session/package-info.java diff --git a/spring-cql/src/main/java/org/springframework/cassandra/core/AsyncCqlTemplate.java b/spring-cql/src/main/java/org/springframework/cassandra/core/AsyncCqlTemplate.java index 9af749ff6..048247a99 100644 --- a/spring-cql/src/main/java/org/springframework/cassandra/core/AsyncCqlTemplate.java +++ b/spring-cql/src/main/java/org/springframework/cassandra/core/AsyncCqlTemplate.java @@ -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; - /** * This is the central class in the CQL core package for asynchronous Cassandra data access. 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 ListenableFuture query(String cql, ResultSetExtractor resultSetExtractor) throws DataAccessException { + public ListenableFuture query(String cql, ResultSetExtractor 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> 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 ListenableFuture query(Statement statement, ResultSetExtractor resultSetExtractor) throws DataAccessException { + public ListenableFuture query(Statement statement, ResultSetExtractor 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 query(Statement statement, RowCallbackHandler rowCallbackHandler) throws DataAccessException { + public ListenableFuture 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 ListenableFuture> queryForList(Statement statement, Class elementType) throws DataAccessException { + public ListenableFuture> queryForList(Statement statement, Class elementType) + throws DataAccessException { return query(statement, newResultSetExtractor(newSingleColumnRowMapper(elementType))); } @@ -355,8 +374,8 @@ public class AsyncCqlTemplate extends CassandraAccessor implements AsyncCqlOpera ListenableFuture> 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 execute(AsyncPreparedStatementCreator preparedStatementCreator) throws DataAccessException { + public ListenableFuture 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 execute(String cql, PreparedStatementBinder preparedStatementBinder) throws DataAccessException { + public ListenableFuture 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 ListenableFuture execute(AsyncPreparedStatementCreator preparedStatementCreator, PreparedStatementCallback action) - throws DataAccessException { + public ListenableFuture execute(AsyncPreparedStatementCreator preparedStatementCreator, + PreparedStatementCallback 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 ListenableFuture query(AsyncPreparedStatementCreator preparedStatementCreator, ResultSetExtractor resultSetExtractor) - throws DataAccessException { + public ListenableFuture query(AsyncPreparedStatementCreator preparedStatementCreator, + ResultSetExtractor 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 query(AsyncPreparedStatementCreator preparedStatementCreator, RowCallbackHandler rowCallbackHandler) - throws DataAccessException { + public ListenableFuture 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 ListenableFuture> query(AsyncPreparedStatementCreator preparedStatementCreator, RowMapper rowMapper) - throws DataAccessException { + public ListenableFuture> query(AsyncPreparedStatementCreator preparedStatementCreator, + RowMapper rowMapper) throws DataAccessException { return query(preparedStatementCreator, null, newResultSetExtractor(rowMapper)); } @@ -482,21 +502,21 @@ public class AsyncCqlTemplate extends CassandraAccessor implements AsyncCqlOpera */ @Override public ListenableFuture query(AsyncPreparedStatementCreator preparedStatementCreator, - PreparedStatementBinder preparedStatementBinder, ResultSetExtractor resultSetExtractor) throws DataAccessException { + PreparedStatementBinder preparedStatementBinder, ResultSetExtractor 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 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 settableListenableFuture = new SettableListenableFuture<>(); - statementFuture.addCallback(boundStatement -> Futures.addCallback(session.executeAsync(boundStatement), - new FutureCallback() { - @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() { + @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 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 ListenableFuture> query(String cql, RowMapper 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 ListenableFuture> query(String cql, PreparedStatementBinder preparedStatementBinder, RowMapper 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 ListenableFuture queryForObject(String cql, RowMapper rowMapper, Object... args) throws DataAccessException { - ListenableFuture> results = query(newAsyncPreparedStatementCreator(cql), - newPreparedStatementBinder(args), newResultSetExtractor(rowMapper, 1)); + ListenableFuture> 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 createPreparedStatement(Session session) throws DriverException { - return new GuavaListenableFutureAdapter<>(session.prepareAsync(getCql()), - this.persistenceExceptionTranslator); + return new GuavaListenableFutureAdapter<>(session.prepareAsync(getCql()), this.persistenceExceptionTranslator); } } 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 31c0ff242..9abd25e5b 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 @@ -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; - /** * This is the central class in the CQL core package. 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 query(PreparedStatementCreator preparedStatementCreator, ResultSetExtractor resultSetExtractor) throws DataAccessException { + public T query(PreparedStatementCreator preparedStatementCreator, ResultSetExtractor 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 List query(PreparedStatementCreator preparedStatementCreator, RowMapper rowMapper) throws DataAccessException { + public List query(PreparedStatementCreator preparedStatementCreator, RowMapper 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 List query(PreparedStatementCreator preparedStatementCreator, PreparedStatementBinder preparedStatementBinder, - RowMapper rowMapper) throws DataAccessException { + public List query(PreparedStatementCreator preparedStatementCreator, + PreparedStatementBinder preparedStatementBinder, RowMapper 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 query(String cql, PreparedStatementBinder preparedStatementBinder, ResultSetExtractor resultSetExtractor) - throws DataAccessException { + public T query(String cql, PreparedStatementBinder preparedStatementBinder, + ResultSetExtractor resultSetExtractor) throws DataAccessException { return query(newPreparedStatementCreator(cql), preparedStatementBinder, resultSetExtractor); } @@ -577,7 +595,7 @@ public class CqlTemplate extends CassandraAccessor implements CqlOperations { @Override public List> 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 List queryForList(String cql, Class 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 queryForObject(String cql, RowMapper 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 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; diff --git a/spring-cql/src/main/java/org/springframework/cassandra/core/PreparedStatementCallback.java b/spring-cql/src/main/java/org/springframework/cassandra/core/PreparedStatementCallback.java index 4d03ba522..4d72c8039 100644 --- a/spring-cql/src/main/java/org/springframework/cassandra/core/PreparedStatementCallback.java +++ b/spring-cql/src/main/java/org/springframework/cassandra/core/PreparedStatementCallback.java @@ -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 { /** - * 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}. *

* 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 { * @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; } diff --git a/spring-cql/src/main/java/org/springframework/cassandra/core/session/DefaultSessionFactory.java b/spring-cql/src/main/java/org/springframework/cassandra/core/session/DefaultSessionFactory.java new file mode 100644 index 000000000..68e8e1f87 --- /dev/null +++ b/spring-cql/src/main/java/org/springframework/cassandra/core/session/DefaultSessionFactory.java @@ -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. + *

+ * 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; + } +} diff --git a/spring-cql/src/main/java/org/springframework/cassandra/core/session/SessionFactory.java b/spring-cql/src/main/java/org/springframework/cassandra/core/session/SessionFactory.java new file mode 100644 index 000000000..14f029f77 --- /dev/null +++ b/spring-cql/src/main/java/org/springframework/cassandra/core/session/SessionFactory.java @@ -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. + *

+ * 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. + *

+ * 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(); +} diff --git a/spring-cql/src/main/java/org/springframework/cassandra/core/session/package-info.java b/spring-cql/src/main/java/org/springframework/cassandra/core/session/package-info.java new file mode 100644 index 000000000..268a7666b --- /dev/null +++ b/spring-cql/src/main/java/org/springframework/cassandra/core/session/package-info.java @@ -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; diff --git a/spring-cql/src/main/java/org/springframework/cassandra/support/CassandraAccessor.java b/spring-cql/src/main/java/org/springframework/cassandra/support/CassandraAccessor.java index 3ed3ec910..595aa4b9c 100644 --- a/spring-cql/src/main/java/org/springframework/cassandra/support/CassandraAccessor.java +++ b/spring-cql/src/main/java/org/springframework/cassandra/support/CassandraAccessor.java @@ -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}. *

* 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}. *

diff --git a/spring-cql/src/test/java/org/springframework/cassandra/core/AsyncCqlTemplateUnitTests.java b/spring-cql/src/test/java/org/springframework/cassandra/core/AsyncCqlTemplateUnitTests.java index b0f18b4bb..2229a6295 100644 --- a/spring-cql/src/test/java/org/springframework/cassandra/core/AsyncCqlTemplateUnitTests.java +++ b/spring-cql/src/test/java/org/springframework/cassandra/core/AsyncCqlTemplateUnitTests.java @@ -481,8 +481,7 @@ public class AsyncCqlTemplateUnitTests { doTestStrings(null, null, null, asyncCqlTemplate -> { ListenableFuture futureOfFuture = asyncCqlTemplate.execute("SELECT * from USERS", - (PreparedStatementCallback) (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 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 future = template.execute(session -> new AsyncResult<>(preparedStatement), - (ps) -> { + (session, ps) -> { throw new NoHostAvailableException(Collections.emptyMap()); }); diff --git a/spring-cql/src/test/java/org/springframework/cassandra/core/CqlTemplateUnitTests.java b/spring-cql/src/test/java/org/springframework/cassandra/core/CqlTemplateUnitTests.java index a0dc5f408..431ea90b4 100644 --- a/spring-cql/src/test/java/org/springframework/cassandra/core/CqlTemplateUnitTests.java +++ b/spring-cql/src/test/java/org/springframework/cassandra/core/CqlTemplateUnitTests.java @@ -442,7 +442,7 @@ public class CqlTemplateUnitTests { doTestStrings(null, null, null, cqlTemplate -> { ResultSet resultSet = cqlTemplate.execute("SELECT * from USERS", - (PreparedStatementCallback) (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()); }); diff --git a/spring-cql/src/test/java/org/springframework/cassandra/support/CassandraAccessorUnitTests.java b/spring-cql/src/test/java/org/springframework/cassandra/support/CassandraAccessorUnitTests.java index 0a66af5a1..8c02e7f5f 100755 --- a/spring-cql/src/test/java/org/springframework/cassandra/support/CassandraAccessorUnitTests.java +++ b/spring-cql/src/test/java/org/springframework/cassandra/support/CassandraAccessorUnitTests.java @@ -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"); } } } diff --git a/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/AsyncCassandraTemplate.java b/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/AsyncCassandraTemplate.java index 131bbc834..f9f11ccf1 100644 --- a/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/AsyncCassandraTemplate.java +++ b/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/AsyncCassandraTemplate.java @@ -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 ListenableFuture selectOne(Statement statement, Class 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 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 diff --git a/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/CassandraTemplate.java b/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/CassandraTemplate.java index a1c3c9971..76f0c0907 100644 --- a/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/CassandraTemplate.java +++ b/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/CassandraTemplate.java @@ -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;