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

DATACASS-485 - Reactive Fluent Cassandra API.

Browse Source
This commit is contained in:
Mark Paluch
2018-01-29 12:25:36 +01:00
committed by John Blum
parent 1a3314a650
commit 81326e43a5
15 changed files with 1890 additions and 15 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/ReactiveCassandraOperations.java
View File

@@ -42,7 +42,7 @@ import com.datastax.driver.core.Statement;
* @see Flux
* @see Mono
*/
public interface ReactiveCassandraOperations {
public interface ReactiveCassandraOperations extends ReactiveFluentCassandraOperations {
/**
* Returns the underlying {@link CassandraConverter}.

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

@@ -17,10 +17,11 @@ package org.springframework.data.cassandra.core;
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;
@@ -36,16 +37,19 @@ import org.springframework.data.cassandra.core.cql.QueryOptions;
import org.springframework.data.cassandra.core.cql.ReactiveCqlOperations;
import org.springframework.data.cassandra.core.cql.ReactiveCqlTemplate;
import org.springframework.data.cassandra.core.cql.ReactiveSessionCallback;
import org.springframework.data.cassandra.core.cql.WriteOptions;
import org.springframework.data.cassandra.core.cql.session.DefaultReactiveSessionFactory;
import org.springframework.data.cassandra.core.mapping.CassandraMappingContext;
import org.springframework.data.cassandra.core.mapping.CassandraPersistentEntity;
import org.springframework.data.cassandra.core.mapping.CassandraPersistentProperty;
import org.springframework.data.cassandra.core.query.Query;
import org.springframework.data.mapping.context.MappingContext;
import org.springframework.data.projection.SpelAwareProxyProjectionFactory;
import org.springframework.util.Assert;
import org.springframework.util.ClassUtils;
import com.datastax.driver.core.RegularStatement;
import com.datastax.driver.core.Row;
import com.datastax.driver.core.Session;
import com.datastax.driver.core.SimpleStatement;
import com.datastax.driver.core.Statement;
@@ -83,6 +87,8 @@ public class ReactiveCassandraTemplate implements ReactiveCassandraOperations {
private final StatementFactory statementFactory;
private final SpelAwareProxyProjectionFactory projectionFactory;
/**
* Creates an instance of {@link ReactiveCassandraTemplate} initialized with the given {@link ReactiveSession} and a
* default {@link MappingCassandraConverter}.
@@ -128,6 +134,7 @@ public class ReactiveCassandraTemplate implements ReactiveCassandraOperations {
this.cqlOperations = new ReactiveCqlTemplate(sessionFactory);
this.mappingContext = this.converter.getMappingContext();
this.statementFactory = new StatementFactory(new QueryMapper(converter), new UpdateMapper(converter));
this.projectionFactory = new SpelAwareProxyProjectionFactory();
}
/**
@@ -150,6 +157,7 @@ public class ReactiveCassandraTemplate implements ReactiveCassandraOperations {
this.cqlOperations = reactiveCqlOperations;
this.mappingContext = this.converter.getMappingContext();
this.statementFactory = new StatementFactory(new QueryMapper(converter), new UpdateMapper(converter));
this.projectionFactory = new SpelAwareProxyProjectionFactory();
}
/*
@@ -212,7 +220,7 @@ public class ReactiveCassandraTemplate implements ReactiveCassandraOperations {
return getRequiredPersistentEntity(entity).getTableName();
}
private CqlIdentifier getTableName(Class<?> entityType) {
CqlIdentifier getTableName(Class<?> entityType) {
return getRequiredPersistentEntity(entityType).getTableName();
}
@@ -243,8 +251,7 @@ public class ReactiveCassandraTemplate implements ReactiveCassandraOperations {
// Methods dealing with com.datastax.driver.core.Statement
// -------------------------------------------------------------------------
/*
* (non-Javadoc)
/* (non-Javadoc)
* @see org.springframework.data.cassandra.core.ReactiveCassandraOperations#select(com.datastax.driver.core.Statement, java.lang.Class)
*/
@Override
@@ -253,7 +260,9 @@ public class ReactiveCassandraTemplate implements ReactiveCassandraOperations {
Assert.notNull(cql, "Statement must not be null");
Assert.notNull(entityClass, "Entity type must not be null");
return getReactiveCqlOperations().query(cql, (row, rowNum) -> getConverter().read(entityClass, row));
Function<Row, T> mapper = getMapper(entityClass, entityClass);
return getReactiveCqlOperations().query(cql, (row, rowNum) -> mapper.apply(row));
}
/* (non-Javadoc)
@@ -277,8 +286,16 @@ public class ReactiveCassandraTemplate implements ReactiveCassandraOperations {
Assert.notNull(query, "Query must not be null");
Assert.notNull(entityClass, "Entity type must not be null");
return select(getStatementFactory().select(query, getMappingContext().getRequiredPersistentEntity(entityClass)),
entityClass);
return doSelect(query, entityClass, getTableName(entityClass), entityClass);
}
<T> Flux<T> doSelect(Query query, Class<?> entityClass, CqlIdentifier tableName, Class<T> returnType) {
Function<Row, T> mapper = getMapper(entityClass, returnType);
RegularStatement select = getStatementFactory().select(query,
getMappingContext().getRequiredPersistentEntity(entityClass), tableName);
return getReactiveCqlOperations().query(select, (row, rowNum) -> mapper.apply(row));
}
/* (non-Javadoc)
@@ -305,8 +322,15 @@ public class ReactiveCassandraTemplate implements ReactiveCassandraOperations {
Assert.notNull(update, "Update must not be null");
Assert.notNull(entityClass, "Entity type must not be null");
return getReactiveCqlOperations().execute(
getStatementFactory().update(query, update, getMappingContext().getRequiredPersistentEntity(entityClass)));
return doUpdate(query, update, entityClass, getTableName(entityClass)).map(WriteResult::wasApplied);
}
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);
return getReactiveCqlOperations().execute(new StatementCallback(statement)).next();
}
/* (non-Javadoc)
@@ -318,8 +342,15 @@ public class ReactiveCassandraTemplate implements ReactiveCassandraOperations {
Assert.notNull(query, "Query must not be null");
Assert.notNull(entityClass, "Entity type must not be null");
return doDelete(query, entityClass, getTableName(entityClass)).map(WriteResult::wasApplied);
}
Mono<WriteResult> doDelete(Query query, Class<?> entityClass, CqlIdentifier tableName) {
RegularStatement delete = getStatementFactory().delete(query, getRequiredPersistentEntity(entityClass), tableName);
return getReactiveCqlOperations()
.execute(getStatementFactory().delete(query, getMappingContext().getRequiredPersistentEntity(entityClass)));
.execute(new StatementCallback(delete)).next();
}
// -------------------------------------------------------------------------
@@ -350,7 +381,14 @@ public class ReactiveCassandraTemplate implements ReactiveCassandraOperations {
RegularStatement count = getStatementFactory().count(query, getRequiredPersistentEntity(entityClass));
return getReactiveCqlOperations().queryForObject(count, Long.class);
return doCount(query, entityClass, getTableName(entityClass));
}
Mono<Long> doCount(Query query, Class<?> entityClass, CqlIdentifier tableName) {
RegularStatement count = getStatementFactory().count(query, getRequiredPersistentEntity(entityClass), tableName);
return getReactiveCqlOperations().queryForObject(count, Long.class).switchIfEmpty(Mono.just(0L));
}
/* (non-Javadoc)
@@ -380,8 +418,13 @@ public class ReactiveCassandraTemplate implements ReactiveCassandraOperations {
Assert.notNull(query, "Query must not be null");
Assert.notNull(entityClass, "Entity type must not be null");
RegularStatement select = getStatementFactory()
.select(query.limit(1), getRequiredPersistentEntity(entityClass));
return doExists(query, entityClass, getTableName(entityClass));
}
Mono<Boolean> doExists(Query query, Class<?> entityClass, CqlIdentifier tableName) {
RegularStatement select = getStatementFactory().select(query.limit(1), getRequiredPersistentEntity(entityClass),
tableName);
return getReactiveCqlOperations().queryForRows(select).hasElements();
}
@@ -421,8 +464,15 @@ public class ReactiveCassandraTemplate implements ReactiveCassandraOperations {
Assert.notNull(entity, "Entity must not be null");
Assert.notNull(options, "InsertOptions must not be null");
Insert insert = QueryUtils.createInsertQuery(getTableName(entity).toCql(), entity, options, getConverter());
CqlIdentifier tableName = getTableName(entity);
return doInsert(entity, options, tableName);
}
Mono<WriteResult> doInsert(Object entity, WriteOptions options, CqlIdentifier tableName) {
Insert insert = QueryUtils.createInsertQuery(tableName.toCql(), entity, options, getConverter());
// noinspection ConstantConditions
return getReactiveCqlOperations().execute(new StatementCallback(insert)).next();
}
@@ -501,6 +551,59 @@ public class ReactiveCassandraTemplate implements ReactiveCassandraOperations {
return getReactiveCqlOperations().execute(truncate).then();
}
// -------------------------------------------------------------------------
// Fluent API entry points
// -------------------------------------------------------------------------
/* (non-Javadoc)
* @see org.springframework.data.cassandra.core.ReactiveSelectOperation#query(java.lang.Class)
*/
@Override
public <T> ReactiveSelect<T> query(Class<T> domainType) {
return new ReactiveSelectOperationSupport(this).query(domainType);
}
/* (non-Javadoc)
* @see org.springframework.data.cassandra.core.ReactiveInsertOperation#insert(java.lang.Class)
*/
@Override
public <T> ReactiveInsert<T> insert(Class<T> domainType) {
return new ReactiveInsertOperationSupport(this).insert(domainType);
}
/* (non-Javadoc)
* @see org.springframework.data.cassandra.core.ReactiveUpdateOperation#update(java.lang.Class)
*/
@Override
public <T> ReactiveUpdate<T> update(Class<T> domainType) {
return new ReactiveUpdateOperationSupport(this).update(domainType);
}
/* (non-Javadoc)
* @see org.springframework.data.cassandra.core.ReactiveDeleteOperation#remove(java.lang.Class)
*/
@Override
public ReactiveDelete delete(Class<?> domainType) {
return new ReactiveDeleteOperationSupport(this).delete(domainType);
}
// -------------------------------------------------------------------------
// Implementation hooks and helper methods
// -------------------------------------------------------------------------
@SuppressWarnings("unchecked")
private <T> Function<Row, T> getMapper(Class<?> entityType, Class<T> targetType) {
Class<?> typeToRead = targetType.isInterface() || targetType.isAssignableFrom(entityType) ? entityType : targetType;
return row -> {
Object source = getConverter().read(typeToRead, row);
return (T) (targetType.isInterface() ? projectionFactory.createProjection(targetType, source) : source);
};
}
@Value
static class StatementCallback implements ReactiveSessionCallback<WriteResult>, CqlProvider {

109
spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/ReactiveDeleteOperation.java Normal file
View File

@@ -0,0 +1,109 @@
/*
* Copyright 2018 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.data.cassandra.core;
import reactor.core.publisher.Mono;
import org.springframework.data.cassandra.core.cql.CqlIdentifier;
import org.springframework.data.cassandra.core.query.Query;
/**
* {@link ReactiveDeleteOperation} 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.
*
* <pre>
* <code>
* delete(Jedi.class)
* .inTable("star_wars")
* .matching(query(where("firstname").is("luke")))
* .all();
* </code>
* </pre>
*
* @author Mark Paluch
* @since 2.1
*/
public interface ReactiveDeleteOperation {
/**
* Start creating a {@code DELETE} operation for the given {@literal domainType}.
*
* @param domainType must not be {@literal null}.
* @return new instance of {@link ReactiveDelete}.
* @throws IllegalArgumentException if domainType is {@literal null}.
*/
ReactiveDelete delete(Class<?> domainType);
/**
* Table override (optional).
*/
interface DeleteWithTable {
/**
* 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 DeleteWithTable}.
* @throws IllegalArgumentException if {@code table} is {@literal null} or empty.
*/
DeleteWithQuery 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 DeleteWithTable}.
* @throws IllegalArgumentException if {@link CqlIdentifier} is {@literal null}.
*/
DeleteWithQuery inTable(CqlIdentifier table);
}
interface TerminatingDelete {
/**
* Remove all matching rows.
*
* @return the {@link WriteResult}. Never {@literal null}.
*/
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.
*/
interface ReactiveDelete extends DeleteWithTable, DeleteWithQuery {}
}

107
spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/ReactiveDeleteOperationSupport.java Normal file
View File

@@ -0,0 +1,107 @@
/*
* Copyright 2018 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.data.cassandra.core;
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;
import org.springframework.data.cassandra.core.query.Query;
import org.springframework.lang.Nullable;
import org.springframework.util.Assert;
/**
* Implementation of {@link ReactiveDeleteOperation}.
*
* @author Mark Paluch
* @since 2.1
*/
@RequiredArgsConstructor
class ReactiveDeleteOperationSupport implements ReactiveDeleteOperation {
private final @NonNull ReactiveCassandraTemplate template;
/* (non-Javadoc)
* @see org.springframework.data.cassandra.core.ReactiveDeleteOperation#remove(java.lang.Class)
*/
@Override
public ReactiveDelete delete(Class<?> domainType) {
Assert.notNull(domainType, "DomainType must not be null!");
return new ReactiveDeleteSupport(template, domainType, Query.empty(), null);
}
@RequiredArgsConstructor
@FieldDefaults(level = AccessLevel.PRIVATE, makeFinal = true)
static class ReactiveDeleteSupport implements ReactiveDelete, DeleteWithTable, TerminatingDelete {
@NonNull ReactiveCassandraTemplate template;
@NonNull Class<?> domainType;
@NonNull Query query;
@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)
*/
@Override
public DeleteWithQuery inTable(CqlIdentifier tableName) {
Assert.notNull(tableName, "Table name must not be null");
return new ReactiveDeleteSupport(template, domainType, query, tableName);
}
/* (non-Javadoc)
* @see org.springframework.data.cassandra.core.ReactiveDeleteOperation.DeleteWithQuery#matching(org.springframework.data.cassandra.core.query.Query)
*/
@Override
public TerminatingDelete matching(Query query) {
Assert.notNull(query, "Query must not be null!");
return new ReactiveDeleteSupport(template, domainType, query, tableName);
}
/* (non-Javadoc)
* @see org.springframework.data.cassandra.core.ReactiveDeleteOperation.TerminatingDelete#all()
*/
public Mono<WriteResult> all() {
return template.doDelete(query, domainType, getTableName());
}
private CqlIdentifier getTableName() {
return tableName != null ? tableName : template.getTableName(domainType);
}
}
}

26
spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/ReactiveFluentCassandraOperations.java Normal file
View File

@@ -0,0 +1,26 @@
/*
* Copyright 2018 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.data.cassandra.core;
/**
* Stripped down interface providing access to a fluent API that specifies a basic set of reactive Cassandra operations.
*
* @author Mark Paluch
* @since 2.1
* @see ReactiveCassandraOperations
*/
public interface ReactiveFluentCassandraOperations
extends ReactiveSelectOperation, ReactiveInsertOperation, ReactiveUpdateOperation, ReactiveDeleteOperation {}

113
spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/ReactiveInsertOperation.java Normal file
View File

@@ -0,0 +1,113 @@
/*
* Copyright 2018 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.data.cassandra.core;
import reactor.core.publisher.Mono;
import org.springframework.data.cassandra.core.cql.CqlIdentifier;
/**
* {@link ReactiveInsertOperation} allows creation and execution of Cassandra {@code INSERT} 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.
*
* <pre>
* <code>
* insert(Jedi.class)
* .inTable("star_wars")
* .one(luke);
* </code>
* </pre>
*
* @author Mark Paluch
* @since 2.1
*/
public interface ReactiveInsertOperation {
/**
* Start creating an {@code INSERT} operation for given {@literal domainType}.
*
* @param domainType must not be {@literal null}.
* @return new instance of {@link ReactiveInsert}.
* @throws IllegalArgumentException if domainType is {@literal null}.
*/
<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).
*/
interface InsertWithTable<T> extends InsertWithOptions<T> {
/**
* Explicitly set the name of the table.
* <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 TerminatingInsert}.
* @throws IllegalArgumentException if {@code table} is {@literal null} or empty.
*/
InsertWithOptions<T> inTable(String table);
/**
* Explicitly set the name of the table.
* <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 TerminatingInsert}.
* @throws IllegalArgumentException if {@link CqlIdentifier} is {@literal null}.
*/
InsertWithOptions<T> inTable(CqlIdentifier table);
}
/**
* Apply {@link InsertOptions} (optional).
*/
interface InsertWithOptions<T> extends TerminatingInsert<T> {
/**
* Set insert options.
*
* @param insertOptions insertOptions not be {@literal null}.
* @return new instance of {@link TerminatingInsert}.
* @throws IllegalArgumentException if {@link InsertOptions} is {@literal null}.
*/
TerminatingInsert<T> withOptions(InsertOptions insertOptions);
}
/**
* {@link ReactiveInsert} provides methods for constructing {@code INSERT} operations in a fluent way.
*/
interface ReactiveInsert<T> extends TerminatingInsert<T>, InsertWithTable<T>, InsertWithOptions<T> {}
}

110
spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/ReactiveInsertOperationSupport.java Normal file
View File

@@ -0,0 +1,110 @@
/*
* Copyright 2018 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.data.cassandra.core;
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;
import org.springframework.lang.Nullable;
import org.springframework.util.Assert;
/**
* Implementation of {@link ReactiveInsertOperation}.
*
* @author Mark Paluch
* @since 2.1
*/
@RequiredArgsConstructor
class ReactiveInsertOperationSupport implements ReactiveInsertOperation {
private final @NonNull ReactiveCassandraTemplate template;
/* (non-Javadoc)
* @see org.springframework.data.cassandra.core.ReactiveInsertOperation#insert(java.lang.Class)
*/
@Override
public <T> ReactiveInsert<T> insert(Class<T> domainType) {
Assert.notNull(domainType, "DomainType must not be null!");
return new ReactiveInsertSupport<>(template, domainType, null, InsertOptions.empty());
}
@RequiredArgsConstructor
@FieldDefaults(level = AccessLevel.PRIVATE, makeFinal = true)
static class ReactiveInsertSupport<T> implements ReactiveInsert<T> {
@NonNull ReactiveCassandraTemplate template;
@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);
}
/* (non-Javadoc)
* @see org.springframework.data.cassandra.core.ReactiveInsertOperation.InsertWithTable#inTable(org.springframework.data.cassandra.core.cql.CqlIdentifier)
*/
@Override
public InsertWithOptions<T> inTable(CqlIdentifier tableName) {
Assert.notNull(tableName, "Table name must not be null");
return new ReactiveInsertSupport<>(template, domainType, tableName, insertOptions);
}
/* (non-Javadoc)
* @see org.springframework.data.cassandra.core.ReactiveInsertOperation.InsertWithOptions#withOptions(org.springframework.data.cassandra.core.InsertOptions)
*/
@Override
public TerminatingInsert<T> withOptions(InsertOptions insertOptions) {
Assert.notNull(insertOptions, "InsertOptions must not be null");
return new ReactiveInsertSupport<>(template, domainType, tableName, insertOptions);
}
/* (non-Javadoc)
* @see org.springframework.data.cassandra.core.ReactiveInsertOperation.TerminatingInsert#one(java.lang.Object)
*/
@Override
public Mono<WriteResult> one(T object) {
Assert.notNull(object, "Object must not be null!");
return template.doInsert(object, insertOptions, getTableName());
}
private CqlIdentifier getTableName() {
return tableName != null ? tableName : template.getTableName(domainType);
}
}
}

