From 159c678f29cff0f3de3bf3794e54bdb9d33495c2 Mon Sep 17 00:00:00 2001 From: John Blum Date: Thu, 28 Feb 2019 14:49:10 -0800 Subject: [PATCH] DATACASS-576 - Polish. Resolves gh-136. --- .../core/AsyncCassandraTemplate.java | 284 +++++++++--------- .../cassandra/core/CassandraTemplate.java | 274 +++++++++-------- .../data/cassandra/core/EntityOperations.java | 139 +++++---- .../core/ReactiveCassandraTemplate.java | 277 +++++++++-------- .../MappingCassandraEntityInformation.java | 13 +- .../support/SimpleCassandraRepository.java | 44 +-- .../SimpleReactiveCassandraRepository.java | 41 +-- .../CassandraSessionFactoryBeanUnitTests.java | 73 +++-- ...syncOptimisticLockingIntegrationTests.java | 12 +- .../OptimisticLockingIntegrationTests.java | 9 +- ...tiveOptimisticLockingIntegrationTests.java | 29 +- .../CassandraRepositoryFactoryUnitTests.java | 8 +- ...veCassandraRepositoryFactoryUnitTests.java | 13 +- .../SimpleCassandraRepositoryUnitTests.java | 16 +- ...eReactiveCassandraRepositoryUnitTests.java | 14 +- 15 files changed, 647 insertions(+), 599 deletions(-) diff --git a/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/AsyncCassandraTemplate.java b/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/AsyncCassandraTemplate.java index 6c5dcef9c..0ba78eafd 100644 --- a/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/AsyncCassandraTemplate.java +++ b/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/AsyncCassandraTemplate.java @@ -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, 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 ListenableFuture> select(String cql, Class 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 ListenableFuture select(String cql, Consumer entityConsumer, Class 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 ListenableFuture selectOne(String cql, Class 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 ListenableFuture> slice(Statement statement, Class entityClass) { - - Assert.notNull(statement, "Statement must not be null"); - Assert.notNull(entityClass, "Entity type must not be null"); - - ListenableFuture resultSet = getAsyncCqlOperations().queryForResultSet(statement); - - Function 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 ListenableFuture select(Statement statement, Consumer entityConsumer, Class 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 ListenableFuture> slice(Statement statement, Class entityClass) { + + Assert.notNull(statement, "Statement must not be null"); + Assert.notNull(entityClass, "Entity type must not be null"); + + ListenableFuture resultSet = getAsyncCqlOperations().queryForResultSet(statement); + + Function 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 ListenableFuture> slice(Query query, Class 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 ListenableFuture select(Query query, Consumer entityConsumer, Class 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 ListenableFuture> slice(Query query, Class 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 source = operations.forEntity(entity, converter.getConversionService()); + AdaptibleEntity 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 ListenableFuture> doInsertVersioned(Insert insert, T entity, AdaptibleEntity 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 ListenableFuture> doInsert(Insert insert, T entity, AdaptibleEntity 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 source = getEntityOperations().forEntity(entity, getConverter().getConversionService()); CassandraPersistentEntity persistentEntity = getRequiredPersistentEntity(entity.getClass()); CqlIdentifier tableName = persistentEntity.getTableName(); - AdaptibleEntity 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 ListenableFuture> doUpdateVersioned(AdaptibleEntity 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 ListenableFuture> doUpdate(T entity, UpdateOptions options, CqlIdentifier tableName, @@ -574,23 +628,6 @@ public class AsyncCassandraTemplate implements AsyncCassandraOperations, Applica return executeSave(entity, tableName, update); } - private ListenableFuture> doUpdateVersioned(AdaptibleEntity 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 source = getEntityOperations().forEntity(entity, getConverter().getConversionService()); CassandraPersistentEntity persistentEntity = getRequiredPersistentEntity(entity.getClass()); CqlIdentifier tableName = persistentEntity.getTableName(); - AdaptibleEntity 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 doDeleteVersioned(Delete delete, Object entity, AdaptibleEntity 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, 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 ListenableFuture> executeSave(T entity, CqlIdentifier tableName, Statement statement) { - return executeSave(entity, tableName, statement, ignore -> { - }); + return executeSave(entity, tableName, statement, ignore -> { }); } private ListenableFuture> executeSave(T entity, CqlIdentifier tableName, Statement statement, @@ -733,6 +733,7 @@ public class AsyncCassandraTemplate implements AsyncCassandraOperations, Applica ListenableFuture result = getAsyncCqlOperations().execute(new AsyncStatementCallback(statement)); return new MappingListenableFutureAdapter<>(result, resultSet -> { + EntityWriteResult 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) 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 extends org.springframework.util.concurrent.ListenableFutureAdapter { @@ -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 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)); diff --git a/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/CassandraTemplate.java b/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/CassandraTemplate.java index fe26b5be3..d7e903ec5 100644 --- a/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/CassandraTemplate.java +++ b/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/CassandraTemplate.java @@ -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, 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 List select(String cql, Class 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 Stream stream(String cql, Class 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 selectOne(String cql, Class 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 Stream stream(String cql, Class 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 selectOne(Statement statement, Class 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 selectOne(Statement statement, Class 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 List doSelect(Query query, Class entityClass, CqlIdentifier tableName, Class returnType) { - Function 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 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 selectOne(Query query, Class entityClass) throws DataAccessException { + + List 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 Stream doStream(Query query, Class entityClass, CqlIdentifier tableName, Class 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 selectOne(Query query, Class entityClass) throws DataAccessException { - - List 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 mapper = getMapper(entityClass, entityClass, tableName); + List 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())); } EntityWriteResult doInsert(T entity, WriteOptions options, CqlIdentifier tableName) { - AdaptibleEntity source = operations.forEntity(entity, converter.getConversionService()); + AdaptibleEntity 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 EntityWriteResult doInsertVersioned(Insert insert, T entity, AdaptibleEntity 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 source = getEntityOperations().forEntity(entity, getConverter().getConversionService()); CassandraPersistentEntity persistentEntity = getRequiredPersistentEntity(entity.getClass()); CqlIdentifier tableName = persistentEntity.getTableName(); - AdaptibleEntity 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 EntityWriteResult doUpdateVersioned(AdaptibleEntity 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 source = getEntityOperations().forEntity(entity, getConverter().getConversionService()); CassandraPersistentEntity persistentEntity = getRequiredPersistentEntity(entity.getClass()); CqlIdentifier tableName = persistentEntity.getTableName(); - AdaptibleEntity 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 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, 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 EntityWriteResult 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, CqlProvider { diff --git a/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/EntityOperations.java b/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/EntityOperations.java index 6a19b4ec0..3e7640bb6 100644 --- a/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/EntityOperations.java +++ b/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/EntityOperations.java @@ -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, CassandraPersistentProperty> context; - - private CassandraPersistentEntity getRequiredPersistentEntity(Class entityType) { - return context.getRequiredPersistentEntity(ClassUtils.getUserClass(entityType)); - } + @NonNull @Getter(AccessLevel.PROTECTED) + private final MappingContext, CassandraPersistentProperty> mappingContext; /** * Creates a new {@link Entity} for the given bean. @@ -61,19 +60,9 @@ class EntityOperations { */ public Entity 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 AdaptibleEntity 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 propertyAccessor; + private static AdaptibleEntity of(T bean, + MappingContext, CassandraPersistentProperty> mappingContext, + ConversionService conversionService) { + + CassandraPersistentEntity entity = mappingContext.getRequiredPersistentEntity(bean.getClass()); + + PersistentPropertyAccessor propertyAccessor = entity.getPropertyAccessor(bean); + + return new AdaptibleMappedEntity<>(entity, + new ConvertingPropertyAccessor<>(propertyAccessor, conversionService)); + } + private AdaptibleMappedEntity(CassandraPersistentEntity entity, ConvertingPropertyAccessor propertyAccessor) { super(entity, propertyAccessor); @@ -241,22 +263,13 @@ class EntityOperations { this.propertyAccessor = propertyAccessor; } - private static AdaptibleEntity of(T bean, - MappingContext, CassandraPersistentProperty> context, - ConversionService conversionService) { - - CassandraPersistentEntity entity = context.getRequiredPersistentEntity(bean.getClass()); - PersistentPropertyAccessor 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(); } } } diff --git a/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/ReactiveCassandraTemplate.java b/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/ReactiveCassandraTemplate.java index aff8a6358..d2915e5dc 100644 --- a/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/ReactiveCassandraTemplate.java +++ b/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/ReactiveCassandraTemplate.java @@ -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, 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 Flux select(String cql, Class 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 Flux select(Statement cql, Class entityClass) { + public Flux select(Statement statement, Class 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 mapper = getMapper(entityClass, entityClass, EntityQueryUtils.getTableName(cql)); + Function 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 Mono selectOne(Statement statement, Class entityClass) { + return select(statement, entityClass).next(); } /* (non-Javadoc) @@ -266,24 +310,17 @@ public class ReactiveCassandraTemplate implements ReactiveCassandraOperations, A Mono effectiveFetchSizeMono = getEffectiveFetchSize(statement); RowMapper 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 Mono selectOne(Statement statement, Class 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 Mono selectOne(Query query, Class 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 Mono selectOne(Query query, Class 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())); } Mono> doInsert(T entity, WriteOptions options, CqlIdentifier tableName) { - AdaptibleEntity source = operations.forEntity(entity, converter.getConversionService()); + AdaptibleEntity 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 Mono> doInsertVersioned(Insert insert, T entity, AdaptibleEntity 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 source = this.entityOperations.forEntity(entity, getConverter().getConversionService()); CassandraPersistentEntity persistentEntity = getRequiredPersistentEntity(entity.getClass()); CqlIdentifier tableName = persistentEntity.getTableName(); - AdaptibleEntity 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 Mono> doUpdateVersioned(AdaptibleEntity 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 source = this.entityOperations.forEntity(entity, getConverter().getConversionService()); CassandraPersistentEntity persistentEntity = getRequiredPersistentEntity(entity.getClass()); CqlIdentifier tableName = persistentEntity.getTableName(); - AdaptibleEntity 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 doDeleteVersioned(Delete delete, Object entity, AdaptibleEntity 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 ReactiveSelect query(Class 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 ReactiveSelect query(Class 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, 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 Mono> 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 getEffectiveFetchSize(Statement statement) { @@ -815,8 +806,8 @@ public class ReactiveCassandraTemplate implements ReactiveCassandraOperations, A } } - return getReactiveCqlOperations().execute((ReactiveSessionCallback) session -> Mono - .just(session.getCluster().getConfiguration().getQueryOptions().getFetchSize())).single(); + return getReactiveCqlOperations().execute((ReactiveSessionCallback) 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, CqlProvider { diff --git a/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/repository/support/MappingCassandraEntityInformation.java b/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/repository/support/MappingCassandraEntityInformation.java index 2629d7654..6eb53a89a 100644 --- a/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/repository/support/MappingCassandraEntityInformation.java +++ b/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/repository/support/MappingCassandraEntityInformation.java @@ -63,9 +63,10 @@ public class MappingCassandraEntityInformation 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 extends PersistentEntityIn @Override public Class getIdType() { - if (entityMetadata.getIdProperty() != null) { - return (Class) entityMetadata.getRequiredIdProperty().getType(); + if (this.entityMetadata.getIdProperty() != null) { + return (Class) this.entityMetadata.getRequiredIdProperty().getType(); } return (Class) MapId.class; @@ -88,7 +89,7 @@ public class MappingCassandraEntityInformation extends PersistentEntityIn */ @Override public String getIdAttribute() { - return entityMetadata.getRequiredIdProperty().getName(); + return this.entityMetadata.getRequiredIdProperty().getName(); } /* (non-Javadoc) @@ -96,6 +97,6 @@ public class MappingCassandraEntityInformation extends PersistentEntityIn */ @Override public CqlIdentifier getTableName() { - return entityMetadata.getTableName(); + return this.entityMetadata.getTableName(); } } diff --git a/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/repository/support/SimpleCassandraRepository.java b/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/repository/support/SimpleCassandraRepository.java index 92ec8416c..15c6e9fdf 100644 --- a/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/repository/support/SimpleCassandraRepository.java +++ b/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/repository/support/SimpleCassandraRepository.java @@ -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 implements CassandraRepository { private static final InsertOptions INSERT_NULLS = InsertOptions.builder().withInsertNulls().build(); + private final AbstractMappingContext, CassandraPersistentProperty> mappingContext; + private final CassandraEntityInformation entityInformation; private final CassandraOperations operations; - private final AbstractMappingContext, CassandraPersistentProperty> mappingContext; - /** * Create a new {@link SimpleCassandraRepository} for the given {@link CassandraEntityInformation} and * {@link CassandraTemplate}. @@ -83,15 +84,16 @@ public class SimpleCassandraRepository implements CassandraRepository 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 implements CassandraRepository 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 implements CassandraRepository implements CassandraRepository 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 implements CassandraRepository implements CassandraRepository implements CassandraRepository implements CassandraRepository 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 implements CassandraRepository 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 implements CassandraRepository implements CassandraRepository implements CassandraRepository implements CassandraRepository implements CassandraRepository implements ReactiveCassand private static final InsertOptions INSERT_NULLS = InsertOptions.builder().withInsertNulls().build(); + private final AbstractMappingContext, CassandraPersistentProperty> mappingContext; + private final CassandraEntityInformation entityInformation; private final ReactiveCassandraOperations operations; - private final AbstractMappingContext, CassandraPersistentProperty> mappingContext; - /** * Create a new {@link SimpleReactiveCassandraRepository} for the given {@link CassandraEntityInformation} and * {@link ReactiveCassandraOperations}. @@ -76,15 +76,16 @@ public class SimpleReactiveCassandraRepository 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 implements ReactiveCassand * {@link ReactiveCassandraOperations#insert(Object, InsertOptions)}. */ private 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 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 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 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 implements ReactiveCassand */ @Override public Mono count() { - return operations.count(entityInformation.getJavaType()); + return this.operations.count(this.entityInformation.getJavaType()); } /* @@ -172,7 +173,7 @@ public class SimpleReactiveCassandraRepository 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 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 implements ReactiveCassand @Override public Flux 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 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 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 implements ReactiveCassand */ @Override public Mono deleteAll() { - return operations.truncate(entityInformation.getJavaType()); + return this.operations.truncate(this.entityInformation.getJavaType()); } /* (non-Javadoc) @@ -297,7 +298,7 @@ public class SimpleReactiveCassandraRepository 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 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(); } } diff --git a/spring-data-cassandra/src/test/java/org/springframework/data/cassandra/config/CassandraSessionFactoryBeanUnitTests.java b/spring-data-cassandra/src/test/java/org/springframework/data/cassandra/config/CassandraSessionFactoryBeanUnitTests.java index d3f99f536..2ab27664d 100755 --- a/spring-data-cassandra/src/test/java/org/springframework/data/cassandra/config/CassandraSessionFactoryBeanUnitTests.java +++ b/spring-data-cassandra/src/test/java/org/springframework/data/cassandra/config/CassandraSessionFactoryBeanUnitTests.java @@ -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 {} + } diff --git a/spring-data-cassandra/src/test/java/org/springframework/data/cassandra/core/AsyncOptimisticLockingIntegrationTests.java b/spring-data-cassandra/src/test/java/org/springframework/data/cassandra/core/AsyncOptimisticLockingIntegrationTests.java index d897b4d4e..875cd5e57 100644 --- a/spring-data-cassandra/src/test/java/org/springframework/data/cassandra/core/AsyncOptimisticLockingIntegrationTests.java +++ b/spring-data-cassandra/src/test/java/org/springframework/data/cassandra/core/AsyncOptimisticLockingIntegrationTests.java @@ -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(); } diff --git a/spring-data-cassandra/src/test/java/org/springframework/data/cassandra/core/OptimisticLockingIntegrationTests.java b/spring-data-cassandra/src/test/java/org/springframework/data/cassandra/core/OptimisticLockingIntegrationTests.java index 1217d3000..da85f9dc0 100644 --- a/spring-data-cassandra/src/test/java/org/springframework/data/cassandra/core/OptimisticLockingIntegrationTests.java +++ b/spring-data-cassandra/src/test/java/org/springframework/data/cassandra/core/OptimisticLockingIntegrationTests.java @@ -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(); } diff --git a/spring-data-cassandra/src/test/java/org/springframework/data/cassandra/core/ReactiveOptimisticLockingIntegrationTests.java b/spring-data-cassandra/src/test/java/org/springframework/data/cassandra/core/ReactiveOptimisticLockingIntegrationTests.java index 0d53118e7..08d3ae788 100644 --- a/spring-data-cassandra/src/test/java/org/springframework/data/cassandra/core/ReactiveOptimisticLockingIntegrationTests.java +++ b/spring-data-cassandra/src/test/java/org/springframework/data/cassandra/core/ReactiveOptimisticLockingIntegrationTests.java @@ -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 diff --git a/spring-data-cassandra/src/test/java/org/springframework/data/cassandra/repository/support/CassandraRepositoryFactoryUnitTests.java b/spring-data-cassandra/src/test/java/org/springframework/data/cassandra/repository/support/CassandraRepositoryFactoryUnitTests.java index b6ec53580..001611ae6 100644 --- a/spring-data-cassandra/src/test/java/org/springframework/data/cassandra/repository/support/CassandraRepositoryFactoryUnitTests.java +++ b/spring-data-cassandra/src/test/java/org/springframework/data/cassandra/repository/support/CassandraRepositoryFactoryUnitTests.java @@ -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 {} + } diff --git a/spring-data-cassandra/src/test/java/org/springframework/data/cassandra/repository/support/ReactiveCassandraRepositoryFactoryUnitTests.java b/spring-data-cassandra/src/test/java/org/springframework/data/cassandra/repository/support/ReactiveCassandraRepositoryFactoryUnitTests.java index e3566b4ab..019ee6fdd 100644 --- a/spring-data-cassandra/src/test/java/org/springframework/data/cassandra/repository/support/ReactiveCassandraRepositoryFactoryUnitTests.java +++ b/spring-data-cassandra/src/test/java/org/springframework/data/cassandra/repository/support/ReactiveCassandraRepositoryFactoryUnitTests.java @@ -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 entityInformation = repositoryFactory - .getEntityInformation(Person.class); + CassandraEntityInformation entityInformation = + repositoryFactory.getEntityInformation(Person.class); assertThat(entityInformation).isInstanceOf(MappingCassandraEntityInformation.class); } @@ -78,4 +80,5 @@ public class ReactiveCassandraRepositoryFactoryUnitTests { } interface MyPersonRepository extends Repository {} + } diff --git a/spring-data-cassandra/src/test/java/org/springframework/data/cassandra/repository/support/SimpleCassandraRepositoryUnitTests.java b/spring-data-cassandra/src/test/java/org/springframework/data/cassandra/repository/support/SimpleCassandraRepositoryUnitTests.java index df1209a34..1445cb5ef 100644 --- a/spring-data-cassandra/src/test/java/org/springframework/data/cassandra/repository/support/SimpleCassandraRepositoryUnitTests.java +++ b/spring-data-cassandra/src/test/java/org/springframework/data/cassandra/repository/support/SimpleCassandraRepositoryUnitTests.java @@ -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 repository; @Mock CassandraOperations cassandraOperations; @Mock CqlOperations cqlOperations; - @Mock UserTypeResolver userTypeResolver; @Mock UserType userType; + @Mock UserTypeResolver userTypeResolver; @Mock EntityWriteResult writeResult; @Captor ArgumentCaptor 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; } - } diff --git a/spring-data-cassandra/src/test/java/org/springframework/data/cassandra/repository/support/SimpleReactiveCassandraRepositoryUnitTests.java b/spring-data-cassandra/src/test/java/org/springframework/data/cassandra/repository/support/SimpleReactiveCassandraRepositoryUnitTests.java index e56499b24..6b63c5335 100644 --- a/spring-data-cassandra/src/test/java/org/springframework/data/cassandra/repository/support/SimpleReactiveCassandraRepositoryUnitTests.java +++ b/spring-data-cassandra/src/test/java/org/springframework/data/cassandra/repository/support/SimpleReactiveCassandraRepositoryUnitTests.java @@ -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; } - }