DATACASS-335 - Polish.

spring-cql/src/main/java/org/springframework/cassandra/core/ColumnMapRowMapper.java

@@ -53,6 +53,7 @@ public class ColumnMapRowMapper implements RowMapper<Map<String, Object>> {
 			Object obj = getColumnValue(rs, i);
 			mapOfColValues.put(key, obj);
 		}
+
 		return mapOfColValues;
 	}
 
@@ -60,7 +61,7 @@ public class ColumnMapRowMapper implements RowMapper<Map<String, Object>> {
 	 * Create a {@link Map} instance to be used as column map.
 	 * <p>
 	 * 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<Map<String, Object>> {
 
 	/**
 	 * 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<Map<String, Object>> {
 	 * Retrieve a CQL object value for the specified column.
 	 * <p>
 	 * 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

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));
 			}
 		}));
 	}

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<Row> rows() {
 
 			int prefetch = Math.max(1, resultSet.getAvailableWithoutFetching());
+
 			return Flux.fromIterable(resultSet) //
 					.subscribeOn(scheduler) //
 					.publishOn(Schedulers.immediate(), prefetch); // limit prefetching to available size

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 <T> Flux<T> query(String cql, ReactiveResultSetExtractor<T> rse) throws DataAccessException {
+	public <T> Flux<T> query(String cql, ReactiveResultSetExtractor<T> 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 <T> Mono<T> queryForObject(String cql, RowMapper<T> 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 <T> Mono<T> queryForObject(Statement statement, RowMapper<T> 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 <T> Flux<T> query(ReactivePreparedStatementCreator psc, PreparedStatementBinder psb,
+	public <T> Flux<T> query(ReactivePreparedStatementCreator psc, PreparedStatementBinder preparedStatementBinder,
 			ReactiveResultSetExtractor<T> 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 <T> Flux<T> query(ReactivePreparedStatementCreator psc, ReactiveResultSetExtractor<T> rse)
 			throws DataAccessException {
+
 		return query(psc, null, rse);
 	}
 
@@ -509,6 +513,7 @@ public class ReactiveCqlTemplate extends ReactiveCassandraAccessor implements Re
 	@Override
 	public <T> Flux<T> query(String cql, PreparedStatementBinder psb, ReactiveResultSetExtractor<T> rse)
 			throws DataAccessException {
+
 		return query(new SimpleReactivePreparedStatementCreator(cql), psb, rse);
 	}
 
@@ -542,6 +547,7 @@ public class ReactiveCqlTemplate extends ReactiveCassandraAccessor implements Re
 	@Override
 	public <T> Flux<T> query(ReactivePreparedStatementCreator psc, PreparedStatementBinder psb, RowMapper<T> rowMapper)
 			throws DataAccessException {
+
 		return query(psc, psb, new ReactiveRowMapperResultSetExtractor<>(rowMapper));
 	}
 
@@ -558,8 +564,8 @@ public class ReactiveCqlTemplate extends ReactiveCassandraAccessor implements Re
 	 */
 	@Override
 	public <T> Mono<T> queryForObject(String cql, RowMapper<T> 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<Boolean> 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 <T> Function<Throwable, Mono<? extends T>> 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 <T> Function<Throwable, Mono<? extends T>> 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 {

spring-cql/src/main/java/org/springframework/cassandra/core/ReactiveRowMapperResultSetExtractor.java

@@ -44,7 +44,7 @@ public class ReactiveRowMapperResultSetExtractor<T> 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<T> rowMapper) {
@@ -67,6 +67,7 @@ public class ReactiveRowMapperResultSetExtractor<T> implements ReactiveResultSet
 			if (value == null) {
 				return Mono.empty();
 			}
+
 			return Mono.just(value);
 		});
 	}

spring-cql/src/main/java/org/springframework/cassandra/core/SingleColumnRowMapper.java

@@ -43,7 +43,7 @@ public class SingleColumnRowMapper<T> implements RowMapper<T> {
 
 	/**
 	 * Create a new {@link SingleColumnRowMapper} for bean-style configuration.
-	 * 
+	 *
 	 * @see #setRequiredType
 	 */
 	public SingleColumnRowMapper() {}
@@ -53,7 +53,7 @@ public class SingleColumnRowMapper<T> implements RowMapper<T> {
 	 * <p>
 	 * 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<T> requiredType) {
@@ -74,7 +74,7 @@ public class SingleColumnRowMapper<T> implements RowMapper<T> {
 	 * <p>
 	 * 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<T> implements RowMapper<T> {
 		// 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<T> implements RowMapper<T> {
 								definitions.getType(0), ex.getMessage()));
 			}
 		}
+
 		return (T) result;
 	}
 
@@ -111,7 +114,7 @@ public class SingleColumnRowMapper<T> implements RowMapper<T> {
 	 * 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<T> implements RowMapper<T> {
 	 * 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<T> implements RowMapper<T> {
 		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<Number>) requiredType);
@@ -183,7 +185,7 @@ public class SingleColumnRowMapper<T> implements RowMapper<T> {
 
 	/**
 	 * 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 <T> SingleColumnRowMapper<T> newInstance(Class<T> requiredType) {

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<T> 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 <T> Flux<T> insert(Publisher<? extends T> 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<T> 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 <T> Flux<T> update(Publisher<? extends T> 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<T> 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 <T> Flux<T> delete(Publisher<? extends T> 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<Void> 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();

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;
 	}
-
 }

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<Class<? extends Annotation>> getIdentifyingAnnotations() {
-		return Collections.<Class<? extends Annotation>> singleton(Table.class);
+		return Collections.<Class<? extends Annotation>>singleton(Table.class);
 	}
 
 	/*
@@ -106,7 +107,7 @@ public class ReactiveCassandraRepositoryConfigurationExtension extends Repositor
 	 */
 	@Override
 	protected Collection<Class<?>> getIdentifyingTypes() {
-		return Collections.<Class<?>> singleton(ReactiveCassandraRepository.class);
+		return Collections.<Class<?>>singleton(ReactiveCassandraRepository.class);
 	}
 
 	/*

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<Object, Object> 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);
+
 }

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> 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;
 	}
 

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;
 	}
 }

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);
 	}
 

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<Object> 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)));
 					}

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<?, Serializable> 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()) {

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<T extends Repository<S, ID>, S, ID extends Serializable>
 		extends RepositoryFactoryBeanSupport<T, S, ID> {
 
-	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<T extends Repository<S, ID>,
 	protected void setMappingContext(MappingContext<?, ?> mappingContext) {
 
 		super.setMappingContext(mappingContext);
+
 		this.mappingContextConfigured = true;
 	}
 
@@ -92,6 +93,7 @@ public class ReactiveCassandraRepositoryFactoryBean<T extends Repository<S, ID>,
 	public void afterPropertiesSet() {
 
 		super.afterPropertiesSet();
+
 		Assert.notNull(operations, "ReactiveCassandraOperations must not be null!");
 
 		if (!mappingContextConfigured) {