167
spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/ReactiveSelectOperation.java Normal file
View File

@@ -0,0 +1,167 @@
/*
* Copyright 2018 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.data.cassandra.core;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import org.springframework.data.cassandra.core.cql.CqlIdentifier;
import org.springframework.data.cassandra.core.query.Query;
/**
* {@link ReactiveSelectOperation} 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
* 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.
*
* <pre>
* <code>
* query(Human.class)
* .inTable("star_wars")
* .as(Jedi.class)
* .matching(query(where("firstname").is("luke")))
* .all();
* </code>
* </pre>
*
* @author Mark Paluch
* @since 2.1
*/
public interface ReactiveSelectOperation {
/**
* Start creating a {@code SELECT} operation for the given {@literal domainType}.
*
* @param domainType must not be {@literal null}.
* @return new instance of {@link ReactiveSelect}.
* @throws IllegalArgumentException if domainType is {@literal null}.
*/
<T> ReactiveSelect<T> query(Class<T> domainType);
/**
* 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}.
*/
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}.
*/
Mono<Boolean> exists();
}
/**
* 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.
*/
interface ReactiveSelect<T> extends SelectWithTable<T>, SelectWithProjection<T> {}
}

172
spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/ReactiveSelectOperationSupport.java Normal file
View File

@@ -0,0 +1,172 @@
/*
* Copyright 2018 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.data.cassandra.core;
import lombok.AccessLevel;
import lombok.NonNull;
import lombok.RequiredArgsConstructor;
import lombok.experimental.FieldDefaults;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import org.springframework.dao.IncorrectResultSizeDataAccessException;
import org.springframework.data.cassandra.core.cql.CqlIdentifier;
import org.springframework.data.cassandra.core.query.Query;
import org.springframework.lang.Nullable;
import org.springframework.util.Assert;
/**
* Implementation of {@link ReactiveSelectOperation}.
*
* @author Mark Paluch
* @since 2.1
*/
@RequiredArgsConstructor
class ReactiveSelectOperationSupport implements ReactiveSelectOperation {
private final @NonNull ReactiveCassandraTemplate template;
/* (non-Javadoc)
* @see org.springframework.data.cassandra.core.ReactiveSelectOperation#query(java.lang.Class)
*/
@Override
public <T> ReactiveSelect<T> query(Class<T> domainType) {
Assert.notNull(domainType, "DomainType must not be null!");
return new ReactiveSelectSupport<>(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> {
@NonNull ReactiveCassandraTemplate template;
@NonNull Class<?> domainType;
@NonNull Class<T> returnType;
@NonNull Query query;
@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!");
return new ReactiveSelectSupport<>(template, domainType, returnType, query, tableName);
}
/* (non-Javadoc)
* @see org.springframework.data.cassandra.core.ReactiveSelectOperation.SelectWithProjection#as(java.lang.Class)
*/
@Override
public <R> SelectWithQuery<R> as(Class<R> returnType) {
Assert.notNull(returnType, "ReturnType must not be null!");
return new ReactiveSelectSupport<>(template, domainType, returnType, query, tableName);
}
/* (non-Javadoc)
* @see org.springframework.data.cassandra.core.ReactiveSelectOperation.SelectWithQuery#matching(org.springframework.data.cassandra.core.query.Query)
*/
@Override
public TerminatingSelect<T> matching(Query query) {
Assert.notNull(query, "Query must not be null!");
return new ReactiveSelectSupport<>(template, domainType, returnType, query, tableName);
}
/* (non-Javadoc)
* @see org.springframework.data.cassandra.core.ReactiveSelectOperation.TerminatingSelect#first()
*/
@Override
public Mono<T> first() {
return template.doSelect(query.limit(1), domainType, getTableName(), returnType).next();
}
/* (non-Javadoc)
* @see org.springframework.data.cassandra.core.ReactiveSelectOperation.TerminatingSelect#one()
*/
@Override
public Mono<T> one() {
Flux<T> result = template.doSelect(query.limit(2), domainType, getTableName(), returnType);
return result.collectList() //
.flatMap(it -> {
if (it.isEmpty()) {
return Mono.empty();
}
if (it.size() > 1) {
return Mono.error(
new IncorrectResultSizeDataAccessException("Query " + query + " returned non unique result.", 1));
}
return Mono.just(it.get(0));
});
}
/* (non-Javadoc)
* @see org.springframework.data.cassandra.core.ReactiveSelectOperation.TerminatingSelect#all()
*/
@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());
}
private CqlIdentifier getTableName() {
return tableName != null ? tableName : template.getTableName(domainType);
}
}
}

