DATACASS-106 - Support persistence lifecycle callbacks.

We now support persistence lifecycle callbacks via Spring's ApplicationEvents. Events are fired upon select, insert, update, delete, and truncate statements. The following events are available:

* BeforeSaveEvent: Before inserting/updating a row in the database, via insert(…) and update(…).
* AfterSaveEvent: After inserting/updating a row in the database, via insert(…) and update(…).
* BeforeDeleteEvent: Before deleting row from the database, via delete(…) and truncate(…).
* AfterDeleteEvent: After deleting row from the database, via delete(…) and truncate(…).
* AfterLoadEvent: After retrieving a row from the database, via select(…), slice(…), and stream(…).
* AfterConvertEvent: After converting a row from the database to a POJO, via select(…), slice(…), and stream(…).

Original pull request: #123.

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<Row, T> mapper = getMapper(entityClass, entityClass);
+		Function<Row, T> 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<Row, T> mapper = getMapper(entityClass, entityClass);
+		Function<Row, T> 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 {
 
 	<T> List<T> doSelect(Query query, Class<?> entityClass, CqlIdentifier tableName, Class<T> returnType) {
 
-		Function<Row, T> mapper = getMapper(entityClass, returnType);
+		Function<Row, T> 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<Object>(entity, tableName.toCql(), insert));
+
 		// noinspection ConstantConditions
-		return getCqlOperations().execute(new StatementCallback(insert));
+		WriteResult result = getCqlOperations().execute(new StatementCallback(insert));
+
+		maybeEmitEvent(new AfterSaveEvent<Object>(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<Object>(entity, guessTableName(entity), update));
+
 		// noinspection ConstantConditions
-		return getCqlOperations().execute(new StatementCallback(update));
+		WriteResult result = getCqlOperations().execute(new StatementCallback(update));
+
+		maybeEmitEvent(new AfterSaveEvent<Object>(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<Object>(entity, guessTableName(entity), delete));
+
 		// noinspection ConstantConditions
-		return getCqlOperations().execute(new StatementCallback(delete));
+		WriteResult result = getCqlOperations().execute(new StatementCallback(delete));
+
+		maybeEmitEvent(new AfterDeleteEvent<Object>(entity, guessTableName(entity)));
+
+		return result;
 	}
 
 	/* (non-Javadoc)
@@ -723,7 +762,7 @@ public class CassandraTemplate implements CassandraOperations {
 	}
 
 	@SuppressWarnings("unchecked")
-	private <T> Function<Row, T> getMapper(Class<?> entityType, Class<T> targetType) {
+	private <T> Function<Row, T> getMapper(Class<?> entityType, Class<T> 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<T>(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<WriteResult>, CqlProvider {
 

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<Row, T> mapper = getMapper(entityClass, entityClass);
+		Function<Row, T> mapper = getMapper(entityClass, entityClass, true);
 
 		return getReactiveCqlOperations().query(cql, (row, rowNum) -> mapper.apply(row));
 	}
@@ -284,7 +306,7 @@ public class ReactiveCassandraTemplate implements ReactiveCassandraOperations {
 
 	<T> Flux<T> doSelect(Query query, Class<?> entityClass, CqlIdentifier tableName, Class<T> returnType) {
 
-		Function<Row, T> mapper = getMapper(entityClass, returnType);
+		Function<Row, T> 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<Object>(entity, tableName.toCql(), insert));
+
 		// noinspection ConstantConditions
-		return getReactiveCqlOperations().execute(new StatementCallback(insert)).next();
+		Mono<WriteResult> result = getReactiveCqlOperations().execute(new StatementCallback(insert)).next();
+
+		maybeEmitEvent(new AfterSaveEvent<Object>(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<Object>(entity, guessTableName(entity), update));
+
+		Mono<WriteResult> result = getReactiveCqlOperations().execute(new StatementCallback(update)).next();
+
+		maybeEmitEvent(new AfterSaveEvent<Object>(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<Object>(entity, guessTableName(entity), delete));
+
+		Mono<WriteResult> result = getReactiveCqlOperations().execute(new StatementCallback(delete)).next();
+
+		maybeEmitEvent(new AfterDeleteEvent<Object>(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 <T> Function<Row, T> getMapper(Class<?> entityType, Class<T> targetType) {
+	private <T> Function<Row, T> getMapper(Class<?> entityType, Class<T> 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<T>(result, guessTableName(result)));
+			}
+
+			return result;
 		};
 	}
 

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<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;
+	}
+
+	@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<E>) event);
+		} 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);
+		}
+	}
+
+	/**
+	 * Captures {@link BeforeSaveEvent}.
+	 *
+	 * @param event will never be {@literal null}.
+	 */
+	public void onBeforeSave(BeforeSaveEvent<E> event) {
+		if (log.isDebugEnabled()) {
+			log.debug("onBeforeSave({})", event.getSource());
+		}
+	}
+
+	/**
+	 * Captures {@link AfterSaveEvent}.
+	 *
+	 * @param event will never be {@literal null}.
+	 */
+	public void onAfterSave(AfterSaveEvent<E> event) {
+		if (log.isDebugEnabled()) {
+			log.debug("onAfterSave({})", event.getSource());
+		}
+	}
+
+	/**
+	 * Captures {@link BeforeDeleteEvent}.
+	 *
+	 * @param event will never be {@literal null}.
+	 */
+	public void onBeforeDelete(BeforeDeleteEvent<E> event) {
+		if (log.isDebugEnabled()) {
+			log.debug("onBeforeDelete({})", event.getSource());
+		}
+	}
+
+	/**
+	 * Captures {@link AfterDeleteEvent}.
+	 *
+	 * @param event will never be {@literal null}.
+	 */
+	public void onAfterDelete(AfterDeleteEvent<E> event) {
+		if (log.isDebugEnabled()) {
+			log.debug("onAfterDelete({})", event.getSource());
+		}
+	}
+
+	/**
+	 * Captures {@link AfterLoadEvent}.
+	 *
+	 * @param event will never be {@literal null}.
+	 */
+	public void onAfterLoad(AfterLoadEvent<E> event) {
+		if (log.isDebugEnabled()) {
+			log.debug("onAfterLoad({})", event.getSource());
+		}
+	}
+}

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<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}.
+	 */
+	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;
+	}
+}

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<E> extends CassandraMappingEvent<E> {
+	private static final long serialVersionUID = 1L;
+
+	public AfterDeleteEvent(E source, @Nullable String table) {
+		super(source, table);
+	}
+}

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

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<E> extends CassandraMappingEvent<E> {
+	private static final long serialVersionUID = 1L;
+
+	public AfterSaveEvent(E source, @Nullable String table) {
+		super(source, table);
+	}
+}

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<E> extends AbstractStatementAwareMappingEvent<E> {
+	private static final long serialVersionUID = 1L;
+
+	public BeforeDeleteEvent(E source, @Nullable String table, Statement statement) {
+		super(source, table, statement);
+	}
+}

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<E> extends AbstractStatementAwareMappingEvent<E> {
+	private static final long serialVersionUID = 1L;
+
+	public BeforeSaveEvent(E source, @Nullable String table, Statement statement) {
+		super(source, table, statement);
+	}
+}

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<T> 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();
+	}
+}

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;

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<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);
+	}
+}

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

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<Person> {
+  @Override
+  public void onBeforeSave(BeforeSaveEvent<Person> 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.
+