DATAMONGO-2261 - Adapt to changes in DATACMNS-1467.

Use the reworked version of the EntityCallback method lookup.
Also fix issues with callbacks not invoked when intended and rework the reactive flow by removing deeply nested constructs.
Update documentation and add EntityCallbacks to BulkOperations.

Original Pull Request: #742

spring-data-mongodb/src/main/java/org/springframework/data/mongodb/core/DefaultBulkOperations.java

@@ -27,9 +27,12 @@ import java.util.stream.Collectors;
 import org.bson.Document;
 import org.bson.conversions.Bson;
 import org.springframework.dao.support.PersistenceExceptionTranslator;
+import org.springframework.data.mapping.callback.EntityCallbacks;
 import org.springframework.data.mongodb.core.convert.QueryMapper;
 import org.springframework.data.mongodb.core.convert.UpdateMapper;
 import org.springframework.data.mongodb.core.mapping.MongoPersistentEntity;
+import org.springframework.data.mongodb.core.mapping.event.BeforeConvertCallback;
+import org.springframework.data.mongodb.core.mapping.event.BeforeSaveCallback;
 import org.springframework.data.mongodb.core.query.Collation;
 import org.springframework.data.mongodb.core.query.Query;
 import org.springframework.data.mongodb.core.query.Update;
