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

DATACASS-576 - Polish.

Browse Source 
Resolves gh-136.
This commit is contained in:
John Blum
2019-02-28 14:49:10 -08:00
parent 16268f101f
commit 159c678f29
15 changed files with 647 additions and 599 deletions
Download Patch File Download Diff File Expand all files Collapse all files

284
spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/AsyncCassandraTemplate.java
View File

@@ -16,7 +16,6 @@
package org.springframework.data.cassandra.core;
import java.util.List;
import java.util.concurrent.ExecutionException;
import java.util.function.Consumer;
import java.util.function.Function;
@@ -43,9 +42,7 @@ import org.springframework.data.cassandra.core.cql.CqlProvider;
import org.springframework.data.cassandra.core.cql.GuavaListenableFutureAdapter;
import org.springframework.data.cassandra.core.cql.QueryOptions;
import org.springframework.data.cassandra.core.cql.session.DefaultSessionFactory;
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.mapping.event.AfterConvertEvent;
import org.springframework.data.cassandra.core.mapping.event.AfterDeleteEvent;
import org.springframework.data.cassandra.core.mapping.event.AfterLoadEvent;
@@ -54,13 +51,11 @@ import org.springframework.data.cassandra.core.mapping.event.BeforeDeleteEvent;
import org.springframework.data.cassandra.core.mapping.event.BeforeSaveEvent;
import org.springframework.data.cassandra.core.query.Query;
import org.springframework.data.domain.Slice;
import org.springframework.data.mapping.context.MappingContext;
import org.springframework.data.projection.ProjectionFactory;
import org.springframework.data.projection.SpelAwareProxyProjectionFactory;
import org.springframework.lang.Nullable;
import org.springframework.scheduling.annotation.AsyncResult;
import org.springframework.util.Assert;
import org.springframework.util.ClassUtils;
import org.springframework.util.concurrent.ListenableFuture;
import com.datastax.driver.core.RegularStatement;
@@ -100,13 +95,11 @@ public class AsyncCassandraTemplate implements AsyncCassandraOperations, Applica
private final CassandraConverter converter;
private final MappingContext<? extends CassandraPersistentEntity<?>, CassandraPersistentProperty> mappingContext;
private final CqlExceptionTranslator exceptionTranslator;
private final SpelAwareProxyProjectionFactory projectionFactory;
private final EntityOperations entityOperations;
private final EntityOperations operations;
private final SpelAwareProxyProjectionFactory projectionFactory;
private final StatementFactory statementFactory;
@@ -168,14 +161,21 @@ public class AsyncCassandraTemplate implements AsyncCassandraOperations, Applica
Assert.notNull(converter, "CassandraConverter must not be null");
this.converter = converter;
this.mappingContext = converter.getMappingContext();
this.cqlOperations = asyncCqlTemplate;
this.entityOperations = new EntityOperations(converter.getMappingContext());
this.exceptionTranslator = asyncCqlTemplate.getExceptionTranslator();
this.projectionFactory = new SpelAwareProxyProjectionFactory();
this.operations = new EntityOperations(converter.getMappingContext());
this.statementFactory = new StatementFactory(new QueryMapper(converter), new UpdateMapper(converter));
}
/* (non-Javadoc)
* @see org.springframework.context.ApplicationEventPublisherAware#setApplicationEventPublisher(org.springframework.context.ApplicationEventPublisher)
*/
@Override
public void setApplicationEventPublisher(ApplicationEventPublisher applicationEventPublisher) {
this.eventPublisher = applicationEventPublisher;
}
/* (non-Javadoc)
* @see org.springframework.data.cassandra.core.AsyncCassandraOperations#getAsyncCqlOperations()
*/
@@ -192,12 +192,47 @@ public class AsyncCassandraTemplate implements AsyncCassandraOperations, Applica
return this.converter;
}
/* (non-Javadoc)
* @see org.springframework.context.ApplicationEventPublisherAware#setApplicationEventPublisher(org.springframework.context.ApplicationEventPublisher)
/**
* Returns the {@link EntityOperations} used to perform data access operations on an entity
* inside a Cassandra data source.
*
* @return the configured {@link EntityOperations} for this template.
* @see org.springframework.data.cassandra.core.EntityOperations
*/
@Override
public void setApplicationEventPublisher(ApplicationEventPublisher applicationEventPublisher) {
this.eventPublisher = applicationEventPublisher;
protected EntityOperations getEntityOperations() {
return this.entityOperations;
}
/**
* Returns a reference to the configured {@link ProjectionFactory} used by this template to process CQL query
* projections.
*
* @return a reference to the configured {@link ProjectionFactory} used by this template to process CQL query
* projections.
* @see org.springframework.data.projection.SpelAwareProxyProjectionFactory
* @since 2.1
*/
protected SpelAwareProxyProjectionFactory getProjectionFactory() {
return this.projectionFactory;
}
private CassandraPersistentEntity<?> getRequiredPersistentEntity(Class<?> entityType) {
return getEntityOperations().getRequiredPersistentEntity(entityType);
}
/**
* Returns the {@link StatementFactory} used by this template to construct and run Cassandra CQL statements.
*
* @return the {@link StatementFactory} used by this template to construct and run Cassandra CQL statements.
* @see org.springframework.data.cassandra.core.StatementFactory
* @since 2.1
*/
protected StatementFactory getStatementFactory() {
return this.statementFactory;
}
private CqlIdentifier getTableName(Class<?> entityClass) {
return getEntityOperations().getTableName(entityClass);
}
// -------------------------------------------------------------------------
@@ -210,7 +245,7 @@ public class AsyncCassandraTemplate implements AsyncCassandraOperations, Applica
@Override
public <T> ListenableFuture<List<T>> select(String cql, Class<T> entityClass) {
Assert.hasText(cql, "Statement must not be empty");
Assert.hasText(cql, "CQL must not be empty");
return select(new SimpleStatement(cql), entityClass);
}
@@ -222,7 +257,7 @@ public class AsyncCassandraTemplate implements AsyncCassandraOperations, Applica
public <T> ListenableFuture<Void> select(String cql, Consumer<T> entityConsumer, Class<T> entityClass)
throws DataAccessException {
Assert.hasText(cql, "Statement must not be empty");
Assert.hasText(cql, "CQL must not be empty");
Assert.notNull(entityConsumer, "Entity Consumer must not be null");
Assert.notNull(entityClass, "Entity type must not be null");
@@ -235,7 +270,7 @@ public class AsyncCassandraTemplate implements AsyncCassandraOperations, Applica
@Override
public <T> ListenableFuture<T> selectOne(String cql, Class<T> entityClass) {
Assert.hasText(cql, "Statement must not be empty");
Assert.hasText(cql, "CQL must not be empty");
Assert.notNull(entityClass, "Entity type must not be null");
return selectOne(new SimpleStatement(cql), entityClass);
@@ -259,29 +294,12 @@ public class AsyncCassandraTemplate implements AsyncCassandraOperations, Applica
return getAsyncCqlOperations().query(statement, (row, rowNum) -> mapper.apply(row));
}
/* (non-Javadoc)
* @see org.springframework.data.cassandra.core.AsyncCassandraOperations#slice(com.datastax.driver.core.Statement, java.lang.Class)
*/
@Override
public <T> ListenableFuture<Slice<T>> slice(Statement statement, Class<T> entityClass) {
Assert.notNull(statement, "Statement must not be null");
Assert.notNull(entityClass, "Entity type must not be null");
ListenableFuture<ResultSet> resultSet = getAsyncCqlOperations().queryForResultSet(statement);
Function<Row, T> mapper = getMapper(entityClass, entityClass, EntityQueryUtils.getTableName(statement));
return new MappingListenableFutureAdapter<>(resultSet,
rs -> EntityQueryUtils.readSlice(rs, (row, rowNum) -> mapper.apply(row), 0, getEffectiveFetchSize(statement)));
}
/* (non-Javadoc)
* @see org.springframework.data.cassandra.core.AsyncCassandraOperations#select(com.datastax.driver.core.Statement, java.util.function.Consumer, java.lang.Class)
*/
@Override
public <T> ListenableFuture<Void> select(Statement statement, Consumer<T> entityConsumer, Class<T> entityClass)
throws DataAccessException {
throws DataAccessException {
Assert.notNull(statement, "Statement must not be null");
Assert.notNull(entityConsumer, "Entity Consumer must not be empty");
@@ -304,6 +322,23 @@ public class AsyncCassandraTemplate implements AsyncCassandraOperations, Applica
list -> list.stream().findFirst().orElse(null));
}
/* (non-Javadoc)
* @see org.springframework.data.cassandra.core.AsyncCassandraOperations#slice(com.datastax.driver.core.Statement, java.lang.Class)
*/
@Override
public <T> ListenableFuture<Slice<T>> slice(Statement statement, Class<T> entityClass) {
Assert.notNull(statement, "Statement must not be null");
Assert.notNull(entityClass, "Entity type must not be null");
ListenableFuture<ResultSet> resultSet = getAsyncCqlOperations().queryForResultSet(statement);
Function<Row, T> mapper = getMapper(entityClass, entityClass, EntityQueryUtils.getTableName(statement));
return new MappingListenableFutureAdapter<>(resultSet,
rs -> EntityQueryUtils.readSlice(rs, (row, rowNum) -> mapper.apply(row), 0, getEffectiveFetchSize(statement)));
}
// -------------------------------------------------------------------------
// Methods dealing with org.springframework.data.cassandra.core.query.Query
// -------------------------------------------------------------------------
@@ -320,24 +355,12 @@ public class AsyncCassandraTemplate implements AsyncCassandraOperations, Applica
return select(getStatementFactory().select(query, getRequiredPersistentEntity(entityClass)), entityClass);
}
/* (non-Javadoc)
* @see org.springframework.data.cassandra.core.AsyncCassandraOperations#slice(org.springframework.data.cassandra.core.query.Query, java.lang.Class)
*/
@Override
public <T> ListenableFuture<Slice<T>> slice(Query query, Class<T> entityClass) throws DataAccessException {
Assert.notNull(query, "Query must not be null");
Assert.notNull(entityClass, "Entity type must not be null");
return slice(getStatementFactory().select(query, getRequiredPersistentEntity(entityClass)), entityClass);
}
/* (non-Javadoc)
* @see org.springframework.data.cassandra.core.AsyncCassandraOperations#select(org.springframework.data.cassandra.core.query.Query, java.util.function.Consumer, java.lang.Class)
*/
@Override
public <T> ListenableFuture<Void> select(Query query, Consumer<T> entityConsumer, Class<T> entityClass)
throws DataAccessException {
throws DataAccessException {
Assert.notNull(query, "Query must not be null");
Assert.notNull(entityConsumer, "Entity Consumer must not be empty");
@@ -359,6 +382,18 @@ public class AsyncCassandraTemplate implements AsyncCassandraOperations, Applica
return selectOne(getStatementFactory().select(query, getRequiredPersistentEntity(entityClass)), entityClass);
}
/* (non-Javadoc)
* @see org.springframework.data.cassandra.core.AsyncCassandraOperations#slice(org.springframework.data.cassandra.core.query.Query, java.lang.Class)
*/
@Override
public <T> ListenableFuture<Slice<T>> slice(Query query, Class<T> entityClass) throws DataAccessException {
Assert.notNull(query, "Query must not be null");
Assert.notNull(entityClass, "Entity type must not be null");
return slice(getStatementFactory().select(query, getRequiredPersistentEntity(entityClass)), entityClass);
}
/* (non-Javadoc)
* @see org.springframework.data.cassandra.core.AsyncCassandraOperations#update(org.springframework.data.cassandra.core.query.Query, org.springframework.data.cassandra.core.query.Update, java.lang.Class)
*/
@@ -445,6 +480,7 @@ public class AsyncCassandraTemplate implements AsyncCassandraOperations, Applica
CassandraPersistentEntity<?> entity = getRequiredPersistentEntity(entityClass);
Select select = QueryBuilder.select().from(entity.getTableName().toCql());
getConverter().write(id, select.where(), entity);
return new MappingListenableFutureAdapter<>(getAsyncCqlOperations().queryForResultSet(select),
@@ -505,7 +541,7 @@ public class AsyncCassandraTemplate implements AsyncCassandraOperations, Applica
Assert.notNull(entity, "Entity must not be null");
Assert.notNull(options, "InsertOptions must not be null");
AdaptibleEntity<T> source = operations.forEntity(entity, converter.getConversionService());
AdaptibleEntity<T> source = getEntityOperations().forEntity(entity, getConverter().getConversionService());
CassandraPersistentEntity<?> persistentEntity = getRequiredPersistentEntity(entity.getClass());
CqlIdentifier tableName = persistentEntity.getTableName();
@@ -514,27 +550,28 @@ public class AsyncCassandraTemplate implements AsyncCassandraOperations, Applica
Insert insert = EntityQueryUtils.createInsertQuery(tableName.toCql(), entityToUse, options, getConverter(),
persistentEntity);
if (source.isVersionedEntity()) {
return doInsertVersioned(insert.ifNotExists(), entityToUse, source, tableName);
}
return doInsert(insert, entityToUse, source, tableName);
return source.isVersionedEntity()
? doInsertVersioned(insert.ifNotExists(), entityToUse, source, tableName)
: doInsert(insert, entityToUse, source, tableName);
}
private <T> ListenableFuture<EntityWriteResult<T>> doInsertVersioned(Insert insert, T entity,
AdaptibleEntity<T> source, CqlIdentifier tableName) {
return executeSave(entity, tableName, insert, result -> {
if (!result.wasApplied()) {
throw new OptimisticLockingFailureException(
String.format("Cannot insert entity %s with version, %s into table %s as it already exists", entity,
source.getVersion(), tableName));
String.format("Cannot insert entity %s with version %s into table %s as it already exists",
entity, source.getVersion(), tableName));
}
});
}
@SuppressWarnings("unused")
private <T> ListenableFuture<EntityWriteResult<T>> doInsert(Insert insert, T entity, AdaptibleEntity<T> source,
CqlIdentifier tableName) {
return executeSave(entity, tableName, insert);
}
@@ -555,15 +592,32 @@ public class AsyncCassandraTemplate implements AsyncCassandraOperations, Applica
Assert.notNull(entity, "Entity must not be null");
Assert.notNull(options, "UpdateOptions must not be null");
AdaptibleEntity<T> source = getEntityOperations().forEntity(entity, getConverter().getConversionService());
CassandraPersistentEntity<?> persistentEntity = getRequiredPersistentEntity(entity.getClass());
CqlIdentifier tableName = persistentEntity.getTableName();
AdaptibleEntity<T> source = operations.forEntity(entity, converter.getConversionService());
if (source.isVersionedEntity()) {
return doUpdateVersioned(source, options, tableName, persistentEntity);
}
return source.isVersionedEntity()
? doUpdateVersioned(source, options, tableName, persistentEntity)
: doUpdate(entity, options, tableName, persistentEntity);
}
return doUpdate(entity, options, tableName, persistentEntity);
private <T> ListenableFuture<EntityWriteResult<T>> doUpdateVersioned(AdaptibleEntity<T> source, UpdateOptions options,
CqlIdentifier tableName, CassandraPersistentEntity<?> persistentEntity) {
Number previousVersion = source.getVersion();
T entity = source.incrementVersion();
Update update = getStatementFactory().update(entity, options, getConverter(), persistentEntity, tableName);
return executeSave(entity, tableName, source.appendVersionCondition(update, previousVersion), result -> {
if (!result.wasApplied()) {
throw new OptimisticLockingFailureException(
String.format("Cannot save entity %s with version %s to table %s. Has it been modified meanwhile?",
entity, source.getVersion(), tableName));
}
});
}
private <T> ListenableFuture<EntityWriteResult<T>> doUpdate(T entity, UpdateOptions options, CqlIdentifier tableName,
@@ -574,23 +628,6 @@ public class AsyncCassandraTemplate implements AsyncCassandraOperations, Applica
return executeSave(entity, tableName, update);
}
private <T> ListenableFuture<EntityWriteResult<T>> doUpdateVersioned(AdaptibleEntity<T> source, UpdateOptions options,
CqlIdentifier tableName, CassandraPersistentEntity<?> persistentEntity) {
Number previousVersion = source.getVersion();
T entity = source.incrementVersion();
Update update = getStatementFactory().update(entity, options, getConverter(), persistentEntity, tableName);
return executeSave(entity, tableName, source.appendVersionCondition(update, previousVersion), result -> {
if (!result.wasApplied()) {
throw new OptimisticLockingFailureException(
String.format("Cannot save entity %s with version %s to table %s. Has it been modified meanwhile?", entity,
source.getVersion(), tableName));
}
});
}
/* (non-Javadoc)
* @see org.springframework.data.cassandra.core.AsyncCassandraOperations#delete(java.lang.Object)
*/
@@ -608,26 +645,25 @@ public class AsyncCassandraTemplate implements AsyncCassandraOperations, Applica
Assert.notNull(entity, "Entity must not be null");
Assert.notNull(options, "QueryOptions must not be null");
AdaptibleEntity<Object> source = getEntityOperations().forEntity(entity, getConverter().getConversionService());
CassandraPersistentEntity<?> persistentEntity = getRequiredPersistentEntity(entity.getClass());
CqlIdentifier tableName = persistentEntity.getTableName();
AdaptibleEntity<Object> source = operations.forEntity(entity, converter.getConversionService());
Delete delete = getStatementFactory().delete(entity, options, getConverter(), persistentEntity, tableName);
if (source.isVersionedEntity()) {
return doDeleteVersioned(delete, entity, source, tableName);
}
return doDelete(delete, entity, tableName);
return source.isVersionedEntity()
? doDeleteVersioned(delete, entity, source, tableName)
: doDelete(delete, entity, tableName);
}
private ListenableFuture<WriteResult> doDeleteVersioned(Delete delete, Object entity, AdaptibleEntity<Object> source,
CqlIdentifier tableName) {
return executeDelete(entity, tableName, source.appendVersionCondition(delete), result -> {
if (!result.wasApplied()) {
throw new OptimisticLockingFailureException(
String.format("Cannot delete entity %s with version, %s in table %s. Has it been modified meanwhile?",
String.format("Cannot delete entity %s with version %s in table %s. Has it been modified meanwhile?",
entity, source.getVersion(), tableName));
}
});
@@ -680,49 +716,13 @@ public class AsyncCassandraTemplate implements AsyncCassandraOperations, Applica
}
// -------------------------------------------------------------------------
// Implementation hooks and helper methods
// Implementation hooks and utility methods
// -------------------------------------------------------------------------
/**
* Returns the {@link CassandraMappingContext} used by this template to access mapping meta-data in order to store
* (map) object to Cassandra tables.
*
* @return the {@link CassandraMappingContext} used by this template.
* @see org.springframework.data.cassandra.core.mapping.CassandraMappingContext
*/
protected MappingContext<? extends CassandraPersistentEntity<?>, CassandraPersistentProperty> getMappingContext() {
return this.mappingContext;
}
/**
* Returns a reference to the configured {@link ProjectionFactory} used by this template to process CQL query
* projections.
*
* @return a reference to the configured {@link ProjectionFactory} used by this template to process CQL query
* projections.
* @see org.springframework.data.projection.SpelAwareProxyProjectionFactory
* @since 2.1
*/
protected SpelAwareProxyProjectionFactory getProjectionFactory() {
return this.projectionFactory;
}
/**
* Returns the {@link StatementFactory} used by this template to construct and run Cassandra CQL statements.
*
* @return the {@link StatementFactory} used by this template to construct and run Cassandra CQL statements.
* @see org.springframework.data.cassandra.core.StatementFactory
* @since 2.1
*/
protected StatementFactory getStatementFactory() {
return this.statementFactory;
}
private <T> ListenableFuture<EntityWriteResult<T>> executeSave(T entity, CqlIdentifier tableName,
Statement statement) {
return executeSave(entity, tableName, statement, ignore -> {
});
return executeSave(entity, tableName, statement, ignore -> { });
}
private <T> ListenableFuture<EntityWriteResult<T>> executeSave(T entity, CqlIdentifier tableName, Statement statement,
@@ -733,6 +733,7 @@ public class AsyncCassandraTemplate implements AsyncCassandraOperations, Applica
ListenableFuture<ResultSet> result = getAsyncCqlOperations().execute(new AsyncStatementCallback(statement));
return new MappingListenableFutureAdapter<>(result, resultSet -> {
EntityWriteResult<T> writeResult = EntityWriteResult.of(resultSet, entity);
beforeAfterSaveEvent.accept(writeResult);
@@ -762,18 +763,6 @@ public class AsyncCassandraTemplate implements AsyncCassandraOperations, Applica
});
}
private CqlIdentifier getTableName(Class<?> entityClass) {
return operations.getTableName(entityClass);
}
private CqlIdentifier getTableName(Object entity) {
return getRequiredPersistentEntity(entity.getClass()).getTableName();
}
private CassandraPersistentEntity<?> getRequiredPersistentEntity(Class<?> entityType) {
return getMappingContext().getRequiredPersistentEntity(ClassUtils.getUserClass(entityType));
}
private int getConfiguredFetchSize(Session session) {
return session.getCluster().getConfiguration().getQueryOptions().getFetchSize();
}
@@ -795,7 +784,8 @@ public class AsyncCassandraTemplate implements AsyncCassandraOperations, Applica
return getAsyncCqlOperations()
.execute((AsyncSessionCallback<Integer>) session -> AsyncResult.forValue(getConfiguredFetchSize(session)))
.completable().join();
.completable()
.join();
}
@SuppressWarnings("unchecked")
@@ -823,13 +813,6 @@ public class AsyncCassandraTemplate implements AsyncCassandraOperations, Applica
return targetType.isInterface() || targetType.isAssignableFrom(entityType) ? entityType : targetType;
}
private void maybeEmitEvent(ApplicationEvent event) {
if (this.eventPublisher != null) {
this.eventPublisher.publishEvent(event);
}
}
private static MappingCassandraConverter newConverter() {
MappingCassandraConverter converter = new MappingCassandraConverter();
@@ -839,6 +822,13 @@ public class AsyncCassandraTemplate implements AsyncCassandraOperations, Applica
return converter;
}
private void maybeEmitEvent(ApplicationEvent event) {
if (this.eventPublisher != null) {
this.eventPublisher.publishEvent(event);
}
}
static class MappingListenableFutureAdapter<T, S>
extends org.springframework.util.concurrent.ListenableFutureAdapter<T, S> {
@@ -853,7 +843,7 @@ public class AsyncCassandraTemplate implements AsyncCassandraOperations, Applica
* @see org.springframework.util.concurrent.FutureAdapter#adapt(java.lang.Object)
*/
@Override
protected T adapt(@Nullable S adapteeResult) throws ExecutionException {
protected T adapt(@Nullable S adapteeResult) {
return this.mapper.apply(adapteeResult);
}
}
@@ -872,7 +862,7 @@ public class AsyncCassandraTemplate implements AsyncCassandraOperations, Applica
*/
@Override
public ListenableFuture<ResultSet> doInSession(Session session) throws DriverException, DataAccessException {
return new GuavaListenableFutureAdapter<>(session.executeAsync(statement),
return new GuavaListenableFutureAdapter<>(session.executeAsync(this.statement),
e -> e instanceof DriverException
? exceptionTranslator.translate("AsyncStatementCallback", getCql(), (DriverException) e)
: exceptionTranslator.translateExceptionIfPossible(e));

274
spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/CassandraTemplate.java
View File

@@ -15,14 +15,14 @@
*/
package org.springframework.data.cassandra.core;
import lombok.Value;
import java.util.List;
import java.util.function.Consumer;
import java.util.function.Function;
import java.util.stream.Stream;
import java.util.stream.StreamSupport;
import lombok.Value;
import org.springframework.context.ApplicationEvent;
import org.springframework.context.ApplicationEventPublisher;
import org.springframework.context.ApplicationEventPublisherAware;
@@ -43,7 +43,6 @@ import org.springframework.data.cassandra.core.cql.QueryOptions;
import org.springframework.data.cassandra.core.cql.SessionCallback;
import org.springframework.data.cassandra.core.cql.WriteOptions;
import org.springframework.data.cassandra.core.cql.session.DefaultSessionFactory;
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.mapping.event.AfterConvertEvent;
@@ -60,7 +59,6 @@ import org.springframework.data.projection.ProjectionFactory;
import org.springframework.data.projection.SpelAwareProxyProjectionFactory;
import org.springframework.lang.Nullable;
import org.springframework.util.Assert;
import org.springframework.util.ClassUtils;
import com.datastax.driver.core.RegularStatement;
import com.datastax.driver.core.ResultSet;
@@ -96,20 +94,20 @@ import com.datastax.driver.core.querybuilder.Update;
*/
public class CassandraTemplate implements CassandraOperations, ApplicationEventPublisherAware {
private @Nullable ApplicationEventPublisher eventPublisher;
private final CassandraConverter converter;
private final CqlOperations cqlOperations;
private final EntityOperations entityOperations;
private final MappingContext<? extends CassandraPersistentEntity<?>, CassandraPersistentProperty> mappingContext;
private final SpelAwareProxyProjectionFactory projectionFactory;
private final EntityOperations operations;
private final StatementFactory statementFactory;
private @Nullable ApplicationEventPublisher eventPublisher;
/**
* Creates an instance of {@link CassandraTemplate} initialized with the given {@link Session} and a default
* {@link MappingCassandraConverter}.
@@ -167,9 +165,9 @@ public class CassandraTemplate implements CassandraOperations, ApplicationEventP
this.converter = converter;
this.cqlOperations = cqlOperations;
this.entityOperations = new EntityOperations(converter.getMappingContext());
this.mappingContext = converter.getMappingContext();
this.projectionFactory = new SpelAwareProxyProjectionFactory();
this.operations = new EntityOperations(converter.getMappingContext());
this.statementFactory = new StatementFactory(new QueryMapper(converter), new UpdateMapper(converter));
}
@@ -181,6 +179,14 @@ public class CassandraTemplate implements CassandraOperations, ApplicationEventP
return new CassandraBatchTemplate(this);
}
/* (non-Javadoc)
* @see org.springframework.context.ApplicationEventPublisherAware#setApplicationEventPublisher(org.springframework.context.ApplicationEventPublisher)
*/
@Override
public void setApplicationEventPublisher(ApplicationEventPublisher applicationEventPublisher) {
this.eventPublisher = applicationEventPublisher;
}
/* (non-Javadoc)
* @see org.springframework.data.cassandra.core.CassandraOperations#getConverter()
*/
@@ -197,20 +203,51 @@ public class CassandraTemplate implements CassandraOperations, ApplicationEventP
return this.cqlOperations;
}
/**
* Returns the {@link EntityOperations} used to perform data access operations on an entity
* inside a Cassandra data source.
*
* @return the configured {@link EntityOperations} for this template.
* @see org.springframework.data.cassandra.core.EntityOperations
*/
protected EntityOperations getEntityOperations() {
return this.entityOperations;
}
/**
* Returns a reference to the configured {@link ProjectionFactory} used by this template to process CQL query
* projections.
*
* @return a reference to the configured {@link ProjectionFactory} used by this template to process CQL query
* projections.
* @see org.springframework.data.projection.SpelAwareProxyProjectionFactory
* @since 2.1
*/
protected SpelAwareProxyProjectionFactory getProjectionFactory() {
return this.projectionFactory;
}
private CassandraPersistentEntity<?> getRequiredPersistentEntity(Class<?> entityType) {
return getEntityOperations().getRequiredPersistentEntity(entityType);
}
/**
* Returns the {@link StatementFactory} used by this template to construct and run Cassandra CQL statements.
*
* @return the {@link StatementFactory} used by this template to construct and run Cassandra CQL statements.
* @see org.springframework.data.cassandra.core.StatementFactory
* @since 2.1
*/
protected StatementFactory getStatementFactory() {
return this.statementFactory;
}
/* (non-Javadoc)
* @see org.springframework.data.cassandra.core.CassandraOperations#getTableName(java.lang.Class)
*/
@Override
public CqlIdentifier getTableName(Class<?> entityClass) {
return operations.getTableName(entityClass);
}
/* (non-Javadoc)
* @see org.springframework.context.ApplicationEventPublisherAware#setApplicationEventPublisher(org.springframework.context.ApplicationEventPublisher)
*/
@Override
public void setApplicationEventPublisher(ApplicationEventPublisher applicationEventPublisher) {
this.eventPublisher = applicationEventPublisher;
return getEntityOperations().getTableName(entityClass);
}
// -------------------------------------------------------------------------
@@ -223,35 +260,35 @@ public class CassandraTemplate implements CassandraOperations, ApplicationEventP
@Override
public <T> List<T> select(String cql, Class<T> entityClass) {
Assert.hasText(cql, "Statement must not be empty");
Assert.hasText(cql, "CQL must not be empty");
return select(new SimpleStatement(cql), entityClass);
}
/* (non-Javadoc)
* @see org.springframework.data.cassandra.core.CassandraOperations#stream(java.lang.String, java.lang.Class)
*/
@Override
public <T> Stream<T> stream(String cql, Class<T> entityClass) throws DataAccessException {
Assert.hasText(cql, "Statement must not be empty");
Assert.notNull(entityClass, "Entity type must not be null");
return stream(new SimpleStatement(cql), entityClass);
}
/* (non-Javadoc)
* @see org.springframework.data.cassandra.core.CassandraOperations#selectOne(java.lang.String, java.lang.Class)
*/
@Override
public <T> T selectOne(String cql, Class<T> entityClass) {
Assert.hasText(cql, "Statement must not be empty");
Assert.hasText(cql, "CQL must not be empty");
Assert.notNull(entityClass, "Entity type must not be null");
return selectOne(new SimpleStatement(cql), entityClass);
}
/* (non-Javadoc)
* @see org.springframework.data.cassandra.core.CassandraOperations#stream(java.lang.String, java.lang.Class)
*/
@Override
public <T> Stream<T> stream(String cql, Class<T> entityClass) throws DataAccessException {
Assert.hasText(cql, "CQL must not be empty");
Assert.notNull(entityClass, "Entity type must not be null");
return stream(new SimpleStatement(cql), entityClass);
}
// -------------------------------------------------------------------------
// Methods dealing with com.datastax.driver.core.Statement
// -------------------------------------------------------------------------
@@ -270,6 +307,14 @@ public class CassandraTemplate implements CassandraOperations, ApplicationEventP
return getCqlOperations().query(statement, (row, rowNum) -> mapper.apply(row));
}
/* (non-Javadoc)
* @see org.springframework.data.cassandra.core.CassandraOperations#selectOne(com.datastax.driver.core.Statement, java.lang.Class)
*/
@Override
public <T> T selectOne(Statement statement, Class<T> entityClass) {
return select(statement, entityClass).stream().findFirst().orElse(null);
}
/* (non-Javadoc)
* @see org.springframework.data.cassandra.core.CassandraOperations#slice(com.datastax.driver.core.Statement, java.lang.Class)
*/
@@ -302,14 +347,6 @@ public class CassandraTemplate implements CassandraOperations, ApplicationEventP
.map(getMapper(entityClass, entityClass, EntityQueryUtils.getTableName(statement)));
}
/* (non-Javadoc)
* @see org.springframework.data.cassandra.core.CassandraOperations#selectOne(com.datastax.driver.core.Statement, java.lang.Class)
*/
@Override
public <T> T selectOne(Statement statement, Class<T> entityClass) {
return select(statement, entityClass).stream().findFirst().orElse(null);
}
// -------------------------------------------------------------------------
// Methods dealing with org.springframework.data.cassandra.core.query.Query
// -------------------------------------------------------------------------
@@ -328,19 +365,31 @@ public class CassandraTemplate implements CassandraOperations, ApplicationEventP
<T> List<T> doSelect(Query query, Class<?> entityClass, CqlIdentifier tableName, Class<T> returnType) {
Function<Row, T> mapper = getMapper(entityClass, returnType, tableName);
CassandraPersistentEntity<?> persistentEntity = getRequiredPersistentEntity(entityClass);
Columns columns = getStatementFactory().computeColumnsForProjection(query.getColumns(), persistentEntity,
returnType);
Columns columns = getStatementFactory()
.computeColumnsForProjection(query.getColumns(), persistentEntity, returnType);
Query queryToUse = query.columns(columns);
RegularStatement select = getStatementFactory().select(queryToUse, persistentEntity, tableName);
Function<Row, T> mapper = getMapper(entityClass, returnType, tableName);
return getCqlOperations().query(select, (row, rowNum) -> mapper.apply(row));
}
/* (non-Javadoc)
* @see org.springframework.data.cassandra.core.CassandraOperations#selectOne(org.springframework.data.cassandra.core.query.Query, java.lang.Class)
*/
@Override
public <T> T selectOne(Query query, Class<T> entityClass) throws DataAccessException {
List<T> result = select(query, entityClass);
return result.isEmpty() ? null : result.get(0);
}
/* (non-Javadoc)
* @see org.springframework.data.cassandra.core.CassandraOperations#slice(org.springframework.data.cassandra.core.query.Query, java.lang.Class)
*/
@@ -369,23 +418,13 @@ public class CassandraTemplate implements CassandraOperations, ApplicationEventP
<T> Stream<T> doStream(Query query, Class<?> entityClass, CqlIdentifier tableName, Class<T> returnType) {
RegularStatement statement = getStatementFactory().select(query, getRequiredPersistentEntity(entityClass),
tableName);
RegularStatement statement = getStatementFactory()
.select(query, getRequiredPersistentEntity(entityClass), tableName);
ResultSet resultSet = getCqlOperations().queryForResultSet(statement);
return StreamSupport.stream(resultSet.spliterator(), false).map(getMapper(entityClass, returnType, tableName));
}
/* (non-Javadoc)
* @see org.springframework.data.cassandra.core.CassandraOperations#selectOne(org.springframework.data.cassandra.core.query.Query, java.lang.Class)
*/
@Override
public <T> T selectOne(Query query, Class<T> entityClass) throws DataAccessException {
List<T> result = select(query, entityClass);
return result.isEmpty() ? null : result.get(0);
return StreamSupport.stream(resultSet.spliterator(), false)
.map(getMapper(entityClass, returnType, tableName));
}
/* (non-Javadoc)
@@ -399,7 +438,8 @@ public class CassandraTemplate implements CassandraOperations, ApplicationEventP
Assert.notNull(update, "Update must not be null");
Assert.notNull(entityClass, "Entity type must not be null");
Statement updateStatement = getStatementFactory().update(query, update, getRequiredPersistentEntity(entityClass));
Statement updateStatement = getStatementFactory()
.update(query, update, getRequiredPersistentEntity(entityClass));
return getCqlOperations().execute(updateStatement);
}
@@ -408,8 +448,8 @@ public class CassandraTemplate implements CassandraOperations, ApplicationEventP
WriteResult doUpdate(Query query, org.springframework.data.cassandra.core.query.Update update, Class<?> entityClass,
CqlIdentifier tableName) {
RegularStatement updateStatement = getStatementFactory().update(query, update,
getRequiredPersistentEntity(entityClass), tableName);
RegularStatement updateStatement = getStatementFactory()
.update(query, update, getRequiredPersistentEntity(entityClass), tableName);
return getCqlOperations().execute(new StatementCallback(updateStatement));
}
@@ -475,8 +515,8 @@ public class CassandraTemplate implements CassandraOperations, ApplicationEventP
long doCount(Query query, Class<?> entityClass, CqlIdentifier tableName) {
RegularStatement countStatement = getStatementFactory().count(query, getRequiredPersistentEntity(entityClass),
tableName);
RegularStatement countStatement = getStatementFactory()
.count(query, getRequiredPersistentEntity(entityClass), tableName);
Long count = getCqlOperations().queryForObject(countStatement, Long.class);
@@ -515,8 +555,8 @@ public class CassandraTemplate implements CassandraOperations, ApplicationEventP
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 getCqlOperations().queryForResultSet(select).iterator().hasNext();
}
@@ -533,9 +573,11 @@ public class CassandraTemplate implements CassandraOperations, ApplicationEventP
CqlIdentifier tableName = getTableName(entityClass);
Select select = QueryBuilder.select().all().from(tableName.toCql());
getConverter().write(id, select.where(), getRequiredPersistentEntity(entityClass));
Function<Row, T> mapper = getMapper(entityClass, entityClass, tableName);
List<T> result = getCqlOperations().query(select, (row, rowNum) -> mapper.apply(row));
return result.isEmpty() ? null : result.get(0);
@@ -558,12 +600,12 @@ public class CassandraTemplate implements CassandraOperations, ApplicationEventP
Assert.notNull(entity, "Entity must not be null");
Assert.notNull(options, "InsertOptions must not be null");
return doInsert(entity, options, getTableName(entity));
return doInsert(entity, options, getTableName(entity.getClass()));
}
<T> EntityWriteResult<T> doInsert(T entity, WriteOptions options, CqlIdentifier tableName) {
AdaptibleEntity<T> source = operations.forEntity(entity, converter.getConversionService());
AdaptibleEntity<T> source = getEntityOperations().forEntity(entity, getConverter().getConversionService());
CassandraPersistentEntity<?> persistentEntity = getRequiredPersistentEntity(entity.getClass());
T entityToUse = source.isVersionedEntity() ? source.initializeVersionProperty() : entity;
@@ -571,11 +613,9 @@ public class CassandraTemplate implements CassandraOperations, ApplicationEventP
Insert insert = EntityQueryUtils.createInsertQuery(tableName.toCql(), entityToUse, options, getConverter(),
persistentEntity);
if (source.isVersionedEntity()) {
return doInsertVersioned(insert.ifNotExists(), entityToUse, source, tableName);
}
return doInsert(insert, entityToUse, tableName);
return source.isVersionedEntity()
? doInsertVersioned(insert.ifNotExists(), entityToUse, source, tableName)
: doInsert(insert, entityToUse, tableName);
}
private <T> EntityWriteResult<T> doInsertVersioned(Insert insert, T entity, AdaptibleEntity<T> source,
@@ -585,8 +625,8 @@ public class CassandraTemplate implements CassandraOperations, ApplicationEventP
if (!result.wasApplied()) {
throw new OptimisticLockingFailureException(
String.format("Cannot insert entity %s with version, %s into table %s as it already exists", entity,
source.getVersion(), tableName));
String.format("Cannot insert entity %s with version %s into table %s as it already exists",
entity, source.getVersion(), tableName));
}
});
}
@@ -612,21 +652,20 @@ public class CassandraTemplate implements CassandraOperations, ApplicationEventP
Assert.notNull(entity, "Entity must not be null");
Assert.notNull(options, "UpdateOptions must not be null");
AdaptibleEntity<T> source = getEntityOperations().forEntity(entity, getConverter().getConversionService());
CassandraPersistentEntity<?> persistentEntity = getRequiredPersistentEntity(entity.getClass());
CqlIdentifier tableName = persistentEntity.getTableName();
AdaptibleEntity<T> source = operations.forEntity(entity, converter.getConversionService());
if (source.isVersionedEntity()) {
return doUpdateVersioned(source, options, tableName, persistentEntity);
}
return doUpdate(entity, options, tableName, persistentEntity);
return source.isVersionedEntity()
? doUpdateVersioned(source, options, tableName, persistentEntity)
: doUpdate(entity, options, tableName, persistentEntity);
}
private <T> EntityWriteResult<T> doUpdateVersioned(AdaptibleEntity<T> source, UpdateOptions options,
CqlIdentifier tableName, CassandraPersistentEntity<?> persistentEntity) {
Number previousVersion = source.getVersion();
T entity = source.incrementVersion();
Update update = getStatementFactory().update(entity, options, getConverter(), persistentEntity, tableName);
@@ -635,8 +674,8 @@ public class CassandraTemplate implements CassandraOperations, ApplicationEventP
if (!result.wasApplied()) {
throw new OptimisticLockingFailureException(
String.format("Cannot save entity %s with version %s to table %s. Has it been modified meanwhile?", entity,
source.getVersion(), tableName));
String.format("Cannot save entity %s with version %s to table %s. Has it been modified meanwhile?",
entity, source.getVersion(), tableName));
}
});
}
@@ -666,17 +705,15 @@ public class CassandraTemplate implements CassandraOperations, ApplicationEventP
Assert.notNull(entity, "Entity must not be null");
Assert.notNull(options, "QueryOptions must not be null");
AdaptibleEntity<Object> source = getEntityOperations().forEntity(entity, getConverter().getConversionService());
CassandraPersistentEntity<?> persistentEntity = getRequiredPersistentEntity(entity.getClass());
CqlIdentifier tableName = persistentEntity.getTableName();
AdaptibleEntity<Object> source = operations.forEntity(entity, converter.getConversionService());
Delete delete = getStatementFactory().delete(entity, options, getConverter(), persistentEntity, tableName);
if (source.isVersionedEntity()) {
return doDeleteVersioned(delete, entity, source, tableName);
}
return doDelete(delete, entity, tableName);
return source.isVersionedEntity()
? doDeleteVersioned(delete, entity, source, tableName)
: doDelete(delete, entity, tableName);
}
private WriteResult doDeleteVersioned(Delete delete, Object entity, AdaptibleEntity<Object> source,
@@ -686,7 +723,7 @@ public class CassandraTemplate implements CassandraOperations, ApplicationEventP
if (!result.wasApplied()) {
throw new OptimisticLockingFailureException(
String.format("Cannot delete entity %s with version, %s in table %s. Has it been modified meanwhile?",
String.format("Cannot delete entity %s with version %s in table %s. Has it been modified meanwhile?",
entity, source.getVersion(), tableName));
}
});
@@ -706,7 +743,6 @@ public class CassandraTemplate implements CassandraOperations, ApplicationEventP
Assert.notNull(entityClass, "Entity type must not be null");
CassandraPersistentEntity<?> entity = getRequiredPersistentEntity(entityClass);
CqlIdentifier tableName = entity.getTableName();
Delete delete = QueryBuilder.delete().from(tableName.toCql());
@@ -776,44 +812,9 @@ public class CassandraTemplate implements CassandraOperations, ApplicationEventP
}
// -------------------------------------------------------------------------
// Implementation hooks and helper methods
// Implementation hooks and utility methods
// -------------------------------------------------------------------------
/**
* Returns the {@link CassandraMappingContext} used by this template to access mapping meta-data in order to store
* (map) object to Cassandra tables.
*
* @return the {@link CassandraMappingContext} used by this template.
* @see org.springframework.data.cassandra.core.mapping.CassandraMappingContext
*/
protected MappingContext<? extends CassandraPersistentEntity<?>, CassandraPersistentProperty> getMappingContext() {
return this.mappingContext;
}
/**
* Returns a reference to the configured {@link ProjectionFactory} used by this template to process CQL query
* projections.
*
* @return a reference to the configured {@link ProjectionFactory} used by this template to process CQL query
* projections.
* @see org.springframework.data.projection.SpelAwareProxyProjectionFactory
* @since 2.1
*/
protected SpelAwareProxyProjectionFactory getProjectionFactory() {
return this.projectionFactory;
}
/**
* Returns the {@link StatementFactory} used by this template to construct and run Cassandra CQL statements.
*
* @return the {@link StatementFactory} used by this template to construct and run Cassandra CQL statements.
* @see org.springframework.data.cassandra.core.StatementFactory
* @since 2.1
*/
protected StatementFactory getStatementFactory() {
return this.statementFactory;
}
private <T> EntityWriteResult<T> executeSave(T entity, CqlIdentifier tableName, Statement statement) {
return executeSave(entity, tableName, statement, ignore -> {});
}
@@ -837,6 +838,7 @@ public class CassandraTemplate implements CassandraOperations, ApplicationEventP
maybeEmitEvent(new BeforeDeleteEvent<>(statement, entity.getClass(), tableName));
WriteResult result = getCqlOperations().execute(new StatementCallback(statement));
resultConsumer.accept(result);
maybeEmitEvent(new AfterDeleteEvent<>(statement, entity.getClass(), tableName));
@@ -844,14 +846,6 @@ public class CassandraTemplate implements CassandraOperations, ApplicationEventP
return result;
}
private CqlIdentifier getTableName(Object entity) {
return getRequiredPersistentEntity(entity.getClass()).getTableName();
}
private CassandraPersistentEntity<?> getRequiredPersistentEntity(Class<?> entityType) {
return getMappingContext().getRequiredPersistentEntity(ClassUtils.getUserClass(entityType));
}
private int getConfiguredFetchSize(Session session) {
return session.getCluster().getConfiguration().getQueryOptions().getFetchSize();
}
@@ -898,13 +892,6 @@ public class CassandraTemplate implements CassandraOperations, ApplicationEventP
return targetType.isInterface() || targetType.isAssignableFrom(entityType) ? entityType : targetType;
}
private void maybeEmitEvent(ApplicationEvent event) {
if (eventPublisher != null) {
eventPublisher.publishEvent(event);
}
}
private static MappingCassandraConverter newConverter() {
MappingCassandraConverter converter = new MappingCassandraConverter();
@@ -914,6 +901,13 @@ public class CassandraTemplate implements CassandraOperations, ApplicationEventP
return converter;
}
private void maybeEmitEvent(ApplicationEvent event) {
if (this.eventPublisher != null) {
this.eventPublisher.publishEvent(event);
}
}
@Value
static class StatementCallback implements SessionCallback<WriteResult>, CqlProvider {

139
spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/EntityOperations.java
View File

@@ -16,6 +16,7 @@
package org.springframework.data.cassandra.core;
import lombok.AccessLevel;
import lombok.Getter;
import lombok.NonNull;
import lombok.RequiredArgsConstructor;
@@ -36,22 +37,20 @@ import com.datastax.driver.core.querybuilder.QueryBuilder;
import com.datastax.driver.core.querybuilder.Update;
/**
* Common operations performed on an entity in the context of it's mapping metadata.
* Common data access operations performed on an entity using a {@link MappingContext} containing mapping metadata.
*
* @author Mark Paluch
* @since 2.2
* @author John Blum
* @see CassandraTemplate
* @see AsyncCassandraTemplate
* @see ReactiveCassandraTemplate
* @since 2.2
*/
@RequiredArgsConstructor
class EntityOperations {
private final @NonNull MappingContext<? extends CassandraPersistentEntity<?>, CassandraPersistentProperty> context;
private CassandraPersistentEntity<?> getRequiredPersistentEntity(Class<?> entityType) {
return context.getRequiredPersistentEntity(ClassUtils.getUserClass(entityType));
}
@NonNull @Getter(AccessLevel.PROTECTED)
private final MappingContext<? extends CassandraPersistentEntity<?>, CassandraPersistentProperty> mappingContext;
/**
* Creates a new {@link Entity} for the given bean.
@@ -61,19 +60,9 @@ class EntityOperations {
*/
public <T> Entity<T> forEntity(T entity) {
Assert.notNull(entity, "Bean must not be null!");
Assert.notNull(entity, "Bean must not be null");
return MappedEntity.of(entity, context);
}
/**
* Returns the table name to which the entity shall be persisted.
*
* @param entityClass entity class, must not be {@literal null}.
* @return the table name to which the entity shall be persisted.
*/
public CqlIdentifier getTableName(Class<?> entityClass) {
return getRequiredPersistentEntity(entityClass).getTableName();
return MappedEntity.of(entity, getMappingContext());
}
/**
@@ -85,10 +74,31 @@ class EntityOperations {
*/
public <T> AdaptibleEntity<T> forEntity(T entity, ConversionService conversionService) {
Assert.notNull(entity, "Bean must not be null!");
Assert.notNull(conversionService, "ConversionService must not be null!");
Assert.notNull(entity, "Bean must not be null");
Assert.notNull(conversionService, "ConversionService must not be null");
return AdaptibleMappedEntity.of(entity, context, conversionService);
return AdaptibleMappedEntity.of(entity, getMappingContext(), conversionService);
}
/**
* Returns the {@link MappingContext} used by this entity data access operations class to access mapping meta-data
* used to store (map) object to Cassandra tables.
*
* @return the {@link MappingContext} used by this entity data access operations class.
* @see org.springframework.data.cassandra.core.mapping.CassandraMappingContext
*/
CassandraPersistentEntity<?> getRequiredPersistentEntity(Class<?> entityType) {
return getMappingContext().getRequiredPersistentEntity(ClassUtils.getUserClass(entityType));
}
/**
* Returns the table name to which the entity shall be persisted.
*
* @param entityClass entity class, must not be {@literal null}.
* @return the table name to which the entity shall be persisted.
*/
CqlIdentifier getTableName(Class<?> entityClass) {
return getRequiredPersistentEntity(entityClass).getTableName();
}
/**
@@ -194,12 +204,28 @@ class EntityOperations {
return new MappedEntity<>(entity, propertyAccessor);
}
/* (non-Javadoc)
* @see org.springframework.data.cassandra.core.EntityOperations.Entity#getBean()
*/
@Override
public T getBean() {
return this.propertyAccessor.getBean();
}
/* (non-Javadoc)
* @see org.springframework.data.cassandra.core.EntityOperations.Entity#isNew()
*/
@Override
public boolean isNew() {
return this.entity.isNew(getBean());
}
/* (non-Javadoc)
* @see org.springframework.data.cassandra.core.EntityOperations.Entity#isVersionedEntity()
*/
@Override
public boolean isVersionedEntity() {
return entity.hasVersionProperty();
return this.entity.hasVersionProperty();
}
/* (non-Javadoc)
@@ -208,23 +234,7 @@ class EntityOperations {
@Override
@Nullable
public Object getVersion() {
return propertyAccessor.getProperty(entity.getRequiredVersionProperty());
}
/* (non-Javadoc)
* @see org.springframework.data.cassandra.core.EntityOperations.Entity#getBean()
*/
@Override
public T getBean() {
return propertyAccessor.getBean();
}
/* (non-Javadoc)
* @see org.springframework.data.cassandra.core.EntityOperations.Entity#isNew()
*/
@Override
public boolean isNew() {
return entity.isNew(propertyAccessor.getBean());
return this.propertyAccessor.getProperty(this.entity.getRequiredVersionProperty());
}
}
@@ -233,6 +243,18 @@ class EntityOperations {
private final CassandraPersistentEntity<?> entity;
private final ConvertingPropertyAccessor<T> propertyAccessor;
private static <T> AdaptibleEntity<T> of(T bean,
MappingContext<? extends CassandraPersistentEntity<?>, CassandraPersistentProperty> mappingContext,
ConversionService conversionService) {
CassandraPersistentEntity<?> entity = mappingContext.getRequiredPersistentEntity(bean.getClass());
PersistentPropertyAccessor<T> propertyAccessor = entity.getPropertyAccessor(bean);
return new AdaptibleMappedEntity<>(entity,
new ConvertingPropertyAccessor<>(propertyAccessor, conversionService));
}
private AdaptibleMappedEntity(CassandraPersistentEntity<?> entity, ConvertingPropertyAccessor<T> propertyAccessor) {
super(entity, propertyAccessor);
@@ -241,22 +263,13 @@ class EntityOperations {
this.propertyAccessor = propertyAccessor;
}
private static <T> AdaptibleEntity<T> of(T bean,
MappingContext<? extends CassandraPersistentEntity<?>, CassandraPersistentProperty> context,
ConversionService conversionService) {
CassandraPersistentEntity<?> entity = context.getRequiredPersistentEntity(bean.getClass());
PersistentPropertyAccessor<T> propertyAccessor = entity.getPropertyAccessor(bean);
return new AdaptibleMappedEntity<>(entity, new ConvertingPropertyAccessor<>(propertyAccessor, conversionService));
}
/* (non-Javadoc)
* @see org.springframework.data.cassandra.core.EntityOperations.AdaptibleEntity#appendVersionCondition(com.datastax.driver.core.querybuilder.Update, java.lang.Number)
*/
@Override
public Statement appendVersionCondition(com.datastax.driver.core.querybuilder.Update update,
Number currentVersionNumber) {
return update.onlyIf(QueryBuilder.eq(getVersionColumnName().toCql(), currentVersionNumber));
}
@@ -274,15 +287,14 @@ class EntityOperations {
@Override
public T initializeVersionProperty() {
if (!entity.hasVersionProperty()) {
return propertyAccessor.getBean();
if (this.entity.hasVersionProperty()) {
CassandraPersistentProperty versionProperty = this.entity.getRequiredVersionProperty();
this.propertyAccessor.setProperty(versionProperty, versionProperty.getType().isPrimitive() ? 1 : 0);
}
CassandraPersistentProperty versionProperty = entity.getRequiredVersionProperty();
propertyAccessor.setProperty(versionProperty, versionProperty.getType().isPrimitive() ? 1 : 0);
return propertyAccessor.getBean();
return this.propertyAccessor.getBean();
}
/* (non-Javadoc)
@@ -291,13 +303,14 @@ class EntityOperations {
@Override
public T incrementVersion() {
CassandraPersistentProperty versionProperty = entity.getRequiredVersionProperty();
CassandraPersistentProperty versionProperty = this.entity.getRequiredVersionProperty();
Number version = getVersion();
Number nextVersion = version == null ? 0 : version.longValue() + 1;
propertyAccessor.setProperty(versionProperty, nextVersion);
this.propertyAccessor.setProperty(versionProperty, nextVersion);
return propertyAccessor.getBean();
return this.propertyAccessor.getBean();
}
/* (non-Javadoc)
@@ -307,13 +320,13 @@ class EntityOperations {
@Nullable
public Number getVersion() {
CassandraPersistentProperty versionProperty = entity.getRequiredVersionProperty();
CassandraPersistentProperty versionProperty = this.entity.getRequiredVersionProperty();
return propertyAccessor.getProperty(versionProperty, Number.class);
return this.propertyAccessor.getProperty(versionProperty, Number.class);
}
private CqlIdentifier getVersionColumnName() {
return entity.getRequiredVersionProperty().getColumnName();
return this.entity.getRequiredVersionProperty().getColumnName();
}
}
}

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

@@ -49,9 +49,7 @@ import org.springframework.data.cassandra.core.cql.ReactiveSessionCallback;
import org.springframework.data.cassandra.core.cql.RowMapper;
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.mapping.event.AfterConvertEvent;
import org.springframework.data.cassandra.core.mapping.event.AfterDeleteEvent;
import org.springframework.data.cassandra.core.mapping.event.AfterLoadEvent;
@@ -62,12 +60,10 @@ import org.springframework.data.cassandra.core.query.Columns;
import org.springframework.data.cassandra.core.query.Query;
import org.springframework.data.domain.Slice;
import org.springframework.data.domain.SliceImpl;
import org.springframework.data.mapping.context.MappingContext;
import org.springframework.data.projection.ProjectionFactory;
import org.springframework.data.projection.SpelAwareProxyProjectionFactory;
import org.springframework.lang.Nullable;
import org.springframework.util.Assert;
import org.springframework.util.ClassUtils;
import com.datastax.driver.core.RegularStatement;
import com.datastax.driver.core.Row;
@@ -102,20 +98,18 @@ import com.datastax.driver.core.querybuilder.Update;
*/
public class ReactiveCassandraTemplate implements ReactiveCassandraOperations, ApplicationEventPublisherAware {
private @Nullable ApplicationEventPublisher eventPublisher;
private final CassandraConverter converter;
private final EntityOperations entityOperations;
private final ReactiveCqlOperations cqlOperations;
private final MappingContext<? extends CassandraPersistentEntity<?>, CassandraPersistentProperty> mappingContext;
private final SpelAwareProxyProjectionFactory projectionFactory;
private final EntityOperations operations;
private final StatementFactory statementFactory;
private @Nullable ApplicationEventPublisher eventPublisher;
/**
* Creates an instance of {@link ReactiveCassandraTemplate} initialized with the given {@link ReactiveSession} and a
* default {@link MappingCassandraConverter}.
@@ -174,9 +168,8 @@ public class ReactiveCassandraTemplate implements ReactiveCassandraOperations, A
this.converter = converter;
this.cqlOperations = reactiveCqlOperations;
this.mappingContext = this.converter.getMappingContext();
this.entityOperations = new EntityOperations(converter.getMappingContext());
this.projectionFactory = new SpelAwareProxyProjectionFactory();
this.operations = new EntityOperations(converter.getMappingContext());
this.statementFactory = new StatementFactory(new QueryMapper(converter), new UpdateMapper(converter));
}
@@ -188,6 +181,14 @@ public class ReactiveCassandraTemplate implements ReactiveCassandraOperations, A
return new ReactiveCassandraBatchTemplate(this);
}
/* (non-Javadoc)
* @see org.springframework.context.ApplicationEventPublisherAware#setApplicationEventPublisher(org.springframework.context.ApplicationEventPublisher)
*/
@Override
public void setApplicationEventPublisher(ApplicationEventPublisher applicationEventPublisher) {
this.eventPublisher = applicationEventPublisher;
}
/* (non-Javadoc)
* @see org.springframework.data.cassandra.core.ReactiveCassandraOperations#getConverter()
*/
@@ -196,6 +197,30 @@ public class ReactiveCassandraTemplate implements ReactiveCassandraOperations, A
return this.converter;
}
/**
* Returns the {@link EntityOperations} used to perform data access operations on an entity
* inside a Cassandra data source.
*
* @return the configured {@link EntityOperations} for this template.
* @see org.springframework.data.cassandra.core.EntityOperations
*/
protected EntityOperations getEntityOperations() {
return this.entityOperations;
}
/**
* Returns a reference to the configured {@link ProjectionFactory} used by this template to process CQL query
* projections.
*
* @return a reference to the configured {@link ProjectionFactory} used by this template to process CQL query
* projections.
* @see org.springframework.data.projection.SpelAwareProxyProjectionFactory
* @since 2.1
*/
protected SpelAwareProxyProjectionFactory getProjectionFactory() {
return this.projectionFactory;
}
/* (non-Javadoc)
* @see org.springframework.data.cassandra.core.ReactiveCassandraOperations#getReactiveCqlOperations()
*/
@@ -204,12 +229,23 @@ public class ReactiveCassandraTemplate implements ReactiveCassandraOperations, A
return this.cqlOperations;
}
/* (non-Javadoc)
* @see org.springframework.context.ApplicationEventPublisherAware#setApplicationEventPublisher(org.springframework.context.ApplicationEventPublisher)
private CassandraPersistentEntity<?> getRequiredPersistentEntity(Class<?> entityType) {
return getEntityOperations().getRequiredPersistentEntity(entityType);
}
/**
* Returns the {@link StatementFactory} used by this template to construct and run Cassandra CQL statements.
*
* @return the {@link StatementFactory} used by this template to construct and run Cassandra CQL statements.
* @see org.springframework.data.cassandra.core.StatementFactory
* @since 2.1
*/
@Override
public void setApplicationEventPublisher(ApplicationEventPublisher applicationEventPublisher) {
this.eventPublisher = applicationEventPublisher;
protected StatementFactory getStatementFactory() {
return this.statementFactory;
}
CqlIdentifier getTableName(Class<?> entityClass) {
return getEntityOperations().getTableName(entityClass);
}
// -------------------------------------------------------------------------
@@ -222,7 +258,7 @@ public class ReactiveCassandraTemplate implements ReactiveCassandraOperations, A
@Override
public <T> Flux<T> select(String cql, Class<T> entityClass) {
Assert.hasText(cql, "Statement must not be empty");
Assert.hasText(cql, "CQL must not be empty");
return select(new SimpleStatement(cql), entityClass);
}
@@ -243,14 +279,22 @@ public class ReactiveCassandraTemplate implements ReactiveCassandraOperations, A
* @see org.springframework.data.cassandra.core.ReactiveCassandraOperations#select(com.datastax.driver.core.Statement, java.lang.Class)
*/
@Override
public <T> Flux<T> select(Statement cql, Class<T> entityClass) {
public <T> Flux<T> select(Statement statement, Class<T> entityClass) {
Assert.notNull(cql, "Statement must not be null");
Assert.notNull(statement, "Statement must not be null");
Assert.notNull(entityClass, "Entity type must not be null");
Function<Row, T> mapper = getMapper(entityClass, entityClass, EntityQueryUtils.getTableName(cql));
Function<Row, T> mapper = getMapper(entityClass, entityClass, EntityQueryUtils.getTableName(statement));
return getReactiveCqlOperations().query(cql, (row, rowNum) -> mapper.apply(row));
return getReactiveCqlOperations().query(statement, (row, rowNum) -> mapper.apply(row));
}
/* (non-Javadoc)
* @see org.springframework.data.cassandra.core.ReactiveCassandraOperations#selectOne(com.datastax.driver.core.Statement, java.lang.Class)
*/
@Override
public <T> Mono<T> selectOne(Statement statement, Class<T> entityClass) {
return select(statement, entityClass).next();
}
/* (non-Javadoc)
@@ -266,24 +310,17 @@ public class ReactiveCassandraTemplate implements ReactiveCassandraOperations, A
Mono<Integer> effectiveFetchSizeMono = getEffectiveFetchSize(statement);
RowMapper<T> rowMapper = (row, i) -> getConverter().read(entityClass, row);
return resultSetMono.zipWith(effectiveFetchSizeMono).flatMap(tuple -> {
return resultSetMono.zipWith(effectiveFetchSizeMono)
.flatMap(tuple -> {
ReactiveResultSet resultSet = tuple.getT1();
Integer effectiveFetchSize = tuple.getT2();
ReactiveResultSet resultSet = tuple.getT1();
Integer effectiveFetchSize = tuple.getT2();
return resultSet.availableRows().collectList().map(it ->
EntityQueryUtils.readSlice(it, resultSet.getExecutionInfo().getPagingState(), rowMapper, 1,
effectiveFetchSize));
return resultSet.availableRows().collectList().map(it ->
EntityQueryUtils.readSlice(it, resultSet.getExecutionInfo().getPagingState(), rowMapper,
1, effectiveFetchSize));
}).defaultIfEmpty(new SliceImpl<>(Collections.emptyList()));
}
/* (non-Javadoc)
* @see org.springframework.data.cassandra.core.ReactiveCassandraOperations#selectOne(com.datastax.driver.core.Statement, java.lang.Class)
*/
@Override
public <T> Mono<T> selectOne(Statement statement, Class<T> entityClass) {
return select(statement, entityClass).next();
}).defaultIfEmpty(new SliceImpl<>(Collections.emptyList()));
}
// -------------------------------------------------------------------------
@@ -318,6 +355,18 @@ public class ReactiveCassandraTemplate implements ReactiveCassandraOperations, A
return getReactiveCqlOperations().query(select, (row, rowNum) -> mapper.apply(row));
}
/* (non-Javadoc)
* @see org.springframework.data.cassandra.core.ReactiveCassandraOperations#selectOne(org.springframework.data.cassandra.core.query.Query, java.lang.Class)
*/
@Override
public <T> Mono<T> selectOne(Query query, Class<T> entityClass) throws DataAccessException {
Assert.notNull(query, "Query must not be null");
Assert.notNull(entityClass, "Entity type must not be null");
return select(query, entityClass).next();
}
/* (non-Javadoc)
* @see org.springframework.data.cassandra.core.ReactiveCassandraOperations#slice(org.springframework.data.cassandra.core.query.Query, java.lang.Class)
*/
@@ -332,18 +381,6 @@ public class ReactiveCassandraTemplate implements ReactiveCassandraOperations, A
return slice(select, entityClass);
}
/* (non-Javadoc)
* @see org.springframework.data.cassandra.core.ReactiveCassandraOperations#selectOne(org.springframework.data.cassandra.core.query.Query, java.lang.Class)
*/
@Override
public <T> Mono<T> selectOne(Query query, Class<T> entityClass) throws DataAccessException {
Assert.notNull(query, "Query must not be null");
Assert.notNull(entityClass, "Entity type must not be null");
return select(query, entityClass).next();
}
/* (non-Javadoc)
* @see org.springframework.data.cassandra.core.ReactiveCassandraOperations#update(org.springframework.data.cassandra.core.query.Query, org.springframework.data.cassandra.core.query.Update, java.lang.Class)
*/
@@ -499,12 +536,12 @@ public class ReactiveCassandraTemplate implements ReactiveCassandraOperations, A
Assert.notNull(entity, "Entity must not be null");
Assert.notNull(options, "InsertOptions must not be null");
return doInsert(entity, options, getTableName(entity));
return doInsert(entity, options, getTableName(entity.getClass()));
}
<T> Mono<EntityWriteResult<T>> doInsert(T entity, WriteOptions options, CqlIdentifier tableName) {
AdaptibleEntity<T> source = operations.forEntity(entity, converter.getConversionService());
AdaptibleEntity<T> source = this.entityOperations.forEntity(entity, getConverter().getConversionService());
CassandraPersistentEntity<?> persistentEntity = getRequiredPersistentEntity(entity.getClass());
T entityToUse = source.isVersionedEntity() ? source.initializeVersionProperty() : entity;
@@ -512,11 +549,9 @@ public class ReactiveCassandraTemplate implements ReactiveCassandraOperations, A
Insert insert = EntityQueryUtils.createInsertQuery(tableName.toCql(), entityToUse, options, getConverter(),
persistentEntity);
if (source.isVersionedEntity()) {
return doInsertVersioned(insert.ifNotExists(), entityToUse, source, tableName);
}
return doInsert(insert, entityToUse, tableName);
return source.isVersionedEntity()
? doInsertVersioned(insert.ifNotExists(), entityToUse, source, tableName)
: doInsert(insert, entityToUse, tableName);
}
private <T> Mono<EntityWriteResult<T>> doInsertVersioned(Insert insert, T entity, AdaptibleEntity<T> source,
@@ -525,9 +560,11 @@ public class ReactiveCassandraTemplate implements ReactiveCassandraOperations, A
return executeSave(entity, tableName, insert, (result, sink) -> {
if (!result.wasApplied()) {
sink.error(new OptimisticLockingFailureException(
String.format("Cannot insert entity %s with version, %s into table %s as it already exists", entity,
source.getVersion(), tableName)));
String.format("Cannot insert entity %s with version %s into table %s as it already exists",
entity, source.getVersion(), tableName)));
return;
}
@@ -556,21 +593,20 @@ public class ReactiveCassandraTemplate implements ReactiveCassandraOperations, A
Assert.notNull(entity, "Entity must not be null");
Assert.notNull(options, "UpdateOptions must not be null");
AdaptibleEntity<T> source = this.entityOperations.forEntity(entity, getConverter().getConversionService());
CassandraPersistentEntity<?> persistentEntity = getRequiredPersistentEntity(entity.getClass());
CqlIdentifier tableName = persistentEntity.getTableName();
AdaptibleEntity<T> source = operations.forEntity(entity, converter.getConversionService());
if (source.isVersionedEntity()) {
return doUpdateVersioned(source, options, tableName, persistentEntity);
}
return doUpdate(entity, options, tableName, persistentEntity);
return source.isVersionedEntity()
? doUpdateVersioned(source, options, tableName, persistentEntity)
: doUpdate(entity, options, tableName, persistentEntity);
}
private <T> Mono<EntityWriteResult<T>> doUpdateVersioned(AdaptibleEntity<T> source, UpdateOptions options,
CqlIdentifier tableName, CassandraPersistentEntity<?> persistentEntity) {
Number previousVersion = source.getVersion();
T entity = source.incrementVersion();
Update update = getStatementFactory().update(entity, options, getConverter(), persistentEntity, tableName);
@@ -578,9 +614,11 @@ public class ReactiveCassandraTemplate implements ReactiveCassandraOperations, A
return executeSave(entity, tableName, source.appendVersionCondition(update, previousVersion), (result, sink) -> {
if (!result.wasApplied()) {
sink.error(new OptimisticLockingFailureException(
String.format("Cannot save entity %s with version %s to table %s. Has it been modified meanwhile?", entity,
source.getVersion(), tableName)));
String.format("Cannot save entity %s with version %s to table %s. Has it been modified meanwhile?",
entity, source.getVersion(), tableName)));
return;
}
@@ -613,17 +651,15 @@ public class ReactiveCassandraTemplate implements ReactiveCassandraOperations, A
Assert.notNull(entity, "Entity must not be null");
Assert.notNull(options, "QueryOptions must not be null");
AdaptibleEntity<Object> source = this.entityOperations.forEntity(entity, getConverter().getConversionService());
CassandraPersistentEntity<?> persistentEntity = getRequiredPersistentEntity(entity.getClass());
CqlIdentifier tableName = persistentEntity.getTableName();
AdaptibleEntity<Object> source = operations.forEntity(entity, converter.getConversionService());
Delete delete = getStatementFactory().delete(entity, options, getConverter(), persistentEntity, tableName);
if (source.isVersionedEntity()) {
return doDeleteVersioned(delete, entity, source, tableName);
}
return doDelete(delete, entity, tableName);
return source.isVersionedEntity()
? doDeleteVersioned(delete, entity, source, tableName)
: doDelete(delete, entity, tableName);
}
private Mono<WriteResult> doDeleteVersioned(Delete delete, Object entity, AdaptibleEntity<Object> source,
@@ -632,9 +668,11 @@ public class ReactiveCassandraTemplate implements ReactiveCassandraOperations, A
return executeDelete(entity, tableName, source.appendVersionCondition(delete), (result, sink) -> {
if (!result.wasApplied()) {
sink.error(new OptimisticLockingFailureException(
String.format("Cannot delete entity %s with version, %s in table %s. Has it been modified meanwhile?",
String.format("Cannot delete entity %s with version %s in table %s. Has it been modified meanwhile?",
entity, source.getVersion(), tableName)));
return;
}
@@ -689,11 +727,11 @@ public class ReactiveCassandraTemplate implements ReactiveCassandraOperations, A
// -------------------------------------------------------------------------
/* (non-Javadoc)
* @see org.springframework.data.cassandra.core.ReactiveSelectOperation#query(java.lang.Class)
* @see org.springframework.data.cassandra.core.ReactiveDeleteOperation#remove(java.lang.Class)
*/
@Override
public <T> ReactiveSelect<T> query(Class<T> domainType) {
return new ReactiveSelectOperationSupport(this).query(domainType);
public ReactiveDelete delete(Class<?> domainType) {
return new ReactiveDeleteOperationSupport(this).delete(domainType);
}
/* (non-Javadoc)
@@ -704,6 +742,14 @@ public class ReactiveCassandraTemplate implements ReactiveCassandraOperations, A
return new ReactiveInsertOperationSupport(this).insert(domainType);
}
/* (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.ReactiveUpdateOperation#update(java.lang.Class)
*/
@@ -712,57 +758,10 @@ public class ReactiveCassandraTemplate implements ReactiveCassandraOperations, A
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
// Implementation hooks and utility methods
// -------------------------------------------------------------------------
/**
* Returns the {@link CassandraMappingContext} used by this template to access mapping meta-data in order to store
* (map) object to Cassandra tables.
*
* @return the {@link CassandraMappingContext} used by this template.
* @see org.springframework.data.cassandra.core.mapping.CassandraMappingContext
*/
protected MappingContext<? extends CassandraPersistentEntity<?>, CassandraPersistentProperty> getMappingContext() {
return this.mappingContext;
}
/**
* Returns a reference to the configured {@link ProjectionFactory} used by this template to process CQL query
* projections.
*
* @return a reference to the configured {@link ProjectionFactory} used by this template to process CQL query
* projections.
* @see org.springframework.data.projection.SpelAwareProxyProjectionFactory
* @since 2.1
*/
protected SpelAwareProxyProjectionFactory getProjectionFactory() {
return this.projectionFactory;
}
/**
* Returns the {@link StatementFactory} used by this template to construct and run Cassandra CQL statements.
*
* @return the {@link StatementFactory} used by this template to construct and run Cassandra CQL statements.
* @see org.springframework.data.cassandra.core.StatementFactory
* @since 2.1
*/
protected StatementFactory getStatementFactory() {
return this.statementFactory;
}
CqlIdentifier getTableName(Class<?> entityClass) {
return operations.getTableName(entityClass);
}
private <T> Mono<EntityWriteResult<T>> executeSave(T entity, CqlIdentifier tableName, Statement statement) {
return executeSave(entity, tableName, statement, (writeResult, sink) -> sink.next(writeResult));
}
@@ -793,14 +792,6 @@ public class ReactiveCassandraTemplate implements ReactiveCassandraOperations, A
.next();
}
private CqlIdentifier getTableName(Object entity) {
return getRequiredPersistentEntity(entity.getClass()).getTableName();
}
private CassandraPersistentEntity<?> getRequiredPersistentEntity(Class<?> entityType) {
return getMappingContext().getRequiredPersistentEntity(ClassUtils.getUserClass(entityType));
}
@SuppressWarnings("ConstantConditions")
private Mono<Integer> getEffectiveFetchSize(Statement statement) {
@@ -815,8 +806,8 @@ public class ReactiveCassandraTemplate implements ReactiveCassandraOperations, A
}
}
return getReactiveCqlOperations().execute((ReactiveSessionCallback<Integer>) session -> Mono
.just(session.getCluster().getConfiguration().getQueryOptions().getFetchSize())).single();
return getReactiveCqlOperations().execute((ReactiveSessionCallback<Integer>) session ->
Mono.just(session.getCluster().getConfiguration().getQueryOptions().getFetchSize())).single();
}
@SuppressWarnings("unchecked")
@@ -830,7 +821,9 @@ public class ReactiveCassandraTemplate implements ReactiveCassandraOperations, A
Object source = getConverter().read(typeToRead, row);
T result = (T) (targetType.isInterface() ? getProjectionFactory().createProjection(targetType, source) : source);
T result = (T) (targetType.isInterface()
? getProjectionFactory().createProjection(targetType, source)
: source);
maybeEmitEvent(new AfterConvertEvent<>(row, result, tableName));
@@ -842,13 +835,6 @@ public class ReactiveCassandraTemplate implements ReactiveCassandraOperations, A
return targetType.isInterface() || targetType.isAssignableFrom(entityType) ? entityType : targetType;
}
private void maybeEmitEvent(ApplicationEvent event) {
if (this.eventPublisher != null) {
this.eventPublisher.publishEvent(event);
}
}
private static MappingCassandraConverter newConverter() {
MappingCassandraConverter converter = new MappingCassandraConverter();
@@ -858,6 +844,13 @@ public class ReactiveCassandraTemplate implements ReactiveCassandraOperations, A
return converter;
}
private void maybeEmitEvent(ApplicationEvent event) {
if (this.eventPublisher != null) {
this.eventPublisher.publishEvent(event);
}
}
@Value
static class StatementCallback implements ReactiveSessionCallback<WriteResult>, CqlProvider {

13
spring-data-cassandra/src/main/java/org/springframework/data/cassandra/repository/support/MappingCassandraEntityInformation.java
View File

@@ -63,9 +63,10 @@ public class MappingCassandraEntityInformation<T, ID> extends PersistentEntityIn
Assert.notNull(entity, "Entity must not be null");
CassandraPersistentProperty idProperty = entityMetadata.getIdProperty();
CassandraPersistentProperty idProperty = this.entityMetadata.getIdProperty();
return idProperty != null ? (ID) entityMetadata.getIdentifierAccessor(entity).getIdentifier()
return idProperty != null
? (ID) this.entityMetadata.getIdentifierAccessor(entity).getIdentifier()
: (ID) converter.getId(entity, entityMetadata);
}
@@ -76,8 +77,8 @@ public class MappingCassandraEntityInformation<T, ID> extends PersistentEntityIn
@Override
public Class<ID> getIdType() {
if (entityMetadata.getIdProperty() != null) {
return (Class<ID>) entityMetadata.getRequiredIdProperty().getType();
if (this.entityMetadata.getIdProperty() != null) {
return (Class<ID>) this.entityMetadata.getRequiredIdProperty().getType();
}
return (Class<ID>) MapId.class;
@@ -88,7 +89,7 @@ public class MappingCassandraEntityInformation<T, ID> extends PersistentEntityIn
*/
@Override
public String getIdAttribute() {
return entityMetadata.getRequiredIdProperty().getName();
return this.entityMetadata.getRequiredIdProperty().getName();
}
/* (non-Javadoc)
@@ -96,6 +97,6 @@ public class MappingCassandraEntityInformation<T, ID> extends PersistentEntityIn
*/
@Override
public CqlIdentifier getTableName() {
return entityMetadata.getTableName();
return this.entityMetadata.getTableName();
}
}

44
spring-data-cassandra/src/main/java/org/springframework/data/cassandra/repository/support/SimpleCassandraRepository.java
View File

@@ -15,7 +15,7 @@
*/
package org.springframework.data.cassandra.repository.support;
import static org.springframework.data.cassandra.core.query.Criteria.*;
import static org.springframework.data.cassandra.core.query.Criteria.where;
import java.util.ArrayList;
import java.util.List;
@@ -46,18 +46,19 @@ import com.datastax.driver.core.querybuilder.Select;
* @author Alex Shvid
* @author Matthew T. Adams
* @author Mark Paluch
* @author John Blum
* @see org.springframework.data.cassandra.repository.CassandraRepository
*/
public class SimpleCassandraRepository<T, ID> implements CassandraRepository<T, ID> {
private static final InsertOptions INSERT_NULLS = InsertOptions.builder().withInsertNulls().build();
private final AbstractMappingContext<BasicCassandraPersistentEntity<?>, CassandraPersistentProperty> mappingContext;
private final CassandraEntityInformation<T, ID> entityInformation;
private final CassandraOperations operations;
private final AbstractMappingContext<BasicCassandraPersistentEntity<?>, CassandraPersistentProperty> mappingContext;
/**
* Create a new {@link SimpleCassandraRepository} for the given {@link CassandraEntityInformation} and
* {@link CassandraTemplate}.
@@ -83,15 +84,16 @@ public class SimpleCassandraRepository<T, ID> implements CassandraRepository<T,
Assert.notNull(entity, "Entity must not be null");
BasicCassandraPersistentEntity<?> persistentEntity = mappingContext.getPersistentEntity(entity.getClass());
BasicCassandraPersistentEntity<?> persistentEntity = this.mappingContext.getPersistentEntity(entity.getClass());
if (persistentEntity != null && persistentEntity.hasVersionProperty()) {
if (!entityInformation.isNew(entity)) {
return operations.update(entity);
return this.operations.update(entity);
}
}
return operations.insert(entity, INSERT_NULLS).getEntity();
return this.operations.insert(entity, INSERT_NULLS).getEntity();
}
/* (non-Javadoc)
@@ -120,7 +122,7 @@ public class SimpleCassandraRepository<T, ID> implements CassandraRepository<T,
* {@link CassandraOperations#insert(Object, InsertOptions)}.
*/
protected <S extends T> Insert createInsert(S entity) {
return InsertUtil.createInsert(operations.getConverter(), entity);
return InsertUtil.createInsert(this.operations.getConverter(), entity);
}
/* (non-Javadoc)
@@ -131,7 +133,7 @@ public class SimpleCassandraRepository<T, ID> implements CassandraRepository<T,
Assert.notNull(entity, "Entity must not be null");
return operations.insert(entity);
return this.operations.insert(entity);
}
/* (non-Javadoc)
@@ -145,7 +147,7 @@ public class SimpleCassandraRepository<T, ID> implements CassandraRepository<T,
List<S> result = new ArrayList<>();
for (S entity : entities) {
result.add(operations.insert(entity));
result.add(this.operations.insert(entity));
}
return result;
@@ -159,7 +161,7 @@ public class SimpleCassandraRepository<T, ID> implements CassandraRepository<T,
Assert.notNull(id, "The given id must not be null");
return Optional.ofNullable(operations.selectOneById(id, entityInformation.getJavaType()));
return Optional.ofNullable(this.operations.selectOneById(id, this.entityInformation.getJavaType()));
}
/* (non-Javadoc)
@@ -170,7 +172,7 @@ public class SimpleCassandraRepository<T, ID> implements CassandraRepository<T,
Assert.notNull(id, "The given id must not be null");
return operations.exists(id, entityInformation.getJavaType());
return this.operations.exists(id, this.entityInformation.getJavaType());
}
/* (non-Javadoc)
@@ -178,7 +180,7 @@ public class SimpleCassandraRepository<T, ID> implements CassandraRepository<T,
*/
@Override
public long count() {
return operations.count(entityInformation.getJavaType());
return this.operations.count(this.entityInformation.getJavaType());
}
/* (non-Javadoc)
@@ -187,9 +189,9 @@ public class SimpleCassandraRepository<T, ID> implements CassandraRepository<T,
@Override
public List<T> findAll() {
Select select = QueryBuilder.select().all().from(entityInformation.getTableName().toCql());
Select select = QueryBuilder.select().all().from(this.entityInformation.getTableName().toCql());
return operations.select(select, entityInformation.getJavaType());
return this.operations.select(select, this.entityInformation.getJavaType());
}
/* (non-Javadoc)
@@ -202,8 +204,8 @@ public class SimpleCassandraRepository<T, ID> implements CassandraRepository<T,
List<ID> idCollection = Streamable.of(ids).stream().collect(StreamUtils.toUnmodifiableList());
return operations.select(Query.query(where(entityInformation.getIdAttribute()).in(idCollection)),
entityInformation.getJavaType());
return this.operations.select(Query.query(where(this.entityInformation.getIdAttribute()).in(idCollection)),
this.entityInformation.getJavaType());
}
/* (non-Javadoc)
@@ -214,7 +216,7 @@ public class SimpleCassandraRepository<T, ID> implements CassandraRepository<T,
Assert.notNull(pageable, "Pageable must not be null");
return operations.slice(Query.empty().pageRequest(pageable), entityInformation.getJavaType());
return this.operations.slice(Query.empty().pageRequest(pageable), this.entityInformation.getJavaType());
}
/* (non-Javadoc)
@@ -225,7 +227,7 @@ public class SimpleCassandraRepository<T, ID> implements CassandraRepository<T,
Assert.notNull(id, "The given id must not be null");
operations.deleteById(id, entityInformation.getJavaType());
this.operations.deleteById(id, this.entityInformation.getJavaType());
}
/* (non-Javadoc)
@@ -236,7 +238,7 @@ public class SimpleCassandraRepository<T, ID> implements CassandraRepository<T,
Assert.notNull(entity, "The given entity must not be null");
deleteById(entityInformation.getRequiredId(entity));
deleteById(this.entityInformation.getRequiredId(entity));
}
/* (non-Javadoc)
@@ -247,7 +249,7 @@ public class SimpleCassandraRepository<T, ID> implements CassandraRepository<T,
Assert.notNull(entities, "The given Iterable of entities must not be null");
entities.forEach(operations::delete);
entities.forEach(this.operations::delete);
}
/* (non-Javadoc)
@@ -255,6 +257,6 @@ public class SimpleCassandraRepository<T, ID> implements CassandraRepository<T,
*/
@Override
public void deleteAll() {
operations.truncate(entityInformation.getJavaType());
this.operations.truncate(this.entityInformation.getJavaType());
}
}

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

@@ -44,12 +44,12 @@ public class SimpleReactiveCassandraRepository<T, ID> implements ReactiveCassand
private static final InsertOptions INSERT_NULLS = InsertOptions.builder().withInsertNulls().build();
private final AbstractMappingContext<BasicCassandraPersistentEntity<?>, CassandraPersistentProperty> mappingContext;
private final CassandraEntityInformation<T, ID> entityInformation;
private final ReactiveCassandraOperations operations;
private final AbstractMappingContext<BasicCassandraPersistentEntity<?>, CassandraPersistentProperty> mappingContext;
/**
* Create a new {@link SimpleReactiveCassandraRepository} for the given {@link CassandraEntityInformation} and
* {@link ReactiveCassandraOperations}.
@@ -76,15 +76,16 @@ public class SimpleReactiveCassandraRepository<T, ID> implements ReactiveCassand
Assert.notNull(entity, "Entity must not be null");
BasicCassandraPersistentEntity<?> persistentEntity = mappingContext.getPersistentEntity(entity.getClass());
BasicCassandraPersistentEntity<?> persistentEntity = this.mappingContext.getPersistentEntity(entity.getClass());
if (persistentEntity != null && persistentEntity.hasVersionProperty()) {
if (!entityInformation.isNew(entity)) {
return operations.update(entity);
if (!this.entityInformation.isNew(entity)) {
return this.operations.update(entity);
}
}
return operations.insert(entity, INSERT_NULLS).map(EntityWriteResult::getEntity);
return this.operations.insert(entity, INSERT_NULLS).map(EntityWriteResult::getEntity);
}
/**
@@ -96,7 +97,7 @@ public class SimpleReactiveCassandraRepository<T, ID> implements ReactiveCassand
* {@link ReactiveCassandraOperations#insert(Object, InsertOptions)}.
*/
private <S extends T> Insert createInsert(S entity) {
return InsertUtil.createInsert(operations.getConverter(), entity);
return InsertUtil.createInsert(this.operations.getConverter(), entity);
}
/* (non-Javadoc)
@@ -129,7 +130,7 @@ public class SimpleReactiveCassandraRepository<T, ID> implements ReactiveCassand
Assert.notNull(entity, "Entity must not be null");
return operations.insert(entity);
return this.operations.insert(entity);
}
/* (non-Javadoc)
@@ -140,7 +141,7 @@ public class SimpleReactiveCassandraRepository<T, ID> implements ReactiveCassand
Assert.notNull(entities, "The given Iterable of entities must not be null");
return Flux.fromIterable(entities).flatMap(operations::insert);
return Flux.fromIterable(entities).flatMap(this.operations::insert);
}
/* (non-Javadoc)
@@ -151,7 +152,7 @@ public class SimpleReactiveCassandraRepository<T, ID> implements ReactiveCassand
Assert.notNull(entityStream, "The given Publisher of entities must not be null");
return Flux.from(entityStream).flatMap(operations::insert);
return Flux.from(entityStream).flatMap(this.operations::insert);
}
/*
@@ -160,7 +161,7 @@ public class SimpleReactiveCassandraRepository<T, ID> implements ReactiveCassand
*/
@Override
public Mono<Long> count() {
return operations.count(entityInformation.getJavaType());
return this.operations.count(this.entityInformation.getJavaType());
}
/*
@@ -172,7 +173,7 @@ public class SimpleReactiveCassandraRepository<T, ID> implements ReactiveCassand
Assert.notNull(id, "The given id must not be null");
return operations.exists(id, entityInformation.getJavaType());
return this.operations.exists(id, this.entityInformation.getJavaType());
}
/*
@@ -196,7 +197,7 @@ public class SimpleReactiveCassandraRepository<T, ID> implements ReactiveCassand
Assert.notNull(id, "The given id must not be null");
return operations.selectOneById(id, entityInformation.getJavaType());
return this.operations.selectOneById(id, this.entityInformation.getJavaType());
}
/*
@@ -218,9 +219,9 @@ public class SimpleReactiveCassandraRepository<T, ID> implements ReactiveCassand
@Override
public Flux<T> findAll() {
Select select = QueryBuilder.select().from(entityInformation.getTableName().toCql());
Select select = QueryBuilder.select().from(this.entityInformation.getTableName().toCql());
return operations.select(select, entityInformation.getJavaType());
return this.operations.select(select, this.entityInformation.getJavaType());
}
/*
@@ -256,7 +257,7 @@ public class SimpleReactiveCassandraRepository<T, ID> implements ReactiveCassand
Assert.notNull(entity, "The given entity must not be null");
return operations.delete(entity).then();
return this.operations.delete(entity).then();
}
/* (non-Javadoc)
@@ -267,7 +268,7 @@ public class SimpleReactiveCassandraRepository<T, ID> implements ReactiveCassand
Assert.notNull(id, "The given id must not be null");
return operations.deleteById(id, entityInformation.getJavaType()).then();
return this.operations.deleteById(id, this.entityInformation.getJavaType()).then();
}
/* (non-Javadoc)
@@ -286,7 +287,7 @@ public class SimpleReactiveCassandraRepository<T, ID> implements ReactiveCassand
*/
@Override
public Mono<Void> deleteAll() {
return operations.truncate(entityInformation.getJavaType());
return this.operations.truncate(this.entityInformation.getJavaType());
}
/* (non-Javadoc)
@@ -297,7 +298,7 @@ public class SimpleReactiveCassandraRepository<T, ID> implements ReactiveCassand
Assert.notNull(entities, "The given Iterable of entities must not be null");
return Flux.fromIterable(entities).flatMap(operations::delete).then();
return Flux.fromIterable(entities).flatMap(this.operations::delete).then();
}
/* (non-Javadoc)
@@ -308,6 +309,6 @@ public class SimpleReactiveCassandraRepository<T, ID> implements ReactiveCassand
Assert.notNull(entityStream, "The given Publisher of entities must not be null");
return Flux.from(entityStream).flatMap(operations::delete).then();
return Flux.from(entityStream).flatMap(this.operations::delete).then();
}
}

73
spring-data-cassandra/src/test/java/org/springframework/data/cassandra/config/CassandraSessionFactoryBeanUnitTests.java
View File

@@ -15,18 +15,26 @@
*/
package org.springframework.data.cassandra.config;
import static org.assertj.core.api.Assertions.*;
import static org.mockito.ArgumentMatchers.*;
import static org.mockito.Mockito.*;
import static org.springframework.data.cassandra.config.CassandraSessionFactoryBean.*;
import static org.assertj.core.api.Assertions.assertThat;
import static org.mockito.ArgumentMatchers.anyBoolean;
import static org.mockito.ArgumentMatchers.eq;
import static org.mockito.Mockito.doAnswer;
import static org.mockito.Mockito.never;
import static org.mockito.Mockito.spy;
import static org.mockito.Mockito.times;
import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.verifyZeroInteractions;
import static org.mockito.Mockito.when;
import static org.springframework.data.cassandra.config.CassandraSessionFactoryBean.DEFAULT_CREATE_IF_NOT_EXISTS;
import static org.springframework.data.cassandra.config.CassandraSessionFactoryBean.DEFAULT_DROP_TABLES;
import static org.springframework.data.cassandra.config.CassandraSessionFactoryBean.DEFAULT_DROP_UNUSED_TABLES;
import org.junit.Before;
import org.junit.Rule;
import org.junit.Test;
import org.junit.rules.ExpectedException;
import org.junit.runner.RunWith;
import org.mockito.Mock;
import org.mockito.junit.MockitoJUnitRunner;
import org.springframework.data.cassandra.core.convert.CassandraConverter;
import org.springframework.data.cassandra.core.mapping.CassandraMappingContext;
@@ -43,8 +51,6 @@ import com.datastax.driver.core.Session;
@RunWith(MockitoJUnitRunner.class)
public class CassandraSessionFactoryBeanUnitTests {
@Rule public ExpectedException exception = ExpectedException.none();
@Mock CassandraConverter mockConverter;
@Mock Cluster mockCluster;
@Mock Session mockSession;
@@ -83,14 +89,19 @@ public class CassandraSessionFactoryBeanUnitTests {
verify(factoryBean, times(1)).performSchemaAction();
}
@Test // DATACASS-219
@Test(expected = IllegalStateException.class) // DATACASS-219
public void afterPropertiesSetThrowsIllegalStateExceptionWhenConverterIsNull() throws Exception {
exception.expect(IllegalStateException.class);
exception.expectMessage("Converter was not properly initialized");
try {
factoryBean.setCluster(mockCluster);
factoryBean.afterPropertiesSet();
} catch (IllegalStateException expected) {
factoryBean.setCluster(mockCluster);
factoryBean.afterPropertiesSet();
assertThat(expected).hasMessage("Converter was not properly initialized");
assertThat(expected).hasNoCause();
throw expected;
}
}
private void performSchemaActionCallsCreateTableWithArgumentsMatchingTheSchemaAction(SchemaAction schemaAction,
@@ -104,9 +115,11 @@ public class CassandraSessionFactoryBeanUnitTests {
}).when(factoryBean).createTables(anyBoolean(), anyBoolean(), anyBoolean());
factoryBean.setSchemaAction(schemaAction);
assertThat(factoryBean.getSchemaAction()).isEqualTo(schemaAction);
factoryBean.performSchemaAction();
verify(factoryBean, times(1)).createTables(eq(dropTables), eq(dropUnused), eq(ifNotExists));
}
@@ -150,38 +163,56 @@ public class CassandraSessionFactoryBeanUnitTests {
public void setAndGetConverter() {
assertThat(factoryBean.getConverter()).isNull();
factoryBean.setConverter(mockConverter);
assertThat(factoryBean.getConverter()).isEqualTo(mockConverter);
verifyZeroInteractions(mockConverter);
}
@Test // DATACASS-219
@Test(expected = IllegalArgumentException.class) // DATACASS-219
public void setConverterToNull() {
exception.expect(IllegalArgumentException.class);
exception.expectMessage("CassandraConverter must not be null");
try {
factoryBean.setConverter(null);
} catch (IllegalArgumentException expected) {
factoryBean.setConverter(null);
assertThat(expected).hasMessage("CassandraConverter must not be null");
assertThat(expected).hasNoCause();
throw expected;
}
}
@Test // DATACASS-219
public void setAndGetSchemaAction() {
assertThat(factoryBean.getSchemaAction()).isEqualTo(SchemaAction.NONE);
factoryBean.setSchemaAction(SchemaAction.CREATE);
assertThat(factoryBean.getSchemaAction()).isEqualTo(SchemaAction.CREATE);
factoryBean.setSchemaAction(SchemaAction.NONE);
assertThat(factoryBean.getSchemaAction()).isEqualTo(SchemaAction.NONE);
}
@Test // DATACASS-219
@Test(expected = IllegalArgumentException.class) // DATACASS-219
public void setSchemaActionToNullThrowsIllegalArgumentException() {
exception.expect(IllegalArgumentException.class);
exception.expectMessage("SchemaAction must not be null");
try {
factoryBean.setSchemaAction(null);
} catch (IllegalArgumentException expected) {
factoryBean.setSchemaAction(null);
assertThat(expected).hasMessage("SchemaAction must not be null");
assertThat(expected).hasNoCause();
throw expected;
}
}
static class Person {}
}

12
spring-data-cassandra/src/test/java/org/springframework/data/cassandra/core/AsyncOptimisticLockingIntegrationTests.java
View File

@@ -15,15 +15,17 @@
*/
package org.springframework.data.cassandra.core;
import static org.assertj.core.api.Assertions.*;
import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.assertThatThrownBy;
import java.util.concurrent.Future;
import lombok.Data;
import lombok.experimental.Wither;
import java.util.concurrent.Future;
import org.junit.Before;
import org.junit.Test;
import org.springframework.dao.OptimisticLockingFailureException;
import org.springframework.data.annotation.Id;
import org.springframework.data.annotation.PersistenceConstructor;
@@ -47,6 +49,7 @@ public class AsyncOptimisticLockingIntegrationTests extends AbstractKeyspaceCrea
public void setUp() {
MappingCassandraConverter converter = new MappingCassandraConverter();
converter.afterPropertiesSet();
template = new AsyncCassandraTemplate(session, converter);
@@ -110,9 +113,11 @@ public class AsyncOptimisticLockingIntegrationTests extends AbstractKeyspaceCrea
VersionedEntity versionedEntity = new VersionedEntity(42);
VersionedEntity saved = getUninterruptibly(template.insert(versionedEntity));
getUninterruptibly(template.delete(saved));
VersionedEntity loaded = getUninterruptibly(template.selectOne(Query.empty(), VersionedEntity.class));
assertThat(loaded).isNull();
}
@@ -125,6 +130,7 @@ public class AsyncOptimisticLockingIntegrationTests extends AbstractKeyspaceCrea
.hasRootCauseInstanceOf(OptimisticLockingFailureException.class);
VersionedEntity loaded = getUninterruptibly(template.selectOne(Query.empty(), VersionedEntity.class));
assertThat(loaded).isNotNull();
}

9
spring-data-cassandra/src/test/java/org/springframework/data/cassandra/core/OptimisticLockingIntegrationTests.java
View File

@@ -15,13 +15,15 @@
*/
package org.springframework.data.cassandra.core;
import static org.assertj.core.api.Assertions.*;
import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.assertThatThrownBy;
import lombok.Data;
import lombok.experimental.Wither;
import org.junit.Before;
import org.junit.Test;
import org.springframework.dao.OptimisticLockingFailureException;
import org.springframework.data.annotation.Id;
import org.springframework.data.annotation.PersistenceConstructor;
@@ -43,6 +45,7 @@ public class OptimisticLockingIntegrationTests extends AbstractKeyspaceCreatingI
public void setUp() {
MappingCassandraConverter converter = new MappingCassandraConverter();
converter.afterPropertiesSet();
template = new CassandraTemplate(session, converter);
@@ -94,6 +97,7 @@ public class OptimisticLockingIntegrationTests extends AbstractKeyspaceCreatingI
VersionedEntity versionedEntity = new VersionedEntity(42);
template.insert(versionedEntity);
assertThatThrownBy(() -> template.update(new VersionedEntity(42, 5, "f")))
.isInstanceOf(OptimisticLockingFailureException.class);
}
@@ -104,9 +108,11 @@ public class OptimisticLockingIntegrationTests extends AbstractKeyspaceCreatingI
VersionedEntity versionedEntity = new VersionedEntity(42);
VersionedEntity saved = template.insert(versionedEntity);
template.delete(saved);
VersionedEntity loaded = template.query(VersionedEntity.class).firstValue();
assertThat(loaded).isNull();
}
@@ -119,6 +125,7 @@ public class OptimisticLockingIntegrationTests extends AbstractKeyspaceCreatingI
.isInstanceOf(OptimisticLockingFailureException.class);
VersionedEntity loaded = template.query(VersionedEntity.class).firstValue();
assertThat(loaded).isNotNull();
}

