From 7a8f91ade397eea9d80e9f96fcc0c8fcb9af1c98 Mon Sep 17 00:00:00 2001 From: Mark Paluch Date: Wed, 30 Jan 2019 21:11:01 +0100 Subject: [PATCH] DATACASS-576 - Add support for Optimistic Locking. We now support Optimistic Locking for insert, update and delete operations leveraging Cassandra's lightweight transaction support. Modifying statements are enhanced with IF conditions to conditionally insert and modify rows and to prevent concurrent modifications by throwing OptimisticLockingFailureException. @Table class Person { @Id String id; String firstname; String lastname; @Version Long version; } Person daenerys = template.insert(new Person("Daenerys")); Person tmp = template.findOne(query(where("id").is(daenerys.getId())), Person.class); daenerys.setLastname("Targaryen"); template.save(daenerys); template.save(tmp); // throws OptimisticLockingFailureException --- .../core/AsyncCassandraTemplate.java | 149 ++++++-- .../cassandra/core/CassandraTemplate.java | 136 ++++++-- .../data/cassandra/core/EntityOperations.java | 319 ++++++++++++++++++ .../core/ReactiveCassandraTemplate.java | 153 +++++++-- .../MappingCassandraEntityInformation.java | 6 +- .../support/SimpleCassandraRepository.java | 16 +- .../SimpleReactiveCassandraRepository.java | 20 +- .../CassandraSessionFactoryBeanUnitTests.java | 5 +- ...syncOptimisticLockingIntegrationTests.java | 161 +++++++++ .../OptimisticLockingIntegrationTests.java | 146 ++++++++ ...tiveOptimisticLockingIntegrationTests.java | 180 ++++++++++ .../CassandraRepositoryFactoryUnitTests.java | 2 - ...veCassandraRepositoryFactoryUnitTests.java | 2 - .../SimpleCassandraRepositoryUnitTests.java | 45 +++ ...eReactiveCassandraRepositoryUnitTests.java | 114 +++++++ src/main/asciidoc/new-features.adoc | 1 + src/main/asciidoc/reference/cassandra.adoc | 39 +++ src/main/asciidoc/reference/mapping.adoc | 1 + 18 files changed, 1414 insertions(+), 81 deletions(-) create mode 100644 spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/EntityOperations.java create mode 100644 spring-data-cassandra/src/test/java/org/springframework/data/cassandra/core/AsyncOptimisticLockingIntegrationTests.java create mode 100644 spring-data-cassandra/src/test/java/org/springframework/data/cassandra/core/OptimisticLockingIntegrationTests.java create mode 100644 spring-data-cassandra/src/test/java/org/springframework/data/cassandra/core/ReactiveOptimisticLockingIntegrationTests.java create mode 100644 spring-data-cassandra/src/test/java/org/springframework/data/cassandra/repository/support/SimpleReactiveCassandraRepositoryUnitTests.java 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 25c3ece26..37094416e 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 @@ -26,7 +26,9 @@ import org.springframework.context.ApplicationEvent; import org.springframework.context.ApplicationEventPublisher; import org.springframework.context.ApplicationEventPublisherAware; import org.springframework.dao.DataAccessException; +import org.springframework.dao.OptimisticLockingFailureException; import org.springframework.data.cassandra.SessionFactory; +import org.springframework.data.cassandra.core.EntityOperations.AdaptibleEntity; import org.springframework.data.cassandra.core.convert.CassandraConverter; import org.springframework.data.cassandra.core.convert.MappingCassandraConverter; import org.springframework.data.cassandra.core.convert.QueryMapper; @@ -104,6 +106,8 @@ public class AsyncCassandraTemplate implements AsyncCassandraOperations, Applica private final SpelAwareProxyProjectionFactory projectionFactory; + private final EntityOperations operations; + private final StatementFactory statementFactory; private @Nullable ApplicationEventPublisher eventPublisher; @@ -168,6 +172,7 @@ public class AsyncCassandraTemplate implements AsyncCassandraOperations, Applica this.cqlOperations = asyncCqlTemplate; this.exceptionTranslator = asyncCqlTemplate.getExceptionTranslator(); this.projectionFactory = new SpelAwareProxyProjectionFactory(); + this.operations = new EntityOperations(converter.getMappingContext()); this.statementFactory = new StatementFactory(new QueryMapper(converter), new UpdateMapper(converter)); } @@ -488,7 +493,7 @@ public class AsyncCassandraTemplate implements AsyncCassandraOperations, Applica */ @Override public ListenableFuture insert(T entity) { - return new MappingListenableFutureAdapter<>(insert(entity, InsertOptions.empty()), writeResult -> entity); + return new MappingListenableFutureAdapter<>(insert(entity, InsertOptions.empty()), EntityWriteResult::getEntity); } /* (non-Javadoc) @@ -500,18 +505,37 @@ 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()); CassandraPersistentEntity persistentEntity = getRequiredPersistentEntity(entity.getClass()); CqlIdentifier tableName = persistentEntity.getTableName(); - Insert insert = EntityQueryUtils.createInsertQuery(tableName.toCql(), entity, options, getConverter(), + + T entityToUse = source.isVersionedEntity() ? source.initializeVersionProperty() : entity; + + Insert insert = EntityQueryUtils.createInsertQuery(tableName.toCql(), entityToUse, options, getConverter(), persistentEntity); - maybeEmitEvent(new BeforeSaveEvent<>(entity, tableName, insert)); + if (source.isVersionedEntity()) { + return doInsertVersioned(insert.ifNotExists(), entityToUse, source, tableName); + } - return new MappingListenableFutureAdapter<>(getAsyncCqlOperations().execute(new AsyncStatementCallback(insert)), - resultSet -> { - maybeEmitEvent(new AfterSaveEvent<>(entity, tableName)); - return EntityWriteResult.of(resultSet, entity); - }); + return 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)); + } + }); + } + + private ListenableFuture> doInsert(Insert insert, T entity, AdaptibleEntity source, + CqlIdentifier tableName) { + return executeSave(entity, tableName, insert); } /* (non-Javadoc) @@ -533,15 +557,38 @@ public class AsyncCassandraTemplate implements AsyncCassandraOperations, Applica 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); + } + + private ListenableFuture> doUpdate(T entity, UpdateOptions options, CqlIdentifier tableName, + CassandraPersistentEntity persistentEntity) { + Update update = getStatementFactory().update(entity, options, getConverter(), persistentEntity, tableName); - maybeEmitEvent(new BeforeSaveEvent<>(entity, tableName, update)); + return executeSave(entity, tableName, update); + } - return new MappingListenableFutureAdapter<>(getAsyncCqlOperations().execute(new AsyncStatementCallback(update)), - resultSet -> { - maybeEmitEvent(new AfterSaveEvent<>(entity, tableName)); - return EntityWriteResult.of(resultSet, entity); - }); + 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) @@ -563,15 +610,31 @@ public class AsyncCassandraTemplate implements AsyncCassandraOperations, Applica 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); - maybeEmitEvent(new BeforeDeleteEvent<>(delete, entity.getClass(), tableName)); + if (source.isVersionedEntity()) { + return doDeleteVersioned(delete, entity, source, tableName); + } - return new MappingListenableFutureAdapter<>(getAsyncCqlOperations().execute(new AsyncStatementCallback(delete)), - resultSet -> { - maybeEmitEvent(new AfterDeleteEvent<>(delete, entity.getClass(), tableName)); - return WriteResult.of(resultSet); - }); + return 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?", + entity, source.getVersion(), tableName)); + } + }); + } + + private ListenableFuture doDelete(Delete delete, Object entity, CqlIdentifier tableName) { + return executeDelete(entity, tableName, delete, result -> {}); } /* (non-Javadoc) @@ -655,8 +718,52 @@ public class AsyncCassandraTemplate implements AsyncCassandraOperations, Applica return this.statementFactory; } + private ListenableFuture> executeSave(T entity, CqlIdentifier tableName, + Statement statement) { + return executeSave(entity, tableName, statement, ignore -> { + + }); + } + + private ListenableFuture> executeSave(T entity, CqlIdentifier tableName, Statement statement, + Consumer beforeAfterSaveEvent) { + + maybeEmitEvent(new BeforeSaveEvent<>(entity, tableName, statement)); + + ListenableFuture result = getAsyncCqlOperations().execute(new AsyncStatementCallback(statement)); + + return new MappingListenableFutureAdapter<>(result, resultSet -> { + EntityWriteResult writeResult = EntityWriteResult.of(resultSet, entity); + + beforeAfterSaveEvent.accept(writeResult); + + maybeEmitEvent(new AfterSaveEvent<>(entity, tableName)); + + return writeResult; + }); + } + + private ListenableFuture executeDelete(Object entity, CqlIdentifier tableName, Statement statement, + Consumer resultConsumer) { + + maybeEmitEvent(new BeforeDeleteEvent<>(statement, entity.getClass(), tableName)); + + ListenableFuture result = getAsyncCqlOperations().execute(new AsyncStatementCallback(statement)); + + return new MappingListenableFutureAdapter<>(result, resultSet -> { + + WriteResult writeResult = WriteResult.of(resultSet); + + resultConsumer.accept(writeResult); + + maybeEmitEvent(new AfterDeleteEvent<>(statement, entity.getClass(), tableName)); + + return writeResult; + }); + } + private CqlIdentifier getTableName(Class entityClass) { - return getRequiredPersistentEntity(entityClass).getTableName(); + return operations.getTableName(entityClass); } private CqlIdentifier getTableName(Object entity) { 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 996fe0889..fe26b5be3 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 @@ -18,6 +18,7 @@ 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; @@ -26,7 +27,9 @@ import org.springframework.context.ApplicationEvent; import org.springframework.context.ApplicationEventPublisher; import org.springframework.context.ApplicationEventPublisherAware; import org.springframework.dao.DataAccessException; +import org.springframework.dao.OptimisticLockingFailureException; import org.springframework.data.cassandra.SessionFactory; +import org.springframework.data.cassandra.core.EntityOperations.AdaptibleEntity; import org.springframework.data.cassandra.core.convert.CassandraConverter; import org.springframework.data.cassandra.core.convert.MappingCassandraConverter; import org.springframework.data.cassandra.core.convert.QueryMapper; @@ -101,6 +104,8 @@ public class CassandraTemplate implements CassandraOperations, ApplicationEventP private final SpelAwareProxyProjectionFactory projectionFactory; + private final EntityOperations operations; + private final StatementFactory statementFactory; private @Nullable ApplicationEventPublisher eventPublisher; @@ -164,6 +169,7 @@ public class CassandraTemplate implements CassandraOperations, ApplicationEventP this.cqlOperations = cqlOperations; this.mappingContext = converter.getMappingContext(); this.projectionFactory = new SpelAwareProxyProjectionFactory(); + this.operations = new EntityOperations(converter.getMappingContext()); this.statementFactory = new StatementFactory(new QueryMapper(converter), new UpdateMapper(converter)); } @@ -196,7 +202,7 @@ public class CassandraTemplate implements CassandraOperations, ApplicationEventP */ @Override public CqlIdentifier getTableName(Class entityClass) { - return getRequiredPersistentEntity(entityClass).getTableName(); + return operations.getTableName(entityClass); } /* (non-Javadoc) @@ -557,19 +563,36 @@ public class CassandraTemplate implements CassandraOperations, ApplicationEventP EntityWriteResult doInsert(T entity, WriteOptions options, CqlIdentifier tableName) { + AdaptibleEntity source = operations.forEntity(entity, converter.getConversionService()); CassandraPersistentEntity persistentEntity = getRequiredPersistentEntity(entity.getClass()); - Insert insert = EntityQueryUtils.createInsertQuery(tableName.toCql(), entity, options, getConverter(), + T entityToUse = source.isVersionedEntity() ? source.initializeVersionProperty() : entity; + + Insert insert = EntityQueryUtils.createInsertQuery(tableName.toCql(), entityToUse, options, getConverter(), persistentEntity); - maybeEmitEvent(new BeforeSaveEvent<>(entity, tableName, insert)); + if (source.isVersionedEntity()) { + return doInsertVersioned(insert.ifNotExists(), entityToUse, source, tableName); + } - // noinspection ConstantConditions - WriteResult result = getCqlOperations().execute(new StatementCallback(insert)); + return doInsert(insert, entityToUse, tableName); + } - maybeEmitEvent(new AfterSaveEvent<>(entity, tableName)); + private EntityWriteResult doInsertVersioned(Insert insert, T entity, AdaptibleEntity source, + CqlIdentifier tableName) { - return EntityWriteResult.of(result, entity); + 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)); + } + }); + } + + private EntityWriteResult doInsert(Insert insert, T entity, CqlIdentifier tableName) { + return executeSave(entity, tableName, insert); } /* (non-Javadoc) @@ -591,16 +614,39 @@ public class CassandraTemplate implements CassandraOperations, ApplicationEventP 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); + } + + 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); - maybeEmitEvent(new BeforeSaveEvent<>(entity, tableName, update)); + return executeSave(entity, tableName, source.appendVersionCondition(update, previousVersion), result -> { - // noinspection ConstantConditions - WriteResult result = getCqlOperations().execute(new StatementCallback(update)); + 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)); + } + }); + } - maybeEmitEvent(new AfterSaveEvent<>(entity, tableName)); + private EntityWriteResult doUpdate(T entity, UpdateOptions options, CqlIdentifier tableName, + CassandraPersistentEntity persistentEntity) { - return EntityWriteResult.of(result, entity); + Update update = getStatementFactory().update(entity, options, getConverter(), persistentEntity, tableName); + + return executeSave(entity, tableName, update); } /* (non-Javadoc) @@ -622,16 +668,32 @@ public class CassandraTemplate implements CassandraOperations, ApplicationEventP 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); - maybeEmitEvent(new BeforeDeleteEvent<>(delete, entity.getClass(), tableName)); + if (source.isVersionedEntity()) { + return doDeleteVersioned(delete, entity, source, tableName); + } - // noinspection ConstantConditions - WriteResult result = getCqlOperations().execute(new StatementCallback(delete)); + return doDelete(delete, entity, tableName); + } - maybeEmitEvent(new AfterDeleteEvent<>(delete, entity.getClass(), tableName)); + private WriteResult doDeleteVersioned(Delete delete, Object entity, AdaptibleEntity source, + CqlIdentifier tableName) { - return result; + 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?", + entity, source.getVersion(), tableName)); + } + }); + } + + private WriteResult doDelete(Delete delete, Object entity, CqlIdentifier tableName) { + return executeDelete(entity, tableName, delete, result -> {}); } /* (non-Javadoc) @@ -752,6 +814,36 @@ public class CassandraTemplate implements CassandraOperations, ApplicationEventP return this.statementFactory; } + private EntityWriteResult executeSave(T entity, CqlIdentifier tableName, Statement statement) { + return executeSave(entity, tableName, statement, ignore -> {}); + } + + private EntityWriteResult executeSave(T entity, CqlIdentifier tableName, Statement statement, + Consumer resultConsumer) { + + maybeEmitEvent(new BeforeSaveEvent<>(entity, tableName, statement)); + + WriteResult result = getCqlOperations().execute(new StatementCallback(statement)); + resultConsumer.accept(result); + + maybeEmitEvent(new AfterSaveEvent<>(entity, tableName)); + + return EntityWriteResult.of(result, entity); + } + + private WriteResult executeDelete(Object entity, CqlIdentifier tableName, Statement statement, + Consumer resultConsumer) { + + maybeEmitEvent(new BeforeDeleteEvent<>(statement, entity.getClass(), tableName)); + + WriteResult result = getCqlOperations().execute(new StatementCallback(statement)); + resultConsumer.accept(result); + + maybeEmitEvent(new AfterDeleteEvent<>(statement, entity.getClass(), tableName)); + + return result; + } + private CqlIdentifier getTableName(Object entity) { return getRequiredPersistentEntity(entity.getClass()).getTableName(); } @@ -790,15 +882,13 @@ public class CassandraTemplate implements CassandraOperations, ApplicationEventP return row -> { - maybeEmitEvent(new AfterLoadEvent(row, targetType, tableName)); + maybeEmitEvent(new AfterLoadEvent<>(row, targetType, tableName)); Object source = getConverter().read(typeToRead, row); T result = (T) (targetType.isInterface() ? getProjectionFactory().createProjection(targetType, source) : source); - if (result != null) { - maybeEmitEvent(new AfterConvertEvent<>(row, result, tableName)); - } + maybeEmitEvent(new AfterConvertEvent<>(row, result, tableName)); return result; }; @@ -834,7 +924,7 @@ public class CassandraTemplate implements CassandraOperations, ApplicationEventP */ @Override public WriteResult doInSession(Session session) throws DriverException, DataAccessException { - return WriteResult.of(session.execute(statement)); + return WriteResult.of(session.execute(this.statement)); } /* (non-Javadoc) @@ -842,7 +932,7 @@ public class CassandraTemplate implements CassandraOperations, ApplicationEventP */ @Override public String getCql() { - return statement.toString(); + return this.statement.toString(); } } } 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 new file mode 100644 index 000000000..6a19b4ec0 --- /dev/null +++ b/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/EntityOperations.java @@ -0,0 +1,319 @@ +/* + * Copyright 2019 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.springframework.data.cassandra.core; + +import lombok.AccessLevel; +import lombok.NonNull; +import lombok.RequiredArgsConstructor; + +import org.springframework.core.convert.ConversionService; +import org.springframework.data.cassandra.core.cql.CqlIdentifier; +import org.springframework.data.cassandra.core.mapping.CassandraPersistentEntity; +import org.springframework.data.cassandra.core.mapping.CassandraPersistentProperty; +import org.springframework.data.mapping.PersistentPropertyAccessor; +import org.springframework.data.mapping.context.MappingContext; +import org.springframework.data.mapping.model.ConvertingPropertyAccessor; +import org.springframework.lang.Nullable; +import org.springframework.util.Assert; +import org.springframework.util.ClassUtils; + +import com.datastax.driver.core.Statement; +import com.datastax.driver.core.querybuilder.Delete; +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. + * + * @author Mark Paluch + * @since 2.2 + * @see CassandraTemplate + * @see AsyncCassandraTemplate + * @see ReactiveCassandraTemplate + */ +@RequiredArgsConstructor +class EntityOperations { + + private final @NonNull MappingContext, CassandraPersistentProperty> context; + + private CassandraPersistentEntity getRequiredPersistentEntity(Class entityType) { + return context.getRequiredPersistentEntity(ClassUtils.getUserClass(entityType)); + } + + /** + * Creates a new {@link Entity} for the given bean. + * + * @param entity must not be {@literal null}. + * @return + */ + public Entity forEntity(T entity) { + + 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(); + } + + /** + * Creates a new {@link AdaptibleEntity} for the given bean and {@link ConversionService}. + * + * @param entity must not be {@literal null}. + * @param conversionService must not be {@literal null}. + * @return + */ + public AdaptibleEntity forEntity(T entity, ConversionService conversionService) { + + Assert.notNull(entity, "Bean must not be null!"); + Assert.notNull(conversionService, "ConversionService must not be null!"); + + return AdaptibleMappedEntity.of(entity, context, conversionService); + } + + /** + * A representation of information about an entity. + */ + interface Entity { + + /** + * Returns whether the entity is versioned, i.e. if it contains a version property. + * + * @return + */ + default boolean isVersionedEntity() { + return false; + } + + /** + * Returns the value of the version if the entity has a version property, {@literal null} otherwise. + * + * @return + */ + @Nullable + Object getVersion(); + + /** + * Returns the underlying bean. + * + * @return + */ + T getBean(); + + /** + * Returns whether the entity is considered to be new. + * + * @return + */ + boolean isNew(); + } + + /** + * Information and commands on an entity. + */ + interface AdaptibleEntity extends Entity { + + /** + * Appends a {@code IF} condition to an {@link Update} statement for optimistic locking to perform the update only + * if the version number matches. This method accepts {@code currentVersionNumber} as the {@link Update} typically + * requires to increment the version number upon assembly time. + * + * @param update the {@link Update} statement to append the condition to. + * @param currentVersionNumber previous version number. + * @return the altered {@link Update} containing the {@code IF} condition for optimistic locking. + */ + Statement appendVersionCondition(Update update, Number currentVersionNumber); + + /** + * Appends a {@code IF} condition to an {@link Delete} statement for optimistic locking to perform the delete only + * if the version number matches. The {@link #getVersion() version number} is derived from the actual state as + * delete statements typically do not increment the version prior to statement creation. + * + * @param delete the {@link Delete} statement to append the condition to. + * @return the altered {@link Delete} containing the {@code IF} condition for optimistic locking. + * @see #getVersion() + */ + Statement appendVersionCondition(Delete delete); + + /** + * Initializes the version property of the of the current entity if available. + * + * @return the entity with the version property updated if available. + */ + T initializeVersionProperty(); + + /** + * Increments the value of the version property if available. + * + * @return the entity with the version property incremented if available. + */ + T incrementVersion(); + + /** + * Returns the current version value if the entity has a version property. + * + * @return the current version or {@literal null} in case it's uninitialized or the entity doesn't expose a version + * property. + */ + @Nullable + Number getVersion(); + } + + @RequiredArgsConstructor(access = AccessLevel.PROTECTED) + private static class MappedEntity implements Entity { + + private final @NonNull CassandraPersistentEntity entity; + private final @NonNull PersistentPropertyAccessor propertyAccessor; + + private static MappedEntity of(T bean, + MappingContext, CassandraPersistentProperty> context) { + + CassandraPersistentEntity entity = context.getRequiredPersistentEntity(bean.getClass()); + PersistentPropertyAccessor propertyAccessor = entity.getPropertyAccessor(bean); + + return new MappedEntity<>(entity, propertyAccessor); + } + + /* (non-Javadoc) + * @see org.springframework.data.cassandra.core.EntityOperations.Entity#isVersionedEntity() + */ + @Override + public boolean isVersionedEntity() { + return entity.hasVersionProperty(); + } + + /* (non-Javadoc) + * @see org.springframework.data.cassandra.core.EntityOperations.Entity#getVersion() + */ + @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()); + } + } + + private static class AdaptibleMappedEntity extends MappedEntity implements AdaptibleEntity { + + private final CassandraPersistentEntity entity; + private final ConvertingPropertyAccessor propertyAccessor; + + private AdaptibleMappedEntity(CassandraPersistentEntity entity, ConvertingPropertyAccessor propertyAccessor) { + + super(entity, propertyAccessor); + + this.entity = entity; + 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)); + } + + /* (non-Javadoc) + * @see org.springframework.data.cassandra.core.EntityOperations.AdaptibleEntity#appendVersionCondition(com.datastax.driver.core.querybuilder.Delete) + */ + @Override + public Statement appendVersionCondition(Delete delete) { + return delete.onlyIf(QueryBuilder.eq(getVersionColumnName().toCql(), getVersion())); + } + + /* (non-Javadoc) + * @see org.springframework.data.cassandra.core.EntityOperations.AdaptibleEntity#initializeVersionProperty() + */ + @Override + public T initializeVersionProperty() { + + if (!entity.hasVersionProperty()) { + return propertyAccessor.getBean(); + } + + CassandraPersistentProperty versionProperty = entity.getRequiredVersionProperty(); + + propertyAccessor.setProperty(versionProperty, versionProperty.getType().isPrimitive() ? 1 : 0); + + return propertyAccessor.getBean(); + } + + /* (non-Javadoc) + * @see org.springframework.data.cassandra.core.EntityOperations.AdaptibleEntity#incrementVersion() + */ + @Override + public T incrementVersion() { + + CassandraPersistentProperty versionProperty = entity.getRequiredVersionProperty(); + Number version = getVersion(); + Number nextVersion = version == null ? 0 : version.longValue() + 1; + + propertyAccessor.setProperty(versionProperty, nextVersion); + + return propertyAccessor.getBean(); + } + + /* (non-Javadoc) + * @see org.springframework.data.cassandra.core.EntityOperations.MappedEntity#getVersion() + */ + @Override + @Nullable + public Number getVersion() { + + CassandraPersistentProperty versionProperty = entity.getRequiredVersionProperty(); + + return propertyAccessor.getProperty(versionProperty, Number.class); + } + + private CqlIdentifier getVersionColumnName() { + return 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 602d3ce9e..6292a656e 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 @@ -16,21 +16,25 @@ package org.springframework.data.cassandra.core; import java.util.Collections; +import java.util.function.BiConsumer; import java.util.function.Function; import lombok.Value; import reactor.core.publisher.Flux; import reactor.core.publisher.Mono; +import reactor.core.publisher.SynchronousSink; import org.reactivestreams.Publisher; import org.springframework.context.ApplicationEvent; import org.springframework.context.ApplicationEventPublisher; import org.springframework.context.ApplicationEventPublisherAware; import org.springframework.dao.DataAccessException; +import org.springframework.dao.OptimisticLockingFailureException; import org.springframework.data.cassandra.ReactiveResultSet; import org.springframework.data.cassandra.ReactiveSession; import org.springframework.data.cassandra.ReactiveSessionFactory; +import org.springframework.data.cassandra.core.EntityOperations.AdaptibleEntity; import org.springframework.data.cassandra.core.convert.CassandraConverter; import org.springframework.data.cassandra.core.convert.MappingCassandraConverter; import org.springframework.data.cassandra.core.convert.QueryMapper; @@ -100,14 +104,16 @@ public class ReactiveCassandraTemplate implements ReactiveCassandraOperations, A private final CassandraConverter converter; - private final MappingContext, CassandraPersistentProperty> mappingContext; - private final ReactiveCqlOperations cqlOperations; - private final StatementFactory statementFactory; + private final MappingContext, CassandraPersistentProperty> mappingContext; private final SpelAwareProxyProjectionFactory projectionFactory; + private final EntityOperations operations; + + private final StatementFactory statementFactory; + private @Nullable ApplicationEventPublisher eventPublisher; /** @@ -170,6 +176,7 @@ public class ReactiveCassandraTemplate implements ReactiveCassandraOperations, A this.cqlOperations = reactiveCqlOperations; this.mappingContext = this.converter.getMappingContext(); this.projectionFactory = new SpelAwareProxyProjectionFactory(); + this.operations = new EntityOperations(converter.getMappingContext()); this.statementFactory = new StatementFactory(new QueryMapper(converter), new UpdateMapper(converter)); } @@ -497,19 +504,39 @@ public class ReactiveCassandraTemplate implements ReactiveCassandraOperations, A Mono> doInsert(T entity, WriteOptions options, CqlIdentifier tableName) { + AdaptibleEntity source = operations.forEntity(entity, converter.getConversionService()); CassandraPersistentEntity persistentEntity = getRequiredPersistentEntity(entity.getClass()); - Insert insert = EntityQueryUtils.createInsertQuery(tableName.toCql(), entity, options, getConverter(), + T entityToUse = source.isVersionedEntity() ? source.initializeVersionProperty() : entity; + + Insert insert = EntityQueryUtils.createInsertQuery(tableName.toCql(), entityToUse, options, getConverter(), persistentEntity); - // noinspection ConstantConditions - Mono> result = getReactiveCqlOperations() // - .execute(new StatementCallback(insert)) // - .doOnSubscribe(it -> maybeEmitEvent(new BeforeSaveEvent<>(entity, tableName, insert))) // - .map(it -> EntityWriteResult.of(it, entity)) // - .next(); + if (source.isVersionedEntity()) { + return doInsertVersioned(insert.ifNotExists(), entityToUse, source, tableName); + } - return result.doOnNext(it -> maybeEmitEvent(new AfterSaveEvent<>(entity, tableName))); + return doInsert(insert, entityToUse, tableName); + } + + private Mono> doInsertVersioned(Insert insert, T entity, AdaptibleEntity source, + CqlIdentifier tableName) { + + 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))); + return; + } + + sink.next(result); + }); + } + + private Mono> doInsert(Insert insert, T entity, CqlIdentifier tableName) { + return executeSave(entity, tableName, insert); } /* (non-Javadoc) @@ -517,7 +544,7 @@ public class ReactiveCassandraTemplate implements ReactiveCassandraOperations, A */ @Override public Mono update(T entity) { - return update(entity, UpdateOptions.empty()).map(writeResult -> entity); + return update(entity, UpdateOptions.empty()).map(EntityWriteResult::getEntity); } /* (non-Javadoc) @@ -531,15 +558,42 @@ public class ReactiveCassandraTemplate implements ReactiveCassandraOperations, A 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); + } + + 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); - Mono> result = getReactiveCqlOperations() // - .execute(new StatementCallback(update)) // - .doOnSubscribe(it -> maybeEmitEvent(new BeforeSaveEvent<>(entity, tableName, update))) // - .map(it -> EntityWriteResult.of(it, entity)) // - .next(); + return executeSave(entity, tableName, source.appendVersionCondition(update, previousVersion), (result, sink) -> { - return result.doOnNext(it -> maybeEmitEvent(new AfterSaveEvent<>(entity, tableName))); + 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))); + return; + } + + sink.next(result); + }); + } + + private Mono> doUpdate(T entity, UpdateOptions options, CqlIdentifier tableName, + CassandraPersistentEntity persistentEntity) { + + Update update = getStatementFactory().update(entity, options, getConverter(), persistentEntity, tableName); + + return executeSave(entity, tableName, update); } /* (non-Javadoc) @@ -561,14 +615,35 @@ public class ReactiveCassandraTemplate implements ReactiveCassandraOperations, A 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); - Mono result = getReactiveCqlOperations() // - .execute(new StatementCallback(delete)) // - .doOnSubscribe(it -> maybeEmitEvent(new BeforeDeleteEvent<>(delete, entity.getClass(), tableName))) // - .next(); + if (source.isVersionedEntity()) { + return doDeleteVersioned(delete, entity, source, tableName); + } - return result.doOnNext(it -> maybeEmitEvent(new AfterDeleteEvent<>(delete, entity.getClass(), tableName))); + return doDelete(delete, entity, tableName); + } + + private Mono doDeleteVersioned(Delete delete, Object entity, AdaptibleEntity source, + CqlIdentifier tableName) { + + 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?", + entity, source.getVersion(), tableName))); + return; + } + + sink.next(result); + }); + } + + private Mono doDelete(Delete delete, Object entity, CqlIdentifier tableName) { + return executeDelete(entity, tableName, delete, (result, sink) -> sink.next(result)); } /* (non-Javadoc) @@ -685,7 +760,37 @@ public class ReactiveCassandraTemplate implements ReactiveCassandraOperations, A } CqlIdentifier getTableName(Class entityClass) { - return getRequiredPersistentEntity(entityClass).getTableName(); + return operations.getTableName(entityClass); + } + + private Mono> executeSave(T entity, CqlIdentifier tableName, Statement statement) { + return executeSave(entity, tableName, statement, (writeResult, sink) -> sink.next(writeResult)); + } + + private Mono> executeSave(T entity, CqlIdentifier tableName, Statement statement, + BiConsumer, SynchronousSink>> handler) { + + maybeEmitEvent(new BeforeSaveEvent<>(entity, tableName, statement)); + + Flux execute = getReactiveCqlOperations().execute(new StatementCallback(statement)); + + return execute.map(it -> EntityWriteResult.of(it, entity)).handle(handler) // + .doOnSubscribe(it -> maybeEmitEvent(new BeforeSaveEvent<>(entity, tableName, statement))) // + .doOnNext(it -> maybeEmitEvent(new AfterSaveEvent<>(it, tableName))) // + .next(); + } + + private Mono executeDelete(Object entity, CqlIdentifier tableName, Statement statement, + BiConsumer> handler) { + + maybeEmitEvent(new BeforeDeleteEvent<>(statement, entity.getClass(), tableName)); + + Flux execute = getReactiveCqlOperations().execute(new StatementCallback(statement)); + + return execute.map(it -> EntityWriteResult.of(it, entity)).handle(handler) // + .doOnSubscribe(it -> maybeEmitEvent(new BeforeSaveEvent<>(entity, tableName, statement))) // + .doOnNext(it -> maybeEmitEvent(new AfterDeleteEvent<>(statement, entity.getClass(), tableName))) // + .next(); } private CqlIdentifier getTableName(Object entity) { 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 397704cb5..2629d7654 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 @@ -21,7 +21,7 @@ import org.springframework.data.cassandra.core.mapping.CassandraPersistentEntity import org.springframework.data.cassandra.core.mapping.CassandraPersistentProperty; import org.springframework.data.cassandra.core.mapping.MapId; import org.springframework.data.cassandra.repository.query.CassandraEntityInformation; -import org.springframework.data.repository.core.support.AbstractEntityInformation; +import org.springframework.data.repository.core.support.PersistentEntityInformation; import org.springframework.lang.Nullable; import org.springframework.util.Assert; @@ -33,7 +33,7 @@ import org.springframework.util.Assert; * @author Matthew T. Adams * @author Mark Paluch */ -public class MappingCassandraEntityInformation extends AbstractEntityInformation +public class MappingCassandraEntityInformation extends PersistentEntityInformation implements CassandraEntityInformation { private final CassandraPersistentEntity entityMetadata; @@ -47,7 +47,7 @@ public class MappingCassandraEntityInformation extends AbstractEntityInfo */ public MappingCassandraEntityInformation(CassandraPersistentEntity entity, CassandraConverter converter) { - super(entity.getType()); + super(entity); this.entityMetadata = entity; this.converter = converter; 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 af2292019..92ec8416c 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 @@ -24,11 +24,14 @@ import java.util.Optional; import org.springframework.data.cassandra.core.CassandraOperations; import org.springframework.data.cassandra.core.CassandraTemplate; import org.springframework.data.cassandra.core.InsertOptions; +import org.springframework.data.cassandra.core.mapping.BasicCassandraPersistentEntity; +import org.springframework.data.cassandra.core.mapping.CassandraPersistentProperty; import org.springframework.data.cassandra.core.query.Query; import org.springframework.data.cassandra.repository.CassandraRepository; import org.springframework.data.cassandra.repository.query.CassandraEntityInformation; import org.springframework.data.domain.Pageable; import org.springframework.data.domain.Slice; +import org.springframework.data.mapping.context.AbstractMappingContext; import org.springframework.data.util.StreamUtils; import org.springframework.data.util.Streamable; import org.springframework.util.Assert; @@ -53,6 +56,8 @@ public class SimpleCassandraRepository implements CassandraRepository, CassandraPersistentProperty> mappingContext; + /** * Create a new {@link SimpleCassandraRepository} for the given {@link CassandraEntityInformation} and * {@link CassandraTemplate}. @@ -67,6 +72,7 @@ public class SimpleCassandraRepository implements CassandraRepository implements CassandraRepository persistentEntity = mappingContext.getPersistentEntity(entity.getClass()); + if (persistentEntity != null && persistentEntity.hasVersionProperty()) { + + if (!entityInformation.isNew(entity)) { + return operations.update(entity); + } + } + return operations.insert(entity, INSERT_NULLS).getEntity(); } @@ -91,7 +105,7 @@ public class SimpleCassandraRepository implements CassandraRepository result = new ArrayList<>(); for (S entity : entities) { - result.add(operations.insert(entity, INSERT_NULLS).getEntity()); + result.add(save(entity)); } return result; diff --git a/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/repository/support/SimpleReactiveCassandraRepository.java b/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/repository/support/SimpleReactiveCassandraRepository.java index 22e5e1afe..275a16125 100644 --- a/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/repository/support/SimpleReactiveCassandraRepository.java +++ b/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/repository/support/SimpleReactiveCassandraRepository.java @@ -19,10 +19,14 @@ import reactor.core.publisher.Flux; import reactor.core.publisher.Mono; import org.reactivestreams.Publisher; +import org.springframework.data.cassandra.core.EntityWriteResult; import org.springframework.data.cassandra.core.InsertOptions; import org.springframework.data.cassandra.core.ReactiveCassandraOperations; +import org.springframework.data.cassandra.core.mapping.BasicCassandraPersistentEntity; +import org.springframework.data.cassandra.core.mapping.CassandraPersistentProperty; import org.springframework.data.cassandra.repository.ReactiveCassandraRepository; import org.springframework.data.cassandra.repository.query.CassandraEntityInformation; +import org.springframework.data.mapping.context.AbstractMappingContext; import org.springframework.util.Assert; import com.datastax.driver.core.querybuilder.Insert; @@ -44,6 +48,8 @@ public class SimpleReactiveCassandraRepository implements ReactiveCassand private final ReactiveCassandraOperations operations; + private final AbstractMappingContext, CassandraPersistentProperty> mappingContext; + /** * Create a new {@link SimpleReactiveCassandraRepository} for the given {@link CassandraEntityInformation} and * {@link ReactiveCassandraOperations}. @@ -59,6 +65,7 @@ public class SimpleReactiveCassandraRepository implements ReactiveCassand this.entityInformation = metadata; this.operations = operations; + this.mappingContext = operations.getConverter().getMappingContext(); } /* (non-Javadoc) @@ -69,7 +76,15 @@ public class SimpleReactiveCassandraRepository implements ReactiveCassand Assert.notNull(entity, "Entity must not be null"); - return operations.insert(entity, INSERT_NULLS).thenReturn(entity); + BasicCassandraPersistentEntity persistentEntity = mappingContext.getPersistentEntity(entity.getClass()); + if (persistentEntity != null && persistentEntity.hasVersionProperty()) { + + if (!entityInformation.isNew(entity)) { + return operations.update(entity); + } + } + + return operations.insert(entity, INSERT_NULLS).map(EntityWriteResult::getEntity); } /** @@ -103,8 +118,7 @@ public class SimpleReactiveCassandraRepository implements ReactiveCassand Assert.notNull(entityStream, "The given Publisher of entities must not be null"); - return Flux.from(entityStream) - .flatMap(entity -> operations.insert(entity, INSERT_NULLS).thenReturn(entity)); + return Flux.from(entityStream).flatMap(this::save); } /* (non-Javadoc) 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 738e29fd8..d3f99f536 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 @@ -16,8 +16,7 @@ package org.springframework.data.cassandra.config; import static org.assertj.core.api.Assertions.*; -import static org.mockito.ArgumentMatchers.anyBoolean; -import static org.mockito.ArgumentMatchers.eq; +import static org.mockito.ArgumentMatchers.*; import static org.mockito.Mockito.*; import static org.springframework.data.cassandra.config.CassandraSessionFactoryBean.*; @@ -29,6 +28,7 @@ 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; import com.datastax.driver.core.Cluster; import com.datastax.driver.core.Session; @@ -55,6 +55,7 @@ public class CassandraSessionFactoryBeanUnitTests { public void setup() { when(mockCluster.connect()).thenReturn(mockSession); + when(mockConverter.getMappingContext()).thenReturn(new CassandraMappingContext()); factoryBean = spy(new CassandraSessionFactoryBean()); factoryBean.setCluster(mockCluster); 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 new file mode 100644 index 000000000..d897b4d4e --- /dev/null +++ b/spring-data-cassandra/src/test/java/org/springframework/data/cassandra/core/AsyncOptimisticLockingIntegrationTests.java @@ -0,0 +1,161 @@ +/* + * Copyright 2019 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.springframework.data.cassandra.core; + +import static org.assertj.core.api.Assertions.*; + +import lombok.Data; +import 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; +import org.springframework.data.annotation.Version; +import org.springframework.data.cassandra.core.convert.MappingCassandraConverter; +import org.springframework.data.cassandra.core.cql.CqlTemplate; +import org.springframework.data.cassandra.core.query.Query; +import org.springframework.data.cassandra.repository.support.SchemaTestUtils; +import org.springframework.data.cassandra.test.util.AbstractKeyspaceCreatingIntegrationTest; + +/** + * Integration tests for optimistic locking through {@link AsyncCassandraTemplate}. + * + * @author Mark Paluch + */ +public class AsyncOptimisticLockingIntegrationTests extends AbstractKeyspaceCreatingIntegrationTest { + + AsyncCassandraTemplate template; + + @Before + public void setUp() { + + MappingCassandraConverter converter = new MappingCassandraConverter(); + converter.afterPropertiesSet(); + + template = new AsyncCassandraTemplate(session, converter); + + CassandraTemplate syncTemplate = new CassandraTemplate(new CqlTemplate(session), converter); + + SchemaTestUtils.potentiallyCreateTableFor(VersionedEntity.class, syncTemplate); + SchemaTestUtils.truncate(VersionedEntity.class, syncTemplate); + } + + @Test // DATACASS-576 + public void shouldInsertVersioned() { + + VersionedEntity versionedEntity = new VersionedEntity(42); + + VersionedEntity saved = getUninterruptibly(template.insert(versionedEntity)); + VersionedEntity loaded = getUninterruptibly(template.selectOne(Query.empty(), VersionedEntity.class)); + + assertThat(saved.version).isEqualTo(1); + assertThat(loaded).isNotNull(); + assertThat(loaded.version).isEqualTo(1); + } + + @Test // DATACASS-576 + public void duplicateInsertShouldFail() { + + getUninterruptibly(template.insert(new VersionedEntity(42))); + + assertThatThrownBy(() -> getUninterruptibly(template.insert(new VersionedEntity(42)))) + .hasRootCauseInstanceOf(OptimisticLockingFailureException.class); + } + + @Test // DATACASS-576 + public void shouldUpdateVersioned() { + + VersionedEntity versionedEntity = new VersionedEntity(42); + + VersionedEntity saved = getUninterruptibly(template.insert(versionedEntity)); + VersionedEntity updated = getUninterruptibly(template.update(saved)); + VersionedEntity loaded = getUninterruptibly(template.selectOne(Query.empty(), VersionedEntity.class)); + + assertThat(saved.version).isEqualTo(1); + assertThat(updated.version).isEqualTo(2); + assertThat(loaded).isNotNull(); + assertThat(loaded.version).isEqualTo(2); + } + + @Test // DATACASS-576 + public void updateForOutdatedEntityShouldFail() { + + VersionedEntity versionedEntity = new VersionedEntity(42); + + getUninterruptibly(template.insert(versionedEntity)); + assertThatThrownBy(() -> getUninterruptibly(template.update(new VersionedEntity(42, 5, "f")))) + .hasRootCauseInstanceOf(OptimisticLockingFailureException.class); + } + + @Test // DATACASS-576 + public void shouldDeleteVersionedEntity() { + + 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(); + } + + @Test // DATACASS-576 + public void deleteForOutdatedEntityShouldFail() { + + getUninterruptibly(template.insert(new VersionedEntity(42))); + + assertThatThrownBy(() -> getUninterruptibly(template.delete(new VersionedEntity(42)))) + .hasRootCauseInstanceOf(OptimisticLockingFailureException.class); + + VersionedEntity loaded = getUninterruptibly(template.selectOne(Query.empty(), VersionedEntity.class)); + assertThat(loaded).isNotNull(); + } + + private static T getUninterruptibly(Future future) { + + try { + return future.get(); + } catch (Exception cause) { + throw new IllegalStateException(cause); + } + } + + @Data + @Wither + static class VersionedEntity { + + @Id final long id; + + @Version final long version; + + final String name; + + public VersionedEntity(long id) { + this(id, 0, null); + } + + @PersistenceConstructor + public VersionedEntity(long id, long version, String name) { + this.id = id; + this.version = version; + this.name = name; + } + } +} 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 new file mode 100644 index 000000000..1217d3000 --- /dev/null +++ b/spring-data-cassandra/src/test/java/org/springframework/data/cassandra/core/OptimisticLockingIntegrationTests.java @@ -0,0 +1,146 @@ +/* + * Copyright 2019 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.springframework.data.cassandra.core; + +import static org.assertj.core.api.Assertions.*; + +import lombok.Data; +import 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; +import org.springframework.data.annotation.Version; +import org.springframework.data.cassandra.core.convert.MappingCassandraConverter; +import org.springframework.data.cassandra.repository.support.SchemaTestUtils; +import org.springframework.data.cassandra.test.util.AbstractKeyspaceCreatingIntegrationTest; + +/** + * Integration tests for optimistic locking through {@link CassandraTemplate}. + * + * @author Mark Paluch + */ +public class OptimisticLockingIntegrationTests extends AbstractKeyspaceCreatingIntegrationTest { + + CassandraTemplate template; + + @Before + public void setUp() { + + MappingCassandraConverter converter = new MappingCassandraConverter(); + converter.afterPropertiesSet(); + + template = new CassandraTemplate(session, converter); + + SchemaTestUtils.potentiallyCreateTableFor(VersionedEntity.class, template); + SchemaTestUtils.truncate(VersionedEntity.class, template); + } + + @Test // DATACASS-576 + public void shouldInsertVersioned() { + + VersionedEntity versionedEntity = new VersionedEntity(42); + + VersionedEntity saved = template.insert(versionedEntity); + VersionedEntity loaded = template.query(VersionedEntity.class).firstValue(); + + assertThat(saved.version).isEqualTo(1); + assertThat(loaded).isNotNull(); + assertThat(loaded.version).isEqualTo(1); + } + + @Test // DATACASS-576 + public void duplicateInsertShouldFail() { + + template.insert(new VersionedEntity(42)); + + assertThatThrownBy(() -> template.insert(new VersionedEntity(42))) + .isInstanceOf(OptimisticLockingFailureException.class); + } + + @Test // DATACASS-576 + public void shouldUpdateVersioned() { + + VersionedEntity versionedEntity = new VersionedEntity(42); + + VersionedEntity saved = template.insert(versionedEntity); + VersionedEntity updated = template.update(saved); + VersionedEntity loaded = template.query(VersionedEntity.class).firstValue(); + + assertThat(saved.version).isEqualTo(1); + assertThat(updated.version).isEqualTo(2); + assertThat(loaded).isNotNull(); + assertThat(loaded.version).isEqualTo(2); + } + + @Test // DATACASS-576 + public void updateForOutdatedEntityShouldFail() { + + VersionedEntity versionedEntity = new VersionedEntity(42); + + template.insert(versionedEntity); + assertThatThrownBy(() -> template.update(new VersionedEntity(42, 5, "f"))) + .isInstanceOf(OptimisticLockingFailureException.class); + } + + @Test // DATACASS-576 + public void shouldDeleteVersionedEntity() { + + VersionedEntity versionedEntity = new VersionedEntity(42); + + VersionedEntity saved = template.insert(versionedEntity); + template.delete(saved); + + VersionedEntity loaded = template.query(VersionedEntity.class).firstValue(); + assertThat(loaded).isNull(); + } + + @Test // DATACASS-576 + public void deleteForOutdatedEntityShouldFail() { + + template.insert(new VersionedEntity(42)); + + assertThatThrownBy(() -> template.delete(new VersionedEntity(42))) + .isInstanceOf(OptimisticLockingFailureException.class); + + VersionedEntity loaded = template.query(VersionedEntity.class).firstValue(); + assertThat(loaded).isNotNull(); + } + + @Data + @Wither + static class VersionedEntity { + + @Id final long id; + + @Version final long version; + + final String name; + + public VersionedEntity(long id) { + this(id, 0, null); + } + + @PersistenceConstructor + public VersionedEntity(long id, long version, String name) { + this.id = id; + this.version = version; + this.name = name; + } + } +} 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 new file mode 100644 index 000000000..0d53118e7 --- /dev/null +++ b/spring-data-cassandra/src/test/java/org/springframework/data/cassandra/core/ReactiveOptimisticLockingIntegrationTests.java @@ -0,0 +1,180 @@ +/* + * Copyright 2019 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.springframework.data.cassandra.core; + +import static org.assertj.core.api.Assertions.*; + +import lombok.Data; +import 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; +import org.springframework.data.annotation.Version; +import org.springframework.data.cassandra.core.convert.MappingCassandraConverter; +import org.springframework.data.cassandra.core.cql.CqlTemplate; +import org.springframework.data.cassandra.core.cql.session.DefaultBridgedReactiveSession; +import org.springframework.data.cassandra.core.query.Query; +import org.springframework.data.cassandra.repository.support.SchemaTestUtils; +import org.springframework.data.cassandra.test.util.AbstractKeyspaceCreatingIntegrationTest; + +/** + * Integration tests for optimistic locking through {@link ReactiveCassandraTemplate}. + * + * @author Mark Paluch + */ +public class ReactiveOptimisticLockingIntegrationTests extends AbstractKeyspaceCreatingIntegrationTest { + + ReactiveCassandraTemplate template; + + @Before + public void setUp() { + + MappingCassandraConverter converter = new MappingCassandraConverter(); + converter.afterPropertiesSet(); + + template = new ReactiveCassandraTemplate(new DefaultBridgedReactiveSession(session), converter); + + CassandraTemplate syncTemplate = new CassandraTemplate(new CqlTemplate(session), converter); + + SchemaTestUtils.potentiallyCreateTableFor(VersionedEntity.class, syncTemplate); + SchemaTestUtils.truncate(VersionedEntity.class, syncTemplate); + } + + @Test // DATACASS-576 + public void shouldInsertVersioned() { + + VersionedEntity versionedEntity = new VersionedEntity(42); + + template.insert(versionedEntity) // + .as(StepVerifier::create) // + .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(); + } + + @Test // DATACASS-576 + public void duplicateInsertShouldFail() { + + template.insert(new VersionedEntity(42)) // + .as(StepVerifier::create) // + .expectNextCount(1) // + .verifyComplete(); + + template.insert(new VersionedEntity(42)) // + .as(StepVerifier::create) // + .verifyError(OptimisticLockingFailureException.class); + } + + @Test // DATACASS-576 + public void shouldUpdateVersioned() { + + VersionedEntity versionedEntity = new VersionedEntity(42); + + template.insert(versionedEntity).flatMap(template::update) // + .as(StepVerifier::create) // + .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(); + } + + @Test // DATACASS-576 + public void updateForOutdatedEntityShouldFail() { + + template.insert(new VersionedEntity(42)) // + .as(StepVerifier::create) // + .expectNextCount(1) // + .verifyComplete(); + + template.update(new VersionedEntity(42, 5, "f")) // + .as(StepVerifier::create) // + .verifyError(OptimisticLockingFailureException.class); + } + + @Test // DATACASS-576 + public void shouldDeleteVersionedEntity() { + + VersionedEntity versionedEntity = new VersionedEntity(42); + + template.insert(versionedEntity).flatMap(template::delete) // + .as(StepVerifier::create) // + .expectNextCount(1) // + .verifyComplete(); + + template.query(VersionedEntity.class).first() // + .as(StepVerifier::create) // + .verifyComplete(); + } + + @Test // DATACASS-576 + public void deleteForOutdatedEntityShouldFail() { + + template.insert(new VersionedEntity(42))// + .as(StepVerifier::create) // + .expectNextCount(1) // + .verifyComplete(); + + template.delete(new VersionedEntity(42)) // + .as(StepVerifier::create) // + .verifyError(OptimisticLockingFailureException.class); + + template.query(VersionedEntity.class).first() // + .as(StepVerifier::create) // + .expectNextCount(1) // + .verifyComplete(); + } + + @Data + @Wither + static class VersionedEntity { + + @Id final long id; + + @Version final long version; + + final String name; + + public VersionedEntity(long id) { + this(id, 0, null); + } + + @PersistenceConstructor + public VersionedEntity(long id, long version, String name) { + this.id = id; + this.version = version; + this.name = name; + } + } +} 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 afec7d463..33d97597d 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 @@ -61,7 +61,6 @@ public class CassandraRepositoryFactoryUnitTests { public void usesMappingCassandraEntityInformationIfMappingContextSet() { when(mappingContext.getRequiredPersistentEntity(Person.class)).thenReturn(entity); - when(entity.getType()).thenReturn(Person.class); CassandraRepositoryFactory repositoryFactory = new CassandraRepositoryFactory(template); @@ -75,7 +74,6 @@ public class CassandraRepositoryFactoryUnitTests { public void createsRepositoryWithIdTypeLong() { when(mappingContext.getRequiredPersistentEntity(Person.class)).thenReturn(entity); - when(entity.getType()).thenReturn(Person.class); CassandraRepositoryFactory repositoryFactory = new CassandraRepositoryFactory(template); MyPersonRepository repository = repositoryFactory.getRepository(MyPersonRepository.class); 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 39136f2aa..659e9c8e0 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 @@ -60,7 +60,6 @@ public class ReactiveCassandraRepositoryFactoryUnitTests { public void usesMappingCassandraEntityInformationIfMappingContextSet() { when(mappingContext.getRequiredPersistentEntity(Person.class)).thenReturn(entity); - when(entity.getType()).thenReturn(Person.class); ReactiveCassandraRepositoryFactory repositoryFactory = new ReactiveCassandraRepositoryFactory(template); @@ -74,7 +73,6 @@ public class ReactiveCassandraRepositoryFactoryUnitTests { public void createsRepositoryWithIdTypeLong() { when(mappingContext.getRequiredPersistentEntity(Person.class)).thenReturn(entity); - when(entity.getType()).thenReturn(Person.class); ReactiveCassandraRepositoryFactory repositoryFactory = new ReactiveCassandraRepositoryFactory(template); MyPersonRepository repository = repositoryFactory.getRepository(MyPersonRepository.class); 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 9dc58b091..df1209a34 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 @@ -29,6 +29,7 @@ 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; import org.springframework.data.cassandra.core.EntityWriteResult; import org.springframework.data.cassandra.core.InsertOptions; @@ -71,6 +72,7 @@ public class SimpleCassandraRepositoryUnitTests { @Before public void before() { mappingContext.setUserTypeResolver(userTypeResolver); + when(cassandraOperations.getConverter()).thenReturn(converter); } @Test // DATACASS-428, DATACASS-560, DATACASS-573 @@ -91,6 +93,41 @@ public class SimpleCassandraRepositoryUnitTests { verify(cassandraOperations).insert(person, InsertOptions.builder().withInsertNulls().build()); } + @Test // DATACASS-576 + public void shouldInsertNewVersionedEntity() { + + when(cassandraOperations.insert(any(), any(InsertOptions.class))).thenReturn(writeResult); + + CassandraPersistentEntity entity = converter.getMappingContext() + .getRequiredPersistentEntity(VersionedPerson.class); + + repository = new SimpleCassandraRepository(new MappingCassandraEntityInformation(entity, converter), + cassandraOperations); + + VersionedPerson versionedPerson = new VersionedPerson(); + + repository.save(versionedPerson); + + verify(cassandraOperations).insert(versionedPerson, InsertOptions.builder().withInsertNulls().build()); + } + + @Test // DATACASS-576 + public void shouldUpdateExistingVersionedEntity() { + + CassandraPersistentEntity entity = converter.getMappingContext() + .getRequiredPersistentEntity(VersionedPerson.class); + + repository = new SimpleCassandraRepository(new MappingCassandraEntityInformation(entity, converter), + cassandraOperations); + + VersionedPerson versionedPerson = new VersionedPerson(); + versionedPerson.setVersion(2); + + repository.save(versionedPerson); + + verify(cassandraOperations).update(versionedPerson); + } + @Test // DATACASS-428, DATACASS-560, DATACASS-573 public void saveShouldUpdateNewEntity() { @@ -166,4 +203,12 @@ public class SimpleCassandraRepositoryUnitTests { @Id String id; } + + @Data + static class VersionedPerson { + + @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 new file mode 100644 index 000000000..e56499b24 --- /dev/null +++ b/spring-data-cassandra/src/test/java/org/springframework/data/cassandra/repository/support/SimpleReactiveCassandraRepositoryUnitTests.java @@ -0,0 +1,114 @@ +/* + * Copyright 2019 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.springframework.data.cassandra.repository.support; + +import static org.mockito.Mockito.*; + +import lombok.Data; +import reactor.core.publisher.Mono; + +import java.io.Serializable; + +import org.junit.Before; +import org.junit.Test; +import org.junit.runner.RunWith; +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; +import org.springframework.data.cassandra.core.InsertOptions; +import org.springframework.data.cassandra.core.ReactiveCassandraOperations; +import org.springframework.data.cassandra.core.convert.MappingCassandraConverter; +import org.springframework.data.cassandra.core.mapping.CassandraMappingContext; +import org.springframework.data.cassandra.core.mapping.CassandraPersistentEntity; +import org.springframework.data.cassandra.core.mapping.UserTypeResolver; + +import com.datastax.driver.core.UserType; +import com.datastax.driver.core.querybuilder.Insert; + +/** + * Unit tests for {@link SimpleReactiveCassandraRepository}. + * + * @author Mark Paluch + */ +@RunWith(MockitoJUnitRunner.class) +@SuppressWarnings("unchecked") +public class SimpleReactiveCassandraRepositoryUnitTests { + + CassandraMappingContext mappingContext = new CassandraMappingContext(); + MappingCassandraConverter converter = new MappingCassandraConverter(mappingContext); + + SimpleReactiveCassandraRepository repository; + + @Mock ReactiveCassandraOperations cassandraOperations; + @Mock UserTypeResolver userTypeResolver; + @Mock UserType userType; + @Mock EntityWriteResult writeResult; + + @Captor ArgumentCaptor insertCaptor; + + @Before + public void before() { + mappingContext.setUserTypeResolver(userTypeResolver); + when(cassandraOperations.getConverter()).thenReturn(converter); + } + + @Test // DATACASS-576 + public void shouldInsertNewVersionedEntity() { + + when(cassandraOperations.insert(any(), any(InsertOptions.class))).thenReturn(Mono.just(writeResult)); + + CassandraPersistentEntity entity = converter.getMappingContext() + .getRequiredPersistentEntity(VersionedPerson.class); + + repository = new SimpleReactiveCassandraRepository( + new MappingCassandraEntityInformation(entity, converter), cassandraOperations); + + VersionedPerson versionedPerson = new VersionedPerson(); + + repository.save(versionedPerson); + + verify(cassandraOperations).insert(versionedPerson, InsertOptions.builder().withInsertNulls().build()); + } + + @Test // DATACASS-576 + public void shouldUpdateExistingVersionedEntity() { + + CassandraPersistentEntity entity = converter.getMappingContext() + .getRequiredPersistentEntity(VersionedPerson.class); + + repository = new SimpleReactiveCassandraRepository( + new MappingCassandraEntityInformation(entity, converter), cassandraOperations); + + VersionedPerson versionedPerson = new VersionedPerson(); + versionedPerson.setVersion(2); + + repository.save(versionedPerson); + + verify(cassandraOperations).update(versionedPerson); + } + + @Data + static class VersionedPerson { + + @Id String id; + @Version long version; + } + +} diff --git a/src/main/asciidoc/new-features.adoc b/src/main/asciidoc/new-features.adoc index 2f4572888..ee7a63076 100644 --- a/src/main/asciidoc/new-features.adoc +++ b/src/main/asciidoc/new-features.adoc @@ -10,6 +10,7 @@ This chapter summarizes changes and new features for each release. * Kotlin Coroutine extensions for `ReactiveFluentCassandraOperations`. * Lightweight transaction support via `DeleteOptions` using the Template API. * Filter conditions for lightweight transaction update and delete (`UPDATE … IF `, `DELETE … IF `). +* Optimistic Locking support. [[new-features.2-1-0]] == What's new in Spring Data for Apache Cassandra 2.1 diff --git a/src/main/asciidoc/reference/cassandra.adoc b/src/main/asciidoc/reference/cassandra.adoc index 9652a9f50..8ddc1455b 100644 --- a/src/main/asciidoc/reference/cassandra.adoc +++ b/src/main/asciidoc/reference/cassandra.adoc @@ -1167,6 +1167,45 @@ You can use the following overloaded methods to remove an object from the databa * `T` *delete* `(T entity, QueryOptions queryOptions)`: Deletes the given object applying `QueryOptions`. * `boolean` *deleteById* `(Object id, Class entityClass)`: Deletes the object using the given Id. +[[cassandra.template.optimistic-locking]] +=== Optimistic Locking + +The `@Version` annotation provides syntax similar to that of JPA in the context of Cassandra and makes sure updates are only applied to rows with a matching version. +Optimistic Locking leverages Cassandra's lightweight transactions to conditionally insert, update and delete rows. +Therefore, `INSERT` statements are executed with the `IF NOT EXISTS` condition. +For updates and deletes, the actual value of the version property is added to the `UPDATE` condition in such a way that the modification does not have any effect if another operation altered the row in the meantime. +In that case, an `OptimisticLockingFailureException` is thrown. +The following example shows these features: + +==== +[source,java] +---- +@Table +class Person { + + @Id String id; + String firstname; + String lastname; + @Version Long version; +} + +Person daenerys = template.insert(new Person("Daenerys")); <1> + +Person tmp = template.findOne(query(where("id").is(daenerys.getId())), Person.class); <2> + +daenerys.setLastname("Targaryen"); +template.save(daenerys); <3> + +template.save(tmp); // throws OptimisticLockingFailureException <4> +---- +<1> Intially insert document. `version` is set to `0`. +<2> Load the just inserted document. `version` is still `0`. +<3> Update the document with `version = 0`. Set the `lastname` and bump `version` to `1`. +<4> Try to update the previously loaded document that still has `version = 0`. The operation fails with an `OptimisticLockingFailureException`, as the current `version` is `1`. +==== + +NOTE: Optimistic Locking is only supported with single-entity operations and not for batch operations. + [[cassandra.template.query]] == Querying Rows diff --git a/src/main/asciidoc/reference/mapping.adoc b/src/main/asciidoc/reference/mapping.adoc index 2de17c6e7..e34dc6262 100644 --- a/src/main/asciidoc/reference/mapping.adoc +++ b/src/main/asciidoc/reference/mapping.adoc @@ -427,6 +427,7 @@ Types are derived from the declaration by default. * `@Tuple`: Applied at the type level to use a type as a mapped tuple. * `@Element`: Applied at the field level to specify element or field ordinals within a mapped tuple. Types are derived from the property declaration by default. +* `@Version`: Applied at field level is used for optimistic locking and checked for modification on save operations. The initial value is `zero` which is bumped automatically on every update. The mapping metadata infrastructure is defined in the separate, spring-data-commons project that is both technology- and data store-agnostic.