133
spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/ReactiveUpdateOperation.java Normal file
View File

@@ -0,0 +1,133 @@
/*
* Copyright 2018 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.data.cassandra.core;
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;
/**
* {@link ReactiveUpdateOperation} allows creation and execution of Cassandra {@code UPDATE} operation 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.
*
* <pre>
* <code>
* update(Jedi.class)
* .inTable("star_wars")
* .matching(query(where("firstname").is("luke")))
* .apply(update("lastname", "skywalker"))
* .all();
* </code>
* </pre>
*
* @author Mark Paluch
* @since 2.1
*/
public interface ReactiveUpdateOperation {
/**
* Start creating an {@code UPDATE} operation for the given {@literal domainType}.
*
* @param domainType must not be {@literal null}.
* @return new instance of {@link ReactiveUpdate}.
* @throws IllegalArgumentException if domainType is {@literal null}.
*/
<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.
*/
interface UpdateWithTable<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 UpdateWithTable}.
* @throws IllegalArgumentException if {@code table} is {@literal null} or empty.
*/
UpdateWithQuery<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 UpdateWithTable}.
* @throws IllegalArgumentException if {@link CqlIdentifier} is {@literal null}.
*/
UpdateWithQuery<T> inTable(CqlIdentifier table);
}
/**
* Define a filter query for the {@link Update}.
*/
interface UpdateWithQuery<T> {
/**
* Filter documents by given {@literal query}.
*
* @param query must not be {@literal null}.
* @return new instance of {@link UpdateWithQuery}.
* @throws IllegalArgumentException if query is {@literal null}.
*/
UpdateWithUpdate<T> matching(Query query);
}
/**
* Trigger update execution by calling one of the terminating methods.
*/
interface TerminatingUpdate<T> {
/**
* Update all matching rows in the table.
*
* @return never {@literal null}.
*/
Mono<WriteResult> all();
}
/**
* {@link ReactiveUpdate} provides methods for constructing {@code UPDATE} operations in a fluent way.
*/
interface ReactiveUpdate<T> extends UpdateWithTable<T>, UpdateWithQuery<T> {}
}

