DATACASS-335 - Adopt changes in Spring Data Commons.

- Remove version placeholders for reactor and rxjava
- Adopt type migration of ReactiveWrappers
- Adopt RxJava to RxJava1 repository interface renaming
- Use ReactiveQueryMethod in ReactiveMongoQuery.
- Remove trailing whitespaces.
- Use ReflectionUtils for method iteration in ReactiveType.

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

@@ -63,7 +63,7 @@ public class DefaultBridgedReactiveSession implements ReactiveSession {
 
 	/**
 	 * Creates a new {@link DefaultBridgedReactiveSession} for a {@link Session} and {@link Scheduler}.
-	 * 
+	 *
 	 * @param session must not be {@literal null}.
 	 * @param scheduler must not be {@literal null}.
 	 */

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

@@ -19,7 +19,7 @@ package org.springframework.cassandra.core;
  * Default implementation of {@link ReactiveSessionFactory}.
  * <p>
  * This implementation returns always the same {@link ReactiveSession}.
- * 
+ *
  * @author Mark Paluch
  * @since 2.0
  */

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

@@ -52,7 +52,7 @@ public interface ReactiveCqlOperations {
 	 * {@link DataAccessException} hierarchy.
 	 * <p>
 	 * The callback action can return a result object, for example a domain object or a collection of domain objects.
-	 * 
+	 *
 	 * @param action the callback object that specifies the action.
 	 * @return a result object returned by the action, or {@literal null}.
 	 * @throws DataAccessException if there is any problem executing the query.
@@ -65,7 +65,7 @@ public interface ReactiveCqlOperations {
 
 	/**
 	 * Issue a single CQL execute, typically a DDL statement, insert, update or delete statement.
-	 * 
+	 *
 	 * @param cql static CQL to execute, must not be empty or {@literal null}.
 	 * @return boolean value whether the statement was applied.
 	 * @throws DataAccessException if there is any problem executing the query.
@@ -77,7 +77,7 @@ public interface ReactiveCqlOperations {
 	 * <p>
 	 * Uses a CQL Statement, not a {@link PreparedStatement}. If you want to execute a static query with a
 	 * {@link PreparedStatement}, use the overloaded {@code query} method with {@literal null} as argument array.
-	 * 
+	 *
 	 * @param cql static CQL to execute, must not be empty or {@literal null}.
 	 * @param rse object that will extract all rows of results, must not be {@literal null}.
 	 * @return an arbitrary result object, as returned by the ReactiveResultSetExtractor.
@@ -91,7 +91,7 @@ public interface ReactiveCqlOperations {
 	 * <p>
 	 * Uses a CQL Statement, not a {@link PreparedStatement}. If you want to execute a static query with a
 	 * {@link PreparedStatement}, use the overloaded {@code query} method with {@literal null} as argument array.
-	 * 
+	 *
 	 * @param cql static CQL to execute, must not be empty or {@literal null}.
 	 * @param rowMapper object that will map one object per row, must not be {@literal null}.
 	 * @return the result {@link Flux}, containing mapped objects.
@@ -106,7 +106,7 @@ public interface ReactiveCqlOperations {
 	 * Uses a CQL Statement, not a {@link PreparedStatement}. If you want to execute a static query with a
 	 * {@link PreparedStatement}, use the overloaded {@link #queryForObject(String, RowMapper, Object...)} method with
 	 * {@literal null} as argument array.
-	 * 
+	 *
 	 * @param cql static CQL to execute, must not be empty or {@literal null}.
 	 * @param rowMapper object that will map one object per row, must not be {@literal null}.
 	 * @return the single mapped object.
@@ -125,7 +125,7 @@ public interface ReactiveCqlOperations {
 	 * <p>
 	 * This method is useful for running static CQL with a known outcome. The query is expected to be a single row/single
 	 * column query; the returned result will be directly mapped to the corresponding object type.
-	 * 
+	 *
 	 * @param cql static CQL to execute, must not be empty or {@literal null}.
 	 * @param requiredType the type that the result object is expected to match, must not be {@literal null}.
 	 * @return the result object of the required type, or {@link Mono#empty()} in case of CQL NULL.
@@ -145,7 +145,7 @@ public interface ReactiveCqlOperations {
 	 * <p>
 	 * The query is expected to be a single row query; the result row will be mapped to a Map (one entry for each column,
 	 * using the column name as the key).
-	 * 
+	 *
 	 * @param cql static CQL to execute, must not be empty or {@literal null}.
 	 * @return the result Map (one entry for each column, using the column name as the key), must not be {@literal null}.
 	 * @throws IncorrectResultSizeDataAccessException if the query does not return exactly one row.
@@ -163,7 +163,7 @@ public interface ReactiveCqlOperations {
 	 * <p>
 	 * The results will be mapped to a {@link Flux} (one item for each row) of result objects, each of them matching the
 	 * specified element type.
-	 * 
+	 *
 	 * @param cql static CQL to execute, must not be empty or {@literal null}.
 	 * @param elementType the required type of element in the result {@link Flux} (for example, {@code Integer.class}),
 	 *          must not be {@literal null}.
@@ -183,7 +183,7 @@ public interface ReactiveCqlOperations {
 	 * The results will be mapped to a {@link Flux} (one item for each row) of {@link Map}s (one entry for each column
 	 * using the column name as the key). Each item in the {@link Flux} will be of the form returned by this interface's
 	 * queryForMap() methods.
-	 * 
+	 *
 	 * @param cql static CQL to execute, must not be empty or {@literal null}.
 	 * @return a {@link Flux} that contains a {@link Map} per row.
 	 * @throws DataAccessException if there is any problem executing the query.
@@ -407,7 +407,7 @@ public interface ReactiveCqlOperations {
 	 * {@link com.datastax.driver.core.exceptions.DriverException}s into Spring's {@link DataAccessException} hierarchy.
 	 * <p>
 	 * The callback action can return a result object, for example a domain object or a collection of domain objects.
-	 * 
+	 *
 	 * @param psc object that can create a {@link PreparedStatement} given a {@link ReactiveSession}, must not be
 	 *          {@literal null}.
 	 * @param action callback object that specifies the action, must not be {@literal null}.
@@ -424,7 +424,7 @@ public interface ReactiveCqlOperations {
 	 * {@link com.datastax.driver.core.exceptions.DriverException}s into Spring's {@link DataAccessException} hierarchy.
 	 * <p>
 	 * The callback action can return a result object, for example a domain object or a collection of domain objects.
-	 * 
+	 *
 	 * @param cql static CQL to execute, must not be empty or {@literal null}.
 	 * @param action callback object that specifies the action, must not be {@literal null}.
 	 * @return a result object returned by the action, or {@literal null}
@@ -434,7 +434,7 @@ public interface ReactiveCqlOperations {
 
 	/**
 	 * 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}, must not be
 	 *          {@literal null}.
 	 * @param rse object that will extract results, must not be {@literal null}.
@@ -445,7 +445,7 @@ public interface ReactiveCqlOperations {
 
 	/**
 	 * Query using a prepared statement, reading the {@link ReactiveResultSet} with a {@link ReactiveResultSetExtractor}.
-	 * 
+	 *
 	 * @param cql static CQL to execute, must not be empty or {@literal null}.
 	 * @param psb 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. Even if there are no bind parameters, this object may be used to
@@ -476,7 +476,7 @@ public interface ReactiveCqlOperations {
 	/**
 	 * Query given CQL to create a prepared statement from CQL and a list of arguments to bind to the query, reading the
 	 * {@link ReactiveResultSet} with a {@link ReactiveResultSetExtractor}.
-	 * 
+	 *
 	 * @param cql static CQL to execute, must not be empty or {@literal null}.
 	 * @param rse object that will extract results, must not be {@literal null}.
 	 * @param args arguments to bind to the query (leaving it to the {@link PreparedStatement} to guess the corresponding
@@ -488,7 +488,7 @@ public interface ReactiveCqlOperations {
 
 	/**
 	 * Query using a prepared statement, mapping each row to a Java object via a {@link RowMapper}.
-	 * 
+	 *
 	 * @param psc object that can create a {@link PreparedStatement} given a {@link ReactiveSession}, must not be
 	 *          {@literal null}.
 	 * @param rowMapper object that will map one object per row, must not be {@literal null}.
@@ -500,7 +500,7 @@ public interface ReactiveCqlOperations {
 	/**
 	 * Query given CQL to create a prepared statement from CQL and a {@link PreparedStatement}Binder implementation that
 	 * knows how to bind values to the query, mapping each row to a Java object via a {@link RowMapper}.
-	 * 
+	 *
 	 * @param cql static CQL to execute, must not be empty or {@literal null}.
 	 * @param psb 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. Even if there are no bind parameters, this object may be used to
@@ -530,7 +530,7 @@ public interface ReactiveCqlOperations {
 	/**
 	 * Query given CQL to create a prepared statement from CQL and a list of arguments to bind to the query, mapping each
 	 * row to a Java object via a {@link RowMapper}.
-	 * 
+	 *
 	 * @param cql static CQL to execute, must not be empty or {@literal null}.
 	 * @param rowMapper object that will map one object per row
 	 * @param args arguments to bind to the query (leaving it to the {@link PreparedStatement} to guess the corresponding
@@ -543,7 +543,7 @@ public interface ReactiveCqlOperations {
 	/**
 	 * Query given CQL to create a prepared statement from CQL and a list of arguments to bind to the query, mapping a
 	 * single result row to a Java object via a {@link RowMapper}.
-	 * 
+	 *
 	 * @param cql static CQL to execute, must not be empty or {@literal null}.
 	 * @param rowMapper object that will map one object per row, must not be {@literal null}.
 	 * @param args arguments to bind to the query (leaving it to the {@link PreparedStatement} to guess the corresponding
@@ -560,7 +560,7 @@ public interface ReactiveCqlOperations {
 	 * <p>
 	 * The query is expected to be a single row/single column query; the returned result will be directly mapped to the
 	 * corresponding object type.
-	 * 
+	 *
 	 * @param cql static CQL to execute, must not be empty or {@literal null}.
 	 * @param requiredType the type that the result object is expected to match, must not be {@literal null}.
 	 * @param args arguments to bind to the query (leaving it to the PreparedStatement to guess the corresponding CQL
@@ -580,7 +580,7 @@ public interface ReactiveCqlOperations {
 	 * <p>
 	 * The query is expected to be a single row query; the result row will be mapped to a Map (one entry for each column,
 	 * using the column name as the key).
-	 * 
+	 *
 	 * @param cql static CQL to execute, must not be empty or {@literal null}.
 	 * @param args arguments to bind to the query (leaving it to the {@link PreparedStatement} to guess the corresponding
 	 *          CQL type).
@@ -598,7 +598,7 @@ public interface ReactiveCqlOperations {
 	 * <p>
 	 * The results will be mapped to a {@link Flux} (one item for each row) of result objects, each of them matching the
 	 * specified element type.
-	 * 
+	 *
 	 * @param cql static CQL to execute, must not be empty or {@literal null}.
 	 * @param elementType the required type of element in the result {@link Flux} (for example, {@code Integer.class}),
 	 *          must not be {@literal null}.
@@ -618,7 +618,7 @@ public interface ReactiveCqlOperations {
 	 * The results will be mapped to a {@link Flux} (one item for each row) of {@link Map}s (one entry for each column,
 	 * using the column name as the key). Each item in the {@link Flux} will be of the form returned by this interface's
 	 * queryForMap() methods.
-	 * 
+	 *
 	 * @param cql static CQL to execute, must not be empty or {@literal null}.
 	 * @param args arguments to bind to the query (leaving it to the {@link PreparedStatement} to guess the corresponding
 	 *          CQL type).
@@ -698,7 +698,7 @@ public interface ReactiveCqlOperations {
 	/**
 	 * Issue a single CQL operation (such as an insert, update or delete statement) via a prepared statement, binding the
 	 * given arguments.
-	 * 
+	 *
 	 * @param cql static CQL to execute containing bind parameters, must not be empty or {@literal null}.
 	 * @param args arguments to bind to the query (leaving it to the {@link PreparedStatement} to guess the corresponding
 	 *          CQL type).

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

@@ -831,7 +831,7 @@ public class ReactiveCqlTemplate extends ReactiveCassandraAccessor implements Re
 
 	/**
 	 * Determine CQL from potential provider object.
-	 * 
+	 *
 	 * @param cqlProvider object that's potentially a {@link CqlProvider}
 	 * @return the CQL string, or {@code null}
 	 * @see CqlProvider

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

@@ -48,7 +48,7 @@ public interface ReactivePreparedStatementCallback<T> {
 	 * objects. Note that there's special support for single step actions: see
 	 * {@link ReactiveCqlTemplate#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 ps the {@link PreparedStatement}, must not be {@literal null}.
 	 * @return a result object publisher.

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

@@ -43,7 +43,7 @@ public interface ReactiveResultSetExtractor<T> {
 
 	/**
 	 * Implementations must implement this method to process the entire {@link ReactiveResultSet}.
-	 * 
+	 *
 	 * @param resultSet {@link ReactiveResultSet} to extract data from, must not be {@literal null}.
 	 * @return an arbitrary result object {@link Publisher}.
 	 * @throws DriverException if a {@link DriverException} is encountered getting column values or navigating (that is,

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

@@ -144,14 +144,14 @@ public interface ReactiveSession extends Closeable {
 	 * <p/>
 	 * This method behaves like {@link #prepare(String)}, but note that the resulting {@code PreparedStatement} will
 	 * inherit the query properties set on {@code statement}. Concretely, this means that in the following code:
-	 * 
+	 *
 	 * <pre>
 	 * RegularStatement toPrepare = new SimpleStatement("SELECT * FROM test WHERE k=?")
 	 * 		.setConsistencyLevel(ConsistencyLevel.QUORUM);
 	 * PreparedStatement prepared = session.prepare(toPrepare);
 	 * session.execute(prepared.bind("someValue"));
 	 * </pre>
-	 * 
+	 *
 	 * the final execution will be performed with Quorum consistency.
 	 * <p/>
 	 * Please note that if the same CQL statement is prepared more than once, all calls to this method will return the

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

@@ -38,14 +38,14 @@ import com.datastax.driver.core.exceptions.DriverException;
 public interface ReactiveSessionCallback<T> {
 
 	/**
-	 * Gets called by {@link ReactiveCqlTemplate#execute(ReactiveSessionCallback)} with an active Cassandra session. Does not
-	 * need to care about activating or closing the {@link ReactiveSession}.
+	 * Gets called by {@link ReactiveCqlTemplate#execute(ReactiveSessionCallback)} with an active Cassandra session. Does
+	 * not need to care about activating or closing the {@link ReactiveSession}.
 	 * <p>
 	 * Allows for returning a result object created within the callback, i.e. a domain object or a collection of domain
 	 * objects. Note that there's special support for single step actions: see
 	 * {@link ReactiveCqlTemplate#queryForObject(Statement, Class)} etc. A thrown {@link RuntimeException} is treated as
 	 * application exception: it gets propagated to the caller of the template.
-	 * 
+	 *
 	 * @param session active Cassandra session.
 	 * @return a result object publisher
 	 * @throws DriverException if thrown by a session method, to be auto-converted to a DataAccessException

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

@@ -35,14 +35,14 @@ import com.datastax.driver.core.exceptions.DriverException;
 public interface ReactiveStatementCallback<T> {
 
 	/**
-	 * Gets called by {@link ReactiveCqlTemplate#execute(String)} with an active Cassandra session. Does not need to care about
-	 * closing the the session: this will all be handled by Spring's {@link ReactiveCqlTemplate}.
+	 * Gets called by {@link ReactiveCqlTemplate#execute(String)} with an active Cassandra session. Does not need to care
+	 * about closing the the session: this will all be handled by Spring's {@link ReactiveCqlTemplate}.
 	 * <p>
 	 * Allows for returning a result object created within the callback, i.e. a domain object or a collection of domain
 	 * objects. Note that there's special support for single step actions: see
 	 * {@link ReactiveCqlTemplate#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.
 	 * @param stmt CQL Statement
 	 * @return a result object publisher

spring-data-cassandra/pom.xml

@@ -86,6 +86,13 @@
 			<optional>true</optional>
 		</dependency>
 
+		<dependency>
+			<groupId>io.reactivex</groupId>
+			<artifactId>rxjava-reactive-streams</artifactId>
+			<version>${rxjava-reactive-streams}</version>
+			<optional>true</optional>
+		</dependency>
+
 		<!-- CDI -->
 		<dependency>
 			<groupId>javax.enterprise</groupId>

spring-data-cassandra/src/main/java/org/springframework/data/cassandra/config/java/AbstractReactiveCassandraConfiguration.java

@@ -17,11 +17,13 @@ package org.springframework.data.cassandra.config.java;
 
 import org.springframework.cassandra.core.DefaultBridgedReactiveSession;
 import org.springframework.cassandra.core.DefaultReactiveSessionFactory;
+import org.springframework.cassandra.core.ReactiveCqlOperations;
 import org.springframework.cassandra.core.ReactiveCqlTemplate;
 import org.springframework.cassandra.core.ReactiveSession;
 import org.springframework.cassandra.core.ReactiveSessionFactory;
 import org.springframework.context.annotation.Bean;
 import org.springframework.data.cassandra.core.CassandraAdminTemplate;
+import org.springframework.data.cassandra.core.ReactiveCassandraOperations;
 import org.springframework.data.cassandra.core.ReactiveCassandraTemplate;
 
 import reactor.core.scheduler.Schedulers;
@@ -29,7 +31,7 @@ import reactor.core.scheduler.Schedulers;
 /**
  * Extension to {@link AbstractCassandraConfiguration} providing Spring Data Cassandra configuration for Spring Data's
  * Reactive Cassandra support using JavaConfig.
- * 
+ *
  * @author Mark Paluch
  * @since 2.0
  */
@@ -38,7 +40,7 @@ public abstract class AbstractReactiveCassandraConfiguration extends AbstractCas
 	/**
 	 * Creates a {@link ReactiveSession} object. This wraps a {@link com.datastax.driver.core.Session} to expose Cassandra
 	 * access in a reactive style.
-	 * 
+	 *
 	 * @return
 	 * @see #session()
 	 * @see DefaultBridgedReactiveSession
@@ -51,7 +53,7 @@ public abstract class AbstractReactiveCassandraConfiguration extends AbstractCas
 	/**
 	 * Creates a {@link ReactiveSessionFactory} to be used by the {@link ReactiveCassandraTemplate}. Will use the
 	 * {@link ReactiveSession} instance configured in {@link #reactiveSession()}.
-	 * 
+	 *
 	 * @return
 	 * @see #reactiveSession()
 	 * @see #reactiveCassandraTemplate()
@@ -63,24 +65,24 @@ public abstract class AbstractReactiveCassandraConfiguration extends AbstractCas
 
 	/**
 	 * Creates a {@link CassandraAdminTemplate}.
-	 * 
+	 *
 	 * @return
 	 * @see #reactiveSessionFactory()
 	 * @see #cassandraConverter()
 	 */
 	@Bean
-	public ReactiveCassandraTemplate reactiveCassandraTemplate() throws Exception {
+	public ReactiveCassandraOperations reactiveCassandraTemplate() throws Exception {
 		return new ReactiveCassandraTemplate(reactiveSessionFactory(), cassandraConverter());
 	}
 
 	/**
 	 * Creates a {@link ReactiveCqlTemplate} using the configured {@link ReactiveSessionFactory}.
-	 * 
+	 *
 	 * @return
 	 * @see #reactiveSessionFactory()
 	 */
 	@Bean
-	public ReactiveCqlTemplate reactiveCqlTemplate() throws Exception {
+	public ReactiveCqlOperations reactiveCqlTemplate() throws Exception {
 		return new ReactiveCqlTemplate(reactiveSessionFactory());
 	}
 }

spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/ReactiveCassandraTemplate.java

@@ -18,16 +18,9 @@ package org.springframework.data.cassandra.core;
 import static org.springframework.data.cassandra.core.CassandraTemplate.*;
 
 import org.reactivestreams.Publisher;
-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.*;
 import org.springframework.cassandra.core.cql.CqlIdentifier;
+import org.springframework.dao.DataAccessException;
 import org.springframework.dao.InvalidDataAccessApiUsageException;
 import org.springframework.data.cassandra.convert.CassandraConverter;
 import org.springframework.data.cassandra.convert.MappingCassandraConverter;
@@ -39,6 +32,7 @@ import org.springframework.util.ClassUtils;
 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.Delete;
 import com.datastax.driver.core.querybuilder.Insert;
 import com.datastax.driver.core.querybuilder.QueryBuilder;
@@ -149,7 +143,7 @@ public class ReactiveCassandraTemplate implements ReactiveCassandraOperations {
 	// Methods dealing with static CQL
 	// -------------------------------------------------------------------------
 
-	/* 
+	/*
 	 * (non-Javadoc)
 	 * @see org.springframework.data.cassandra.core.ReactiveCassandraOperations#select(java.lang.String, java.lang.Class)
 	 */
@@ -174,7 +168,7 @@ public class ReactiveCassandraTemplate implements ReactiveCassandraOperations {
 	// Methods dealing with com.datastax.driver.core.Statement
 	// -------------------------------------------------------------------------
 
-	/* 
+	/*
 	 * (non-Javadoc)
 	 * @see org.springframework.data.cassandra.core.ReactiveCassandraOperations#select(com.datastax.driver.core.Statement, java.lang.Class)
 	 */
@@ -200,7 +194,7 @@ public class ReactiveCassandraTemplate implements ReactiveCassandraOperations {
 	// Methods dealing with entities
 	// -------------------------------------------------------------------------
 
-	/* 
+	/*
 	 * (non-Javadoc)
 	 * @see org.springframework.data.cassandra.core.ReactiveCassandraOperations#selectOneById(java.lang.Object, java.lang.Class)
 	 */
@@ -218,7 +212,7 @@ public class ReactiveCassandraTemplate implements ReactiveCassandraOperations {
 		return selectOne(select, entityClass);
 	}
 
-	/* 
+	/*
 	 * (non-Javadoc)
 	 * @see org.springframework.data.cassandra.core.ReactiveCassandraOperations#exists(java.lang.Object, java.lang.Class)
 	 */
@@ -235,7 +229,7 @@ public class ReactiveCassandraTemplate implements ReactiveCassandraOperations {
 		return cqlOperations.queryForRows(select).hasElements();
 	}
 
-	/* 
+	/*
 	 * (non-Javadoc)
 	 * @see org.springframework.data.cassandra.core.ReactiveCassandraOperations#count(java.lang.Class)
 	 */
@@ -249,7 +243,7 @@ public class ReactiveCassandraTemplate implements ReactiveCassandraOperations {
 		return cqlOperations.queryForObject(select, Long.class);
 	}
 
-	/* 
+	/*
 	 * (non-Javadoc)
 	 * @see org.springframework.data.cassandra.core.ReactiveCassandraOperations#insert(java.lang.Object)
 	 */
@@ -258,7 +252,7 @@ public class ReactiveCassandraTemplate implements ReactiveCassandraOperations {
 		return insert(entity, null);
 	}
 
-	/* 
+	/*
 	 * (non-Javadoc)
 	 * @see org.springframework.data.cassandra.core.ReactiveCassandraOperations#insert(java.lang.Object, org.springframework.cassandra.core.WriteOptions)
 	 */
@@ -269,10 +263,23 @@ public class ReactiveCassandraTemplate implements ReactiveCassandraOperations {
 
 		CqlIdentifier tableName = getTableName(entity);
 
-		Insert insertQuery = createInsertQuery(tableName.toCql(), entity, options, converter);
+		Insert insert = createInsertQuery(tableName.toCql(), entity, options, converter);
 
-		return cqlOperations.execute((ReactiveSessionCallback<T>) session -> (Publisher<T>) session.execute(insertQuery)
-				.flatMap(reactiveResultSet -> reactiveResultSet.wasApplied() ? Mono.just(entity) : Mono.empty())).next();
+		class InsertCallback implements ReactiveSessionCallback<T>, CqlProvider {
+
+			@Override
+			public Publisher<T> doInSession(ReactiveSession session) throws DriverException, DataAccessException {
+				return session.execute(insert)
+						.flatMap(reactiveResultSet -> reactiveResultSet.wasApplied() ? Mono.just(entity) : Mono.empty());
+			}
+
+			@Override
+			public String getCql() {
+				return insert.toString();
+			}
+		}
+
+		return cqlOperations.execute(new InsertCallback()).next();
 	}
 
 	/*
@@ -284,7 +291,7 @@ public class ReactiveCassandraTemplate implements ReactiveCassandraOperations {
 		return insert(entities, null);
 	}
 
-	/* 
+	/*
 	 * (non-Javadoc)
 	 * @see org.springframework.data.cassandra.core.ReactiveCassandraOperations#insert(org.reactivestreams.Publisher, org.springframework.cassandra.core.WriteOptions)
 	 */
@@ -295,7 +302,7 @@ public class ReactiveCassandraTemplate implements ReactiveCassandraOperations {
 		return Flux.from(entities).flatMap(entity -> insert(entity, options));
 	}
 
-	/* 
+	/*
 	 * (non-Javadoc)
 	 * @see org.springframework.data.cassandra.core.ReactiveCassandraOperations#update(java.lang.Object)
 	 */
@@ -304,7 +311,7 @@ public class ReactiveCassandraTemplate implements ReactiveCassandraOperations {
 		return update(entity, null);
 	}
 
-	/* 
+	/*
 	 * (non-Javadoc)
 	 * @see org.springframework.data.cassandra.core.ReactiveCassandraOperations#update(java.lang.Object, org.springframework.cassandra.core.WriteOptions)
 	 */
@@ -317,8 +324,21 @@ public class ReactiveCassandraTemplate implements ReactiveCassandraOperations {
 
 		Update update = createUpdateQuery(tableName.toCql(), entity, options, converter);
 
-		return cqlOperations.execute((ReactiveSessionCallback<T>) session -> (Publisher<T>) session.execute(update)
-				.flatMap(reactiveResultSet -> reactiveResultSet.wasApplied() ? Mono.just(entity) : Mono.empty())).next();
+		class UpdateCallback implements ReactiveSessionCallback<T>, CqlProvider {
+
+			@Override
+			public Publisher<T> doInSession(ReactiveSession session) throws DriverException, DataAccessException {
+				return session.execute(update)
+						.flatMap(reactiveResultSet -> reactiveResultSet.wasApplied() ? Mono.just(entity) : Mono.empty());
+			}
+
+			@Override
+			public String getCql() {
+				return update.toString();
+			}
+		}
+
+		return cqlOperations.execute(new UpdateCallback()).next();
 	}
 
 	/*
@@ -330,7 +350,7 @@ public class ReactiveCassandraTemplate implements ReactiveCassandraOperations {
 		return update(entities, null);
 	}
 
-	/* 
+	/*
 	 * (non-Javadoc)
 	 * @see org.springframework.data.cassandra.core.ReactiveCassandraOperations#update(org.reactivestreams.Publisher, org.springframework.cassandra.core.WriteOptions)
 	 */
@@ -341,7 +361,7 @@ public class ReactiveCassandraTemplate implements ReactiveCassandraOperations {
 		return Flux.from(entities).flatMap(entity -> update(entity, options));
 	}
 
-	/* 
+	/*
 	 * (non-Javadoc)
 	 * @see org.springframework.data.cassandra.core.ReactiveCassandraOperations#deleteById(java.lang.Object, java.lang.Class)
 	 */
@@ -359,7 +379,7 @@ public class ReactiveCassandraTemplate implements ReactiveCassandraOperations {
 		return cqlOperations.execute(delete);
 	}
 
-	/* 
+	/*
 	 * (non-Javadoc)
 	 * @see org.springframework.data.cassandra.core.ReactiveCassandraOperations#delete(java.lang.Object)
 	 */
@@ -368,7 +388,7 @@ public class ReactiveCassandraTemplate implements ReactiveCassandraOperations {
 		return delete(entity, null);
 	}
 
-	/* 
+	/*
 	 * (non-Javadoc)
 	 * @see org.springframework.data.cassandra.core.ReactiveCassandraOperations#delete(java.lang.Object, org.springframework.cassandra.core.QueryOptions)
 	 */
@@ -381,8 +401,21 @@ public class ReactiveCassandraTemplate implements ReactiveCassandraOperations {
 
 		Delete delete = createDeleteQuery(tableName.toCql(), entity, options, converter);
 
-		return cqlOperations.execute((ReactiveSessionCallback<T>) session -> (Publisher<T>) session.execute(delete)
-				.flatMap(reactiveResultSet -> reactiveResultSet.wasApplied() ? Mono.just(entity) : Mono.empty())).next();
+		class DeleteCallback implements ReactiveSessionCallback<T>, CqlProvider {
+
+			@Override
+			public Publisher<T> doInSession(ReactiveSession session) throws DriverException, DataAccessException {
+				return session.execute(delete)
+						.flatMap(reactiveResultSet -> reactiveResultSet.wasApplied() ? Mono.just(entity) : Mono.empty());
+			}
+
+			@Override
+			public String getCql() {
+				return delete.toString();
+			}
+		}
+
+		return cqlOperations.execute(new DeleteCallback()).next();
 	}
 
 	/*
@@ -394,7 +427,7 @@ public class ReactiveCassandraTemplate implements ReactiveCassandraOperations {
 		return delete(entities, null);
 	}
 
-	/* 
+	/*
 	 * (non-Javadoc)
 	 * @see org.springframework.data.cassandra.core.ReactiveCassandraOperations#delete(org.reactivestreams.Publisher, org.springframework.cassandra.core.QueryOptions)
 	 */
@@ -405,7 +438,7 @@ public class ReactiveCassandraTemplate implements ReactiveCassandraOperations {
 		return Flux.from(entities).flatMap(entity -> delete(entity, options));
 	}
 
-	/* 
+	/*
 	 * (non-Javadoc)
 	 * @see org.springframework.data.cassandra.core.ReactiveCassandraOperations#truncate(java.lang.Class)
 	 */
@@ -418,7 +451,7 @@ public class ReactiveCassandraTemplate implements ReactiveCassandraOperations {
 		return cqlOperations.execute(truncate).then();
 	}
 
-	/* 
+	/*
 	 * (non-Javadoc)
 	 * @see org.springframework.data.cassandra.core.ReactiveCassandraOperations#getConverter()
 	 */

spring-data-cassandra/src/main/java/org/springframework/data/cassandra/repository/config/CassandraRepositoryConfigurationExtension.java

@@ -34,7 +34,7 @@ import org.springframework.data.repository.config.RepositoryConfigurationExtensi
 import org.springframework.data.repository.config.RepositoryConfigurationExtensionSupport;
 import org.springframework.data.repository.config.RepositoryConfigurationSource;
 import org.springframework.data.repository.config.XmlRepositoryConfigurationSource;
-import org.springframework.data.repository.query.ReactiveWrappers;
+import org.springframework.data.repository.util.ReactiveWrappers;
 import org.springframework.util.StringUtils;
 import org.w3c.dom.Element;
 

spring-data-cassandra/src/main/java/org/springframework/data/cassandra/repository/config/EnableReactiveCassandraRepositories.java

@@ -25,7 +25,6 @@ import java.lang.annotation.Target;
 import org.springframework.beans.factory.FactoryBean;
 import org.springframework.context.annotation.ComponentScan.Filter;
 import org.springframework.context.annotation.Import;
-import org.springframework.data.cassandra.repository.support.CassandraRepositoryFactoryBean;
 import org.springframework.data.cassandra.repository.support.ReactiveCassandraRepositoryFactoryBean;
 import org.springframework.data.repository.config.DefaultRepositoryBaseClass;
 import org.springframework.data.repository.query.QueryLookupStrategy;
@@ -35,7 +34,7 @@ import org.springframework.data.repository.query.QueryLookupStrategy.Key;
  * Annotation to activate reactive Cassandra repositories. If no base package is configured through either
  * {@link #value()}, {@link #basePackages()} or {@link #basePackageClasses()} it will trigger scanning of the package of
  * annotated class.
- * 
+ *
  * @author Mark Paluch
  * @since 2.0
  */
@@ -81,7 +80,7 @@ public @interface EnableReactiveCassandraRepositories {
 	 * Returns the postfix to be used when looking up custom repository implementations. Defaults to {@literal Impl}. So
 	 * for a repository named {@code UserRepository} the corresponding implementation class will be looked up scanning for
 	 * {@code UserRepositoryImpl}.
-	 * 
+	 *
 	 * @return
 	 */
 	String repositoryImplementationPostfix() default "Impl";
@@ -89,7 +88,7 @@ public @interface EnableReactiveCassandraRepositories {
 	/**
 	 * Configures the location of where to find the Spring Data named queries properties file. Will default to
 	 * {@code META-INF/cassandra-named-queries.properties}.
-	 * 
+	 *
 	 * @return
 	 */
 	String namedQueriesLocation() default "";
@@ -97,7 +96,7 @@ public @interface EnableReactiveCassandraRepositories {
 	/**
 	 * Returns the key of the {@link QueryLookupStrategy} to be used for lookup queries for query methods. Defaults to
 	 * {@link Key#CREATE_IF_NOT_FOUND}.
-	 * 
+	 *
 	 * @return
 	 */
 	Key queryLookupStrategy() default Key.CREATE_IF_NOT_FOUND;
@@ -105,14 +104,14 @@ public @interface EnableReactiveCassandraRepositories {
 	/**
 	 * Returns the {@link FactoryBean} class to be used for each repository instance. Defaults to
 	 * {@link ReactiveCassandraRepositoryFactoryBean}.
-	 * 
+	 *
 	 * @return
 	 */
 	Class<?> repositoryFactoryBeanClass() default ReactiveCassandraRepositoryFactoryBean.class;
 
 	/**
 	 * Configure the repository base class to be used to create repository proxies for this particular configuration.
-	 * 
+	 *
 	 * @return
 	 */
 	Class<?> repositoryBaseClass() default DefaultRepositoryBaseClass.class;
@@ -120,7 +119,7 @@ public @interface EnableReactiveCassandraRepositories {
 	/**
 	 * Configures the name of the {@link org.springframework.data.cassandra.core.ReactiveCassandraTemplate} bean to be
 	 * used with the repositories detected.
-	 * 
+	 *
 	 * @return
 	 */
 	String reactiveCassandraTemplateRef() default "reactiveCassandraTemplate";

spring-data-cassandra/src/main/java/org/springframework/data/cassandra/repository/config/ReactiveCassandraRepositoriesRegistrar.java

@@ -24,13 +24,13 @@ import org.springframework.data.repository.config.RepositoryConfigurationExtensi
 /**
  * {@link ImportBeanDefinitionRegistrar} to setup Cassandra repositories via
  * {@link EnableReactiveCassandraRepositories}.
- * 
+ *
  * @author Mark Paluch
  * @since 2.0
  */
 class ReactiveCassandraRepositoriesRegistrar extends RepositoryBeanDefinitionRegistrarSupport {
 
-	/* 
+	/*
 	 * (non-Javadoc)
 	 * @see org.springframework.data.repository.config.RepositoryBeanDefinitionRegistrarSupport#getAnnotation()
 	 */
@@ -39,7 +39,7 @@ class ReactiveCassandraRepositoriesRegistrar extends RepositoryBeanDefinitionReg
 		return EnableReactiveCassandraRepositories.class;
 	}
 
-	/* 
+	/*
 	 * (non-Javadoc)
 	 * @see org.springframework.data.repository.config.RepositoryBeanDefinitionRegistrarSupport#getExtension()
 	 */

spring-data-cassandra/src/main/java/org/springframework/data/cassandra/repository/config/ReactiveCassandraRepositoryConfigurationExtension.java

@@ -36,7 +36,7 @@ import org.springframework.util.StringUtils;
 
 /**
  * {@link RepositoryConfigurationExtension} for Cassandra.
- * 
+ *
  * @author Mark Paluch
  * @since 2.0
  */
@@ -69,14 +69,14 @@ public class ReactiveCassandraRepositoryConfigurationExtension extends Repositor
 		return ReactiveCassandraRepositoryFactoryBean.class.getName();
 	}
 
-	/* 
+	/*
 	 * (non-Javadoc)
 	 * @see org.springframework.data.repository.config.RepositoryConfigurationExtensionSupport#postProcess(org.springframework.beans.factory.support.BeanDefinitionBuilder, org.springframework.data.repository.config.XmlRepositoryConfigurationSource)
 	 */
 	@Override
 	public void postProcess(BeanDefinitionBuilder builder, XmlRepositoryConfigurationSource config) {}
 
-	/* 
+	/*
 	 * (non-Javadoc)
 	 * @see org.springframework.data.repository.config.RepositoryConfigurationExtensionSupport#postProcess(org.springframework.beans.factory.support.BeanDefinitionBuilder, org.springframework.data.repository.config.AnnotationRepositoryConfigurationSource)
 	 */
@@ -97,7 +97,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,10 +106,10 @@ public class ReactiveCassandraRepositoryConfigurationExtension extends Repositor
 	 */
 	@Override
 	protected Collection<Class<?>> getIdentifyingTypes() {
-		return Collections.<Class<?>>singleton(ReactiveCassandraRepository.class);
+		return Collections.<Class<?>> singleton(ReactiveCassandraRepository.class);
 	}
 
-	/* 
+	/*
 	 * (non-Javadoc)
 	 * @see org.springframework.data.repository.config.RepositoryConfigurationExtensionSupport#getRepositoryConfigurations(T, org.springframework.core.io.ResourceLoader, boolean)
 	 */
@@ -120,11 +120,8 @@ public class ReactiveCassandraRepositoryConfigurationExtension extends Repositor
 		Collection<RepositoryConfiguration<T>> repositoryConfigurations = super.getRepositoryConfigurations(configSource,
 				loader, strictMatchesOnly);
 
-		return repositoryConfigurations.stream().filter(configuration -> {
-
-			Class<?> repositoryInterface = super.loadRepositoryInterface(configuration, loader);
-			return RepositoryType.isReactiveRepository(repositoryInterface);
-
-		}).collect(Collectors.toList());
+		return repositoryConfigurations.stream()
+				.filter(configuration -> RepositoryType.isReactiveRepository(loadRepositoryInterface(configuration, loader)))
+				.collect(Collectors.toList());
 	}
 }

spring-data-cassandra/src/main/java/org/springframework/data/cassandra/repository/config/RepositoryType.java

@@ -16,14 +16,17 @@
 package org.springframework.data.cassandra.repository.config;
 
 import java.lang.reflect.Method;
+import java.util.ArrayList;
+import java.util.List;
 
-import org.springframework.data.repository.query.ReactiveWrappers;
+import org.springframework.data.repository.util.ReactiveWrappers;
+import org.springframework.util.ReflectionUtils;
 
 import lombok.experimental.UtilityClass;
 
 /**
  * Utility class to discover whether a repository interface uses reactive wrapper types.
- * 
+ *
  * @author Mark Paluch
  * @since 2.0
  */
@@ -33,7 +36,7 @@ class RepositoryType {
 	/**
 	 * Check whether {@code repositoryInterface} uses reactive wrapper types as return type or parameter types in its
 	 * methods.
-	 * 
+	 *
 	 * @param repositoryInterface must not be {@literal null}.
 	 * @return {@literal true} if the {@code repositoryInterface} uses reactive wrapper types.
 	 * @see ReactiveWrappers
@@ -45,16 +48,9 @@ class RepositoryType {
 			return false;
 		}
 
-		Method[] methods = repositoryInterface.getMethods();
-
-		for (Method method : methods) {
-
-			if (usesReactiveWrappers(method)) {
-				return true;
-			}
-		}
-
-		return false;
+		List<Method> reactiveMethods = new ArrayList<>();
+		ReflectionUtils.doWithMethods(repositoryInterface, reactiveMethods::add, RepositoryType::usesReactiveWrappers);
+		return !reactiveMethods.isEmpty();
 	}
 
 	private static boolean usesReactiveWrappers(Method method) {

spring-data-cassandra/src/main/java/org/springframework/data/cassandra/repository/query/AbstractReactiveCassandraQuery.java

@@ -16,8 +16,6 @@
 package org.springframework.data.cassandra.repository.query;
 
 import org.reactivestreams.Publisher;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
 import org.springframework.core.convert.converter.Converter;
 import org.springframework.data.cassandra.core.CassandraOperations;
 import org.springframework.data.cassandra.core.ReactiveCassandraOperations;
@@ -26,7 +24,6 @@ import org.springframework.data.cassandra.repository.query.ReactiveCassandraQuer
 import org.springframework.data.cassandra.repository.query.ReactiveCassandraQueryExecution.ResultProcessingExecution;
 import org.springframework.data.cassandra.repository.query.ReactiveCassandraQueryExecution.SingleEntityExecution;
 import org.springframework.data.repository.query.ParameterAccessor;
-import org.springframework.data.repository.query.ReactiveWrapperConverters;
 import org.springframework.data.repository.query.RepositoryQuery;
 import org.springframework.data.repository.query.ResultProcessor;
 import org.springframework.util.Assert;
@@ -42,9 +39,7 @@ import reactor.core.publisher.Mono;
  */
 public abstract class AbstractReactiveCassandraQuery implements RepositoryQuery {
 
-	protected static Logger log = LoggerFactory.getLogger(AbstractReactiveCassandraQuery.class);
-
-	private final CassandraQueryMethod method;
+	private final ReactiveCassandraQueryMethod method;
 	private final ReactiveCassandraOperations operations;
 
 	/**
@@ -54,9 +49,9 @@ public abstract class AbstractReactiveCassandraQuery implements RepositoryQuery
 	 * @param method must not be {@literal null}.
 	 * @param operations must not be {@literal null}.
 	 */
-	public AbstractReactiveCassandraQuery(CassandraQueryMethod method, ReactiveCassandraOperations operations) {
+	public AbstractReactiveCassandraQuery(ReactiveCassandraQueryMethod method, ReactiveCassandraOperations operations) {
 
-		Assert.notNull(method, "CassandraQueryMethod must not be null");
+		Assert.notNull(method, "ReactiveCassandraQueryMethod must not be null");
 		Assert.notNull(operations, "ReactiveCassandraOperations must not be null");
 
 		this.method = method;
@@ -77,11 +72,8 @@ public abstract class AbstractReactiveCassandraQuery implements RepositoryQuery
 	@Override
 	public Object execute(Object[] parameters) {
 
-		if (hasReactiveWrapperParameter()) {
-			return executeDeferred(parameters);
-		}
-
-		return execute(new ReactiveCassandraParameterAccessor(method, parameters));
+		return method.hasReactiveWrapperParameter() ? executeDeferred(parameters)
+				: execute(new ReactiveCassandraParameterAccessor(method, parameters));
 	}
 
 	@SuppressWarnings("unchecked")
@@ -139,16 +131,6 @@ public abstract class AbstractReactiveCassandraQuery implements RepositoryQuery
 		}
 	}
 
-	private boolean hasReactiveWrapperParameter() {
-
-		for (CassandraParameters.CassandraParameter cassandraParameter : method.getParameters()) {
-			if (ReactiveWrapperConverters.supports(cassandraParameter.getType())) {
-				return true;
-			}
-		}
-		return false;
-	}
-
 	/**
 	 * Creates a string query using the given {@link ParameterAccessor}
 	 *

spring-data-cassandra/src/main/java/org/springframework/data/cassandra/repository/query/CassandraParameters.java

@@ -24,8 +24,8 @@ import org.springframework.data.cassandra.mapping.CassandraType;
 import org.springframework.data.cassandra.repository.query.CassandraParameters.CassandraParameter;
 import org.springframework.data.repository.query.Parameter;
 import org.springframework.data.repository.query.Parameters;
-import org.springframework.data.repository.query.ReactiveWrappers;
 import org.springframework.data.repository.util.QueryExecutionConverters;
+import org.springframework.data.repository.util.ReactiveWrappers;
 import org.springframework.util.Assert;
 
 /**

spring-data-cassandra/src/main/java/org/springframework/data/cassandra/repository/query/ReactiveCassandraParameterAccessor.java

@@ -15,7 +15,11 @@
  */
 package org.springframework.data.cassandra.repository.query;
 
-import org.springframework.data.repository.query.ReactiveWrapperConverters;
+import java.util.ArrayList;
+import java.util.List;
+
+import org.springframework.data.repository.util.ReactiveWrapperConverters;
+import org.springframework.data.repository.util.ReactiveWrappers;
 
 import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
@@ -26,35 +30,31 @@ import reactor.core.publisher.MonoProcessor;
  * to reactive parameter wrapper types upon creation. This class performs synchronization when acessing parameters.
  *
  * @author Mark Paluch
+ * @since 2.0
  */
 class ReactiveCassandraParameterAccessor extends CassandraParametersParameterAccessor {
 
 	private final Object[] values;
-	private final MonoProcessor<?>[] subscriptions;
+	private final List<MonoProcessor<?>> subscriptions;
 
 	public ReactiveCassandraParameterAccessor(CassandraQueryMethod method, Object[] values) {
 
 		super(method, values);
 
 		this.values = values;
-		this.subscriptions = new MonoProcessor<?>[values.length];
+		this.subscriptions = new ArrayList<>(values.length);
 
-		for (int i = 0; i < values.length; i++) {
+		for (Object value : values) {
 
-			Object value = values[i];
-
-			if (value == null) {
+			if (value == null || !ReactiveWrappers.supports(value.getClass())) {
+				subscriptions.add(null);
 				continue;
 			}
 
-			if (!ReactiveWrapperConverters.supports(value.getClass())) {
-				continue;
-			}
-
-			if (ReactiveWrapperConverters.isSingleLike(value.getClass())) {
-				subscriptions[i] = ReactiveWrapperConverters.toWrapper(value, Mono.class).subscribe();
+			if (ReactiveWrappers.isSingleValueType(value.getClass())) {
+				subscriptions.add(ReactiveWrapperConverters.toWrapper(value, Mono.class).subscribe());
 			} else {
-				subscriptions[i] = ReactiveWrapperConverters.toWrapper(value, Flux.class).collectList().subscribe();
+				subscriptions.add(ReactiveWrapperConverters.toWrapper(value, Flux.class).collectList().subscribe());
 			}
 		}
 	}
@@ -66,8 +66,8 @@ class ReactiveCassandraParameterAccessor extends CassandraParametersParameterAcc
 	@Override
 	protected <T> T getValue(int index) {
 
-		if (subscriptions[index] != null) {
-			return (T) subscriptions[index].block();
+		if (subscriptions.get(index) != null) {
+			return (T) subscriptions.get(index).block();
 		}
 
 		return super.getValue(index);

spring-data-cassandra/src/main/java/org/springframework/data/cassandra/repository/query/ReactiveCassandraQueryMethod.java

@@ -15,13 +15,14 @@
  */
 package org.springframework.data.cassandra.repository.query;
 
-import static org.springframework.data.repository.query.ReactiveWrappers.*;
-
 import java.lang.reflect.Method;
 
 import org.springframework.data.cassandra.mapping.CassandraMappingContext;
+import org.springframework.data.cassandra.repository.query.CassandraParameters.CassandraParameter;
 import org.springframework.data.projection.ProjectionFactory;
 import org.springframework.data.repository.core.RepositoryMetadata;
+import org.springframework.data.repository.util.ReactiveWrapperConverters;
+import org.springframework.data.repository.util.ReactiveWrappers;
 
 /**
  * Reactive specific implementation of {@link CassandraQueryMethod}.
@@ -55,7 +56,7 @@ public class ReactiveCassandraQueryMethod extends CassandraQueryMethod {
 	 */
 	@Override
 	public boolean isCollectionQuery() {
-		return !(isPageQuery() || isSliceQuery()) && isMultiType(method.getReturnType());
+		return !(isPageQuery() || isSliceQuery()) && ReactiveWrappers.isMultiValueType(method.getReturnType());
 	}
 
 	/*
@@ -67,4 +68,20 @@ public class ReactiveCassandraQueryMethod extends CassandraQueryMethod {
 	public boolean isStreamQuery() {
 		return true;
 	}
+
+	/**
+	 * Check if the given {@link org.springframework.data.repository.query.QueryMethod} receives a reactive parameter
+	 * wrapper as one of its parameters.
+	 *
+	 * @return
+	 */
+	public boolean hasReactiveWrapperParameter() {
+
+		for (CassandraParameter cassandraParameter : getParameters()) {
+			if (ReactiveWrapperConverters.supports(cassandraParameter.getType())) {
+				return true;
+			}
+		}
+		return false;
+	}
 }

spring-data-cassandra/src/main/java/org/springframework/data/cassandra/repository/query/ReactivePartTreeCassandraQuery.java

@@ -17,7 +17,6 @@ package org.springframework.data.cassandra.repository.query;
 
 import org.springframework.data.cassandra.core.ReactiveCassandraOperations;
 import org.springframework.data.cassandra.mapping.CassandraMappingContext;
-import org.springframework.data.repository.query.QueryMethod;
 import org.springframework.data.repository.query.RepositoryQuery;
 import org.springframework.data.repository.query.parser.PartTree;
 
@@ -34,13 +33,14 @@ public class ReactivePartTreeCassandraQuery extends AbstractReactiveCassandraQue
 	private final PartTree tree;
 
 	/**
-	 * Creates a new {@link ReactivePartTreeCassandraQuery} from the given {@link QueryMethod} and
+	 * Creates a new {@link ReactivePartTreeCassandraQuery} from the given {@link ReactiveCassandraQueryMethod} and
 	 * {@link ReactiveCassandraOperations}.
 	 *
 	 * @param queryMethod must not be {@literal null}.
 	 * @param operations must not be {@literal null}.
 	 */
-	public ReactivePartTreeCassandraQuery(CassandraQueryMethod queryMethod, ReactiveCassandraOperations operations) {
+	public ReactivePartTreeCassandraQuery(ReactiveCassandraQueryMethod queryMethod,
+			ReactiveCassandraOperations operations) {
 
 		super(queryMethod, operations);
 

spring-data-cassandra/src/main/java/org/springframework/data/cassandra/repository/query/ReactiveStringBasedCassandraQuery.java

@@ -50,14 +50,15 @@ public class ReactiveStringBasedCassandraQuery extends AbstractReactiveCassandra
 	 * Creates a new {@link ReactiveStringBasedCassandraQuery} for the given {@link CassandraQueryMethod},
 	 * {@link ReactiveCassandraOperations}, {@link SpelExpressionParser}, and {@link EvaluationContextProvider}.
 	 *
-	 * @param queryMethod {@link CassandraQueryMethod} on which this query is based.
+	 * @param queryMethod {@link ReactiveCassandraQueryMethod} on which this query is based.
 	 * @param operations {@link ReactiveCassandraOperations} used to perform data access in Cassandra.
 	 * @param expressionParser {@link SpelExpressionParser} used to parse expressions in the query.
 	 * @param evaluationContextProvider {@link EvaluationContextProvider} used to access the potentially shared
 	 *          {@link org.springframework.expression.spel.support.StandardEvaluationContext}.
 	 */
-	public ReactiveStringBasedCassandraQuery(CassandraQueryMethod queryMethod, ReactiveCassandraOperations operations,
-			SpelExpressionParser expressionParser, EvaluationContextProvider evaluationContextProvider) {
+	public ReactiveStringBasedCassandraQuery(ReactiveCassandraQueryMethod queryMethod,
+			ReactiveCassandraOperations operations, SpelExpressionParser expressionParser,
+			EvaluationContextProvider evaluationContextProvider) {
 		this(queryMethod.getAnnotatedQuery(), queryMethod, operations, expressionParser, evaluationContextProvider);
 	}
 
@@ -65,13 +66,13 @@ public class ReactiveStringBasedCassandraQuery extends AbstractReactiveCassandra
 	 * Creates a new {@link ReactiveStringBasedCassandraQuery} for the given {@code query}, {@link CassandraQueryMethod},
 	 * {@link ReactiveCassandraOperations}, {@link SpelExpressionParser}, and {@link EvaluationContextProvider}.
 	 *
-	 * @param queryMethod {@link CassandraQueryMethod} on which this query is based.
+	 * @param queryMethod {@link ReactiveCassandraQueryMethod} on which this query is based.
 	 * @param operations {@link ReactiveCassandraOperations} used to perform data access in Cassandra.
 	 * @param expressionParser {@link SpelExpressionParser} used to parse expressions in the query.
 	 * @param evaluationContextProvider {@link EvaluationContextProvider} used to access the potentially shared
 	 *          {@link org.springframework.expression.spel.support.StandardEvaluationContext}.
 	 */
-	public ReactiveStringBasedCassandraQuery(String query, CassandraQueryMethod queryMethod,
+	public ReactiveStringBasedCassandraQuery(String query, ReactiveCassandraQueryMethod queryMethod,
 			ReactiveCassandraOperations operations, SpelExpressionParser expressionParser,
 			EvaluationContextProvider evaluationContextProvider) {
 

spring-data-cassandra/src/main/java/org/springframework/data/cassandra/repository/support/ReactiveCassandraRepositoryFactory.java

@@ -17,14 +17,16 @@ package org.springframework.data.cassandra.repository.support;
 
 import java.io.Serializable;
 import java.lang.reflect.Method;
+import java.util.Arrays;
 
+import org.reactivestreams.Publisher;
 import org.springframework.core.convert.ConversionService;
 import org.springframework.core.convert.support.DefaultConversionService;
+import org.springframework.dao.InvalidDataAccessApiUsageException;
 import org.springframework.data.cassandra.core.ReactiveCassandraOperations;
 import org.springframework.data.cassandra.mapping.CassandraMappingContext;
 import org.springframework.data.cassandra.mapping.CassandraPersistentEntity;
 import org.springframework.data.cassandra.repository.query.CassandraEntityInformation;
-import org.springframework.data.cassandra.repository.query.CassandraQueryMethod;
 import org.springframework.data.cassandra.repository.query.ReactiveCassandraQueryMethod;
 import org.springframework.data.cassandra.repository.query.ReactivePartTreeCassandraQuery;
 import org.springframework.data.cassandra.repository.query.ReactiveStringBasedCassandraQuery;
@@ -39,8 +41,11 @@ 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;
 import org.springframework.util.Assert;
+import org.springframework.util.ClassUtils;
 
 /**
  * Factory to create {@link org.springframework.data.cassandra.repository.ReactiveCassandraRepository} instances.
@@ -69,7 +74,7 @@ public class ReactiveCassandraRepositoryFactory extends RepositoryFactorySupport
 		this.mappingContext = cassandraOperations.getConverter().getMappingContext();
 
 		DefaultConversionService conversionService = new DefaultConversionService();
-		QueryExecutionConverters.registerConvertersIn(conversionService);
+		ReactiveWrapperConverters.registerConvertersIn(conversionService);
 
 		this.conversionService = conversionService;
 		setConversionService(conversionService);
@@ -105,6 +110,53 @@ public class ReactiveCassandraRepositoryFactory extends RepositoryFactorySupport
 		return new CassandraQueryLookupStrategy(operations, evaluationContextProvider, mappingContext, conversionService);
 	}
 
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.repository.core.support.RepositoryFactorySupport#validate(org.springframework.data.repository.core.RepositoryMetadata)
+	 */
+	@Override
+	protected void validate(RepositoryMetadata repositoryMetadata) {
+
+		if (!ReactiveWrappers.isAvailable()) {
+			throw new InvalidDataAccessApiUsageException(
+					String.format("Cannot implement Repository %s without reactive library support.",
+							repositoryMetadata.getRepositoryInterface().getName()));
+		}
+
+		Arrays.stream(repositoryMetadata.getRepositoryInterface().getMethods())
+				.forEach(ReactiveCassandraRepositoryFactory::validate);
+	}
+
+	/**
+	 * Reactive Cassandra support requires reactive wrapper support. If return type/parameters are reactive wrapper types,
+	 * then it's required to be able to convert these into Publisher.
+	 *
+	 * @param method the method to validate.
+	 */
+	private static void validate(Method method) {
+
+		if (ReactiveWrappers.supports(method.getReturnType())
+				&& !ClassUtils.isAssignable(Publisher.class, method.getReturnType())) {
+
+			if (!ReactiveWrapperConverters.supports(method.getReturnType())) {
+
+				throw new InvalidDataAccessApiUsageException(
+						String.format("No reactive type converter found for type %s used in %s, method %s.",
+								method.getReturnType().getName(), method.getDeclaringClass().getName(), method));
+			}
+		}
+
+		Arrays.stream(method.getParameterTypes()) //
+				.filter(ReactiveWrappers::supports) //
+				.filter(parameterType -> !ClassUtils.isAssignable(Publisher.class, parameterType)) //
+				.filter(parameterType -> !ReactiveWrapperConverters.supports(parameterType)) //
+				.forEach(parameterType -> {
+					throw new InvalidDataAccessApiUsageException(
+							String.format("No reactive type converter found for type %s used in %s, method %s.",
+									parameterType.getName(), method.getDeclaringClass().getName(), method));
+				});
+	}
+
 	/*
 	 * (non-Javadoc)
 	 * @see org.springframework.data.repository.core.support.RepositoryFactorySupport#getEntityInformation(java.lang.Class)
@@ -141,7 +193,8 @@ public class ReactiveCassandraRepositoryFactory extends RepositoryFactorySupport
 		private final CassandraMappingContext mappingContext;
 		private final ConversionService conversionService;
 
-		CassandraQueryLookupStrategy(ReactiveCassandraOperations operations, EvaluationContextProvider evaluationContextProvider, CassandraMappingContext mappingContext,
+		CassandraQueryLookupStrategy(ReactiveCassandraOperations operations,
+				EvaluationContextProvider evaluationContextProvider, CassandraMappingContext mappingContext,
 				ConversionService conversionService) {
 
 			this.evaluationContextProvider = evaluationContextProvider;
@@ -158,7 +211,8 @@ public class ReactiveCassandraRepositoryFactory extends RepositoryFactorySupport
 		public RepositoryQuery resolveQuery(Method method, RepositoryMetadata metadata, ProjectionFactory factory,
 				NamedQueries namedQueries) {
 
-			CassandraQueryMethod queryMethod = new ReactiveCassandraQueryMethod(method, metadata, factory, mappingContext);
+			ReactiveCassandraQueryMethod queryMethod = new ReactiveCassandraQueryMethod(method, metadata, factory,
+					mappingContext);
 			String namedQueryName = queryMethod.getNamedQueryName();
 
 			if (namedQueries.hasQuery(namedQueryName)) {

spring-data-cassandra/src/main/java/org/springframework/data/cassandra/repository/support/ReactiveCassandraRepositoryFactoryBean.java

@@ -29,7 +29,7 @@ import org.springframework.util.Assert;
 /**
  * {@link org.springframework.beans.factory.FactoryBean} to create
  * {@link org.springframework.data.cassandra.repository.ReactiveCassandraRepository} instances.
- * 
+ *
  * @author Mark Paluch
  * @since 2.0
  * @see org.springframework.data.repository.reactive.ReactivePagingAndSortingRepository
@@ -64,7 +64,7 @@ public class ReactiveCassandraRepositoryFactoryBean<T extends Repository<S, ID>,
 
 	/*
 	 * (non-Javadoc)
-	 * 
+	 *
 	 * @see
 	 * org.springframework.data.repository.support.RepositoryFactoryBeanSupport
 	 * #createRepositoryFactory()
@@ -76,7 +76,7 @@ public class ReactiveCassandraRepositoryFactoryBean<T extends Repository<S, ID>,
 
 	/**
 	 * Creates and initializes a {@link RepositoryFactorySupport} instance.
-	 * 
+	 *
 	 * @param operations
 	 * @return
 	 */

spring-data-cassandra/src/main/java/org/springframework/data/cassandra/repository/support/SimpleReactiveCassandraRepository.java

@@ -31,7 +31,7 @@ import reactor.core.publisher.Mono;
 
 /**
  * Reactive repository base implementation for Cassandra.
- * 
+ *
  * @author Mark Paluch
  * @since 2.0
  */

spring-data-cassandra/src/test/java/org/springframework/data/cassandra/core/ReactiveCassandraTemplateIntegrationTests.java

@@ -31,7 +31,7 @@ import reactor.core.scheduler.Schedulers;
 
 /**
  * Integration tests for {@link ReactiveCassandraTemplate}.
- * 
+ *
  * @author Mark Paluch
  */
 public class ReactiveCassandraTemplateIntegrationTests extends AbstractKeyspaceCreatingIntegrationTest {

spring-data-cassandra/src/test/java/org/springframework/data/cassandra/repository/ConvertingReactiveCassandraRepositoryTests.java

@@ -32,7 +32,7 @@ import org.springframework.data.cassandra.domain.Person;
 import org.springframework.data.cassandra.repository.config.EnableReactiveCassandraRepositories;
 import org.springframework.data.cassandra.test.integration.support.IntegrationTestConfig;
 import org.springframework.data.repository.reactive.ReactiveCrudRepository;
-import org.springframework.data.repository.reactive.RxJavaCrudRepository;
+import org.springframework.data.repository.reactive.RxJava1CrudRepository;
 import org.springframework.stereotype.Repository;
 import org.springframework.test.context.ContextConfiguration;
 import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
@@ -71,7 +71,7 @@ public class ConvertingReactiveCassandraRepositoryTests extends AbstractKeyspace
 	@Autowired ReactiveCassandraTemplate template;
 	@Autowired MixedPersonRepostitory reactiveRepository;
 	@Autowired PersonRepostitory reactivePersonRepostitory;
-	@Autowired RxJavaPersonRepostitory rxJavaPersonRepostitory;
+	@Autowired RxJava1PersonRepostitory rxJava1PersonRepostitory;
 
 	Person dave, oliver, carter, boyd;
 
@@ -130,7 +130,7 @@ public class ConvertingReactiveCassandraRepositoryTests extends AbstractKeyspace
 	public void simpleRxJavaMethodsShouldWork() {
 
 		rx.observers.TestSubscriber<Boolean> subscriber = new rx.observers.TestSubscriber<>();
-		rxJavaPersonRepostitory.exists(dave.getId()).subscribe(subscriber);
+		rxJava1PersonRepostitory.exists(dave.getId()).subscribe(subscriber);
 
 		subscriber.awaitTerminalEvent();
 		subscriber.assertCompleted();
@@ -145,7 +145,7 @@ public class ConvertingReactiveCassandraRepositoryTests extends AbstractKeyspace
 	public void existsWithSingleRxJavaIdMethodsShouldWork() {
 
 		rx.observers.TestSubscriber<Boolean> subscriber = new rx.observers.TestSubscriber<>();
-		rxJavaPersonRepostitory.exists(Single.just(dave.getId())).subscribe(subscriber);
+		rxJava1PersonRepostitory.exists(Single.just(dave.getId())).subscribe(subscriber);
 
 		subscriber.awaitTerminalEvent();
 		subscriber.assertCompleted();
@@ -160,7 +160,7 @@ public class ConvertingReactiveCassandraRepositoryTests extends AbstractKeyspace
 	public void singleRxJavaQueryMethodShouldWork() {
 
 		rx.observers.TestSubscriber<Person> subscriber = new rx.observers.TestSubscriber<>();
-		rxJavaPersonRepostitory.findManyByLastname(dave.getLastname()).subscribe(subscriber);
+		rxJava1PersonRepostitory.findManyByLastname(dave.getLastname()).subscribe(subscriber);
 
 		subscriber.awaitTerminalEvent();
 		subscriber.assertNoErrors();
@@ -175,7 +175,7 @@ public class ConvertingReactiveCassandraRepositoryTests extends AbstractKeyspace
 	public void singleProjectedRxJavaQueryMethodShouldWork() {
 
 		rx.observers.TestSubscriber<ProjectedPerson> subscriber = new rx.observers.TestSubscriber<>();
-		rxJavaPersonRepostitory.findProjectedByLastname(carter.getLastname()).subscribe(subscriber);
+		rxJava1PersonRepostitory.findProjectedByLastname(carter.getLastname()).subscribe(subscriber);
 
 		subscriber.awaitTerminalEvent();
 		subscriber.assertCompleted();
@@ -192,7 +192,7 @@ public class ConvertingReactiveCassandraRepositoryTests extends AbstractKeyspace
 	public void observableRxJavaQueryMethodShouldWork() {
 
 		rx.observers.TestSubscriber<Person> subscriber = new rx.observers.TestSubscriber<>();
-		rxJavaPersonRepostitory.findByLastname(boyd.getLastname()).subscribe(subscriber);
+		rxJava1PersonRepostitory.findByLastname(boyd.getLastname()).subscribe(subscriber);
 
 		subscriber.awaitTerminalEvent();
 		subscriber.assertCompleted();
@@ -229,7 +229,7 @@ public class ConvertingReactiveCassandraRepositoryTests extends AbstractKeyspace
 	}
 
 	@Repository
-	interface RxJavaPersonRepostitory extends RxJavaCrudRepository<Person, String> {
+	interface RxJava1PersonRepostitory extends RxJava1CrudRepository<Person, String> {
 
 		Observable<Person> findManyByLastname(String lastname);
 

spring-data-cassandra/src/test/java/org/springframework/data/cassandra/repository/config/ReactiveCassandraRepositoryConfigurationExtensionUnitTests.java

@@ -33,7 +33,7 @@ import org.springframework.data.repository.config.AnnotationRepositoryConfigurat
 import org.springframework.data.repository.config.RepositoryConfiguration;
 import org.springframework.data.repository.config.RepositoryConfigurationSource;
 import org.springframework.data.repository.reactive.ReactiveCrudRepository;
-import org.springframework.data.repository.reactive.RxJavaCrudRepository;
+import org.springframework.data.repository.reactive.RxJava1CrudRepository;
 
 /**
  * Unit tests for {@link ReactiveCassandraRepositoryConfigurationExtension}.
@@ -114,7 +114,7 @@ public class ReactiveCassandraRepositoryConfigurationExtensionUnitTests {
 	@Table
 	static class Sample {}
 
-	interface SampleRepository extends RxJavaCrudRepository<Sample, Long> {}
+	interface SampleRepository extends RxJava1CrudRepository<Sample, Long> {}
 
 	interface UnannotatedRepository extends ReactiveCrudRepository<Object, Long> {}
 

spring-data-cassandra/src/test/java/org/springframework/data/cassandra/repository/query/ReactivePartTreeCassandraQueryUnitTests.java

@@ -125,8 +125,8 @@ public class ReactivePartTreeCassandraQueryUnitTests {
 		try {
 			Method method = Repo.class.getMethod(methodName, paramTypes);
 			ProjectionFactory factory = new SpelAwareProxyProjectionFactory();
-			CassandraQueryMethod queryMethod = new CassandraQueryMethod(method, new DefaultRepositoryMetadata(Repo.class),
-					factory, mappingContext);
+			ReactiveCassandraQueryMethod queryMethod = new ReactiveCassandraQueryMethod(method,
+					new DefaultRepositoryMetadata(Repo.class), factory, mappingContext);
 
 			return new ReactivePartTreeCassandraQuery(queryMethod, mockCassandraOperations);
 		} catch (NoSuchMethodException e) {

spring-data-cassandra/src/test/java/org/springframework/data/cassandra/repository/query/ReactiveStringBasedCassandraQueryUnitTests.java

@@ -111,7 +111,7 @@ public class ReactiveStringBasedCassandraQueryUnitTests {
 	private ReactiveStringBasedCassandraQuery getQueryMethod(String name, Class<?>... args) {
 
 		Method method = ReflectionUtils.findMethod(SampleRepository.class, name, args);
-		CassandraQueryMethod queryMethod = new CassandraQueryMethod(method, metadata, factory,
+		ReactiveCassandraQueryMethod queryMethod = new ReactiveCassandraQueryMethod(method, metadata, factory,
 				converter.getMappingContext());
 		return new ReactiveStringBasedCassandraQuery(queryMethod, operations, PARSER,
 				new ExtensionAwareEvaluationContextProvider());