29
spring-data-cassandra/src/test/java/org/springframework/data/cassandra/core/ReactiveOptimisticLockingIntegrationTests.java
View File

@@ -15,14 +15,16 @@
*/
package org.springframework.data.cassandra.core;
import static org.assertj.core.api.Assertions.*;
import static org.assertj.core.api.Assertions.assertThat;
import lombok.Data;
import lombok.experimental.Wither;
import reactor.test.StepVerifier;
import org.junit.Before;
import org.junit.Test;
import org.springframework.dao.OptimisticLockingFailureException;
import org.springframework.data.annotation.Id;
import org.springframework.data.annotation.PersistenceConstructor;
@@ -47,6 +49,7 @@ public class ReactiveOptimisticLockingIntegrationTests extends AbstractKeyspaceC
public void setUp() {
MappingCassandraConverter converter = new MappingCassandraConverter();
converter.afterPropertiesSet();
template = new ReactiveCassandraTemplate(new DefaultBridgedReactiveSession(session), converter);
@@ -64,17 +67,13 @@ public class ReactiveOptimisticLockingIntegrationTests extends AbstractKeyspaceC
template.insert(versionedEntity) //
.as(StepVerifier::create) //
.consumeNextWith(actual -> {
assertThat(actual.version).isEqualTo(1);
}).verifyComplete();
.consumeNextWith(actual -> assertThat(actual.version).isEqualTo(1))
.verifyComplete();
template.selectOne(Query.empty(), VersionedEntity.class) //
.as(StepVerifier::create) //
.consumeNextWith(actual -> {
assertThat(actual.version).isEqualTo(1);
}).verifyComplete();
.consumeNextWith(actual -> assertThat(actual.version).isEqualTo(1))
.verifyComplete();
}
@Test // DATACASS-576
@@ -97,17 +96,13 @@ public class ReactiveOptimisticLockingIntegrationTests extends AbstractKeyspaceC
template.insert(versionedEntity).flatMap(template::update) //
.as(StepVerifier::create) //
.consumeNextWith(actual -> {
assertThat(actual.version).isEqualTo(2);
}).verifyComplete();
.consumeNextWith(actual -> assertThat(actual.version).isEqualTo(2))
.verifyComplete();
template.selectOne(Query.empty(), VersionedEntity.class) //
.as(StepVerifier::create) //
.consumeNextWith(actual -> {
assertThat(actual.version).isEqualTo(2);
}).verifyComplete();
.consumeNextWith(actual -> assertThat(actual.version).isEqualTo(2))
.verifyComplete();
}
@Test // DATACASS-576

