diff --git a/spring-cql/src/main/java/org/springframework/cassandra/core/DefaultBridgedReactiveSession.java b/spring-cql/src/main/java/org/springframework/cassandra/core/DefaultBridgedReactiveSession.java index cd6a70130..472972260 100644 --- a/spring-cql/src/main/java/org/springframework/cassandra/core/DefaultBridgedReactiveSession.java +++ b/spring-cql/src/main/java/org/springframework/cassandra/core/DefaultBridgedReactiveSession.java @@ -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}. */ diff --git a/spring-cql/src/main/java/org/springframework/cassandra/core/DefaultReactiveSessionFactory.java b/spring-cql/src/main/java/org/springframework/cassandra/core/DefaultReactiveSessionFactory.java index 7cfc9b62d..19275fc0b 100644 --- a/spring-cql/src/main/java/org/springframework/cassandra/core/DefaultReactiveSessionFactory.java +++ b/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}. *

* This implementation returns always the same {@link ReactiveSession}. - * + * * @author Mark Paluch * @since 2.0 */ diff --git a/spring-cql/src/main/java/org/springframework/cassandra/core/ReactiveCqlOperations.java b/spring-cql/src/main/java/org/springframework/cassandra/core/ReactiveCqlOperations.java index 1ad4b9cba..98beec224 100644 --- a/spring-cql/src/main/java/org/springframework/cassandra/core/ReactiveCqlOperations.java +++ b/spring-cql/src/main/java/org/springframework/cassandra/core/ReactiveCqlOperations.java @@ -52,7 +52,7 @@ public interface ReactiveCqlOperations { * {@link DataAccessException} hierarchy. *

* 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 { *

* 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 { *

* 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 { *

* 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 { *

* 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 { *

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

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

* 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 { *

* 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 { *

* 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 { *

* 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). diff --git a/spring-cql/src/main/java/org/springframework/cassandra/core/ReactiveCqlTemplate.java b/spring-cql/src/main/java/org/springframework/cassandra/core/ReactiveCqlTemplate.java index 5c92b5ac9..b3182180b 100644 --- a/spring-cql/src/main/java/org/springframework/cassandra/core/ReactiveCqlTemplate.java +++ b/spring-cql/src/main/java/org/springframework/cassandra/core/ReactiveCqlTemplate.java @@ -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 diff --git a/spring-cql/src/main/java/org/springframework/cassandra/core/ReactivePreparedStatementCallback.java b/spring-cql/src/main/java/org/springframework/cassandra/core/ReactivePreparedStatementCallback.java index 799367dfb..66598c836 100644 --- a/spring-cql/src/main/java/org/springframework/cassandra/core/ReactivePreparedStatementCallback.java +++ b/spring-cql/src/main/java/org/springframework/cassandra/core/ReactivePreparedStatementCallback.java @@ -48,7 +48,7 @@ public interface ReactivePreparedStatementCallback { * 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. diff --git a/spring-cql/src/main/java/org/springframework/cassandra/core/ReactiveResultSetExtractor.java b/spring-cql/src/main/java/org/springframework/cassandra/core/ReactiveResultSetExtractor.java index 3882d3852..1a701405c 100644 --- a/spring-cql/src/main/java/org/springframework/cassandra/core/ReactiveResultSetExtractor.java +++ b/spring-cql/src/main/java/org/springframework/cassandra/core/ReactiveResultSetExtractor.java @@ -43,7 +43,7 @@ public interface ReactiveResultSetExtractor { /** * 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, diff --git a/spring-cql/src/main/java/org/springframework/cassandra/core/ReactiveSession.java b/spring-cql/src/main/java/org/springframework/cassandra/core/ReactiveSession.java index 27c0861cd..74f23f5af 100644 --- a/spring-cql/src/main/java/org/springframework/cassandra/core/ReactiveSession.java +++ b/spring-cql/src/main/java/org/springframework/cassandra/core/ReactiveSession.java @@ -144,14 +144,14 @@ public interface ReactiveSession extends Closeable { *

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

 	 * RegularStatement toPrepare = new SimpleStatement("SELECT * FROM test WHERE k=?")
 	 * 		.setConsistencyLevel(ConsistencyLevel.QUORUM);
 	 * PreparedStatement prepared = session.prepare(toPrepare);
 	 * session.execute(prepared.bind("someValue"));
 	 *
- * + * * the final execution will be performed with Quorum consistency. *

* Please note that if the same CQL statement is prepared more than once, all calls to this method will return the diff --git a/spring-cql/src/main/java/org/springframework/cassandra/core/ReactiveSessionCallback.java b/spring-cql/src/main/java/org/springframework/cassandra/core/ReactiveSessionCallback.java index 7457613ff..74a3d4e19 100644 --- a/spring-cql/src/main/java/org/springframework/cassandra/core/ReactiveSessionCallback.java +++ b/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 { /** - * 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}. *

* 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 diff --git a/spring-cql/src/main/java/org/springframework/cassandra/core/ReactiveStatementCallback.java b/spring-cql/src/main/java/org/springframework/cassandra/core/ReactiveStatementCallback.java index 8fac20054..1de249b00 100644 --- a/spring-cql/src/main/java/org/springframework/cassandra/core/ReactiveStatementCallback.java +++ b/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 { /** - * 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}. *

* 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 diff --git a/spring-data-cassandra/pom.xml b/spring-data-cassandra/pom.xml index 94a4f95fe..043986dad 100644 --- a/spring-data-cassandra/pom.xml +++ b/spring-data-cassandra/pom.xml @@ -86,6 +86,13 @@ true + + io.reactivex + rxjava-reactive-streams + ${rxjava-reactive-streams} + true + + javax.enterprise diff --git a/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/config/java/AbstractReactiveCassandraConfiguration.java b/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/config/java/AbstractReactiveCassandraConfiguration.java index e57aa7b22..d573f637f 100644 --- a/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/config/java/AbstractReactiveCassandraConfiguration.java +++ b/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()); } } diff --git a/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/ReactiveCassandraTemplate.java b/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/ReactiveCassandraTemplate.java index 0d55d3f1c..029bfef51 100644 --- a/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/ReactiveCassandraTemplate.java +++ b/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/ReactiveCassandraTemplate.java @@ -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) session -> (Publisher) session.execute(insertQuery) - .flatMap(reactiveResultSet -> reactiveResultSet.wasApplied() ? Mono.just(entity) : Mono.empty())).next(); + class InsertCallback implements ReactiveSessionCallback, CqlProvider { + + @Override + public Publisher 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) session -> (Publisher) session.execute(update) - .flatMap(reactiveResultSet -> reactiveResultSet.wasApplied() ? Mono.just(entity) : Mono.empty())).next(); + class UpdateCallback implements ReactiveSessionCallback, CqlProvider { + + @Override + public Publisher 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) session -> (Publisher) session.execute(delete) - .flatMap(reactiveResultSet -> reactiveResultSet.wasApplied() ? Mono.just(entity) : Mono.empty())).next(); + class DeleteCallback implements ReactiveSessionCallback, CqlProvider { + + @Override + public Publisher 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() */ diff --git a/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/repository/config/CassandraRepositoryConfigurationExtension.java b/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/repository/config/CassandraRepositoryConfigurationExtension.java index 813154878..b8db98170 100644 --- a/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/repository/config/CassandraRepositoryConfigurationExtension.java +++ b/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/repository/config/CassandraRepositoryConfigurationExtension.java @@ -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; diff --git a/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/repository/config/EnableReactiveCassandraRepositories.java b/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/repository/config/EnableReactiveCassandraRepositories.java index 3568f67fd..07b3492e8 100644 --- a/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/repository/config/EnableReactiveCassandraRepositories.java +++ b/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"; diff --git a/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/repository/config/ReactiveCassandraRepositoriesRegistrar.java b/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/repository/config/ReactiveCassandraRepositoriesRegistrar.java index dda1195db..e4467a3e5 100644 --- a/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/repository/config/ReactiveCassandraRepositoriesRegistrar.java +++ b/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() */ diff --git a/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/repository/config/ReactiveCassandraRepositoryConfigurationExtension.java b/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/repository/config/ReactiveCassandraRepositoryConfigurationExtension.java index fb192beb3..ff1d7cc42 100644 --- a/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/repository/config/ReactiveCassandraRepositoryConfigurationExtension.java +++ b/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/repository/config/ReactiveCassandraRepositoryConfigurationExtension.java @@ -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> getIdentifyingAnnotations() { - return Collections.>singleton(Table.class); + return Collections.> singleton(Table.class); } /* @@ -106,10 +106,10 @@ public class ReactiveCassandraRepositoryConfigurationExtension extends Repositor */ @Override protected Collection> getIdentifyingTypes() { - return Collections.>singleton(ReactiveCassandraRepository.class); + return Collections.> 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> 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()); } } diff --git a/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/repository/config/RepositoryType.java b/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/repository/config/RepositoryType.java index 3a23e2d0b..530c8844c 100644 --- a/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/repository/config/RepositoryType.java +++ b/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 reactiveMethods = new ArrayList<>(); + ReflectionUtils.doWithMethods(repositoryInterface, reactiveMethods::add, RepositoryType::usesReactiveWrappers); + return !reactiveMethods.isEmpty(); } private static boolean usesReactiveWrappers(Method method) { diff --git a/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/repository/query/AbstractReactiveCassandraQuery.java b/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/repository/query/AbstractReactiveCassandraQuery.java index 7465cb200..fc69e5c63 100644 --- a/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/repository/query/AbstractReactiveCassandraQuery.java +++ b/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/repository/query/AbstractReactiveCassandraQuery.java @@ -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} * diff --git a/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/repository/query/CassandraParameters.java b/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/repository/query/CassandraParameters.java index e6f6b2017..20683cf80 100644 --- a/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/repository/query/CassandraParameters.java +++ b/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; /** diff --git a/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/repository/query/ReactiveCassandraParameterAccessor.java b/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/repository/query/ReactiveCassandraParameterAccessor.java index 2ae914aee..1f17267d7 100644 --- a/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/repository/query/ReactiveCassandraParameterAccessor.java +++ b/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/repository/query/ReactiveCassandraParameterAccessor.java @@ -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> 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 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); diff --git a/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/repository/query/ReactiveCassandraQueryMethod.java b/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/repository/query/ReactiveCassandraQueryMethod.java index 7d9fe9af6..68f9ec6d2 100644 --- a/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/repository/query/ReactiveCassandraQueryMethod.java +++ b/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/repository/query/ReactiveCassandraQueryMethod.java @@ -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; + } } diff --git a/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/repository/query/ReactivePartTreeCassandraQuery.java b/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/repository/query/ReactivePartTreeCassandraQuery.java index 42685089f..d9e6fa5d5 100644 --- a/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/repository/query/ReactivePartTreeCassandraQuery.java +++ b/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); diff --git a/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/repository/query/ReactiveStringBasedCassandraQuery.java b/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/repository/query/ReactiveStringBasedCassandraQuery.java index eacfc78e6..d3c539716 100644 --- a/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/repository/query/ReactiveStringBasedCassandraQuery.java +++ b/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/repository/query/ReactiveStringBasedCassandraQuery.java @@ -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) { diff --git a/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/repository/support/ReactiveCassandraRepositoryFactory.java b/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/repository/support/ReactiveCassandraRepositoryFactory.java index c45fd6e11..8139167f5 100644 --- a/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/repository/support/ReactiveCassandraRepositoryFactory.java +++ b/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/repository/support/ReactiveCassandraRepositoryFactory.java @@ -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)) { diff --git a/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/repository/support/ReactiveCassandraRepositoryFactoryBean.java b/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/repository/support/ReactiveCassandraRepositoryFactoryBean.java index ee0e93f44..a364f8be8 100644 --- a/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/repository/support/ReactiveCassandraRepositoryFactoryBean.java +++ b/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/repository/support/ReactiveCassandraRepositoryFactoryBean.java @@ -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, /* * (non-Javadoc) - * + * * @see * org.springframework.data.repository.support.RepositoryFactoryBeanSupport * #createRepositoryFactory() @@ -76,7 +76,7 @@ public class ReactiveCassandraRepositoryFactoryBean, /** * Creates and initializes a {@link RepositoryFactorySupport} instance. - * + * * @param operations * @return */ diff --git a/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/repository/support/SimpleReactiveCassandraRepository.java b/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/repository/support/SimpleReactiveCassandraRepository.java index 9e24988a3..6afb16205 100644 --- a/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/repository/support/SimpleReactiveCassandraRepository.java +++ b/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 */ diff --git a/spring-data-cassandra/src/test/java/org/springframework/data/cassandra/core/ReactiveCassandraTemplateIntegrationTests.java b/spring-data-cassandra/src/test/java/org/springframework/data/cassandra/core/ReactiveCassandraTemplateIntegrationTests.java index ecdedd86b..5b2d10780 100644 --- a/spring-data-cassandra/src/test/java/org/springframework/data/cassandra/core/ReactiveCassandraTemplateIntegrationTests.java +++ b/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 { diff --git a/spring-data-cassandra/src/test/java/org/springframework/data/cassandra/repository/ConvertingReactiveCassandraRepositoryTests.java b/spring-data-cassandra/src/test/java/org/springframework/data/cassandra/repository/ConvertingReactiveCassandraRepositoryTests.java index 0de3b8953..cdfff9865 100644 --- a/spring-data-cassandra/src/test/java/org/springframework/data/cassandra/repository/ConvertingReactiveCassandraRepositoryTests.java +++ b/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 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 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 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 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 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 { + interface RxJava1PersonRepostitory extends RxJava1CrudRepository { Observable findManyByLastname(String lastname); diff --git a/spring-data-cassandra/src/test/java/org/springframework/data/cassandra/repository/config/ReactiveCassandraRepositoryConfigurationExtensionUnitTests.java b/spring-data-cassandra/src/test/java/org/springframework/data/cassandra/repository/config/ReactiveCassandraRepositoryConfigurationExtensionUnitTests.java index a027bf061..c340af888 100644 --- a/spring-data-cassandra/src/test/java/org/springframework/data/cassandra/repository/config/ReactiveCassandraRepositoryConfigurationExtensionUnitTests.java +++ b/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 {} + interface SampleRepository extends RxJava1CrudRepository {} interface UnannotatedRepository extends ReactiveCrudRepository {} diff --git a/spring-data-cassandra/src/test/java/org/springframework/data/cassandra/repository/query/ReactivePartTreeCassandraQueryUnitTests.java b/spring-data-cassandra/src/test/java/org/springframework/data/cassandra/repository/query/ReactivePartTreeCassandraQueryUnitTests.java index 02e9d3060..b908bea67 100644 --- a/spring-data-cassandra/src/test/java/org/springframework/data/cassandra/repository/query/ReactivePartTreeCassandraQueryUnitTests.java +++ b/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) { diff --git a/spring-data-cassandra/src/test/java/org/springframework/data/cassandra/repository/query/ReactiveStringBasedCassandraQueryUnitTests.java b/spring-data-cassandra/src/test/java/org/springframework/data/cassandra/repository/query/ReactiveStringBasedCassandraQueryUnitTests.java index a670d0499..4a7315568 100644 --- a/spring-data-cassandra/src/test/java/org/springframework/data/cassandra/repository/query/ReactiveStringBasedCassandraQueryUnitTests.java +++ b/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());