123
spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/ReactiveUpdateOperationSupport.java Normal file
View File

@@ -0,0 +1,123 @@
/*
* Copyright 2018 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.data.cassandra.core;
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;
import org.springframework.data.cassandra.core.query.Query;
import org.springframework.data.cassandra.core.query.Update;
import org.springframework.lang.Nullable;
import org.springframework.util.Assert;
/**
* Implementation of {@link ReactiveUpdateOperation}.
*
* @author Mark Paluch
* @since 2.1
*/
@RequiredArgsConstructor
class ReactiveUpdateOperationSupport implements ReactiveUpdateOperation {
private final @NonNull ReactiveCassandraTemplate template;
/* (non-Javadoc)
* @see org.springframework.data.cassandra.core.ReactiveUpdateOperation#update(java.lang.Class)
*/
@Override
public <T> ReactiveUpdate<T> update(Class<T> domainType) {
Assert.notNull(domainType, "DomainType must not be null!");
return new ReactiveUpdateSupport<>(template, domainType, Query.empty(), null, null);
}
@RequiredArgsConstructor
@FieldDefaults(level = AccessLevel.PRIVATE, makeFinal = true)
static class ReactiveUpdateSupport<T>
implements ReactiveUpdate<T>, UpdateWithTable<T>, UpdateWithQuery<T>, UpdateWithUpdate<T>, TerminatingUpdate<T> {
@NonNull ReactiveCassandraTemplate template;
@NonNull Class<T> domainType;
@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!");
return new ReactiveUpdateSupport<>(template, domainType, query, update, 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) {
Assert.notNull(query, "Query must not be null!");
return new ReactiveUpdateSupport<>(template, domainType, query, update, tableName);
}
/* (non-Javadoc)
* @see org.springframework.data.cassandra.core.ReactiveUpdateOperation.TerminatingUpdate#all()
*/
@Override
public Mono<WriteResult> all() {
return template.doUpdate(query, update, domainType, getTableName());
}
private CqlIdentifier getTableName() {
return tableName != null ? tableName : template.getTableName(domainType);
}
}
}

