diff --git a/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/CassandraTemplate.java b/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/CassandraTemplate.java index e5ef7d4d2..2f32979dd 100644 --- a/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/CassandraTemplate.java +++ b/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/CassandraTemplate.java @@ -23,6 +23,11 @@ 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.dao.DataAccessException; import org.springframework.data.cassandra.SessionFactory; import org.springframework.data.cassandra.core.convert.CassandraConverter; @@ -41,6 +46,11 @@ 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.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; @@ -81,7 +91,7 @@ import com.datastax.driver.core.querybuilder.Update; * @see org.springframework.data.cassandra.core.CassandraOperations * @since 2.0 */ -public class CassandraTemplate implements CassandraOperations { +public class CassandraTemplate implements CassandraOperations, ApplicationContextAware { private final CassandraConverter converter; @@ -93,6 +103,8 @@ public class CassandraTemplate implements CassandraOperations { private final StatementFactory statementFactory; + private @Nullable ApplicationEventPublisher eventPublisher; + /** * Creates an instance of {@link CassandraTemplate} initialized with the given {@link Session} and a default * {@link MappingCassandraConverter}. @@ -202,6 +214,15 @@ public class CassandraTemplate implements CassandraOperations { 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. @@ -290,7 +311,7 @@ public class CassandraTemplate implements CassandraOperations { Assert.notNull(statement, "Statement must not be null"); Assert.notNull(entityClass, "Entity type must not be null"); - Function mapper = getMapper(entityClass, entityClass); + Function mapper = getMapper(entityClass, entityClass, true); return getCqlOperations().query(statement, (row, rowNum) -> mapper.apply(row)); } @@ -306,7 +327,7 @@ public class CassandraTemplate implements CassandraOperations { ResultSet resultSet = getCqlOperations().queryForResultSet(statement); - Function mapper = getMapper(entityClass, entityClass); + Function mapper = getMapper(entityClass, entityClass, true); return QueryUtils.readSlice(resultSet, (row, rowNum) -> mapper.apply(row), 0, getEffectiveFetchSize(statement)); @@ -323,7 +344,7 @@ public class CassandraTemplate implements CassandraOperations { ResultSet resultSet = getCqlOperations().queryForResultSet(statement); - return StreamSupport.stream(resultSet.spliterator(), false).map(getMapper(entityClass, entityClass)); + return StreamSupport.stream(resultSet.spliterator(), false).map(getMapper(entityClass, entityClass, true)); } /* (non-Javadoc) @@ -352,7 +373,7 @@ public class CassandraTemplate implements CassandraOperations { List doSelect(Query query, Class entityClass, CqlIdentifier tableName, Class returnType) { - Function mapper = getMapper(entityClass, returnType); + Function mapper = getMapper(entityClass, returnType, true); RegularStatement select = getStatementFactory() .select(query, getRequiredPersistentEntity(entityClass), tableName); @@ -393,7 +414,7 @@ public class CassandraTemplate implements CassandraOperations { ResultSet resultSet = getCqlOperations().queryForResultSet(statement); - return StreamSupport.stream(resultSet.spliterator(), false).map(getMapper(entityClass, returnType)); + return StreamSupport.stream(resultSet.spliterator(), false).map(getMapper(entityClass, returnType, true)); } /* (non-Javadoc) @@ -578,8 +599,14 @@ public class CassandraTemplate implements CassandraOperations { Insert insert = QueryUtils.createInsertQuery(tableName.toCql(), entity, options, getConverter()); + maybeEmitEvent(new BeforeSaveEvent(entity, tableName.toCql(), insert)); + // noinspection ConstantConditions - return getCqlOperations().execute(new StatementCallback(insert)); + WriteResult result = getCqlOperations().execute(new StatementCallback(insert)); + + maybeEmitEvent(new AfterSaveEvent(entity, tableName.toCql())); + + return result; } /* (non-Javadoc) @@ -601,8 +628,14 @@ public class CassandraTemplate implements CassandraOperations { Update update = QueryUtils.createUpdateQuery(getTableName(entity).toCql(), entity, options, getConverter()); + maybeEmitEvent(new BeforeSaveEvent(entity, guessTableName(entity), update)); + // noinspection ConstantConditions - return getCqlOperations().execute(new StatementCallback(update)); + WriteResult result = getCqlOperations().execute(new StatementCallback(update)); + + maybeEmitEvent(new AfterSaveEvent(entity, guessTableName(entity))); + + return result; } /* (non-Javadoc) @@ -624,8 +657,14 @@ public class CassandraTemplate implements CassandraOperations { Delete delete = QueryUtils.createDeleteQuery(getTableName(entity).toCql(), entity, options, getConverter()); + maybeEmitEvent(new BeforeDeleteEvent(entity, guessTableName(entity), delete)); + // noinspection ConstantConditions - return getCqlOperations().execute(new StatementCallback(delete)); + WriteResult result = getCqlOperations().execute(new StatementCallback(delete)); + + maybeEmitEvent(new AfterDeleteEvent(entity, guessTableName(entity))); + + return result; } /* (non-Javadoc) @@ -723,7 +762,7 @@ public class CassandraTemplate implements CassandraOperations { } @SuppressWarnings("unchecked") - private Function getMapper(Class entityType, Class targetType) { + private Function getMapper(Class entityType, Class targetType, boolean emitEvents) { Class typeToRead = resolveTypeToRead(entityType, targetType); @@ -731,8 +770,13 @@ public class CassandraTemplate implements CassandraOperations { Object source = getConverter().read(typeToRead, row); - return (T) (targetType.isInterface() - ? getProjectionFactory().createProjection(targetType, source) : source); + T result = (T) (targetType.isInterface() ? getProjectionFactory().createProjection(targetType, source) : source); + + if (emitEvents) { + maybeEmitEvent(new AfterLoadEvent(result, guessTableName(result))); + } + + return result; }; } @@ -748,6 +792,17 @@ public class CassandraTemplate implements CassandraOperations { 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); + } + } + @Value static class StatementCallback implements SessionCallback, CqlProvider { diff --git a/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/ReactiveCassandraTemplate.java b/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/ReactiveCassandraTemplate.java index cfec42dc1..18533dd6c 100644 --- a/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/ReactiveCassandraTemplate.java +++ b/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/ReactiveCassandraTemplate.java @@ -20,6 +20,17 @@ 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; @@ -78,7 +89,7 @@ import com.datastax.driver.core.querybuilder.Update; * @author John Blum * @since 2.0 */ -public class ReactiveCassandraTemplate implements ReactiveCassandraOperations { +public class ReactiveCassandraTemplate implements ReactiveCassandraOperations, ApplicationContextAware { private final CassandraConverter converter; @@ -90,6 +101,8 @@ public class ReactiveCassandraTemplate implements ReactiveCassandraOperations { private final SpelAwareProxyProjectionFactory projectionFactory; + private @Nullable ApplicationEventPublisher eventPublisher; + /** * Creates an instance of {@link ReactiveCassandraTemplate} initialized with the given {@link ReactiveSession} and a * default {@link MappingCassandraConverter}. @@ -213,6 +226,15 @@ public class ReactiveCassandraTemplate implements ReactiveCassandraOperations { 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(); } @@ -253,7 +275,7 @@ public class ReactiveCassandraTemplate implements ReactiveCassandraOperations { Assert.notNull(cql, "Statement must not be null"); Assert.notNull(entityClass, "Entity type must not be null"); - Function mapper = getMapper(entityClass, entityClass); + Function mapper = getMapper(entityClass, entityClass, true); return getReactiveCqlOperations().query(cql, (row, rowNum) -> mapper.apply(row)); } @@ -284,7 +306,7 @@ public class ReactiveCassandraTemplate implements ReactiveCassandraOperations { Flux doSelect(Query query, Class entityClass, CqlIdentifier tableName, Class returnType) { - Function mapper = getMapper(entityClass, returnType); + Function mapper = getMapper(entityClass, returnType, true); RegularStatement select = getStatementFactory() .select(query, getRequiredPersistentEntity(entityClass), tableName); @@ -463,8 +485,14 @@ public class ReactiveCassandraTemplate implements ReactiveCassandraOperations { Insert insert = QueryUtils.createInsertQuery(tableName.toCql(), entity, options, getConverter()); + maybeEmitEvent(new BeforeSaveEvent(entity, tableName.toCql(), insert)); + // noinspection ConstantConditions - return getReactiveCqlOperations().execute(new StatementCallback(insert)).next(); + Mono result = getReactiveCqlOperations().execute(new StatementCallback(insert)).next(); + + maybeEmitEvent(new AfterSaveEvent(entity, tableName.toCql())); + + return result; } /* (non-Javadoc) @@ -486,7 +514,13 @@ public class ReactiveCassandraTemplate implements ReactiveCassandraOperations { Update update = QueryUtils.createUpdateQuery(getTableName(entity).toCql(), entity, options, getConverter()); - return getReactiveCqlOperations().execute(new StatementCallback(update)).next(); + maybeEmitEvent(new BeforeSaveEvent(entity, guessTableName(entity), update)); + + Mono result = getReactiveCqlOperations().execute(new StatementCallback(update)).next(); + + maybeEmitEvent(new AfterSaveEvent(entity, guessTableName(entity))); + + return result; } /* (non-Javadoc) @@ -508,7 +542,13 @@ public class ReactiveCassandraTemplate implements ReactiveCassandraOperations { Delete delete = QueryUtils.createDeleteQuery(getTableName(entity).toCql(), entity, options, getConverter()); - return getReactiveCqlOperations().execute(new StatementCallback(delete)).next(); + maybeEmitEvent(new BeforeDeleteEvent(entity, guessTableName(entity), delete)); + + Mono result = getReactiveCqlOperations().execute(new StatementCallback(delete)).next(); + + maybeEmitEvent(new AfterDeleteEvent(entity, guessTableName(entity))); + + return result; } /* (non-Javadoc) @@ -578,12 +618,23 @@ public class ReactiveCassandraTemplate implements ReactiveCassandraOperations { 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); + } + } + @SuppressWarnings("unchecked") - private Function getMapper(Class entityType, Class targetType) { + private Function getMapper(Class entityType, Class targetType, boolean emitEvents) { Class typeToRead = resolveTypeToRead(entityType, targetType); @@ -591,8 +642,13 @@ public class ReactiveCassandraTemplate implements ReactiveCassandraOperations { Object source = getConverter().read(typeToRead, row); - return (T) (targetType.isInterface() - ? this.projectionFactory.createProjection(targetType, source) : source); + T result = (T) (targetType.isInterface() ? this.projectionFactory.createProjection(targetType, source) : source); + + if (emitEvents) { + maybeEmitEvent(new AfterLoadEvent(result, guessTableName(result))); + } + + return result; }; } diff --git a/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/mapping/event/AbstractCassandraEventListener.java b/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/mapping/event/AbstractCassandraEventListener.java new file mode 100644 index 000000000..6a9cdefda --- /dev/null +++ b/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/mapping/event/AbstractCassandraEventListener.java @@ -0,0 +1,119 @@ +/* + * Copyright 2013-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.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. + * + * @author Lukasz Antoniak + */ +public abstract class AbstractCassandraEventListener implements ApplicationListener> { + protected static final Logger log = LoggerFactory.getLogger(AbstractCassandraEventListener.class); + private final Class domainClass; + + /** + * Creates a new {@link AbstractCassandraEventListener}. + */ + public AbstractCassandraEventListener() { + Class typeArgument = GenericTypeResolver.resolveTypeArgument(getClass(), AbstractCassandraEventListener.class); + this.domainClass = typeArgument == null ? Object.class : typeArgument; + } + + @SuppressWarnings({ "unchecked" }) + @Override + public void onApplicationEvent(CassandraMappingEvent event) { + Object source = event.getSource(); + + // Check for matching domain type and invoke callbacks. + if (!domainClass.isAssignableFrom(source.getClass())) { + return; + } + + if (event instanceof BeforeSaveEvent) { + onBeforeSave((BeforeSaveEvent) event); + } else if ( event instanceof AfterSaveEvent ) { + onAfterSave((AfterSaveEvent) event); + } else if ( event instanceof BeforeDeleteEvent ) { + onBeforeDelete((BeforeDeleteEvent) event); + } else if ( event instanceof AfterDeleteEvent ) { + onAfterDelete((AfterDeleteEvent) event); + } else if ( event instanceof AfterLoadEvent ) { + onAfterLoad((AfterLoadEvent) event); + } + } + + /** + * Captures {@link BeforeSaveEvent}. + * + * @param event will never be {@literal null}. + */ + public void onBeforeSave(BeforeSaveEvent event) { + if (log.isDebugEnabled()) { + log.debug("onBeforeSave({})", event.getSource()); + } + } + + /** + * Captures {@link AfterSaveEvent}. + * + * @param event will never be {@literal null}. + */ + public void onAfterSave(AfterSaveEvent event) { + if (log.isDebugEnabled()) { + log.debug("onAfterSave({})", event.getSource()); + } + } + + /** + * Captures {@link BeforeDeleteEvent}. + * + * @param event will never be {@literal null}. + */ + public void onBeforeDelete(BeforeDeleteEvent event) { + if (log.isDebugEnabled()) { + log.debug("onBeforeDelete({})", event.getSource()); + } + } + + /** + * Captures {@link AfterDeleteEvent}. + * + * @param event will never be {@literal null}. + */ + public void onAfterDelete(AfterDeleteEvent event) { + if (log.isDebugEnabled()) { + log.debug("onAfterDelete({})", event.getSource()); + } + } + + /** + * Captures {@link AfterLoadEvent}. + * + * @param event will never be {@literal null}. + */ + public void onAfterLoad(AfterLoadEvent event) { + if (log.isDebugEnabled()) { + log.debug("onAfterLoad({})", event.getSource()); + } + } +} diff --git a/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/mapping/event/AbstractStatementAwareMappingEvent.java b/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/mapping/event/AbstractStatementAwareMappingEvent.java new file mode 100644 index 000000000..23f56a85c --- /dev/null +++ b/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/mapping/event/AbstractStatementAwareMappingEvent.java @@ -0,0 +1,46 @@ +/* + * Copyright 2013-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 com.datastax.driver.core.Statement; +import org.springframework.lang.Nullable; + +/** + * Event encapsulating Cassandra CQL statement. + * + * @author Lukasz Antoniak + */ +public abstract class AbstractStatementAwareMappingEvent extends CassandraMappingEvent { + private final Statement statement; + /** + * Creates new {@link AbstractStatementAwareMappingEvent}. + * + * @param source must not be {@literal null}. + * @param table may be {@literal null}. + * @param statement must not be {@literal null}. + */ + public AbstractStatementAwareMappingEvent(T source, @Nullable String table, Statement statement) { + super(source, table); + this.statement = statement; + } + + /** + * @return CQL statement that is going to be executed. + */ + public Statement getStatement() { + return statement; + } +} diff --git a/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/mapping/event/AfterDeleteEvent.java b/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/mapping/event/AfterDeleteEvent.java new file mode 100644 index 000000000..c12fa0fa1 --- /dev/null +++ b/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/mapping/event/AfterDeleteEvent.java @@ -0,0 +1,31 @@ +/* + * Copyright 2013-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.lang.Nullable; + +/** + * Event thrown after a single row has been deleted. + * + * @author Lukasz Antoniak + */ +public class AfterDeleteEvent extends CassandraMappingEvent { + private static final long serialVersionUID = 1L; + + public AfterDeleteEvent(E source, @Nullable String table) { + super(source, table); + } +} diff --git a/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/mapping/event/AfterLoadEvent.java b/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/mapping/event/AfterLoadEvent.java new file mode 100644 index 000000000..ec080d5ea --- /dev/null +++ b/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/mapping/event/AfterLoadEvent.java @@ -0,0 +1,31 @@ +/* + * Copyright 2013-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.lang.Nullable; + +/** + * Event thrown after loading one or multiple rows which are further mapped into given type. + * + * @author Lukasz Antoniak + */ +public class AfterLoadEvent extends CassandraMappingEvent { + private static final long serialVersionUID = 1L; + + public AfterLoadEvent(E source, @Nullable String table) { + super(source, table); + } +} diff --git a/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/mapping/event/AfterSaveEvent.java b/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/mapping/event/AfterSaveEvent.java new file mode 100644 index 000000000..2f462770a --- /dev/null +++ b/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/mapping/event/AfterSaveEvent.java @@ -0,0 +1,31 @@ +/* + * Copyright 2013-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.lang.Nullable; + +/** + * Event thrown after a single row has been inserted. + * + * @author Lukasz Antoniak + */ +public class AfterSaveEvent extends CassandraMappingEvent { + private static final long serialVersionUID = 1L; + + public AfterSaveEvent(E source, @Nullable String table) { + super(source, table); + } +} diff --git a/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/mapping/event/BeforeDeleteEvent.java b/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/mapping/event/BeforeDeleteEvent.java new file mode 100644 index 000000000..a91fbd2f9 --- /dev/null +++ b/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/mapping/event/BeforeDeleteEvent.java @@ -0,0 +1,32 @@ +/* + * Copyright 2013-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 com.datastax.driver.core.Statement; +import org.springframework.lang.Nullable; + +/** + * Event thrown before a single row is deleted. + * + * @author Lukasz Antoniak + */ +public class BeforeDeleteEvent extends AbstractStatementAwareMappingEvent { + private static final long serialVersionUID = 1L; + + public BeforeDeleteEvent(E source, @Nullable String table, Statement statement) { + super(source, table, statement); + } +} diff --git a/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/mapping/event/BeforeSaveEvent.java b/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/mapping/event/BeforeSaveEvent.java new file mode 100644 index 000000000..dc194a859 --- /dev/null +++ b/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/mapping/event/BeforeSaveEvent.java @@ -0,0 +1,32 @@ +/* + * Copyright 2013-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 com.datastax.driver.core.Statement; +import org.springframework.lang.Nullable; + +/** + * Event thrown before a single row has been inserted. + * + * @author Lukasz Antoniak + */ +public class BeforeSaveEvent extends AbstractStatementAwareMappingEvent { + private static final long serialVersionUID = 1L; + + public BeforeSaveEvent(E source, @Nullable String table, Statement statement) { + super(source, table, statement); + } +} diff --git a/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/mapping/event/CassandraMappingEvent.java b/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/mapping/event/CassandraMappingEvent.java new file mode 100644 index 000000000..3f90042e0 --- /dev/null +++ b/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/mapping/event/CassandraMappingEvent.java @@ -0,0 +1,58 @@ +/* + * Copyright 2013-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.context.ApplicationEvent; +import org.springframework.lang.Nullable; + +/** + * Base {@link ApplicationEvent} triggered by Spring Data Cassandra. + * + * @author Lukasz Antoniak + */ +public class CassandraMappingEvent extends ApplicationEvent { + private static final long serialVersionUID = 1L; + private final @Nullable String table; + + /** + * Creates new {@link CassandraMappingEvent}. + * + * @param source must not be {@literal null}. + * @param table may be {@literal null}. + */ + public CassandraMappingEvent(T source, @Nullable String table) { + super(source); + this.table = table; + } + + /** + * @return Table that event refers to. May return {@literal null} for not entity objects. + */ + @Nullable + public String getTable() { + return table; + } + + /* + * (non-Javadoc) + * @see java.util.EventObject#getSource() + */ + @SuppressWarnings({ "unchecked" }) + @Override + public T getSource() { + return (T) super.getSource(); + } +} diff --git a/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/mapping/event/package-info.java b/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/mapping/event/package-info.java new file mode 100644 index 000000000..bf7281add --- /dev/null +++ b/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/mapping/event/package-info.java @@ -0,0 +1,7 @@ +/** + * Event callback infrastructure for Cassandra mapping subsystem. + */ +@NonNullApi +package org.springframework.data.cassandra.core.mapping.event; + +import org.springframework.lang.NonNullApi; diff --git a/spring-data-cassandra/src/test/java/org/springframework/data/cassandra/core/mapping/EventListenerIntegrationTests.java b/spring-data-cassandra/src/test/java/org/springframework/data/cassandra/core/mapping/EventListenerIntegrationTests.java new file mode 100644 index 000000000..f1ff9cad1 --- /dev/null +++ b/spring-data-cassandra/src/test/java/org/springframework/data/cassandra/core/mapping/EventListenerIntegrationTests.java @@ -0,0 +1,259 @@ +/* + * Copyright 2016-2018 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.springframework.data.cassandra.core.mapping; + +import java.util.Arrays; +import java.util.Collections; +import java.util.LinkedList; +import java.util.List; +import java.util.stream.Stream; + +import org.junit.After; +import org.junit.Before; +import org.junit.Test; + +import com.datastax.driver.core.Session; +import org.springframework.context.ApplicationListener; +import org.springframework.context.ConfigurableApplicationContext; +import org.springframework.context.annotation.AnnotationConfigApplicationContext; +import org.springframework.context.annotation.Bean; +import org.springframework.context.annotation.Configuration; +import org.springframework.data.cassandra.core.CassandraTemplate; +import org.springframework.data.cassandra.core.mapping.event.AbstractCassandraEventListener; +import org.springframework.data.cassandra.core.mapping.event.AfterDeleteEvent; +import org.springframework.data.cassandra.core.mapping.event.AfterLoadEvent; +import org.springframework.data.cassandra.core.mapping.event.AfterSaveEvent; +import org.springframework.data.cassandra.core.mapping.event.BeforeDeleteEvent; +import org.springframework.data.cassandra.core.mapping.event.BeforeSaveEvent; +import org.springframework.data.cassandra.core.query.Query; +import org.springframework.data.cassandra.domain.User; +import org.springframework.data.cassandra.repository.support.SchemaTestUtils; +import org.springframework.data.cassandra.test.util.AbstractKeyspaceCreatingIntegrationTest; +import org.springframework.data.domain.Slice; + +import static org.assertj.core.api.Assertions.*; + +/** + * Integration tests for callback events. + * + * @author Lukasz Antoniak + */ +public class EventListenerIntegrationTests extends AbstractKeyspaceCreatingIntegrationTest { + private static final CaptureEventListener listener = new CaptureEventListener(); + + private CassandraTemplate template = null; + private ConfigurableApplicationContext context = null; + private User firstUser = null; + + @Before + public void setUp() { + context = new AnnotationConfigApplicationContext(ListenerConfiguration.class); + setUpTemplate(session, context); + + firstUser = new User("id-1", "Johny", "Bravo"); + insert(firstUser); + + listener.clear(); + } + + @After + public void tearDown() { + tearDownTemplate(); + if (context != null) { + context.close(); + context = null; + } + } + + @Test // DATACASS-106 + public void shouldEmitInsertEvents() { + User user = new User("id-2", "Lukasz", "Antoniak"); + insert(user); + + assertThat(listener.getBeforeSave()).isEqualTo(Collections.singletonList(user)); + assertThat(listener.getAfterSave()).isEqualTo(Collections.singletonList(user)); + } + + @Test // DATACASS-106 + public void shouldEmitUpdateEvents() { + firstUser.setLastname("Wayne"); + update(firstUser); + + assertThat(listener.getBeforeSave()).isEqualTo(Collections.singletonList(firstUser)); + assertThat(listener.getAfterSave()).isEqualTo(Collections.singletonList(firstUser)); + } + + @Test // DATACASS-106 + public void shouldEmitDeleteEvents() { + delete(firstUser); + + assertThat(listener.getBeforeDelete()).isEqualTo(Collections.singletonList(firstUser)); + assertThat(listener.getAfterDelete()).isEqualTo(Collections.singletonList(firstUser)); + } + + @Test // DATACASS-106 + public void shouldEmitLoadEvents() { + User user = new User("id-2", "Lukasz", "Antoniak"); + insert(user); + + User loaded = selectOneById(firstUser.getId(), User.class); + assertThat(listener.getAfterLoad()).isEqualTo(Collections.singletonList(loaded)); + + listener.clear(); + + stream("SELECT * FROM users;", User.class).count(); // Just load entire stream. + assertThat(listener.getAfterLoad()).isEqualTo(Arrays.asList(loaded, user)); + + listener.clear(); + + slice(Query.empty(), User.class).getSize(); // Force load entire collection. + assertThat(listener.getAfterLoad()).isEqualTo(Arrays.asList(loaded, user)); + } + + @Test // DATACASS-106 + public void shouldEmitMultipleEvents() { + User user = new User("id-2", "Lukasz", "Antoniak"); + insert(user); + user.setFirstname("Robert"); + update(user); + User loaded = selectOneById("id-1", User.class); + delete(loaded); + + List modificationHistory = Arrays.asList(new User("id-2", "Lukasz", "Antoniak"), new User("id-2", "Robert", "Antoniak")); + assertThat(listener.getBeforeSave()).isEqualTo(modificationHistory); + assertThat(listener.getAfterSave()).isEqualTo(modificationHistory); + assertThat(listener.getBeforeDelete()).isEqualTo(Collections.singletonList(loaded)); + assertThat(listener.getAfterDelete()).isEqualTo(Collections.singletonList(loaded)); + assertThat(listener.getAfterLoad()).isEqualTo(Collections.singletonList(loaded)); + } + + @Configuration + static abstract class ListenerConfiguration { + @Bean + public ApplicationListener listener() { + return listener; + } + } + + private static class CaptureEventListener extends AbstractCassandraEventListener { + private final List beforeSave = new LinkedList(); + private final List afterSave = new LinkedList(); + private final List beforeDelete = new LinkedList(); + private final List afterDelete = new LinkedList(); + private final List afterLoad = new LinkedList(); + + @Override + public void onBeforeSave(BeforeSaveEvent event) { + super.onBeforeSave(event); + beforeSave.add(event.getSource()); + assertThat(event.getTable()).isEqualTo("users"); + assertThat(event.getStatement()).isNotNull(); + } + + @Override + public void onAfterSave(AfterSaveEvent event) { + super.onAfterSave(event); + afterSave.add(event.getSource()); + assertThat(event.getTable()).isEqualTo("users"); + } + + @Override + public void onBeforeDelete(BeforeDeleteEvent event) { + super.onBeforeDelete(event); + beforeDelete.add(event.getSource()); + assertThat(event.getTable()).isEqualTo("users"); + assertThat(event.getStatement()).isNotNull(); + } + + @Override + public void onAfterDelete(AfterDeleteEvent event) { + super.onAfterDelete(event); + afterDelete.add(event.getSource()); + assertThat(event.getTable()).isEqualTo("users"); + } + + @Override + public void onAfterLoad(AfterLoadEvent event) { + super.onAfterLoad(event); + afterLoad.add(event.getSource()); + assertThat(event.getTable()).isEqualTo("users"); + } + + private void clear() { + beforeSave.clear(); + afterSave.clear(); + beforeDelete.clear(); + afterDelete.clear(); + afterLoad.clear(); + } + + private List getBeforeSave() { + return beforeSave; + } + + private List getAfterSave() { + return afterSave; + } + + private List getBeforeDelete() { + return beforeDelete; + } + + private List getAfterDelete() { + return afterDelete; + } + + private List getAfterLoad() { + return afterLoad; + } + } + + protected void setUpTemplate(Session session, ConfigurableApplicationContext context) { + template = new CassandraTemplate(session); + template.setApplicationContext(context); + + SchemaTestUtils.potentiallyCreateTableFor(User.class, template); + SchemaTestUtils.truncate(User.class, template); + } + + protected void tearDownTemplate() { + template = null; + } + + protected void insert(Object entity) { + template.insert(entity); + } + + protected void update(Object entity) { + template.update(entity); + } + + protected void delete(Object entity) { + template.delete(entity); + } + + protected T selectOneById(String id, Class entityClass) { + return template.selectOneById(id, entityClass); + } + + protected Slice slice(Query query, Class entityClass) { + return template.slice(query, entityClass); + } + + protected Stream stream(String statement, Class entityClass) { + return template.stream(statement, entityClass); + } +} diff --git a/spring-data-cassandra/src/test/java/org/springframework/data/cassandra/core/mapping/ReactiveEventListenerIntegrationTests.java b/spring-data-cassandra/src/test/java/org/springframework/data/cassandra/core/mapping/ReactiveEventListenerIntegrationTests.java new file mode 100644 index 000000000..03e3c17e6 --- /dev/null +++ b/spring-data-cassandra/src/test/java/org/springframework/data/cassandra/core/mapping/ReactiveEventListenerIntegrationTests.java @@ -0,0 +1,63 @@ +/* + * Copyright 2016-2018 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.springframework.data.cassandra.core.mapping; + +import com.datastax.driver.core.Session; +import org.springframework.context.ConfigurableApplicationContext; +import org.springframework.data.cassandra.core.ReactiveCassandraTemplate; +import org.springframework.data.cassandra.core.cql.session.DefaultBridgedReactiveSession; + +/** + * Integration tests for callback events with reactive Cassandra template. + * + * @author Lukasz Antoniak + */ +public class ReactiveEventListenerIntegrationTests extends EventListenerIntegrationTests { + private ReactiveCassandraTemplate reactiveTemplate = null; + + @Override + protected void setUpTemplate(Session session, ConfigurableApplicationContext context) { + super.setUpTemplate(session, context); + reactiveTemplate = new ReactiveCassandraTemplate(new DefaultBridgedReactiveSession(session)); + reactiveTemplate.setApplicationContext(context); + } + + @Override + protected void tearDownTemplate() { + super.tearDownTemplate(); + reactiveTemplate = null; + } + + @Override + protected void insert(Object entity) { + reactiveTemplate.insert(entity).block(); + } + + @Override + protected void update(Object entity) { + reactiveTemplate.update(entity).block(); + } + + @Override + protected void delete(Object entity) { + reactiveTemplate.delete(entity).block(); + } + + @Override + protected T selectOneById(String id, Class entityClass) { + return reactiveTemplate.selectOneById(id, entityClass).block(); + } +} diff --git a/src/main/asciidoc/reference/mapping.adoc b/src/main/asciidoc/reference/mapping.adoc index bf0292d3d..913a9c5d0 100644 --- a/src/main/asciidoc/reference/mapping.adoc +++ b/src/main/asciidoc/reference/mapping.adoc @@ -582,3 +582,33 @@ Below is an example of a Spring `Converter` implementation that converts from a } ---- + +[[cassandra.mapping-usage.events]] +== Lifecycle Events + +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). + +==== +[source,java] +---- +public class BeforeSaveListener extends AbstractCassandraEventListener { + @Override + public void onBeforeSave(BeforeSaveEvent event) { + ... change values, delete them, whatever ... + } +} +---- +==== + +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: + +* `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. +