@@ -55,7 +58,7 @@ class DefaultBulkOperations implements BulkOperations {
 	private final MongoOperations mongoOperations;
 	private final String collectionName;
 	private final BulkOperationContext bulkOperationContext;
-	private final List<WriteModel<Document>> models = new ArrayList<>();
+	private final List<SourceAwareWriteModelHolder> models = new ArrayList<>();
 
 	private PersistenceExceptionTranslator exceptionTranslator;
 	private @Nullable WriteConcern defaultWriteConcern;
@@ -112,7 +115,8 @@ class DefaultBulkOperations implements BulkOperations {
 
 		Assert.notNull(document, "Document must not be null!");
 
-		models.add(new InsertOneModel<>(getMappedObject(document)));
+		Object source = maybeInvokeBeforeConvertCallback(document);
+		addModel(source, new InsertOneModel<>(getMappedObject(source)));
 
 		return this;
 	}
@@ -226,7 +230,7 @@ class DefaultBulkOperations implements BulkOperations {
 		DeleteOptions deleteOptions = new DeleteOptions();
 		query.getCollation().map(Collation::toMongoCollation).ifPresent(deleteOptions::collation);
 
-		models.add(new DeleteManyModel<>(query.getQueryObject(), deleteOptions));
+		addModel(query, new DeleteManyModel<>(query.getQueryObject(), deleteOptions));
 
 		return this;
 	}
@@ -262,8 +266,9 @@ class DefaultBulkOperations implements BulkOperations {
 		replaceOptions.upsert(options.isUpsert());
 		query.getCollation().map(Collation::toMongoCollation).ifPresent(replaceOptions::collation);
 
-		models.add(
-				new ReplaceOneModel<>(getMappedQuery(query.getQueryObject()), getMappedObject(replacement), replaceOptions));
+		Object source = maybeInvokeBeforeConvertCallback(replacement);
+		addModel(source,
+				new ReplaceOneModel<>(getMappedQuery(query.getQueryObject()), getMappedObject(source), replaceOptions));
 
 		return this;
 	}
@@ -278,7 +283,17 @@ class DefaultBulkOperations implements BulkOperations {
 		try {
 
 			return mongoOperations.execute(collectionName, collection -> {
-				return collection.bulkWrite(models.stream().map(this::mapWriteModel).collect(Collectors.toList()), bulkOptions);
+				return collection.bulkWrite(models.stream().map(it -> {
+
+					if (it.getModel() instanceof InsertOneModel) {
+						maybeInvokeBeforeSaveCallback(it.getSource(), ((InsertOneModel<Document>) it.getModel()).getDocument());
+					}
+					if (it.getModel() instanceof ReplaceOneModel) {
+						maybeInvokeBeforeSaveCallback(it.getSource(), ((ReplaceOneModel<Document>) it.getModel()).getReplacement());
+					}
+
+					return mapWriteModel(it.getModel());
+				}).collect(Collectors.toList()), bulkOptions);
 			});
 		} finally {
 			this.bulkOptions = getBulkWriteOptions(bulkOperationContext.getBulkMode());
@@ -304,9 +319,9 @@ class DefaultBulkOperations implements BulkOperations {
 		query.getCollation().map(Collation::toMongoCollation).ifPresent(options::collation);
 
 		if (multi) {
-			models.add(new UpdateManyModel<>(query.getQueryObject(), update.getUpdateObject(), options));
+			addModel(update, new UpdateManyModel<>(query.getQueryObject(), update.getUpdateObject(), options));
 		} else {
-			models.add(new UpdateOneModel<>(query.getQueryObject(), update.getUpdateObject(), options));
+			addModel(update, new UpdateOneModel<>(query.getQueryObject(), update.getUpdateObject(), options));
 		}
 
 		return this;
@@ -362,10 +377,34 @@ class DefaultBulkOperations implements BulkOperations {
 		}
 
 		Document sink = new Document();
+
 		mongoOperations.getConverter().write(source, sink);
 		return sink;
 	}
 
+	private void addModel(Object source, WriteModel<Document> model) {
+		models.add(new SourceAwareWriteModelHolder(source, model));
+	}
+
+	private Object maybeInvokeBeforeConvertCallback(Object value) {
+
+		if (bulkOperationContext.getEntityCallbacks() == null) {
+			return value;
+		}
+
+		return bulkOperationContext.getEntityCallbacks().callback(BeforeConvertCallback.class, value, collectionName);
+	}
+
+	private Object maybeInvokeBeforeSaveCallback(Object value, Document mappedDocument) {
+
+		if (bulkOperationContext.getEntityCallbacks() == null) {
+			return value;
+		}
+
+		return bulkOperationContext.getEntityCallbacks().callback(BeforeSaveCallback.class, value, mappedDocument,
+				collectionName);
+	}
+
 	private static BulkWriteOptions getBulkWriteOptions(BulkMode bulkMode) {
 
 		BulkWriteOptions options = new BulkWriteOptions();
@@ -395,5 +434,19 @@ class DefaultBulkOperations implements BulkOperations {
 		@NonNull Optional<? extends MongoPersistentEntity<?>> entity;
 		@NonNull QueryMapper queryMapper;
 		@NonNull UpdateMapper updateMapper;
+		EntityCallbacks entityCallbacks;
+	}
+
+	/**
+	 * Value object chaining together an actual source with its {@link WriteModel} representation.
+	 *
+	 * @since 2.2
+	 * @author Christoph Strobl
+	 */
+	@Value
+	private static class SourceAwareWriteModelHolder {
+
+		Object source;
+		WriteModel<Document> model;
 	}
 }

spring-data-mongodb/src/main/java/org/springframework/data/mongodb/core/MongoTemplate.java

@@ -35,7 +35,6 @@ import org.bson.codecs.Codec;
 import org.bson.conversions.Bson;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
-
 import org.springframework.beans.BeansException;
 import org.springframework.context.ApplicationContext;
 import org.springframework.context.ApplicationContextAware;
@@ -56,7 +55,7 @@ import org.springframework.data.geo.GeoResults;
 import org.springframework.data.geo.Metric;
 import org.springframework.data.mapping.PropertyPath;
 import org.springframework.data.mapping.PropertyReferenceException;
-import org.springframework.data.mapping.callback.SimpleEntityCallbacks;
+import org.springframework.data.mapping.callback.EntityCallbacks;
 import org.springframework.data.mapping.context.MappingContext;
 import org.springframework.data.mongodb.MongoDatabaseUtils;
 import org.springframework.data.mongodb.MongoDbFactory;
@@ -143,17 +142,7 @@ import com.mongodb.client.MongoCollection;
 import com.mongodb.client.MongoCursor;
 import com.mongodb.client.MongoDatabase;
 import com.mongodb.client.MongoIterable;
-import com.mongodb.client.model.CountOptions;
-import com.mongodb.client.model.CreateCollectionOptions;
-import com.mongodb.client.model.DeleteOptions;
-import com.mongodb.client.model.FindOneAndDeleteOptions;
-import com.mongodb.client.model.FindOneAndReplaceOptions;
-import com.mongodb.client.model.FindOneAndUpdateOptions;
-import com.mongodb.client.model.ReplaceOptions;
-import com.mongodb.client.model.ReturnDocument;
-import com.mongodb.client.model.UpdateOptions;
-import com.mongodb.client.model.ValidationAction;
-import com.mongodb.client.model.ValidationLevel;
+import com.mongodb.client.model.*;
 import com.mongodb.client.result.DeleteResult;
 import com.mongodb.client.result.UpdateResult;
 
@@ -214,7 +203,7 @@ public class MongoTemplate implements MongoOperations, ApplicationContextAware,
 	private WriteResultChecking writeResultChecking = WriteResultChecking.NONE;
 	private @Nullable ReadPreference readPreference;
 	private @Nullable ApplicationEventPublisher eventPublisher;
-	private @Nullable SimpleEntityCallbacks entityCallbacks;
+	private @Nullable EntityCallbacks entityCallbacks;
 	private @Nullable ResourceLoader resourceLoader;
 	private @Nullable MongoPersistentEntityIndexCreator indexCreator;
 
@@ -360,7 +349,9 @@ public class MongoTemplate implements MongoOperations, ApplicationContextAware,
 
 		eventPublisher = applicationContext;
 
-		entityCallbacks = new SimpleEntityCallbacks(applicationContext);
+		if (entityCallbacks == null) {
+			setEntityCallbacks(EntityCallbacks.create(applicationContext));
+		}
 
 		if (mappingContext instanceof ApplicationEventPublisherAware) {
 			((ApplicationEventPublisherAware) mappingContext).setApplicationEventPublisher(eventPublisher);
@@ -372,6 +363,22 @@ public class MongoTemplate implements MongoOperations, ApplicationContextAware,
 		projectionFactory.setBeanClassLoader(applicationContext.getClassLoader());
 	}
 
+	/**
+	 * Set the {@link EntityCallbacks} instance to use when invoking
+	 * {@link org.springframework.data.mapping.callback.EntityCallback callbacks} like the {@link BeforeSaveCallback}.
+	 * <p />
+	 * Overrides potentially existing {@link EntityCallbacks}.
+	 *
+	 * @param entityCallbacks must not be {@literal null}.
+	 * @throws IllegalArgumentException if the given instance is {@literal null}.
+	 * @since 2.2
+	 */
+	public void setEntityCallbacks(EntityCallbacks entityCallbacks) {
+
+		Assert.notNull(entityCallbacks, "EntityCallbacks must not be null!");
+		this.entityCallbacks = entityCallbacks;
+	}
+
 	/**
 	 * Inspects the given {@link ApplicationContext} for {@link MongoPersistentEntityIndexCreator} and those in turn if
 	 * they were registered for the current {@link MappingContext}. If no creator for the current {@link MappingContext}
@@ -771,7 +778,7 @@ public class MongoTemplate implements MongoOperations, ApplicationContextAware,
 		Assert.hasText(collectionName, "Collection name must not be null or empty!");
 
 		DefaultBulkOperations operations = new DefaultBulkOperations(this, collectionName, new BulkOperationContext(mode,
-				Optional.ofNullable(getPersistentEntity(entityType)), queryMapper, updateMapper));
+				Optional.ofNullable(getPersistentEntity(entityType)), queryMapper, updateMapper, entityCallbacks));
 
 		operations.setExceptionTranslator(exceptionTranslator);
 		operations.setDefaultWriteConcern(writeConcern);
@@ -1098,8 +1105,12 @@ public class MongoTemplate implements MongoOperations, ApplicationContextAware,
 		Document mappedFields = queryMapper.getMappedFields(query.getFieldsObject(), entity);
 		Document mappedSort = queryMapper.getMappedSort(query.getSortObject(), entity);
 
+		replacement = maybeCallBeforeConvert(replacement, collectionName);
 		Document mappedReplacement = operations.forEntity(replacement).toMappedDocument(this.mongoConverter).getDocument();
 
+		maybeEmitEvent(new BeforeSaveEvent<>(replacement, mappedReplacement, collectionName));
+		maybeCallBeforeSave(replacement, mappedReplacement, collectionName);
+
 		return doFindAndReplace(collectionName, mappedQuery, mappedFields, mappedSort,
 				operations.forType(entityType).getCollation(query).map(Collation::toMongoCollation).orElse(null), entityType,
 				mappedReplacement, options, resultType);
@@ -2340,8 +2351,7 @@ public class MongoTemplate implements MongoOperations, ApplicationContextAware,
 	protected <T> T maybeCallBeforeConvert(T object, String collection) {
 
 		if (null != entityCallbacks) {
-			return (T) entityCallbacks.callback(object, BeforeConvertCallback.class,
-					(cb, t) -> cb.onBeforeConvert(t, collection));
+			return entityCallbacks.callback(BeforeConvertCallback.class, object, collection);
 		}
 
 		return object;
@@ -2351,8 +2361,7 @@ public class MongoTemplate implements MongoOperations, ApplicationContextAware,
 	protected <T> T maybeCallBeforeSave(T object, Document document, String collection) {
 
 		if (null != entityCallbacks) {
-			return (T) entityCallbacks.callback(object, BeforeSaveCallback.class,
-					(cb, t) -> cb.onBeforeSave(t, document, collection));
+			return entityCallbacks.callback(BeforeSaveCallback.class, object, document, collection);
 		}
 
 		return object;
@@ -2679,9 +2688,6 @@ public class MongoTemplate implements MongoOperations, ApplicationContextAware,
 					entityType, serializeToJsonSafely(replacement), collectionName);
 		}
 
-		maybeEmitEvent(new BeforeSaveEvent<>(replacement, replacement, collectionName));
-		replacement = maybeCallBeforeSave(replacement, replacement, collectionName);
-
 		return executeFindOneInternal(
 				new FindAndReplaceCallback(mappedQuery, mappedFields, mappedSort, replacement, collation, options),
 				new ProjectingReadCallback<>(mongoConverter, entityType, resultType, collectionName), collectionName);

spring-data-mongodb/src/main/java/org/springframework/data/mongodb/core/ReactiveMongoTemplate.java

@@ -20,7 +20,6 @@ import static org.springframework.data.mongodb.core.query.SerializationUtils.*;
 import lombok.AccessLevel;
 import lombok.NonNull;
 import lombok.RequiredArgsConstructor;
-import org.springframework.util.NumberUtils;
 import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
 import reactor.util.function.Tuple2;
@@ -115,6 +114,7 @@ import org.springframework.lang.Nullable;
 import org.springframework.util.Assert;
 import org.springframework.util.ClassUtils;
 import org.springframework.util.CollectionUtils;
+import org.springframework.util.NumberUtils;
 import org.springframework.util.ObjectUtils;
 import org.springframework.util.ResourceUtils;
 import org.springframework.util.StringUtils;
@@ -358,7 +358,11 @@ public class ReactiveMongoTemplate implements ReactiveMongoOperations, Applicati
 		prepareIndexCreator(applicationContext);
 
 		eventPublisher = applicationContext;
-		entityCallbacks = new ReactiveEntityCallbacks(applicationContext);
+
+		if (entityCallbacks == null) {
+			setEntityCallbacks(ReactiveEntityCallbacks.create(applicationContext));
+		}
+
 		if (mappingContext instanceof ApplicationEventPublisherAware) {
 			((ApplicationEventPublisherAware) mappingContext).setApplicationEventPublisher(eventPublisher);
 		}
@@ -367,6 +371,23 @@ public class ReactiveMongoTemplate implements ReactiveMongoOperations, Applicati
 		projectionFactory.setBeanClassLoader(applicationContext.getClassLoader());
 	}
 
+	/**
+	 * Set the {@link ReactiveEntityCallbacks} instance to use when invoking
+	 * {@link org.springframework.data.mapping.callback.EntityCallback callbacks} like the
+	 * {@link ReactiveBeforeSaveCallback}.
+	 * <p />
+	 * Overrides potentially existing {@link ReactiveEntityCallbacks}.
+	 *
+	 * @param entityCallbacks must not be {@literal null}.
+	 * @throws IllegalArgumentException if the given instance is {@literal null}.
+	 * @since 2.2
+	 */
+	public void setEntityCallbacks(ReactiveEntityCallbacks entityCallbacks) {
+
+		Assert.notNull(entityCallbacks, "EntityCallbacks must not be null!");
+		this.entityCallbacks = entityCallbacks;
+	}
+
 	/**
 	 * Inspects the given {@link ApplicationContext} for {@link ReactiveMongoPersistentEntityIndexCreator} and those in
 	 * turn if they were registered for the current {@link MappingContext}. If no creator for the current
@@ -1166,11 +1187,25 @@ public class ReactiveMongoTemplate implements ReactiveMongoOperations, Applicati
 		Document mappedFields = queryMapper.getMappedFields(query.getFieldsObject(), entity);
 		Document mappedSort = queryMapper.getMappedSort(query.getSortObject(), entity);
 
-		Document mappedReplacement = operations.forEntity(replacement).toMappedDocument(this.mongoConverter).getDocument();
-
-		return doFindAndReplace(collectionName, mappedQuery, mappedFields, mappedSort,
-				operations.forType(entityType).getCollation(query).map(Collation::toMongoCollation).orElse(null), entityType,
-				mappedReplacement, options, resultType);
+		return Mono.just(PersistableEntityModel.of(replacement, collectionName)) //
+				.doOnNext(it -> maybeEmitEvent(new BeforeConvertEvent<>(it.getSource(), it.getCollection()))) //
+				.flatMap(it -> maybeCallBeforeConvert(it.getSource(), it.getCollection()).map(it::mutate))
+				.map(it -> it
+						.addTargetDocument(operations.forEntity(it.getSource()).toMappedDocument(mongoConverter).getDocument())) //
+				.doOnNext(it -> maybeEmitEvent(new BeforeSaveEvent(it.getSource(), it.getTarget(), it.getCollection()))) //
+				.flatMap(it -> {
+
+					PersistableEntityModel<S> flowObject = (PersistableEntityModel<S>) it;
+					return maybeCallBeforeSave(flowObject.getSource(), flowObject.getTarget(), flowObject.getCollection())
+							.map(potentiallyModified -> PersistableEntityModel.of(potentiallyModified, flowObject.getTarget(),
+									flowObject.getCollection()));
+				}).flatMap(it -> {
+
+					PersistableEntityModel<S> flowObject = (PersistableEntityModel<S>) it;
+					return doFindAndReplace(flowObject.getCollection(), mappedQuery, mappedFields, mappedSort,
+							operations.forType(entityType).getCollation(query).map(Collation::toMongoCollation).orElse(null),
+							entityType, flowObject.getTarget(), options, resultType);
+				});
 	}
 
 	/*
@@ -1321,30 +1356,29 @@ public class ReactiveMongoTemplate implements ReactiveMongoOperations, Applicati
 
 	protected <T> Mono<T> doInsert(String collectionName, T objectToSave, MongoWriter<Object> writer) {
 
-		return Mono.defer(() -> {
-
-			BeforeConvertEvent<T> event = new BeforeConvertEvent<>(objectToSave, collectionName);
-			T toConvert = maybeEmitEvent(event).getSource();
-			return maybeCallBeforeConvert(toConvert, collectionName).flatMap(toSave -> {
+		return Mono.just(PersistableEntityModel.of(objectToSave, collectionName)) //
+				.doOnNext(it -> maybeEmitEvent(new BeforeConvertEvent<>(it.getSource(), it.getCollection()))) //
+				.flatMap(it -> maybeCallBeforeConvert(it.getSource(), it.getCollection()).map(it::mutate)) //
+				.map(it -> {
 
-				AdaptibleEntity<T> entity = operations.forEntity(toConvert, mongoConverter.getConversionService());
+					AdaptibleEntity<T> entity = operations.forEntity(it.getSource(), mongoConverter.getConversionService());
 					entity.assertUpdateableIdIfNotSet();
 
-				T initialized = entity.initializeVersionProperty();
-				Document dbDoc = entity.toMappedDocument(writer).getDocument();
+					return PersistableEntityModel.of(entity.initializeVersionProperty(),
+							entity.toMappedDocument(writer).getDocument(), it.getCollection());
+				}).doOnNext(it -> maybeEmitEvent(new BeforeSaveEvent<>(it.getSource(), it.getTarget(), it.getCollection()))) //
+				.flatMap(it -> {
 
-				maybeEmitEvent(new BeforeSaveEvent<>(initialized, dbDoc, collectionName));
-				return maybeCallBeforeSave(initialized, dbDoc, collectionName).flatMap(it -> {
+					return maybeCallBeforeSave(it.getSource(), it.getTarget(), it.getCollection()).map(it::mutate);
 
-					Mono<T> afterInsert = insertDocument(collectionName, dbDoc, it.getClass()).map(id -> {
+				}).flatMap(it -> {
 
-						T saved = entity.populateIdIfNecessary(id);
-						maybeEmitEvent(new AfterSaveEvent<>(saved, dbDoc, collectionName));
-						return saved;
-					});
+					return insertDocument(it.getCollection(), it.getTarget(), it.getSource().getClass()).map(id -> {
 
-					return afterInsert;
-				});
+						T saved = operations.forEntity(it.getSource(), mongoConverter.getConversionService())
+								.populateIdIfNecessary(id);
+						maybeEmitEvent(new AfterSaveEvent<>(saved, it.getTarget(), collectionName));
+						return saved;
 					});
 				});
 	}
@@ -2514,16 +2548,11 @@ public class ReactiveMongoTemplate implements ReactiveMongoOperations, Applicati
 						serializeToJsonSafely(replacement), collectionName);
 			}
 
-			maybeEmitEvent(new BeforeSaveEvent<>(replacement, replacement, collectionName));
-
-			return maybeCallBeforeSave(replacement, replacement, collectionName).flatMap(it -> {
-
 			return executeFindOneInternal(
-						new FindAndReplaceCallback(mappedQuery, mappedFields, mappedSort, it, collation, options),
+					new FindAndReplaceCallback(mappedQuery, mappedFields, mappedSort, replacement, collation, options),
 					new ProjectingReadCallback<>(this.mongoConverter, entityType, resultType, collectionName), collectionName);
 
 		});
-		});
 	}
 
 	protected <E extends MongoMappingEvent<T>, T> E maybeEmitEvent(E event) {
@@ -2539,8 +2568,7 @@ public class ReactiveMongoTemplate implements ReactiveMongoOperations, Applicati
 	protected <T> Mono<T> maybeCallBeforeConvert(T object, String collection) {
 
 		if (null != entityCallbacks) {
-			return entityCallbacks.callbackLater(object, ReactiveBeforeConvertCallback.class,
-					(cb, t) -> cb.onBeforeConvert(t, collection));
+			return entityCallbacks.callback(ReactiveBeforeConvertCallback.class, object, collection);
 		}
 
 		return Mono.just(object);
@@ -2550,8 +2578,7 @@ public class ReactiveMongoTemplate implements ReactiveMongoOperations, Applicati
 	protected <T> Mono<T> maybeCallBeforeSave(T object, Document document, String collection) {
 
 		if (null != entityCallbacks) {
-			return entityCallbacks.callbackLater(object, ReactiveBeforeSaveCallback.class,
-					(cb, t) -> cb.onBeforeSave(t, document, collection));
+			return entityCallbacks.callback(ReactiveBeforeSaveCallback.class, object, document, collection);
 		}
 
 		return Mono.just(object);
@@ -3307,4 +3334,54 @@ public class ReactiveMongoTemplate implements ReactiveMongoOperations, Applicati
 		}
 	}
 
+	/**
+	 * Value object chaining together a given source document with its mapped representation and the collection to persist
+	 * it to.
+	 *
+	 * @param <T>
+	 * @author Christoph Strobl
+	 * @since 2.2
+	 */
+	private static class PersistableEntityModel<T> {
+
+		private final T source;
+		private final @Nullable Document target;
+		private final String collection;
+
+		private PersistableEntityModel(T source, @Nullable Document target, String collection) {
+
+			this.source = source;
+			this.target = target;
+			this.collection = collection;
+		}
+
+		static <T> PersistableEntityModel<T> of(T source, String collection) {
+			return new PersistableEntityModel<>(source, null, collection);
+		}
+
+		static <T> PersistableEntityModel<T> of(T source, Document target, String collection) {
+			return new PersistableEntityModel<>(source, target, collection);
+		}
+
+		PersistableEntityModel<T> mutate(T source) {
+			return new PersistableEntityModel(source, target, collection);
+		}
+
+		PersistableEntityModel<T> addTargetDocument(Document target) {
+			return new PersistableEntityModel(source, target, collection);
+		}
+
+		T getSource() {
+			return source;
+		}
+
+		@Nullable
+		Document getTarget() {
+			return target;
+		}
+
+		String getCollection() {
+			return collection;
+		}
+	}
 }

spring-data-mongodb/src/main/java/org/springframework/data/mongodb/core/mapping/event/BeforeConvertCallback.java

@@ -16,20 +16,18 @@
 package org.springframework.data.mongodb.core.mapping.event;
 
 import org.springframework.data.mapping.callback.EntityCallback;
-import org.springframework.data.mapping.callback.SimpleEntityCallbacks;
 
 /**
  * Callback being invoked before a domain object is converted to be persisted.
  *
  * @author Mark Paluch
  * @since 2.2
- * @see SimpleEntityCallbacks
  */
 @FunctionalInterface
 public interface BeforeConvertCallback<T> extends EntityCallback<T> {
 
 	/**
-	 * Entity callback method invoked before a domain object is converted to be persisted. Can return either the same of a
+	 * Entity callback method invoked before a domain object is converted to be persisted. Can return either the same or a
 	 * modified instance of the domain object.
 	 *
 	 * @param entity the domain object to save.

spring-data-mongodb/src/main/java/org/springframework/data/mongodb/core/mapping/event/BeforeSaveCallback.java

@@ -16,22 +16,19 @@
 package org.springframework.data.mongodb.core.mapping.event;
 
 import org.bson.Document;
-
 import org.springframework.data.mapping.callback.EntityCallback;
-import org.springframework.data.mapping.callback.SimpleEntityCallbacks;
 
 /**
  * Entity callback triggered before save of a document.
  *
  * @author Mark Paluch
  * @since 2.2
- * @see SimpleEntityCallbacks
  */
 @FunctionalInterface
 public interface BeforeSaveCallback<T> extends EntityCallback<T> {
 
 	/**
-	 * Entity callback method invoked before a domain object is saved. Can return either the same of a modified instance
+	 * Entity callback method invoked before a domain object is saved. Can return either the same or a modified instance
 	 * of the domain object and can modify {@link Document} contents. This method called after converting the
 	 * {@code entity} to {@link Document} so effectively the document is used as outcome of invoking this callback.
 	 *

spring-data-mongodb/src/main/java/org/springframework/data/mongodb/core/mapping/event/ReactiveBeforeSaveCallback.java

@@ -32,7 +32,7 @@ import org.springframework.data.mapping.callback.ReactiveEntityCallbacks;
 public interface ReactiveBeforeSaveCallback<T> extends EntityCallback<T> {
 
 	/**
-	 * Entity callback method invoked before a domain object is saved. Can return either the same of a modified instance
+	 * Entity callback method invoked before a domain object is saved. Can return either the same or a modified instance
 	 * of the domain object and can modify {@link Document} contents. This method is called after converting the
 	 * {@code entity} to {@link Document} so effectively the document is used as outcome of invoking this callback.
 	 *

spring-data-mongodb/src/test/java/org/springframework/data/mongodb/config/AuditingIntegrationTests.java

@@ -20,13 +20,12 @@ import static org.junit.Assert.*;
 
 import org.joda.time.DateTime;
 import org.junit.Test;
-
 import org.springframework.context.support.AbstractApplicationContext;
 import org.springframework.context.support.ClassPathXmlApplicationContext;
 import org.springframework.data.annotation.CreatedDate;
 import org.springframework.data.annotation.Id;
 import org.springframework.data.annotation.LastModifiedDate;
-import org.springframework.data.mapping.callback.SimpleEntityCallbacks;
+import org.springframework.data.mapping.callback.EntityCallbacks;
 import org.springframework.data.mongodb.core.mapping.MongoMappingContext;
 import org.springframework.data.mongodb.core.mapping.event.BeforeConvertCallback;
 
@@ -38,7 +37,7 @@ import org.springframework.data.mongodb.core.mapping.event.BeforeConvertCallback
  */
 public class AuditingIntegrationTests {
 
-	@Test // DATAMONGO-577, DATAMONGO-800, DATAMONGO-883, 2261
+	@Test // DATAMONGO-577, DATAMONGO-800, DATAMONGO-883, DATAMONGO-2261
 	public void enablesAuditingAndSetsPropertiesAccordingly() throws Exception {
 
 		AbstractApplicationContext context = new ClassPathXmlApplicationContext("auditing.xml", getClass());
@@ -46,10 +45,10 @@ public class AuditingIntegrationTests {
 		MongoMappingContext mappingContext = context.getBean(MongoMappingContext.class);
 		mappingContext.getPersistentEntity(Entity.class);
 
-		SimpleEntityCallbacks callbacks = new SimpleEntityCallbacks(context);
+		EntityCallbacks callbacks = EntityCallbacks.create(context);
 
 		Entity entity = new Entity();
-		entity = callbacks.callback(entity, BeforeConvertCallback.class, (cb, e) -> cb.onBeforeConvert(e, "collection-1"));
+		entity = callbacks.callback(BeforeConvertCallback.class, entity, "collection-1");
 
 		assertThat(entity.created, is(notNullValue()));
 		assertThat(entity.modified, is(entity.created));
@@ -57,7 +56,7 @@ public class AuditingIntegrationTests {
 		Thread.sleep(10);
 		entity.id = 1L;
 
-		entity = callbacks.callback(entity, BeforeConvertCallback.class, (cb, e) -> cb.onBeforeConvert(e, "collection-1"));
+		entity = callbacks.callback(BeforeConvertCallback.class, entity, "collection-1");
 
 		assertThat(entity.created, is(notNullValue()));
 		assertThat(entity.modified, is(not(entity.created)));

spring-data-mongodb/src/test/java/org/springframework/data/mongodb/core/DefaultBulkOperationsIntegrationTests.java

@@ -72,19 +72,19 @@ public class DefaultBulkOperationsIntegrationTests {
 	@Test(expected = IllegalArgumentException.class) // DATAMONGO-934
 	public void rejectsNullMongoOperations() {
 		new DefaultBulkOperations(null, COLLECTION_NAME,
-				new BulkOperationContext(BulkMode.ORDERED, Optional.empty(), null, null));
+				new BulkOperationContext(BulkMode.ORDERED, Optional.empty(), null, null, null));
 
 	}
 
 	@Test(expected = IllegalArgumentException.class) // DATAMONGO-934
 	public void rejectsNullCollectionName() {
 		new DefaultBulkOperations(operations, null,
-				new BulkOperationContext(BulkMode.ORDERED, Optional.empty(), null, null));
+				new BulkOperationContext(BulkMode.ORDERED, Optional.empty(), null, null, null));
 	}
 
 	@Test(expected = IllegalArgumentException.class) // DATAMONGO-934
 	public void rejectsEmptyCollectionName() {
-		new DefaultBulkOperations(operations, "", new BulkOperationContext(BulkMode.ORDERED, Optional.empty(), null, null));
+		new DefaultBulkOperations(operations, "", new BulkOperationContext(BulkMode.ORDERED, Optional.empty(), null, null, null));
 	}
 
 	@Test // DATAMONGO-934
@@ -341,7 +341,7 @@ public class DefaultBulkOperationsIntegrationTests {
 				: Optional.empty();
 
 		BulkOperationContext bulkOperationContext = new BulkOperationContext(mode, entity,
-				new QueryMapper(operations.getConverter()), new UpdateMapper(operations.getConverter()));
+				new QueryMapper(operations.getConverter()), new UpdateMapper(operations.getConverter()), null);
 
 		DefaultBulkOperations bulkOps = new DefaultBulkOperations(operations, COLLECTION_NAME, bulkOperationContext);
 		bulkOps.setDefaultWriteConcern(WriteConcern.ACKNOWLEDGED);

spring-data-mongodb/src/test/java/org/springframework/data/mongodb/core/DefaultBulkOperationsUnitTests.java

@@ -26,7 +26,6 @@ import static org.springframework.data.mongodb.core.query.Query.*;
 import java.util.List;
 import java.util.Optional;
 
-import com.mongodb.client.model.*;
 import org.bson.Document;
 import org.junit.Before;
 import org.junit.Test;
@@ -36,6 +35,7 @@ import org.mockito.Captor;
 import org.mockito.Mock;
 import org.mockito.junit.MockitoJUnitRunner;
 import org.springframework.data.annotation.Id;
+import org.springframework.data.mapping.callback.EntityCallbacks;
 import org.springframework.data.mongodb.MongoDbFactory;
 import org.springframework.data.mongodb.core.BulkOperations.BulkMode;
 import org.springframework.data.mongodb.core.DefaultBulkOperations.BulkOperationContext;
@@ -46,12 +46,20 @@ import org.springframework.data.mongodb.core.convert.QueryMapper;
 import org.springframework.data.mongodb.core.convert.UpdateMapper;
 import org.springframework.data.mongodb.core.mapping.Field;
 import org.springframework.data.mongodb.core.mapping.MongoMappingContext;
+import org.springframework.data.mongodb.core.mapping.event.BeforeConvertCallback;
+import org.springframework.data.mongodb.core.mapping.event.BeforeSaveCallback;
 import org.springframework.data.mongodb.core.query.BasicQuery;
 import org.springframework.data.mongodb.core.query.Collation;
 import org.springframework.data.mongodb.core.query.Update;
 
 import com.mongodb.client.MongoCollection;
 import com.mongodb.client.MongoDatabase;
+import com.mongodb.client.model.DeleteManyModel;
+import com.mongodb.client.model.InsertOneModel;
+import com.mongodb.client.model.ReplaceOneModel;
+import com.mongodb.client.model.UpdateManyModel;
+import com.mongodb.client.model.UpdateOneModel;
+import com.mongodb.client.model.WriteModel;
 
 /**
  * Unit tests for {@link DefaultBulkOperations}.
@@ -90,7 +98,7 @@ public class DefaultBulkOperationsUnitTests {
 		ops = new DefaultBulkOperations(template, "collection-1",
 				new BulkOperationContext(BulkMode.ORDERED,
 						Optional.of(mappingContext.getPersistentEntity(SomeDomainType.class)), new QueryMapper(converter),
-						new UpdateMapper(converter)));
+						new UpdateMapper(converter), null));
 	}
 
 	@Test // DATAMONGO-1518
@@ -183,6 +191,34 @@ public class DefaultBulkOperationsUnitTests {
 		assertThat(updateModel.getReplacement().getString("lastName")).isEqualTo("Kim");
 	}
 
+	@Test // DATAMONGO-2261
+	public void bulkInsertInvokesEntityCallbacks() {
+
+		BeforeConvertPersonCallback beforeConvertCallback = spy(new BeforeConvertPersonCallback());
+		BeforeSavePersonCallback beforeSaveCallback = spy(new BeforeSavePersonCallback());
+
+		ops = new DefaultBulkOperations(template, "collection-1",
+				new BulkOperationContext(BulkMode.ORDERED, Optional.of(mappingContext.getPersistentEntity(Person.class)),
+						new QueryMapper(converter), new UpdateMapper(converter),
+						EntityCallbacks.create(beforeConvertCallback, beforeSaveCallback)));
+
+		Person entity = new Person("init");
+		ops.insert(entity);
+
+		ArgumentCaptor<Person> personArgumentCaptor = ArgumentCaptor.forClass(Person.class);
+		verify(beforeConvertCallback).onBeforeConvert(personArgumentCaptor.capture(), eq("collection-1"));
+		verifyZeroInteractions(beforeSaveCallback);
+
+		ops.execute();
+
+		verify(beforeSaveCallback).onBeforeSave(personArgumentCaptor.capture(), any(), eq("collection-1"));
+		assertThat(personArgumentCaptor.getAllValues()).extracting("firstName").containsExactly("init", "before-convert");
+		verify(collection).bulkWrite(captor.capture(), any());
+
+		InsertOneModel<Document> updateModel = (InsertOneModel<Document>) captor.getValue().get(0);
+		assertThat(updateModel.getDocument()).containsEntry("firstName", "before-save");
+	}
+
 	class SomeDomainType {
 
 		@Id String id;
@@ -194,4 +230,22 @@ public class DefaultBulkOperationsUnitTests {
 	enum Gender {
 		M, F
 	}
+
+	static class BeforeConvertPersonCallback implements BeforeConvertCallback<Person> {
+
+		@Override
+		public Person onBeforeConvert(Person entity, String collection) {
+			return new Person("before-convert");
+		}
+	}
+
+	static class BeforeSavePersonCallback implements BeforeSaveCallback<Person> {
+
+		@Override
+		public Person onBeforeSave(Person entity, Document document, String collection) {
+
+			document.put("firstName", "before-save");
+			return new Person("before-save");
+		}
+	}
 }

spring-data-mongodb/src/test/java/org/springframework/data/mongodb/core/MongoTemplateUnitTests.java

@@ -22,6 +22,8 @@ import static org.springframework.data.mongodb.test.util.Assertions.*;
 import lombok.Data;
 
 import java.math.BigInteger;
+import java.util.ArrayList;
+import java.util.Arrays;
 import java.util.Collections;
 import java.util.LinkedHashMap;
 import java.util.List;
@@ -42,9 +44,11 @@ import org.mockito.ArgumentMatcher;
 import org.mockito.Mock;
 import org.mockito.Mockito;
 import org.mockito.junit.MockitoJUnitRunner;
-
 import org.springframework.beans.factory.annotation.Value;
+import org.springframework.context.ApplicationContext;
+import org.springframework.context.ApplicationListener;
 import org.springframework.context.support.GenericApplicationContext;
+import org.springframework.context.support.StaticApplicationContext;
 import org.springframework.core.convert.converter.Converter;
 import org.springframework.dao.DataAccessException;
 import org.springframework.dao.InvalidDataAccessApiUsageException;
@@ -53,6 +57,7 @@ import org.springframework.data.annotation.Version;
 import org.springframework.data.convert.CustomConversions;
 import org.springframework.data.domain.Sort;
 import org.springframework.data.geo.Point;
+import org.springframework.data.mapping.callback.EntityCallbacks;
 import org.springframework.data.mongodb.MongoDbFactory;
 import org.springframework.data.mongodb.core.aggregation.Aggregation;
 import org.springframework.data.mongodb.core.aggregation.AggregationOptions;
@@ -65,7 +70,10 @@ import org.springframework.data.mongodb.core.index.MongoPersistentEntityIndexCre
 import org.springframework.data.mongodb.core.mapping.Field;
 import org.springframework.data.mongodb.core.mapping.MongoMappingContext;
 import org.springframework.data.mongodb.core.mapping.event.AbstractMongoEventListener;
+import org.springframework.data.mongodb.core.mapping.event.BeforeConvertCallback;
 import org.springframework.data.mongodb.core.mapping.event.BeforeConvertEvent;
+import org.springframework.data.mongodb.core.mapping.event.BeforeSaveCallback;
+import org.springframework.data.mongodb.core.mapping.event.BeforeSaveEvent;
 import org.springframework.data.mongodb.core.mapreduce.GroupBy;
 import org.springframework.data.mongodb.core.mapreduce.MapReduceOptions;
 import org.springframework.data.mongodb.core.query.BasicQuery;
@@ -74,7 +82,9 @@ import org.springframework.data.mongodb.core.query.Criteria;
 import org.springframework.data.mongodb.core.query.NearQuery;
 import org.springframework.data.mongodb.core.query.Query;
 import org.springframework.data.mongodb.core.query.Update;
+import org.springframework.lang.Nullable;
 import org.springframework.test.util.ReflectionTestUtils;
+import org.springframework.util.CollectionUtils;
 
 import com.mongodb.DB;
 import com.mongodb.MongoClient;
@@ -1382,6 +1392,212 @@ public class MongoTemplateUnitTests extends MongoOperationsUnitTests {
 		verify(mapReduceIterable).collation(eq(com.mongodb.client.model.Collation.builder().locale("fr").build()));
 	}
 
+	@Test // DATAMONGO-2261
+	public void saveShouldInvokeCallbacks() {
+
+		ValueCapturingBeforeConvertCallback beforeConvertCallback = spy(new ValueCapturingBeforeConvertCallback());
+		ValueCapturingBeforeSaveCallback beforeSaveCallback = spy(new ValueCapturingBeforeSaveCallback());
+
+		template.setEntityCallbacks(EntityCallbacks.create(beforeConvertCallback, beforeSaveCallback));
+
+		Person entity = new Person();
+		entity.id = "init";
+		entity.firstname = "luke";
+
+		template.save(entity);
+
+		verify(beforeConvertCallback).onBeforeConvert(eq(entity), anyString());
+		verify(beforeSaveCallback).onBeforeSave(eq(entity), any(), anyString());
+	}
+
+	@Test // DATAMONGO-2261
+	public void insertShouldInvokeCallbacks() {
+
+		ValueCapturingBeforeConvertCallback beforeConvertCallback = spy(new ValueCapturingBeforeConvertCallback());
+		ValueCapturingBeforeSaveCallback beforeSaveCallback = spy(new ValueCapturingBeforeSaveCallback());
+
+		template.setEntityCallbacks(EntityCallbacks.create(beforeConvertCallback, beforeSaveCallback));
+
+		Person entity = new Person();
+		entity.id = "init";
+		entity.firstname = "luke";
+
+		template.insert(entity);
+
+		verify(beforeConvertCallback).onBeforeConvert(eq(entity), anyString());
+		verify(beforeSaveCallback).onBeforeSave(eq(entity), any(), anyString());
+	}
+
+	@Test // DATAMONGO-2261
+	public void insertAllShouldInvokeCallbacks() {
+
+		ValueCapturingBeforeConvertCallback beforeConvertCallback = spy(new ValueCapturingBeforeConvertCallback());
+		ValueCapturingBeforeSaveCallback beforeSaveCallback = spy(new ValueCapturingBeforeSaveCallback());
+
+		template.setEntityCallbacks(EntityCallbacks.create(beforeConvertCallback, beforeSaveCallback));
+
+		Person entity1 = new Person();
+		entity1.id = "1";
+		entity1.firstname = "luke";
+
+		Person entity2 = new Person();
+		entity1.id = "2";
+		entity1.firstname = "luke";
+
+		template.insertAll(Arrays.asList(entity1, entity2));
+
+		verify(beforeConvertCallback, times(2)).onBeforeConvert(any(), anyString());
+		verify(beforeSaveCallback, times(2)).onBeforeSave(any(), any(), anyString());
+	}
+
+	@Test // DATAMONGO-2261
+	public void findAndReplaceShouldInvokeCallbacks() {
+
+		ValueCapturingBeforeConvertCallback beforeConvertCallback = spy(new ValueCapturingBeforeConvertCallback());
+		ValueCapturingBeforeSaveCallback beforeSaveCallback = spy(new ValueCapturingBeforeSaveCallback());
+
+		template.setEntityCallbacks(EntityCallbacks.create(beforeConvertCallback, beforeSaveCallback));
+
+		Person entity = new Person();
+		entity.id = "init";
+		entity.firstname = "luke";
+
+		template.findAndReplace(new Query(), entity);
+
+		verify(beforeConvertCallback).onBeforeConvert(eq(entity), anyString());
+		verify(beforeSaveCallback).onBeforeSave(eq(entity), any(), anyString());
+	}
+
+	@Test // DATAMONGO-2261
+	public void publishesEventsAndEntityCallbacksInOrder() {
+
+		BeforeConvertCallback<Person> beforeConvertCallback = new BeforeConvertCallback<Person>() {
+
+			@Override
+			public Person onBeforeConvert(Person entity, String collection) {
+
+				assertThat(entity.id).isEqualTo("before-convert-event");
+				entity.id = "before-convert-callback";
+				return entity;
+			}
+		};
+
+		BeforeSaveCallback<Person> beforeSaveCallback = new BeforeSaveCallback<Person>() {
+
+			@Override
+			public Person onBeforeSave(Person entity, Document document, String collection) {
+
+				assertThat(entity.id).isEqualTo("before-save-event");
+				entity.id = "before-save-callback";
+				return entity;
+			}
+		};
+
+		AbstractMongoEventListener<Person> eventListener = new AbstractMongoEventListener<Person>() {
+
+			@Override
+			public void onBeforeConvert(BeforeConvertEvent<Person> event) {
+
+				assertThat(event.getSource().id).isEqualTo("init");
+				event.getSource().id = "before-convert-event";
+			}
+
+			@Override
+			public void onBeforeSave(BeforeSaveEvent<Person> event) {
+
+				assertThat(event.getSource().id).isEqualTo("before-convert-callback");
+				event.getSource().id = "before-save-event";
+			}
+		};
+
+		StaticApplicationContext ctx = new StaticApplicationContext();
+		ctx.registerBean(ApplicationListener.class, () -> eventListener);
+		ctx.registerBean(BeforeConvertCallback.class, () -> beforeConvertCallback);
+		ctx.registerBean(BeforeSaveCallback.class, () -> beforeSaveCallback);
+		ctx.refresh();
+
+		template.setApplicationContext(ctx);
+
+		Person entity = new Person();
+		entity.id = "init";
+		entity.firstname = "luke";
+
+		Person saved = template.save(entity);
+
+		assertThat(saved.id).isEqualTo("before-save-callback");
+	}
+
+	@Test // DATAMONGO-2261
+	public void beforeSaveCallbackAllowsTargetDocumentModifications() {
+
+		BeforeSaveCallback<Person> beforeSaveCallback = new BeforeSaveCallback<Person>() {
+
+			@Override
+			public Person onBeforeSave(Person entity, Document document, String collection) {
+
+				document.append("added-by", "callback");
+				return entity;
+			}
+		};
+
+		StaticApplicationContext ctx = new StaticApplicationContext();
+		ctx.registerBean(BeforeSaveCallback.class, () -> beforeSaveCallback);
+		ctx.refresh();
+
+		template.setApplicationContext(ctx);
+
+		Person entity = new Person();
+		entity.id = "luke-skywalker";
+		entity.firstname = "luke";
+
+		template.save(entity);
+
+		ArgumentCaptor<Document> captor = ArgumentCaptor.forClass(Document.class);
+
+		verify(collection).replaceOne(any(), captor.capture(), any(ReplaceOptions.class));
+		assertThat(captor.getValue()).containsEntry("added-by", "callback");
+	}
+
+	// TODO: additional tests for what is when saved.
+
+	@Test // DATAMONGO-2261
+	public void entityCallbacksAreNotSetByDefault() {
+		Assertions.assertThat(ReflectionTestUtils.getField(template, "entityCallbacks")).isNull();
+	}
+
+	@Test // DATAMONGO-2261
+	public void entityCallbacksShouldBeInitiatedOnSettingApplicationContext() {
+
+		ApplicationContext ctx = new StaticApplicationContext();
+		template.setApplicationContext(ctx);
+
+		Assertions.assertThat(ReflectionTestUtils.getField(template, "entityCallbacks")).isNotNull();
+	}
+
+	@Test // DATAMONGO-2261
+	public void setterForEntityCallbackOverridesContextInitializedOnes() {
+
+		ApplicationContext ctx = new StaticApplicationContext();
+		template.setApplicationContext(ctx);
+
+		EntityCallbacks callbacks = EntityCallbacks.create();
+		template.setEntityCallbacks(callbacks);
+
+		Assertions.assertThat(ReflectionTestUtils.getField(template, "entityCallbacks")).isSameAs(callbacks);
+	}
+
+	@Test // DATAMONGO-2261
+	public void setterForApplicationContextShouldNotOverrideAlreadySetEntityCallbacks() {
+
+		EntityCallbacks callbacks = EntityCallbacks.create();
+		ApplicationContext ctx = new StaticApplicationContext();
+
+		template.setEntityCallbacks(callbacks);
+		template.setApplicationContext(ctx);
+
+		Assertions.assertThat(ReflectionTestUtils.getField(template, "entityCallbacks")).isSameAs(callbacks);
+	}
+
 	class AutogenerateableId {
 
 		@Id BigInteger id;
@@ -1500,4 +1716,45 @@ public class MongoTemplateUnitTests extends MongoOperationsUnitTests {
 	protected MongoOperations getOperations() {
 		return this.template;
 	}
+
+	static class ValueCapturingEntityCallback<T> {
+
+		private final List<T> values = new ArrayList<>(1);
+
+		protected void capture(T value) {
+			values.add(value);
+		}
+
+		public List<T> getValues() {
+			return values;
+		}
+
+		@Nullable
+		public T getValue() {
+			return CollectionUtils.lastElement(values);
+		}
+
+	}
+
+	static class ValueCapturingBeforeConvertCallback extends ValueCapturingEntityCallback<Person>
+			implements BeforeConvertCallback<Person> {
+
+		@Override
+		public Person onBeforeConvert(Person entity, String collection) {
+
+			capture(entity);
+			return entity;
+		}
+	}
+
+	static class ValueCapturingBeforeSaveCallback extends ValueCapturingEntityCallback<Person>
+			implements BeforeSaveCallback<Person> {
+
+		@Override
+		public Person onBeforeSave(Person entity, Document document, String collection) {
+
+			capture(entity);
+			return entity;
+		}
+	}
 }

spring-data-mongodb/src/test/java/org/springframework/data/mongodb/core/ReactiveMongoTemplateUnitTests.java

@@ -25,6 +25,8 @@ import lombok.Data;
 import reactor.core.publisher.Mono;
 import reactor.test.StepVerifier;
 
+import java.util.ArrayList;
+import java.util.Arrays;
 import java.util.LinkedHashMap;
 import java.util.List;
 import java.util.Map;
@@ -41,13 +43,18 @@ import org.mockito.Mock;
 import org.mockito.junit.MockitoJUnitRunner;
 import org.reactivestreams.Publisher;
 import org.springframework.beans.factory.annotation.Value;
+import org.springframework.context.ApplicationContext;
+import org.springframework.context.support.StaticApplicationContext;
 import org.springframework.data.annotation.Id;
+import org.springframework.data.mapping.callback.ReactiveEntityCallbacks;
 import org.springframework.data.mongodb.core.MongoTemplateUnitTests.AutogenerateableId;
 import org.springframework.data.mongodb.core.aggregation.AggregationOptions;
 import org.springframework.data.mongodb.core.convert.MappingMongoConverter;
 import org.springframework.data.mongodb.core.convert.NoOpDbRefResolver;
 import org.springframework.data.mongodb.core.mapping.Field;
 import org.springframework.data.mongodb.core.mapping.MongoMappingContext;
+import org.springframework.data.mongodb.core.mapping.event.ReactiveBeforeConvertCallback;
+import org.springframework.data.mongodb.core.mapping.event.ReactiveBeforeSaveCallback;
 import org.springframework.data.mongodb.core.mapreduce.MapReduceOptions;
 import org.springframework.data.mongodb.core.query.BasicQuery;
 import org.springframework.data.mongodb.core.query.Collation;
@@ -55,7 +62,9 @@ import org.springframework.data.mongodb.core.query.Criteria;
 import org.springframework.data.mongodb.core.query.NearQuery;
 import org.springframework.data.mongodb.core.query.Query;
 import org.springframework.data.mongodb.core.query.Update;
+import org.springframework.lang.Nullable;
 import org.springframework.test.util.ReflectionTestUtils;
+import org.springframework.util.CollectionUtils;
 
 import com.mongodb.client.model.CountOptions;
 import com.mongodb.client.model.CreateCollectionOptions;
@@ -65,6 +74,7 @@ import com.mongodb.client.model.FindOneAndReplaceOptions;
 import com.mongodb.client.model.FindOneAndUpdateOptions;
 import com.mongodb.client.model.ReplaceOptions;
 import com.mongodb.client.model.UpdateOptions;
+import com.mongodb.client.result.UpdateResult;
 import com.mongodb.reactivestreams.client.AggregatePublisher;
 import com.mongodb.reactivestreams.client.DistinctPublisher;
 import com.mongodb.reactivestreams.client.FindPublisher;
@@ -72,6 +82,7 @@ import com.mongodb.reactivestreams.client.MapReducePublisher;
 import com.mongodb.reactivestreams.client.MongoClient;
 import com.mongodb.reactivestreams.client.MongoCollection;
 import com.mongodb.reactivestreams.client.MongoDatabase;
+import com.mongodb.reactivestreams.client.Success;
 
 /**
  * Unit tests for {@link ReactiveMongoTemplate}.
@@ -91,8 +102,9 @@ public class ReactiveMongoTemplateUnitTests {
 	@Mock FindPublisher findPublisher;
 	@Mock AggregatePublisher aggregatePublisher;
 	@Mock Publisher runCommandPublisher;
-	@Mock Publisher updatePublisher;
+	@Mock Publisher<UpdateResult> updateResultPublisher;
 	@Mock Publisher findAndUpdatePublisher;
+	@Mock Publisher<Success> successPublisher;
 	@Mock DistinctPublisher distinctPublisher;
 	@Mock Publisher deletePublisher;
 	@Mock MapReducePublisher mapReducePublisher;
@@ -115,8 +127,8 @@ public class ReactiveMongoTemplateUnitTests {
 		when(collection.aggregate(anyList())).thenReturn(aggregatePublisher);
 		when(collection.aggregate(anyList(), any(Class.class))).thenReturn(aggregatePublisher);
 		when(collection.count(any(), any(CountOptions.class))).thenReturn(Mono.just(0L));
-		when(collection.updateOne(any(), any(), any(UpdateOptions.class))).thenReturn(updatePublisher);
-		when(collection.updateMany(any(Bson.class), any(), any())).thenReturn(updatePublisher);
+		when(collection.updateOne(any(), any(), any(UpdateOptions.class))).thenReturn(updateResultPublisher);
+		when(collection.updateMany(any(Bson.class), any(), any())).thenReturn(updateResultPublisher);
 		when(collection.findOneAndUpdate(any(), any(), any(FindOneAndUpdateOptions.class)))
 				.thenReturn(findAndUpdatePublisher);
 		when(collection.findOneAndReplace(any(Bson.class), any(), any())).thenReturn(findPublisher);
@@ -126,6 +138,9 @@ public class ReactiveMongoTemplateUnitTests {
 		when(collection.findOneAndUpdate(any(), any(), any(FindOneAndUpdateOptions.class)))
 				.thenReturn(findAndUpdatePublisher);
 		when(collection.mapReduce(anyString(), anyString(), any())).thenReturn(mapReducePublisher);
+		when(collection.replaceOne(any(Bson.class), any(), any(ReplaceOptions.class))).thenReturn(updateResultPublisher);
+		when(collection.insertOne(any(Bson.class))).thenReturn(successPublisher);
+		when(collection.insertMany(anyList())).thenReturn(successPublisher);
 		when(findPublisher.projection(any())).thenReturn(findPublisher);
 		when(findPublisher.limit(anyInt())).thenReturn(findPublisher);
 		when(findPublisher.collation(any())).thenReturn(findPublisher);
@@ -676,6 +691,120 @@ public class ReactiveMongoTemplateUnitTests {
 				is(com.mongodb.client.model.Collation.builder().locale("fr").build()));
 	}
 
+	@Test // DATAMONGO-2261
+	public void saveShouldInvokeCallbacks() {
+
+		ValueCapturingBeforeConvertCallback beforeConvertCallback = spy(new ValueCapturingBeforeConvertCallback());
+		ValueCapturingBeforeSaveCallback beforeSaveCallback = spy(new ValueCapturingBeforeSaveCallback());
+
+		template.setEntityCallbacks(ReactiveEntityCallbacks.create(beforeConvertCallback, beforeSaveCallback));
+
+		Person entity = new Person();
+		entity.id = "init";
+		entity.firstname = "luke";
+
+		template.save(entity).subscribe();
+
+		verify(beforeConvertCallback).onBeforeConvert(eq(entity), anyString());
+		verify(beforeSaveCallback).onBeforeSave(eq(entity), any(), anyString());
+	}
+
+	@Test // DATAMONGO-2261
+	public void insertShouldInvokeCallbacks() {
+
+		ValueCapturingBeforeConvertCallback beforeConvertCallback = spy(new ValueCapturingBeforeConvertCallback());
+		ValueCapturingBeforeSaveCallback beforeSaveCallback = spy(new ValueCapturingBeforeSaveCallback());
+
+		template.setEntityCallbacks(ReactiveEntityCallbacks.create(beforeConvertCallback, beforeSaveCallback));
+
+		Person entity = new Person();
+		entity.id = "init";
+		entity.firstname = "luke";
+
+		template.insert(entity).subscribe();
+
+		verify(beforeConvertCallback).onBeforeConvert(eq(entity), anyString());
+		verify(beforeSaveCallback).onBeforeSave(eq(entity), any(), anyString());
+	}
+
+	@Test // DATAMONGO-2261
+	public void insertAllShouldInvokeCallbacks() {
+
+		ValueCapturingBeforeConvertCallback beforeConvertCallback = spy(new ValueCapturingBeforeConvertCallback());
+		ValueCapturingBeforeSaveCallback beforeSaveCallback = spy(new ValueCapturingBeforeSaveCallback());
+
+		template.setEntityCallbacks(ReactiveEntityCallbacks.create(beforeConvertCallback, beforeSaveCallback));
+
+		Person entity1 = new Person();
+		entity1.id = "1";
+		entity1.firstname = "luke";
+
+		Person entity2 = new Person();
+		entity1.id = "2";
+		entity1.firstname = "luke";
+
+		template.insertAll(Arrays.asList(entity1, entity2)).subscribe();
+
+		verify(beforeConvertCallback, times(2)).onBeforeConvert(any(), anyString());
+		verify(beforeSaveCallback, times(2)).onBeforeSave(any(), any(), anyString());
+	}
+
+	@Test // DATAMONGO-2261
+	public void findAndReplaceShouldInvokeCallbacks() {
+
+		ValueCapturingBeforeConvertCallback beforeConvertCallback = spy(new ValueCapturingBeforeConvertCallback());
+		ValueCapturingBeforeSaveCallback beforeSaveCallback = spy(new ValueCapturingBeforeSaveCallback());
+
+		template.setEntityCallbacks(ReactiveEntityCallbacks.create(beforeConvertCallback, beforeSaveCallback));
+
+		Person entity = new Person();
+		entity.id = "init";
+		entity.firstname = "luke";
+
+		template.findAndReplace(new Query(), entity).subscribe();
+
+		verify(beforeConvertCallback).onBeforeConvert(eq(entity), anyString());
+		verify(beforeSaveCallback).onBeforeSave(eq(entity), any(), anyString());
+	}
+
+	@Test // DATAMONGO-2261
+	public void entityCallbacksAreNotSetByDefault() {
+		Assertions.assertThat(ReflectionTestUtils.getField(template, "entityCallbacks")).isNull();
+	}
+
+	@Test // DATAMONGO-2261
+	public void entityCallbacksShouldBeInitiatedOnSettingApplicationContext() {
+
+		ApplicationContext ctx = new StaticApplicationContext();
+		template.setApplicationContext(ctx);
+
+		Assertions.assertThat(ReflectionTestUtils.getField(template, "entityCallbacks")).isNotNull();
+	}
+
+	@Test // DATAMONGO-2261
+	public void setterForEntityCallbackOverridesContextInitializedOnes() {
+
+		ApplicationContext ctx = new StaticApplicationContext();
+		template.setApplicationContext(ctx);
+
+		ReactiveEntityCallbacks callbacks = ReactiveEntityCallbacks.create();
+		template.setEntityCallbacks(callbacks);
+
+		Assertions.assertThat(ReflectionTestUtils.getField(template, "entityCallbacks")).isSameAs(callbacks);
+	}
+
+	@Test // DATAMONGO-2261
+	public void setterForApplicationContextShouldNotOverrideAlreadySetEntityCallbacks() {
+
+		ReactiveEntityCallbacks callbacks = ReactiveEntityCallbacks.create();
+		ApplicationContext ctx = new StaticApplicationContext();
+
+		template.setEntityCallbacks(callbacks);
+		template.setApplicationContext(ctx);
+
+		Assertions.assertThat(ReflectionTestUtils.getField(template, "entityCallbacks")).isSameAs(callbacks);
+	}
+
 	@Data
 	@org.springframework.data.mongodb.core.mapping.Document(collection = "star-wars")
 	static class Person {
@@ -714,4 +843,44 @@ public class ReactiveMongoTemplateUnitTests {
 	static class EntityWithListOfSimple {
 		List<Integer> grades;
 	}
+
+	static class ValueCapturingEntityCallback<T> {
+
+		private final List<T> values = new ArrayList<>(1);
+
+		protected void capture(T value) {
+			values.add(value);
+		}
+
+		public List<T> getValues() {
+			return values;
+		}
+
+		@Nullable
+		public T getValue() {
+			return CollectionUtils.lastElement(values);
+		}
+	}
+
+	static class ValueCapturingBeforeConvertCallback extends ValueCapturingEntityCallback<Person>
+			implements ReactiveBeforeConvertCallback<Person> {
+
+		@Override
+		public Mono<Person> onBeforeConvert(Person entity, String collection) {
+
+			capture(entity);
+			return Mono.just(entity);
+		}
+	}
+
+	static class ValueCapturingBeforeSaveCallback extends ValueCapturingEntityCallback<Person>
+			implements ReactiveBeforeSaveCallback<Person> {
+
+		@Override
+		public Mono<Person> onBeforeSave(Person entity, Document document, String collection) {
+
+			capture(entity);
+			return Mono.just(entity);
+		}
+	}
 }

src/main/asciidoc/reference/mongo-entity-callbacks.adoc

@@ -0,0 +1,30 @@
+= Store specific EntityCallbacks
+
+Spring Data MongoDB uses the `EntityCallback` API for its auditing support and reacts on the following callbacks.
+
+.Supported Entity Callbacks
+[%header,cols="4"]
+|===
+| Callback
+| Method
+| Description
+| Order
+
+| Reactive/BeforeConvertCallback
+| onBeforeConvert(T entity, String collection)
+| Invoked before a domain object is converted to `org.bson.Document`.
+| `Ordered.LOWEST_PRECEDENCE`
+
+| Reactive/AuditingEntityCallback
+| onBeforeConvert(Object entity, String collection)
+| Marks an auditable entity _created_ or _modified_
+| 100
+
+| Reactive/BeforeSaveCallback
+| onBeforeSave(T entity, org.bson.Document target, String collection)
+| Invoked before a domain object is saved. +
+ Can modify the target, to be persisted, `Document` containing all mapped entity information.
+| `Ordered.LOWEST_PRECEDENCE`
+
+|===
+

src/main/asciidoc/reference/mongodb.adoc

@@ -3159,6 +3159,11 @@ The following callback methods are present in `AbstractMappingEventListener`:
 
 NOTE: Lifecycle events are only emitted for root level types. Complex types used as properties within a document root are not subject to event publication unless they are document references annotated with `@DBRef`.
 
+WARNING: Lifecycle events depend on an `ApplicationEventMulticaster`, which in case of the `SimpleApplicationEventMulticaster` can be configured with a `TaskExecutor`, and therefore gives no guarantees when an Event is processed.
+
+include::../{spring-data-commons-docs}/entity-callbacks.adoc[leveloffset=+1]
+include::./mongo-entity-callbacks.adoc[leveloffset=+2]
+
 [[mongo.exception]]
 == Exception Translation