8
spring-data-cassandra/src/test/java/org/springframework/data/cassandra/repository/support/CassandraRepositoryFactoryUnitTests.java
View File

@@ -15,8 +15,8 @@
*/
package org.springframework.data.cassandra.repository.support;
import static org.assertj.core.api.Assertions.*;
import static org.mockito.Mockito.*;
import static org.assertj.core.api.Assertions.assertThat;
import static org.mockito.Mockito.when;
import java.io.Serializable;
@@ -25,6 +25,7 @@ import org.junit.Test;
import org.junit.runner.RunWith;
import org.mockito.Mock;
import org.mockito.junit.MockitoJUnitRunner;
import org.springframework.data.cassandra.core.CassandraTemplate;
import org.springframework.data.cassandra.core.convert.CassandraConverter;
import org.springframework.data.cassandra.core.mapping.BasicCassandraPersistentEntity;
@@ -42,9 +43,9 @@ import org.springframework.data.repository.Repository;
@SuppressWarnings({ "rawtypes", "unchecked" })
public class CassandraRepositoryFactoryUnitTests {
@Mock BasicCassandraPersistentEntity entity;
@Mock CassandraConverter converter;
@Mock CassandraMappingContext mappingContext;
@Mock BasicCassandraPersistentEntity entity;
@Mock CassandraTemplate template;
@Before
@@ -79,4 +80,5 @@ public class CassandraRepositoryFactoryUnitTests {
}
interface MyPersonRepository extends Repository<Person, Long> {}
}