104
spring-data-cassandra/src/test/java/org/springframework/data/cassandra/core/ReactiveDeleteOperationSupportTests.java Normal file
View File

@@ -0,0 +1,104 @@
/*
* Copyright 2018 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
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 java.util.Collections;
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;
import org.springframework.data.cassandra.core.cql.session.DefaultBridgedReactiveSession;
import org.springframework.data.cassandra.core.mapping.Column;
import org.springframework.data.cassandra.core.mapping.Indexed;
import org.springframework.data.cassandra.core.mapping.Table;
import org.springframework.data.cassandra.core.query.Query;
import org.springframework.data.cassandra.test.util.AbstractKeyspaceCreatingIntegrationTest;
/**
* Integration tests for {@link ExecutableDeleteOperationSupport}.
*
* @author Mark Paluch
*/
public class ReactiveDeleteOperationSupportTests extends AbstractKeyspaceCreatingIntegrationTest {
CassandraAdminTemplate admin;
ReactiveCassandraTemplate template;
Person han;
Person luke;
@Before
public void setUp() {
admin = new CassandraAdminTemplate(session, new MappingCassandraConverter());
template = new ReactiveCassandraTemplate(new DefaultBridgedReactiveSession(session));
admin.dropTable(true, CqlIdentifier.of("person"));
admin.createTable(true, CqlIdentifier.of("person"), ExecutableInsertOperationSupportTests.Person.class,
Collections.emptyMap());
han = new Person();
han.firstname = "han";
han.id = "id-1";
luke = new Person();
luke.firstname = "luke";
luke.id = "id-2";
admin.insert(han);
admin.insert(luke);
}
@Test // DATACASS-485
public void removeAllMatching() {
Mono<WriteResult> writeResult = template.delete(Person.class).matching(query(where("id").is(han.id))).all();
StepVerifier.create(writeResult.map(WriteResult::wasApplied)).expectNext(true).verifyComplete();
}
@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();
StepVerifier.create(writeResult.map(WriteResult::wasApplied)).expectNext(true).verifyComplete();
StepVerifier.create(template.select(Query.empty(), Person.class)).verifyComplete();
}
@Data
@Table
static class Person {
@Id String id;
@Indexed String firstname;
}
@Data
static class Jedi {
@Column("firstname") String name;
}
}

