Spring/spring-data-cassandra
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

DATACASS-106 - Polishing.

Browse Source 
Add AfterConvertEvent, introduce base class for AbstractDeleteEvent. Turn table name in mapping events to non-nullable. Replace guessTableName(…) with getTableName(…) and table name extraction from statements. Pass table name to converter mapper function for event propagation. Refactor tests to base class and test operations accessor.

Introduce lifecycle events and ProjectionFactory to AsyncCassandraTemplate.

Extend JavaDoc, add author and since tags. Reduce copyright year to inception year of new classes. Extend reference documentation.

Original pull request: #123.
This commit is contained in:
Mark Paluch
2018-03-05 14:36:37 +01:00
parent 774d6b2732
commit 56df77efdf
23 changed files with 1534 additions and 717 deletions
Download Patch File Download Diff File Expand all files Collapse all files

277
spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/AsyncCassandraTemplate.java
View File

@@ -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<? extends CassandraPersistentEntity<?>, 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<Row, T> 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> resultSet = getAsyncCqlOperations().queryForResultSet(statement);
CassandraConverter converter = getConverter();
Function<Row, T> 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<Row, T> 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<Boolean> doDelete(Query query, Class<?> entityClass, CqlIdentifier tableName) {
RegularStatement delete = getStatementFactory().delete(query, getRequiredPersistentEntity(entityClass), tableName);
maybeEmitEvent(new BeforeDeleteEvent<>(delete, entityClass, tableName));
ListenableFuture<Boolean> 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<Row, T> 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<Boolean> 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<Boolean> 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<? extends CassandraPersistentEntity<?>, 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<Integer>) session ->
AsyncResult.forValue(getConfiguredFetchSize(session))).completable().join();
return getAsyncCqlOperations()
.execute((AsyncSessionCallback<Integer>) session -> AsyncResult.forValue(getConfiguredFetchSize(session)))
.completable().join();
}
@SuppressWarnings("unchecked")
private <T> Function<Row, T> getMapper(Class<?> entityType, Class<T> 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<T, S>
@@ -633,9 +751,10 @@ public class AsyncCassandraTemplate implements AsyncCassandraOperations {
}
}
private class AsyncStatementCallback implements AsyncSessionCallback<ResultSet>, CqlProvider {
@Value
class AsyncStatementCallback implements AsyncSessionCallback<ResultSet>, CqlProvider {
private final Statement statement;
@lombok.NonNull Statement statement;
AsyncStatementCallback(Statement statement) {
this.statement = statement;

272
spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/CassandraTemplate.java
View File

@@ -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<? extends CassandraPersistentEntity<?>, 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<Row, T> mapper = getMapper(entityClass, entityClass, true);
Function<Row, T> 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<Row, T> mapper = getMapper(entityClass, entityClass, true);
Function<Row, T> 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
<T> List<T> doSelect(Query query, Class<?> entityClass, CqlIdentifier tableName, Class<T> returnType) {
Function<Row, T> mapper = getMapper(entityClass, returnType, true);
Function<Row, T> 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
<T> Stream<T> doStream(Query query, Class<?> entityClass, CqlIdentifier tableName, Class<T> 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<Row, T> mapper = getMapper(entityClass, entityClass, tableName);
List<T> 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<Object>(entity, tableName.toCql(), insert));
maybeEmitEvent(new BeforeSaveEvent<>(entity, tableName, insert));
// noinspection ConstantConditions
WriteResult result = getCqlOperations().execute(new StatementCallback(insert));
maybeEmitEvent(new AfterSaveEvent<Object>(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<Object>(entity, guessTableName(entity), update));
maybeEmitEvent(new BeforeSaveEvent<>(entity, tableName, update));
// noinspection ConstantConditions
WriteResult result = getCqlOperations().execute(new StatementCallback(update));
maybeEmitEvent(new AfterSaveEvent<Object>(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<Object>(entity, guessTableName(entity), delete));
maybeEmitEvent(new BeforeDeleteEvent<>(delete, entity.getClass(), tableName));
// noinspection ConstantConditions
WriteResult result = getCqlOperations().execute(new StatementCallback(delete));
maybeEmitEvent(new AfterDeleteEvent<Object>(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<? extends CassandraPersistentEntity<?>, 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 <T> Function<Row, T> getMapper(Class<?> entityType, Class<T> targetType, boolean emitEvents) {
private <T> Function<Row, T> getMapper(Class<?> entityType, Class<T> 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<T>(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<WriteResult>, 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();

52
spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/QueryUtils.java
View File

@@ -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");
}
}

252
spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/ReactiveCassandraTemplate.java
View File

@@ -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<? extends CassandraPersistentEntity<?>, 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<Row, T> mapper = getMapper(entityClass, entityClass, true);
Function<Row, T> 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
<T> Flux<T> doSelect(Query query, Class<?> entityClass, CqlIdentifier tableName, Class<T> returnType) {
Function<Row, T> mapper = getMapper(entityClass, returnType, true);
RegularStatement select = getStatementFactory().select(query, getRequiredPersistentEntity(entityClass), tableName);
RegularStatement select = getStatementFactory()
.select(query, getRequiredPersistentEntity(entityClass), tableName);
Function<Row, T> 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<WriteResult> 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> 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<Long> 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<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 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<Object>(entity, tableName.toCql(), insert));
// noinspection ConstantConditions
Mono<WriteResult> result = getReactiveCqlOperations().execute(new StatementCallback(insert)).next();
Mono<WriteResult> result = getReactiveCqlOperations().execute(new StatementCallback(insert))
.doOnSubscribe(it -> maybeEmitEvent(new BeforeSaveEvent<>(entity, tableName, insert))).next();
maybeEmitEvent(new AfterSaveEvent<Object>(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<Object>(entity, guessTableName(entity), update));
Mono<WriteResult> result = getReactiveCqlOperations().execute(new StatementCallback(update))
.doOnSubscribe(it -> maybeEmitEvent(new BeforeSaveEvent<>(entity, tableName, update))).next();
Mono<WriteResult> result = getReactiveCqlOperations().execute(new StatementCallback(update)).next();
maybeEmitEvent(new AfterSaveEvent<Object>(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<Object>(entity, guessTableName(entity), delete));
Mono<WriteResult> result = getReactiveCqlOperations().execute(new StatementCallback(delete))
.doOnSubscribe(it -> maybeEmitEvent(new BeforeDeleteEvent<>(delete, entity.getClass(), tableName))).next();
Mono<WriteResult> result = getReactiveCqlOperations().execute(new StatementCallback(delete)).next();
maybeEmitEvent(new AfterDeleteEvent<Object>(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<Boolean> 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<Boolean> 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<? extends CassandraPersistentEntity<?>, 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 <T> Function<Row, T> getMapper(Class<?> entityType, Class<T> targetType, boolean emitEvents) {
private <T> Function<Row, T> getMapper(Class<?> entityType, Class<T> 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<T>(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<WriteResult>, CqlProvider {
@NonNull Statement statement;
@lombok.NonNull Statement statement;
/* (non-Javadoc)
* @see org.springframework.data.cassandra.core.cql.ReactiveSessionCallback#doInSession(org.springframework.data.cassandra.ReactiveSession)

65
spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/mapping/event/AbstractCassandraEventListener.java
View File

@@ -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<E> implements ApplicationListener<CassandraMappingEvent<?>> {
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<E>) event);
}
return;
}
if (event instanceof AbstractDeleteEvent) {
Class<?> eventDomainType = ((AbstractDeleteEvent<?>) event).getType();
if (eventDomainType != null && domainClass.isAssignableFrom(eventDomainType)) {
if (event instanceof BeforeDeleteEvent) {
onBeforeDelete((BeforeDeleteEvent<E>) event);
}
if (event instanceof AfterDeleteEvent) {
onAfterDelete((AfterDeleteEvent<E>) event);
}
}
return;
}
// Check for matching domain type and invoke callbacks.
if (!domainClass.isAssignableFrom(source.getClass())) {
return;
@@ -51,14 +87,10 @@ public abstract class AbstractCassandraEventListener<E> implements ApplicationLi
if (event instanceof BeforeSaveEvent) {
onBeforeSave((BeforeSaveEvent<E>) event);
} else if ( event instanceof AfterSaveEvent ) {
} else if (event instanceof AfterSaveEvent) {
onAfterSave((AfterSaveEvent<E>) event);
} else if ( event instanceof BeforeDeleteEvent ) {
onBeforeDelete((BeforeDeleteEvent<E>) event);
} else if ( event instanceof AfterDeleteEvent ) {
onAfterDelete((AfterDeleteEvent<E>) event);
} else if ( event instanceof AfterLoadEvent ) {
onAfterLoad((AfterLoadEvent<E>) event);
} else if (event instanceof AfterConvertEvent) {
onAfterConvert((AfterConvertEvent<E>) event);
}
}
@@ -116,4 +148,15 @@ public abstract class AbstractCassandraEventListener<E> implements ApplicationLi
log.debug("onAfterLoad({})", event.getSource());
}
}
/**
* Captures {@link AfterConvertEvent}.
*
* @param event will never be {@literal null}.
*/
public void onAfterConvert(AfterConvertEvent<E> event) {
if (log.isDebugEnabled()) {
log.debug("onAfterConvert({})", event.getSource());
}
}
}

56
spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/mapping/event/AbstractDeleteEvent.java Normal file
View File

@@ -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<T> extends AbstractStatementAwareMappingEvent<Statement> {
private final Class<T> 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<T> 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<T> getType() {
return type;
}
}

16
spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/mapping/event/AbstractStatementAwareMappingEvent.java
View File

@@ -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<T> extends CassandraMappingEvent<T> {
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;
}

57
spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/mapping/event/AfterConvertEvent.java Normal file
View File

@@ -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<E> extends CassandraMappingEvent<E> {
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;
}
}

24
spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/mapping/event/AfterDeleteEvent.java
View File

@@ -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<E> extends CassandraMappingEvent<E> {
public class AfterDeleteEvent<T> extends AbstractDeleteEvent<T> {
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<T> type, CqlIdentifier tableName) {
super(source, type, tableName);
}
}

40
spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/mapping/event/AfterLoadEvent.java
View File

@@ -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<E> extends CassandraMappingEvent<E> {
public class AfterLoadEvent<T> extends CassandraMappingEvent<Row> {
private static final long serialVersionUID = 1L;
public AfterLoadEvent(E source, @Nullable String table) {
super(source, table);
private final Class<T> 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<T> 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<T> getType() {
return type;
}
}

19
spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/mapping/event/AfterSaveEvent.java
View File

@@ -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<E> extends CassandraMappingEvent<E> {
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);
}
}

23
spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/mapping/event/BeforeDeleteEvent.java
View File

@@ -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<E> extends AbstractStatementAwareMappingEvent<E> {
public class BeforeDeleteEvent<T> extends AbstractDeleteEvent<T> {
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<T> type, CqlIdentifier tableName) {
super(source, type, tableName);
}
}

21
spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/mapping/event/BeforeSaveEvent.java
View File

@@ -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<E> extends AbstractStatementAwareMappingEvent<E> {
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);
}
}

27
spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/mapping/event/CassandraMappingEvent.java
View File

@@ -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<T> 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;
}
/*

60
spring-data-cassandra/src/test/java/org/springframework/data/cassandra/core/QueryUtilsUnitTests.java Normal file
View File

@@ -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"));
}
}

259
spring-data-cassandra/src/test/java/org/springframework/data/cassandra/core/mapping/EventListenerIntegrationTests.java
View File

@@ -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<User> 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<User> {
private final List<User> beforeSave = new LinkedList<User>();
private final List<User> afterSave = new LinkedList<User>();
private final List<User> beforeDelete = new LinkedList<User>();
private final List<User> afterDelete = new LinkedList<User>();
private final List<User> afterLoad = new LinkedList<User>();
@Override
public void onBeforeSave(BeforeSaveEvent<User> event) {
super.onBeforeSave(event);
beforeSave.add(event.getSource());
assertThat(event.getTable()).isEqualTo("users");
assertThat(event.getStatement()).isNotNull();
}
@Override
public void onAfterSave(AfterSaveEvent<User> event) {
super.onAfterSave(event);
afterSave.add(event.getSource());
assertThat(event.getTable()).isEqualTo("users");
}
@Override
public void onBeforeDelete(BeforeDeleteEvent<User> event) {
super.onBeforeDelete(event);
beforeDelete.add(event.getSource());
assertThat(event.getTable()).isEqualTo("users");
assertThat(event.getStatement()).isNotNull();
}
@Override
public void onAfterDelete(AfterDeleteEvent<User> event) {
super.onAfterDelete(event);
afterDelete.add(event.getSource());
assertThat(event.getTable()).isEqualTo("users");
}
@Override
public void onAfterLoad(AfterLoadEvent<User> 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<User> getBeforeSave() {
return beforeSave;
}
private List<User> getAfterSave() {
return afterSave;
}
private List<User> getBeforeDelete() {
return beforeDelete;
}
private List<User> getAfterDelete() {
return afterDelete;
}
private List<User> 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> T selectOneById(String id, Class<T> entityClass) {
return template.selectOneById(id, entityClass);
}
protected <T> Slice<T> slice(Query query, Class<T> entityClass) {
return template.slice(query, entityClass);
}
protected <T> Stream<T> stream(String statement, Class<T> entityClass) {
return template.stream(statement, entityClass);
}
}

63
spring-data-cassandra/src/test/java/org/springframework/data/cassandra/core/mapping/ReactiveEventListenerIntegrationTests.java
View File

@@ -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> T selectOneById(String id, Class<T> entityClass) {
return reactiveTemplate.selectOneById(id, entityClass).block();
}
}

142
spring-data-cassandra/src/test/java/org/springframework/data/cassandra/core/mapping/event/AsyncCassandraTemplateEventIntegrationTests.java Normal file
View File

@@ -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> T selectOneById(String id, Class<T> entityClass) {
return getUninterruptibly(template.selectOneById(id, entityClass));
}
@Override
public <T> List<T> select(Query query, Class<T> entityClass) {
return getUninterruptibly(template.select(query, entityClass));
}
@Override
public <T> List<T> select(Statement statement, Class<T> entityClass) {
return getUninterruptibly(template.select(statement, entityClass));
}
};
}
private static <T> T getUninterruptibly(Future<T> future) {
try {
return future.get();
} catch (Exception cause) {
throw new IllegalStateException(cause);
}
}
}

119
spring-data-cassandra/src/test/java/org/springframework/data/cassandra/core/mapping/event/CassandraTemplateEventIntegrationTests.java Normal file
View File

@@ -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> T selectOneById(String id, Class<T> entityClass) {
return template.selectOneById(id, entityClass);
}
@Override
public <T> List<T> select(Query query, Class<T> entityClass) {
return template.select(query, entityClass);
}
@Override
public <T> List<T> select(Statement statement, Class<T> entityClass) {
return template.select(statement, entityClass);
}
};
}
}

274
spring-data-cassandra/src/test/java/org/springframework/data/cassandra/core/mapping/event/EventListenerIntegrationTestSupport.java Normal file
View File

@@ -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<User> 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<User> 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<User> {
private final List<CassandraMappingEvent<?>> events = new CopyOnWriteArrayList<>();
@Override
public void onBeforeSave(BeforeSaveEvent<User> event) {
events.add(event);
}
@Override
public void onAfterSave(AfterSaveEvent<User> event) {
events.add(event);
}
@Override
public void onBeforeDelete(BeforeDeleteEvent<User> event) {
events.add(event);
}
@Override
public void onAfterDelete(AfterDeleteEvent<User> event) {
events.add(event);
}
@Override
public void onAfterLoad(AfterLoadEvent<User> event) {
events.add(event);
}
@Override
public void onAfterConvert(AfterConvertEvent<User> event) {
events.add(event);
}
private void clear() {
events.clear();
}
List<BeforeSaveEvent<User>> getBeforeSave() {
return filter(BeforeSaveEvent.class);
}
List<AfterSaveEvent<User>> getAfterSave() {
return filter(AfterSaveEvent.class);
}
List<BeforeDeleteEvent<User>> getBeforeDelete() {
return filter(BeforeDeleteEvent.class);
}
List<AfterDeleteEvent<User>> getAfterDelete() {
return filter(AfterDeleteEvent.class);
}
List<AfterConvertEvent<User>> getAfterConvert() {
return filter(AfterConvertEvent.class);
}
List<AfterLoadEvent<User>> getAfterLoad() {
return filter(AfterLoadEvent.class);
}
@SuppressWarnings("unchecked")
private <T> List<T> filter(Class<? super T> 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> T selectOneById(String id, Class<T> entityClass);
<T> List<T> select(Query query, Class<T> entityClass);
<T> List<T> select(Statement statement, Class<T> entityClass);
}
}

113
spring-data-cassandra/src/test/java/org/springframework/data/cassandra/core/mapping/event/ReactiveEventListenerIntegrationTestSupport.java Normal file
View File

@@ -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> T selectOneById(String id, Class<T> entityClass) {
List<T> result = new CopyOnWriteArrayList<>();
template.selectOneById(id, entityClass).as(StepVerifier::create).recordWith(() -> result).expectNextCount(1)
.verifyComplete();
return result.get(0);
}
@Override
public <T> List<T> select(Query query, Class<T> entityClass) {
List<T> result = new CopyOnWriteArrayList<>();
template.select(query, entityClass).as(StepVerifier::create).recordWith(() -> result).expectNextCount(1)
.verifyComplete();
return result;
}
@Override
public <T> List<T> select(Statement statement, Class<T> entityClass) {
List<T> result = new CopyOnWriteArrayList<>();
template.select(statement, entityClass).as(StepVerifier::create).recordWith(() -> result).expectNextCount(1)
.verifyComplete();
return result;
}
};
}
}

1
src/main/asciidoc/new-features.adoc
View File

@@ -10,6 +10,7 @@ This chapter summarizes changes and new features for each release.
* <<cassandra.template.query.fluent-template-api,Fluent API>> for CRUD operations.
* Cassandra Tuple support via `TupleValue`.
* Support for `map` columns using User-defined/converted types.
* <<cassandra.mapping-usage.events>>
[[new-features.2-0-0]]
== What's new in Spring Data for Apache Cassandra 2.0

19
src/main/asciidoc/reference/mapping.adoc
View File

@@ -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<Person> {
@Override
public void onBeforeSave(BeforeSaveEvent<Person> event) {
... change values, delete them, whatever ...
… change values, delete them, whatever …
}
}
----
@@ -604,11 +604,14 @@ public class BeforeSaveListener extends AbstractCassandraEventListener<Person> {
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.