13
spring-data-cassandra/src/test/java/org/springframework/data/cassandra/repository/support/ReactiveCassandraRepositoryFactoryUnitTests.java
View File

@@ -15,8 +15,8 @@
*/
package org.springframework.data.cassandra.repository.support;
import static org.assertj.core.api.Assertions.*;
import static org.mockito.Mockito.*;
import static org.assertj.core.api.Assertions.assertThat;
import static org.mockito.Mockito.when;
import java.io.Serializable;
@@ -25,6 +25,7 @@ import org.junit.Test;
import org.junit.runner.RunWith;
import org.mockito.Mock;
import org.mockito.junit.MockitoJUnitRunner;
import org.springframework.data.cassandra.core.ReactiveCassandraTemplate;
import org.springframework.data.cassandra.core.convert.CassandraConverter;
import org.springframework.data.cassandra.core.mapping.BasicCassandraPersistentEntity;
@@ -42,13 +43,14 @@ import org.springframework.data.repository.Repository;
@SuppressWarnings({ "rawtypes", "unchecked" })
public class ReactiveCassandraRepositoryFactoryUnitTests {
@Mock BasicCassandraPersistentEntity entity;
@Mock CassandraConverter converter;
@Mock CassandraMappingContext mappingContext;
@Mock BasicCassandraPersistentEntity entity;
@Mock ReactiveCassandraTemplate template;
@Before
public void setUp() {
when(template.getConverter()).thenReturn(converter);
when(converter.getMappingContext()).thenReturn(mappingContext);
}
@@ -60,8 +62,8 @@ public class ReactiveCassandraRepositoryFactoryUnitTests {
ReactiveCassandraRepositoryFactory repositoryFactory = new ReactiveCassandraRepositoryFactory(template);
CassandraEntityInformation<Person, Serializable> entityInformation = repositoryFactory
.getEntityInformation(Person.class);
CassandraEntityInformation<Person, Serializable> entityInformation =
repositoryFactory.getEntityInformation(Person.class);
assertThat(entityInformation).isInstanceOf(MappingCassandraEntityInformation.class);
}
@@ -78,4 +80,5 @@ public class ReactiveCassandraRepositoryFactoryUnitTests {
}
interface MyPersonRepository extends Repository<Person, Long> {}
}

