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.