118
spring-data-cassandra/src/test/java/org/springframework/data/cassandra/core/ReactiveInsertOperationSupportTests.java Normal file
View File

@@ -0,0 +1,118 @@
/*
* Copyright 2018 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
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 java.util.Collections;
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;
import org.springframework.data.cassandra.core.cql.session.DefaultBridgedReactiveSession;
import org.springframework.data.cassandra.core.mapping.Indexed;
import org.springframework.data.cassandra.core.mapping.Table;
import org.springframework.data.cassandra.test.util.AbstractKeyspaceCreatingIntegrationTest;
/**
* Integration tests for {@link ReactiveInsertOperationSupport}.
*
* @author Mark Paluch
*/
public class ReactiveInsertOperationSupportTests extends AbstractKeyspaceCreatingIntegrationTest {
CassandraAdminTemplate admin;
ReactiveCassandraTemplate template;
Person han;
Person luke;
@Before
public void setUp() {
admin = new CassandraAdminTemplate(session, new MappingCassandraConverter());
template = new ReactiveCassandraTemplate(new DefaultBridgedReactiveSession(session));
admin.dropTable(true, CqlIdentifier.of("person"));
admin.createTable(true, CqlIdentifier.of("person"), Person.class, Collections.emptyMap());
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() {
han = new Person();
han.firstname = "han";
han.lastname = "solo";
han.id = "id-1";
luke = new Person();
luke.firstname = "luke";
luke.lastname = "skywalker";
luke.id = "id-2";
}
}

363
spring-data-cassandra/src/test/java/org/springframework/data/cassandra/core/ReactiveSelectOperationSupportTests.java Normal file
View File

@@ -0,0 +1,363 @@
/*
* Copyright 2018 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
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 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;
import org.springframework.data.cassandra.core.convert.MappingCassandraConverter;
import org.springframework.data.cassandra.core.cql.CqlIdentifier;
import org.springframework.data.cassandra.core.cql.session.DefaultBridgedReactiveSession;
import org.springframework.data.cassandra.core.mapping.Column;
import org.springframework.data.cassandra.core.mapping.Indexed;
import org.springframework.data.cassandra.core.mapping.Table;
import org.springframework.data.cassandra.core.query.Query;
import org.springframework.data.cassandra.test.util.AbstractKeyspaceCreatingIntegrationTest;
/**
* Integration tests for {@link ExecutableSelectOperationSupport}.
*
* @author Mark Paluch
*/
public class ReactiveSelectOperationSupportTests extends AbstractKeyspaceCreatingIntegrationTest {
CassandraAdminTemplate admin;
ReactiveCassandraTemplate template;
Person han;
Person luke;
@Before
public void setUp() {
admin = new CassandraAdminTemplate(session, new MappingCassandraConverter());
template = new ReactiveCassandraTemplate(new DefaultBridgedReactiveSession(session));
admin.dropTable(true, CqlIdentifier.of("person"));
admin.createTable(true, CqlIdentifier.of("person"), Person.class, Collections.emptyMap());
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();
han.firstname = "han";
han.lastname = "solo";
han.id = "id-1";
luke = new Person();
luke.firstname = "luke";
luke.lastname = "skywalker";
luke.id = "id-2";
admin.insert(han);
admin.insert(luke);
}
}

127
spring-data-cassandra/src/test/java/org/springframework/data/cassandra/core/ReactiveUpdateOperationSupportTests.java Normal file
View File

@@ -0,0 +1,127 @@
/*
* Copyright 2018 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
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 java.util.Collections;
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;
import org.springframework.data.cassandra.core.cql.session.DefaultBridgedReactiveSession;
import org.springframework.data.cassandra.core.mapping.Column;
import org.springframework.data.cassandra.core.mapping.Indexed;
import org.springframework.data.cassandra.core.mapping.Table;
import org.springframework.data.cassandra.core.query.Query;
import org.springframework.data.cassandra.test.util.AbstractKeyspaceCreatingIntegrationTest;
/**
* Integration tests for {@link ReactiveUpdateOperationSupport}.
*
* @author Mark Paluch
*/
public class ReactiveUpdateOperationSupportTests extends AbstractKeyspaceCreatingIntegrationTest {
CassandraAdminTemplate admin;
ReactiveCassandraTemplate template;
Person han;
Person luke;
@Before
public void setUp() {
admin = new CassandraAdminTemplate(session, new MappingCassandraConverter());
template = new ReactiveCassandraTemplate(new DefaultBridgedReactiveSession(session));
admin.dropTable(true, CqlIdentifier.of("person"));
admin.createTable(false, CqlIdentifier.of("person"), Person.class, Collections.emptyMap());
han = new Person();
han.firstname = "han";
han.id = "id-1";
luke = new Person();
luke.firstname = "luke";
luke.id = "id-2";
admin.insert(han);
admin.insert(luke);
}
@Test(expected = IllegalArgumentException.class) // DATACASS-485
public void domainTypeIsRequired() {
template.update(null);
}
@Test(expected = IllegalArgumentException.class) // DATACASS-485
public void queryIsRequired() {
template.update(Person.class).matching(null);
}
@Test(expected = IllegalArgumentException.class) // DATACASS-485
public void tableIsRequiredOnSet() {
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();
StepVerifier.create(writeResult.map(WriteResult::wasApplied)).expectNext(true).verifyComplete();
}
@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();
StepVerifier.create(writeResult.map(WriteResult::wasApplied)).expectNext(true).verifyComplete();
assertThat(admin.selectOne(queryHan(), Person.class)).isNotEqualTo(han).hasFieldOrPropertyWithValue("firstname",
"Han");
}
private Query queryHan() {
return query(where("id").is(han.getId()));
}
@Data
@Table
static class Person {
@Id String id;
@Indexed String firstname;
}
@Data
static class Jedi {
@Column("firstname") String name;
}
}