16
spring-data-cassandra/src/test/java/org/springframework/data/cassandra/repository/support/SimpleCassandraRepositoryUnitTests.java
View File

@@ -15,12 +15,15 @@
*/
package org.springframework.data.cassandra.repository.support;
import static org.mockito.Mockito.*;
import lombok.Data;
import static org.mockito.Mockito.any;
import static org.mockito.Mockito.eq;
import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;
import java.io.Serializable;
import lombok.Data;
import org.junit.Before;
import org.junit.Test;
import org.junit.runner.RunWith;
@@ -28,6 +31,7 @@ import org.mockito.ArgumentCaptor;
import org.mockito.Captor;
import org.mockito.Mock;
import org.mockito.junit.MockitoJUnitRunner;
import org.springframework.data.annotation.Id;
import org.springframework.data.annotation.Version;
import org.springframework.data.cassandra.core.CassandraOperations;
@@ -57,14 +61,15 @@ import com.datastax.driver.core.querybuilder.Insert;
public class SimpleCassandraRepositoryUnitTests {
CassandraMappingContext mappingContext = new CassandraMappingContext();
MappingCassandraConverter converter = new MappingCassandraConverter(mappingContext);
SimpleCassandraRepository<Object, ? extends Serializable> repository;
@Mock CassandraOperations cassandraOperations;
@Mock CqlOperations cqlOperations;
@Mock UserTypeResolver userTypeResolver;
@Mock UserType userType;
@Mock UserTypeResolver userTypeResolver;
@Mock EntityWriteResult writeResult;
@Captor ArgumentCaptor<Insert> insertCaptor;
@@ -121,6 +126,7 @@ public class SimpleCassandraRepositoryUnitTests {
cassandraOperations);
VersionedPerson versionedPerson = new VersionedPerson();
versionedPerson.setVersion(2);
repository.save(versionedPerson);
@@ -155,6 +161,7 @@ public class SimpleCassandraRepositoryUnitTests {
cassandraOperations);
Person person = new Person();
person.setFirstname("foo");
person.setLastname("bar");
@@ -210,5 +217,4 @@ public class SimpleCassandraRepositoryUnitTests {
@Id String id;
@Version long version;
}
}

