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 679cf8c1c..8e7312c62 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 @@ -15,11 +15,16 @@ */ package org.springframework.data.cassandra.core; +import lombok.Value; + import java.util.List; import java.util.concurrent.ExecutionException; import java.util.function.Consumer; import java.util.function.Function; +import org.springframework.context.ApplicationEvent; +import org.springframework.context.ApplicationEventPublisher; +import org.springframework.context.ApplicationEventPublisherAware; import org.springframework.dao.DataAccessException; import org.springframework.data.cassandra.SessionFactory; import org.springframework.data.cassandra.core.convert.CassandraConverter; @@ -39,9 +44,17 @@ import org.springframework.data.cassandra.core.cql.session.DefaultSessionFactory import org.springframework.data.cassandra.core.mapping.CassandraMappingContext; import org.springframework.data.cassandra.core.mapping.CassandraPersistentEntity; import org.springframework.data.cassandra.core.mapping.CassandraPersistentProperty; +import org.springframework.data.cassandra.core.mapping.event.AfterConvertEvent; +import org.springframework.data.cassandra.core.mapping.event.AfterDeleteEvent; +import org.springframework.data.cassandra.core.mapping.event.AfterLoadEvent; +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.query.Query; import org.springframework.data.domain.Slice; import org.springframework.data.mapping.context.MappingContext; +import org.springframework.data.projection.ProjectionFactory; +import org.springframework.data.projection.SpelAwareProxyProjectionFactory; import org.springframework.lang.Nullable; import org.springframework.scheduling.annotation.AsyncResult; import org.springframework.util.Assert; @@ -50,6 +63,7 @@ import org.springframework.util.concurrent.ListenableFuture; import com.datastax.driver.core.RegularStatement; import com.datastax.driver.core.ResultSet; +import com.datastax.driver.core.Row; import com.datastax.driver.core.Session; import com.datastax.driver.core.SimpleStatement; import com.datastax.driver.core.Statement; @@ -78,7 +92,7 @@ import com.datastax.driver.core.querybuilder.Update; * @see org.springframework.data.cassandra.core.AsyncCassandraOperations * @since 2.0 */ -public class AsyncCassandraTemplate implements AsyncCassandraOperations { +public class AsyncCassandraTemplate implements AsyncCassandraOperations, ApplicationEventPublisherAware { private final AsyncCqlOperations cqlOperations; @@ -88,8 +102,12 @@ public class AsyncCassandraTemplate implements AsyncCassandraOperations { private final CqlExceptionTranslator exceptionTranslator; + private final SpelAwareProxyProjectionFactory projectionFactory; + private final StatementFactory statementFactory; + private @Nullable ApplicationEventPublisher eventPublisher; + /** * Creates an instance of {@link AsyncCassandraTemplate} initialized with the given {@link Session} and a default * {@link MappingCassandraConverter}. @@ -149,6 +167,7 @@ public class AsyncCassandraTemplate implements AsyncCassandraOperations { this.mappingContext = converter.getMappingContext(); this.cqlOperations = asyncCqlTemplate; this.exceptionTranslator = asyncCqlTemplate.getExceptionTranslator(); + this.projectionFactory = new SpelAwareProxyProjectionFactory(); this.statementFactory = new StatementFactory(new QueryMapper(converter), new UpdateMapper(converter)); } @@ -168,50 +187,12 @@ public class AsyncCassandraTemplate implements AsyncCassandraOperations { return this.converter; } - private static MappingCassandraConverter newConverter() { - - MappingCassandraConverter converter = new MappingCassandraConverter(); - - converter.afterPropertiesSet(); - - return converter; - } - - /** - * Returns the {@link CassandraMappingContext} used by this template to access mapping meta-data - * in order to store (map) objects to Cassandra tables. - * - * @return the {@link CassandraMappingContext} used by this template. - * @see org.springframework.data.cassandra.core.mapping.CassandraMappingContext + /* (non-Javadoc) + * @see org.springframework.context.ApplicationEventPublisherAware#setApplicationEventPublisher(org.springframework.context.ApplicationEventPublisher) */ - protected MappingContext, CassandraPersistentProperty> getMappingContext() { - return this.mappingContext; - } - - private CassandraPersistentEntity getRequiredPersistentEntity(Object entity) { - return getRequiredPersistentEntity(entity.getClass()); - } - - private CassandraPersistentEntity getRequiredPersistentEntity(Class entityType) { - return getMappingContext().getRequiredPersistentEntity(ClassUtils.getUserClass(entityType)); - } - - /** - * Returns the {@link StatementFactory} used by this template to construct and run Cassandra CQL statements. - * - * @return the {@link StatementFactory} used by this template to construct and run Cassandra CQL statements. - * @see org.springframework.data.cassandra.core.StatementFactory - */ - protected StatementFactory getStatementFactory() { - return this.statementFactory; - } - - private CqlIdentifier getTableName(Object entity) { - return getRequiredPersistentEntity(entity).getTableName(); - } - - private CqlIdentifier getTableName(Class entityType) { - return getRequiredPersistentEntity(entityType).getTableName(); + @Override + public void setApplicationEventPublisher(ApplicationEventPublisher applicationEventPublisher) { + this.eventPublisher = applicationEventPublisher; } // ------------------------------------------------------------------------- @@ -268,7 +249,9 @@ public class AsyncCassandraTemplate implements AsyncCassandraOperations { Assert.notNull(statement, "Statement must not be null"); Assert.notNull(entityClass, "Entity type must not be null"); - return getAsyncCqlOperations().query(statement, (row, rowNum) -> getConverter().read(entityClass, row)); + Function mapper = getMapper(entityClass, entityClass, QueryUtils.getTableName(statement)); + + return getAsyncCqlOperations().query(statement, (row, rowNum) -> mapper.apply(row)); } /* (non-Javadoc) @@ -282,10 +265,10 @@ public class AsyncCassandraTemplate implements AsyncCassandraOperations { ListenableFuture resultSet = getAsyncCqlOperations().queryForResultSet(statement); - CassandraConverter converter = getConverter(); + Function mapper = getMapper(entityClass, entityClass, QueryUtils.getTableName(statement)); - return new MappingListenableFutureAdapter<>(resultSet, rs -> QueryUtils.readSlice(rs, - (row, rowNum) -> converter.read(entityClass, row), 0, getEffectiveFetchSize(statement))); + return new MappingListenableFutureAdapter<>(resultSet, + rs -> QueryUtils.readSlice(rs, (row, rowNum) -> mapper.apply(row), 0, getEffectiveFetchSize(statement))); } /* (non-Javadoc) @@ -299,8 +282,10 @@ public class AsyncCassandraTemplate implements AsyncCassandraOperations { Assert.notNull(entityConsumer, "Entity Consumer must not be empty"); Assert.notNull(entityClass, "Entity type must not be null"); + Function mapper = getMapper(entityClass, entityClass, QueryUtils.getTableName(statement)); + return getAsyncCqlOperations().query(statement, row -> { - entityConsumer.accept(getConverter().read(entityClass, row)); + entityConsumer.accept(mapper.apply(row)); }); } @@ -327,7 +312,7 @@ public class AsyncCassandraTemplate implements AsyncCassandraOperations { Assert.notNull(query, "Query must not be null"); Assert.notNull(entityClass, "Entity type must not be null"); - return select(getStatementFactory().select(query, getMappingContext().getRequiredPersistentEntity(entityClass)), + return select(getStatementFactory().select(query, getRequiredPersistentEntity(entityClass)), entityClass); } @@ -340,7 +325,7 @@ public class AsyncCassandraTemplate implements AsyncCassandraOperations { Assert.notNull(query, "Query must not be null"); Assert.notNull(entityClass, "Entity type must not be null"); - return slice(this.statementFactory.select(query, getMappingContext().getRequiredPersistentEntity(entityClass)), + return slice(this.statementFactory.select(query, getRequiredPersistentEntity(entityClass)), entityClass); } @@ -355,7 +340,7 @@ public class AsyncCassandraTemplate implements AsyncCassandraOperations { Assert.notNull(entityConsumer, "Entity Consumer must not be empty"); Assert.notNull(entityClass, "Entity type must not be null"); - return select(getStatementFactory().select(query, getMappingContext().getRequiredPersistentEntity(entityClass)), + return select(getStatementFactory().select(query, getRequiredPersistentEntity(entityClass)), entityConsumer, entityClass); } @@ -368,7 +353,7 @@ public class AsyncCassandraTemplate implements AsyncCassandraOperations { Assert.notNull(query, "Query must not be null"); Assert.notNull(entityClass, "Entity type must not be null"); - return selectOne(getStatementFactory().select(query, getMappingContext().getRequiredPersistentEntity(entityClass)), + return selectOne(getStatementFactory().select(query, getRequiredPersistentEntity(entityClass)), entityClass); } @@ -384,7 +369,7 @@ public class AsyncCassandraTemplate implements AsyncCassandraOperations { Assert.notNull(entityClass, "Entity type must not be null"); return getAsyncCqlOperations().execute( - getStatementFactory().update(query, update, getMappingContext().getRequiredPersistentEntity(entityClass))); + getStatementFactory().update(query, update, getRequiredPersistentEntity(entityClass))); } /* (non-Javadoc) @@ -396,8 +381,21 @@ public class AsyncCassandraTemplate implements AsyncCassandraOperations { Assert.notNull(query, "Query must not be null"); Assert.notNull(entityClass, "Entity type must not be null"); - return getAsyncCqlOperations() - .execute(getStatementFactory().delete(query, getMappingContext().getRequiredPersistentEntity(entityClass))); + return doDelete(query, entityClass, getTableName(entityClass)); + } + + private ListenableFuture doDelete(Query query, Class entityClass, CqlIdentifier tableName) { + + RegularStatement delete = getStatementFactory().delete(query, getRequiredPersistentEntity(entityClass), tableName); + + maybeEmitEvent(new BeforeDeleteEvent<>(delete, entityClass, tableName)); + + ListenableFuture future = getAsyncCqlOperations() + .execute(getStatementFactory().delete(query, getRequiredPersistentEntity(entityClass))); + + future.addCallback(success -> maybeEmitEvent(new AfterDeleteEvent<>(delete, entityClass, tableName)), e -> {}); + + return future; } // ------------------------------------------------------------------------- @@ -445,7 +443,6 @@ public class AsyncCassandraTemplate implements AsyncCassandraOperations { CassandraPersistentEntity entity = getRequiredPersistentEntity(entityClass); Select select = QueryBuilder.select().from(entity.getTableName().toCql()); - getConverter().write(id, select.where(), entity); return new MappingListenableFutureAdapter<>(getAsyncCqlOperations().queryForResultSet(select), @@ -461,8 +458,7 @@ public class AsyncCassandraTemplate implements AsyncCassandraOperations { Assert.notNull(query, "Query must not be null"); Assert.notNull(entityClass, "Entity type must not be null"); - RegularStatement select = getStatementFactory() - .select(query.limit(1), getRequiredPersistentEntity(entityClass)); + RegularStatement select = getStatementFactory().select(query.limit(1), getRequiredPersistentEntity(entityClass)); return new MappingListenableFutureAdapter<>(getAsyncCqlOperations().queryForResultSet(select), resultSet -> resultSet.iterator().hasNext()); @@ -480,10 +476,14 @@ public class AsyncCassandraTemplate implements AsyncCassandraOperations { CassandraPersistentEntity entity = getRequiredPersistentEntity(entityClass); Select select = QueryBuilder.select().all().from(entity.getTableName().toCql()); - getConverter().write(id, select.where(), entity); - return selectOne(select, entityClass); + Function mapper = getMapper(entityClass, entityClass, entity.getTableName()); + + return new MappingListenableFutureAdapter<>( + getAsyncCqlOperations().query(select, (row, rowNum) -> mapper.apply(row)), it -> { + return it.isEmpty() ? null : (T) it.get(0); + }); } /* (non-Javadoc) @@ -503,10 +503,16 @@ public class AsyncCassandraTemplate implements AsyncCassandraOperations { Assert.notNull(entity, "Entity must not be null"); Assert.notNull(options, "InsertOptions must not be null"); - Insert insert = QueryUtils.createInsertQuery(getTableName(entity).toCql(), entity, options, getConverter()); + CqlIdentifier tableName = getTableName(entity); + Insert insert = QueryUtils.createInsertQuery(tableName.toCql(), entity, options, getConverter()); - return new MappingListenableFutureAdapter<>( - getAsyncCqlOperations().execute(new AsyncStatementCallback(insert)), WriteResult::of); + maybeEmitEvent(new BeforeSaveEvent<>(entity, tableName, insert)); + + return new MappingListenableFutureAdapter<>(getAsyncCqlOperations().execute(new AsyncStatementCallback(insert)), + resultSet -> { + maybeEmitEvent(new AfterSaveEvent<>(entity, tableName)); + return WriteResult.of(resultSet); + }); } /* (non-Javadoc) @@ -526,10 +532,16 @@ public class AsyncCassandraTemplate implements AsyncCassandraOperations { Assert.notNull(entity, "Entity must not be null"); Assert.notNull(options, "UpdateOptions must not be null"); - Update update = QueryUtils.createUpdateQuery(getTableName(entity).toCql(), entity, options, getConverter()); + CqlIdentifier tableName = getTableName(entity); + Update update = QueryUtils.createUpdateQuery(tableName.toCql(), entity, options, getConverter()); - return new MappingListenableFutureAdapter<>( - getAsyncCqlOperations().execute(new AsyncStatementCallback(update)), WriteResult::of); + maybeEmitEvent(new BeforeSaveEvent<>(entity, tableName, update)); + + return new MappingListenableFutureAdapter<>(getAsyncCqlOperations().execute(new AsyncStatementCallback(update)), + resultSet -> { + maybeEmitEvent(new AfterSaveEvent<>(entity, tableName)); + return WriteResult.of(resultSet); + }); } /* (non-Javadoc) @@ -549,10 +561,16 @@ public class AsyncCassandraTemplate implements AsyncCassandraOperations { Assert.notNull(entity, "Entity must not be null"); Assert.notNull(options, "QueryOptions must not be null"); - Delete delete = QueryUtils.createDeleteQuery(getTableName(entity).toCql(), entity, options, getConverter()); + CqlIdentifier tableName = getTableName(entity); + Delete delete = QueryUtils.createDeleteQuery(tableName.toCql(), entity, options, getConverter()); - return new MappingListenableFutureAdapter<>(getAsyncCqlOperations() - .execute(new AsyncStatementCallback(delete)), WriteResult::of); + maybeEmitEvent(new BeforeDeleteEvent<>(delete, entity.getClass(), tableName)); + + return new MappingListenableFutureAdapter<>(getAsyncCqlOperations().execute(new AsyncStatementCallback(delete)), + resultSet -> { + maybeEmitEvent(new AfterDeleteEvent<>(delete, entity.getClass(), tableName)); + return WriteResult.of(resultSet); + }); } /* (non-Javadoc) @@ -566,11 +584,16 @@ public class AsyncCassandraTemplate implements AsyncCassandraOperations { CassandraPersistentEntity entity = getRequiredPersistentEntity(entityClass); - Delete delete = QueryBuilder.delete().from(entity.getTableName().toCql()); - + CqlIdentifier tableName = entity.getTableName(); + Delete delete = QueryBuilder.delete().from(tableName.toCql()); getConverter().write(id, delete.where(), entity); - return getAsyncCqlOperations().execute(delete); + maybeEmitEvent(new BeforeDeleteEvent<>(delete, entityClass, tableName)); + + ListenableFuture future = getAsyncCqlOperations().execute(delete); + future.addCallback(success -> maybeEmitEvent(new AfterDeleteEvent<>(delete, entityClass, tableName)), e -> {}); + + return future; } /* (non-Javadoc) @@ -581,15 +604,68 @@ public class AsyncCassandraTemplate implements AsyncCassandraOperations { Assert.notNull(entityClass, "Entity type must not be null"); - Truncate truncate = QueryBuilder.truncate(getTableName(entityClass).toCql()); + CqlIdentifier tableName = getTableName(entityClass); + Truncate truncate = QueryBuilder.truncate(tableName.toCql()); - return new MappingListenableFutureAdapter<>(getAsyncCqlOperations().execute(truncate), aBoolean -> null); + maybeEmitEvent(new BeforeDeleteEvent<>(truncate, entityClass, tableName)); + + ListenableFuture future = getAsyncCqlOperations().execute(truncate); + future.addCallback(success -> maybeEmitEvent(new AfterDeleteEvent<>(truncate, entityClass, tableName)), e -> {}); + + return new MappingListenableFutureAdapter<>(future, aBoolean -> null); } // ------------------------------------------------------------------------- // Implementation hooks and helper methods // ------------------------------------------------------------------------- + /** + * Returns the {@link CassandraMappingContext} used by this template to access mapping meta-data in order to store + * (map) object to Cassandra tables. + * + * @return the {@link CassandraMappingContext} used by this template. + * @see org.springframework.data.cassandra.core.mapping.CassandraMappingContext + */ + protected MappingContext, CassandraPersistentProperty> getMappingContext() { + return this.mappingContext; + } + + /** + * Returns a reference to the configured {@link ProjectionFactory} used by this template to process CQL query + * projections. + * + * @return a reference to the configured {@link ProjectionFactory} used by this template to process CQL query + * projections. + * @see org.springframework.data.projection.SpelAwareProxyProjectionFactory + * @since 2.1 + */ + protected SpelAwareProxyProjectionFactory getProjectionFactory() { + return this.projectionFactory; + } + + /** + * Returns the {@link StatementFactory} used by this template to construct and run Cassandra CQL statements. + * + * @return the {@link StatementFactory} used by this template to construct and run Cassandra CQL statements. + * @see org.springframework.data.cassandra.core.StatementFactory + * @since 2.1 + */ + protected StatementFactory getStatementFactory() { + return this.statementFactory; + } + + private CqlIdentifier getTableName(Class entityClass) { + return getRequiredPersistentEntity(entityClass).getTableName(); + } + + private CqlIdentifier getTableName(Object entity) { + return getRequiredPersistentEntity(entity.getClass()).getTableName(); + } + + private CassandraPersistentEntity getRequiredPersistentEntity(Class entityType) { + return getMappingContext().getRequiredPersistentEntity(ClassUtils.getUserClass(entityType)); + } + private int getConfiguredFetchSize(Session session) { return session.getCluster().getConfiguration().getQueryOptions().getFetchSize(); } @@ -610,8 +686,50 @@ public class AsyncCassandraTemplate implements AsyncCassandraOperations { } } - return getAsyncCqlOperations().execute((AsyncSessionCallback) session -> - AsyncResult.forValue(getConfiguredFetchSize(session))).completable().join(); + return getAsyncCqlOperations() + .execute((AsyncSessionCallback) session -> AsyncResult.forValue(getConfiguredFetchSize(session))) + .completable().join(); + } + + @SuppressWarnings("unchecked") + private Function getMapper(Class entityType, Class targetType, CqlIdentifier tableName) { + + Class typeToRead = resolveTypeToRead(entityType, targetType); + + return row -> { + + 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)); + } + + return result; + }; + } + + private Class resolveTypeToRead(Class entityType, Class targetType) { + return targetType.isInterface() || targetType.isAssignableFrom(entityType) ? entityType : targetType; + } + + private void maybeEmitEvent(ApplicationEvent event) { + + if (eventPublisher != null) { + eventPublisher.publishEvent(event); + } + } + + private static MappingCassandraConverter newConverter() { + + MappingCassandraConverter converter = new MappingCassandraConverter(); + + converter.afterPropertiesSet(); + + return converter; } static class MappingListenableFutureAdapter @@ -633,9 +751,10 @@ public class AsyncCassandraTemplate implements AsyncCassandraOperations { } } - private class AsyncStatementCallback implements AsyncSessionCallback, CqlProvider { + @Value + class AsyncStatementCallback implements AsyncSessionCallback, CqlProvider { - private final Statement statement; + @lombok.NonNull Statement statement; AsyncStatementCallback(Statement statement) { this.statement = statement; 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 2f32979dd..373c7291d 100644 --- a/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/CassandraTemplate.java +++ b/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/CassandraTemplate.java @@ -15,19 +15,16 @@ */ package org.springframework.data.cassandra.core; +import lombok.Value; + import java.util.List; import java.util.function.Function; import java.util.stream.Stream; import java.util.stream.StreamSupport; -import lombok.NonNull; -import lombok.Value; - -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; import org.springframework.dao.DataAccessException; import org.springframework.data.cassandra.SessionFactory; import org.springframework.data.cassandra.core.convert.CassandraConverter; @@ -46,6 +43,7 @@ import org.springframework.data.cassandra.core.cql.session.DefaultSessionFactory import org.springframework.data.cassandra.core.mapping.CassandraMappingContext; import org.springframework.data.cassandra.core.mapping.CassandraPersistentEntity; import org.springframework.data.cassandra.core.mapping.CassandraPersistentProperty; +import org.springframework.data.cassandra.core.mapping.event.AfterConvertEvent; import org.springframework.data.cassandra.core.mapping.event.AfterDeleteEvent; import org.springframework.data.cassandra.core.mapping.event.AfterLoadEvent; import org.springframework.data.cassandra.core.mapping.event.AfterSaveEvent; @@ -88,10 +86,11 @@ import com.datastax.driver.core.querybuilder.Update; * * @author Mark Paluch * @author John Blum + * @author Lukasz Antoniak * @see org.springframework.data.cassandra.core.CassandraOperations * @since 2.0 */ -public class CassandraTemplate implements CassandraOperations, ApplicationContextAware { +public class CassandraTemplate implements CassandraOperations, ApplicationEventPublisherAware { private final CassandraConverter converter; @@ -167,6 +166,14 @@ public class CassandraTemplate implements CassandraOperations, ApplicationContex this.statementFactory = new StatementFactory(new QueryMapper(converter), new UpdateMapper(converter)); } + /* (non-Javadoc) + * @see org.springframework.data.cassandra.core.CassandraOperations#batchOps() + */ + @Override + public CassandraBatchOperations batchOps() { + return new CassandraBatchTemplate(this); + } + /* (non-Javadoc) * @see org.springframework.data.cassandra.core.CassandraOperations#getConverter() */ @@ -175,90 +182,30 @@ public class CassandraTemplate implements CassandraOperations, ApplicationContex return this.converter; } - private static MappingCassandraConverter newConverter() { - - MappingCassandraConverter converter = new MappingCassandraConverter(); - - converter.afterPropertiesSet(); - - return converter; - } - /* (non-Javadoc) * @see org.springframework.data.cassandra.core.CassandraOperations#CqlOperations() */ @Override - @org.springframework.lang.NonNull public CqlOperations getCqlOperations() { return this.cqlOperations; } - /** - * Returns the {@link CassandraMappingContext} used by this template to access mapping meta-data in order to store - * (map) object to Cassandra tables. - * - * @return the {@link CassandraMappingContext} used by this template. - * @see org.springframework.data.cassandra.core.mapping.CassandraMappingContext - */ - protected MappingContext, CassandraPersistentProperty> getMappingContext() { - return this.mappingContext; - } - - @org.springframework.lang.NonNull - private CassandraPersistentEntity getRequiredPersistentEntity(Object entity) { - return getRequiredPersistentEntity(entity.getClass()); - } - - @org.springframework.lang.NonNull - private CassandraPersistentEntity getRequiredPersistentEntity(Class entityType) { - return getMappingContext().getRequiredPersistentEntity(ClassUtils.getUserClass(entityType)); - } - - @org.springframework.lang.Nullable - private String guessTableName(Object entity) { - if (getMappingContext().hasPersistentEntityFor(entity.getClass())) { - return getMappingContext().getRequiredPersistentEntity(ClassUtils.getUserClass(entity.getClass())).getTableName().toCql(); - } - // Not an entity. - return null; - } - - /** - * Returns a reference to the configured {@link ProjectionFactory} used by this template - * to process CQL query projections. - * - * @return a reference to the configured {@link ProjectionFactory} used by this template - * to process CQL query projections. - * @see org.springframework.data.projection.SpelAwareProxyProjectionFactory - */ - protected SpelAwareProxyProjectionFactory getProjectionFactory() { - return this.projectionFactory; - } - - /** - * Returns the {@link StatementFactory} used by this template to construct and run Cassandra CQL statements. - * - * @return the {@link StatementFactory} used by this template to construct and run Cassandra CQL statements. - * @see org.springframework.data.cassandra.core.StatementFactory - */ - protected StatementFactory getStatementFactory() { - return this.statementFactory; - } - - @org.springframework.lang.NonNull - private CqlIdentifier getTableName(Object entity) { - return getRequiredPersistentEntity(entity).getTableName(); - } - /* (non-Javadoc) * @see org.springframework.data.cassandra.core.CassandraOperations#getTableName(java.lang.Class) */ @Override - @org.springframework.lang.NonNull public CqlIdentifier getTableName(Class entityClass) { return getRequiredPersistentEntity(entityClass).getTableName(); } + /* (non-Javadoc) + * @see org.springframework.context.ApplicationEventPublisherAware#setApplicationEventPublisher(org.springframework.context.ApplicationEventPublisher) + */ + @Override + public void setApplicationEventPublisher(ApplicationEventPublisher applicationEventPublisher) { + this.eventPublisher = applicationEventPublisher; + } + // ------------------------------------------------------------------------- // Methods dealing with static CQL // ------------------------------------------------------------------------- @@ -311,7 +258,7 @@ public class CassandraTemplate implements CassandraOperations, ApplicationContex Assert.notNull(statement, "Statement must not be null"); Assert.notNull(entityClass, "Entity type must not be null"); - Function mapper = getMapper(entityClass, entityClass, true); + Function mapper = getMapper(entityClass, entityClass, QueryUtils.getTableName(statement)); return getCqlOperations().query(statement, (row, rowNum) -> mapper.apply(row)); } @@ -327,10 +274,9 @@ public class CassandraTemplate implements CassandraOperations, ApplicationContex ResultSet resultSet = getCqlOperations().queryForResultSet(statement); - Function mapper = getMapper(entityClass, entityClass, true); + Function mapper = getMapper(entityClass, entityClass, QueryUtils.getTableName(statement)); - return QueryUtils.readSlice(resultSet, (row, rowNum) -> mapper.apply(row), - 0, getEffectiveFetchSize(statement)); + return QueryUtils.readSlice(resultSet, (row, rowNum) -> mapper.apply(row), 0, getEffectiveFetchSize(statement)); } /* (non-Javadoc) @@ -344,7 +290,8 @@ public class CassandraTemplate implements CassandraOperations, ApplicationContex ResultSet resultSet = getCqlOperations().queryForResultSet(statement); - return StreamSupport.stream(resultSet.spliterator(), false).map(getMapper(entityClass, entityClass, true)); + return StreamSupport.stream(resultSet.spliterator(), false) + .map(getMapper(entityClass, entityClass, QueryUtils.getTableName(statement))); } /* (non-Javadoc) @@ -373,10 +320,9 @@ public class CassandraTemplate implements CassandraOperations, ApplicationContex List doSelect(Query query, Class entityClass, CqlIdentifier tableName, Class returnType) { - Function mapper = getMapper(entityClass, returnType, true); + Function mapper = getMapper(entityClass, returnType, tableName); - RegularStatement select = getStatementFactory() - .select(query, getRequiredPersistentEntity(entityClass), tableName); + RegularStatement select = getStatementFactory().select(query, getRequiredPersistentEntity(entityClass), tableName); return getCqlOperations().query(select, (row, rowNum) -> mapper.apply(row)); } @@ -409,12 +355,12 @@ public class CassandraTemplate implements CassandraOperations, ApplicationContex Stream doStream(Query query, Class entityClass, CqlIdentifier tableName, Class returnType) { - RegularStatement statement = getStatementFactory() - .select(query, getRequiredPersistentEntity(entityClass), tableName); + RegularStatement statement = getStatementFactory().select(query, getRequiredPersistentEntity(entityClass), + tableName); ResultSet resultSet = getCqlOperations().queryForResultSet(statement); - return StreamSupport.stream(resultSet.spliterator(), false).map(getMapper(entityClass, returnType, true)); + return StreamSupport.stream(resultSet.spliterator(), false).map(getMapper(entityClass, returnType, tableName)); } /* (non-Javadoc) @@ -439,8 +385,7 @@ public class CassandraTemplate implements CassandraOperations, ApplicationContex Assert.notNull(update, "Update must not be null"); Assert.notNull(entityClass, "Entity type must not be null"); - Statement updateStatement = getStatementFactory() - .update(query, update, getRequiredPersistentEntity(entityClass)); + Statement updateStatement = getStatementFactory().update(query, update, getRequiredPersistentEntity(entityClass)); return getCqlOperations().execute(updateStatement); } @@ -449,8 +394,8 @@ public class CassandraTemplate implements CassandraOperations, ApplicationContex WriteResult doUpdate(Query query, org.springframework.data.cassandra.core.query.Update update, Class entityClass, CqlIdentifier tableName) { - RegularStatement updateStatement = getStatementFactory() - .update(query, update, getRequiredPersistentEntity(entityClass), tableName); + RegularStatement updateStatement = getStatementFactory().update(query, update, + getRequiredPersistentEntity(entityClass), tableName); return getCqlOperations().execute(new StatementCallback(updateStatement)); } @@ -472,10 +417,15 @@ public class CassandraTemplate implements CassandraOperations, ApplicationContex @Nullable WriteResult doDelete(Query query, Class entityClass, CqlIdentifier tableName) { - RegularStatement delete = getStatementFactory() - .delete(query, getRequiredPersistentEntity(entityClass), tableName); + RegularStatement delete = getStatementFactory().delete(query, getRequiredPersistentEntity(entityClass), tableName); - return getCqlOperations().execute(new StatementCallback(delete)); + maybeEmitEvent(new BeforeDeleteEvent<>(delete, entityClass, tableName)); + + WriteResult writeResult = getCqlOperations().execute(new StatementCallback(delete)); + + maybeEmitEvent(new AfterDeleteEvent<>(delete, entityClass, tableName)); + + return writeResult; } // ------------------------------------------------------------------------- @@ -511,8 +461,8 @@ public class CassandraTemplate implements CassandraOperations, ApplicationContex long doCount(Query query, Class entityClass, CqlIdentifier tableName) { - RegularStatement countStatement = getStatementFactory() - .count(query, getRequiredPersistentEntity(entityClass), tableName); + RegularStatement countStatement = getStatementFactory().count(query, getRequiredPersistentEntity(entityClass), + tableName); Long count = getCqlOperations().queryForObject(countStatement, Long.class); @@ -551,8 +501,8 @@ public class CassandraTemplate implements CassandraOperations, ApplicationContex boolean doExists(Query query, Class entityClass, CqlIdentifier tableName) { - RegularStatement select = getStatementFactory() - .select(query.limit(1), getRequiredPersistentEntity(entityClass), tableName); + RegularStatement select = getStatementFactory().select(query.limit(1), getRequiredPersistentEntity(entityClass), + tableName); return getCqlOperations().queryForResultSet(select).iterator().hasNext(); } @@ -566,13 +516,15 @@ public class CassandraTemplate implements CassandraOperations, ApplicationContex Assert.notNull(id, "Id must not be null"); Assert.notNull(entityClass, "Entity type must not be null"); - CassandraPersistentEntity entity = getRequiredPersistentEntity(entityClass); + CqlIdentifier tableName = getTableName(entityClass); - Select select = QueryBuilder.select().all().from(entity.getTableName().toCql()); + Select select = QueryBuilder.select().all().from(tableName.toCql()); + getConverter().write(id, select.where(), getRequiredPersistentEntity(entityClass)); - getConverter().write(id, select.where(), entity); + Function mapper = getMapper(entityClass, entityClass, tableName); + List result = getCqlOperations().query(select, (row, rowNum) -> mapper.apply(row)); - return selectOne(select, entityClass); + return result.isEmpty() ? null : result.get(0); } /* (non-Javadoc) @@ -599,12 +551,12 @@ public class CassandraTemplate implements CassandraOperations, ApplicationContex Insert insert = QueryUtils.createInsertQuery(tableName.toCql(), entity, options, getConverter()); - maybeEmitEvent(new BeforeSaveEvent(entity, tableName.toCql(), insert)); + maybeEmitEvent(new BeforeSaveEvent<>(entity, tableName, insert)); // noinspection ConstantConditions WriteResult result = getCqlOperations().execute(new StatementCallback(insert)); - maybeEmitEvent(new AfterSaveEvent(entity, tableName.toCql())); + maybeEmitEvent(new AfterSaveEvent<>(entity, tableName)); return result; } @@ -626,14 +578,15 @@ public class CassandraTemplate implements CassandraOperations, ApplicationContex Assert.notNull(entity, "Entity must not be null"); Assert.notNull(options, "UpdateOptions must not be null"); - Update update = QueryUtils.createUpdateQuery(getTableName(entity).toCql(), entity, options, getConverter()); + CqlIdentifier tableName = getTableName(entity); + Update update = QueryUtils.createUpdateQuery(tableName.toCql(), entity, options, getConverter()); - maybeEmitEvent(new BeforeSaveEvent(entity, guessTableName(entity), update)); + maybeEmitEvent(new BeforeSaveEvent<>(entity, tableName, update)); // noinspection ConstantConditions WriteResult result = getCqlOperations().execute(new StatementCallback(update)); - maybeEmitEvent(new AfterSaveEvent(entity, guessTableName(entity))); + maybeEmitEvent(new AfterSaveEvent<>(entity, tableName)); return result; } @@ -655,14 +608,15 @@ public class CassandraTemplate implements CassandraOperations, ApplicationContex Assert.notNull(entity, "Entity must not be null"); Assert.notNull(options, "QueryOptions must not be null"); - Delete delete = QueryUtils.createDeleteQuery(getTableName(entity).toCql(), entity, options, getConverter()); + CqlIdentifier tableName = getTableName(entity); + Delete delete = QueryUtils.createDeleteQuery(tableName.toCql(), entity, options, getConverter()); - maybeEmitEvent(new BeforeDeleteEvent(entity, guessTableName(entity), delete)); + maybeEmitEvent(new BeforeDeleteEvent<>(delete, entity.getClass(), tableName)); // noinspection ConstantConditions WriteResult result = getCqlOperations().execute(new StatementCallback(delete)); - maybeEmitEvent(new AfterDeleteEvent(entity, guessTableName(entity))); + maybeEmitEvent(new AfterDeleteEvent<>(delete, entity.getClass(), tableName)); return result; } @@ -678,11 +632,18 @@ public class CassandraTemplate implements CassandraOperations, ApplicationContex CassandraPersistentEntity entity = getRequiredPersistentEntity(entityClass); - Delete delete = QueryBuilder.delete().from(entity.getTableName().toCql()); + CqlIdentifier tableName = entity.getTableName(); + Delete delete = QueryBuilder.delete().from(tableName.toCql()); getConverter().write(id, delete.where(), entity); - return getCqlOperations().execute(delete); + maybeEmitEvent(new BeforeDeleteEvent<>(delete, entityClass, tableName)); + + boolean result = getCqlOperations().execute(delete); + + maybeEmitEvent(new AfterDeleteEvent<>(delete, entityClass, tableName)); + + return result; } /* (non-Javadoc) @@ -693,9 +654,14 @@ public class CassandraTemplate implements CassandraOperations, ApplicationContex Assert.notNull(entityClass, "Entity type must not be null"); - Truncate truncate = QueryBuilder.truncate(getTableName(entityClass).toCql()); + CqlIdentifier tableName = getTableName(entityClass); + Truncate truncate = QueryBuilder.truncate(tableName.toCql()); + + maybeEmitEvent(new BeforeDeleteEvent<>(truncate, entityClass, tableName)); getCqlOperations().execute(truncate); + + maybeEmitEvent(new AfterDeleteEvent<>(truncate, entityClass, tableName)); } // ------------------------------------------------------------------------- @@ -738,6 +704,49 @@ public class CassandraTemplate implements CassandraOperations, ApplicationContex // Implementation hooks and helper methods // ------------------------------------------------------------------------- + /** + * Returns the {@link CassandraMappingContext} used by this template to access mapping meta-data in order to store + * (map) object to Cassandra tables. + * + * @return the {@link CassandraMappingContext} used by this template. + * @see org.springframework.data.cassandra.core.mapping.CassandraMappingContext + */ + protected MappingContext, CassandraPersistentProperty> getMappingContext() { + return this.mappingContext; + } + + /** + * Returns a reference to the configured {@link ProjectionFactory} used by this template to process CQL query + * projections. + * + * @return a reference to the configured {@link ProjectionFactory} used by this template to process CQL query + * projections. + * @see org.springframework.data.projection.SpelAwareProxyProjectionFactory + * @since 2.1 + */ + protected SpelAwareProxyProjectionFactory getProjectionFactory() { + return this.projectionFactory; + } + + /** + * Returns the {@link StatementFactory} used by this template to construct and run Cassandra CQL statements. + * + * @return the {@link StatementFactory} used by this template to construct and run Cassandra CQL statements. + * @see org.springframework.data.cassandra.core.StatementFactory + * @since 2.1 + */ + protected StatementFactory getStatementFactory() { + return this.statementFactory; + } + + private CqlIdentifier getTableName(Object entity) { + return getRequiredPersistentEntity(entity.getClass()).getTableName(); + } + + private CassandraPersistentEntity getRequiredPersistentEntity(Class entityType) { + return getMappingContext().getRequiredPersistentEntity(ClassUtils.getUserClass(entityType)); + } + private int getConfiguredFetchSize(Session session) { return session.getCluster().getConfiguration().getQueryOptions().getFetchSize(); } @@ -762,18 +771,20 @@ public class CassandraTemplate implements CassandraOperations, ApplicationContex } @SuppressWarnings("unchecked") - private Function getMapper(Class entityType, Class targetType, boolean emitEvents) { + private Function getMapper(Class entityType, Class targetType, CqlIdentifier tableName) { Class typeToRead = resolveTypeToRead(entityType, targetType); return row -> { + maybeEmitEvent(new AfterLoadEvent(row, targetType, tableName)); + Object source = getConverter().read(typeToRead, row); T result = (T) (targetType.isInterface() ? getProjectionFactory().createProjection(targetType, source) : source); - if (emitEvents) { - maybeEmitEvent(new AfterLoadEvent(result, guessTableName(result))); + if (result != null) { + maybeEmitEvent(new AfterConvertEvent<>(row, result, tableName)); } return result; @@ -784,35 +795,38 @@ public class CassandraTemplate implements CassandraOperations, ApplicationContex return targetType.isInterface() || targetType.isAssignableFrom(entityType) ? entityType : targetType; } - /* (non-Javadoc) - * @see org.springframework.data.cassandra.core.CassandraOperations#batchOps() - */ - @Override - public CassandraBatchOperations batchOps() { - return new CassandraBatchTemplate(this); - } - - @Override - public void setApplicationContext(ApplicationContext applicationContext) throws BeansException { - this.eventPublisher = applicationContext; - } - private void maybeEmitEvent(ApplicationEvent event) { + if (eventPublisher != null) { eventPublisher.publishEvent(event); } } + private static MappingCassandraConverter newConverter() { + + MappingCassandraConverter converter = new MappingCassandraConverter(); + + converter.afterPropertiesSet(); + + return converter; + } + @Value static class StatementCallback implements SessionCallback, CqlProvider { - @NonNull Statement statement; + @lombok.NonNull Statement statement; + /* (non-Javadoc) + * @see org.springframework.data.cassandra.core.cql.SessionCallback#doInSession(org.springframework.data.cassandra.Session) + */ @Override public WriteResult doInSession(Session session) throws DriverException, DataAccessException { return WriteResult.of(session.execute(statement)); } + /* (non-Javadoc) + * @see org.springframework.data.cassandra.core.cql.CqlProvider#getCql() + */ @Override public String getCql() { return statement.toString(); diff --git a/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/QueryUtils.java b/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/QueryUtils.java index 113a55c55..16c3e79f8 100644 --- a/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/QueryUtils.java +++ b/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/QueryUtils.java @@ -17,7 +17,11 @@ package org.springframework.data.cassandra.core; import java.util.ArrayList; import java.util.List; +import java.util.regex.Matcher; +import java.util.regex.Pattern; +import org.springframework.beans.DirectFieldAccessor; +import org.springframework.data.cassandra.core.cql.CqlIdentifier; import org.springframework.data.cassandra.core.cql.QueryOptions; import org.springframework.data.cassandra.core.cql.QueryOptionsUtil; import org.springframework.data.cassandra.core.cql.RowMapper; @@ -32,10 +36,12 @@ import org.springframework.util.Assert; import com.datastax.driver.core.PagingState; import com.datastax.driver.core.ResultSet; +import com.datastax.driver.core.Statement; import com.datastax.driver.core.querybuilder.Delete; import com.datastax.driver.core.querybuilder.Delete.Where; import com.datastax.driver.core.querybuilder.Insert; import com.datastax.driver.core.querybuilder.QueryBuilder; +import com.datastax.driver.core.querybuilder.Select; import com.datastax.driver.core.querybuilder.Update; /** @@ -48,6 +54,9 @@ import com.datastax.driver.core.querybuilder.Update; */ class QueryUtils { + private static final Pattern FROM_REGEX = Pattern.compile(" FROM ([\"]?[\\w]*[\\\\.]?[\\w]*[\"]?)[\\s]?", + Pattern.CASE_INSENSITIVE); + /** * Creates a Query Object for an insert. * @@ -172,4 +181,47 @@ class QueryUtils { return new SliceImpl<>(result, pageRequest, pagingState != null); } + + /** + * Extract the table name from a {@link Statement}. + * + * @param statement + * @return + * @since 2.1 + */ + static CqlIdentifier getTableName(Statement statement) { + + if (statement instanceof Select) { + + Select select = (Select) statement; + + DirectFieldAccessor accessor = new DirectFieldAccessor(select); + String table = (String) accessor.getPropertyValue("table"); + + if (table != null) { + return CqlIdentifier.of(table); + } + } + + String cql = statement.toString(); + Matcher matcher = FROM_REGEX.matcher(cql); + + if (matcher.find()) { + + String cqlTableName = matcher.group(1); + if (cqlTableName.startsWith("\"")) { + return CqlIdentifier.quoted(cqlTableName.substring(1, cqlTableName.length() - 1)); + } + + int separator = cqlTableName.indexOf('.'); + + if (separator != -1) { + return CqlIdentifier.of(cqlTableName.substring(separator + 1)); + } + + return CqlIdentifier.of(cqlTableName); + } + + return CqlIdentifier.of("unknown"); + } } 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 18533dd6c..7f2b051d8 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 @@ -15,26 +15,16 @@ */ package org.springframework.data.cassandra.core; -import java.util.function.Function; - -import lombok.NonNull; import lombok.Value; - -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.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.BeforeDeleteEvent; -import org.springframework.data.cassandra.core.mapping.event.BeforeSaveEvent; -import org.springframework.lang.Nullable; import reactor.core.publisher.Flux; import reactor.core.publisher.Mono; +import java.util.function.Function; + 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.data.cassandra.ReactiveResultSet; import org.springframework.data.cassandra.ReactiveSession; @@ -54,9 +44,17 @@ import org.springframework.data.cassandra.core.cql.session.DefaultReactiveSessio import org.springframework.data.cassandra.core.mapping.CassandraMappingContext; import org.springframework.data.cassandra.core.mapping.CassandraPersistentEntity; import org.springframework.data.cassandra.core.mapping.CassandraPersistentProperty; +import org.springframework.data.cassandra.core.mapping.event.AfterConvertEvent; +import org.springframework.data.cassandra.core.mapping.event.AfterDeleteEvent; +import org.springframework.data.cassandra.core.mapping.event.AfterLoadEvent; +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.query.Query; import org.springframework.data.mapping.context.MappingContext; +import org.springframework.data.projection.ProjectionFactory; import org.springframework.data.projection.SpelAwareProxyProjectionFactory; +import org.springframework.lang.Nullable; import org.springframework.util.Assert; import org.springframework.util.ClassUtils; @@ -87,9 +85,10 @@ import com.datastax.driver.core.querybuilder.Update; * * @author Mark Paluch * @author John Blum + * @author Lukasz Antoniak * @since 2.0 */ -public class ReactiveCassandraTemplate implements ReactiveCassandraOperations, ApplicationContextAware { +public class ReactiveCassandraTemplate implements ReactiveCassandraOperations, ApplicationEventPublisherAware { private final CassandraConverter converter; @@ -166,8 +165,7 @@ public class ReactiveCassandraTemplate implements ReactiveCassandraOperations, A this.projectionFactory = new SpelAwareProxyProjectionFactory(); } - /* - * (non-Javadoc) + /* (non-Javadoc) * @see org.springframework.data.cassandra.core.ReactiveCassandraOperations#getConverter() */ @Override @@ -175,28 +173,7 @@ public class ReactiveCassandraTemplate implements ReactiveCassandraOperations, A return this.converter; } - private static MappingCassandraConverter newConverter() { - - MappingCassandraConverter converter = new MappingCassandraConverter(); - - converter.afterPropertiesSet(); - - return converter; - } - - /** - * Returns the {@link CassandraMappingContext} used by this template to access mapping meta-data used to store (map) - * objects to Cassandra tables. - * - * @return the {@link CassandraMappingContext} used by this template. - * @see CassandraMappingContext - */ - protected MappingContext, CassandraPersistentProperty> getMappingContext() { - return this.mappingContext; - } - - /* - * (non-Javadoc) + /* (non-Javadoc) * @see org.springframework.data.cassandra.core.ReactiveCassandraOperations#getReactiveCqlOperations() */ @Override @@ -204,39 +181,12 @@ public class ReactiveCassandraTemplate implements ReactiveCassandraOperations, A return this.cqlOperations; } - private CassandraPersistentEntity getRequiredPersistentEntity(Object entity) { - return getRequiredPersistentEntity(entity.getClass()); - } - - private CassandraPersistentEntity getRequiredPersistentEntity(Class entityType) { - return getMappingContext().getRequiredPersistentEntity(ClassUtils.getUserClass(entityType)); - } - - /** - * Returns the {@link StatementFactory} used by this template to construct and run Cassandra CQL statements. - * - * @return the {@link StatementFactory} used by this template to construct and run Cassandra CQL statements. - * @see org.springframework.data.cassandra.core.StatementFactory + /* (non-Javadoc) + * @see org.springframework.context.ApplicationEventPublisherAware#setApplicationEventPublisher(org.springframework.context.ApplicationEventPublisher) */ - protected StatementFactory getStatementFactory() { - return this.statementFactory; - } - - private CqlIdentifier getTableName(Object entity) { - return getRequiredPersistentEntity(entity).getTableName(); - } - - @org.springframework.lang.Nullable - private String guessTableName(Object entity) { - if (getMappingContext().hasPersistentEntityFor(entity.getClass())) { - return getMappingContext().getRequiredPersistentEntity(ClassUtils.getUserClass(entity.getClass())).getTableName().toCql(); - } - // Not an entity. - return null; - } - - CqlIdentifier getTableName(Class entityType) { - return getRequiredPersistentEntity(entityType).getTableName(); + @Override + public void setApplicationEventPublisher(ApplicationEventPublisher applicationEventPublisher) { + this.eventPublisher = applicationEventPublisher; } // ------------------------------------------------------------------------- @@ -275,7 +225,7 @@ public class ReactiveCassandraTemplate implements ReactiveCassandraOperations, A Assert.notNull(cql, "Statement must not be null"); Assert.notNull(entityClass, "Entity type must not be null"); - Function mapper = getMapper(entityClass, entityClass, true); + Function mapper = getMapper(entityClass, entityClass, QueryUtils.getTableName(cql)); return getReactiveCqlOperations().query(cql, (row, rowNum) -> mapper.apply(row)); } @@ -306,10 +256,9 @@ public class ReactiveCassandraTemplate implements ReactiveCassandraOperations, A Flux doSelect(Query query, Class entityClass, CqlIdentifier tableName, Class returnType) { - Function mapper = getMapper(entityClass, returnType, true); + RegularStatement select = getStatementFactory().select(query, getRequiredPersistentEntity(entityClass), tableName); - RegularStatement select = getStatementFactory() - .select(query, getRequiredPersistentEntity(entityClass), tableName); + Function mapper = getMapper(entityClass, returnType, tableName); return getReactiveCqlOperations().query(select, (row, rowNum) -> mapper.apply(row)); } @@ -323,7 +272,7 @@ public class ReactiveCassandraTemplate implements ReactiveCassandraOperations, A Assert.notNull(query, "Query must not be null"); Assert.notNull(entityClass, "Entity type must not be null"); - return selectOne(getStatementFactory().select(query, getRequiredPersistentEntity(entityClass)), entityClass); + return select(query, entityClass).next(); } /* (non-Javadoc) @@ -343,8 +292,8 @@ public class ReactiveCassandraTemplate implements ReactiveCassandraOperations, A Mono doUpdate(Query query, org.springframework.data.cassandra.core.query.Update update, Class entityClass, CqlIdentifier tableName) { - RegularStatement statement = getStatementFactory() - .update(query, update, getRequiredPersistentEntity(entityClass), tableName); + RegularStatement statement = getStatementFactory().update(query, update, getRequiredPersistentEntity(entityClass), + tableName); return getReactiveCqlOperations().execute(new StatementCallback(statement)).next(); } @@ -365,7 +314,10 @@ public class ReactiveCassandraTemplate implements ReactiveCassandraOperations, A RegularStatement delete = getStatementFactory().delete(query, getRequiredPersistentEntity(entityClass), tableName); - return getReactiveCqlOperations().execute(new StatementCallback(delete)).next(); + Mono writeResult = getReactiveCqlOperations().execute(new StatementCallback(delete)) + .doOnSubscribe(it -> maybeEmitEvent(new BeforeDeleteEvent<>(delete, entityClass, tableName))).next(); + + return writeResult.doOnNext(it -> maybeEmitEvent(new AfterDeleteEvent<>(delete, entityClass, tableName))); } // ------------------------------------------------------------------------- @@ -399,8 +351,7 @@ public class ReactiveCassandraTemplate implements ReactiveCassandraOperations, A Mono doCount(Query query, Class entityClass, CqlIdentifier tableName) { - RegularStatement count = - getStatementFactory().count(query, getRequiredPersistentEntity(entityClass), tableName); + RegularStatement count = getStatementFactory().count(query, getRequiredPersistentEntity(entityClass), tableName); return getReactiveCqlOperations().queryForObject(count, Long.class).switchIfEmpty(Mono.just(0L)); } @@ -417,7 +368,6 @@ public class ReactiveCassandraTemplate implements ReactiveCassandraOperations, A CassandraPersistentEntity entity = getRequiredPersistentEntity(entityClass); Select select = QueryBuilder.select().from(entity.getTableName().toCql()); - getConverter().write(id, select.where(), entity); return getReactiveCqlOperations().queryForRows(select).hasElements(); @@ -437,8 +387,8 @@ public class ReactiveCassandraTemplate implements ReactiveCassandraOperations, A Mono doExists(Query query, Class entityClass, CqlIdentifier tableName) { - RegularStatement select = - getStatementFactory().select(query.limit(1), getRequiredPersistentEntity(entityClass), tableName); + RegularStatement select = getStatementFactory().select(query.limit(1), getRequiredPersistentEntity(entityClass), + tableName); return getReactiveCqlOperations().queryForRows(select).hasElements(); } @@ -455,7 +405,6 @@ public class ReactiveCassandraTemplate implements ReactiveCassandraOperations, A CassandraPersistentEntity entity = getRequiredPersistentEntity(entityClass); Select select = QueryBuilder.select().all().from(entity.getTableName().toCql()); - getConverter().write(id, select.where(), entity); return selectOne(select, entityClass); @@ -485,14 +434,11 @@ public class ReactiveCassandraTemplate implements ReactiveCassandraOperations, A Insert insert = QueryUtils.createInsertQuery(tableName.toCql(), entity, options, getConverter()); - maybeEmitEvent(new BeforeSaveEvent(entity, tableName.toCql(), insert)); - // noinspection ConstantConditions - Mono result = getReactiveCqlOperations().execute(new StatementCallback(insert)).next(); + Mono result = getReactiveCqlOperations().execute(new StatementCallback(insert)) + .doOnSubscribe(it -> maybeEmitEvent(new BeforeSaveEvent<>(entity, tableName, insert))).next(); - maybeEmitEvent(new AfterSaveEvent(entity, tableName.toCql())); - - return result; + return result.doOnNext(it -> maybeEmitEvent(new AfterSaveEvent<>(entity, tableName))); } /* (non-Javadoc) @@ -512,15 +458,13 @@ public class ReactiveCassandraTemplate implements ReactiveCassandraOperations, A Assert.notNull(entity, "Entity must not be null"); Assert.notNull(options, "UpdateOptions must not be null"); - Update update = QueryUtils.createUpdateQuery(getTableName(entity).toCql(), entity, options, getConverter()); + CqlIdentifier tableName = getTableName(entity); + Update update = QueryUtils.createUpdateQuery(tableName.toCql(), entity, options, getConverter()); - maybeEmitEvent(new BeforeSaveEvent(entity, guessTableName(entity), update)); + Mono result = getReactiveCqlOperations().execute(new StatementCallback(update)) + .doOnSubscribe(it -> maybeEmitEvent(new BeforeSaveEvent<>(entity, tableName, update))).next(); - Mono result = getReactiveCqlOperations().execute(new StatementCallback(update)).next(); - - maybeEmitEvent(new AfterSaveEvent(entity, guessTableName(entity))); - - return result; + return result.doOnNext(it -> maybeEmitEvent(new AfterSaveEvent<>(entity, tableName))); } /* (non-Javadoc) @@ -540,15 +484,13 @@ public class ReactiveCassandraTemplate implements ReactiveCassandraOperations, A Assert.notNull(entity, "Entity must not be null"); Assert.notNull(options, "QueryOptions must not be null"); - Delete delete = QueryUtils.createDeleteQuery(getTableName(entity).toCql(), entity, options, getConverter()); + CqlIdentifier tableName = getTableName(entity); + Delete delete = QueryUtils.createDeleteQuery(tableName.toCql(), entity, options, getConverter()); - maybeEmitEvent(new BeforeDeleteEvent(entity, guessTableName(entity), delete)); + Mono result = getReactiveCqlOperations().execute(new StatementCallback(delete)) + .doOnSubscribe(it -> maybeEmitEvent(new BeforeDeleteEvent<>(delete, entity.getClass(), tableName))).next(); - Mono result = getReactiveCqlOperations().execute(new StatementCallback(delete)).next(); - - maybeEmitEvent(new AfterDeleteEvent(entity, guessTableName(entity))); - - return result; + return result.doOnNext(it -> maybeEmitEvent(new AfterDeleteEvent<>(delete, entity.getClass(), tableName))); } /* (non-Javadoc) @@ -562,11 +504,15 @@ public class ReactiveCassandraTemplate implements ReactiveCassandraOperations, A CassandraPersistentEntity entity = getRequiredPersistentEntity(entityClass); - Delete delete = QueryBuilder.delete().from(entity.getTableName().toCql()); + CqlIdentifier tableName = entity.getTableName(); + Delete delete = QueryBuilder.delete().from(tableName.toCql()); getConverter().write(id, delete.where(), entity); - return getReactiveCqlOperations().execute(delete); + Mono result = getReactiveCqlOperations().execute(delete) + .doOnSubscribe(it -> maybeEmitEvent(new BeforeDeleteEvent<>(delete, entityClass, tableName))); + + return result.doOnNext(it -> maybeEmitEvent(new AfterDeleteEvent<>(delete, entityClass, tableName))); } /* (non-Javadoc) @@ -577,9 +523,13 @@ public class ReactiveCassandraTemplate implements ReactiveCassandraOperations, A Assert.notNull(entityClass, "Entity type must not be null"); - Truncate truncate = QueryBuilder.truncate(getTableName(entityClass).toCql()); + CqlIdentifier tableName = getTableName(entityClass); + Truncate truncate = QueryBuilder.truncate(tableName.toCql()); - return getReactiveCqlOperations().execute(truncate).then(); + Mono result = getReactiveCqlOperations().execute(truncate) + .doOnSubscribe(it -> maybeEmitEvent(new BeforeDeleteEvent<>(truncate, entityClass, tableName))); + + return result.doOnNext(it -> maybeEmitEvent(new AfterDeleteEvent<>(truncate, entityClass, tableName))).then(); } // ------------------------------------------------------------------------- @@ -618,35 +568,71 @@ public class ReactiveCassandraTemplate implements ReactiveCassandraOperations, A return new ReactiveDeleteOperationSupport(this).delete(domainType); } - @Override - public void setApplicationContext(ApplicationContext applicationContext) throws BeansException { - this.eventPublisher = applicationContext; - } - // ------------------------------------------------------------------------- // Implementation hooks and helper methods // ------------------------------------------------------------------------- - private void maybeEmitEvent(ApplicationEvent event) { - if (eventPublisher != null) { - eventPublisher.publishEvent(event); - } + /** + * Returns the {@link CassandraMappingContext} used by this template to access mapping meta-data in order to store + * (map) object to Cassandra tables. + * + * @return the {@link CassandraMappingContext} used by this template. + * @see org.springframework.data.cassandra.core.mapping.CassandraMappingContext + */ + protected MappingContext, CassandraPersistentProperty> getMappingContext() { + return this.mappingContext; + } + + /** + * Returns a reference to the configured {@link ProjectionFactory} used by this template to process CQL query + * projections. + * + * @return a reference to the configured {@link ProjectionFactory} used by this template to process CQL query + * projections. + * @see org.springframework.data.projection.SpelAwareProxyProjectionFactory + * @since 2.1 + */ + protected SpelAwareProxyProjectionFactory getProjectionFactory() { + return this.projectionFactory; + } + + /** + * Returns the {@link StatementFactory} used by this template to construct and run Cassandra CQL statements. + * + * @return the {@link StatementFactory} used by this template to construct and run Cassandra CQL statements. + * @see org.springframework.data.cassandra.core.StatementFactory + * @since 2.1 + */ + protected StatementFactory getStatementFactory() { + return this.statementFactory; + } + + CqlIdentifier getTableName(Class entityClass) { + return getRequiredPersistentEntity(entityClass).getTableName(); + } + + private CqlIdentifier getTableName(Object entity) { + return getRequiredPersistentEntity(entity.getClass()).getTableName(); + } + + private CassandraPersistentEntity getRequiredPersistentEntity(Class entityType) { + return getMappingContext().getRequiredPersistentEntity(ClassUtils.getUserClass(entityType)); } @SuppressWarnings("unchecked") - private Function getMapper(Class entityType, Class targetType, boolean emitEvents) { + private Function getMapper(Class entityType, Class targetType, CqlIdentifier tableName) { Class typeToRead = resolveTypeToRead(entityType, targetType); return row -> { + maybeEmitEvent(new AfterLoadEvent<>(row, targetType, tableName)); + Object source = getConverter().read(typeToRead, row); - T result = (T) (targetType.isInterface() ? this.projectionFactory.createProjection(targetType, source) : source); + T result = (T) (targetType.isInterface() ? getProjectionFactory().createProjection(targetType, source) : source); - if (emitEvents) { - maybeEmitEvent(new AfterLoadEvent(result, guessTableName(result))); - } + maybeEmitEvent(new AfterConvertEvent<>(row, result, tableName)); return result; }; @@ -656,10 +642,26 @@ public class ReactiveCassandraTemplate implements ReactiveCassandraOperations, A return targetType.isInterface() || targetType.isAssignableFrom(entityType) ? entityType : targetType; } + private void maybeEmitEvent(ApplicationEvent event) { + + if (eventPublisher != null) { + eventPublisher.publishEvent(event); + } + } + + private static MappingCassandraConverter newConverter() { + + MappingCassandraConverter converter = new MappingCassandraConverter(); + + converter.afterPropertiesSet(); + + return converter; + } + @Value static class StatementCallback implements ReactiveSessionCallback, CqlProvider { - @NonNull Statement statement; + @lombok.NonNull Statement statement; /* (non-Javadoc) * @see org.springframework.data.cassandra.core.cql.ReactiveSessionCallback#doInSession(org.springframework.data.cassandra.ReactiveSession) diff --git a/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/mapping/event/AbstractCassandraEventListener.java b/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/mapping/event/AbstractCassandraEventListener.java index 6a9cdefda..3371b1b7a 100644 --- a/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/mapping/event/AbstractCassandraEventListener.java +++ b/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/mapping/event/AbstractCassandraEventListener.java @@ -1,5 +1,5 @@ /* - * Copyright 2013-2018 the original author or authors. + * Copyright 2018 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. @@ -17,33 +17,69 @@ package org.springframework.data.cassandra.core.mapping.event; import org.slf4j.Logger; import org.slf4j.LoggerFactory; - import org.springframework.context.ApplicationListener; import org.springframework.core.GenericTypeResolver; -import org.springframework.data.cassandra.config.CassandraCqlClusterFactoryBean; /** - * Base class to implement domain specific {@link ApplicationListener}s. + * Base class to implement domain specific {@link ApplicationListener}s for {@link CassandraMappingEvent}. * * @author Lukasz Antoniak + * @author Mark Paluch + * @since 2.1 */ public abstract class AbstractCassandraEventListener implements ApplicationListener> { + protected static final Logger log = LoggerFactory.getLogger(AbstractCassandraEventListener.class); + private final Class domainClass; /** * Creates a new {@link AbstractCassandraEventListener}. */ public AbstractCassandraEventListener() { + Class typeArgument = GenericTypeResolver.resolveTypeArgument(getClass(), AbstractCassandraEventListener.class); + this.domainClass = typeArgument == null ? Object.class : typeArgument; } + /* (non-Javadoc) + * @see org.springframework.context.ApplicationListener#onApplicationEvent(org.springframework.context.ApplicationEvent) + */ @SuppressWarnings({ "unchecked" }) @Override public void onApplicationEvent(CassandraMappingEvent event) { + Object source = event.getSource(); + if (event instanceof AfterLoadEvent) { + + AfterLoadEvent afterLoadEvent = (AfterLoadEvent) event; + + if (domainClass.isAssignableFrom(afterLoadEvent.getType())) { + onAfterLoad((AfterLoadEvent) event); + } + + return; + } + + if (event instanceof AbstractDeleteEvent) { + + Class eventDomainType = ((AbstractDeleteEvent) event).getType(); + + if (eventDomainType != null && domainClass.isAssignableFrom(eventDomainType)) { + + if (event instanceof BeforeDeleteEvent) { + onBeforeDelete((BeforeDeleteEvent) event); + } + if (event instanceof AfterDeleteEvent) { + onAfterDelete((AfterDeleteEvent) event); + } + } + + return; + } + // Check for matching domain type and invoke callbacks. if (!domainClass.isAssignableFrom(source.getClass())) { return; @@ -51,14 +87,10 @@ public abstract class AbstractCassandraEventListener implements ApplicationLi if (event instanceof BeforeSaveEvent) { onBeforeSave((BeforeSaveEvent) event); - } else if ( event instanceof AfterSaveEvent ) { + } else if (event instanceof AfterSaveEvent) { onAfterSave((AfterSaveEvent) event); - } else if ( event instanceof BeforeDeleteEvent ) { - onBeforeDelete((BeforeDeleteEvent) event); - } else if ( event instanceof AfterDeleteEvent ) { - onAfterDelete((AfterDeleteEvent) event); - } else if ( event instanceof AfterLoadEvent ) { - onAfterLoad((AfterLoadEvent) event); + } else if (event instanceof AfterConvertEvent) { + onAfterConvert((AfterConvertEvent) event); } } @@ -116,4 +148,15 @@ public abstract class AbstractCassandraEventListener implements ApplicationLi log.debug("onAfterLoad({})", event.getSource()); } } + + /** + * Captures {@link AfterConvertEvent}. + * + * @param event will never be {@literal null}. + */ + public void onAfterConvert(AfterConvertEvent event) { + if (log.isDebugEnabled()) { + log.debug("onAfterConvert({})", event.getSource()); + } + } } diff --git a/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/mapping/event/AbstractDeleteEvent.java b/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/mapping/event/AbstractDeleteEvent.java new file mode 100644 index 000000000..683823ee5 --- /dev/null +++ b/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/mapping/event/AbstractDeleteEvent.java @@ -0,0 +1,56 @@ +/* + * Copyright 2018 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.mapping.event; + +import org.springframework.data.cassandra.core.cql.CqlIdentifier; +import org.springframework.util.Assert; + +import com.datastax.driver.core.Statement; + +/** + * Base class for delete events. + * + * @author Mark Paluch + * @since 2.1 + */ +public class AbstractDeleteEvent extends AbstractStatementAwareMappingEvent { + + private final Class type; + + /** + * Creates new {@link AbstractDeleteEvent}. + * + * @param source must not be {@literal null}. + * @param type must not be {@literal null}. + * @param tableName may be {@literal null}. + */ + public AbstractDeleteEvent(Statement source, Class type, CqlIdentifier tableName) { + + super(source, source, tableName); + + Assert.notNull(type, "Type must not be null!"); + this.type = type; + } + + /** + * Returns the type for which the {@link AbstractDeleteEvent} shall be invoked for. + * + * @return + */ + public Class getType() { + return type; + } +} diff --git a/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/mapping/event/AbstractStatementAwareMappingEvent.java b/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/mapping/event/AbstractStatementAwareMappingEvent.java index 23f56a85c..861eb3207 100644 --- a/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/mapping/event/AbstractStatementAwareMappingEvent.java +++ b/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/mapping/event/AbstractStatementAwareMappingEvent.java @@ -1,5 +1,5 @@ /* - * Copyright 2013-2018 the original author or authors. + * Copyright 2018 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. @@ -15,25 +15,31 @@ */ package org.springframework.data.cassandra.core.mapping.event; +import org.springframework.data.cassandra.core.cql.CqlIdentifier; + import com.datastax.driver.core.Statement; -import org.springframework.lang.Nullable; /** * Event encapsulating Cassandra CQL statement. * * @author Lukasz Antoniak + * @author Mark Paluch + * @since 2.1 */ public abstract class AbstractStatementAwareMappingEvent extends CassandraMappingEvent { + private final Statement statement; + /** * Creates new {@link AbstractStatementAwareMappingEvent}. * * @param source must not be {@literal null}. - * @param table may be {@literal null}. * @param statement must not be {@literal null}. + * @param tableName must not be {@literal null}. */ - public AbstractStatementAwareMappingEvent(T source, @Nullable String table, Statement statement) { - super(source, table); + public AbstractStatementAwareMappingEvent(T source, Statement statement, CqlIdentifier tableName) { + + super(source, tableName); this.statement = statement; } diff --git a/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/mapping/event/AfterConvertEvent.java b/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/mapping/event/AfterConvertEvent.java new file mode 100644 index 000000000..7a3682df4 --- /dev/null +++ b/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/mapping/event/AfterConvertEvent.java @@ -0,0 +1,57 @@ +/* + * Copyright 2018 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.mapping.event; + +import org.springframework.data.cassandra.core.cql.CqlIdentifier; +import org.springframework.util.Assert; + +import com.datastax.driver.core.Row; + +/** + * Event to be triggered after converting a {@link Row}. + * + * @author Mark Paluch + * @since 2.1 + */ +public class AfterConvertEvent extends CassandraMappingEvent { + + private static final long serialVersionUID = 1L; + + private final Row row; + + /** + * Creates a new {@link AfterConvertEvent} for the given {@code source} and {@link CqlIdentifier tableName}. + * + * @param source must not be {@literal null}. + * @param tableName must not be {@literal null}. + */ + public AfterConvertEvent(Row row, E source, CqlIdentifier tableName) { + + super(source, tableName); + + Assert.notNull(row, "Row must not be null"); + this.row = row; + } + + /** + * Returns the {@link Row} from which this {@link AfterConvertEvent} was derived. + * + * @return + */ + public Row getRow() { + return row; + } +} diff --git a/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/mapping/event/AfterDeleteEvent.java b/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/mapping/event/AfterDeleteEvent.java index c12fa0fa1..062b8a6ac 100644 --- a/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/mapping/event/AfterDeleteEvent.java +++ b/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/mapping/event/AfterDeleteEvent.java @@ -1,5 +1,5 @@ /* - * Copyright 2013-2018 the original author or authors. + * Copyright 2018 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. @@ -15,17 +15,29 @@ */ package org.springframework.data.cassandra.core.mapping.event; -import org.springframework.lang.Nullable; +import org.springframework.data.cassandra.core.cql.CqlIdentifier; + +import com.datastax.driver.core.Statement; /** - * Event thrown after a single row has been deleted. + * Event being thrown after a single or a set of rows has/have been deleted. * * @author Lukasz Antoniak + * @author Mark Paluch + * @since 2.1 */ -public class AfterDeleteEvent extends CassandraMappingEvent { +public class AfterDeleteEvent extends AbstractDeleteEvent { + private static final long serialVersionUID = 1L; - public AfterDeleteEvent(E source, @Nullable String table) { - super(source, table); + /** + * Create a new {@link AfterDeleteEvent}. + * + * @param source must not be {@literal null}. + * @param type must not be {@literal null}. + * @param tableName must not be {@literal null}. + */ + public AfterDeleteEvent(Statement source, Class type, CqlIdentifier tableName) { + super(source, type, tableName); } } diff --git a/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/mapping/event/AfterLoadEvent.java b/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/mapping/event/AfterLoadEvent.java index ec080d5ea..05e836689 100644 --- a/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/mapping/event/AfterLoadEvent.java +++ b/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/mapping/event/AfterLoadEvent.java @@ -1,5 +1,5 @@ /* - * Copyright 2013-2018 the original author or authors. + * Copyright 2018 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. @@ -15,17 +15,45 @@ */ package org.springframework.data.cassandra.core.mapping.event; -import org.springframework.lang.Nullable; +import org.springframework.data.cassandra.core.cql.CqlIdentifier; +import org.springframework.util.Assert; + +import com.datastax.driver.core.Row; /** - * Event thrown after loading one or multiple rows which are further mapped into given type. + * Event to be triggered after loading {@link com.datastax.driver.core.Row}s to be mapped onto a given type. * * @author Lukasz Antoniak + * @author Mark Paluch + * @since 2.1 */ -public class AfterLoadEvent extends CassandraMappingEvent { +public class AfterLoadEvent extends CassandraMappingEvent { + private static final long serialVersionUID = 1L; - public AfterLoadEvent(E source, @Nullable String table) { - super(source, table); + private final Class type; + + /** + * Creates a new {@link AfterLoadEvent} for the given {@link Row}, type and {@link CqlIdentifier tableName}. + * + * @param source must not be {@literal null}. + * @param type must not be {@literal null}. + * @param tableName must not be {@literal null}. + */ + public AfterLoadEvent(Row source, Class type, CqlIdentifier tableName) { + + super(source, tableName); + + Assert.notNull(type, "Type must not be null!"); + this.type = type; + } + + /** + * Returns the type for which the {@link AfterLoadEvent} shall be invoked for. + * + * @return + */ + public Class getType() { + return type; } } diff --git a/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/mapping/event/AfterSaveEvent.java b/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/mapping/event/AfterSaveEvent.java index 2f462770a..c92438b36 100644 --- a/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/mapping/event/AfterSaveEvent.java +++ b/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/mapping/event/AfterSaveEvent.java @@ -1,5 +1,5 @@ /* - * Copyright 2013-2018 the original author or authors. + * Copyright 2018 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. @@ -15,17 +15,26 @@ */ package org.springframework.data.cassandra.core.mapping.event; -import org.springframework.lang.Nullable; +import org.springframework.data.cassandra.core.cql.CqlIdentifier; /** - * Event thrown after a single row has been inserted. + * {@link CassandraMappingEvent} triggered after save of an object. * * @author Lukasz Antoniak + * @author Mark Paluch + * @since 2.1 */ public class AfterSaveEvent extends CassandraMappingEvent { + private static final long serialVersionUID = 1L; - public AfterSaveEvent(E source, @Nullable String table) { - super(source, table); + /** + * Creates a new {@link AfterSaveEvent}. + * + * @param source must not be {@literal null}. + * @param tableName must not be {@literal null}. + */ + public AfterSaveEvent(E source, CqlIdentifier tableName) { + super(source, tableName); } } diff --git a/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/mapping/event/BeforeDeleteEvent.java b/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/mapping/event/BeforeDeleteEvent.java index a91fbd2f9..6db6410db 100644 --- a/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/mapping/event/BeforeDeleteEvent.java +++ b/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/mapping/event/BeforeDeleteEvent.java @@ -1,5 +1,5 @@ /* - * Copyright 2013-2018 the original author or authors. + * Copyright 2018 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. @@ -15,18 +15,29 @@ */ package org.springframework.data.cassandra.core.mapping.event; +import org.springframework.data.cassandra.core.cql.CqlIdentifier; + import com.datastax.driver.core.Statement; -import org.springframework.lang.Nullable; /** - * Event thrown before a single row is deleted. + * Event thrown before a row or a set of rows is deleted. * * @author Lukasz Antoniak + * @author Mark Paluch + * @since 2.1 */ -public class BeforeDeleteEvent extends AbstractStatementAwareMappingEvent { +public class BeforeDeleteEvent extends AbstractDeleteEvent { + private static final long serialVersionUID = 1L; - public BeforeDeleteEvent(E source, @Nullable String table, Statement statement) { - super(source, table, statement); + /** + * Create a new {@link BeforeDeleteEvent}. + * + * @param source must not be {@literal null}. + * @param type must not be {@literal null}. + * @param tableName must not be {@literal null}. + */ + public BeforeDeleteEvent(Statement source, Class type, CqlIdentifier tableName) { + super(source, type, tableName); } } diff --git a/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/mapping/event/BeforeSaveEvent.java b/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/mapping/event/BeforeSaveEvent.java index dc194a859..700a7743d 100644 --- a/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/mapping/event/BeforeSaveEvent.java +++ b/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/mapping/event/BeforeSaveEvent.java @@ -1,5 +1,5 @@ /* - * Copyright 2013-2018 the original author or authors. + * Copyright 2018 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. @@ -15,18 +15,29 @@ */ package org.springframework.data.cassandra.core.mapping.event; +import org.springframework.data.cassandra.core.cql.CqlIdentifier; + import com.datastax.driver.core.Statement; -import org.springframework.lang.Nullable; /** - * Event thrown before a single row has been inserted. + * {@link CassandraMappingEvent} triggered before save of an object. * * @author Lukasz Antoniak + * @author Mark Paluch + * @since 2.1 */ public class BeforeSaveEvent extends AbstractStatementAwareMappingEvent { + private static final long serialVersionUID = 1L; - public BeforeSaveEvent(E source, @Nullable String table, Statement statement) { - super(source, table, statement); + /** + * Create a new {@link BeforeSaveEvent}. + * + * @param source must not be {@literal null}. + * @param table must not be {@literal null}. + * @param statement must not be {@literal null}. + */ + public BeforeSaveEvent(E source, CqlIdentifier table, Statement statement) { + super(source, statement, table); } } diff --git a/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/mapping/event/CassandraMappingEvent.java b/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/mapping/event/CassandraMappingEvent.java index 3f90042e0..efef6a628 100644 --- a/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/mapping/event/CassandraMappingEvent.java +++ b/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/mapping/event/CassandraMappingEvent.java @@ -1,5 +1,5 @@ /* - * Copyright 2013-2018 the original author or authors. + * Copyright 2018 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. @@ -16,34 +16,41 @@ package org.springframework.data.cassandra.core.mapping.event; import org.springframework.context.ApplicationEvent; -import org.springframework.lang.Nullable; +import org.springframework.data.cassandra.core.cql.CqlIdentifier; +import org.springframework.util.Assert; /** * Base {@link ApplicationEvent} triggered by Spring Data Cassandra. * * @author Lukasz Antoniak + * @author Mark Paluch + * @since 2.1 */ public class CassandraMappingEvent extends ApplicationEvent { + private static final long serialVersionUID = 1L; - private final @Nullable String table; + + private final CqlIdentifier tableName; /** * Creates new {@link CassandraMappingEvent}. * * @param source must not be {@literal null}. - * @param table may be {@literal null}. + * @param tableName must not be {@literal null}. */ - public CassandraMappingEvent(T source, @Nullable String table) { + public CassandraMappingEvent(T source, CqlIdentifier tableName) { + super(source); - this.table = table; + + Assert.notNull(tableName, "Table name must not be null!"); + this.tableName = tableName; } /** - * @return Table that event refers to. May return {@literal null} for not entity objects. + * @return table name that event refers to. May return {@literal null} for non-entity objects. */ - @Nullable - public String getTable() { - return table; + public CqlIdentifier getTableName() { + return tableName; } /* diff --git a/spring-data-cassandra/src/test/java/org/springframework/data/cassandra/core/QueryUtilsUnitTests.java b/spring-data-cassandra/src/test/java/org/springframework/data/cassandra/core/QueryUtilsUnitTests.java new file mode 100644 index 000000000..dee39a2d3 --- /dev/null +++ b/spring-data-cassandra/src/test/java/org/springframework/data/cassandra/core/QueryUtilsUnitTests.java @@ -0,0 +1,60 @@ +/* + * Copyright 2018 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 org.junit.Test; +import org.springframework.data.cassandra.core.cql.CqlIdentifier; + +import com.datastax.driver.core.SimpleStatement; +import com.datastax.driver.core.querybuilder.QueryBuilder; +import com.datastax.driver.core.querybuilder.Select; + +/** + * Unit tests for {@link QueryUtils}. + * + * @author Mark Paluch + */ +public class QueryUtilsUnitTests { + + @Test // DATACASS-106 + public void shouldRetrieveTableNameFromSelect() { + + Select select = QueryBuilder.select().from("keyspace", "table"); + + CqlIdentifier tableName = QueryUtils.getTableName(select); + + assertThat(tableName).isEqualTo(CqlIdentifier.of("table")); + } + + @Test // DATACASS-106 + public void shouldRetrieveTableNameFromSimpleStatement() { + + assertThat(QueryUtils.getTableName(new SimpleStatement("SELECT * FROM table"))) + .isEqualTo(CqlIdentifier.of("table")); + assertThat(QueryUtils.getTableName(new SimpleStatement("SELECT * FROM foo.table where"))) + .isEqualTo(CqlIdentifier.of("table")); + } + + @Test // DATACASS-106 + public void shouldRetrieveQuotedTableNameFromSimpleStatement() { + + CqlIdentifier tableName = QueryUtils.getTableName(new SimpleStatement("SELECT * from \"table\"")); + + assertThat(tableName).isEqualTo(CqlIdentifier.of("table")); + } +} diff --git a/spring-data-cassandra/src/test/java/org/springframework/data/cassandra/core/mapping/EventListenerIntegrationTests.java b/spring-data-cassandra/src/test/java/org/springframework/data/cassandra/core/mapping/EventListenerIntegrationTests.java deleted file mode 100644 index f1ff9cad1..000000000 --- a/spring-data-cassandra/src/test/java/org/springframework/data/cassandra/core/mapping/EventListenerIntegrationTests.java +++ /dev/null @@ -1,259 +0,0 @@ -/* - * Copyright 2016-2018 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.mapping; - -import java.util.Arrays; -import java.util.Collections; -import java.util.LinkedList; -import java.util.List; -import java.util.stream.Stream; - -import org.junit.After; -import org.junit.Before; -import org.junit.Test; - -import com.datastax.driver.core.Session; -import org.springframework.context.ApplicationListener; -import org.springframework.context.ConfigurableApplicationContext; -import org.springframework.context.annotation.AnnotationConfigApplicationContext; -import org.springframework.context.annotation.Bean; -import org.springframework.context.annotation.Configuration; -import org.springframework.data.cassandra.core.CassandraTemplate; -import org.springframework.data.cassandra.core.mapping.event.AbstractCassandraEventListener; -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.BeforeDeleteEvent; -import org.springframework.data.cassandra.core.mapping.event.BeforeSaveEvent; -import org.springframework.data.cassandra.core.query.Query; -import org.springframework.data.cassandra.domain.User; -import org.springframework.data.cassandra.repository.support.SchemaTestUtils; -import org.springframework.data.cassandra.test.util.AbstractKeyspaceCreatingIntegrationTest; -import org.springframework.data.domain.Slice; - -import static org.assertj.core.api.Assertions.*; - -/** - * Integration tests for callback events. - * - * @author Lukasz Antoniak - */ -public class EventListenerIntegrationTests extends AbstractKeyspaceCreatingIntegrationTest { - private static final CaptureEventListener listener = new CaptureEventListener(); - - private CassandraTemplate template = null; - private ConfigurableApplicationContext context = null; - private User firstUser = null; - - @Before - public void setUp() { - context = new AnnotationConfigApplicationContext(ListenerConfiguration.class); - setUpTemplate(session, context); - - firstUser = new User("id-1", "Johny", "Bravo"); - insert(firstUser); - - listener.clear(); - } - - @After - public void tearDown() { - tearDownTemplate(); - if (context != null) { - context.close(); - context = null; - } - } - - @Test // DATACASS-106 - public void shouldEmitInsertEvents() { - User user = new User("id-2", "Lukasz", "Antoniak"); - insert(user); - - assertThat(listener.getBeforeSave()).isEqualTo(Collections.singletonList(user)); - assertThat(listener.getAfterSave()).isEqualTo(Collections.singletonList(user)); - } - - @Test // DATACASS-106 - public void shouldEmitUpdateEvents() { - firstUser.setLastname("Wayne"); - update(firstUser); - - assertThat(listener.getBeforeSave()).isEqualTo(Collections.singletonList(firstUser)); - assertThat(listener.getAfterSave()).isEqualTo(Collections.singletonList(firstUser)); - } - - @Test // DATACASS-106 - public void shouldEmitDeleteEvents() { - delete(firstUser); - - assertThat(listener.getBeforeDelete()).isEqualTo(Collections.singletonList(firstUser)); - assertThat(listener.getAfterDelete()).isEqualTo(Collections.singletonList(firstUser)); - } - - @Test // DATACASS-106 - public void shouldEmitLoadEvents() { - User user = new User("id-2", "Lukasz", "Antoniak"); - insert(user); - - User loaded = selectOneById(firstUser.getId(), User.class); - assertThat(listener.getAfterLoad()).isEqualTo(Collections.singletonList(loaded)); - - listener.clear(); - - stream("SELECT * FROM users;", User.class).count(); // Just load entire stream. - assertThat(listener.getAfterLoad()).isEqualTo(Arrays.asList(loaded, user)); - - listener.clear(); - - slice(Query.empty(), User.class).getSize(); // Force load entire collection. - assertThat(listener.getAfterLoad()).isEqualTo(Arrays.asList(loaded, user)); - } - - @Test // DATACASS-106 - public void shouldEmitMultipleEvents() { - User user = new User("id-2", "Lukasz", "Antoniak"); - insert(user); - user.setFirstname("Robert"); - update(user); - User loaded = selectOneById("id-1", User.class); - delete(loaded); - - List modificationHistory = Arrays.asList(new User("id-2", "Lukasz", "Antoniak"), new User("id-2", "Robert", "Antoniak")); - assertThat(listener.getBeforeSave()).isEqualTo(modificationHistory); - assertThat(listener.getAfterSave()).isEqualTo(modificationHistory); - assertThat(listener.getBeforeDelete()).isEqualTo(Collections.singletonList(loaded)); - assertThat(listener.getAfterDelete()).isEqualTo(Collections.singletonList(loaded)); - assertThat(listener.getAfterLoad()).isEqualTo(Collections.singletonList(loaded)); - } - - @Configuration - static abstract class ListenerConfiguration { - @Bean - public ApplicationListener listener() { - return listener; - } - } - - private static class CaptureEventListener extends AbstractCassandraEventListener { - private final List beforeSave = new LinkedList(); - private final List afterSave = new LinkedList(); - private final List beforeDelete = new LinkedList(); - private final List afterDelete = new LinkedList(); - private final List afterLoad = new LinkedList(); - - @Override - public void onBeforeSave(BeforeSaveEvent event) { - super.onBeforeSave(event); - beforeSave.add(event.getSource()); - assertThat(event.getTable()).isEqualTo("users"); - assertThat(event.getStatement()).isNotNull(); - } - - @Override - public void onAfterSave(AfterSaveEvent event) { - super.onAfterSave(event); - afterSave.add(event.getSource()); - assertThat(event.getTable()).isEqualTo("users"); - } - - @Override - public void onBeforeDelete(BeforeDeleteEvent event) { - super.onBeforeDelete(event); - beforeDelete.add(event.getSource()); - assertThat(event.getTable()).isEqualTo("users"); - assertThat(event.getStatement()).isNotNull(); - } - - @Override - public void onAfterDelete(AfterDeleteEvent event) { - super.onAfterDelete(event); - afterDelete.add(event.getSource()); - assertThat(event.getTable()).isEqualTo("users"); - } - - @Override - public void onAfterLoad(AfterLoadEvent event) { - super.onAfterLoad(event); - afterLoad.add(event.getSource()); - assertThat(event.getTable()).isEqualTo("users"); - } - - private void clear() { - beforeSave.clear(); - afterSave.clear(); - beforeDelete.clear(); - afterDelete.clear(); - afterLoad.clear(); - } - - private List getBeforeSave() { - return beforeSave; - } - - private List getAfterSave() { - return afterSave; - } - - private List getBeforeDelete() { - return beforeDelete; - } - - private List getAfterDelete() { - return afterDelete; - } - - private List getAfterLoad() { - return afterLoad; - } - } - - protected void setUpTemplate(Session session, ConfigurableApplicationContext context) { - template = new CassandraTemplate(session); - template.setApplicationContext(context); - - SchemaTestUtils.potentiallyCreateTableFor(User.class, template); - SchemaTestUtils.truncate(User.class, template); - } - - protected void tearDownTemplate() { - template = null; - } - - protected void insert(Object entity) { - template.insert(entity); - } - - protected void update(Object entity) { - template.update(entity); - } - - protected void delete(Object entity) { - template.delete(entity); - } - - protected T selectOneById(String id, Class entityClass) { - return template.selectOneById(id, entityClass); - } - - protected Slice slice(Query query, Class entityClass) { - return template.slice(query, entityClass); - } - - protected Stream stream(String statement, Class entityClass) { - return template.stream(statement, entityClass); - } -} diff --git a/spring-data-cassandra/src/test/java/org/springframework/data/cassandra/core/mapping/ReactiveEventListenerIntegrationTests.java b/spring-data-cassandra/src/test/java/org/springframework/data/cassandra/core/mapping/ReactiveEventListenerIntegrationTests.java deleted file mode 100644 index 03e3c17e6..000000000 --- a/spring-data-cassandra/src/test/java/org/springframework/data/cassandra/core/mapping/ReactiveEventListenerIntegrationTests.java +++ /dev/null @@ -1,63 +0,0 @@ -/* - * Copyright 2016-2018 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.mapping; - -import com.datastax.driver.core.Session; -import org.springframework.context.ConfigurableApplicationContext; -import org.springframework.data.cassandra.core.ReactiveCassandraTemplate; -import org.springframework.data.cassandra.core.cql.session.DefaultBridgedReactiveSession; - -/** - * Integration tests for callback events with reactive Cassandra template. - * - * @author Lukasz Antoniak - */ -public class ReactiveEventListenerIntegrationTests extends EventListenerIntegrationTests { - private ReactiveCassandraTemplate reactiveTemplate = null; - - @Override - protected void setUpTemplate(Session session, ConfigurableApplicationContext context) { - super.setUpTemplate(session, context); - reactiveTemplate = new ReactiveCassandraTemplate(new DefaultBridgedReactiveSession(session)); - reactiveTemplate.setApplicationContext(context); - } - - @Override - protected void tearDownTemplate() { - super.tearDownTemplate(); - reactiveTemplate = null; - } - - @Override - protected void insert(Object entity) { - reactiveTemplate.insert(entity).block(); - } - - @Override - protected void update(Object entity) { - reactiveTemplate.update(entity).block(); - } - - @Override - protected void delete(Object entity) { - reactiveTemplate.delete(entity).block(); - } - - @Override - protected T selectOneById(String id, Class entityClass) { - return reactiveTemplate.selectOneById(id, entityClass).block(); - } -} diff --git a/spring-data-cassandra/src/test/java/org/springframework/data/cassandra/core/mapping/event/AsyncCassandraTemplateEventIntegrationTests.java b/spring-data-cassandra/src/test/java/org/springframework/data/cassandra/core/mapping/event/AsyncCassandraTemplateEventIntegrationTests.java new file mode 100644 index 000000000..f4ca1e1a7 --- /dev/null +++ b/spring-data-cassandra/src/test/java/org/springframework/data/cassandra/core/mapping/event/AsyncCassandraTemplateEventIntegrationTests.java @@ -0,0 +1,142 @@ +/* + * Copyright 2018 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.mapping.event; + +import static org.assertj.core.api.Assertions.*; +import static org.springframework.data.cassandra.core.query.Criteria.*; +import static org.springframework.data.cassandra.core.query.Query.*; + +import java.util.List; +import java.util.concurrent.Future; + +import org.junit.Before; +import org.junit.Test; +import org.springframework.data.cassandra.core.AsyncCassandraTemplate; +import org.springframework.data.cassandra.core.cql.CqlIdentifier; +import org.springframework.data.cassandra.core.query.Query; +import org.springframework.data.cassandra.domain.User; + +import com.datastax.driver.core.Statement; + +/** + * Integration test for mapping events via {@link AsyncCassandraTemplate}. + * + * @author Mark Paluch + */ +public class AsyncCassandraTemplateEventIntegrationTests extends EventListenerIntegrationTestSupport { + + AsyncCassandraTemplate template; + + @Before + public void setUp() { + + template = new AsyncCassandraTemplate(session); + template.setApplicationEventPublisher(getApplicationEventPublisher()); + + super.setUp(); + } + + @Test // DATACASS-106 + public void selectWithCallbackShouldEmitEvents() { + + getUninterruptibly(template.select("SELECT * FROM users;", it -> {}, User.class)); + + assertThat(getListener().getAfterLoad()).extracting(CassandraMappingEvent::getTableName) + .contains(CqlIdentifier.of("users")); + assertThat(getListener().getAfterConvert()).extracting(CassandraMappingEvent::getSource).containsOnly(firstUser); + } + + @Test // DATACASS-106 + public void selectByQueryWithCallbackShouldEmitEvents() { + + getUninterruptibly(template.select(query(where("id").is(firstUser.getId())), it -> {}, User.class)); + + assertThat(getListener().getAfterLoad()).extracting(CassandraMappingEvent::getTableName) + .contains(CqlIdentifier.of("users")); + assertThat(getListener().getAfterConvert()).extracting(CassandraMappingEvent::getSource).containsOnly(firstUser); + } + + @Test // DATACASS-106 + public void sliceShouldEmitEvents() { + + getUninterruptibly(template.slice(Query.empty(), User.class)); + + assertThat(getListener().getAfterLoad()).extracting(CassandraMappingEvent::getTableName) + .contains(CqlIdentifier.of("users")); + assertThat(getListener().getAfterConvert()).extracting(CassandraMappingEvent::getSource).containsOnly(firstUser); + } + + @Override + public CassandraOperationsAccessor getAccessor() { + + return new CassandraOperationsAccessor() { + + @Override + public void insert(Object entity) { + getUninterruptibly(template.insert(entity)); + } + + @Override + public void update(Object entity) { + getUninterruptibly(template.update(entity)); + } + + @Override + public void delete(Object entity) { + getUninterruptibly(template.delete(entity)); + } + + @Override + public void deleteById(Object id, Class entityClass) { + getUninterruptibly(template.deleteById(id, entityClass)); + } + + @Override + public void delete(Query query, Class entityClass) { + getUninterruptibly(template.delete(query, entityClass)); + } + + @Override + public void truncate(Class entityClass) { + getUninterruptibly(template.truncate(entityClass)); + } + + @Override + public T selectOneById(String id, Class entityClass) { + return getUninterruptibly(template.selectOneById(id, entityClass)); + } + + @Override + public List select(Query query, Class entityClass) { + return getUninterruptibly(template.select(query, entityClass)); + } + + @Override + public List select(Statement statement, Class entityClass) { + return getUninterruptibly(template.select(statement, entityClass)); + } + }; + } + + private static T getUninterruptibly(Future future) { + + try { + return future.get(); + } catch (Exception cause) { + throw new IllegalStateException(cause); + } + } +} diff --git a/spring-data-cassandra/src/test/java/org/springframework/data/cassandra/core/mapping/event/CassandraTemplateEventIntegrationTests.java b/spring-data-cassandra/src/test/java/org/springframework/data/cassandra/core/mapping/event/CassandraTemplateEventIntegrationTests.java new file mode 100644 index 000000000..fe6d32f3d --- /dev/null +++ b/spring-data-cassandra/src/test/java/org/springframework/data/cassandra/core/mapping/event/CassandraTemplateEventIntegrationTests.java @@ -0,0 +1,119 @@ +/* + * Copyright 2018 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.mapping.event; + +import static org.assertj.core.api.Assertions.*; + +import java.util.List; + +import org.junit.Before; +import org.junit.Test; +import org.springframework.data.cassandra.core.CassandraTemplate; +import org.springframework.data.cassandra.core.cql.CqlIdentifier; +import org.springframework.data.cassandra.core.query.Query; +import org.springframework.data.cassandra.domain.User; + +import com.datastax.driver.core.Statement; + +/** + * Integration test for mapping events via {@link CassandraTemplate}. + * + * @author Mark Paluch + */ +public class CassandraTemplateEventIntegrationTests extends EventListenerIntegrationTestSupport { + + CassandraTemplate template; + + @Before + public void setUp() { + + template = new CassandraTemplate(session); + template.setApplicationEventPublisher(getApplicationEventPublisher()); + + super.setUp(); + } + + @Test // DATACASS-106 + public void streamShouldEmitEvents() { + + template.stream("SELECT * FROM users;", User.class).count(); // Just load entire stream. + + assertThat(getListener().getAfterLoad()).extracting(CassandraMappingEvent::getTableName) + .contains(CqlIdentifier.of("users")); + assertThat(getListener().getAfterConvert()).extracting(CassandraMappingEvent::getSource).containsOnly(firstUser); + } + + @Test // DATACASS-106 + public void sliceShouldEmitEvents() { + + template.slice(Query.empty(), User.class).getSize(); // Force load entire collection. + + assertThat(getListener().getAfterLoad()).extracting(CassandraMappingEvent::getTableName) + .contains(CqlIdentifier.of("users")); + assertThat(getListener().getAfterConvert()).extracting(CassandraMappingEvent::getSource).containsOnly(firstUser); + } + + @Override + public CassandraOperationsAccessor getAccessor() { + + return new CassandraOperationsAccessor() { + @Override + public void insert(Object entity) { + template.insert(entity); + } + + @Override + public void update(Object entity) { + template.update(entity); + } + + @Override + public void delete(Object entity) { + template.delete(entity); + } + + @Override + public void deleteById(Object id, Class entityClass) { + template.deleteById(id, entityClass); + } + + @Override + public void delete(Query query, Class entityClass) { + template.delete(query, entityClass); + } + + @Override + public void truncate(Class entityClass) { + template.truncate(entityClass); + } + + @Override + public T selectOneById(String id, Class entityClass) { + return template.selectOneById(id, entityClass); + } + + @Override + public List select(Query query, Class entityClass) { + return template.select(query, entityClass); + } + + @Override + public List select(Statement statement, Class entityClass) { + return template.select(statement, entityClass); + } + }; + } +} diff --git a/spring-data-cassandra/src/test/java/org/springframework/data/cassandra/core/mapping/event/EventListenerIntegrationTestSupport.java b/spring-data-cassandra/src/test/java/org/springframework/data/cassandra/core/mapping/event/EventListenerIntegrationTestSupport.java new file mode 100644 index 000000000..6d4c5acc2 --- /dev/null +++ b/spring-data-cassandra/src/test/java/org/springframework/data/cassandra/core/mapping/event/EventListenerIntegrationTestSupport.java @@ -0,0 +1,274 @@ +/* + * Copyright 2018 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.mapping.event; + +import static org.assertj.core.api.Assertions.*; +import static org.springframework.data.cassandra.core.query.Criteria.*; +import static org.springframework.data.cassandra.core.query.Query.*; + +import java.util.List; +import java.util.concurrent.CopyOnWriteArrayList; +import java.util.stream.Collectors; + +import org.junit.Before; +import org.junit.Test; +import org.springframework.context.ApplicationEventPublisher; +import org.springframework.data.cassandra.core.CassandraTemplate; +import org.springframework.data.cassandra.core.cql.CqlIdentifier; +import org.springframework.data.cassandra.core.query.Query; +import org.springframework.data.cassandra.domain.User; +import org.springframework.data.cassandra.repository.support.SchemaTestUtils; +import org.springframework.data.cassandra.test.util.AbstractKeyspaceCreatingIntegrationTest; + +import com.datastax.driver.core.SimpleStatement; +import com.datastax.driver.core.Statement; + +/** + * Integration tests for lifecycle events. + * + * @author Lukasz Antoniak + * @author Mark Paluch + */ +public abstract class EventListenerIntegrationTestSupport extends AbstractKeyspaceCreatingIntegrationTest { + + private CaptureEventListener listener = new CaptureEventListener(); + User firstUser = null; + + @Before + public void setUp() { + + CassandraTemplate setup = new CassandraTemplate(session); + + SchemaTestUtils.potentiallyCreateTableFor(User.class, setup); + SchemaTestUtils.truncate(User.class, setup); + + firstUser = new User("id-1", "Johny", "Bravo"); + setup.insert(firstUser); + + setup.setApplicationEventPublisher(getApplicationEventPublisher()); + listener.clear(); + } + + public CaptureEventListener getListener() { + return listener; + } + + public ApplicationEventPublisher getApplicationEventPublisher() { + return it -> listener.onApplicationEvent((CassandraMappingEvent) it); + } + + public abstract CassandraOperationsAccessor getAccessor(); + + @Test // DATACASS-106 + public void selectByIdShouldEmitLoadEvents() { + + User loaded = getAccessor().selectOneById(firstUser.getId(), User.class); + + assertThat(listener.getAfterLoad()).extracting(CassandraMappingEvent::getTableName) + .containsOnly(CqlIdentifier.of("users")); + assertThat(listener.getAfterConvert()).extracting(CassandraMappingEvent::getSource).containsOnly(loaded); + } + + @Test // DATACASS-106 + public void selectByQueryShouldEmitLoadEvents() { + + List loaded = getAccessor().select(query(where("id").is(firstUser.getId())), User.class); + + assertThat(listener.getAfterLoad()).extracting(CassandraMappingEvent::getTableName) + .containsOnly(CqlIdentifier.of("users")); + assertThat(listener.getAfterConvert()).extracting(CassandraMappingEvent::getSource).containsOnly(loaded.get(0)); + } + + @Test // DATACASS-106 + public void selectByStatementShouldEmitLoadEvents() { + + List loaded = getAccessor().select(new SimpleStatement("SELECT * FROM users"), User.class); + + assertThat(listener.getAfterLoad()).extracting(CassandraMappingEvent::getTableName) + .containsOnly(CqlIdentifier.of("users")); + assertThat(listener.getAfterConvert()).extracting(CassandraMappingEvent::getSource).containsOnly(loaded.get(0)); + } + + @Test // DATACASS-106 + public void insertShouldEmitEvents() { + + User user = new User("id-2", "Lukasz", "Antoniak"); + getAccessor().insert(user); + + assertThat(listener.getBeforeSave()).extracting(CassandraMappingEvent::getSource).containsOnly(user); + assertThat(listener.getAfterSave()).extracting(CassandraMappingEvent::getSource).containsOnly(user); + } + + @Test // DATACASS-106 + public void updateShouldEmitEvents() { + + firstUser.setLastname("Wayne"); + getAccessor().update(firstUser); + + assertThat(listener.getBeforeSave()).extracting(CassandraMappingEvent::getSource).containsOnly(firstUser); + assertThat(listener.getAfterSave()).extracting(CassandraMappingEvent::getSource).containsOnly(firstUser); + } + + @Test // DATACASS-106 + public void deleteShouldEmitEvents() { + + getAccessor().delete(firstUser); + + assertThat(listener.getBeforeDelete()).extracting(CassandraMappingEvent::getTableName) + .containsExactly(CqlIdentifier.of("users")); + assertThat(listener.getAfterDelete()).extracting(CassandraMappingEvent::getTableName) + .containsExactly(CqlIdentifier.of("users")); + } + + @Test // DATACASS-106 + public void deleteByIdShouldEmitEvents() { + + getAccessor().deleteById(firstUser.getId(), User.class); + + assertThat(listener.getBeforeDelete()).extracting(CassandraMappingEvent::getTableName) + .containsExactly(CqlIdentifier.of("users")); + assertThat(listener.getAfterDelete()).extracting(CassandraMappingEvent::getTableName) + .containsExactly(CqlIdentifier.of("users")); + } + + @Test // DATACASS-106 + public void deleteByQueryShouldEmitEvents() { + + getAccessor().delete(query(where("id").is(firstUser.getId())), User.class); + + assertThat(listener.getBeforeDelete()).extracting(CassandraMappingEvent::getTableName) + .containsExactly(CqlIdentifier.of("users")); + assertThat(listener.getAfterDelete()).extracting(CassandraMappingEvent::getTableName) + .containsExactly(CqlIdentifier.of("users")); + } + + @Test // DATACASS-106 + public void truncateShouldEmitEvents() { + + getAccessor().truncate(User.class); + + assertThat(listener.getBeforeDelete()).extracting(CassandraMappingEvent::getTableName) + .containsExactly(CqlIdentifier.of("users")); + assertThat(listener.getAfterDelete()).extracting(CassandraMappingEvent::getTableName) + .containsExactly(CqlIdentifier.of("users")); + } + + @Test // DATACASS-106 + public void shouldEmitMultipleEvents() { + + User user = new User("id-2", "Lukasz", "Antoniak"); + getAccessor().insert(user); + + assertThat(listener.getBeforeSave()).hasSize(1); + assertThat(listener.getAfterSave()).hasSize(1); + listener.clear(); + + user.setFirstname("Robert"); + getAccessor().update(user); + + assertThat(listener.getBeforeSave()).hasSize(1); + assertThat(listener.getAfterSave()).hasSize(1); + } + + static class CaptureEventListener extends AbstractCassandraEventListener { + + private final List> events = new CopyOnWriteArrayList<>(); + + @Override + public void onBeforeSave(BeforeSaveEvent event) { + events.add(event); + } + + @Override + public void onAfterSave(AfterSaveEvent event) { + events.add(event); + } + + @Override + public void onBeforeDelete(BeforeDeleteEvent event) { + events.add(event); + } + + @Override + public void onAfterDelete(AfterDeleteEvent event) { + events.add(event); + } + + @Override + public void onAfterLoad(AfterLoadEvent event) { + events.add(event); + } + + @Override + public void onAfterConvert(AfterConvertEvent event) { + events.add(event); + } + + private void clear() { + events.clear(); + } + + List> getBeforeSave() { + return filter(BeforeSaveEvent.class); + } + + List> getAfterSave() { + return filter(AfterSaveEvent.class); + } + + List> getBeforeDelete() { + return filter(BeforeDeleteEvent.class); + } + + List> getAfterDelete() { + return filter(AfterDeleteEvent.class); + } + + List> getAfterConvert() { + return filter(AfterConvertEvent.class); + } + + List> getAfterLoad() { + return filter(AfterLoadEvent.class); + } + + @SuppressWarnings("unchecked") + private List filter(Class targetType) { + return (List) events.stream().filter(targetType::isInstance).map(targetType::cast).collect(Collectors.toList()); + } + } + + interface CassandraOperationsAccessor { + + void insert(Object entity); + + void update(Object entity); + + void delete(Object entity); + + void deleteById(Object id, Class entityClass); + + void delete(Query query, Class entityClass); + + void truncate(Class entityClass); + + T selectOneById(String id, Class entityClass); + + List select(Query query, Class entityClass); + + List select(Statement statement, Class entityClass); + } +} diff --git a/spring-data-cassandra/src/test/java/org/springframework/data/cassandra/core/mapping/event/ReactiveEventListenerIntegrationTestSupport.java b/spring-data-cassandra/src/test/java/org/springframework/data/cassandra/core/mapping/event/ReactiveEventListenerIntegrationTestSupport.java new file mode 100644 index 000000000..1a8c9ff54 --- /dev/null +++ b/spring-data-cassandra/src/test/java/org/springframework/data/cassandra/core/mapping/event/ReactiveEventListenerIntegrationTestSupport.java @@ -0,0 +1,113 @@ +/* + * Copyright 2018 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.mapping.event; + +import reactor.test.StepVerifier; + +import java.util.List; +import java.util.concurrent.CopyOnWriteArrayList; + +import org.junit.Before; +import org.springframework.data.cassandra.core.ReactiveCassandraTemplate; +import org.springframework.data.cassandra.core.cql.session.DefaultBridgedReactiveSession; +import org.springframework.data.cassandra.core.query.Query; + +import com.datastax.driver.core.Statement; + +/** + * Integration test for mapping events via {@link ReactiveCassandraTemplate}. + * + * @author Lukasz Antoniak + * @author Mark Paluch + */ +public class ReactiveEventListenerIntegrationTestSupport extends EventListenerIntegrationTestSupport { + + ReactiveCassandraTemplate template; + + @Before + public void setUp() { + + template = new ReactiveCassandraTemplate(new DefaultBridgedReactiveSession(session)); + template.setApplicationEventPublisher(getApplicationEventPublisher()); + + super.setUp(); + } + + @Override + public CassandraOperationsAccessor getAccessor() { + return new CassandraOperationsAccessor() { + @Override + public void insert(Object entity) { + template.insert(entity).as(StepVerifier::create).expectNextCount(1).verifyComplete(); + } + + @Override + public void update(Object entity) { + template.update(entity).as(StepVerifier::create).expectNextCount(1).verifyComplete(); + } + + @Override + public void delete(Object entity) { + template.delete(entity).as(StepVerifier::create).expectNextCount(1).verifyComplete(); + } + + @Override + public void deleteById(Object id, Class entityClass) { + template.deleteById(id, entityClass).as(StepVerifier::create).expectNextCount(1).verifyComplete(); + } + + @Override + public void delete(Query query, Class entityClass) { + template.delete(query, entityClass).as(StepVerifier::create).expectNextCount(1).verifyComplete(); + } + + @Override + public void truncate(Class entityClass) { + template.truncate(entityClass).as(StepVerifier::create).verifyComplete(); + } + + @Override + public T selectOneById(String id, Class entityClass) { + + List result = new CopyOnWriteArrayList<>(); + template.selectOneById(id, entityClass).as(StepVerifier::create).recordWith(() -> result).expectNextCount(1) + .verifyComplete(); + + return result.get(0); + } + + @Override + public List select(Query query, Class entityClass) { + + List result = new CopyOnWriteArrayList<>(); + template.select(query, entityClass).as(StepVerifier::create).recordWith(() -> result).expectNextCount(1) + .verifyComplete(); + + return result; + } + + @Override + public List select(Statement statement, Class entityClass) { + + List result = new CopyOnWriteArrayList<>(); + template.select(statement, entityClass).as(StepVerifier::create).recordWith(() -> result).expectNextCount(1) + .verifyComplete(); + + return result; + } + }; + } +} diff --git a/src/main/asciidoc/new-features.adoc b/src/main/asciidoc/new-features.adoc index 35dd03116..38e5d8256 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. * <> for CRUD operations. * Cassandra Tuple support via `TupleValue`. * Support for `map` columns using User-defined/converted types. +* <> [[new-features.2-0-0]] == What's new in Spring Data for Apache Cassandra 2.0 diff --git a/src/main/asciidoc/reference/mapping.adoc b/src/main/asciidoc/reference/mapping.adoc index 913a9c5d0..ff7c71419 100644 --- a/src/main/asciidoc/reference/mapping.adoc +++ b/src/main/asciidoc/reference/mapping.adoc @@ -588,7 +588,7 @@ Below is an example of a Spring `Converter` implementation that converts from a Built into the Cassandra mapping framework are several `org.springframework.context.ApplicationEvent` events that your application can respond to by registering special beans in the `ApplicationContext`. By being based on Spring's application context event infrastructure this enables other products, such as Spring Integration, to easily receive these events as they are a well known eventing mechanism in Spring based applications. -To intercept an object before it goes into the database, you'd register a subclass of `org.springframework.data.cassandra.core.mapping.event.AbstractCassandraEventListener` that overrides the `onBeforeSave` method. When the event is dispatched, your listener will be called and passed the domain object (Java entity). +To intercept an object before it goes into the database, you'd register a subclass of `org.springframework.data.cassandra.core.mapping.event.AbstractCassandraEventListener` that overrides the `onBeforeSave(…)` method. When the event is dispatched, your listener will be called and passed the domain object (Java entity). ==== [source,java] @@ -596,7 +596,7 @@ To intercept an object before it goes into the database, you'd register a subcla public class BeforeSaveListener extends AbstractCassandraEventListener { @Override public void onBeforeSave(BeforeSaveEvent event) { - ... change values, delete them, whatever ... + … change values, delete them, whatever … } } ---- @@ -604,11 +604,14 @@ public class BeforeSaveListener extends AbstractCassandraEventListener { Simply declaring these beans in your Spring `ApplicationContext` will cause them to be invoked whenever the event is dispatched. -The list of callback methods that are present in AbstractMappingEventListener are: +The list of callback methods that are present in `AbstractCassandraEventListener` are: -* `onBeforeSave` - called in `CassandraTemplate#insert(...)` and `#update(...)` operations *before* inserting/updating record in the database. -* `onAfterSave` - called in `CassandraTemplate#insert(...)` and `#update(...)` operations *after* inserting/updating record in the database. -* `onBeforeDelete` - called in `CassandraTemplate#delete(Object, QueryOptions)` and `#delete(Object)` operations *before* deleting record from the database. -* `onAfterDelete` - called in `CassandraTemplate#delete(Object, QueryOptions)` and `#delete(Object)` operations *after* deleting record from the database. -* `onAfterLoad` - called in `CassandraTemplate#select(...)`, `#slice(...)`, and `#stream(...)` methods after record is retrieved from the database. +* `onBeforeSave` - called in `CassandraTemplate.insert(…)` and `.update(…)` operations *before* inserting/updating a row in the database. +* `onAfterSave` - called in `CassandraTemplate…insert(…)` and `.update(…)` operations *after* inserting/updating a row in the database. +* `onBeforeDelete` - called in `CassandraTemplate.delete(…)` operations *before* deleting row from the database. +* `onAfterDelete` - called in `CassandraTemplate.delete(…)` operations *after* deleting row from the database. +* `onAfterLoad` - called in `CassandraTemplate.#select(…)`, `.slice(…)`, and `.stream(…)` methods after each row retrieved from the database. +* `onAfterConvert` - called in `CassandraTemplate.#select(…)`, `.slice(…)`, and `.stream(…)` methods after converting a row retrieved from the database to a POJO. + +NOTE: Lifecycle events are only emitted for root level types. Complex types used as properties within an aggregate root are not subject of event publication.