From 1c7413bf1f4102220895680d8bcc18683802dcd2 Mon Sep 17 00:00:00 2001 From: John Blum Date: Fri, 11 Nov 2016 04:32:59 -0800 Subject: [PATCH] DATACASS-335 - Polish. --- .../cassandra/core/ColumnMapRowMapper.java | 7 +- .../cassandra/core/CqlTemplate.java | 19 ++- .../core/DefaultBridgedReactiveSession.java | 17 ++- .../cassandra/core/ReactiveCqlTemplate.java | 112 ++++++++++-------- .../ReactiveRowMapperResultSetExtractor.java | 3 +- .../cassandra/core/SingleColumnRowMapper.java | 16 +-- .../core/ReactiveCassandraTemplate.java | 31 +++-- ...andraRepositoryConfigurationExtension.java | 12 +- ...andraRepositoryConfigurationExtension.java | 5 +- .../query/AbstractReactiveCassandraQuery.java | 8 +- .../ReactiveCassandraParameterAccessor.java | 14 +-- .../query/ReactiveCassandraQueryMethod.java | 1 + .../ReactiveStringBasedCassandraQuery.java | 1 + .../repository/query/StringBasedQuery.java | 14 +-- .../ReactiveCassandraRepositoryFactory.java | 4 +- ...eactiveCassandraRepositoryFactoryBean.java | 6 +- 16 files changed, 158 insertions(+), 112 deletions(-) diff --git a/spring-cql/src/main/java/org/springframework/cassandra/core/ColumnMapRowMapper.java b/spring-cql/src/main/java/org/springframework/cassandra/core/ColumnMapRowMapper.java index 86b21e0cf..64b9c7097 100644 --- a/spring-cql/src/main/java/org/springframework/cassandra/core/ColumnMapRowMapper.java +++ b/spring-cql/src/main/java/org/springframework/cassandra/core/ColumnMapRowMapper.java @@ -53,6 +53,7 @@ public class ColumnMapRowMapper implements RowMapper> { Object obj = getColumnValue(rs, i); mapOfColValues.put(key, obj); } + return mapOfColValues; } @@ -60,7 +61,7 @@ public class ColumnMapRowMapper implements RowMapper> { * Create a {@link Map} instance to be used as column map. *

* By default, a linked case-insensitive Map will be created. - * + * * @param columnCount the column count, to be used as initial capacity for the {@link Map}, must not be {@literal null}. * @return the new Map instance. * @see org.springframework.util.LinkedCaseInsensitiveMap @@ -71,7 +72,7 @@ public class ColumnMapRowMapper implements RowMapper> { /** * Determine the key to use for the given column in the column Map. - * + * * @param columnName the column name as returned by the {@link Row}, must not be {@literal null}. * @return the column key to use. * @see ColumnDefinitions#getName(int) @@ -84,7 +85,7 @@ public class ColumnMapRowMapper implements RowMapper> { * Retrieve a CQL object value for the specified column. *

* The default implementation uses the {@code getObject} method. - * + * * @param row is the {@link Row} holding the data, must not be {@literal null}. * @param index is the column index. * @return the Object returned 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 bf97804aa..ad10d72b1 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 @@ -15,7 +15,7 @@ */ package org.springframework.cassandra.core; -import static org.springframework.cassandra.core.cql.CqlIdentifier.*; +import static org.springframework.cassandra.core.cql.CqlIdentifier.cqlId; import java.util.ArrayList; import java.util.Collection; @@ -55,8 +55,17 @@ import org.springframework.dao.QueryTimeoutException; import org.springframework.dao.support.PersistenceExceptionTranslator; import org.springframework.util.Assert; -import com.datastax.driver.core.*; +import com.datastax.driver.core.BoundStatement; +import com.datastax.driver.core.ColumnDefinitions; import com.datastax.driver.core.ColumnDefinitions.Definition; +import com.datastax.driver.core.Host; +import com.datastax.driver.core.PreparedStatement; +import com.datastax.driver.core.ResultSet; +import com.datastax.driver.core.ResultSetFuture; +import com.datastax.driver.core.Row; +import com.datastax.driver.core.Session; +import com.datastax.driver.core.SimpleStatement; +import com.datastax.driver.core.Statement; import com.datastax.driver.core.exceptions.DriverException; import com.datastax.driver.core.querybuilder.Batch; import com.datastax.driver.core.querybuilder.Delete; @@ -131,7 +140,8 @@ public class CqlTemplate extends CassandraAccessor implements CqlOperations { if (queryOptions.getDriverConsistencyLevel() != null) { preparedStatement.setConsistencyLevel(queryOptions.getDriverConsistencyLevel()); } else if (queryOptions.getConsistencyLevel() != null) { - preparedStatement.setConsistencyLevel(ConsistencyLevelResolver.resolve(queryOptions.getConsistencyLevel())); + preparedStatement.setConsistencyLevel( + ConsistencyLevelResolver.resolve(queryOptions.getConsistencyLevel())); } if (queryOptions.getDriverRetryPolicy() != null) { @@ -455,7 +465,8 @@ public class CqlTemplate extends CassandraAccessor implements CqlOperations { @Override public ResultSetFuture doInSession(Session session) { return session - .executeAsync(addQueryOptions(new SimpleStatement(logCql("async execute CQL [{}]", cql)), queryOptions)); + .executeAsync( + addQueryOptions(new SimpleStatement(logCql("async execute CQL [{}]", cql)), queryOptions)); } })); } diff --git a/spring-cql/src/main/java/org/springframework/cassandra/core/DefaultBridgedReactiveSession.java b/spring-cql/src/main/java/org/springframework/cassandra/core/DefaultBridgedReactiveSession.java index 472972260..9941b5e7a 100644 --- a/spring-cql/src/main/java/org/springframework/cassandra/core/DefaultBridgedReactiveSession.java +++ b/spring-cql/src/main/java/org/springframework/cassandra/core/DefaultBridgedReactiveSession.java @@ -23,7 +23,17 @@ import org.slf4j.Logger; import org.slf4j.LoggerFactory; import org.springframework.util.Assert; -import com.datastax.driver.core.*; +import com.datastax.driver.core.Cluster; +import com.datastax.driver.core.ColumnDefinitions; +import com.datastax.driver.core.ExecutionInfo; +import com.datastax.driver.core.PreparedStatement; +import com.datastax.driver.core.RegularStatement; +import com.datastax.driver.core.ResultSet; +import com.datastax.driver.core.ResultSetFuture; +import com.datastax.driver.core.Row; +import com.datastax.driver.core.Session; +import com.datastax.driver.core.SimpleStatement; +import com.datastax.driver.core.Statement; import com.google.common.util.concurrent.ListenableFuture; import reactor.core.publisher.Flux; @@ -118,9 +128,7 @@ public class DefaultBridgedReactiveSession implements ReactiveSession { Assert.notNull(statement, "Statement must not be null"); return Mono.defer(() -> { - try { - if (logger.isDebugEnabled()) { logger.debug("Executing Statement [{}]", statement); } @@ -168,9 +176,7 @@ public class DefaultBridgedReactiveSession implements ReactiveSession { Assert.notNull(statement, "Statement must not be null"); return Mono.defer(() -> { - try { - if (logger.isDebugEnabled()) { logger.debug("Preparing Statement [{}]", statement); } @@ -238,6 +244,7 @@ public class DefaultBridgedReactiveSession implements ReactiveSession { public Flux rows() { int prefetch = Math.max(1, resultSet.getAvailableWithoutFetching()); + return Flux.fromIterable(resultSet) // .subscribeOn(scheduler) // .publishOn(Schedulers.immediate(), prefetch); // limit prefetching to available size diff --git a/spring-cql/src/main/java/org/springframework/cassandra/core/ReactiveCqlTemplate.java b/spring-cql/src/main/java/org/springframework/cassandra/core/ReactiveCqlTemplate.java index b3182180b..aa29e37c0 100644 --- a/spring-cql/src/main/java/org/springframework/cassandra/core/ReactiveCqlTemplate.java +++ b/spring-cql/src/main/java/org/springframework/cassandra/core/ReactiveCqlTemplate.java @@ -133,6 +133,26 @@ public class ReactiveCqlTemplate extends ReactiveCassandraAccessor implements Re afterPropertiesSet(); } + /** + * Set the consistency level for this {@link ReactiveCqlTemplate}. 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 + */ + public void setConsistencyLevel(ConsistencyLevel consistencyLevel) { + this.consistencyLevel = consistencyLevel; + } + + /** + * @return the {@link ConsistencyLevel} specified for this {@link ReactiveCqlTemplate}. + */ + public ConsistencyLevel getConsistencyLevel() { + return consistencyLevel; + } + /** * Set the fetch size for this {@link ReactiveCqlTemplate}. 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; @@ -171,26 +191,6 @@ public class ReactiveCqlTemplate extends ReactiveCassandraAccessor implements Re return retryPolicy; } - /** - * Set the consistency level for this {@link ReactiveCqlTemplate}. 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 - */ - public void setConsistencyLevel(ConsistencyLevel consistencyLevel) { - this.consistencyLevel = consistencyLevel; - } - - /** - * @return the {@link ConsistencyLevel} specified for this {@link ReactiveCqlTemplate}. - */ - public ConsistencyLevel getConsistencyLevel() { - return consistencyLevel; - } - // ------------------------------------------------------------------------- // Methods dealing with a plain org.springframework.cassandra.core.ReactiveSession // ------------------------------------------------------------------------- @@ -225,10 +225,10 @@ public class ReactiveCqlTemplate extends ReactiveCassandraAccessor implements Re * @see org.springframework.cassandra.core.ReactiveCqlOperations#query(java.lang.String, org.springframework.cassandra.core.ReactiveResultSetExtractor) */ @Override - public Flux query(String cql, ReactiveResultSetExtractor rse) throws DataAccessException { + public Flux query(String cql, ReactiveResultSetExtractor resultSetExtractor) throws DataAccessException { Assert.hasText(cql, "CQL must not be empty"); - Assert.notNull(rse, "ReactiveResultSetExtractor must not be null"); + Assert.notNull(resultSetExtractor, "ReactiveResultSetExtractor must not be null"); return createFlux(new SimpleStatement(cql), (session, stmt) -> { @@ -236,7 +236,7 @@ public class ReactiveCqlTemplate extends ReactiveCassandraAccessor implements Re logger.debug("Executing CQL Statement [{}]", cql); } - return session.execute(stmt).flatMap(rse::extractData); + return session.execute(stmt).flatMap(resultSetExtractor::extractData); }).onErrorResumeWith(translateException("Query", cql)); } @@ -253,8 +253,8 @@ public class ReactiveCqlTemplate extends ReactiveCassandraAccessor implements Re */ @Override public Mono queryForObject(String cql, RowMapper rowMapper) throws DataAccessException { - return query(cql, rowMapper).buffer(2).flatMap(list -> Mono.just(DataAccessUtils.requiredSingleResult(list))) - .next(); + return query(cql, rowMapper).buffer(2).flatMap( + list -> Mono.just(DataAccessUtils.requiredSingleResult(list))).next(); } /* (non-Javadoc) @@ -374,8 +374,8 @@ public class ReactiveCqlTemplate extends ReactiveCassandraAccessor implements Re */ @Override public Mono queryForObject(Statement statement, RowMapper rowMapper) throws DataAccessException { - return query(statement, rowMapper).buffer(2).flatMap(list -> Mono.just(DataAccessUtils.requiredSingleResult(list))) - .next(); + return query(statement, rowMapper).buffer(2).flatMap( + list -> Mono.just(DataAccessUtils.requiredSingleResult(list))).next(); } /* (non-Javadoc) @@ -424,6 +424,7 @@ public class ReactiveCqlTemplate extends ReactiveCassandraAccessor implements Re logger.debug("Executing CQL [{}]", executedStatement); } + return session.execute(executedStatement); }).otherwise(translateException("QueryForResultSet", statement.toString())); } @@ -469,13 +470,13 @@ public class ReactiveCqlTemplate extends ReactiveCassandraAccessor implements Re * Query using a prepared statement, reading the {@link ReactiveResultSet} with a {@link ReactiveResultSetExtractor}. * * @param psc object that can create a {@link PreparedStatement} given a {@link ReactiveSession} - * @param psb object that knows how to set values on the prepared statement. If this is {@literal null}, the CQL will + * @param preparedStatementBinder object that knows how to set values on the prepared statement. If this is {@literal null}, the CQL will * be assumed to contain no bind parameters. * @param rse object that will extract results * @return an arbitrary result object, as returned by the {@link ReactiveResultSetExtractor} * @throws DataAccessException if there is any problem */ - public Flux query(ReactivePreparedStatementCreator psc, PreparedStatementBinder psb, + public Flux query(ReactivePreparedStatementCreator psc, PreparedStatementBinder preparedStatementBinder, ReactiveResultSetExtractor rse) throws DataAccessException { Assert.notNull(psc, "ReactivePreparedStatementCreator must not be null"); @@ -487,9 +488,11 @@ public class ReactiveCqlTemplate extends ReactiveCassandraAccessor implements Re logger.debug("Executing Prepared CQL Statement [{}]", ps.getQueryString()); } - BoundStatement boundStatement = psb != null ? psb.bindValues(ps) : ps.bind(); + BoundStatement boundStatement = (preparedStatementBinder != null + ? preparedStatementBinder.bindValues(ps) : ps.bind()); applyStatementSettings(boundStatement); + return session.execute(boundStatement); }).flatMap(rse::extractData)).onErrorResumeWith(translateException("Query", getCql(psc))); } @@ -500,6 +503,7 @@ public class ReactiveCqlTemplate extends ReactiveCassandraAccessor implements Re @Override public Flux query(ReactivePreparedStatementCreator psc, ReactiveResultSetExtractor rse) throws DataAccessException { + return query(psc, null, rse); } @@ -509,6 +513,7 @@ public class ReactiveCqlTemplate extends ReactiveCassandraAccessor implements Re @Override public Flux query(String cql, PreparedStatementBinder psb, ReactiveResultSetExtractor rse) throws DataAccessException { + return query(new SimpleReactivePreparedStatementCreator(cql), psb, rse); } @@ -542,6 +547,7 @@ public class ReactiveCqlTemplate extends ReactiveCassandraAccessor implements Re @Override public Flux query(ReactivePreparedStatementCreator psc, PreparedStatementBinder psb, RowMapper rowMapper) throws DataAccessException { + return query(psc, psb, new ReactiveRowMapperResultSetExtractor<>(rowMapper)); } @@ -558,8 +564,8 @@ public class ReactiveCqlTemplate extends ReactiveCassandraAccessor implements Re */ @Override public Mono queryForObject(String cql, RowMapper rowMapper, Object... args) throws DataAccessException { - return query(cql, rowMapper, args).buffer(2).flatMap(list -> Mono.just(DataAccessUtils.requiredSingleResult(list))) - .next(); + return query(cql, rowMapper, args).buffer(2).flatMap( + list -> Mono.just(DataAccessUtils.requiredSingleResult(list))).next(); } /* (non-Javadoc) @@ -628,8 +634,8 @@ public class ReactiveCqlTemplate extends ReactiveCassandraAccessor implements Re */ @Override public Mono execute(String cql, PreparedStatementBinder psb) throws DataAccessException { - return query(new SimpleReactivePreparedStatementCreator(cql), psb, resultSet -> Mono.just(resultSet.wasApplied())) - .next(); + return query(new SimpleReactivePreparedStatementCreator(cql), psb, + resultSet -> Mono.just(resultSet.wasApplied())).next(); } /* (non-Javadoc) @@ -658,6 +664,7 @@ public class ReactiveCqlTemplate extends ReactiveCassandraAccessor implements Re BoundStatement boundStatement = newArgPreparedStatementBinder(objects).bindValues(ps); applyStatementSettings(boundStatement); + return session.execute(boundStatement); }).map(ReactiveResultSet::wasApplied)); @@ -724,8 +731,8 @@ public class ReactiveCqlTemplate extends ReactiveCassandraAccessor implements Re @SuppressWarnings("ThrowableResultOfMethodCallIgnored") protected Function> translateException() { - return throwable -> Mono.error( - throwable instanceof DriverException ? translateExceptionIfPossible((DriverException) throwable) : throwable); + return throwable -> Mono.error(throwable instanceof DriverException + ? translateExceptionIfPossible((DriverException) throwable) : throwable); } /** @@ -740,7 +747,7 @@ public class ReactiveCqlTemplate extends ReactiveCassandraAccessor implements Re protected Function> translateException(String task, String cql) { return throwable -> Mono.error(throwable instanceof DriverException - ? ReactiveCqlTemplate.this.translate(task, cql, (DriverException) throwable) : throwable); + ? translate(task, cql, (DriverException) throwable) : throwable); } /** @@ -775,20 +782,23 @@ public class ReactiveCqlTemplate extends ReactiveCassandraAccessor implements Re */ protected void applyStatementSettings(Statement stmt) { + ConsistencyLevel consistencyLevel = getConsistencyLevel(); + + if (consistencyLevel != null && stmt.getConsistencyLevel() == DEFAULTS.getConsistencyLevel()) { + stmt.setConsistencyLevel(consistencyLevel); + } + int fetchSize = getFetchSize(); + if (fetchSize != -1 && stmt.getFetchSize() == DEFAULTS.getFetchSize()) { stmt.setFetchSize(fetchSize); } RetryPolicy retryPolicy = getRetryPolicy(); + if (retryPolicy != null && stmt.getRetryPolicy() == DEFAULTS.getRetryPolicy()) { stmt.setRetryPolicy(retryPolicy); } - - ConsistencyLevel consistencyLevel = getConsistencyLevel(); - if (consistencyLevel != null && stmt.getConsistencyLevel() == DEFAULTS.getConsistencyLevel()) { - stmt.setConsistencyLevel(consistencyLevel); - } } /** @@ -801,15 +811,17 @@ public class ReactiveCqlTemplate extends ReactiveCassandraAccessor implements Re */ protected void applyStatementSettings(PreparedStatement stmt) { - RetryPolicy retryPolicy = getRetryPolicy(); - if (retryPolicy != null) { - stmt.setRetryPolicy(retryPolicy); - } - ConsistencyLevel consistencyLevel = getConsistencyLevel(); + if (consistencyLevel != null) { stmt.setConsistencyLevel(consistencyLevel); } + + RetryPolicy retryPolicy = getRetryPolicy(); + + if (retryPolicy != null) { + stmt.setRetryPolicy(retryPolicy); + } } /** @@ -838,11 +850,7 @@ public class ReactiveCqlTemplate extends ReactiveCassandraAccessor implements Re */ private static String getCql(Object cqlProvider) { - if (cqlProvider instanceof CqlProvider) { - return ((CqlProvider) cqlProvider).getCql(); - } else { - return null; - } + return (cqlProvider instanceof CqlProvider ? ((CqlProvider) cqlProvider).getCql() : null); } private class SimpleReactivePreparedStatementCreator implements ReactivePreparedStatementCreator, CqlProvider { diff --git a/spring-cql/src/main/java/org/springframework/cassandra/core/ReactiveRowMapperResultSetExtractor.java b/spring-cql/src/main/java/org/springframework/cassandra/core/ReactiveRowMapperResultSetExtractor.java index bd4eed455..8c86d24db 100644 --- a/spring-cql/src/main/java/org/springframework/cassandra/core/ReactiveRowMapperResultSetExtractor.java +++ b/spring-cql/src/main/java/org/springframework/cassandra/core/ReactiveRowMapperResultSetExtractor.java @@ -44,7 +44,7 @@ public class ReactiveRowMapperResultSetExtractor implements ReactiveResultSet /** * Create a new {@link ReactiveRowMapperResultSetExtractor}. - * + * * @param rowMapper the {@link RowMapper} which creates an object for each row, must not be {@literal null}. */ public ReactiveRowMapperResultSetExtractor(RowMapper rowMapper) { @@ -67,6 +67,7 @@ public class ReactiveRowMapperResultSetExtractor implements ReactiveResultSet if (value == null) { return Mono.empty(); } + return Mono.just(value); }); } diff --git a/spring-cql/src/main/java/org/springframework/cassandra/core/SingleColumnRowMapper.java b/spring-cql/src/main/java/org/springframework/cassandra/core/SingleColumnRowMapper.java index 2ac6b89d3..587d1aa8c 100644 --- a/spring-cql/src/main/java/org/springframework/cassandra/core/SingleColumnRowMapper.java +++ b/spring-cql/src/main/java/org/springframework/cassandra/core/SingleColumnRowMapper.java @@ -43,7 +43,7 @@ public class SingleColumnRowMapper implements RowMapper { /** * Create a new {@link SingleColumnRowMapper} for bean-style configuration. - * + * * @see #setRequiredType */ public SingleColumnRowMapper() {} @@ -53,7 +53,7 @@ public class SingleColumnRowMapper implements RowMapper { *

* Consider using the {@link #newInstance} factory method instead, which allows for specifying the required type once * only. - * + * * @param requiredType the type that each result object is expected to match */ public SingleColumnRowMapper(Class requiredType) { @@ -74,7 +74,7 @@ public class SingleColumnRowMapper implements RowMapper { *

* Validates that there is only one column selected, then delegates to {@code getColumnValue()} and also * {@code convertValueToRequiredType}, if necessary. - * + * * @see ColumnDefinitions#size() * @see #getColumnValue(Row, int, Class) * @see #convertValueToRequiredType(Object, Class) @@ -86,12 +86,14 @@ public class SingleColumnRowMapper implements RowMapper { // Validate column count. ColumnDefinitions definitions = row.getColumnDefinitions(); int nrOfColumns = definitions.size(); + if (nrOfColumns != 1) { throw new IncorrectResultSetColumnCountException(1, nrOfColumns); } // Extract column value from CQL ResultSet. Object result = getColumnValue(row, 0, this.requiredType); + if (result != null && this.requiredType != null && !this.requiredType.isInstance(result)) { // Extracted value does not match already: try to convert it. try { @@ -102,6 +104,7 @@ public class SingleColumnRowMapper implements RowMapper { definitions.getType(0), ex.getMessage())); } } + return (T) result; } @@ -111,7 +114,7 @@ public class SingleColumnRowMapper implements RowMapper { * The default implementation calls {@link RowUtils#getRowValue(Row, int, Class)}. If no required type has been * specified, this method delegates to {@code getColumnValue(rs, index)}, which basically calls * {@link Row#getObject(int)} but applies some additional default conversion to appropriate value types. - * + * * @param row is the {@link Row} holding the data, must not be {@literal null}. * @param index is the column index * @param requiredType the type that each result object is expected to match (or {@code null} if none specified). @@ -154,7 +157,7 @@ public class SingleColumnRowMapper implements RowMapper { * If the required type is String, the value will simply get stringified via {@code toString()}. In case of a Number, * the value will be converted into a Number, either through number conversion or through String parsing (depending on * the value type). - * + * * @param value the column value as extracted from {@code getColumnValue()} (never {@code null}) * @param requiredType the type that each result object is expected to match (never {@code null}) * @return the converted value @@ -166,7 +169,6 @@ public class SingleColumnRowMapper implements RowMapper { if (String.class == requiredType) { return value.toString(); } else if (Number.class.isAssignableFrom(requiredType)) { - if (value instanceof Number) { // Convert original Number to target Number class. return NumberUtils.convertNumberToTargetClass(((Number) value), (Class) requiredType); @@ -183,7 +185,7 @@ public class SingleColumnRowMapper implements RowMapper { /** * Static factory method to create a new {@code SingleColumnRowMapper} (with the required type specified only once). - * + * * @param requiredType the type that each result object is expected to match */ public static SingleColumnRowMapper newInstance(Class requiredType) { diff --git a/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/ReactiveCassandraTemplate.java b/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/ReactiveCassandraTemplate.java index 029bfef51..89372bc66 100644 --- a/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/ReactiveCassandraTemplate.java +++ b/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/ReactiveCassandraTemplate.java @@ -15,10 +15,21 @@ */ package org.springframework.data.cassandra.core; -import static org.springframework.data.cassandra.core.CassandraTemplate.*; +import static org.springframework.data.cassandra.core.CassandraTemplate.createDeleteQuery; +import static org.springframework.data.cassandra.core.CassandraTemplate.createInsertQuery; +import static org.springframework.data.cassandra.core.CassandraTemplate.createUpdateQuery; import org.reactivestreams.Publisher; -import org.springframework.cassandra.core.*; +import org.springframework.cassandra.core.CqlProvider; +import org.springframework.cassandra.core.DefaultReactiveSessionFactory; +import org.springframework.cassandra.core.QueryOptions; +import org.springframework.cassandra.core.ReactiveCqlOperations; +import org.springframework.cassandra.core.ReactiveCqlTemplate; +import org.springframework.cassandra.core.ReactiveResultSet; +import org.springframework.cassandra.core.ReactiveSession; +import org.springframework.cassandra.core.ReactiveSessionCallback; +import org.springframework.cassandra.core.ReactiveSessionFactory; +import org.springframework.cassandra.core.WriteOptions; import org.springframework.cassandra.core.cql.CqlIdentifier; import org.springframework.dao.DataAccessException; import org.springframework.dao.InvalidDataAccessApiUsageException; @@ -269,8 +280,8 @@ public class ReactiveCassandraTemplate implements ReactiveCassandraOperations { @Override public Publisher doInSession(ReactiveSession session) throws DriverException, DataAccessException { - return session.execute(insert) - .flatMap(reactiveResultSet -> reactiveResultSet.wasApplied() ? Mono.just(entity) : Mono.empty()); + return session.execute(insert).flatMap( + reactiveResultSet -> reactiveResultSet.wasApplied() ? Mono.just(entity) : Mono.empty()); } @Override @@ -299,6 +310,7 @@ public class ReactiveCassandraTemplate implements ReactiveCassandraOperations { public Flux insert(Publisher entities, WriteOptions options) { Assert.notNull(entities, "Entity publisher must not be null"); + return Flux.from(entities).flatMap(entity -> insert(entity, options)); } @@ -328,8 +340,8 @@ public class ReactiveCassandraTemplate implements ReactiveCassandraOperations { @Override public Publisher doInSession(ReactiveSession session) throws DriverException, DataAccessException { - return session.execute(update) - .flatMap(reactiveResultSet -> reactiveResultSet.wasApplied() ? Mono.just(entity) : Mono.empty()); + return session.execute(update).flatMap( + reactiveResultSet -> reactiveResultSet.wasApplied() ? Mono.just(entity) : Mono.empty()); } @Override @@ -358,6 +370,7 @@ public class ReactiveCassandraTemplate implements ReactiveCassandraOperations { public Flux update(Publisher entities, WriteOptions options) { Assert.notNull(entities, "Entity publisher must not be null"); + return Flux.from(entities).flatMap(entity -> update(entity, options)); } @@ -405,8 +418,8 @@ public class ReactiveCassandraTemplate implements ReactiveCassandraOperations { @Override public Publisher doInSession(ReactiveSession session) throws DriverException, DataAccessException { - return session.execute(delete) - .flatMap(reactiveResultSet -> reactiveResultSet.wasApplied() ? Mono.just(entity) : Mono.empty()); + return session.execute(delete).flatMap( + reactiveResultSet -> reactiveResultSet.wasApplied() ? Mono.just(entity) : Mono.empty()); } @Override @@ -435,6 +448,7 @@ public class ReactiveCassandraTemplate implements ReactiveCassandraOperations { public Flux delete(Publisher entities, QueryOptions options) { Assert.notNull(entities, "Entity publisher must not be null"); + return Flux.from(entities).flatMap(entity -> delete(entity, options)); } @@ -446,6 +460,7 @@ public class ReactiveCassandraTemplate implements ReactiveCassandraOperations { public Mono truncate(Class entityClass) { Assert.notNull(entityClass, "Entity type must not be null"); + Truncate truncate = QueryBuilder.truncate(getPersistentEntity(entityClass).getTableName().toCql()); return cqlOperations.execute(truncate).then(); diff --git a/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/repository/config/CassandraRepositoryConfigurationExtension.java b/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/repository/config/CassandraRepositoryConfigurationExtension.java index b8db98170..596b39ed1 100644 --- a/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/repository/config/CassandraRepositoryConfigurationExtension.java +++ b/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/repository/config/CassandraRepositoryConfigurationExtension.java @@ -1,12 +1,12 @@ /* * Copyright 2013-2016 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. @@ -40,7 +40,7 @@ import org.w3c.dom.Element; /** * {@link RepositoryConfigurationExtension} for Cassandra. - * + * * @author Alex Shvid * @author Mark Paluch * @author Christoph Strobl @@ -83,6 +83,7 @@ public class CassandraRepositoryConfigurationExtension extends RepositoryConfigu AnnotationAttributes attributes = config.getAttributes(); String cassandraTemplateRef = attributes.getString("cassandraTemplateRef"); + if (StringUtils.hasText(cassandraTemplateRef)) { builder.addPropertyReference("cassandraTemplate", cassandraTemplateRef); } @@ -118,12 +119,11 @@ public class CassandraRepositoryConfigurationExtension extends RepositoryConfigu return repositoryConfigurations.stream().filter(configuration -> { Class repositoryInterface = super.loadRepositoryInterface(configuration, loader); - return !RepositoryType.isReactiveRepository(repositoryInterface); + return !RepositoryType.isReactiveRepository(repositoryInterface); }).collect(Collectors.toList()); } return repositoryConfigurations; } - } diff --git a/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/repository/config/ReactiveCassandraRepositoryConfigurationExtension.java b/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/repository/config/ReactiveCassandraRepositoryConfigurationExtension.java index ff1d7cc42..b3dd4e48e 100644 --- a/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/repository/config/ReactiveCassandraRepositoryConfigurationExtension.java +++ b/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/repository/config/ReactiveCassandraRepositoryConfigurationExtension.java @@ -86,6 +86,7 @@ public class ReactiveCassandraRepositoryConfigurationExtension extends Repositor AnnotationAttributes attributes = config.getAttributes(); String reactiveCassandraTemplateRef = attributes.getString("reactiveCassandraTemplateRef"); + if (StringUtils.hasText(reactiveCassandraTemplateRef)) { builder.addPropertyReference("reactiveCassandraOperations", reactiveCassandraTemplateRef); } @@ -97,7 +98,7 @@ public class ReactiveCassandraRepositoryConfigurationExtension extends Repositor */ @Override protected Collection> getIdentifyingAnnotations() { - return Collections.> singleton(Table.class); + return Collections.>singleton(Table.class); } /* @@ -106,7 +107,7 @@ public class ReactiveCassandraRepositoryConfigurationExtension extends Repositor */ @Override protected Collection> getIdentifyingTypes() { - return Collections.> singleton(ReactiveCassandraRepository.class); + return Collections.>singleton(ReactiveCassandraRepository.class); } /* diff --git a/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/repository/query/AbstractReactiveCassandraQuery.java b/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/repository/query/AbstractReactiveCassandraQuery.java index fc69e5c63..9834e6ae0 100644 --- a/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/repository/query/AbstractReactiveCassandraQuery.java +++ b/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/repository/query/AbstractReactiveCassandraQuery.java @@ -124,11 +124,8 @@ public abstract class AbstractReactiveCassandraQuery implements RepositoryQuery private ReactiveCassandraQueryExecution getExecutionToWrap(CassandraParameterAccessor accessor, Converter resultProcessing) { - if (method.isCollectionQuery()) { - return new CollectionExecution(operations); - } else { - return new SingleEntityExecution(operations); - } + return (method.isCollectionQuery() ? new CollectionExecution(operations) + : new SingleEntityExecution(operations)); } /** @@ -137,4 +134,5 @@ public abstract class AbstractReactiveCassandraQuery implements RepositoryQuery * @param accessor must not be {@literal null}. */ protected abstract String createQuery(CassandraParameterAccessor accessor); + } diff --git a/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/repository/query/ReactiveCassandraParameterAccessor.java b/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/repository/query/ReactiveCassandraParameterAccessor.java index 1f17267d7..a371d27d4 100644 --- a/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/repository/query/ReactiveCassandraParameterAccessor.java +++ b/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/repository/query/ReactiveCassandraParameterAccessor.java @@ -45,7 +45,6 @@ class ReactiveCassandraParameterAccessor extends CassandraParametersParameterAcc this.subscriptions = new ArrayList<>(values.length); for (Object value : values) { - if (value == null || !ReactiveWrappers.supports(value.getClass())) { subscriptions.add(null); continue; @@ -65,12 +64,7 @@ class ReactiveCassandraParameterAccessor extends CassandraParametersParameterAcc @SuppressWarnings("unchecked") @Override protected T getValue(int index) { - - if (subscriptions.get(index) != null) { - return (T) subscriptions.get(index).block(); - } - - return super.getValue(index); + return (subscriptions.get(index) != null ? (T) subscriptions.get(index).block() : super.getValue(index)); } /* (non-Javadoc) @@ -80,9 +74,11 @@ class ReactiveCassandraParameterAccessor extends CassandraParametersParameterAcc public Object[] getValues() { Object[] result = new Object[values.length]; - for (int i = 0; i < result.length; i++) { - result[i] = getValue(i); + + for (int index = 0; index < result.length; index++) { + result[index] = getValue(index); } + return result; } diff --git a/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/repository/query/ReactiveCassandraQueryMethod.java b/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/repository/query/ReactiveCassandraQueryMethod.java index 68f9ec6d2..e50634d73 100644 --- a/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/repository/query/ReactiveCassandraQueryMethod.java +++ b/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/repository/query/ReactiveCassandraQueryMethod.java @@ -82,6 +82,7 @@ public class ReactiveCassandraQueryMethod extends CassandraQueryMethod { return true; } } + return false; } } diff --git a/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/repository/query/ReactiveStringBasedCassandraQuery.java b/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/repository/query/ReactiveStringBasedCassandraQuery.java index d3c539716..75a859929 100644 --- a/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/repository/query/ReactiveStringBasedCassandraQuery.java +++ b/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/repository/query/ReactiveStringBasedCassandraQuery.java @@ -59,6 +59,7 @@ public class ReactiveStringBasedCassandraQuery extends AbstractReactiveCassandra public ReactiveStringBasedCassandraQuery(ReactiveCassandraQueryMethod queryMethod, ReactiveCassandraOperations operations, SpelExpressionParser expressionParser, EvaluationContextProvider evaluationContextProvider) { + this(queryMethod.getAnnotatedQuery(), queryMethod, operations, expressionParser, evaluationContextProvider); } diff --git a/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/repository/query/StringBasedQuery.java b/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/repository/query/StringBasedQuery.java index d7d81f817..ff97b5605 100644 --- a/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/repository/query/StringBasedQuery.java +++ b/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/repository/query/StringBasedQuery.java @@ -35,7 +35,7 @@ import com.datastax.driver.core.querybuilder.BindMarker; /** * String-based Query abstracting a CQL query with parameter bindings. - * + * * @author Mark Paluch * @since 2.0 */ @@ -49,7 +49,7 @@ class StringBasedQuery { /** * Creates a new {@link StringBasedQuery} given {@code query}, {@link ExpressionEvaluatingParameterBinder} and * {@link CodecRegistry}. - * + * * @param query must not be empty. * @param parameterBinder must not be {@literal null}. * @param codecRegistry must not be {@literal null}. @@ -71,7 +71,7 @@ class StringBasedQuery { /** * Bind the query to actual parameters using {@link CassandraParameterAccessor}, - * + * * @param parameterAccessor must not be {@literal null}. * @param queryMethod must not be {@literal null}. * @return the bound String query containing formatted parameters. @@ -80,10 +80,10 @@ class StringBasedQuery { Assert.notNull(parameterAccessor, "CassandraParameterAccessor must not be null"); Assert.notNull(queryMethod, "CassandraQueryMethod must not be null"); - + List arguments = parameterBinder.bind(parameterAccessor, new BindingContext(queryMethod, queryParameterBindings)); - + return ParameterBinder.INSTANCE.bind(query, codecRegistry, arguments); } @@ -344,8 +344,8 @@ class StringBasedQuery { .expression(input.substring(exprStart + 3, currentPosition - 1), true)); } else { if (matcher.pattern() == INDEX_PARAMETER_BINDING_PATTERN) { - bindings - .add(ExpressionEvaluatingParameterBinder.ParameterBinding.indexed(Integer.parseInt(matcher.group(1)))); + bindings.add(ExpressionEvaluatingParameterBinder.ParameterBinding.indexed( + Integer.parseInt(matcher.group(1)))); } else { bindings.add(ExpressionEvaluatingParameterBinder.ParameterBinding.named(matcher.group(1))); } diff --git a/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/repository/support/ReactiveCassandraRepositoryFactory.java b/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/repository/support/ReactiveCassandraRepositoryFactory.java index 8139167f5..a67acafc7 100644 --- a/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/repository/support/ReactiveCassandraRepositoryFactory.java +++ b/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/repository/support/ReactiveCassandraRepositoryFactory.java @@ -40,7 +40,6 @@ import org.springframework.data.repository.query.EvaluationContextProvider; import org.springframework.data.repository.query.QueryLookupStrategy; import org.springframework.data.repository.query.QueryLookupStrategy.Key; import org.springframework.data.repository.query.RepositoryQuery; -import org.springframework.data.repository.util.QueryExecutionConverters; import org.springframework.data.repository.util.ReactiveWrapperConverters; import org.springframework.data.repository.util.ReactiveWrappers; import org.springframework.expression.spel.standard.SpelExpressionParser; @@ -98,6 +97,7 @@ public class ReactiveCassandraRepositoryFactory extends RepositoryFactorySupport CassandraEntityInformation entityInformation = getEntityInformation(information.getDomainType(), information); + return getTargetRepositoryViaReflection(information, entityInformation, operations); } @@ -213,10 +213,12 @@ public class ReactiveCassandraRepositoryFactory extends RepositoryFactorySupport ReactiveCassandraQueryMethod queryMethod = new ReactiveCassandraQueryMethod(method, metadata, factory, mappingContext); + String namedQueryName = queryMethod.getNamedQueryName(); if (namedQueries.hasQuery(namedQueryName)) { String namedQuery = namedQueries.getQuery(namedQueryName); + return new ReactiveStringBasedCassandraQuery(namedQuery, queryMethod, operations, EXPRESSION_PARSER, evaluationContextProvider); } else if (queryMethod.hasAnnotatedQuery()) { diff --git a/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/repository/support/ReactiveCassandraRepositoryFactoryBean.java b/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/repository/support/ReactiveCassandraRepositoryFactoryBean.java index a364f8be8..7dd27b597 100644 --- a/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/repository/support/ReactiveCassandraRepositoryFactoryBean.java +++ b/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/repository/support/ReactiveCassandraRepositoryFactoryBean.java @@ -18,7 +18,6 @@ package org.springframework.data.cassandra.repository.support; import java.io.Serializable; -import org.springframework.data.cassandra.core.CassandraTemplate; import org.springframework.data.cassandra.core.ReactiveCassandraOperations; import org.springframework.data.mapping.context.MappingContext; import org.springframework.data.repository.Repository; @@ -38,9 +37,10 @@ import org.springframework.util.Assert; public class ReactiveCassandraRepositoryFactoryBean, S, ID extends Serializable> extends RepositoryFactoryBeanSupport { - private ReactiveCassandraOperations operations; private boolean mappingContextConfigured = false; + private ReactiveCassandraOperations operations; + /** * Configures the {@link ReactiveCassandraOperations} used for Cassandra data access operations. * @@ -59,6 +59,7 @@ public class ReactiveCassandraRepositoryFactoryBean, protected void setMappingContext(MappingContext mappingContext) { super.setMappingContext(mappingContext); + this.mappingContextConfigured = true; } @@ -92,6 +93,7 @@ public class ReactiveCassandraRepositoryFactoryBean, public void afterPropertiesSet() { super.afterPropertiesSet(); + Assert.notNull(operations, "ReactiveCassandraOperations must not be null!"); if (!mappingContextConfigured) {