Spring/spring-data-cassandra
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

DATACASS-485 - Polish reactive, fluent Cassandra API.

Browse Source
This commit is contained in:
John Blum
2018-02-01 22:40:27 -08:00
parent 1197805d1b
commit e3efe92c7f
13 changed files with 781 additions and 720 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -15,13 +15,14 @@
*/
package org.springframework.data.cassandra.core;
import java.util.function.Function;
import lombok.NonNull;
import lombok.Value;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import java.util.function.Function;
import org.reactivestreams.Publisher;
import org.springframework.dao.DataAccessException;
import org.springframework.data.cassandra.ReactiveResultSet;
@@ -126,15 +127,7 @@ public class ReactiveCassandraTemplate implements ReactiveCassandraOperations {
* @see com.datastax.driver.core.Session
*/
public ReactiveCassandraTemplate(ReactiveSessionFactory sessionFactory, CassandraConverter converter) {
Assert.notNull(sessionFactory, "ReactiveSessionFactory must not be null");
Assert.notNull(converter, "CassandraConverter must not be null");
this.converter = converter;
this.cqlOperations = new ReactiveCqlTemplate(sessionFactory);
this.mappingContext = this.converter.getMappingContext();
this.statementFactory = new StatementFactory(new QueryMapper(converter), new UpdateMapper(converter));
this.projectionFactory = new SpelAwareProxyProjectionFactory();
this(new ReactiveCqlTemplate(sessionFactory), converter);
}
/**
@@ -293,8 +286,9 @@ public class ReactiveCassandraTemplate implements ReactiveCassandraOperations {
Function<Row, T> mapper = getMapper(entityClass, returnType);
RegularStatement select = getStatementFactory().select(query,
getMappingContext().getRequiredPersistentEntity(entityClass), tableName);
RegularStatement select = getStatementFactory()
.select(query, getRequiredPersistentEntity(entityClass), tableName);
return getReactiveCqlOperations().query(select, (row, rowNum) -> mapper.apply(row));
}
@@ -307,8 +301,7 @@ public class ReactiveCassandraTemplate implements ReactiveCassandraOperations {
Assert.notNull(query, "Query must not be null");
Assert.notNull(entityClass, "Entity type must not be null");
return selectOne(getStatementFactory().select(query, getMappingContext().getRequiredPersistentEntity(entityClass)),
entityClass);
return selectOne(getStatementFactory().select(query, getRequiredPersistentEntity(entityClass)), entityClass);
}
/* (non-Javadoc)
@@ -328,8 +321,9 @@ public class ReactiveCassandraTemplate implements ReactiveCassandraOperations {
Mono<WriteResult> doUpdate(Query query, org.springframework.data.cassandra.core.query.Update update,
Class<?> entityClass, CqlIdentifier tableName) {
RegularStatement statement = getStatementFactory().update(query, update,
getMappingContext().getRequiredPersistentEntity(entityClass), tableName);
RegularStatement statement = getStatementFactory()
.update(query, update, getRequiredPersistentEntity(entityClass), tableName);
return getReactiveCqlOperations().execute(new StatementCallback(statement)).next();
}
@@ -349,8 +343,7 @@ public class ReactiveCassandraTemplate implements ReactiveCassandraOperations {
RegularStatement delete = getStatementFactory().delete(query, getRequiredPersistentEntity(entityClass), tableName);
return getReactiveCqlOperations()
.execute(new StatementCallback(delete)).next();
return getReactiveCqlOperations().execute(new StatementCallback(delete)).next();
}
// -------------------------------------------------------------------------
@@ -379,14 +372,13 @@ public class ReactiveCassandraTemplate implements ReactiveCassandraOperations {
Assert.notNull(query, "Query must not be null");
Assert.notNull(entityClass, "Entity type must not be null");
RegularStatement count = getStatementFactory().count(query, getRequiredPersistentEntity(entityClass));
return doCount(query, entityClass, getTableName(entityClass));
}
Mono<Long> doCount(Query query, Class<?> entityClass, CqlIdentifier tableName) {
RegularStatement count = getStatementFactory().count(query, getRequiredPersistentEntity(entityClass), tableName);
RegularStatement count =
getStatementFactory().count(query, getRequiredPersistentEntity(entityClass), tableName);
return getReactiveCqlOperations().queryForObject(count, Long.class).switchIfEmpty(Mono.just(0L));
}
@@ -423,8 +415,8 @@ public class ReactiveCassandraTemplate implements ReactiveCassandraOperations {
Mono<Boolean> doExists(Query query, Class<?> entityClass, CqlIdentifier tableName) {
RegularStatement select = getStatementFactory().select(query.limit(1), getRequiredPersistentEntity(entityClass),
tableName);
RegularStatement select =
getStatementFactory().select(query.limit(1), getRequiredPersistentEntity(entityClass), tableName);
return getReactiveCqlOperations().queryForRows(select).hasElements();
}
@@ -464,8 +456,7 @@ public class ReactiveCassandraTemplate implements ReactiveCassandraOperations {
Assert.notNull(entity, "Entity must not be null");
Assert.notNull(options, "InsertOptions must not be null");
CqlIdentifier tableName = getTableName(entity);
return doInsert(entity, options, tableName);
return doInsert(entity, options, getTableName(entity));
}
Mono<WriteResult> doInsert(Object entity, WriteOptions options, CqlIdentifier tableName) {
@@ -594,16 +585,21 @@ public class ReactiveCassandraTemplate implements ReactiveCassandraOperations {
@SuppressWarnings("unchecked")
private <T> Function<Row, T> getMapper(Class<?> entityType, Class<T> targetType) {
Class<?> typeToRead = targetType.isInterface() || targetType.isAssignableFrom(entityType) ? entityType : targetType;
Class<?> typeToRead = resolveTypeToRead(entityType, targetType);
return row -> {
Object source = getConverter().read(typeToRead, row);
return (T) (targetType.isInterface() ? projectionFactory.createProjection(targetType, source) : source);
return (T) (targetType.isInterface()
? this.projectionFactory.createProjection(targetType, source) : source);
};
}
private Class<?> resolveTypeToRead(Class<?> entityType, Class<?> targetType) {
return targetType.isInterface() || targetType.isAssignableFrom(entityType) ? entityType : targetType;
}
@Value
static class StatementCallback implements ReactiveSessionCallback<WriteResult>, CqlProvider {
@@ -614,7 +610,7 @@ public class ReactiveCassandraTemplate implements ReactiveCassandraOperations {
*/
@Override
public Publisher<WriteResult> doInSession(ReactiveSession session) throws DriverException, DataAccessException {
return session.execute(statement).flatMap(StatementCallback::toWriteResult);
return session.execute(this.statement).flatMap(StatementCallback::toWriteResult);
}
/* (non-Javadoc)
@@ -622,7 +618,7 @@ public class ReactiveCassandraTemplate implements ReactiveCassandraOperations {
*/
@Override
public String getCql() {
return statement.toString();
return this.statement.toString();
}
private static Mono<WriteResult> toWriteResult(ReactiveResultSet resultSet) {

92
spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/ReactiveDeleteOperation.java
View File

@@ -19,15 +19,16 @@ import reactor.core.publisher.Mono;
import org.springframework.data.cassandra.core.cql.CqlIdentifier;
import org.springframework.data.cassandra.core.query.Query;
import org.springframework.util.Assert;
/**
* {@link ReactiveDeleteOperation} allows creation and execution of Cassandra {@code DELETE} operations in a fluent API
* style.
* The {@link ReactiveDeleteOperation} interface allows creation and execution of Cassandra {@code DELETE} operations
* in a fluent API style.
* <p>
* The starting {@literal domainType} is used for mapping the {@link Query} provided via {@code matching} into the
* Cassandra specific representation. The table to operate on is by default derived from the initial
* {@literal domainType} and can be defined there via {@link org.springframework.data.cassandra.core.mapping.Table}.
* Using {@code inTable} allows to override the table name for the execution.
* The starting {@literal domainType} is used for mapping the {@link Query} provided via {@code matching}
* into the Cassandra specific representation. By default, the table to operate on is derived from the initial
* {@literal domainType} and can be defined there via {@link org.springframework.data.cassandra.core.mapping.Table}
* annotation. Using {@code inTable} allows a developer to override the table name for the execution.
*
* <pre>
* <code>
@@ -39,16 +40,19 @@ import org.springframework.data.cassandra.core.query.Query;
* </pre>
*
* @author Mark Paluch
* @author John Blum
* @see org.springframework.data.cassandra.core.query.Query
* @since 2.1
*/
public interface ReactiveDeleteOperation {
/**
* Start creating a {@code DELETE} operation for the given {@literal domainType}.
* Begin creating a {@code DELETE} operation for the given {@link Class domainType}.
*
* @param domainType must not be {@literal null}.
* @param domainType {@link Class type} of domain object to delete; must not be {@literal null}.
* @return new instance of {@link ReactiveDelete}.
* @throws IllegalArgumentException if domainType is {@literal null}.
* @throws IllegalArgumentException if {@link Class domainType} is {@literal null}.
* @see ReactiveDelete
*/
ReactiveDelete delete(Class<?> domainType);
@@ -58,52 +62,74 @@ public interface ReactiveDeleteOperation {
interface DeleteWithTable {
/**
* Explicitly set the name of the table to perform the query on.
* Explicitly set the {@link String name} of the table on which to perform the delete.
* <p>
* Skip this step to use the default table derived from the domain type.
* Skip this step to use the default table derived from the {@link Class domain type}.
*
* @param table must not be {@literal null} or empty.
* @return new instance of {@link DeleteWithTable}.
* @throws IllegalArgumentException if {@code table} is {@literal null} or empty.
* @param table {@link String name} of the table; must not be {@literal null} or empty.
* @return new instance of {@link DeleteWithQuery}.
* @throws IllegalArgumentException if {@link String table} is {@literal null} or empty.
* @see #inTable(CqlIdentifier)
* @see DeleteWithQuery
*/
DeleteWithQuery inTable(String table);
default DeleteWithQuery inTable(String table) {
Assert.hasText(table, "Table name must not be null or empty");
return inTable(CqlIdentifier.of(table));
}
/**
* Explicitly set the name of the table to perform the query on.
* Explicitly set the {@link CqlIdentifier name} of the table on which to perform the delete.
* <p>
* Skip this step to use the default table derived from the domain type.
* Skip this step to use the default table derived from the {@link Class domain type}.
*
* @param table must not be {@literal null}.
* @return new instance of {@link DeleteWithTable}.
* @throws IllegalArgumentException if {@link CqlIdentifier} is {@literal null}.
* @param table {@link CqlIdentifier name} of the table; must not be {@literal null}.
* @return new instance of {@link DeleteWithQuery}.
* @throws IllegalArgumentException if {@link CqlIdentifier table} is {@literal null}.
* @see org.springframework.data.cassandra.core.cql.CqlIdentifier
* @see DeleteWithQuery
*/
DeleteWithQuery inTable(CqlIdentifier table);
}
/**
* Required {@link Query filter}.
*/
interface DeleteWithQuery {
/**
* Define the {@link Query} used to filter elements in the delete.
*
* @param query {@link Query} used as the filter in the delete; must not be {@literal null}.
* @return new instance of {@link TerminatingDelete}.
* @throws IllegalArgumentException if {@link Query} is {@literal null}.
* @see TerminatingDelete
*/
TerminatingDelete matching(Query query);
}
/**
* Trigger {@code DELETE} operation by calling one of the terminating methods.
*/
interface TerminatingDelete {
/**
* Remove all matching rows.
*
* @return the {@link WriteResult}. Never {@literal null}.
* @return the {@link WriteResult}; never {@literal null}.
* @see org.springframework.data.cassandra.core.WriteResult
* @see reactor.core.publisher.Mono
*/
Mono<WriteResult> all();
}
interface DeleteWithQuery {
/**
* Define the query filtering elements.
*
* @param query must not be {@literal null}.
* @return new instance of {@link TerminatingDelete}.
* @throws IllegalArgumentException if query is {@literal null}.
*/
TerminatingDelete matching(Query query);
}
/**
* {@link ReactiveDelete} provides methods for constructing {@code DELETE} operations in a fluent way.
* The {@link ReactiveDelete} interface provides methods for constructing {@code DELETE} operations
* in a fluent way.
*/
interface ReactiveDelete extends DeleteWithTable, DeleteWithQuery {}
}

30
spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/ReactiveDeleteOperationSupport.java
View File

@@ -19,6 +19,7 @@ import lombok.AccessLevel;
import lombok.NonNull;
import lombok.RequiredArgsConstructor;
import lombok.experimental.FieldDefaults;
import reactor.core.publisher.Mono;
import org.springframework.data.cassandra.core.cql.CqlIdentifier;
@@ -30,6 +31,8 @@ import org.springframework.util.Assert;
* Implementation of {@link ReactiveDeleteOperation}.
*
* @author Mark Paluch
* @see org.springframework.data.cassandra.core.ReactiveDeleteOperation
* @see org.springframework.data.cassandra.core.query.Query
* @since 2.1
*/
@RequiredArgsConstructor
@@ -43,14 +46,14 @@ class ReactiveDeleteOperationSupport implements ReactiveDeleteOperation {
@Override
public ReactiveDelete delete(Class<?> domainType) {
Assert.notNull(domainType, "DomainType must not be null!");
Assert.notNull(domainType, "DomainType must not be null");
return new ReactiveDeleteSupport(template, domainType, Query.empty(), null);
return new ReactiveDeleteSupport(this.template, domainType, Query.empty(), null);
}
@RequiredArgsConstructor
@FieldDefaults(level = AccessLevel.PRIVATE, makeFinal = true)
static class ReactiveDeleteSupport implements ReactiveDelete, DeleteWithTable, TerminatingDelete {
static class ReactiveDeleteSupport implements ReactiveDelete, TerminatingDelete {
@NonNull ReactiveCassandraTemplate template;
@@ -60,17 +63,6 @@ class ReactiveDeleteOperationSupport implements ReactiveDeleteOperation {
@Nullable CqlIdentifier tableName;
/* (non-Javadoc)
* @see org.springframework.data.cassandra.core.ReactiveDeleteOperation.DeleteWithTable#inTable(java.lang.String)
*/
@Override
public DeleteWithQuery inTable(String tableName) {
Assert.hasText(tableName, "Table name must not be null or empty");
return new ReactiveDeleteSupport(template, domainType, query, CqlIdentifier.of(tableName));
}
/* (non-Javadoc)
* @see org.springframework.data.cassandra.core.ReactiveDeleteOperation.DeleteWithTable#inTable(org.springframework.data.cassandra.core.cql.CqlIdentifier)
*/
@@ -79,7 +71,7 @@ class ReactiveDeleteOperationSupport implements ReactiveDeleteOperation {
Assert.notNull(tableName, "Table name must not be null");
return new ReactiveDeleteSupport(template, domainType, query, tableName);
return new ReactiveDeleteSupport(this.template, this.domainType, this.query, tableName);
}
/* (non-Javadoc)
@@ -88,20 +80,20 @@ class ReactiveDeleteOperationSupport implements ReactiveDeleteOperation {
@Override
public TerminatingDelete matching(Query query) {
Assert.notNull(query, "Query must not be null!");
Assert.notNull(query, "Query must not be null");
return new ReactiveDeleteSupport(template, domainType, query, tableName);
return new ReactiveDeleteSupport(this.template, this.domainType, query, this.tableName);
}
/* (non-Javadoc)
* @see org.springframework.data.cassandra.core.ReactiveDeleteOperation.TerminatingDelete#all()
*/
public Mono<WriteResult> all() {
return template.doDelete(query, domainType, getTableName());
return this.template.doDelete(this.query, this.domainType, getTableName());
}
private CqlIdentifier getTableName() {
return tableName != null ? tableName : template.getTableName(domainType);
return this.tableName != null ? this.tableName : this.template.getTableName(this.domainType);
}
}
}

98
spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/ReactiveInsertOperation.java
View File

@@ -18,14 +18,15 @@ package org.springframework.data.cassandra.core;
import reactor.core.publisher.Mono;
import org.springframework.data.cassandra.core.cql.CqlIdentifier;
import org.springframework.util.Assert;
/**
* {@link ReactiveInsertOperation} allows creation and execution of Cassandra {@code INSERT} insert operations in a
* fluent API style.
* The {@link ReactiveInsertOperation} interface allows creation and execution of Cassandra {@code INSERT} operations
* in a fluent API style.
* <p>
* The table to operate on is by default derived from the initial {@literal domainType} and can be defined there via
* {@link org.springframework.data.cassandra.core.mapping.Table}. Using {@code inTable} allows to override the
* collection name for the execution.
* By default,the table to operate on is derived from the initial {@link Class domainType} and can be defined
* there via {@link org.springframework.data.cassandra.core.mapping.Table} annotation. Using {@code inTable}
* allows a developer to override the table name for the execution.
*
* <pre>
* <code>
@@ -36,59 +37,58 @@ import org.springframework.data.cassandra.core.cql.CqlIdentifier;
* </pre>
*
* @author Mark Paluch
* @author John Blum
* @since 2.1
*/
public interface ReactiveInsertOperation {
/**
* Start creating an {@code INSERT} operation for given {@literal domainType}.
* Begin creating an {@code INSERT} operation for given {@link Class domainType}.
*
* @param domainType must not be {@literal null}.
* @param <T> {@link Class type} of the application domain object.
* @param domainType {@link Class type} of the domain object to insert; must not be {@literal null}.
* @return new instance of {@link ReactiveInsert}.
* @throws IllegalArgumentException if domainType is {@literal null}.
* @throws IllegalArgumentException if {@link Class domainType} is {@literal null}.
* @see ReactiveInsert
*/
<T> ReactiveInsert<T> insert(Class<T> domainType);
/**
* Trigger insert execution by calling one of the terminating methods.
*/
interface TerminatingInsert<T> {
/**
* Insert exactly one object.
*
* @param object must not be {@literal null}.
* @throws IllegalArgumentException if object is {@literal null}.
*/
Mono<WriteResult> one(T object);
}
/**
* Collection override (optional).
* Table override (optional).
*/
interface InsertWithTable<T> extends InsertWithOptions<T> {
/**
* Explicitly set the name of the table.
* Explicitly set the {@link String name} of the table.
* <p>
* Skip this step to use the default table derived from the domain type.
* Skip this step to use the default table derived from the {@link Class domain type}.
*
* @param table must not be {@literal null} or empty.
* @return new instance of {@link TerminatingInsert}.
* @throws IllegalArgumentException if {@code table} is {@literal null} or empty.
* @param table {@link String name} of the table; must not be {@literal null} or empty.
* @return new instance of {@link InsertWithOptions}.
* @throws IllegalArgumentException if {@link String table} is {@literal null} or empty.
* @see #inTable(CqlIdentifier)
* @see InsertWithOptions
*/
InsertWithOptions<T> inTable(String table);
default InsertWithOptions<T> inTable(String table) {
Assert.hasText(table, "Table must not be null or empty");
return inTable(CqlIdentifier.of(table));
}
/**
* Explicitly set the name of the table.
* Explicitly set the {@link String name} of the table.
* <p>
* Skip this step to use the default table derived from the domain type.
* Skip this step to use the default table derived from the {@link Class domain type}.
*
* @param table must not be {@literal null}.
* @return new instance of {@link TerminatingInsert}.
* @throws IllegalArgumentException if {@link CqlIdentifier} is {@literal null}.
* @param table {@link String name} of the table; must not be {@literal null}.
* @return new instance of {@link InsertWithOptions}.
* @throws IllegalArgumentException if {@link CqlIdentifier table} is {@literal null}.
* @see org.springframework.data.cassandra.core.cql.CqlIdentifier
* @see InsertWithOptions
*/
InsertWithOptions<T> inTable(CqlIdentifier table);
}
/**
@@ -97,17 +97,39 @@ public interface ReactiveInsertOperation {
interface InsertWithOptions<T> extends TerminatingInsert<T> {
/**
* Set insert options.
* Set {@link InsertOptions}.
*
* @param insertOptions insertOptions not be {@literal null}.
* @param insertOptions {@link InsertOptions options} to use on insert; must not be {@literal null}.
* @return new instance of {@link TerminatingInsert}.
* @throws IllegalArgumentException if {@link InsertOptions} is {@literal null}.
* @see org.springframework.data.cassandra.core.InsertOptions
* @see TerminatingInsert
*/
TerminatingInsert<T> withOptions(InsertOptions insertOptions);
}
/**
* {@link ReactiveInsert} provides methods for constructing {@code INSERT} operations in a fluent way.
* Trigger {@code INSERT} execution by calling one of the terminating methods.
*/
interface ReactiveInsert<T> extends TerminatingInsert<T>, InsertWithTable<T>, InsertWithOptions<T> {}
interface TerminatingInsert<T> {
/**
* Insert exactly one {@link Object}.
*
* @param object {@link Object} to insert; must not be {@literal null}.
* @throws IllegalArgumentException if {@link Object} is {@literal null}.
* @see org.springframework.data.cassandra.core.WriteResult
* @see reactor.core.publisher.Mono
*/
Mono<WriteResult> one(T object);
}
/**
* The {@link ReactiveInsert} interface provides methods for constructing {@code INSERT} operations
* in a fluent way.
*/
interface ReactiveInsert<T> extends InsertWithTable<T> {}
}

28
spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/ReactiveInsertOperationSupport.java
View File

@@ -19,6 +19,7 @@ import lombok.AccessLevel;
import lombok.NonNull;
import lombok.RequiredArgsConstructor;
import lombok.experimental.FieldDefaults;
import reactor.core.publisher.Mono;
import org.springframework.data.cassandra.core.cql.CqlIdentifier;
@@ -42,9 +43,9 @@ class ReactiveInsertOperationSupport implements ReactiveInsertOperation {
@Override
public <T> ReactiveInsert<T> insert(Class<T> domainType) {
Assert.notNull(domainType, "DomainType must not be null!");
Assert.notNull(domainType, "DomainType must not be null");
return new ReactiveInsertSupport<>(template, domainType, null, InsertOptions.empty());
return new ReactiveInsertSupport<>(this.template, domainType, InsertOptions.empty(), null);
}
@RequiredArgsConstructor
@@ -55,20 +56,9 @@ class ReactiveInsertOperationSupport implements ReactiveInsertOperation {
@NonNull Class<T> domainType;
@Nullable CqlIdentifier tableName;
@NonNull InsertOptions insertOptions;
/* (non-Javadoc)
* @see org.springframework.data.cassandra.core.ReactiveInsertOperation.InsertWithTable#inTable(java.lang.String)
*/
@Override
public InsertWithOptions<T> inTable(String tableName) {
Assert.hasText(tableName, "Table name must not be null or empty");
return new ReactiveInsertSupport<>(template, domainType, CqlIdentifier.of(tableName), insertOptions);
}
@Nullable CqlIdentifier tableName;
/* (non-Javadoc)
* @see org.springframework.data.cassandra.core.ReactiveInsertOperation.InsertWithTable#inTable(org.springframework.data.cassandra.core.cql.CqlIdentifier)
@@ -78,7 +68,7 @@ class ReactiveInsertOperationSupport implements ReactiveInsertOperation {
Assert.notNull(tableName, "Table name must not be null");
return new ReactiveInsertSupport<>(template, domainType, tableName, insertOptions);
return new ReactiveInsertSupport<>(this.template, this.domainType, this.insertOptions, tableName);
}
/* (non-Javadoc)
@@ -89,7 +79,7 @@ class ReactiveInsertOperationSupport implements ReactiveInsertOperation {
Assert.notNull(insertOptions, "InsertOptions must not be null");
return new ReactiveInsertSupport<>(template, domainType, tableName, insertOptions);
return new ReactiveInsertSupport<>(this.template, this.domainType, insertOptions, this.tableName);
}
/* (non-Javadoc)
@@ -98,13 +88,13 @@ class ReactiveInsertOperationSupport implements ReactiveInsertOperation {
@Override
public Mono<WriteResult> one(T object) {
Assert.notNull(object, "Object must not be null!");
Assert.notNull(object, "Object must not be null");
return template.doInsert(object, insertOptions, getTableName());
return this.template.doInsert(object, this.insertOptions, getTableName());
}
private CqlIdentifier getTableName() {
return tableName != null ? tableName : template.getTableName(domainType);
return this.tableName != null ? this.tableName : this.template.getTableName(this.domainType);
}
}
}

218
spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/ReactiveSelectOperation.java
View File

@@ -20,19 +20,20 @@ import reactor.core.publisher.Mono;
import org.springframework.data.cassandra.core.cql.CqlIdentifier;
import org.springframework.data.cassandra.core.query.Query;
import org.springframework.util.Assert;
/**
* {@link ReactiveSelectOperation} allows creation and execution of Cassandra {@code SELECT} operations in a fluent API
* style.
* The {@link ReactiveSelectOperation} interface allows creation and execution of Cassandra {@code SELECT} operations
* in a fluent API style.
* <p>
* The starting {@literal domainType} is used for mapping the {@link Query} provided via {@code matching} into the
* The starting {@literal domainType} is used for mapping the {@link Query} provided via {@code matching} int the
* Cassandra specific representation. By default, the originating {@literal domainType} is also used for mapping back
* the result from the {@link com.datastax.driver.core.Row}. However, it is possible to define an different
* {@literal returnType} via {@code as} to mapping the result.
* <p>
* The table to operate on is by default derived from the initial {@literal domainType} and can be defined there via
* {@link org.springframework.data.cassandra.core.mapping.Table}. Using {@code inTable} allows to override the table
* name for the execution.
* By default, the table to operate on is derived from the initial {@literal domainType} and can be defined there via
* the {@link org.springframework.data.cassandra.core.mapping.Table} annotation. Using {@code inTable} allows
* a developer to override the table name for the execution.
*
* <pre>
* <code>
@@ -45,123 +46,152 @@ import org.springframework.data.cassandra.core.query.Query;
* </pre>
*
* @author Mark Paluch
* @author John Blum
* @see org.springframework.data.cassandra.core.query.Query
* @since 2.1
*/
public interface ReactiveSelectOperation {
/**
* Start creating a {@code SELECT} operation for the given {@literal domainType}.
* Begin creating a {@code SELECT} operation for the given {@link Class domainType}.
*
* @param domainType must not be {@literal null}.
* @param <T> {@link Class type} of the application domain object.
* @param domainType {@link Class type} of the domain object to query; must not be {@literal null}.
* @return new instance of {@link ReactiveSelect}.
* @throws IllegalArgumentException if domainType is {@literal null}.
* @throws IllegalArgumentException if {@link Class domainType} is {@literal null}.
* @see ReactiveSelect
*/
<T> ReactiveSelect<T> query(Class<T> domainType);
/**
* Table override (optional).
*/
interface SelectWithTable<T> extends SelectWithQuery<T> {
/**
* Explicitly set the {@link String name} of the table on which to perform the query.
* <p>
* Skip this step to use the default table derived from the {@link Class domain type}.
*
* @param table {@link String name} of the table; must not be {@literal null} or empty.
* @return new instance of {@link SelectWithProjection}.
* @throws IllegalArgumentException if {@link String table} is {@literal null} or empty.
* @see #inTable(CqlIdentifier)
* @see SelectWithProjection
*/
default SelectWithProjection<T> inTable(String table) {
Assert.hasText(table, "Table name must not be null or empty");
return inTable(CqlIdentifier.of(table));
}
/**
* Explicitly set the {@link CqlIdentifier name} of the table on which to perform the query.
* <p>
* Skip this step to use the default table derived from the {@link Class domain type}.
*
* @param table {@link CqlIdentifier name} of the table; must not be {@literal null}.
* @return new instance of {@link SelectWithProjection}.
* @throws IllegalArgumentException if {@link CqlIdentifier table} is {@literal null}.
* @see org.springframework.data.cassandra.core.cql.CqlIdentifier
* @see SelectWithProjection
*/
SelectWithProjection<T> inTable(CqlIdentifier table);
}
/**
* Result type override (optional).
*/
interface SelectWithProjection<T> extends SelectWithQuery<T> {
/**
* Define the {@link Class result target type} that the fields should be mapped to.
* <p>
* Skip this step if you are only interested in the original {@link Class domain type}.
*
* @param <R> {@link Class type} of the result.
* @param resultType desired {@link Class type} of the result; must not be {@literal null}.
* @return new instance of {@link SelectWithQuery}.
* @throws IllegalArgumentException if {@link Class resultType} is {@literal null}.
* @see SelectWithQuery
*/
<R> SelectWithQuery<R> as(Class<R> resultType);
}
/**
* Define a {@link Query} used as the filter for the {@code SELECT}.
*/
interface SelectWithQuery<T> extends TerminatingSelect<T> {
/**
* Set the {@link Query} used as a filter in the {@code SELECT} statement.
*
* @param query {@link Query} used as a filter; must not be {@literal null}.
* @return new instance of {@link TerminatingSelect}.
* @throws IllegalArgumentException if {@link Query} is {@literal null}.
* @see org.springframework.data.cassandra.core.query.Query
* @see TerminatingSelect
*/
TerminatingSelect<T> matching(Query query);
}
/**
* Trigger {@code SELECT} execution by calling one of the terminating methods.
*/
interface TerminatingSelect<T> {
/**
* Get exactly zero or one result.
*
* @return {@link Mono#empty()} if no match found. Never {@literal null}.
* @throws org.springframework.dao.IncorrectResultSizeDataAccessException if more than one match found.
*/
Mono<T> one();
/**
* Get the first or no result.
*
* @return {@link Mono#empty()} if no match found. Never {@literal null}.
*/
Mono<T> first();
/**
* Get all matching elements.
*
* @return never {@literal null}.
*/
Flux<T> all();
/**
* Get the number of matching elements.
*
* @return {@link Mono} emitting total number of matching elements. Never {@literal null}.
* @return a {@link Mono} emitting the total number of matching elements; never {@literal null}.
* @see reactor.core.publisher.Mono
*/
Mono<Long> count();
/**
* Check for the presence of matching elements.
*
* @return {@link Mono} emitting {@literal true} if at least one matching element exists. Never {@literal null}.
* @return a {@link Mono} emitting {@literal true} if at least one matching element exists;
* never {@literal null}.
* @see reactor.core.publisher.Mono
*/
Mono<Boolean> exists();
/**
* Get the first result or no result.
*
* @return the first result or {@link Mono#empty()} if no match found; never {@literal null}.
* @see reactor.core.publisher.Mono
*/
Mono<T> first();
/**
* Get exactly zero or one result.
*
* @return exactly one result or {@link Mono#empty()} if no match found; never {@literal null}.
* @throws org.springframework.dao.IncorrectResultSizeDataAccessException if more than one match found.
* @see reactor.core.publisher.Mono
*/
Mono<T> one();
/**
* Get all matching elements.
*
* @return all matching elements; never {@literal null}.
* @see reactor.core.publisher.Flux
*/
Flux<T> all();
}
/**
* Terminating operations invoking the actual query execution.
*/
interface SelectWithQuery<T> extends TerminatingSelect<T> {
/**
* Set the filter query to be used.
*
* @param query must not be {@literal null}.
* @return new instance of {@link TerminatingSelect}.
* @throws IllegalArgumentException if query is {@literal null}.
*/
TerminatingSelect<T> matching(Query query);
}
/**
* Table override (Optional).
*/
interface SelectWithTable<T> extends SelectWithQuery<T> {
/**
* Explicitly set the name of the table to perform the query on.
* <p>
* Skip this step to use the default table derived from the domain type.
*
* @param table must not be {@literal null} or empty.
* @return new instance of {@link SelectWithProjection}.
* @throws IllegalArgumentException if {@code table} is {@literal null} or empty.
*/
SelectWithProjection<T> inTable(String table);
/**
* Explicitly set the name of the table to perform the query on.
* <p>
* Skip this step to use the default table derived from the domain type.
*
* @param table must not be {@literal null}.
* @return new instance of {@link SelectWithProjection}.
* @throws IllegalArgumentException if {@link CqlIdentifier} is {@literal null}.
*/
SelectWithProjection<T> inTable(CqlIdentifier table);
}
/**
* Result type override (Optional).
*/
interface SelectWithProjection<T> extends SelectWithQuery<T> {
/**
* Define the target type fields should be mapped to. <br />
* Skip this step if you are anyway only interested in the original domain type.
*
* @param resultType must not be {@literal null}.
* @param <R> result type.
* @return new instance of {@link SelectWithProjection}.
* @throws IllegalArgumentException if resultType is {@literal null}.
*/
<R> SelectWithQuery<R> as(Class<R> resultType);
}
/**
* {@link ReactiveSelect} provides methods for constructing {@code SELECT} operations in a fluent way.
* The {@link ReactiveSelect} interface provides methods for constructing {@code SELECT} operations
* in a fluent way.
*/
interface ReactiveSelect<T> extends SelectWithTable<T>, SelectWithProjection<T> {}
}

78
spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/ReactiveSelectOperationSupport.java
View File

@@ -19,6 +19,7 @@ import lombok.AccessLevel;
import lombok.NonNull;
import lombok.RequiredArgsConstructor;
import lombok.experimental.FieldDefaults;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
@@ -32,6 +33,8 @@ import org.springframework.util.Assert;
* Implementation of {@link ReactiveSelectOperation}.
*
* @author Mark Paluch
* @see org.springframework.data.cassandra.core.ReactiveSelectOperation
* @see org.springframework.data.cassandra.core.query.Query
* @since 2.1
*/
@RequiredArgsConstructor
@@ -45,15 +48,14 @@ class ReactiveSelectOperationSupport implements ReactiveSelectOperation {
@Override
public <T> ReactiveSelect<T> query(Class<T> domainType) {
Assert.notNull(domainType, "DomainType must not be null!");
Assert.notNull(domainType, "DomainType must not be null");
return new ReactiveSelectSupport<>(template, domainType, domainType, Query.empty(), null);
return new ReactiveSelectSupport<>(this.template, domainType, domainType, Query.empty(), null);
}
@RequiredArgsConstructor
@FieldDefaults(level = AccessLevel.PRIVATE, makeFinal = true)
static class ReactiveSelectSupport<T>
implements ReactiveSelect<T>, SelectWithTable<T>, SelectWithProjection<T>, SelectWithQuery<T> {
static class ReactiveSelectSupport<T> implements ReactiveSelect<T> {
@NonNull ReactiveCassandraTemplate template;
@@ -65,26 +67,15 @@ class ReactiveSelectOperationSupport implements ReactiveSelectOperation {
@Nullable CqlIdentifier tableName;
/* (non-Javadoc)
* @see org.springframework.data.cassandra.core.ReactiveSelectOperation.SelectWithTable#inTable(java.lang.String)
*/
@Override
public SelectWithProjection<T> inTable(String tableName) {
Assert.hasText(tableName, "Table name must not be null or empty!");
return new ReactiveSelectSupport<>(template, domainType, returnType, query, CqlIdentifier.of(tableName));
}
/* (non-Javadoc)
* @see org.springframework.data.cassandra.core.ReactiveSelectOperation.SelectWithTable#inTable(org.springframework.data.cassandra.core.cql.CqlIdentifier)
*/
@Override
public SelectWithProjection<T> inTable(CqlIdentifier tableName) {
Assert.notNull(tableName, "Table name must not be null!");
Assert.notNull(tableName, "Table name must not be null");
return new ReactiveSelectSupport<>(template, domainType, returnType, query, tableName);
return new ReactiveSelectSupport<>(this.template, this.domainType, this.returnType, this.query, tableName);
}
/* (non-Javadoc)
@@ -93,9 +84,9 @@ class ReactiveSelectOperationSupport implements ReactiveSelectOperation {
@Override
public <R> SelectWithQuery<R> as(Class<R> returnType) {
Assert.notNull(returnType, "ReturnType must not be null!");
Assert.notNull(returnType, "ReturnType must not be null");
return new ReactiveSelectSupport<>(template, domainType, returnType, query, tableName);
return new ReactiveSelectSupport<>(this.template, this.domainType, returnType, this.query, this.tableName);
}
/* (non-Javadoc)
@@ -104,9 +95,25 @@ class ReactiveSelectOperationSupport implements ReactiveSelectOperation {
@Override
public TerminatingSelect<T> matching(Query query) {
Assert.notNull(query, "Query must not be null!");
Assert.notNull(query, "Query must not be null");
return new ReactiveSelectSupport<>(template, domainType, returnType, query, tableName);
return new ReactiveSelectSupport<>(this.template, this.domainType, this.returnType, query, this.tableName);
}
/* (non-Javadoc)
* @see org.springframework.data.cassandra.core.ReactiveSelectOperation.TerminatingSelect#count()
*/
@Override
public Mono<Long> count() {
return this.template.doCount(this.query, this.domainType, getTableName());
}
/* (non-Javadoc)
* @see org.springframework.data.cassandra.core.ReactiveSelectOperation.TerminatingSelect#exists()
*/
@Override
public Mono<Boolean> exists() {
return this.template.doExists(this.query, this.domainType, getTableName());
}
/* (non-Javadoc)
@@ -114,7 +121,7 @@ class ReactiveSelectOperationSupport implements ReactiveSelectOperation {
*/
@Override
public Mono<T> first() {
return template.doSelect(query.limit(1), domainType, getTableName(), returnType).next();
return this.template.doSelect(this.query.limit(1), this.domainType, getTableName(), this.returnType).next();
}
/* (non-Javadoc)
@@ -123,7 +130,8 @@ class ReactiveSelectOperationSupport implements ReactiveSelectOperation {
@Override
public Mono<T> one() {
Flux<T> result = template.doSelect(query.limit(2), domainType, getTableName(), returnType);
Flux<T> result =
this.template.doSelect(this.query.limit(2), this.domainType, getTableName(), this.returnType);
return result.collectList() //
.flatMap(it -> {
@@ -133,8 +141,8 @@ class ReactiveSelectOperationSupport implements ReactiveSelectOperation {
}
if (it.size() > 1) {
return Mono.error(
new IncorrectResultSizeDataAccessException("Query " + query + " returned non unique result.", 1));
return Mono.error(new IncorrectResultSizeDataAccessException(
String.format("Query [%s] returned non unique result.", this.query), 1));
}
return Mono.just(it.get(0));
@@ -146,27 +154,11 @@ class ReactiveSelectOperationSupport implements ReactiveSelectOperation {
*/
@Override
public Flux<T> all() {
return template.doSelect(query, domainType, getTableName(), returnType);
}
/* (non-Javadoc)
* @see org.springframework.data.cassandra.core.ReactiveSelectOperation.TerminatingSelect#count()
*/
@Override
public Mono<Long> count() {
return template.doCount(query, domainType, getTableName());
}
/* (non-Javadoc)
* @see org.springframework.data.cassandra.core.ReactiveSelectOperation.TerminatingSelect#exists()
*/
@Override
public Mono<Boolean> exists() {
return template.doExists(query, domainType, getTableName());
return this.template.doSelect(this.query, this.domainType, getTableName(), this.returnType);
}
private CqlIdentifier getTableName() {
return tableName != null ? tableName : template.getTableName(domainType);
return this.tableName != null ? this.tableName : this.template.getTableName(this.domainType);
}
}
}

103
spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/ReactiveUpdateOperation.java
View File

@@ -20,16 +20,18 @@ import reactor.core.publisher.Mono;
import org.springframework.data.cassandra.core.cql.CqlIdentifier;
import org.springframework.data.cassandra.core.query.Query;
import org.springframework.data.cassandra.core.query.Update;
import org.springframework.util.Assert;
/**
* {@link ReactiveUpdateOperation} allows creation and execution of Cassandra {@code UPDATE} operation in a fluent API
* style.
* The {@link ReactiveUpdateOperation} interface allows creation and execution of Cassandra {@code UPDATE} operations
* in a fluent API style.
* <p>
* The starting {@literal domainType} is used for mapping the {@link Query} provided via {@code matching}, as well as
* the {@link Update} via {@code apply} into the Cassandra specific representations. The table to operate on is by
* default derived from the initial {@literal domainType} and can be defined there via
* {@link org.springframework.data.cassandra.core.mapping.Table}. Using {@code inTable} allows to override the table
* name for the execution.
* the {@link Update} via {@code apply} into the Cassandra specific representations.
* <p>
* By default, the table to operate on is derived from the initial {@literal domainType} and can be defined there
* via the {@link org.springframework.data.cassandra.core.mapping.Table} annotation. Using {@code inTable} allows
* a developer to override the table name for the execution.
*
* <pre>
* <code>
@@ -42,92 +44,101 @@ import org.springframework.data.cassandra.core.query.Update;
* </pre>
*
* @author Mark Paluch
* @author John Blum
* @see org.springframework.data.cassandra.core.query.Query
* @see org.springframework.data.cassandra.core.query.Update
* @since 2.1
*/
public interface ReactiveUpdateOperation {
/**
* Start creating an {@code UPDATE} operation for the given {@literal domainType}.
* Begin creating an {@code UPDATE} operation for the given {@link Class domainType}.
*
* @param domainType must not be {@literal null}.
* @param <T> {@link Class type} of the application domain object.
* @param domainType {@link Class type} of domain object to update; must not be {@literal null}.
* @return new instance of {@link ReactiveUpdate}.
* @throws IllegalArgumentException if domainType is {@literal null}.
* @throws IllegalArgumentException if {@link Class domainType} is {@literal null}.
* @see ReactiveUpdate
*/
<T> ReactiveUpdate<T> update(Class<T> domainType);
/**
* Declare the {@link Update} to apply.
*/
interface UpdateWithUpdate<T> {
/**
* Set the {@link Update} to be applied.
*
* @param update must not be {@literal null}.
* @return new instance of {@link TerminatingUpdate}.
* @throws IllegalArgumentException if update is {@literal null}.
*/
TerminatingUpdate<T> apply(Update update);
}
/**
* Explicitly define the name of the table to perform operation in.
* Table override (optional).
*/
interface UpdateWithTable<T> {
/**
* Explicitly set the name of the table to perform the query on.
* Explicitly set the {@link String name} of the table on which to perform the update.
* <p>
* Skip this step to use the default table derived from the domain type.
* Skip this step to use the default table derived from the {@link Class domain type}.
*
* @param table must not be {@literal null} or empty.
* @return new instance of {@link UpdateWithTable}.
* @throws IllegalArgumentException if {@code table} is {@literal null} or empty.
* @param table {@link String name} of the table; must not be {@literal null} or empty.
* @return new instance of {@link UpdateWithQuery}.
* @throws IllegalArgumentException if {@link String table} is {@literal null} or empty.
* @see #inTable(CqlIdentifier)
* @see UpdateWithQuery
*/
UpdateWithQuery<T> inTable(String table);
default UpdateWithQuery<T> inTable(String table) {
Assert.hasText(table, "Table name must not be null or empty");
return inTable(CqlIdentifier.of(table));
}
/**
* Explicitly set the name of the table to perform the query on.
* Explicitly set the {@link CqlIdentifier name} of the table to on which to perform the update.
* <p>
* Skip this step to use the default table derived from the domain type.
* Skip this step to use the default table derived from the {@link Class domain type}.
*
* @param table must not be {@literal null}.
* @return new instance of {@link UpdateWithTable}.
* @throws IllegalArgumentException if {@link CqlIdentifier} is {@literal null}.
* @param table {@link CqlIdentifier name} of the table; must not be {@literal null}.
* @return new instance of {@link UpdateWithQuery}.
* @throws IllegalArgumentException if {@link CqlIdentifier table} is {@literal null}.
* @see org.springframework.data.cassandra.core.cql.CqlIdentifier
* @see UpdateWithQuery
*/
UpdateWithQuery<T> inTable(CqlIdentifier table);
}
/**
* Define a filter query for the {@link Update}.
* Define a {@link Query} used as the filter for the {@link Update}.
*/
interface UpdateWithQuery<T> {
/**
* Filter documents by given {@literal query}.
* Filter rows to update by the given {@link Query}.
*
* @param query must not be {@literal null}.
* @return new instance of {@link UpdateWithQuery}.
* @throws IllegalArgumentException if query is {@literal null}.
* @param query {@link Query} used as a filter in the update; must not be {@literal null}.
* @return new instance of {@link TerminatingUpdate}.
* @throws IllegalArgumentException if {@link Query} is {@literal null}.
* @see org.springframework.data.cassandra.core.query.Query
* @see TerminatingUpdate
*/
UpdateWithUpdate<T> matching(Query query);
TerminatingUpdate<T> matching(Query query);
}
/**
* Trigger update execution by calling one of the terminating methods.
* Trigger {@code UPDATE} execution by calling one of the terminating methods.
*/
interface TerminatingUpdate<T> {
/**
* Update all matching rows in the table.
*
* @return never {@literal null}.
* @return the {@link WriteResult} of the update; never {@literal null}.
* @see org.springframework.data.cassandra.core.query.Update
* @see org.springframework.data.cassandra.core.WriteResult
* @see reactor.core.publisher.Mono
*/
Mono<WriteResult> all();
Mono<WriteResult> apply(Update update);
}
/**
* {@link ReactiveUpdate} provides methods for constructing {@code UPDATE} operations in a fluent way.
* The {@link ReactiveUpdate} interface provides methods for constructing {@code UPDATE} operations
* in a fluent way.
*/
interface ReactiveUpdate<T> extends UpdateWithTable<T>, UpdateWithQuery<T> {}
}

56
spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/ReactiveUpdateOperationSupport.java
View File

@@ -19,6 +19,7 @@ import lombok.AccessLevel;
import lombok.NonNull;
import lombok.RequiredArgsConstructor;
import lombok.experimental.FieldDefaults;
import reactor.core.publisher.Mono;
import org.springframework.data.cassandra.core.cql.CqlIdentifier;
@@ -31,6 +32,9 @@ import org.springframework.util.Assert;
* Implementation of {@link ReactiveUpdateOperation}.
*
* @author Mark Paluch
* @see org.springframework.data.cassandra.core.ReactiveUpdateOperation
* @see org.springframework.data.cassandra.core.query.Query
* @see org.springframework.data.cassandra.core.query.Update
* @since 2.1
*/
@RequiredArgsConstructor
@@ -44,15 +48,14 @@ class ReactiveUpdateOperationSupport implements ReactiveUpdateOperation {
@Override
public <T> ReactiveUpdate<T> update(Class<T> domainType) {
Assert.notNull(domainType, "DomainType must not be null!");
Assert.notNull(domainType, "DomainType must not be null");
return new ReactiveUpdateSupport<>(template, domainType, Query.empty(), null, null);
return new ReactiveUpdateSupport<>(this.template, domainType, Query.empty(), null);
}
@RequiredArgsConstructor
@FieldDefaults(level = AccessLevel.PRIVATE, makeFinal = true)
static class ReactiveUpdateSupport<T>
implements ReactiveUpdate<T>, UpdateWithTable<T>, UpdateWithQuery<T>, UpdateWithUpdate<T>, TerminatingUpdate<T> {
static class ReactiveUpdateSupport<T> implements ReactiveUpdate<T>, TerminatingUpdate<T> {
@NonNull ReactiveCassandraTemplate template;
@@ -60,64 +63,43 @@ class ReactiveUpdateOperationSupport implements ReactiveUpdateOperation {
@NonNull Query query;
@Nullable Update update;
@Nullable CqlIdentifier tableName;
/* (non-Javadoc)
* @see org.springframework.data.cassandra.core.ReactiveUpdateOperation.UpdateWithUpdate#apply(org.springframework.data.cassandra.core.query.Update)
*/
@Override
public TerminatingUpdate<T> apply(Update update) {
Assert.notNull(update, "Update must not be null!");
return new ReactiveUpdateSupport<>(template, domainType, query, update, tableName);
}
/* (non-Javadoc)
* @see org.springframework.data.cassandra.core.ReactiveUpdateOperation.UpdateWithTable#inTable(java.lang.String)
*/
@Override
public UpdateWithQuery<T> inTable(String tableName) {
Assert.hasText(tableName, "Table name must not be null or empty!");
return new ReactiveUpdateSupport<>(template, domainType, query, update, CqlIdentifier.of(tableName));
}
/* (non-Javadoc)
* @see org.springframework.data.cassandra.core.ReactiveUpdateOperation.UpdateWithTable#inTable(org.springframework.data.cassandra.core.cql.CqlIdentifier)
*/
@Override
public UpdateWithQuery<T> inTable(CqlIdentifier tableName) {
Assert.notNull(tableName, "Table name must not be null!");
Assert.notNull(tableName, "Table name must not be null");
return new ReactiveUpdateSupport<>(template, domainType, query, update, tableName);
return new ReactiveUpdateSupport<>(this.template, this.domainType, this.query, tableName);
}
/* (non-Javadoc)
* @see org.springframework.data.cassandra.core.ReactiveUpdateOperation.UpdateWithQuery#matching(org.springframework.data.cassandra.core.query.Query)
*/
@Override
public UpdateWithUpdate<T> matching(Query query) {
public TerminatingUpdate<T> matching(Query query) {
Assert.notNull(query, "Query must not be null!");
Assert.notNull(query, "Query must not be null");
return new ReactiveUpdateSupport<>(template, domainType, query, update, tableName);
return new ReactiveUpdateSupport<>(this.template, this.domainType, query, this.tableName);
}
/* (non-Javadoc)
* @see org.springframework.data.cassandra.core.ReactiveUpdateOperation.TerminatingUpdate#all()
* @see org.springframework.data.cassandra.core.ReactiveUpdateOperation.UpdateWithUpdate#apply(org.springframework.data.cassandra.core.query.Update)
*/
@Override
public Mono<WriteResult> all() {
return template.doUpdate(query, update, domainType, getTableName());
public Mono<WriteResult> apply(Update update) {
Assert.notNull(update, "Update must not be null");
return this.template.doUpdate(this.query, update, this.domainType, getTableName());
}
private CqlIdentifier getTableName() {
return tableName != null ? tableName : template.getTableName(domainType);
return this.tableName != null ? this.tableName : this.template.getTableName(this.domainType);
}
}
}

27
spring-data-cassandra/src/test/java/org/springframework/data/cassandra/core/ReactiveDeleteOperationSupportTests.java
View File

@@ -15,17 +15,19 @@
*/
package org.springframework.data.cassandra.core;
import static org.springframework.data.cassandra.core.query.Criteria.*;
import static org.springframework.data.cassandra.core.query.Query.*;
import lombok.Data;
import reactor.core.publisher.Mono;
import reactor.test.StepVerifier;
import static org.springframework.data.cassandra.core.query.Criteria.where;
import static org.springframework.data.cassandra.core.query.Query.query;
import java.util.Collections;
import lombok.Data;
import reactor.core.publisher.Mono;
import reactor.test.StepVerifier;
import org.junit.Before;
import org.junit.Test;
import org.springframework.data.annotation.Id;
import org.springframework.data.cassandra.core.convert.MappingCassandraConverter;
import org.springframework.data.cassandra.core.cql.CqlIdentifier;
@@ -44,6 +46,7 @@ import org.springframework.data.cassandra.test.util.AbstractKeyspaceCreatingInte
public class ReactiveDeleteOperationSupportTests extends AbstractKeyspaceCreatingIntegrationTest {
CassandraAdminTemplate admin;
ReactiveCassandraTemplate template;
Person han;
@@ -74,7 +77,10 @@ public class ReactiveDeleteOperationSupportTests extends AbstractKeyspaceCreatin
@Test // DATACASS-485
public void removeAllMatching() {
Mono<WriteResult> writeResult = template.delete(Person.class).matching(query(where("id").is(han.id))).all();
Mono<WriteResult> writeResult = this.template
.delete(Person.class)
.matching(query(where("id").is(han.id)))
.all();
StepVerifier.create(writeResult.map(WriteResult::wasApplied)).expectNext(true).verifyComplete();
}
@@ -82,8 +88,10 @@ public class ReactiveDeleteOperationSupportTests extends AbstractKeyspaceCreatin
@Test // DATACASS-485
public void removeAllMatchingWithAlternateDomainTypeAndCollection() {
Mono<WriteResult> writeResult = template.delete(Jedi.class).inTable("person")
.matching(query(where("id").in(han.id, luke.id))).all();
Mono<WriteResult> writeResult = this.template
.delete(Jedi.class).inTable("person")
.matching(query(where("id").in(han.id, luke.id)))
.all();
StepVerifier.create(writeResult.map(WriteResult::wasApplied)).expectNext(true).verifyComplete();
StepVerifier.create(template.select(Query.empty(), Person.class)).verifyComplete();
@@ -98,7 +106,6 @@ public class ReactiveDeleteOperationSupportTests extends AbstractKeyspaceCreatin
@Data
static class Jedi {
@Column("firstname") String name;
}
}

103
spring-data-cassandra/src/test/java/org/springframework/data/cassandra/core/ReactiveInsertOperationSupportTests.java
View File

@@ -15,16 +15,18 @@
*/
package org.springframework.data.cassandra.core;
import static org.assertj.core.api.Assertions.*;
import lombok.Data;
import reactor.core.publisher.Mono;
import reactor.test.StepVerifier;
import static org.assertj.core.api.Assertions.assertThat;
import java.util.Collections;
import lombok.Data;
import reactor.core.publisher.Mono;
import reactor.test.StepVerifier;
import org.junit.Before;
import org.junit.Test;
import org.springframework.data.annotation.Id;
import org.springframework.data.cassandra.core.convert.MappingCassandraConverter;
import org.springframework.data.cassandra.core.cql.CqlIdentifier;
@@ -41,6 +43,7 @@ import org.springframework.data.cassandra.test.util.AbstractKeyspaceCreatingInte
public class ReactiveInsertOperationSupportTests extends AbstractKeyspaceCreatingIntegrationTest {
CassandraAdminTemplate admin;
ReactiveCassandraTemplate template;
Person han;
@@ -58,50 +61,6 @@ public class ReactiveInsertOperationSupportTests extends AbstractKeyspaceCreatin
initPersons();
}
@Test(expected = IllegalArgumentException.class) // DATACASS-485
public void domainTypeIsRequired() {
template.insert((Class) null);
}
@Test(expected = IllegalArgumentException.class) // DATACASS-485
public void tableIsRequiredOnSet() {
template.insert(Person.class).inTable((String) null);
}
@Test(expected = IllegalArgumentException.class) // DATACASS-485
public void optionsIsRequiredOnSet() {
template.insert(Person.class).withOptions(null);
}
@Test // DATACASS-485
public void insertOne() {
Mono<WriteResult> writeResult = template.insert(Person.class).inTable("person").one(han);
StepVerifier.create(writeResult.map(WriteResult::wasApplied)).expectNext(true).verifyComplete();
StepVerifier.create(template.selectOneById(han.id, Person.class)).expectNext(han).verifyComplete();
}
@Test // DATACASS-485
public void insertOneWithOptions() {
template.insert(Person.class).inTable("person").one(han);
Mono<WriteResult> writeResult = template.insert(Person.class).inTable("person")
.withOptions(InsertOptions.builder().withIfNotExists().build()).one(han);
StepVerifier.create(writeResult).assertNext(it -> assertThat(it.wasApplied()).isTrue()).verifyComplete();
StepVerifier.create(template.selectOneById(han.id, Person.class)).expectNext(han).verifyComplete();
}
@Data
@Table
static class Person {
@Id String id;
@Indexed String firstname;
@Indexed String lastname;
}
private void initPersons() {
@@ -115,4 +74,50 @@ public class ReactiveInsertOperationSupportTests extends AbstractKeyspaceCreatin
luke.lastname = "skywalker";
luke.id = "id-2";
}
@Test(expected = IllegalArgumentException.class) // DATACASS-485
public void domainTypeIsRequired() {
this.template.insert((Class) null);
}
@Test(expected = IllegalArgumentException.class) // DATACASS-485
public void optionsIsRequiredOnSet() {
this.template.insert(Person.class).withOptions(null);
}
@Test(expected = IllegalArgumentException.class) // DATACASS-485
public void tableIsRequiredOnSet() {
this.template.insert(Person.class).inTable((String) null);
}
@Test // DATACASS-485
public void insertOne() {
Mono<WriteResult> writeResult = this.template.insert(Person.class).inTable("person").one(han);
StepVerifier.create(writeResult.map(WriteResult::wasApplied)).expectNext(true).verifyComplete();
StepVerifier.create(template.selectOneById(han.id, Person.class)).expectNext(han).verifyComplete();
}
@Test // DATACASS-485
public void insertOneWithOptions() {
this.template.insert(Person.class).inTable("person").one(han);
Mono<WriteResult> writeResult = this.template
.insert(Person.class).inTable("person")
.withOptions(InsertOptions.builder().withIfNotExists().build())
.one(han);
StepVerifier.create(writeResult).assertNext(it -> assertThat(it.wasApplied()).isTrue()).verifyComplete();
StepVerifier.create(template.selectOneById(han.id, Person.class)).expectNext(han).verifyComplete();
}
@Data
@Table
static class Person {
@Id String id;
@Indexed String firstname;
@Indexed String lastname;
}
}

568
spring-data-cassandra/src/test/java/org/springframework/data/cassandra/core/ReactiveSelectOperationSupportTests.java
View File

@@ -15,21 +15,23 @@
*/
package org.springframework.data.cassandra.core;
import static org.assertj.core.api.Assertions.*;
import static org.springframework.data.cassandra.core.query.Criteria.*;
import static org.springframework.data.cassandra.core.query.Query.*;
import static org.assertj.core.api.Assertions.assertThat;
import static org.springframework.data.cassandra.core.query.Criteria.where;
import static org.springframework.data.cassandra.core.query.Query.query;
import java.util.Collections;
import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import reactor.test.StepVerifier;
import java.util.Collections;
import org.junit.Before;
import org.junit.Test;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.dao.IncorrectResultSizeDataAccessException;
import org.springframework.data.annotation.Id;
@@ -50,6 +52,7 @@ import org.springframework.data.cassandra.test.util.AbstractKeyspaceCreatingInte
public class ReactiveSelectOperationSupportTests extends AbstractKeyspaceCreatingIntegrationTest {
CassandraAdminTemplate admin;
ReactiveCassandraTemplate template;
Person han;
@@ -67,284 +70,6 @@ public class ReactiveSelectOperationSupportTests extends AbstractKeyspaceCreatin
initPersons();
}
@Test(expected = IllegalArgumentException.class) // DATACASS-485
public void domainTypeIsRequired() {
template.query(null);
}
@Test(expected = IllegalArgumentException.class) // DATACASS-485
public void returnTypeIsRequiredOnSet() {
template.query(Person.class).as(null);
}
@Test(expected = IllegalArgumentException.class) // DATACASS-485
public void tableIsRequiredOnSet() {
template.query(Person.class).inTable((String) null);
}
@Test // DATACASS-485
public void findAll() {
Flux<Person> result = template.query(Person.class).all();
StepVerifier.create(result.collectList()).assertNext(actual -> {
assertThat(actual).containsExactlyInAnyOrder(han, luke);
}).verifyComplete();
}
@Test // DATACASS-485
public void findAllWithCollection() {
Flux<Human> result = template.query(Human.class).inTable("person").all();
StepVerifier.create(result).expectNextCount(2).verifyComplete();
}
@Test // DATACASS-485
public void findAllWithProjection() {
Flux<Jedi> result = template.query(Person.class).as(Jedi.class).all();
StepVerifier.create(result.collectList()).assertNext(actual -> {
assertThat(actual).hasOnlyElementsOfType(Jedi.class).hasSize(2);
}).verifyComplete();
}
@Test // DATACASS-485
public void findByReturningAllValuesAsClosedInterfaceProjection() {
Flux<PersonProjection> result = template.query(Person.class).as(PersonProjection.class).all();
StepVerifier.create(result.collectList()).assertNext(actual -> {
assertThat(actual).hasOnlyElementsOfType(PersonProjection.class).hasSize(2);
}).verifyComplete();
}
@Test // DATACASS-485
public void findAllBy() {
Flux<Person> result = template.query(Person.class).matching(queryLuke()).all();
StepVerifier.create(result).expectNext(luke).verifyComplete();
}
@Test // DATACASS-485
public void findAllByWithCollectionUsingMappingInformation() {
Flux<Jedi> result = template.query(Jedi.class).inTable("person").all();
StepVerifier.create(result.collectList()).assertNext(actual -> {
assertThat(actual).isNotEmpty().hasOnlyElementsOfType(Jedi.class);
}).verifyComplete();
}
@Test // DATACASS-485
public void findAllByWithCollection() {
Flux<Human> result = template.query(Human.class).inTable("person").matching(queryLuke()).all();
StepVerifier.create(result.collectList()).expectNextCount(1).verifyComplete();
}
@Test // DATACASS-485
public void findAllByWithProjection() {
Flux<Jedi> result = template.query(Person.class).as(Jedi.class).all();
StepVerifier.create(result.collectList()).assertNext(actual -> {
assertThat(actual).isNotEmpty().hasOnlyElementsOfType(Jedi.class);
}).verifyComplete();
}
@Test // DATACASS-485
public void findBy() {
Mono<Person> result = template.query(Person.class).matching(queryLuke()).one();
StepVerifier.create(result).expectNext(luke).verifyComplete();
}
@Test // DATACASS-485
public void findByNoMatch() {
Mono<Person> result = template.query(Person.class).matching(querySpock()).one();
StepVerifier.create(result).verifyComplete();
}
@Test // DATACASS-485
public void findByTooManyResults() {
Mono<Person> result = template.query(Person.class).one();
StepVerifier.create(result).expectError(IncorrectResultSizeDataAccessException.class).verify();
}
@Test // DATACASS-485
public void findByReturningFirstValue() {
Mono<Person> result = template.query(Person.class).matching(queryLuke()).first();
StepVerifier.create(result).expectNext(luke).verifyComplete();
}
@Test // DATACASS-485
public void findByReturningFirstValueForManyResults() {
Mono<Person> result = template.query(Person.class).first();
StepVerifier.create(result).assertNext(actual -> {
assertThat(actual).isIn(han, luke);
}).verifyComplete();
}
@Test // DATACASS-485
public void findByReturningFirstValueAsClosedInterfaceProjection() {
Mono<PersonProjection> result = template.query(Person.class).as(PersonProjection.class)
.matching(query(where("firstname").is("han")).withAllowFiltering()).first();
StepVerifier.create(result).assertNext(actual -> {
assertThat(actual).isInstanceOf(PersonProjection.class);
assertThat(actual.getFirstname()).isEqualTo("han");
}).verifyComplete();
}
@Test // DATACASS-485
public void findByReturningFirstValueAsOpenInterfaceProjection() {
Mono<PersonSpELProjection> result = template.query(Person.class).as(PersonSpELProjection.class)
.matching(query(where("firstname").is("han")).withAllowFiltering()).first();
StepVerifier.create(result).assertNext(actual -> {
assertThat(actual).isInstanceOf(PersonSpELProjection.class);
assertThat(actual.getName()).isEqualTo("han");
}).verifyComplete();
}
@Test // DATACASS-485
public void countShouldReturnNrOfElementsInCollectionWhenNoQueryPresent() {
Mono<Long> count = template.query(Person.class).count();
StepVerifier.create(count).expectNext(2L).verifyComplete();
}
@Test // DATACASS-485
public void countShouldReturnNrOfElementsMatchingQuery() {
Mono<Long> count = template.query(Person.class)
.matching(query(where("firstname").is(luke.getFirstname())).withAllowFiltering()).count();
StepVerifier.create(count).expectNext(1L).verifyComplete();
}
@Test // DATACASS-485
public void existsShouldReturnTrueIfAtLeastOneElementExistsInCollection() {
Mono<Boolean> exists = template.query(Person.class).exists();
StepVerifier.create(exists).expectNext(true).verifyComplete();
}
@Test // DATACASS-485
public void existsShouldReturnFalseIfNoElementExistsInCollection() {
StepVerifier.create(template.truncate(Person.class)).verifyComplete();
Mono<Boolean> exists = template.query(Person.class).exists();
StepVerifier.create(exists).expectNext(false).verifyComplete();
}
@Test // DATACASS-485
public void existsShouldReturnTrueIfAtLeastOneElementMatchesQuery() {
Mono<Boolean> exists = template.query(Person.class).matching(queryLuke()).exists();
StepVerifier.create(exists).expectNext(true).verifyComplete();
}
@Test // DATACASS-485
public void existsShouldReturnFalseWhenNoElementMatchesQuery() {
Mono<Boolean> exists = template.query(Person.class).matching(querySpock()).exists();
StepVerifier.create(exists).expectNext(false).verifyComplete();
}
@Test // DATACASS-485
public void returnsTargetObjectDirectlyIfProjectionInterfaceIsImplemented() {
Flux<Contact> result = template.query(Person.class).as(Contact.class).all();
StepVerifier.create(result.collectList()).assertNext(actual -> {
assertThat(actual).allMatch(it -> it instanceof Person);
}).verifyComplete();
}
private static Query queryLuke() {
return query(where("firstname").is("luke")).withAllowFiltering();
}
private static Query querySpock() {
return query(where("firstname").is("spock")).withAllowFiltering();
}
interface Contact {}
@Data
@Table
static class Person implements Contact {
@Id String id;
@Indexed String firstname;
@Indexed String lastname;
}
interface PersonProjection {
String getFirstname();
}
public interface PersonSpELProjection {
@Value("#{target.firstname}")
String getName();
}
@Data
static class Human {
@Id String id;
}
@Data
@AllArgsConstructor
@NoArgsConstructor
static class Jedi {
@Column("firstname") String name;
}
@Data
static class Sith {
String rank;
}
interface PlanetProjection {
String getName();
}
interface PlanetSpELProjection {
@Value("#{target.name}")
String getId();
}
private void initPersons() {
han = new Person();
@@ -360,4 +85,281 @@ public class ReactiveSelectOperationSupportTests extends AbstractKeyspaceCreatin
admin.insert(han);
admin.insert(luke);
}
@Test(expected = IllegalArgumentException.class) // DATACASS-485
public void domainTypeIsRequired() {
this.template.query(null);
}
@Test(expected = IllegalArgumentException.class) // DATACASS-485
public void returnTypeIsRequiredOnSet() {
this.template.query(Person.class).as(null);
}
@Test(expected = IllegalArgumentException.class) // DATACASS-485
public void tableIsRequiredOnSet() {
this.template.query(Person.class).inTable((String) null);
}
@Test // DATACASS-485
public void findAll() {
Flux<Person> result = this.template.query(Person.class).all();
StepVerifier.create(result.collectList()).assertNext(actual ->
assertThat(actual).containsExactlyInAnyOrder(han, luke)
).verifyComplete();
}
@Test // DATACASS-485
public void findAllWithCollection() {
Flux<Human> result = this.template.query(Human.class).inTable("person").all();
StepVerifier.create(result).expectNextCount(2).verifyComplete();
}
@Test // DATACASS-485
public void findAllWithProjection() {
Flux<Jedi> result = this.template.query(Person.class).as(Jedi.class).all();
StepVerifier.create(result.collectList()).assertNext(actual ->
assertThat(actual).hasOnlyElementsOfType(Jedi.class).hasSize(2)
).verifyComplete();
}
@Test // DATACASS-485
public void findByReturningAllValuesAsClosedInterfaceProjection() {
Flux<PersonProjection> result = this.template.query(Person.class).as(PersonProjection.class).all();
StepVerifier.create(result.collectList()).assertNext(actual ->
assertThat(actual).hasOnlyElementsOfType(PersonProjection.class).hasSize(2)
).verifyComplete();
}
@Test // DATACASS-485
public void findAllBy() {
Flux<Person> result = this.template.query(Person.class).matching(queryLuke()).all();
StepVerifier.create(result).expectNext(luke).verifyComplete();
}
@Test // DATACASS-485
public void findAllByWithCollectionUsingMappingInformation() {
Flux<Jedi> result = this.template.query(Jedi.class).inTable("person").all();
StepVerifier.create(result.collectList()).assertNext(actual ->
assertThat(actual).isNotEmpty().hasOnlyElementsOfType(Jedi.class)
).verifyComplete();
}
@Test // DATACASS-485
public void findAllByWithCollection() {
Flux<Human> result = this.template.query(Human.class).inTable("person").matching(queryLuke()).all();
StepVerifier.create(result.collectList()).expectNextCount(1).verifyComplete();
}
@Test // DATACASS-485
public void findAllByWithProjection() {
Flux<Jedi> result = this.template.query(Person.class).as(Jedi.class).all();
StepVerifier.create(result.collectList()).assertNext(actual ->
assertThat(actual).isNotEmpty().hasOnlyElementsOfType(Jedi.class)
).verifyComplete();
}
@Test // DATACASS-485
public void findBy() {
Mono<Person> result = this.template.query(Person.class).matching(queryLuke()).one();
StepVerifier.create(result).expectNext(luke).verifyComplete();
}
@Test // DATACASS-485
public void findByNoMatch() {
Mono<Person> result = this.template.query(Person.class).matching(querySpock()).one();
StepVerifier.create(result).verifyComplete();
}
@Test // DATACASS-485
public void findByTooManyResults() {
Mono<Person> result = this.template.query(Person.class).one();
StepVerifier.create(result).expectError(IncorrectResultSizeDataAccessException.class).verify();
}
@Test // DATACASS-485
public void findByReturningFirst() {
Mono<Person> result = this.template.query(Person.class).matching(queryLuke()).first();
StepVerifier.create(result).expectNext(luke).verifyComplete();
}
@Test // DATACASS-485
public void findByReturningFirstForManyResults() {
Mono<Person> result = this.template.query(Person.class).first();
StepVerifier.create(result).assertNext(actual ->
assertThat(actual).isIn(han, luke)
).verifyComplete();
}
@Test // DATACASS-485
public void findByReturningFirstAsClosedInterfaceProjection() {
Mono<PersonProjection> result = this.template
.query(Person.class)
.as(PersonProjection.class)
.matching(query(where("firstname").is("han")).withAllowFiltering())
.first();
StepVerifier.create(result).assertNext(actual -> {
assertThat(actual).isInstanceOf(PersonProjection.class);
assertThat(actual.getFirstname()).isEqualTo("han");
}).verifyComplete();
}
@Test // DATACASS-485
public void findByReturningFirstAsOpenInterfaceProjection() {
Mono<PersonSpELProjection> result = this.template
.query(Person.class)
.as(PersonSpELProjection.class)
.matching(query(where("firstname").is("han")).withAllowFiltering())
.first();
StepVerifier.create(result).assertNext(actual -> {
assertThat(actual).isInstanceOf(PersonSpELProjection.class);
assertThat(actual.getName()).isEqualTo("han");
}).verifyComplete();
}
@Test // DATACASS-485
public void countShouldReturnNumberOfElementsInCollectionWhenNoQueryPresent() {
Mono<Long> count = this.template.query(Person.class).count();
StepVerifier.create(count).expectNext(2L).verifyComplete();
}
@Test // DATACASS-485
public void countShouldReturnNrOfElementsMatchingQuery() {
Mono<Long> count = this.template
.query(Person.class)
.matching(query(where("firstname").is(luke.getFirstname())).withAllowFiltering())
.count();
StepVerifier.create(count).expectNext(1L).verifyComplete();
}
@Test // DATACASS-485
public void existsShouldReturnTrueIfAtLeastOneElementExistsInCollection() {
Mono<Boolean> exists = this.template.query(Person.class).exists();
StepVerifier.create(exists).expectNext(true).verifyComplete();
}
@Test // DATACASS-485
public void existsShouldReturnFalseIfNoElementExistsInCollection() {
StepVerifier.create(this.template.truncate(Person.class)).verifyComplete();
Mono<Boolean> exists = this.template.query(Person.class).exists();
StepVerifier.create(exists).expectNext(false).verifyComplete();
}
@Test // DATACASS-485
public void existsShouldReturnTrueIfAtLeastOneElementMatchesQuery() {
Mono<Boolean> exists = this.template.query(Person.class).matching(queryLuke()).exists();
StepVerifier.create(exists).expectNext(true).verifyComplete();
}
@Test // DATACASS-485
public void existsShouldReturnFalseWhenNoElementMatchesQuery() {
Mono<Boolean> exists = this.template.query(Person.class).matching(querySpock()).exists();
StepVerifier.create(exists).expectNext(false).verifyComplete();
}
@Test // DATACASS-485
public void returnsTargetObjectDirectlyIfProjectionInterfaceIsImplemented() {
Flux<Contact> result = this.template.query(Person.class).as(Contact.class).all();
StepVerifier.create(result.collectList()).assertNext(actual ->
assertThat(actual).allMatch(it -> it instanceof Person)
).verifyComplete();
}
private static Query queryLuke() {
return query(where("firstname").is("luke")).withAllowFiltering();
}
private static Query querySpock() {
return query(where("firstname").is("spock")).withAllowFiltering();
}
interface Contact {}
@Data
@Table
static class Person implements Contact {
@Id String id;
@Indexed String firstname;
@Indexed String lastname;
}
interface PersonProjection {
String getFirstname();
}
public interface PersonSpELProjection {
@Value("#{target.firstname}")
String getName();
}
@Data
static class Human {
@Id String id;
}
@Data
@AllArgsConstructor
@NoArgsConstructor
static class Jedi {
@Column("firstname") String name;
}
@Data
static class Sith {
String rank;
}
interface PlanetProjection {
String getName();
}
interface PlanetSpELProjection {
@Value("#{target.name}")
String getId();
}
}

44
spring-data-cassandra/src/test/java/org/springframework/data/cassandra/core/ReactiveUpdateOperationSupportTests.java
View File

@@ -15,19 +15,21 @@
*/
package org.springframework.data.cassandra.core;
import static org.assertj.core.api.Assertions.*;
import static org.springframework.data.cassandra.core.query.Criteria.*;
import static org.springframework.data.cassandra.core.query.Query.*;
import static org.springframework.data.cassandra.core.query.Update.*;
import lombok.Data;
import reactor.core.publisher.Mono;
import reactor.test.StepVerifier;
import static org.assertj.core.api.Assertions.assertThat;
import static org.springframework.data.cassandra.core.query.Criteria.where;
import static org.springframework.data.cassandra.core.query.Query.query;
import static org.springframework.data.cassandra.core.query.Update.update;
import java.util.Collections;
import lombok.Data;
import reactor.core.publisher.Mono;
import reactor.test.StepVerifier;
import org.junit.Before;
import org.junit.Test;
import org.springframework.data.annotation.Id;
import org.springframework.data.cassandra.core.convert.MappingCassandraConverter;
import org.springframework.data.cassandra.core.cql.CqlIdentifier;
@@ -46,6 +48,7 @@ import org.springframework.data.cassandra.test.util.AbstractKeyspaceCreatingInte
public class ReactiveUpdateOperationSupportTests extends AbstractKeyspaceCreatingIntegrationTest {
CassandraAdminTemplate admin;
ReactiveCassandraTemplate template;
Person han;
@@ -74,24 +77,26 @@ public class ReactiveUpdateOperationSupportTests extends AbstractKeyspaceCreatin
@Test(expected = IllegalArgumentException.class) // DATACASS-485
public void domainTypeIsRequired() {
template.update(null);
this.template.update(null);
}
@Test(expected = IllegalArgumentException.class) // DATACASS-485
public void queryIsRequired() {
template.update(Person.class).matching(null);
this.template.update(Person.class).matching(null);
}
@Test(expected = IllegalArgumentException.class) // DATACASS-485
public void tableIsRequiredOnSet() {
template.update(Person.class).inTable((CqlIdentifier) null);
this.template.update(Person.class).inTable((CqlIdentifier) null);
}
@Test // DATACASS-485
public void updateAllMatching() {
Mono<WriteResult> writeResult = template.update(Person.class).matching(queryHan()).apply(update("firstname", "Han"))
.all();
Mono<WriteResult> writeResult = this.template
.update(Person.class)
.matching(queryHan())
.apply(update("firstname", "Han"));
StepVerifier.create(writeResult.map(WriteResult::wasApplied)).expectNext(true).verifyComplete();
}
@@ -99,12 +104,15 @@ public class ReactiveUpdateOperationSupportTests extends AbstractKeyspaceCreatin
@Test // DATACASS-485
public void updateWithDifferentDomainClassAndCollection() {
Mono<WriteResult> writeResult = template.update(Jedi.class).inTable("person")
.matching(query(where("id").is(han.getId()))).apply(update("name", "Han")).all();
Mono<WriteResult> writeResult = this.template
.update(Jedi.class).inTable("person")
.matching(query(where("id").is(han.getId())))
.apply(update("name", "Han"));
StepVerifier.create(writeResult.map(WriteResult::wasApplied)).expectNext(true).verifyComplete();
assertThat(admin.selectOne(queryHan(), Person.class)).isNotEqualTo(han).hasFieldOrPropertyWithValue("firstname",
"Han");
assertThat(this.admin.selectOne(queryHan(), Person.class))
.isNotEqualTo(han).hasFieldOrPropertyWithValue("firstname", "Han");
}
private Query queryHan() {
@@ -114,14 +122,12 @@ public class ReactiveUpdateOperationSupportTests extends AbstractKeyspaceCreatin
@Data
@Table
static class Person {
@Id String id;
@Indexed String firstname;
}
@Data
static class Jedi {
@Column("firstname") String name;
}
}