14
spring-data-cassandra/src/test/java/org/springframework/data/cassandra/repository/support/SimpleReactiveCassandraRepositoryUnitTests.java
View File

@@ -15,13 +15,16 @@
*/
package org.springframework.data.cassandra.repository.support;
import static org.mockito.Mockito.*;
import lombok.Data;
import reactor.core.publisher.Mono;
import static org.mockito.Mockito.any;
import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;
import java.io.Serializable;
import lombok.Data;
import reactor.core.publisher.Mono;
import org.junit.Before;
import org.junit.Test;
import org.junit.runner.RunWith;
@@ -29,6 +32,7 @@ import org.mockito.ArgumentCaptor;
import org.mockito.Captor;
import org.mockito.Mock;
import org.mockito.junit.MockitoJUnitRunner;
import org.springframework.data.annotation.Id;
import org.springframework.data.annotation.Version;
import org.springframework.data.cassandra.core.EntityWriteResult;
@@ -97,6 +101,7 @@ public class SimpleReactiveCassandraRepositoryUnitTests {
new MappingCassandraEntityInformation(entity, converter), cassandraOperations);
VersionedPerson versionedPerson = new VersionedPerson();
versionedPerson.setVersion(2);
repository.save(versionedPerson);
@@ -110,5 +115,4 @@ public class SimpleReactiveCassandraRepositoryUnitTests {
@Id String id;
@Version long version;
}
}