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 2ad114746..f4f4d31a9 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 @@ -21,6 +21,9 @@ import java.util.List; import java.util.function.Consumer; import java.util.function.Function; +import org.springframework.beans.BeansException; +import org.springframework.context.ApplicationContext; +import org.springframework.context.ApplicationContextAware; import org.springframework.context.ApplicationEvent; import org.springframework.context.ApplicationEventPublisher; import org.springframework.context.ApplicationEventPublisherAware; @@ -41,16 +44,20 @@ import org.springframework.data.cassandra.core.cql.CqlIdentifier; import org.springframework.data.cassandra.core.cql.CqlProvider; import org.springframework.data.cassandra.core.cql.GuavaListenableFutureAdapter; import org.springframework.data.cassandra.core.cql.QueryOptions; +import org.springframework.data.cassandra.core.cql.WriteOptions; import org.springframework.data.cassandra.core.cql.session.DefaultSessionFactory; import org.springframework.data.cassandra.core.mapping.CassandraPersistentEntity; import org.springframework.data.cassandra.core.mapping.event.AfterConvertEvent; import org.springframework.data.cassandra.core.mapping.event.AfterDeleteEvent; import org.springframework.data.cassandra.core.mapping.event.AfterLoadEvent; import org.springframework.data.cassandra.core.mapping.event.AfterSaveEvent; +import org.springframework.data.cassandra.core.mapping.event.BeforeConvertCallback; import org.springframework.data.cassandra.core.mapping.event.BeforeDeleteEvent; +import org.springframework.data.cassandra.core.mapping.event.BeforeSaveCallback; import org.springframework.data.cassandra.core.mapping.event.BeforeSaveEvent; import org.springframework.data.cassandra.core.query.Query; import org.springframework.data.domain.Slice; +import org.springframework.data.mapping.callback.EntityCallbacks; import org.springframework.data.projection.ProjectionFactory; import org.springframework.data.projection.SpelAwareProxyProjectionFactory; import org.springframework.lang.Nullable; @@ -89,7 +96,8 @@ import com.datastax.driver.core.querybuilder.Update; * @see org.springframework.data.cassandra.core.AsyncCassandraOperations * @since 2.0 */ -public class AsyncCassandraTemplate implements AsyncCassandraOperations, ApplicationEventPublisherAware { +public class AsyncCassandraTemplate + implements AsyncCassandraOperations, ApplicationEventPublisherAware, ApplicationContextAware { private final AsyncCqlOperations cqlOperations; @@ -105,6 +113,8 @@ public class AsyncCassandraTemplate implements AsyncCassandraOperations, Applica private @Nullable ApplicationEventPublisher eventPublisher; + private @Nullable EntityCallbacks entityCallbacks; + /** * Creates an instance of {@link AsyncCassandraTemplate} initialized with the given {@link Session} and a default * {@link MappingCassandraConverter}. @@ -176,6 +186,29 @@ public class AsyncCassandraTemplate implements AsyncCassandraOperations, Applica this.eventPublisher = applicationEventPublisher; } + /* (non-Javadoc) + * @see org.springframework.context.ApplicationContextAware(org.springframework.context.ApplicationContext) + */ + @Override + public void setApplicationContext(ApplicationContext applicationContext) throws BeansException { + + if (entityCallbacks == null) { + setEntityCallbacks(EntityCallbacks.create(applicationContext)); + } + + projectionFactory.setBeanFactory(applicationContext); + projectionFactory.setBeanClassLoader(applicationContext.getClassLoader()); + } + + /** + * Configure {@link EntityCallbacks} to pre-/post-process entities during persistence operations. + * + * @param entityCallbacks + */ + public void setEntityCallbacks(@Nullable EntityCallbacks entityCallbacks) { + this.entityCallbacks = entityCallbacks; + } + /* (non-Javadoc) * @see org.springframework.data.cassandra.core.AsyncCassandraOperations#getAsyncCqlOperations() */ @@ -541,9 +574,14 @@ public class AsyncCassandraTemplate implements AsyncCassandraOperations, Applica Assert.notNull(entity, "Entity must not be null"); Assert.notNull(options, "InsertOptions must not be null"); - AdaptibleEntity source = getEntityOperations().forEntity(entity, getConverter().getConversionService()); + return doInsert(entity, options, getTableName(entity.getClass())); + } + + private ListenableFuture> doInsert(T entity, WriteOptions options, CqlIdentifier tableName) { + + AdaptibleEntity source = getEntityOperations().forEntity(maybeCallBeforeConvert(entity, tableName), + getConverter().getConversionService()); CassandraPersistentEntity persistentEntity = getRequiredPersistentEntity(entity.getClass()); - CqlIdentifier tableName = persistentEntity.getTableName(); T entityToUse = source.isVersionedEntity() ? source.initializeVersionProperty() : entity; @@ -595,24 +633,26 @@ public class AsyncCassandraTemplate implements AsyncCassandraOperations, Applica CassandraPersistentEntity persistentEntity = getRequiredPersistentEntity(entity.getClass()); CqlIdentifier tableName = persistentEntity.getTableName(); - return source.isVersionedEntity() ? doUpdateVersioned(source, options, tableName, persistentEntity) - : doUpdate(entity, options, tableName, persistentEntity); + T entityToUpdate = maybeCallBeforeConvert(entity, tableName); + + return source.isVersionedEntity() ? doUpdateVersioned(entityToUpdate, options, tableName, persistentEntity) + : doUpdate(entityToUpdate, options, tableName, persistentEntity); } - private ListenableFuture> doUpdateVersioned(AdaptibleEntity source, UpdateOptions options, + private ListenableFuture> doUpdateVersioned(T entity, UpdateOptions options, CqlIdentifier tableName, CassandraPersistentEntity persistentEntity) { + AdaptibleEntity source = getEntityOperations().forEntity(entity, getConverter().getConversionService()); Number previousVersion = source.getVersion(); + T toSave = source.incrementVersion(); - T entity = source.incrementVersion(); + Update update = getStatementFactory().update(toSave, options, getConverter(), persistentEntity, tableName); - Update update = getStatementFactory().update(entity, options, getConverter(), persistentEntity, tableName); - - return executeSave(entity, tableName, source.appendVersionCondition(update, previousVersion), result -> { + return executeSave(toSave, 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, + String.format("Cannot save entity %s with version %s to table %s. Has it been modified meanwhile?", toSave, source.getVersion(), tableName)); } }); @@ -726,16 +766,17 @@ public class AsyncCassandraTemplate implements AsyncCassandraOperations, Applica Consumer beforeAfterSaveEvent) { maybeEmitEvent(new BeforeSaveEvent<>(entity, tableName, statement)); + T entityToSave = maybeCallBeforeSave(entity, tableName, statement); ListenableFuture result = getAsyncCqlOperations().execute(new AsyncStatementCallback(statement)); return new MappingListenableFutureAdapter<>(result, resultSet -> { - EntityWriteResult writeResult = EntityWriteResult.of(resultSet, entity); + EntityWriteResult writeResult = EntityWriteResult.of(resultSet, entityToSave); beforeAfterSaveEvent.accept(writeResult); - maybeEmitEvent(new AfterSaveEvent<>(entity, tableName)); + maybeEmitEvent(new AfterSaveEvent<>(entityToSave, tableName)); return writeResult; }); @@ -825,6 +866,24 @@ public class AsyncCassandraTemplate implements AsyncCassandraOperations, Applica } } + protected T maybeCallBeforeConvert(T object, CqlIdentifier tableName) { + + if (null != entityCallbacks) { + return (T) entityCallbacks.callback(BeforeConvertCallback.class, object, tableName); + } + + return object; + } + + protected T maybeCallBeforeSave(T object, CqlIdentifier tableName, Statement statement) { + + if (null != entityCallbacks) { + return (T) entityCallbacks.callback(BeforeSaveCallback.class, object, tableName, statement); + } + + return object; + } + static class MappingListenableFutureAdapter extends org.springframework.util.concurrent.ListenableFutureAdapter { 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 9d9cf9a76..ef56d94b0 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 @@ -23,6 +23,9 @@ import java.util.function.Function; import java.util.stream.Stream; import java.util.stream.StreamSupport; +import org.springframework.beans.BeansException; +import org.springframework.context.ApplicationContext; +import org.springframework.context.ApplicationContextAware; import org.springframework.context.ApplicationEvent; import org.springframework.context.ApplicationEventPublisher; import org.springframework.context.ApplicationEventPublisherAware; @@ -49,11 +52,14 @@ import org.springframework.data.cassandra.core.mapping.event.AfterConvertEvent; import org.springframework.data.cassandra.core.mapping.event.AfterDeleteEvent; import org.springframework.data.cassandra.core.mapping.event.AfterLoadEvent; import org.springframework.data.cassandra.core.mapping.event.AfterSaveEvent; +import org.springframework.data.cassandra.core.mapping.event.BeforeConvertCallback; import org.springframework.data.cassandra.core.mapping.event.BeforeDeleteEvent; +import org.springframework.data.cassandra.core.mapping.event.BeforeSaveCallback; import org.springframework.data.cassandra.core.mapping.event.BeforeSaveEvent; import org.springframework.data.cassandra.core.query.Columns; import org.springframework.data.cassandra.core.query.Query; import org.springframework.data.domain.Slice; +import org.springframework.data.mapping.callback.EntityCallbacks; import org.springframework.data.mapping.context.MappingContext; import org.springframework.data.projection.ProjectionFactory; import org.springframework.data.projection.SpelAwareProxyProjectionFactory; @@ -92,10 +98,12 @@ import com.datastax.driver.core.querybuilder.Update; * @see org.springframework.data.cassandra.core.CassandraOperations * @since 2.0 */ -public class CassandraTemplate implements CassandraOperations, ApplicationEventPublisherAware { +public class CassandraTemplate implements CassandraOperations, ApplicationEventPublisherAware, ApplicationContextAware { private @Nullable ApplicationEventPublisher eventPublisher; + private @Nullable EntityCallbacks entityCallbacks; + private final CassandraConverter converter; private final CqlOperations cqlOperations; @@ -187,6 +195,29 @@ public class CassandraTemplate implements CassandraOperations, ApplicationEventP this.eventPublisher = applicationEventPublisher; } + /* (non-Javadoc) + * @see org.springframework.context.ApplicationContextAware(org.springframework.context.ApplicationContext) + */ + @Override + public void setApplicationContext(ApplicationContext applicationContext) throws BeansException { + + if (entityCallbacks == null) { + setEntityCallbacks(EntityCallbacks.create(applicationContext)); + } + + projectionFactory.setBeanFactory(applicationContext); + projectionFactory.setBeanClassLoader(applicationContext.getClassLoader()); + } + + /** + * Configure {@link EntityCallbacks} to pre-/post-process entities during persistence operations. + * + * @param entityCallbacks + */ + public void setEntityCallbacks(@Nullable EntityCallbacks entityCallbacks) { + this.entityCallbacks = entityCallbacks; + } + /* (non-Javadoc) * @see org.springframework.data.cassandra.core.CassandraOperations#getConverter() */ @@ -603,7 +634,8 @@ public class CassandraTemplate implements CassandraOperations, ApplicationEventP EntityWriteResult doInsert(T entity, WriteOptions options, CqlIdentifier tableName) { - AdaptibleEntity source = getEntityOperations().forEntity(entity, getConverter().getConversionService()); + AdaptibleEntity source = getEntityOperations().forEntity(maybeCallBeforeConvert(entity, tableName), + getConverter().getConversionService()); CassandraPersistentEntity persistentEntity = getRequiredPersistentEntity(entity.getClass()); T entityToUse = source.isVersionedEntity() ? source.initializeVersionProperty() : entity; @@ -653,24 +685,27 @@ public class CassandraTemplate implements CassandraOperations, ApplicationEventP CassandraPersistentEntity persistentEntity = getRequiredPersistentEntity(entity.getClass()); CqlIdentifier tableName = persistentEntity.getTableName(); - return source.isVersionedEntity() ? doUpdateVersioned(source, options, tableName, persistentEntity) - : doUpdate(entity, options, tableName, persistentEntity); + T entityToUpdate = maybeCallBeforeConvert(entity, tableName); + + return source.isVersionedEntity() ? doUpdateVersioned(entityToUpdate, options, tableName, persistentEntity) + : doUpdate(entityToUpdate, options, tableName, persistentEntity); } - private EntityWriteResult doUpdateVersioned(AdaptibleEntity source, UpdateOptions options, - CqlIdentifier tableName, CassandraPersistentEntity persistentEntity) { + private EntityWriteResult doUpdateVersioned(T entity, UpdateOptions options, CqlIdentifier tableName, + CassandraPersistentEntity persistentEntity) { + + AdaptibleEntity source = getEntityOperations().forEntity(entity, getConverter().getConversionService()); Number previousVersion = source.getVersion(); + T toSave = source.incrementVersion(); - T entity = source.incrementVersion(); + Update update = getStatementFactory().update(toSave, options, getConverter(), persistentEntity, tableName); - Update update = getStatementFactory().update(entity, options, getConverter(), persistentEntity, tableName); - - return executeSave(entity, tableName, source.appendVersionCondition(update, previousVersion), result -> { + return executeSave(toSave, 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, + String.format("Cannot save entity %s with version %s to table %s. Has it been modified meanwhile?", toSave, source.getVersion(), tableName)); } }); @@ -818,13 +853,14 @@ public class CassandraTemplate implements CassandraOperations, ApplicationEventP Consumer resultConsumer) { maybeEmitEvent(new BeforeSaveEvent<>(entity, tableName, statement)); + T entityToSave = maybeCallBeforeSave(entity, tableName, statement); WriteResult result = getCqlOperations().execute(new StatementCallback(statement)); resultConsumer.accept(result); - maybeEmitEvent(new AfterSaveEvent<>(entity, tableName)); + maybeEmitEvent(new AfterSaveEvent<>(entityToSave, tableName)); - return EntityWriteResult.of(result, entity); + return EntityWriteResult.of(result, entityToSave); } private WriteResult executeDelete(Object entity, CqlIdentifier tableName, Statement statement, @@ -903,6 +939,24 @@ public class CassandraTemplate implements CassandraOperations, ApplicationEventP } } + protected T maybeCallBeforeConvert(T object, CqlIdentifier tableName) { + + if (null != entityCallbacks) { + return (T) entityCallbacks.callback(BeforeConvertCallback.class, object, tableName); + } + + return object; + } + + protected T maybeCallBeforeSave(T object, CqlIdentifier tableName, Statement statement) { + + if (null != entityCallbacks) { + return (T) entityCallbacks.callback(BeforeSaveCallback.class, object, tableName, statement); + } + + return object; + } + @Value static class StatementCallback implements SessionCallback, CqlProvider { 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 f2aa2b2a2..5bda49618 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 @@ -26,6 +26,9 @@ import java.util.function.Function; import org.reactivestreams.Publisher; +import org.springframework.beans.BeansException; +import org.springframework.context.ApplicationContext; +import org.springframework.context.ApplicationContextAware; import org.springframework.context.ApplicationEvent; import org.springframework.context.ApplicationEventPublisher; import org.springframework.context.ApplicationEventPublisherAware; @@ -56,10 +59,14 @@ import org.springframework.data.cassandra.core.mapping.event.AfterLoadEvent; import org.springframework.data.cassandra.core.mapping.event.AfterSaveEvent; import org.springframework.data.cassandra.core.mapping.event.BeforeDeleteEvent; import org.springframework.data.cassandra.core.mapping.event.BeforeSaveEvent; +import org.springframework.data.cassandra.core.mapping.event.ReactiveBeforeConvertCallback; +import org.springframework.data.cassandra.core.mapping.event.ReactiveBeforeSaveCallback; import org.springframework.data.cassandra.core.query.Columns; import org.springframework.data.cassandra.core.query.Query; import org.springframework.data.domain.Slice; import org.springframework.data.domain.SliceImpl; +import org.springframework.data.mapping.callback.EntityCallbacks; +import org.springframework.data.mapping.callback.ReactiveEntityCallbacks; import org.springframework.data.projection.ProjectionFactory; import org.springframework.data.projection.SpelAwareProxyProjectionFactory; import org.springframework.lang.Nullable; @@ -96,10 +103,13 @@ import com.datastax.driver.core.querybuilder.Update; * @author Hleb Albau * @since 2.0 */ -public class ReactiveCassandraTemplate implements ReactiveCassandraOperations, ApplicationEventPublisherAware { +public class ReactiveCassandraTemplate + implements ReactiveCassandraOperations, ApplicationEventPublisherAware, ApplicationContextAware { private @Nullable ApplicationEventPublisher eventPublisher; + private @Nullable ReactiveEntityCallbacks entityCallbacks; + private final CassandraConverter converter; private final EntityOperations entityOperations; @@ -189,6 +199,29 @@ public class ReactiveCassandraTemplate implements ReactiveCassandraOperations, A this.eventPublisher = applicationEventPublisher; } + /* (non-Javadoc) + * @see org.springframework.context.ApplicationContextAware(org.springframework.context.ApplicationContext) + */ + @Override + public void setApplicationContext(ApplicationContext applicationContext) throws BeansException { + + if (entityCallbacks == null) { + setEntityCallbacks(ReactiveEntityCallbacks.create(applicationContext)); + } + + projectionFactory.setBeanFactory(applicationContext); + projectionFactory.setBeanClassLoader(applicationContext.getClassLoader()); + } + + /** + * Configure {@link EntityCallbacks} to pre-/post-process entities during persistence operations. + * + * @param entityCallbacks + */ + public void setEntityCallbacks(@Nullable ReactiveEntityCallbacks entityCallbacks) { + this.entityCallbacks = entityCallbacks; + } + /* (non-Javadoc) * @see org.springframework.data.cassandra.core.ReactiveCassandraOperations#getConverter() */ @@ -538,16 +571,20 @@ public class ReactiveCassandraTemplate implements ReactiveCassandraOperations, A Mono> doInsert(T entity, WriteOptions options, CqlIdentifier tableName) { - AdaptibleEntity source = this.entityOperations.forEntity(entity, getConverter().getConversionService()); - CassandraPersistentEntity persistentEntity = getRequiredPersistentEntity(entity.getClass()); + return maybeCallBeforeConvert(entity, tableName).flatMap(entityToInsert -> { - T entityToUse = source.isVersionedEntity() ? source.initializeVersionProperty() : entity; + AdaptibleEntity source = this.entityOperations.forEntity(entityToInsert, + getConverter().getConversionService()); + CassandraPersistentEntity persistentEntity = getRequiredPersistentEntity(entityToInsert.getClass()); - Insert insert = EntityQueryUtils.createInsertQuery(tableName.toCql(), entityToUse, options, getConverter(), - persistentEntity); + T entityToUse = source.isVersionedEntity() ? source.initializeVersionProperty() : entityToInsert; - return source.isVersionedEntity() ? doInsertVersioned(insert.ifNotExists(), entityToUse, source, tableName) - : doInsert(insert, entityToUse, tableName); + Insert insert = EntityQueryUtils.createInsertQuery(tableName.toCql(), entityToUse, options, getConverter(), + persistentEntity); + + return source.isVersionedEntity() ? doInsertVersioned(insert.ifNotExists(), entityToUse, source, tableName) + : doInsert(insert, entityToUse, tableName); + }); } private Mono> doInsertVersioned(Insert insert, T entity, AdaptibleEntity source, @@ -593,25 +630,28 @@ public class ReactiveCassandraTemplate implements ReactiveCassandraOperations, A CassandraPersistentEntity persistentEntity = getRequiredPersistentEntity(entity.getClass()); CqlIdentifier tableName = persistentEntity.getTableName(); - return source.isVersionedEntity() ? doUpdateVersioned(source, options, tableName, persistentEntity) - : doUpdate(entity, options, tableName, persistentEntity); + return maybeCallBeforeConvert(entity, tableName).flatMap(entityToUpdate -> { + return source.isVersionedEntity() ? doUpdateVersioned(entity, options, tableName, persistentEntity) + : doUpdate(entity, options, tableName, persistentEntity); + }); } - private Mono> doUpdateVersioned(AdaptibleEntity source, UpdateOptions options, - CqlIdentifier tableName, CassandraPersistentEntity persistentEntity) { + private Mono> doUpdateVersioned(T entity, UpdateOptions options, CqlIdentifier tableName, + CassandraPersistentEntity persistentEntity) { + + AdaptibleEntity source = getEntityOperations().forEntity(entity, getConverter().getConversionService()); Number previousVersion = source.getVersion(); + T toSave = source.incrementVersion(); - T entity = source.incrementVersion(); + Update update = getStatementFactory().update(toSave, options, getConverter(), persistentEntity, tableName); - Update update = getStatementFactory().update(entity, options, getConverter(), persistentEntity, tableName); - - return executeSave(entity, tableName, source.appendVersionCondition(update, previousVersion), (result, sink) -> { + return executeSave(toSave, tableName, source.appendVersionCondition(update, previousVersion), (result, sink) -> { if (!result.wasApplied()) { sink.error(new OptimisticLockingFailureException( - String.format("Cannot save entity %s with version %s to table %s. Has it been modified meanwhile?", entity, + String.format("Cannot save entity %s with version %s to table %s. Has it been modified meanwhile?", toSave, source.getVersion(), tableName))); return; @@ -763,14 +803,18 @@ public class ReactiveCassandraTemplate implements ReactiveCassandraOperations, A private Mono> executeSave(T entity, CqlIdentifier tableName, Statement statement, BiConsumer, SynchronousSink>> handler) { - maybeEmitEvent(new BeforeSaveEvent<>(entity, tableName, statement)); + return Mono.defer(() -> { - Flux execute = getReactiveCqlOperations().execute(new StatementCallback(statement)); + maybeEmitEvent(new BeforeSaveEvent<>(entity, tableName, statement)); + + return maybeCallBeforeSave(entity, tableName, statement).flatMapMany(entityToSave -> { + Flux execute = getReactiveCqlOperations().execute(new StatementCallback(statement)); + + return execute.map(it -> EntityWriteResult.of(it, entityToSave)).handle(handler) // + .doOnNext(it -> maybeEmitEvent(new AfterSaveEvent<>(entityToSave, tableName))); + }).next(); + }); - 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, @@ -786,7 +830,6 @@ public class ReactiveCassandraTemplate implements ReactiveCassandraOperations, A .next(); } - @SuppressWarnings("ConstantConditions") private Mono getEffectiveFetchSize(Statement statement) { if (statement.getFetchSize() > 0) { @@ -843,6 +886,24 @@ public class ReactiveCassandraTemplate implements ReactiveCassandraOperations, A } } + protected Mono maybeCallBeforeConvert(T object, CqlIdentifier tableName) { + + if (null != entityCallbacks) { + return entityCallbacks.callback(ReactiveBeforeConvertCallback.class, object, tableName); + } + + return Mono.just(object); + } + + protected Mono maybeCallBeforeSave(T object, CqlIdentifier tableName, Statement statement) { + + if (null != entityCallbacks) { + return entityCallbacks.callback(ReactiveBeforeSaveCallback.class, object, tableName, statement); + } + + return Mono.just(object); + } + @Value static class StatementCallback implements ReactiveSessionCallback, CqlProvider { diff --git a/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/mapping/event/BeforeConvertCallback.java b/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/mapping/event/BeforeConvertCallback.java new file mode 100644 index 000000000..ceb7461b5 --- /dev/null +++ b/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/mapping/event/BeforeConvertCallback.java @@ -0,0 +1,40 @@ +/* + * 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 + * + * https://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.mapping.event; + +import org.springframework.data.cassandra.core.cql.CqlIdentifier; +import org.springframework.data.mapping.callback.EntityCallback; + +/** + * Callback being invoked before a domain object is converted to be persisted. + * + * @author Mark Paluch + * @since 2.2 + * @see org.springframework.data.mapping.callback.EntityCallbacks + */ +@FunctionalInterface +public interface BeforeConvertCallback extends EntityCallback { + + /** + * Entity callback method invoked before a domain object is converted to be persisted. Can return either the same or a + * modified instance of the domain object. + * + * @param entity the domain object to save. + * @param tableName name of the table. + * @return the domain object to be persisted. + */ + T onBeforeConvert(T entity, CqlIdentifier tableName); +} diff --git a/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/mapping/event/BeforeSaveCallback.java b/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/mapping/event/BeforeSaveCallback.java new file mode 100644 index 000000000..1a050dcc2 --- /dev/null +++ b/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/mapping/event/BeforeSaveCallback.java @@ -0,0 +1,44 @@ +/* + * 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 + * + * https://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.mapping.event; + +import org.springframework.data.cassandra.core.cql.CqlIdentifier; +import org.springframework.data.mapping.callback.EntityCallback; + +import com.datastax.driver.core.Statement; + +/** + * Entity callback triggered before save of a row. + * + * @author Mark Paluch + * @since 2.2 + * @see org.springframework.data.mapping.callback.EntityCallbacks + */ +@FunctionalInterface +public interface BeforeSaveCallback extends EntityCallback { + + /** + * Entity callback method invoked before a domain object is saved. Can return either the same of a modified instance + * of the domain object and can modify {@link Statement} contents. This method is called after converting the + * {@code entity} to {@link Statement} so effectively the row is used as outcome of invoking this callback. + * + * @param entity the domain object to save. + * @param tableName name of the table. + * @param statement {@link Statement} representing the {@code entity} operation. + * @return the domain object to be persisted. + */ + T onBeforeSave(T entity, CqlIdentifier tableName, Statement statement); +} diff --git a/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/mapping/event/ReactiveBeforeConvertCallback.java b/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/mapping/event/ReactiveBeforeConvertCallback.java new file mode 100644 index 000000000..d3ef50f84 --- /dev/null +++ b/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/mapping/event/ReactiveBeforeConvertCallback.java @@ -0,0 +1,42 @@ +/* + * 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 + * + * https://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.mapping.event; + +import org.reactivestreams.Publisher; + +import org.springframework.data.cassandra.core.cql.CqlIdentifier; +import org.springframework.data.mapping.callback.EntityCallback; + +/** + * Callback being invoked before a domain object is converted to be persisted. + * + * @author Mark Paluch + * @since 2.2 + * @see org.springframework.data.mapping.callback.ReactiveEntityCallbacks + */ +@FunctionalInterface +public interface ReactiveBeforeConvertCallback extends EntityCallback { + + /** + * Entity callback method invoked before a domain object is converted to be persisted. Can return either the same or a + * modified instance of the domain object. + * + * @param entity the domain object to save. + * @param tableName name of the table. + * @return a {@link Publisher} emitting the domain object to be persisted. + */ + Publisher onBeforeConvert(T entity, CqlIdentifier tableName); +} diff --git a/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/mapping/event/ReactiveBeforeSaveCallback.java b/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/mapping/event/ReactiveBeforeSaveCallback.java new file mode 100644 index 000000000..41f03a39e --- /dev/null +++ b/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/mapping/event/ReactiveBeforeSaveCallback.java @@ -0,0 +1,46 @@ +/* + * 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 + * + * https://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.mapping.event; + +import org.reactivestreams.Publisher; + +import org.springframework.data.cassandra.core.cql.CqlIdentifier; +import org.springframework.data.mapping.callback.EntityCallback; + +import com.datastax.driver.core.Statement; + +/** + * Entity callback triggered before save of a row. + * + * @author Mark Paluch + * @since 2.2 + * @see org.springframework.data.mapping.callback.ReactiveEntityCallbacks + */ +@FunctionalInterface +public interface ReactiveBeforeSaveCallback extends EntityCallback { + + /** + * Entity callback method invoked before a domain object is saved. Can return either the same of a modified instance + * of the domain object and can modify {@link Statement} contents. This method is called after converting the + * {@code entity} to {@link Statement} so effectively the row is used as outcome of invoking this callback. + * + * @param entity the domain object to save. + * @param tableName name of the table. + * @param statement {@link Statement} representing the {@code entity} operation. + * @return a {@link Publisher} emitting the domain object to be persisted. + */ + Publisher onBeforeSave(T entity, CqlIdentifier tableName, Statement statement); +} diff --git a/spring-data-cassandra/src/test/java/org/springframework/data/cassandra/config/AbstractAuditingTests.java b/spring-data-cassandra/src/test/java/org/springframework/data/cassandra/config/AbstractAuditingTests.java new file mode 100644 index 000000000..e69de29bb diff --git a/spring-data-cassandra/src/test/java/org/springframework/data/cassandra/core/AsyncCassandraTemplateUnitTests.java b/spring-data-cassandra/src/test/java/org/springframework/data/cassandra/core/AsyncCassandraTemplateUnitTests.java index ca43eaa2b..68ef8ee27 100644 --- a/spring-data-cassandra/src/test/java/org/springframework/data/cassandra/core/AsyncCassandraTemplateUnitTests.java +++ b/spring-data-cassandra/src/test/java/org/springframework/data/cassandra/core/AsyncCassandraTemplateUnitTests.java @@ -15,15 +15,10 @@ */ package org.springframework.data.cassandra.core; -import static org.assertj.core.api.Assertions.assertThat; -import static org.assertj.core.api.Assertions.fail; -import static org.mockito.ArgumentMatchers.any; -import static org.mockito.ArgumentMatchers.anyInt; -import static org.mockito.ArgumentMatchers.anyString; -import static org.mockito.Mockito.reset; -import static org.mockito.Mockito.verify; -import static org.mockito.Mockito.when; -import static org.springframework.data.cassandra.core.query.Criteria.where; +import static org.assertj.core.api.Assertions.*; +import static org.mockito.ArgumentMatchers.*; +import static org.mockito.Mockito.*; +import static org.springframework.data.cassandra.core.query.Criteria.*; import java.util.ArrayList; import java.util.Collections; @@ -42,10 +37,14 @@ import org.mockito.Mock; import org.mockito.junit.MockitoJUnitRunner; import org.springframework.data.cassandra.CassandraConnectionFailureException; +import org.springframework.data.cassandra.core.mapping.event.BeforeConvertCallback; +import org.springframework.data.cassandra.core.mapping.event.BeforeSaveCallback; import org.springframework.data.cassandra.core.query.Filter; import org.springframework.data.cassandra.core.query.Query; import org.springframework.data.cassandra.core.query.Update; import org.springframework.data.cassandra.domain.User; +import org.springframework.data.cassandra.domain.VersionedUser; +import org.springframework.data.mapping.callback.EntityCallbacks; import org.springframework.util.concurrent.ListenableFuture; import com.datastax.driver.core.ColumnDefinitions; @@ -75,6 +74,10 @@ public class AsyncCassandraTemplateUnitTests { AsyncCassandraTemplate template; + Object beforeSave; + + Object beforeConvert; + @Before public void setUp() { @@ -82,6 +85,24 @@ public class AsyncCassandraTemplateUnitTests { when(session.executeAsync(any(Statement.class))).thenReturn(new TestResultSetFuture(resultSet)); when(row.getColumnDefinitions()).thenReturn(columnDefinitions); + + EntityCallbacks callbacks = EntityCallbacks.create(); + callbacks.addEntityCallback((BeforeSaveCallback) (entity, tableName, statement) -> { + + assertThat(tableName).isNotNull(); + assertThat(statement).isNotNull(); + beforeSave = entity; + return entity; + }); + + callbacks.addEntityCallback((BeforeConvertCallback) (entity, tableName) -> { + + assertThat(tableName).isNotNull(); + beforeConvert = entity; + return entity; + }); + + template.setEntityCallbacks(callbacks); } @Test // DATACASS-292 @@ -256,7 +277,7 @@ public class AsyncCassandraTemplateUnitTests { assertThat(statementCaptor.getValue().toString()).isEqualTo("SELECT COUNT(1) FROM users;"); } - @Test // DATACASS-292 + @Test // DATACASS-292, DATACASS-618 public void insertShouldInsertEntity() { when(resultSet.wasApplied()).thenReturn(true); @@ -269,6 +290,25 @@ public class AsyncCassandraTemplateUnitTests { verify(session).executeAsync(statementCaptor.capture()); assertThat(statementCaptor.getValue().toString()) .isEqualTo("INSERT INTO users (firstname,id,lastname) VALUES ('Walter','heisenberg','White');"); + assertThat(beforeConvert).isSameAs(user); + assertThat(beforeSave).isSameAs(user); + } + + @Test // DATACASS-618 + public void insertShouldInsertVersionedEntity() { + + when(resultSet.wasApplied()).thenReturn(true); + + VersionedUser user = new VersionedUser("heisenberg", "Walter", "White"); + + ListenableFuture future = template.insert(user); + + assertThat(getUninterruptibly(future)).isEqualTo(user); + verify(session).executeAsync(statementCaptor.capture()); + assertThat(statementCaptor.getValue().toString()).isEqualTo( + "INSERT INTO vusers (firstname,id,lastname,version) VALUES ('Walter','heisenberg','White',0) IF NOT EXISTS;"); + assertThat(beforeConvert).isSameAs(user); + assertThat(beforeSave).isSameAs(user); } @Test // DATACASS-292 @@ -290,7 +330,7 @@ public class AsyncCassandraTemplateUnitTests { } } - @Test // DATACASS-292 + @Test // DATACASS-292, DATACASS-618 public void updateShouldUpdateEntity() { when(resultSet.wasApplied()).thenReturn(true); @@ -303,6 +343,26 @@ public class AsyncCassandraTemplateUnitTests { verify(session).executeAsync(statementCaptor.capture()); assertThat(statementCaptor.getValue().toString()) .isEqualTo("UPDATE users SET firstname='Walter',lastname='White' WHERE id='heisenberg';"); + assertThat(beforeConvert).isSameAs(user); + assertThat(beforeSave).isSameAs(user); + } + + @Test // DATACASS-618 + public void updateShouldUpdateVersionedEntity() { + + when(resultSet.wasApplied()).thenReturn(true); + + VersionedUser user = new VersionedUser("heisenberg", "Walter", "White"); + user.setVersion(0L); + + ListenableFuture future = template.update(user); + + assertThat(getUninterruptibly(future)).isEqualTo(user); + verify(session).executeAsync(statementCaptor.capture()); + assertThat(statementCaptor.getValue().toString()).isEqualTo( + "UPDATE vusers SET firstname='Walter',lastname='White',version=1 WHERE id='heisenberg' IF version=0;"); + assertThat(beforeConvert).isSameAs(user); + assertThat(beforeSave).isSameAs(user); } @Test // DATACASS-575 @@ -357,8 +417,8 @@ public class AsyncCassandraTemplateUnitTests { template.update(query, update, User.class); verify(session).executeAsync(statementCaptor.capture()); - assertThat(statementCaptor.getValue().toString()) - .isEqualTo("UPDATE users SET firstname='Walter' WHERE id='heisenberg' IF firstname='Walter' AND lastname='White';"); + assertThat(statementCaptor.getValue().toString()).isEqualTo( + "UPDATE users SET firstname='Walter' WHERE id='heisenberg' IF firstname='Walter' AND lastname='White';"); } @Test // DATACASS-292 diff --git a/spring-data-cassandra/src/test/java/org/springframework/data/cassandra/core/CassandraTemplateUnitTests.java b/spring-data-cassandra/src/test/java/org/springframework/data/cassandra/core/CassandraTemplateUnitTests.java index 63bc5d7bd..8c6828495 100644 --- a/spring-data-cassandra/src/test/java/org/springframework/data/cassandra/core/CassandraTemplateUnitTests.java +++ b/spring-data-cassandra/src/test/java/org/springframework/data/cassandra/core/CassandraTemplateUnitTests.java @@ -15,15 +15,10 @@ */ package org.springframework.data.cassandra.core; -import static org.assertj.core.api.Assertions.assertThat; -import static org.assertj.core.api.Assertions.fail; -import static org.mockito.ArgumentMatchers.any; -import static org.mockito.ArgumentMatchers.anyInt; -import static org.mockito.ArgumentMatchers.anyString; -import static org.mockito.Mockito.reset; -import static org.mockito.Mockito.verify; -import static org.mockito.Mockito.when; -import static org.springframework.data.cassandra.core.query.Criteria.where; +import static org.assertj.core.api.Assertions.*; +import static org.mockito.ArgumentMatchers.*; +import static org.mockito.Mockito.*; +import static org.springframework.data.cassandra.core.query.Criteria.*; import java.util.Collections; import java.util.List; @@ -39,10 +34,14 @@ import org.mockito.Mockito; import org.mockito.junit.MockitoJUnitRunner; import org.springframework.data.cassandra.CassandraConnectionFailureException; +import org.springframework.data.cassandra.core.mapping.event.BeforeConvertCallback; +import org.springframework.data.cassandra.core.mapping.event.BeforeSaveCallback; import org.springframework.data.cassandra.core.query.Filter; import org.springframework.data.cassandra.core.query.Query; import org.springframework.data.cassandra.core.query.Update; import org.springframework.data.cassandra.domain.User; +import org.springframework.data.cassandra.domain.VersionedUser; +import org.springframework.data.mapping.callback.EntityCallbacks; import com.datastax.driver.core.ColumnDefinitions; import com.datastax.driver.core.DataType; @@ -70,6 +69,10 @@ public class CassandraTemplateUnitTests { CassandraTemplate template; + Object beforeSave; + + Object beforeConvert; + @Before public void setUp() { @@ -77,6 +80,24 @@ public class CassandraTemplateUnitTests { when(session.execute(any(Statement.class))).thenReturn(resultSet); when(row.getColumnDefinitions()).thenReturn(columnDefinitions); + + EntityCallbacks callbacks = EntityCallbacks.create(); + callbacks.addEntityCallback((BeforeSaveCallback) (entity, tableName, statement) -> { + + assertThat(tableName).isNotNull(); + assertThat(statement).isNotNull(); + beforeSave = entity; + return entity; + }); + + callbacks.addEntityCallback((BeforeConvertCallback) (entity, tableName) -> { + + assertThat(tableName).isNotNull(); + beforeConvert = entity; + return entity; + }); + + template.setEntityCallbacks(callbacks); } @Test // DATACASS-292 @@ -241,7 +262,7 @@ public class CassandraTemplateUnitTests { assertThat(statementCaptor.getValue().toString()).isEqualTo("SELECT COUNT(1) FROM users;"); } - @Test // DATACASS-292 + @Test // DATACASS-292, DATACASS-618 public void insertShouldInsertEntity() { when(resultSet.wasApplied()).thenReturn(true); @@ -253,6 +274,24 @@ public class CassandraTemplateUnitTests { verify(session).execute(statementCaptor.capture()); assertThat(statementCaptor.getValue().toString()) .isEqualTo("INSERT INTO users (firstname,id,lastname) VALUES ('Walter','heisenberg','White');"); + assertThat(beforeConvert).isSameAs(user); + assertThat(beforeSave).isSameAs(user); + } + + @Test // DATACASS-618 + public void insertShouldInsertVersionedEntity() { + + when(resultSet.wasApplied()).thenReturn(true); + + VersionedUser user = new VersionedUser("heisenberg", "Walter", "White"); + + template.insert(user); + + verify(session).execute(statementCaptor.capture()); + assertThat(statementCaptor.getValue().toString()).isEqualTo( + "INSERT INTO vusers (firstname,id,lastname,version) VALUES ('Walter','heisenberg','White',0) IF NOT EXISTS;"); + assertThat(beforeConvert).isSameAs(user); + assertThat(beforeSave).isSameAs(user); } @Test // DATACASS-250 @@ -314,7 +353,7 @@ public class CassandraTemplateUnitTests { assertThat(writeResult.wasApplied()).isFalse(); } - @Test // DATACASS-292 + @Test // DATACASS-292, DATACASS-618 public void updateShouldUpdateEntity() { when(resultSet.wasApplied()).thenReturn(true); @@ -326,6 +365,25 @@ public class CassandraTemplateUnitTests { verify(session).execute(statementCaptor.capture()); assertThat(statementCaptor.getValue().toString()) .isEqualTo("UPDATE users SET firstname='Walter',lastname='White' WHERE id='heisenberg';"); + assertThat(beforeConvert).isSameAs(user); + assertThat(beforeSave).isSameAs(user); + } + + @Test // DATACASS-618 + public void updateShouldUpdateVersionedEntity() { + + when(resultSet.wasApplied()).thenReturn(true); + + VersionedUser user = new VersionedUser("heisenberg", "Walter", "White"); + user.setVersion(0L); + + template.update(user); + + verify(session).execute(statementCaptor.capture()); + assertThat(statementCaptor.getValue().toString()).isEqualTo( + "UPDATE vusers SET firstname='Walter',lastname='White',version=1 WHERE id='heisenberg' IF version=0;"); + assertThat(beforeConvert).isSameAs(user); + assertThat(beforeSave).isSameAs(user); } @Test // DATACASS-250 @@ -384,8 +442,8 @@ public class CassandraTemplateUnitTests { template.update(query, update, User.class); verify(session).execute(statementCaptor.capture()); - assertThat(statementCaptor.getValue().toString()) - .isEqualTo("UPDATE users SET firstname='Walter' WHERE id='heisenberg' IF firstname='Walter' AND lastname='White';"); + assertThat(statementCaptor.getValue().toString()).isEqualTo( + "UPDATE users SET firstname='Walter' WHERE id='heisenberg' IF firstname='Walter' AND lastname='White';"); } @Test // DATACASS-292 diff --git a/spring-data-cassandra/src/test/java/org/springframework/data/cassandra/core/ReactiveCassandraTemplateUnitTests.java b/spring-data-cassandra/src/test/java/org/springframework/data/cassandra/core/ReactiveCassandraTemplateUnitTests.java index f395319e4..d2a64fe3a 100644 --- a/spring-data-cassandra/src/test/java/org/springframework/data/cassandra/core/ReactiveCassandraTemplateUnitTests.java +++ b/spring-data-cassandra/src/test/java/org/springframework/data/cassandra/core/ReactiveCassandraTemplateUnitTests.java @@ -15,21 +15,17 @@ */ package org.springframework.data.cassandra.core; -import static org.assertj.core.api.Assertions.assertThat; -import static org.mockito.ArgumentMatchers.any; -import static org.mockito.ArgumentMatchers.anyInt; -import static org.mockito.ArgumentMatchers.anyString; -import static org.mockito.Mockito.reset; -import static org.mockito.Mockito.verify; -import static org.mockito.Mockito.when; -import static org.springframework.data.cassandra.core.query.Criteria.where; - -import java.util.Collections; +import static org.assertj.core.api.Assertions.*; +import static org.mockito.ArgumentMatchers.*; +import static org.mockito.Mockito.*; +import static org.springframework.data.cassandra.core.query.Criteria.*; import reactor.core.publisher.Flux; import reactor.core.publisher.Mono; import reactor.test.StepVerifier; +import java.util.Collections; + import org.junit.Before; import org.junit.Test; import org.junit.runner.RunWith; @@ -40,10 +36,14 @@ import org.mockito.junit.MockitoJUnitRunner; import org.springframework.data.cassandra.ReactiveResultSet; import org.springframework.data.cassandra.ReactiveSession; +import org.springframework.data.cassandra.core.mapping.event.ReactiveBeforeConvertCallback; +import org.springframework.data.cassandra.core.mapping.event.ReactiveBeforeSaveCallback; import org.springframework.data.cassandra.core.query.Filter; import org.springframework.data.cassandra.core.query.Query; import org.springframework.data.cassandra.core.query.Update; import org.springframework.data.cassandra.domain.User; +import org.springframework.data.cassandra.domain.VersionedUser; +import org.springframework.data.mapping.callback.ReactiveEntityCallbacks; import com.datastax.driver.core.ColumnDefinitions; import com.datastax.driver.core.DataType; @@ -68,6 +68,10 @@ public class ReactiveCassandraTemplateUnitTests { ReactiveCassandraTemplate template; + Object beforeSave; + + Object beforeConvert; + @Before public void setUp() { @@ -75,6 +79,24 @@ public class ReactiveCassandraTemplateUnitTests { when(session.execute(any(Statement.class))).thenReturn(Mono.just(reactiveResultSet)); when(row.getColumnDefinitions()).thenReturn(columnDefinitions); + + ReactiveEntityCallbacks callbacks = ReactiveEntityCallbacks.create(); + callbacks.addEntityCallback((ReactiveBeforeSaveCallback) (entity, tableName, statement) -> { + + assertThat(tableName).isNotNull(); + assertThat(statement).isNotNull(); + beforeSave = entity; + return Mono.just(entity); + }); + + callbacks.addEntityCallback((ReactiveBeforeConvertCallback) (entity, tableName) -> { + + assertThat(tableName).isNotNull(); + beforeConvert = entity; + return Mono.just(entity); + }); + + template.setEntityCallbacks(callbacks); } @Test // DATACASS-335 @@ -226,7 +248,7 @@ public class ReactiveCassandraTemplateUnitTests { assertThat(statementCaptor.getValue().toString()).isEqualTo("SELECT COUNT(1) FROM users;"); } - @Test // DATACASS-335 + @Test // DATACASS-335, DATACASS-618 public void insertShouldInsertEntity() { when(reactiveResultSet.wasApplied()).thenReturn(true); @@ -238,6 +260,24 @@ public class ReactiveCassandraTemplateUnitTests { verify(session).execute(statementCaptor.capture()); assertThat(statementCaptor.getValue().toString()) .isEqualTo("INSERT INTO users (firstname,id,lastname) VALUES ('Walter','heisenberg','White');"); + assertThat(beforeConvert).isSameAs(user); + assertThat(beforeSave).isSameAs(user); + } + + @Test // DATACASS-618 + public void insertShouldInsertVersionedEntity() { + + when(reactiveResultSet.wasApplied()).thenReturn(true); + when(reactiveResultSet.rows()).thenReturn(Flux.just(row)); + + VersionedUser user = new VersionedUser("heisenberg", "Walter", "White"); + StepVerifier.create(template.insert(user)).expectNext(user).verifyComplete(); + + verify(session).execute(statementCaptor.capture()); + assertThat(statementCaptor.getValue().toString()).isEqualTo( + "INSERT INTO vusers (firstname,id,lastname,version) VALUES ('Walter','heisenberg','White',0) IF NOT EXISTS;"); + assertThat(beforeConvert).isSameAs(user); + assertThat(beforeSave).isSameAs(user); } @Test // DATACASS-335 @@ -254,7 +294,7 @@ public class ReactiveCassandraTemplateUnitTests { }).verify(); } - @Test // DATACASS-335 + @Test // DATACASS-335, DATACASS-618 public void updateShouldUpdateEntity() { when(reactiveResultSet.wasApplied()).thenReturn(true); @@ -267,6 +307,26 @@ public class ReactiveCassandraTemplateUnitTests { verify(session).execute(statementCaptor.capture()); assertThat(statementCaptor.getValue().toString()) .isEqualTo("UPDATE users SET firstname='Walter',lastname='White' WHERE id='heisenberg';"); + assertThat(beforeConvert).isSameAs(user); + assertThat(beforeSave).isSameAs(user); + } + + @Test // DATACASS-618 + public void updateShouldUpdateVersionedEntity() { + + when(reactiveResultSet.wasApplied()).thenReturn(true); + when(reactiveResultSet.rows()).thenReturn(Flux.just(row)); + + VersionedUser user = new VersionedUser("heisenberg", "Walter", "White"); + user.setVersion(0L); + + StepVerifier.create(template.update(user)).expectNext(user).verifyComplete(); + + verify(session).execute(statementCaptor.capture()); + assertThat(statementCaptor.getValue().toString()).isEqualTo( + "UPDATE vusers SET firstname='Walter',lastname='White',version=1 WHERE id='heisenberg' IF version=0;"); + assertThat(beforeConvert).isSameAs(user); + assertThat(beforeSave).isSameAs(user); } @Test // DATACASS-575 @@ -341,8 +401,8 @@ public class ReactiveCassandraTemplateUnitTests { .verifyComplete(); verify(session).execute(statementCaptor.capture()); - assertThat(statementCaptor.getValue().toString()) - .isEqualTo("UPDATE users SET firstname='Walter' WHERE id='heisenberg' IF firstname='Walter' AND lastname='White';"); + assertThat(statementCaptor.getValue().toString()).isEqualTo( + "UPDATE users SET firstname='Walter' WHERE id='heisenberg' IF firstname='Walter' AND lastname='White';"); } @Test // DATACASS-335 diff --git a/spring-data-cassandra/src/test/java/org/springframework/data/cassandra/domain/VersionedUser.java b/spring-data-cassandra/src/test/java/org/springframework/data/cassandra/domain/VersionedUser.java new file mode 100644 index 000000000..3d12c17e1 --- /dev/null +++ b/spring-data-cassandra/src/test/java/org/springframework/data/cassandra/domain/VersionedUser.java @@ -0,0 +1,56 @@ +/* + * 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 + * + * https://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.domain; + +import lombok.Data; +import lombok.EqualsAndHashCode; +import lombok.NoArgsConstructor; + +import org.springframework.data.annotation.Id; +import org.springframework.data.annotation.PersistenceConstructor; +import org.springframework.data.annotation.Version; +import org.springframework.data.cassandra.core.mapping.Table; + +/** + * @author Mark Paluch + */ +@Table("vusers") +@Data +@NoArgsConstructor +@EqualsAndHashCode(of = "id") +public class VersionedUser { + + /* + * Primary Row ID + */ + @Id private String id; + + @Version private Long version; + + /* + * Public information + */ + private String firstname; + private String lastname; + + @PersistenceConstructor + public VersionedUser(String id, String firstname, String lastname) { + + this.id = id; + this.firstname = firstname; + this.lastname = lastname; + } +}