DATACMNS-1467 - Decouple sync and reactive bits.

Add dedicated interfaces for sync and reactive usage.
Hide default implementation and use reflective callback method lookup.
Update documentation and add initial reference documentation snippet for store specific modules.

Original Pull Request: #332

pom.xml

@@ -100,6 +100,12 @@
 			<optional>true</optional>
 		</dependency>
 
+		<dependency>
+			<groupId>io.projectreactor</groupId>
+			<artifactId>reactor-test</artifactId>
+			<scope>test</scope>
+		</dependency>
+
 		<!--  RxJava -->
 
 		<dependency>

src/main/asciidoc/entity-callbacks.adoc

@@ -0,0 +1,156 @@
+[entity-callbacks]
+= Entity Callbacks
+
+The Spring Data infrastructure provides hooks for modifying an entity before and/or after certain methods are invoked.
+Those so called ``EntityCallback``s provide a convenient way to check and potentially modify an entity in a callback fashioned style. +
+An `EntityCallback` looks pretty much like a specialized `ApplicationListener`, with which you might be familiar.
+Some Spring Data implementations publish store specific events, like a `BeforeSaveEvent` that allow to modify the given entity, which in some cases, eg. immutable types, can cause somme trouble.
+Plus, the event publishing relies on the `ApplicationEventMulticaster` which can be configured with an asynchronous `TaskExecutor` leading to unpredictable outcome.
+
+``EntityCallback``s offer integration points with both sync and reactive APIs guaranteeing an in order invocation at fixed checkpoints within the processing chain, returning a potentially modified entity or an reactive wrapper type.
+
+[NOTE]
+====
+The entity callback API has been introduced with Spring Data Commons 2.2 and is the recommended way of dealing with entity modifications. +
+Existing store specific `ApplicationEvents` are still be published *before* the potentially registered ``EntityCallback``s are called.
+====
+
+[entity-callbacks.implement]
+== Implementing Entity Callbacks
+
+The `EntityCallback` is, via its generic type argument, directly associated with the domain type it is meant for.
+Each store typically uses a set of predefined entity callbacks covering the lifecycle of an entity.
+
+.Anatomy of an `EntityCallback`
+====
+[source,java]
+----
+@FunctionalInterface
+public interface BeforeSaveCallback<T> extends EntityCallback<T> {
+
+	/**
+	 * Entity callback method invoked before a domain object is saved.
+	 * Can return either the same or a modified instance.
+	 *
+	 * @return the domain object to be persisted.
+	 */
+	T onBeforeSave(T entity <2>, String collection <3>); <1>
+}
+----
+<1> `BeforeSaveCallback` specific method to be called before an entity is saved. Returns a potentially modifed instance.
+<2> The entity right bevore presisted.
+<3> A number of store specific arguments like the _collection_ the entity is persisted to.
+====
+
+.Anatomy of a reactive `EntityCallback`
+====
+[source,java]
+----
+@FunctionalInterface
+public interface ReactiveBeforeSaveCallback<T> extends EntityCallback<T> {
+
+	/**
+	 * Entity callback method invoked on subscription, before a domain object is saved.
+	 * The returned Publisher can emit either the same or a modified instance.
+	 *
+	 * @return Publisher emitting the domain object to be persisted.
+	 */
+	Publisher<T> onBeforeSave(T entity <2>, String collection <3>); <1>
+}
+----
+<1> `BeforeSaveCallback` specific method to be called on subscription, before an entity is saved. Emits a potentially modifed instance.
+<2> The entity right bevore presisted.
+<3> A number of store specific arguments like the _collection_ the entity is persisted to.
+====
+
+Implement the interface suiting your application needs like shown in the example below.
+
+.Example `BeforeSaveCallback`
+====
+[source,java]
+----
+public class DefaultingEntityCallback implements BeforeSaveCallback<Person>, Ordered <2> {
+
+	@Override
+	public Object onBeforeSave(Person entity, String collection) { <1>
+
+		if(collection == "user) {
+		    return // ...
+		}
+
+		return // ...
+	}
+
+	@Override
+	public int getOrder() {
+		return 100; <2>
+	}
+}
+----
+<1> Implement logic according to application requirements.
+<2> Potentially order the entity callback if multiple ones for the same domain type exist. Ordering follows lowest precedence.
+====
+
+[entity-callbacks.register]
+== Registering Entity Callbacks
+
+``EntityCallback``s get picked up by the store specific implementations in case they are provided with an `ApplicationContext`.
+Most template APIs already implement `ApplicationContextAware` an therefore have a context at hand if registered as a Bean.
+
+The following example provides a collection of valid entity callback registrations.
+
+.Example `EntityCallback` Bean registration
+====
+[source,java]
+----
+@Configuration
+public class EntityCallbackConfiguration {
+
+    @Bean
+    BeforeSaveCallback<Person> annotationOrderedCallback() { <1>
+    	return new First();
+    }
+
+    @Bean
+    BeforeSaveCallback<Person> interfaceOrderedCallback() { <2>
+    	return new DefaultingEntityCallback();
+    }
+
+    @Bean
+    BeforeSaveCallback<Person> unorderedLambdaReceiverCallback() { <3>
+        return (BeforeSaveCallback<Person>) it -> // ...
+    }
+
+    @Bean
+    UserCallbacks multipleCallbacksInOneImplementationClass() { <4>
+        return new UserCallbacks();
+    }
+
+    @Order(1) <1>
+    static class First implements BeforeSaveCallback<Person> {
+
+        @Override
+    	public Person onBeforeSave(Person person) {
+    		return // ...
+    	}
+    }
+
+    static class UserCallbacks implements BeforeConvertCallback<User>, BeforeSaveCallback<User> { <4>
+
+        @Override
+        public Person onBeforeConvert(User user) {
+        	return // ...
+        }
+
+        @Override
+        public Person onBeforeSave(User user) {
+        	return // ...
+        }
+    }
+}
+----
+<1> `BeforeSaveCallback` receiving its order from the `@Order` annotation.
+<2> `BeforeSaveCallback` receiving its order via the `Ordered` interface implementation.
+<3> `BeforeSaveCallback` using a lambda expression. Unordered by default and invoked last.
+<4> Combine multiple entity callback interfaces in one implementation class.
+====

src/main/java/org/springframework/data/mapping/callback/DefaultEntityCallbacks.java

@@ -0,0 +1,138 @@
+/*
+ * Copyright 2019 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
+ *
+ *      https://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.mapping.callback;
+
+import java.lang.reflect.Method;
+import java.util.Map;
+import java.util.function.BiFunction;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.springframework.beans.factory.BeanFactory;
+import org.springframework.core.ResolvableType;
+import org.springframework.util.Assert;
+import org.springframework.util.ClassUtils;
+import org.springframework.util.ConcurrentReferenceHashMap;
+import org.springframework.util.ReflectionUtils;
+
+/**
+ * {@link EntityCallbacks} implementation using an {@link EntityCallbackDiscoverer} to retrieve {@link EntityCallback
+ * EntityCallbacks} from a {@link BeanFactory}.
+ * 
+ * @author Mark Paluch
+ * @author Christoph Strobl
+ * @since 2.2
+ */
+class DefaultEntityCallbacks implements EntityCallbacks {
+
+	private final Map<Class<?>, Method> callbackMethodCache = new ConcurrentReferenceHashMap<>(64);
+	private final SimpleEntityCallbackInvoker callbackInvoker = new SimpleEntityCallbackInvoker();
+	private final EntityCallbackDiscoverer callbackDiscoverer;
+
+	/**
+	 * Create new instance of {@link DefaultEntityCallbacks}.
+	 */
+	DefaultEntityCallbacks() {
+		this.callbackDiscoverer = new EntityCallbackDiscoverer();
+	}
+
+	/**
+	 * Create new instance of {@link DefaultEntityCallbacks} discovering {@link EntityCallback entity callbacks} within
+	 * the given {@link BeanFactory}.
+	 * 
+	 * @param beanFactory must not be {@literal null}.
+	 */
+	DefaultEntityCallbacks(BeanFactory beanFactory) {
+		this.callbackDiscoverer = new EntityCallbackDiscoverer(beanFactory);
+	}
+
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.mapping.callback.EntityCallbacks#callback(java.lang.Class, java.lang.Object, java.lang.Object)
+	 */
+	@Override
+	public <T> T callback(Class<? extends EntityCallback> callbackType, T entity, Object... args) {
+
+		Assert.notNull(entity, "Entity must not be null!");
+
+		Class<T> entityType = (Class<T>) (entity != null ? ClassUtils.getUserClass(entity.getClass())
+				: callbackDiscoverer.resolveDeclaredEntityType(callbackType).getRawClass());
+
+		Method callbackMethod = callbackMethodCache.computeIfAbsent(callbackType, it -> {
+
+			Method method = EntityCallbackDiscoverer.lookupCallbackMethod(it, entityType, args);
+			ReflectionUtils.makeAccessible(method);
+			return method;
+		});
+
+		T value = entity;
+
+		for (EntityCallback<T> callback : callbackDiscoverer.getEntityCallbacks(entityType,
+				ResolvableType.forClass(callbackType))) {
+
+			BiFunction<EntityCallback<T>, T, Object> callbackFunction = EntityCallbackDiscoverer
+					.computeCallbackInvokerFunction(callback, callbackMethod, args);
+			value = callbackInvoker.invokeCallback(callback, value, callbackFunction);
+		}
+
+		return value;
+	}
+
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.mapping.callback.EntityCallbacks#addEntityCallback(org.springframework.data.mapping.callback.EntityCallback)
+	 */
+	@Override
+	public void addEntityCallback(EntityCallback<?> callback) {
+		this.callbackDiscoverer.addEntityCallback(callback);
+	}
+
+	class SimpleEntityCallbackInvoker implements org.springframework.data.mapping.callback.EntityCallbackInvoker {
+
+		@Override
+		public <T> T invokeCallback(EntityCallback<T> callback, T entity,
+				BiFunction<EntityCallback<T>, T, Object> callbackInvokerFunction) {
+
+			try {
+
+				Object value = callbackInvokerFunction.apply(callback, entity);
+
+				if (value != null) {
+					return (T) value;
+				}
+
+				throw new IllegalArgumentException(
+						String.format("Callback invocation on %s returned null value for %s", callback.getClass(), entity));
+
+			} catch (ClassCastException ex) {
+
+				String msg = ex.getMessage();
+				if (msg == null || EntityCallbackInvoker.matchesClassCastMessage(msg, entity.getClass())) {
+
+					// Possibly a lambda-defined listener which we could not resolve the generic event type for
+					// -> let's suppress the exception and just log a debug message.
+					Log logger = LogFactory.getLog(getClass());
+					if (logger.isDebugEnabled()) {
+						logger.debug("Non-matching callback type for entity callback: " + callback, ex);
+					}
+					return entity;
+				} else {
+					throw ex;
+				}
+			}
+		}
+	}
+}

src/main/java/org/springframework/data/mapping/callback/DefaultReactiveEntityCallbacks.java

@@ -0,0 +1,141 @@
+/*
+ * Copyright 2019 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
+ *
+ *      https://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.mapping.callback;
+
+import reactor.core.publisher.Mono;
+
+import java.lang.reflect.Method;
+import java.util.Map;
+import java.util.function.BiFunction;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.reactivestreams.Publisher;
+import org.springframework.beans.factory.BeanFactory;
+import org.springframework.core.ResolvableType;
+import org.springframework.util.Assert;
+import org.springframework.util.ClassUtils;
+import org.springframework.util.ConcurrentReferenceHashMap;
+import org.springframework.util.ReflectionUtils;
+
+/**
+ * {@link ReactiveEntityCallbacks} implementation.
+ *
+ * @author Mark Paluch
+ * @author Christoph Strobl
+ */
+class DefaultReactiveEntityCallbacks implements ReactiveEntityCallbacks {
+
+	private final Map<Class<?>, Method> callbackMethodCache = new ConcurrentReferenceHashMap<>(64);
+	private final ReactiveEntityCallbackInvoker callbackInvoker = new DefaultReactiveEntityCallbackInvoker();
+	private final EntityCallbackDiscoverer callbackDiscoverer;
+
+	/**
+	 * Create new instance of {@link DefaultReactiveEntityCallbacks}.
+	 */
+	DefaultReactiveEntityCallbacks() {
+		this.callbackDiscoverer = new EntityCallbackDiscoverer();
+	}
+
+	/**
+	 * Create new instance of {@link DefaultReactiveEntityCallbacks} discovering {@link EntityCallback entity callbacks}
+	 * within the given {@link BeanFactory}.
+	 *
+	 * @param beanFactory must not be {@literal null}.
+	 */
+	DefaultReactiveEntityCallbacks(BeanFactory beanFactory) {
+		this.callbackDiscoverer = new EntityCallbackDiscoverer(beanFactory);
+	}
+
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.mapping.callback.ReactiveEntityCallbacks#callback(java.lang.Class, java.lang.Object, java.lang.Object)
+	 */
+	@Override
+	public <T> Mono<T> callback(Class<? extends EntityCallback> callbackType, T entity, Object... args) {
+
+		Assert.notNull(entity, "Entity must not be null!");
+
+		Class<T> entityType = (Class<T>) (entity != null ? ClassUtils.getUserClass(entity.getClass())
+				: callbackDiscoverer.resolveDeclaredEntityType(callbackType).getRawClass());
+
+		Method callbackMethod = callbackMethodCache.computeIfAbsent(callbackType, it -> {
+
+			Method method = EntityCallbackDiscoverer.lookupCallbackMethod(it, entityType, args);
+			ReflectionUtils.makeAccessible(method);
+			return method;
+		});
+
+		Mono<T> deferredCallbackChain = Mono.just(entity);
+
+		for (EntityCallback<T> callback : callbackDiscoverer.getEntityCallbacks(entityType,
+				ResolvableType.forClass(callbackType))) {
+
+			BiFunction<EntityCallback<T>, T, Object> callbackFunction = EntityCallbackDiscoverer
+					.computeCallbackInvokerFunction(callback, callbackMethod, args);
+
+			deferredCallbackChain = deferredCallbackChain
+					.flatMap(it -> callbackInvoker.invokeCallback(callback, it, callbackFunction));
+		}
+
+		return deferredCallbackChain;
+	}
+
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.mapping.callback.EntityCallbacks#addEntityCallback(org.springframework.data.mapping.callback.EntityCallback)
+	 */
+	@Override
+	public void addEntityCallback(EntityCallback<?> callback) {
+		this.callbackDiscoverer.addEntityCallback(callback);
+	}
+
+	static class DefaultReactiveEntityCallbackInvoker implements ReactiveEntityCallbackInvoker {
+
+		@Override
+		public <T> Mono<T> invokeCallback(EntityCallback<T> callback, T entity,
+				BiFunction<EntityCallback<T>, T, Object> callbackInvokerFunction) {
+
+			try {
+
+				Object value = callbackInvokerFunction.apply(callback, entity);
+
+				if (value != null) {
+					return value instanceof Publisher ? Mono.from((Publisher<T>) value) : Mono.just((T) value);
+				}
+
+				throw new IllegalArgumentException(
+						String.format("Callback invocation on %s returned null value for %s", callback.getClass(), entity));
+			} catch (ClassCastException ex) {
+
+				String msg = ex.getMessage();
+				if (msg == null || EntityCallbackInvoker.matchesClassCastMessage(msg, entity.getClass())) {
+
+					// Possibly a lambda-defined listener which we could not resolve the generic event type for
+					// -> let's suppress the exception and just log a debug message.
+					Log logger = LogFactory.getLog(getClass());
+					if (logger.isDebugEnabled()) {
+						logger.debug("Non-matching callback type for entity callback: " + callback, ex);
+					}
+
+					return Mono.just(entity);
+				} else {
+					return Mono.error(ex);
+				}
+			}
+		}
+	}
+}

src/main/java/org/springframework/data/mapping/callback/EntityCallback.java

@@ -16,13 +16,37 @@
 package org.springframework.data.mapping.callback;
 
 /**
- * Marker interface for entity callbacks to be implemented in specific callback subtypes. Multiple entity callbacks are
- * invoked sequentially with the result of the previous callback.
+ * Marker interface for entity callbacks to be implemented in specific callback subtypes intended for internal usage
+ * within store specific implementations. <br />
+ * <p />
+ * Multiple entity callbacks are invoked sequentially with the result of the previous callback. Those callbacks do by
+ * default not follow an explicit order of invocation. It is strongly recommended to enforce ordering for callbacks of
+ * the same by implementing {@link org.springframework.core.Ordered} or following the annotation driven approach using
+ * {@link org.springframework.core.annotation.Order}.
+ * <p />
+ * Entity callbacks are invoked after publishing {@link org.springframework.context.ApplicationEvent events}.
+ * <p />
+ * A store specific {@link EntityCallback} needs to define a callback method accepting an object of the parameterized
+ * type as its first argument followed by additional <i>optional</i> arguments.
+ * 
+ * <pre>
+ * <code>
+ *
+ * public interface BeforeSaveCallback&lt;T&gt; extends EntityCallback&lt;T&gt; {
+ *
+ *     T onBeforeSave(T entity, String collection);
+ * }
+ * </code>
+ * </pre>
+ *
+ * The
  *
  * @author Mark Paluch
- * @param <T> Entity type.
- * @see BeforeSaveCallback
+ * @author Christoph Strobl
+ * @param <T> Entity type. Used to detect {@link EntityCallback callbacks} to invoke via their generic type signature.
  * @see org.springframework.core.Ordered
  * @see org.springframework.core.annotation.Order
  */
-public interface EntityCallback<T> {}
+public interface EntityCallback<T> {
+
+}

src/main/java/org/springframework/data/mapping/callback/EntityCallbackDiscoverer.java

@@ -15,6 +15,8 @@
  */
 package org.springframework.data.mapping.callback;
 
+import java.lang.reflect.Method;
+import java.lang.reflect.Modifier;
 import java.util.ArrayList;
 import java.util.Collection;
 import java.util.LinkedHashSet;
@@ -22,12 +24,12 @@ import java.util.List;
 import java.util.Map;
 import java.util.Set;
 import java.util.concurrent.ConcurrentHashMap;
+import java.util.function.BiFunction;
 
 import org.springframework.aop.framework.AopProxyUtils;
 import org.springframework.beans.factory.BeanFactory;
 import org.springframework.beans.factory.NoSuchBeanDefinitionException;
 import org.springframework.beans.factory.config.ConfigurableBeanFactory;
-import org.springframework.context.ApplicationContext;
 import org.springframework.core.ResolvableType;
 import org.springframework.core.annotation.AnnotationAwareOrderComparator;
 import org.springframework.lang.Nullable;
@@ -35,73 +37,45 @@ import org.springframework.util.Assert;
 import org.springframework.util.ClassUtils;
 import org.springframework.util.ConcurrentReferenceHashMap;
 import org.springframework.util.ObjectUtils;
+import org.springframework.util.ReflectionUtils;
 import org.springframework.util.comparator.Comparators;
 
 /**
- * Abstract implementation of the for classes that wish to implement {@link EntityCallback} functionality.
- *
  * @author Mark Paluch
- * @see EntityCallback
- * @see AnnotationAwareOrderComparator
+ * @author Christoph Strobl
+ * @since 2.2
  */
-public abstract class AbstractEntityCallbacks {
-
-	@Nullable private ClassLoader beanClassLoader;
-	@Nullable private BeanFactory beanFactory;
+class EntityCallbackDiscoverer {
 
 	private final CallbackRetriever defaultRetriever = new CallbackRetriever(false);
 	private final Map<CallbackCacheKey, CallbackRetriever> retrieverCache = new ConcurrentHashMap<>(64);
 	private final Map<Class<?>, ResolvableType> entityTypeCache = new ConcurrentReferenceHashMap<>(64);
 
+	@Nullable private ClassLoader beanClassLoader;
+	@Nullable private BeanFactory beanFactory;
+
 	private Object retrievalMutex = this.defaultRetriever;
 
 	/**
 	 * Create a new {@link EntityCallback} instance.
 	 */
-	public AbstractEntityCallbacks() {}
+	EntityCallbackDiscoverer() {}
 
 	/**
-	 * Create a new {@link EntityCallback} instance given {@link ApplicationContext}. Preloads {@link EntityCallback}
-	 * beans by scanning the {@link BeanFactory}.
+	 * Create a new {@link EntityCallback} instance.
+	 * <p />
+	 * Pre loads {@link EntityCallback} beans by scanning the {@link BeanFactory}.
 	 */
-	public AbstractEntityCallbacks(ApplicationContext beanFactory) {
+	EntityCallbackDiscoverer(BeanFactory beanFactory) {
 		setBeanFactory(beanFactory);
 	}
 
-	public void setBeanClassLoader(ClassLoader classLoader) {
-		this.beanClassLoader = classLoader;
-	}
+	void addEntityCallback(EntityCallback<?> callback) {
 
-	/**
-	 * Set the {@link BeanFactory} and optionally {@link #setBeanClassLoader(ClassLoader) class loader} if not set.
-	 * Preloads {@link EntityCallback} beans by scanning the {@link BeanFactory}.
-	 *
-	 * @param beanFactory must not be {@literal null}.
-	 */
-	public void setBeanFactory(BeanFactory beanFactory) {
-		this.beanFactory = beanFactory;
-		if (beanFactory instanceof ConfigurableBeanFactory) {
-			ConfigurableBeanFactory cbf = (ConfigurableBeanFactory) beanFactory;
-			if (this.beanClassLoader == null) {
-				this.beanClassLoader = cbf.getBeanClassLoader();
-			}
-			this.retrievalMutex = cbf.getSingletonMutex();
-		}
+		Assert.notNull(callback, "Callback must not be null!");
 
-		defaultRetriever.discoverEntityCallbacks(this.beanFactory);
-		this.retrieverCache.clear();
-	}
-
-	private BeanFactory getBeanFactory() {
-		if (this.beanFactory == null) {
-			throw new IllegalStateException(
-					"EntityCallbacks cannot retrieve callback beans " + "because it is not associated with a BeanFactory");
-		}
-		return this.beanFactory;
-	}
-
-	public void addEntityCallback(EntityCallback<?> callback) {
 		synchronized (this.retrievalMutex) {
+
 			// Explicitly remove target for a proxy, if registered already,
 			// in order to avoid double invocations of the same callback.
 			Object singletonTarget = AopProxyUtils.getSingletonTarget(callback);
@@ -113,28 +87,32 @@ public abstract class AbstractEntityCallbacks {
 		}
 	}
 
-	public void addEntityCallbackBean(String callbackBeanName) {
+	void addEntityCallbackBean(String callbackBeanName) {
+
 		synchronized (this.retrievalMutex) {
 			this.defaultRetriever.entityCallbackBeans.add(callbackBeanName);
 			this.retrieverCache.clear();
 		}
 	}
 
-	public void removeEntityCallback(EntityCallback<?> callback) {
+	void removeEntityCallback(EntityCallback<?> callback) {
+
 		synchronized (this.retrievalMutex) {
 			this.defaultRetriever.entityCallbacks.remove(callback);
 			this.retrieverCache.clear();
 		}
 	}
 
-	public void removeEntityCallbackBean(String callbackBeanName) {
+	void removeEntityCallbackBean(String callbackBeanName) {
+
 		synchronized (this.retrievalMutex) {
 			this.defaultRetriever.entityCallbackBeans.remove(callbackBeanName);
 			this.retrieverCache.clear();
 		}
 	}
 
-	public void removeAllCallbacks() {
+	void clear() {
+
 		synchronized (this.retrievalMutex) {
 			this.defaultRetriever.entityCallbacks.clear();
 			this.defaultRetriever.entityCallbackBeans.clear();
@@ -142,18 +120,6 @@ public abstract class AbstractEntityCallbacks {
 		}
 	}
 
-	/**
-	 * Return a {@link Collection} of all {@link EntityCallback}s.
-	 *
-	 * @return a {@link Collection} of all {@link EntityCallback}s.
-	 * @see EntityCallback
-	 */
-	protected Collection<EntityCallback<?>> getEntityCallbacks() {
-		synchronized (this.retrievalMutex) {
-			return this.defaultRetriever.getEntityCallbacks();
-		}
-	}
-
 	/**
 	 * Return a {@link Collection} of all {@link EntityCallback}s matching the given entity type. Non-matching callbacks
 	 * get excluded early.
@@ -164,15 +130,15 @@ public abstract class AbstractEntityCallbacks {
 	 * @return a {@link Collection} of {@link EntityCallback}s.
 	 * @see EntityCallback
 	 */
-	protected Collection<EntityCallback<?>> getEntityCallbacks(Object entity, ResolvableType callbackType) {
+	<T extends S, S> Collection<EntityCallback<S>> getEntityCallbacks(Class<T> entity, ResolvableType callbackType) {
 
-		Class<?> sourceType = entity.getClass();
+		Class<?> sourceType = entity;
 		CallbackCacheKey cacheKey = new CallbackCacheKey(callbackType, sourceType);
 
 		// Quick check for existing entry on ConcurrentHashMap...
 		CallbackRetriever retriever = this.retrieverCache.get(cacheKey);
 		if (retriever != null) {
-			return retriever.getEntityCallbacks();
+			return (Collection<EntityCallback<S>>) (Collection) retriever.getEntityCallbacks();
 		}
 
 		if (this.beanClassLoader == null || (ClassUtils.isCacheSafe(entity.getClass(), this.beanClassLoader)
@@ -182,20 +148,33 @@ public abstract class AbstractEntityCallbacks {
 			synchronized (this.retrievalMutex) {
 				retriever = this.retrieverCache.get(cacheKey);
 				if (retriever != null) {
-					return retriever.getEntityCallbacks();
+					return (Collection<EntityCallback<S>>) (Collection) retriever.getEntityCallbacks();
 				}
 				retriever = new CallbackRetriever(true);
 				Collection<EntityCallback<?>> callbacks = retrieveEntityCallbacks(ResolvableType.forClass(sourceType),
 						callbackType, retriever);
 				this.retrieverCache.put(cacheKey, retriever);
-				return callbacks;
+				return (Collection<EntityCallback<S>>) (Collection) callbacks;
 			}
 		} else {
 			// No CallbackRetriever caching -> no synchronization necessary
-			return retrieveEntityCallbacks(callbackType, callbackType, null);
+			return (Collection<EntityCallback<S>>) (Collection) retrieveEntityCallbacks(callbackType, callbackType, null);
 		}
 	}
 
+	@Nullable
+	ResolvableType resolveDeclaredEntityType(Class<?> callbackType) {
+
+		ResolvableType eventType = entityTypeCache.get(callbackType);
+
+		if (eventType == null) {
+			eventType = ResolvableType.forClass(callbackType).as(EntityCallback.class).getGeneric();
+			entityTypeCache.put(callbackType, eventType);
+		}
+
+		return (eventType != ResolvableType.NONE ? eventType : null);
+	}
+
 	/**
 	 * Actually retrieve the callbacks for the given entity and callback type.
 	 *
@@ -226,7 +205,7 @@ public abstract class AbstractEntityCallbacks {
 		}
 
 		if (!callbackBeans.isEmpty()) {
-			BeanFactory beanFactory = getBeanFactory();
+			BeanFactory beanFactory = getRequiredBeanFactory();
 			for (String callbackBeanName : callbackBeans) {
 				try {
 					Class<?> callbackImplType = beanFactory.getType(callbackBeanName);
@@ -286,21 +265,136 @@ public abstract class AbstractEntityCallbacks {
 	 * @return whether the given callback should be included in the candidates for the given callback type.
 	 */
 	protected boolean supportsEvent(EntityCallback<?> callback, ResolvableType entityType, ResolvableType callbackType) {
+
 		return supportsEvent(callback.getClass(), entityType)
 				&& callbackType.isAssignableFrom(ResolvableType.forInstance(callback));
 	}
 
-	@Nullable
-	private ResolvableType resolveDeclaredEntityType(Class<?> callbackType) {
+	/**
+	 * (non-Javadoc)
+	 * 
+	 * @see org.springframework.beans.factory.BeanClassLoaderAware
+	 */
+	public void setBeanClassLoader(ClassLoader classLoader) {
+		this.beanClassLoader = classLoader;
+	}
 
-		ResolvableType eventType = entityTypeCache.get(callbackType);
+	/**
+	 * Set the {@link BeanFactory} and optionally {@link #setBeanClassLoader(ClassLoader) class loader} if not set. Pre
+	 * loads {@link EntityCallback} beans by scanning the {@link BeanFactory}.
+	 *
+	 * @param beanFactory must not be {@literal null}.
+	 * @see org.springframework.beans.factory.BeanFactoryAware#setBeanFactory(BeanFactory)
+	 */
+	public void setBeanFactory(BeanFactory beanFactory) {
 
-		if (eventType == null) {
-			eventType = ResolvableType.forClass(callbackType).as(EntityCallback.class).getGeneric();
-			entityTypeCache.put(callbackType, eventType);
+		this.beanFactory = beanFactory;
+
+		if (beanFactory instanceof ConfigurableBeanFactory) {
+			ConfigurableBeanFactory cbf = (ConfigurableBeanFactory) beanFactory;
+			if (this.beanClassLoader == null) {
+				this.beanClassLoader = cbf.getBeanClassLoader();
+			}
+			this.retrievalMutex = cbf.getSingletonMutex();
 		}
 
-		return (eventType != ResolvableType.NONE ? eventType : null);
+		defaultRetriever.discoverEntityCallbacks(this.beanFactory);
+		this.retrieverCache.clear();
+	}
+
+	@Nullable
+	static Method lookupCallbackMethod(Class<?> callbackType, Class<?> entityType, Object[] args) {
+
+		Collection<Method> methods = new ArrayList<>(1);
+
+		ReflectionUtils.doWithMethods(callbackType, mc -> methods.add(mc), method -> {
+
+			if (!Modifier.isPublic(method.getModifiers()) || method.getParameterCount() != args.length + 1
+					|| method.isBridge() || ReflectionUtils.isObjectMethod(method)) {
+				return false;
+			}
+
+			return ClassUtils.isAssignable(method.getParameterTypes()[0], entityType);
+		});
+
+		if (methods.size() == 1) {
+			return methods.iterator().next();
+		}
+
+		throw new IllegalStateException(
+				String.format("%s does not define a callback method accepting %s and %s additional arguments.",
+						ClassUtils.getShortName(callbackType), ClassUtils.getShortName(entityType), args.length));
+	}
+
+	static <T> BiFunction<EntityCallback<T>, T, Object> computeCallbackInvokerFunction(EntityCallback<T> callback,
+			Method callbackMethod, Object[] args) {
+
+		return (entityCallback, entity) -> {
+
+			Object[] invocationArgs = new Object[args.length + 1];
+			invocationArgs[0] = entity;
+			if (args.length > 0) {
+				System.arraycopy(args, 0, invocationArgs, 1, args.length);
+			}
+
+			return ReflectionUtils.invokeMethod(callbackMethod, callback, invocationArgs);
+		};
+	}
+
+	private BeanFactory getRequiredBeanFactory() {
+
+		Assert.state(beanFactory != null,
+				"EntityCallbacks cannot retrieve callback beans because it is not associated with a BeanFactory");
+		return beanFactory;
+	}
+
+	/**
+	 * Helper class that encapsulates a specific set of target {@link EntityCallback callbacks}, allowing for efficient
+	 * retrieval of pre-filtered callbacks.
+	 */
+	class CallbackRetriever {
+
+		private final Set<EntityCallback<?>> entityCallbacks = new LinkedHashSet<>();
+
+		private final Set<String> entityCallbackBeans = new LinkedHashSet<>();
+
+		private final boolean preFiltered;
+
+		CallbackRetriever(boolean preFiltered) {
+			this.preFiltered = preFiltered;
+		}
+
+		Collection<EntityCallback<?>> getEntityCallbacks() {
+
+			List<EntityCallback<?>> allCallbacks = new ArrayList<>(
+					this.entityCallbacks.size() + this.entityCallbackBeans.size());
+			allCallbacks.addAll(this.entityCallbacks);
+
+			if (!this.entityCallbackBeans.isEmpty()) {
+				BeanFactory beanFactory = getRequiredBeanFactory();
+				for (String callbackBeanName : this.entityCallbackBeans) {
+					try {
+						EntityCallback<?> callback = beanFactory.getBean(callbackBeanName, EntityCallback.class);
+						if (this.preFiltered || !allCallbacks.contains(callback)) {
+							allCallbacks.add(callback);
+						}
+					} catch (NoSuchBeanDefinitionException ex) {
+						// Singleton callback instance (without backing bean definition) disappeared -
+						// probably in the middle of the destruction phase
+					}
+				}
+			}
+
+			if (!this.preFiltered || !this.entityCallbackBeans.isEmpty()) {
+				AnnotationAwareOrderComparator.sort(allCallbacks);
+			}
+
+			return allCallbacks;
+		}
+
+		void discoverEntityCallbacks(BeanFactory beanFactory) {
+			beanFactory.getBeanProvider(EntityCallback.class).stream().forEach(entityCallbacks::add);
+		}
 	}
 
 	/**
@@ -348,52 +442,4 @@ public abstract class AbstractEntityCallbacks {
 		}
 	}
 
-	/**
-	 * Helper class that encapsulates a specific set of target {@link EntityCallback callbacks}, allowing for efficient
-	 * retrieval of pre-filtered callbacks.
-	 */
-	class CallbackRetriever {
-
-		private final Set<EntityCallback<?>> entityCallbacks = new LinkedHashSet<>();
-
-		private final Set<String> entityCallbackBeans = new LinkedHashSet<>();
-
-		private final boolean preFiltered;
-
-		CallbackRetriever(boolean preFiltered) {
-			this.preFiltered = preFiltered;
-		}
-
-		Collection<EntityCallback<?>> getEntityCallbacks() {
-
-			List<EntityCallback<?>> allCallbacks = new ArrayList<>(
-					this.entityCallbacks.size() + this.entityCallbackBeans.size());
-			allCallbacks.addAll(this.entityCallbacks);
-
-			if (!this.entityCallbackBeans.isEmpty()) {
-				BeanFactory beanFactory = getBeanFactory();
-				for (String callbackBeanName : this.entityCallbackBeans) {
-					try {
-						EntityCallback<?> callback = beanFactory.getBean(callbackBeanName, EntityCallback.class);
-						if (this.preFiltered || !allCallbacks.contains(callback)) {
-							allCallbacks.add(callback);
-						}
-					} catch (NoSuchBeanDefinitionException ex) {
-						// Singleton callback instance (without backing bean definition) disappeared -
-						// probably in the middle of the destruction phase
-					}
-				}
-			}
-
-			if (!this.preFiltered || !this.entityCallbackBeans.isEmpty()) {
-				AnnotationAwareOrderComparator.sort(allCallbacks);
-			}
-
-			return allCallbacks;
-		}
-
-		void discoverEntityCallbacks(BeanFactory beanFactory) {
-			beanFactory.getBeanProvider(EntityCallback.class).stream().forEach(entityCallbacks::add);
-		}
-	}
 }

src/main/java/org/springframework/data/mapping/callback/EntityCallbackInvoker.java

@@ -0,0 +1,59 @@
+/*
+ * Copyright 2019 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
+ *
+ *      https://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.mapping.callback;
+
+import java.util.function.BiFunction;
+
+/**
+ * @author Christoph Strobl
+ * @since 2.2
+ */
+interface EntityCallbackInvoker {
+
+	/**
+	 * Invoke the actual {@link EntityCallback} for the given entity via the {@link BiFunction invoker function}.
+	 *
+	 * @param callback must not be {@literal null}.
+	 * @param entity must not be {@literal null}
+	 * @param callbackInvokerFunction must not be {@literal null}.
+	 * @param <T>
+	 * @return never {@literal null}.
+	 */
+	<T> Object invokeCallback(EntityCallback<T> callback, T entity,
+			BiFunction<EntityCallback<T>, T, Object> callbackInvokerFunction);
+
+	static boolean matchesClassCastMessage(String classCastMessage, Class<?> eventClass) {
+
+		// On Java 8, the message starts with the class name: "java.lang.String cannot be cast..."
+		if (classCastMessage.startsWith(eventClass.getName())) {
+			return true;
+		}
+
+		// On Java 11, the message starts with "class ..." a.k.a. Class.toString()
+		if (classCastMessage.startsWith(eventClass.toString())) {
+			return true;
+		}
+
+		// On Java 9, the message used to contain the module name: "java.base/java.lang.String cannot be cast..."
+		int moduleSeparatorIndex = classCastMessage.indexOf('/');
+		if (moduleSeparatorIndex != -1 && classCastMessage.startsWith(eventClass.getName(), moduleSeparatorIndex + 1)) {
+			return true;
+		}
+
+		// Assuming an unrelated class cast failure...
+		return false;
+	}
+}

src/main/java/org/springframework/data/mapping/callback/EntityCallbacks.java

@@ -0,0 +1,83 @@
+/*
+ * Copyright 2019 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
+ *
+ *      https://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.mapping.callback;
+
+import org.springframework.beans.factory.BeanFactory;
+import org.springframework.util.Assert;
+
+/**
+ * @author Christoph Strobl
+ * @since 2.2
+ */
+public interface EntityCallbacks {
+
+	/**
+	 * Add the given {@link EntityCallback callback} using generic type argument detection for identification of supported
+	 * types.
+	 *
+	 * @param callback must not be {@literal null}.
+	 * @throws IllegalArgumentException if the required argument is {@literal null}.
+	 */
+	void addEntityCallback(EntityCallback<?> callback);
+
+	/**
+	 * Invoke matching {@link EntityCallback entity callbacks} with given arguments.
+	 *
+	 * @param callbackType must not be {@literal null}.
+	 * @param entity must not be {@literal null}.
+	 * @param args optional arguments.
+	 * @param <T>
+	 * @return never {@literal null}.
+	 * @throws IllegalArgumentException if a required argument is {@literal null}.
+	 */
+	<T> T callback(Class<? extends EntityCallback> callbackType, T entity, Object... args);
+
+	/**
+	 * Create a new {@link EntityCallbacks} instance with given {@link EntityCallback callbacks}. <br />
+	 * The provided {@link EntityCallback callbacks} are immediately {@link #addEntityCallback(EntityCallback) added}.
+	 */
+	static EntityCallbacks create(EntityCallback<?>... callbacks) {
+
+		EntityCallbacks entityCallbacks = create();
+		for (EntityCallback<?> callback : callbacks) {
+			entityCallbacks.addEntityCallback(callback);
+		}
+		return entityCallbacks;
+	}
+
+	/**
+	 * Obtain a new {@link EntityCallbacks} instance. <br />
+	 * Use {@link #addEntityCallback(EntityCallback)} to register callbacks manually.
+	 */
+	static EntityCallbacks create() {
+		return new DefaultEntityCallbacks();
+	}
+
+	/**
+	 * Obtain a new {@link EntityCallbacks} instance.
+	 * <p />
+	 * {@link EntityCallback callbacks} are pre loaded from the given {@link BeanFactory}. <br />
+	 * Use {@link #addEntityCallback(EntityCallback)} to register additional callbacks manually.
+	 *
+	 * @param beanFactory must not be {@literal null}.
+	 * @throws IllegalArgumentException if a required argument is {@literal null}.
+	 */
+	static EntityCallbacks create(BeanFactory beanFactory) {
+
+		Assert.notNull(beanFactory, "Context must not be null!");
+		return new DefaultEntityCallbacks(beanFactory);
+	}
+}

src/main/java/org/springframework/data/mapping/callback/ReactiveBeforeSaveCallback.java

@@ -1,39 +0,0 @@
-/*
- * Copyright 2019 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
- *
- *      https://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.mapping.callback;
-
-import org.reactivestreams.Publisher;
-
-/**
- * Entity callback before saving the entity using deferred, reactive execution. This interface allows for modifications
- * to the actual entity that are required prior to saving the entity.
- *
- * @author Mark Paluch
- * @see EntityCallback
- * @see ReactiveEntityCallbacks
- */
-@FunctionalInterface
-public interface ReactiveBeforeSaveCallback<T> extends EntityCallback<T> {
-
-	/**
-	 * Handle the entity callback. Modifications that result in creating a new instance should return the changed entity
-	 * as method return value.
-	 *
-	 * @param object the entity for this callback.
-	 * @return {@link Publisher} emitting the resulting entity.
-	 */
-	Publisher<T> onBeforeSave(T object);
-}

src/main/java/org/springframework/data/mapping/callback/ReactiveEntityCallbackInvoker.java

@@ -15,23 +15,28 @@
  */
 package org.springframework.data.mapping.callback;
 
+import reactor.core.publisher.Mono;
+
+import java.util.function.BiFunction;
+
+import org.springframework.lang.NonNull;
+import org.springframework.lang.Nullable;
+
 /**
- * Entity callback before saving the entity. This interface allows for modifications to the actual entity that are
- * required prior to saving the entity.
- *
- * @author Mark Paluch
- * @see EntityCallback
- * @see SimpleEntityCallbacks
+ * @author Christoph Strobl
+ * @since 2.2
  */
-@FunctionalInterface
-public interface BeforeSaveCallback<T> extends EntityCallback<T> {
+interface ReactiveEntityCallbackInvoker extends EntityCallbackInvoker {
 
 	/**
-	 * Handle the entity callback. Modifications that result in creating a new instance should return the changed entity
-	 * as method return value.
-	 *
-	 * @param object the entity for this callback.
-	 * @return the resulting entity.
+	 * @param callback must not be {@literal null}.
+	 * @param entity can be {@literal null}
+	 * @param callbackInvokerFunction must not be {@literal null}.
+	 * @param <T>
+	 * @return a {@link Mono} emitting the result of the invocation.
 	 */
-	T onBeforeSave(T object);
+	@NonNull
+	@Override
+	<T> Mono<T> invokeCallback(EntityCallback<T> callback, @Nullable T entity,
+			BiFunction<EntityCallback<T>, T, Object> callbackInvokerFunction);
 }

src/main/java/org/springframework/data/mapping/callback/ReactiveEntityCallbacks.java

@@ -17,60 +17,70 @@ package org.springframework.data.mapping.callback;
 
 import reactor.core.publisher.Mono;
 
-import java.util.function.BiFunction;
-
-import org.reactivestreams.Publisher;
-import org.springframework.context.ApplicationContext;
-import org.springframework.core.ResolvableType;
+import org.springframework.beans.factory.BeanFactory;
 import org.springframework.util.Assert;
 
 /**
- * Reactive {@link SimpleEntityCallbacks} implementation.
- *
- * @author Mark Paluch
+ * @author Christoph Strobl
+ * @since 2.2
  */
-public class ReactiveEntityCallbacks extends SimpleEntityCallbacks {
+public interface ReactiveEntityCallbacks {
 
-	public ReactiveEntityCallbacks() {
-		super();
-	}
+	/**
+	 * Add the given {@link EntityCallback callback} using generic type argument detection for identification of supported
+	 * types.
+	 *
+	 * @param callback must not be {@literal null}.
+	 * @throws IllegalArgumentException if the required argument is {@literal null}.
+	 */
+	void addEntityCallback(EntityCallback<?> callback);
 
-	public ReactiveEntityCallbacks(ApplicationContext beanFactory) {
-		super(beanFactory);
+	/**
+	 * On {@link Mono#subscribe() subscribe} invoke the matching {@link EntityCallback entity callbacks} with given
+	 * arguments.
+	 *
+	 * @param callbackType must not be {@literal null}.
+	 * @param entity must not be {@literal null}.
+	 * @param args optional arguments.
+	 * @param <T>
+	 * @return a {@link Mono} emitting the result after invoking the callbacks.
+	 * @throws IllegalArgumentException if a required argument is {@literal null}.
+	 */
+	<T> Mono<T> callback(Class<? extends EntityCallback> callbackType, T entity, Object... args);
+
+	/**
+	 * Create a new {@link ReactiveEntityCallbacks} instance with given {@link EntityCallback callbacks}. <br />
+	 * The provided {@link EntityCallback callbacks} are immediately {@link #addEntityCallback(EntityCallback) added}.
+	 */
+	static ReactiveEntityCallbacks create(EntityCallback<?>... callbacks) {
+
+		ReactiveEntityCallbacks entityCallbacks = create();
+		for (EntityCallback<?> callback : callbacks) {
+			entityCallbacks.addEntityCallback(callback);
+		}
+		return entityCallbacks;
 	}
 
 	/**
-	 * Perform an entity callback.
-	 *
-	 * @param entity the entity, must not be {@literal null}.
-	 * @param callbackType desired callback type.
-	 * @param callbackInvoker invocation function for the callback to optionally pass additional parameters.
-	 * @return the resulting entity after invoking all callbacks.
+	 * Obtain a new {@link ReactiveEntityCallbacks} instance. <br />
+	 * Use {@link #addEntityCallback(EntityCallback)} to register callbacks manually.
 	 */
-	@SuppressWarnings("unchecked")
-	public <T, E extends EntityCallback<T>> Mono<T> callbackLater(T entity, Class<? extends E> callbackType,
-			BiFunction<? extends E, T, Publisher<? extends Object>> callbackInvoker) {
+	static ReactiveEntityCallbacks create() {
+		return new DefaultReactiveEntityCallbacks();
+	}
 
-		Assert.notNull(entity, "Entity must not be null!");
+	/**
+	 * Obtain a new {@link ReactiveEntityCallbacks} instance.
+	 * <p />
+	 * {@link EntityCallback callbacks} are pre loaded from the given {@link BeanFactory}. <br />
+	 * Use {@link #addEntityCallback(EntityCallback)} to register additional callbacks manually.
+	 *
+	 * @param beanFactory must not be {@literal null}.
+	 * @throws IllegalArgumentException if a required argument is {@literal null}.
+	 */
+	static ReactiveEntityCallbacks create(BeanFactory beanFactory) {
 
-		ResolvableType resolvedCallbackType = ResolvableType.forClass(callbackType);
-		Mono<T> deferredCallbackChain = Mono.just(entity);
-
-		for (EntityCallback<?> callback : getEntityCallbacks(entity, resolvedCallbackType)) {
-
-			deferredCallbackChain = deferredCallbackChain.flatMap(it -> {
-
-				Object o = invokeCallback(callback, it, (BiFunction) callbackInvoker);
-
-				if (o instanceof Publisher) {
-					return Mono.from((Publisher<T>) o);
-				}
-
-				throw new IllegalStateException("Callback " + callback + " returned a non-Publisher type " + o);
-
-			});
-		}
-
-		return deferredCallbackChain;
+		Assert.notNull(beanFactory, "Context must not be null!");
+		return new DefaultReactiveEntityCallbacks(beanFactory);
 	}
 }

src/main/java/org/springframework/data/mapping/callback/SimpleEntityCallbacks.java

@@ -1,159 +0,0 @@
-/*
- * Copyright 2019 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
- *
- *      https://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.mapping.callback;
-
-import java.util.function.BiFunction;
-
-import org.apache.commons.logging.Log;
-import org.apache.commons.logging.LogFactory;
-import org.springframework.context.ApplicationContext;
-import org.springframework.core.ResolvableType;
-import org.springframework.lang.Nullable;
-import org.springframework.util.Assert;
-import org.springframework.util.ErrorHandler;
-
-/**
- * Simple implementation to invoke {@link EntityCallback}s.
- *
- * @author Mark Paluch
- */
-public class SimpleEntityCallbacks extends AbstractEntityCallbacks {
-
-	@Nullable private ErrorHandler errorHandler;
-
-	public SimpleEntityCallbacks() {
-		super();
-	}
-
-	public SimpleEntityCallbacks(ApplicationContext beanFactory) {
-		super(beanFactory);
-	}
-
-	/**
-	 * Set the {@link ErrorHandler} to invoke in case an exception is thrown from a {@link EntityCallback}.
-	 * <p>
-	 * Default is none, with a callback {@link Exception} stopping the current dispatch and getting propagated to the
-	 * publisher of the current event.
-	 * <p>
-	 * Consider setting an {@link ErrorHandler} implementation that catches and logs exceptions (a la
-	 * {@link org.springframework.scheduling.support.TaskUtils#LOG_AND_SUPPRESS_ERROR_HANDLER}) or an implementation that
-	 * logs exceptions while nevertheless propagating them (e.g.
-	 * {@link org.springframework.scheduling.support.TaskUtils#LOG_AND_PROPAGATE_ERROR_HANDLER}).
-	 */
-	public void setErrorHandler(@Nullable ErrorHandler errorHandler) {
-		this.errorHandler = errorHandler;
-	}
-
-	/**
-	 * Return the current {@link ErrorHandler} for this callback dispatcher.
-	 *
-	 * @return the current {@link ErrorHandler}.
-	 */
-	@Nullable
-	protected ErrorHandler getErrorHandler() {
-		return this.errorHandler;
-	}
-
-	/**
-	 * Perform an entity callback.
-	 *
-	 * @param entity the entity, must not be {@literal null}.
-	 * @param callbackType desired callback type.
-	 * @param callbackInvoker invocation function for the callback to optionally pass additional parameters.
-	 * @return the resulting entity after invoking all callbacks.
-	 */
-	@SuppressWarnings("unchecked")
-	public <T, E extends EntityCallback<T>> T callback(T entity, Class<? extends E> callbackType,
-			BiFunction<? extends E, T, Object> callbackInvoker) {
-
-		Assert.notNull(entity, "Entity must not be null!");
-
-		ResolvableType resolvedCallbackType = ResolvableType.forClass(callbackType);
-		T entityToUse = entity;
-
-		for (EntityCallback<?> callback : getEntityCallbacks(entity, resolvedCallbackType)) {
-			entityToUse = (T) invokeCallback(callback, entityToUse, (BiFunction) callbackInvoker);
-		}
-
-		return entityToUse;
-	}
-
-	/**
-	 * Invoke the given callback with the given entity.
-	 */
-	protected Object invokeCallback(EntityCallback<?> callback, Object entity,
-			BiFunction<EntityCallback<?>, Object, Object> callbackInvokerFunction) {
-
-		ErrorHandler errorHandler = getErrorHandler();
-
-		if (errorHandler != null) {
-			try {
-				return doInvokeCallback(callback, entity, callbackInvokerFunction);
-			} catch (Throwable err) {
-				errorHandler.handleError(err);
-				return entity;
-			}
-		}
-
-		return doInvokeCallback(callback, entity, callbackInvokerFunction);
-
-	}
-
-	@SuppressWarnings({ "rawtypes" })
-	private Object doInvokeCallback(EntityCallback<?> callback, Object entity,
-			BiFunction<EntityCallback<?>, Object, Object> callbackInvokerFunction) {
-
-		try {
-			return callbackInvokerFunction.apply(callback, entity);
-		} catch (ClassCastException ex) {
-			String msg = ex.getMessage();
-			if (msg == null || matchesClassCastMessage(msg, entity.getClass())) {
-				// Possibly a lambda-defined listener which we could not resolve the generic event type for
-				// -> let's suppress the exception and just log a debug message.
-				Log logger = LogFactory.getLog(getClass());
-				if (logger.isDebugEnabled()) {
-					logger.debug("Non-matching callback type for entity callback: " + callback, ex);
-				}
-
-				return entity;
-			} else {
-				throw ex;
-			}
-		}
-	}
-
-	private boolean matchesClassCastMessage(String classCastMessage, Class<?> eventClass) {
-
-		// On Java 8, the message starts with the class name: "java.lang.String cannot be cast..."
-		if (classCastMessage.startsWith(eventClass.getName())) {
-			return true;
-		}
-
-		// On Java 11, the message starts with "class ..." a.k.a. Class.toString()
-		if (classCastMessage.startsWith(eventClass.toString())) {
-			return true;
-		}
-
-		// On Java 9, the message used to contain the module name: "java.base/java.lang.String cannot be cast..."
-		int moduleSeparatorIndex = classCastMessage.indexOf('/');
-		if (moduleSeparatorIndex != -1 && classCastMessage.startsWith(eventClass.getName(), moduleSeparatorIndex + 1)) {
-			return true;
-		}
-
-		// Assuming an unrelated class cast failure...
-		return false;
-	}
-}

src/main/java/org/springframework/data/mapping/callback/package-info.java

@@ -0,0 +1,5 @@
+/**
+ * Mapping callback API and implementation base classes.
+ */
+@org.springframework.lang.NonNullApi
+package org.springframework.data.mapping.callback;

src/test/java/org/springframework/data/mapping/callback/CapturingEntityCallback.java

@@ -0,0 +1,86 @@
+/*
+ * Copyright 2019 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
+ *
+ *      https://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.mapping.callback;
+
+import java.util.ArrayList;
+import java.util.List;
+
+import org.springframework.core.Ordered;
+import org.springframework.data.mapping.Person;
+import org.springframework.data.mapping.PersonDocument;
+import org.springframework.lang.Nullable;
+import org.springframework.util.CollectionUtils;
+
+/**
+ * @author Christoph Strobl
+ */
+class CapturingEntityCallback implements EntityCallback<Person> {
+
+	final List<Person> captured = new ArrayList<>(3);
+	final @Nullable Person returnValue;
+
+	CapturingEntityCallback() {
+		this(new PersonDocument(null, null, null));
+	}
+
+	CapturingEntityCallback(@Nullable Person returnValue) {
+		this.returnValue = returnValue;
+	}
+
+	public Person doSomething(Person person) {
+
+		captured.add(person);
+		return returnValue;
+	}
+
+	Person capturedValue() {
+		return CollectionUtils.lastElement(captured);
+	}
+
+	List<Person> capturedValues() {
+		return captured;
+	}
+
+	static class FirstCallback extends CapturingEntityCallback implements Ordered {
+
+		@Override
+		public int getOrder() {
+			return 1;
+		}
+	}
+
+	static class SecondCallback extends CapturingEntityCallback implements Ordered {
+
+		public SecondCallback() {}
+
+		public SecondCallback(Person returnValue) {
+			super(returnValue);
+		}
+
+		@Override
+		public int getOrder() {
+			return 2;
+		}
+	}
+
+	static class ThirdCallback extends CapturingEntityCallback implements Ordered {
+
+		@Override
+		public int getOrder() {
+			return 3;
+		}
+	}
+}

src/test/java/org/springframework/data/mapping/callback/DefaultEntityCallbacksUnitTests.java

@@ -0,0 +1,291 @@
+/*
+ * Copyright 2019 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
+ *
+ *      https://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.mapping.callback;
+
+import static org.assertj.core.api.Assertions.*;
+
+import java.util.ArrayList;
+import java.util.List;
+
+import org.junit.Test;
+import org.springframework.context.annotation.AnnotationConfigApplicationContext;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.context.annotation.Lazy;
+import org.springframework.core.Ordered;
+import org.springframework.data.mapping.Person;
+import org.springframework.data.mapping.PersonDocument;
+import org.springframework.data.mapping.callback.CapturingEntityCallback.FirstCallback;
+import org.springframework.data.mapping.callback.CapturingEntityCallback.SecondCallback;
+import org.springframework.data.mapping.callback.CapturingEntityCallback.ThirdCallback;
+
+/**
+ * Unit tests for {@link DefaultEntityCallbacks}.
+ *
+ * @author Mark Paluch
+ * @author Christoph Strobl
+ */
+public class DefaultEntityCallbacksUnitTests {
+
+	@Test // DATACMNS-1467
+	public void shouldDispatchCallback() {
+
+		AnnotationConfigApplicationContext ctx = new AnnotationConfigApplicationContext(MyConfig.class);
+
+		DefaultEntityCallbacks callbacks = new DefaultEntityCallbacks(ctx);
+
+		PersonDocument personDocument = new PersonDocument(null, "Walter", null);
+		PersonDocument afterCallback = callbacks.callback(BeforeSaveCallback.class, personDocument);
+
+		assertThat(afterCallback.getSsn()).isEqualTo(6);
+	}
+
+	@Test // DATACMNS-1467
+	public void shouldDispatchCallsToLambdaReceivers() {
+
+		AnnotationConfigApplicationContext ctx = new AnnotationConfigApplicationContext(LambdaConfig.class);
+
+		DefaultEntityCallbacks callbacks = new DefaultEntityCallbacks(ctx);
+
+		PersonDocument personDocument = new PersonDocument(null, "Walter", null);
+		PersonDocument afterCallback = callbacks.callback(BeforeSaveCallback.class, personDocument);
+
+		assertThat(afterCallback).isSameAs(personDocument);
+	}
+
+	@Test // DATACMNS-1467
+	public void invokeGenericEvent() {
+
+		DefaultEntityCallbacks callbacks = new DefaultEntityCallbacks();
+		callbacks.addEntityCallback(new GenericPersonCallback());
+
+		Person afterCallback = callbacks.callback(GenericPersonCallback.class, new PersonDocument(null, "Walter", null));
+
+		assertThat(afterCallback.getSsn()).isEqualTo(6);
+	}
+
+	@Test // DATACMNS-1467
+	public void invokeGenericEventWithArgs() {
+
+		DefaultEntityCallbacks callbacks = new DefaultEntityCallbacks();
+		callbacks.addEntityCallback(new GenericPersonCallbackWithArgs());
+
+		Person afterCallback = callbacks.callback(GenericPersonCallbackWithArgs.class,
+				new PersonDocument(null, "Walter", null), "agr0", Float.POSITIVE_INFINITY);
+
+		assertThat(afterCallback.getSsn()).isEqualTo(6);
+	}
+
+	@Test // DATACMNS-1467
+	public void invokeInvalidEvent() {
+
+		DefaultEntityCallbacks callbacks = new DefaultEntityCallbacks();
+		callbacks.addEntityCallback(new InvalidEntityCallback() {});
+
+		assertThatExceptionOfType(IllegalStateException.class)
+				.isThrownBy(() -> callbacks.callback(InvalidEntityCallback.class, new PersonDocument(null, "Walter", null),
+						"agr0", Float.POSITIVE_INFINITY));
+	}
+
+	@Test // DATACMNS-1467
+	public void passesInvocationResultOnAlongTheChain() {
+
+		CapturingEntityCallback first = new FirstCallback();
+		CapturingEntityCallback second = new SecondCallback();
+
+		DefaultEntityCallbacks callbacks = new DefaultEntityCallbacks();
+		callbacks.addEntityCallback(first);
+		callbacks.addEntityCallback(second);
+
+		PersonDocument initial = new PersonDocument(null, "Walter", null);
+
+		callbacks.callback(CapturingEntityCallback.class, initial);
+
+		assertThat(first.capturedValue()).isSameAs(initial);
+		assertThat(first.capturedValues()).hasSize(1);
+		assertThat(second.capturedValue()).isNotSameAs(initial);
+		assertThat(second.capturedValues()).hasSize(1);
+	}
+
+	@Test // DATACMNS-1467
+	public void errorsOnNullEntity() {
+
+		DefaultEntityCallbacks callbacks = new DefaultEntityCallbacks();
+		callbacks.addEntityCallback(new CapturingEntityCallback());
+
+		assertThatExceptionOfType(IllegalArgumentException.class)
+				.isThrownBy(() -> callbacks.callback(CapturingEntityCallback.class, null));
+	}
+
+	@Test // DATACMNS-1467
+	public void errorsOnNullValueReturnedByCallbackEntity() {
+
+		CapturingEntityCallback first = new FirstCallback();
+		CapturingEntityCallback second = new SecondCallback(null);
+		CapturingEntityCallback third = new ThirdCallback();
+
+		DefaultEntityCallbacks callbacks = new DefaultEntityCallbacks();
+		callbacks.addEntityCallback(first);
+		callbacks.addEntityCallback(second);
+		callbacks.addEntityCallback(third);
+
+		PersonDocument initial = new PersonDocument(null, "Walter", null);
+
+		assertThatExceptionOfType(IllegalArgumentException.class)
+				.isThrownBy(() -> callbacks.callback(CapturingEntityCallback.class, initial));
+
+		assertThat(first.capturedValue()).isSameAs(initial);
+		assertThat(second.capturedValue()).isNotNull().isNotSameAs(initial);
+		assertThat(third.capturedValues()).isEmpty();
+	}
+
+	@Test // DATACMNS-1467
+	public void detectsMultipleCallbacksWithinOneClass() {
+
+		AnnotationConfigApplicationContext ctx = new AnnotationConfigApplicationContext(MultipleCallbacksInOneClassConfig.class);
+
+		DefaultEntityCallbacks callbacks = new DefaultEntityCallbacks(ctx);
+
+		PersonDocument personDocument = new PersonDocument(null, "Walter", null);
+		callbacks.callback(BeforeSaveCallback.class, personDocument);
+
+		assertThat(ctx.getBean("callbacks", MultipleCallbacks.class).invocations).containsExactly("save");
+
+		callbacks.callback(BeforeConvertCallback.class, personDocument);
+
+		assertThat(ctx.getBean("callbacks", MultipleCallbacks.class).invocations).containsExactly("save", "convert");
+	}
+
+	@Configuration
+	static class MyConfig {
+
+		@Bean
+		MyBeforeSaveCallback callback() {
+			return new MyBeforeSaveCallback();
+		}
+
+		@Bean
+		@Lazy
+		Object namedCallback() {
+			return new MyOtherCallback();
+		}
+	}
+
+	@Configuration
+	static class LambdaConfig {
+
+		@Bean
+		BeforeSaveCallback<User> userCallback() {
+			return object -> object;
+		}
+
+		@Bean
+		BeforeSaveCallback<Person> personCallback() {
+
+			return object -> {
+				object.setSsn(object.getFirstName().length());
+				return object;
+			};
+		}
+	}
+
+	@Configuration
+	static class MultipleCallbacksInOneClassConfig {
+
+		@Bean
+		MultipleCallbacks callbacks() {
+			return new MultipleCallbacks();
+		}
+	}
+
+	interface BeforeConvertCallback<T> extends EntityCallback<T>, Ordered {
+		T onBeforeConvert(T object);
+
+		@Override
+		default int getOrder() {
+			return 0;
+		}
+	}
+
+	interface BeforeSaveCallback<T> extends EntityCallback<T> {
+		T onBeforeSave(T object);
+	}
+
+	static class MyBeforeSaveCallback implements BeforeSaveCallback<Person> {
+
+		@Override
+		public Person onBeforeSave(Person object) {
+
+			object.setSsn(object.getFirstName().length());
+			return object;
+		}
+	}
+
+	static class MyOtherCallback implements BeforeSaveCallback<Person> {
+
+		@Override
+		public Person onBeforeSave(Person object) {
+			return object;
+		}
+	}
+
+	static class User {}
+
+	static class GenericPersonCallback implements EntityCallback<Person> {
+
+		public Person onBeforeSave(Person value) {
+
+			value.setSsn(value.getFirstName().length());
+			return value;
+		}
+	}
+
+	static class GenericPersonCallbackWithArgs implements EntityCallback<Person> {
+
+		public Person onBeforeSave(Person value, String agr1, Object arg2) {
+
+			value.setSsn(value.getFirstName().length());
+			return value;
+		}
+	}
+
+	interface InvalidEntityCallback extends EntityCallback<Person> {
+
+		default Person onBeforeSave(String value, Person entity) {
+			return entity;
+		}
+	}
+
+	static class MultipleCallbacks implements BeforeConvertCallback<Person>, BeforeSaveCallback<Person> {
+
+		List<String> invocations = new ArrayList(2);
+
+		@Override
+		public Person onBeforeConvert(Person object) {
+
+			invocations.add("convert");
+			return object;
+		}
+
+		@Override
+		public Person onBeforeSave(Person object) {
+
+			invocations.add("save");
+			return object;
+		}
+	}
+
+}

src/test/java/org/springframework/data/mapping/callback/DefaultReactiveEntityCallbacksUnitTests.java

@@ -0,0 +1,161 @@
+/*
+ * Copyright 2019 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
+ *
+ *      https://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.mapping.callback;
+
+import static org.assertj.core.api.Assertions.*;
+
+import reactor.core.publisher.Mono;
+import reactor.test.StepVerifier;
+
+import org.junit.Test;
+import org.springframework.context.annotation.AnnotationConfigApplicationContext;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.data.mapping.Person;
+import org.springframework.data.mapping.PersonDocument;
+import org.springframework.data.mapping.callback.CapturingEntityCallback.FirstCallback;
+import org.springframework.data.mapping.callback.CapturingEntityCallback.SecondCallback;
+import org.springframework.data.mapping.callback.CapturingEntityCallback.ThirdCallback;
+
+/**
+ * Unit tests for {@link DefaultReactiveEntityCallbacks}.
+ *
+ * @author Mark Paluch
+ * @author Christoph Strobl
+ */
+public class DefaultReactiveEntityCallbacksUnitTests {
+
+	@Test // DATACMNS-1467
+	public void dispatchResolvesOnSubscribe() {
+
+		AnnotationConfigApplicationContext ctx = new AnnotationConfigApplicationContext(MyConfig.class);
+
+		DefaultReactiveEntityCallbacks callbacks = new DefaultReactiveEntityCallbacks(ctx);
+
+		PersonDocument personDocument = new PersonDocument(null, "Walter", null);
+		Mono<PersonDocument> afterCallback = callbacks.callback(ReactiveBeforeSaveCallback.class, personDocument);
+
+		assertThat(personDocument.getSsn()).isNull();
+
+		afterCallback.as(StepVerifier::create) //
+				.consumeNextWith(it -> assertThat(it.getSsn()).isEqualTo(6)) //
+				.verifyComplete();
+	}
+
+	@Test // DATACMNS-1467
+	public void invokeGenericEvent() {
+
+		DefaultReactiveEntityCallbacks callbacks = new DefaultReactiveEntityCallbacks();
+		callbacks.addEntityCallback(new GenericPersonCallback());
+
+		callbacks.callback(GenericPersonCallback.class, new PersonDocument(null, "Walter", null)) //
+				.as(StepVerifier::create) //
+				.consumeNextWith(it -> assertThat(it.getSsn()).isEqualTo(6)) //
+				.verifyComplete();
+	}
+
+	@Test // DATACMNS-1467
+	public void passesInvocationResultOnAlongTheChain() {
+
+		CapturingEntityCallback first = new FirstCallback();
+		CapturingEntityCallback second = new SecondCallback();
+
+		DefaultReactiveEntityCallbacks callbacks = new DefaultReactiveEntityCallbacks();
+		callbacks.addEntityCallback(first);
+		callbacks.addEntityCallback(second);
+
+		PersonDocument initial = new PersonDocument(null, "Walter", null);
+
+		callbacks.callback(CapturingEntityCallback.class, initial) //
+				.as(StepVerifier::create) //
+				.expectNextCount(1) //
+				.verifyComplete();
+
+		assertThat(first.capturedValue()).isSameAs(initial);
+		assertThat(first.capturedValues()).hasSize(1);
+		assertThat(second.capturedValue()).isNotSameAs(initial);
+		assertThat(second.capturedValues()).hasSize(1);
+	}
+
+	@Test // DATACMNS-1467
+	public void errorsOnNullEntity() {
+
+		DefaultReactiveEntityCallbacks callbacks = new DefaultReactiveEntityCallbacks();
+		callbacks.addEntityCallback(new CapturingEntityCallback());
+
+		assertThatExceptionOfType(IllegalArgumentException.class)
+				.isThrownBy(() -> callbacks.callback(CapturingEntityCallback.class, null));
+	}
+
+	@Test // DATACMNS-1467
+	public void errorsOnNullValueReturnedByCallbackEntity() {
+
+		CapturingEntityCallback first = new FirstCallback();
+		CapturingEntityCallback second = new SecondCallback(null);
+		CapturingEntityCallback third = new ThirdCallback();
+
+		DefaultReactiveEntityCallbacks callbacks = new DefaultReactiveEntityCallbacks();
+		callbacks.addEntityCallback(first);
+		callbacks.addEntityCallback(second);
+		callbacks.addEntityCallback(third);
+
+		PersonDocument initial = new PersonDocument(null, "Walter", null);
+
+		callbacks.callback(CapturingEntityCallback.class, initial) //
+				.as(StepVerifier::create) //
+				.expectError(IllegalArgumentException.class) //
+				.verify();
+
+		assertThat(first.capturedValue()).isSameAs(initial);
+		assertThat(second.capturedValue()).isNotNull().isNotSameAs(initial);
+		assertThat(third.capturedValues()).isEmpty();
+	}
+
+	@Configuration
+	static class MyConfig {
+
+		@Bean
+		MyReactiveBeforeSaveCallback callback() {
+			return new MyReactiveBeforeSaveCallback();
+		}
+
+	}
+
+	interface ReactiveBeforeSaveCallback<T> extends EntityCallback<T> {
+		Mono<T> onBeforeSave(T object);
+	}
+
+	static class MyReactiveBeforeSaveCallback implements ReactiveBeforeSaveCallback<Person> {
+
+		@Override
+		public Mono<Person> onBeforeSave(Person object) {
+
+			PersonDocument result = new PersonDocument(object.getFirstName().length(), object.getFirstName(),
+					object.getLastName());
+
+			return Mono.just(result);
+		}
+	}
+
+	static class GenericPersonCallback implements EntityCallback<Person> {
+
+		public Person onBeforeSave(Person value) {
+
+			value.setSsn(value.getFirstName().length());
+			return value;
+		}
+	}
+}

src/test/java/org/springframework/data/mapping/callback/EntityCallbackDiscovererUnitTests.java

@@ -0,0 +1,207 @@
+/*
+ * Copyright 2019 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
+ *
+ *      https://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.mapping.callback;
+
+import static org.assertj.core.api.Assertions.*;
+
+import java.util.Collection;
+
+import org.junit.Test;
+import org.springframework.context.annotation.AnnotationConfigApplicationContext;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.context.annotation.Lazy;
+import org.springframework.core.Ordered;
+import org.springframework.core.ResolvableType;
+import org.springframework.core.annotation.Order;
+import org.springframework.data.mapping.Child;
+import org.springframework.data.mapping.Person;
+import org.springframework.data.mapping.PersonDocument;
+
+/**
+ * @author Christoph Strobl
+ */
+public class EntityCallbackDiscovererUnitTests {
+
+	@Test // DATACMNS-1467
+	public void shouldDiscoverCallbackType() {
+
+		AnnotationConfigApplicationContext ctx = new AnnotationConfigApplicationContext(MyConfig.class);
+
+		EntityCallbackDiscoverer discoverer = new EntityCallbackDiscoverer(ctx);
+
+		Collection<EntityCallback<Person>> entityCallbacks = discoverer.getEntityCallbacks(PersonDocument.class,
+				ResolvableType.forType(BeforeSaveCallback.class));
+
+		assertThat(entityCallbacks).hasSize(1).element(0).isInstanceOf(MyBeforeSaveCallback.class);
+	}
+
+	@Test // DATACMNS-1467
+	public void shouldDiscoverCallbackTypeByName() {
+
+		AnnotationConfigApplicationContext ctx = new AnnotationConfigApplicationContext(MyConfig.class);
+
+		EntityCallbackDiscoverer discoverer = new EntityCallbackDiscoverer(ctx);
+		discoverer.clear();
+		discoverer.addEntityCallbackBean("namedCallback");
+
+		Collection<EntityCallback<Person>> entityCallbacks = discoverer.getEntityCallbacks(PersonDocument.class,
+				ResolvableType.forType(BeforeSaveCallback.class));
+
+		assertThat(entityCallbacks).hasSize(1).element(0).isInstanceOf(MyOtherCallback.class);
+	}
+
+	@Test // DATACMNS-1467
+	public void shouldSupportCallbackTypes() {
+
+		EntityCallbackDiscoverer discoverer = new EntityCallbackDiscoverer();
+
+		assertThat(discoverer.supportsEvent(MyBeforeSaveCallback.class, ResolvableType.forClass(Person.class))).isTrue();
+		assertThat(discoverer.supportsEvent(MyBeforeSaveCallback.class, ResolvableType.forClass(Child.class))).isTrue();
+		assertThat(discoverer.supportsEvent(BeforeSaveCallback.class, ResolvableType.forClass(PersonDocument.class)))
+				.isTrue();
+
+		assertThat(discoverer.supportsEvent(MyBeforeSaveCallback.class, ResolvableType.forClass(Object.class))).isFalse();
+		assertThat(discoverer.supportsEvent(MyBeforeSaveCallback.class, ResolvableType.forClass(User.class))).isFalse();
+	}
+
+	@Test // DATACMNS-1467
+	public void shouldSupportInstanceCallbackTypes() {
+
+		EntityCallbackDiscoverer discoverer = new EntityCallbackDiscoverer();
+
+		MyBeforeSaveCallback callback = new MyBeforeSaveCallback();
+
+		assertThat(discoverer.supportsEvent(callback, ResolvableType.forClass(Person.class),
+				ResolvableType.forClass(BeforeSaveCallback.class))).isTrue();
+		assertThat(discoverer.supportsEvent(callback, ResolvableType.forClass(Child.class),
+				ResolvableType.forClass(BeforeSaveCallback.class))).isTrue();
+		assertThat(discoverer.supportsEvent(callback, ResolvableType.forClass(PersonDocument.class),
+				ResolvableType.forClass(BeforeSaveCallback.class))).isTrue();
+		assertThat(discoverer.supportsEvent(callback, ResolvableType.forClass(PersonDocument.class),
+				ResolvableType.forClass(BeforeSaveCallback.class))).isTrue();
+
+		assertThat(discoverer.supportsEvent(callback, ResolvableType.forClass(User.class),
+				ResolvableType.forClass(BeforeSaveCallback.class))).isFalse();
+	}
+
+	@Test // DATACMNS-1467
+	public void shouldDispatchInOrder() {
+
+		AnnotationConfigApplicationContext ctx = new AnnotationConfigApplicationContext(OrderedConfig.class);
+
+		EntityCallbackDiscoverer discoverer = new EntityCallbackDiscoverer(ctx);
+
+		Collection<EntityCallback<Person>> entityCallbacks = discoverer.getEntityCallbacks(PersonDocument.class,
+				ResolvableType.forType(EntityCallback.class));
+
+		assertThat(entityCallbacks).containsExactly(ctx.getBean("callback1", EntityCallback.class),
+				ctx.getBean("callback2", EntityCallback.class), ctx.getBean("callback3", EntityCallback.class),
+				ctx.getBean("callback4", EntityCallback.class));
+	}
+
+	@Configuration
+	static class MyConfig {
+
+		@Bean
+		MyBeforeSaveCallback callback() {
+			return new MyBeforeSaveCallback();
+		}
+
+		@Bean
+		@Lazy
+		Object namedCallback() {
+			return new MyOtherCallback();
+		}
+	}
+
+	@Configuration
+	static class OrderedConfig {
+
+		@Bean
+		EntityCallback<Person> callback4() {
+
+			return (BeforeSaveCallback<Person>) object -> object;
+		}
+
+		@Bean
+		EntityCallback<Person> callback2() {
+			return new Second();
+		}
+
+		@Bean
+		EntityCallback<Person> callback3() {
+			return new Third();
+		}
+
+		@Bean
+		EntityCallback<Person> callback1() {
+			return new First();
+		}
+
+		@Order(3)
+		static class Third implements EntityCallback<Person> {
+
+			public Person beforeSave(Person object) {
+				return object;
+			}
+		}
+
+		static class Second implements EntityCallback<Person>, Ordered {
+
+			public Person beforeSave(Person object) {
+				return object;
+			}
+
+			@Override
+			public int getOrder() {
+				return 2;
+			}
+		}
+
+		@Order(1)
+		static class First implements EntityCallback<Person> {
+
+			public Person beforeSave(Person object) {
+				return object;
+			}
+		}
+	}
+
+	interface BeforeSaveCallback<T> extends EntityCallback<T> {
+		T onBeforeSave(T object);
+	}
+
+	static class MyBeforeSaveCallback implements BeforeSaveCallback<Person> {
+
+		@Override
+		public Person onBeforeSave(Person object) {
+
+			object.setSsn(object.getFirstName().length());
+			return object;
+		}
+	}
+
+	static class MyOtherCallback implements BeforeSaveCallback<Person> {
+
+		@Override
+		public Person onBeforeSave(Person object) {
+			return object;
+		}
+	}
+
+	static class User {}
+}

src/test/java/org/springframework/data/mapping/callback/ReactiveEntityCallbacksUnitTests.java

@@ -1,72 +0,0 @@
-/*
- * Copyright 2019 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
- *
- *      https://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.mapping.callback;
-
-import static org.assertj.core.api.Assertions.*;
-
-import reactor.core.publisher.Mono;
-
-import org.junit.Test;
-import org.springframework.context.annotation.AnnotationConfigApplicationContext;
-import org.springframework.context.annotation.Bean;
-import org.springframework.context.annotation.Configuration;
-import org.springframework.data.mapping.Person;
-import org.springframework.data.mapping.PersonDocument;
-
-/**
- * Unit tests for {@link ReactiveEntityCallbacks}.
- *
- * @author Mark Paluch
- */
-public class ReactiveEntityCallbacksUnitTests {
-
-	@Test
-	public void shouldDispatchCallback() {
-
-		AnnotationConfigApplicationContext ctx = new AnnotationConfigApplicationContext(MyConfig.class);
-
-		ReactiveEntityCallbacks callbacks = new ReactiveEntityCallbacks(ctx);
-
-		PersonDocument personDocument = new PersonDocument(null, "Walter", null);
-		Mono<PersonDocument> afterCallback = callbacks.callbackLater(personDocument, ReactiveBeforeSaveCallback.class,
-				ReactiveBeforeSaveCallback::onBeforeSave);
-
-		assertThat(personDocument.getSsn()).isNull();
-		assertThat(afterCallback.block().getSsn()).isEqualTo(6);
-	}
-
-	@Configuration
-	static class MyConfig {
-
-		@Bean
-		MyReactiveBeforeSaveCallback callback() {
-			return new MyReactiveBeforeSaveCallback();
-		}
-
-	}
-
-	static class MyReactiveBeforeSaveCallback implements ReactiveBeforeSaveCallback<Person> {
-
-		@Override
-		public Mono<Person> onBeforeSave(Person object) {
-
-			PersonDocument result = new PersonDocument(object.getFirstName().length(), object.getFirstName(),
-					object.getLastName());
-
-			return Mono.just(result);
-		}
-	}
-}

src/test/java/org/springframework/data/mapping/callback/SimpleEntityCallbacksUnitTests.java

@@ -1,179 +0,0 @@
-/*
- * Copyright 2019 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
- *
- *      https://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.mapping.callback;
-
-import static org.assertj.core.api.Assertions.*;
-
-import java.util.Collection;
-
-import org.junit.Test;
-import org.springframework.context.annotation.AnnotationConfigApplicationContext;
-import org.springframework.context.annotation.Bean;
-import org.springframework.context.annotation.Configuration;
-import org.springframework.context.annotation.Lazy;
-import org.springframework.core.ResolvableType;
-import org.springframework.data.mapping.Child;
-import org.springframework.data.mapping.Person;
-import org.springframework.data.mapping.PersonDocument;
-
-/**
- * Unit tests for {@link SimpleEntityCallbacks}.
- *
- * @author Mark Paluch
- */
-public class SimpleEntityCallbacksUnitTests {
-
-	@Test
-	public void shouldDispatchCallback() {
-
-		AnnotationConfigApplicationContext ctx = new AnnotationConfigApplicationContext(MyConfig.class);
-
-		SimpleEntityCallbacks callbacks = new SimpleEntityCallbacks(ctx);
-
-		PersonDocument personDocument = new PersonDocument(null, "Walter", null);
-		PersonDocument afterCallback = callbacks.callback(personDocument, BeforeSaveCallback.class,
-				BeforeSaveCallback::onBeforeSave);
-
-		assertThat(afterCallback.getSsn()).isEqualTo(6);
-	}
-
-	@Test
-	public void shouldDispatchCallsToLambdaReceivers() {
-
-		AnnotationConfigApplicationContext ctx = new AnnotationConfigApplicationContext(LambdaConfig.class);
-
-		SimpleEntityCallbacks callbacks = new SimpleEntityCallbacks(ctx);
-
-		PersonDocument personDocument = new PersonDocument(null, "Walter", null);
-		PersonDocument afterCallback = callbacks.callback(personDocument, BeforeSaveCallback.class,
-				BeforeSaveCallback::onBeforeSave);
-
-		assertThat(afterCallback.getSsn()).isEqualTo(6);
-	}
-
-	@Test
-	public void shouldDiscoverCallbackType() {
-
-		AnnotationConfigApplicationContext ctx = new AnnotationConfigApplicationContext(MyConfig.class);
-
-		SimpleEntityCallbacks callbacks = new SimpleEntityCallbacks(ctx);
-
-		Collection<EntityCallback<?>> entityCallbacks = callbacks.getEntityCallbacks(new PersonDocument(null, null, null),
-				ResolvableType.forType(BeforeSaveCallback.class));
-
-		assertThat(entityCallbacks).hasSize(1).element(0).isInstanceOf(MyBeforeSaveCallback.class);
-	}
-
-	@Test
-	public void shouldDiscoverCallbackTypeByName() {
-
-		AnnotationConfigApplicationContext ctx = new AnnotationConfigApplicationContext(MyConfig.class);
-
-		SimpleEntityCallbacks callbacks = new SimpleEntityCallbacks(ctx);
-		callbacks.removeAllCallbacks();
-		callbacks.addEntityCallbackBean("namedCallback");
-
-		Collection<EntityCallback<?>> entityCallbacks = callbacks.getEntityCallbacks(new PersonDocument(null, null, null),
-				ResolvableType.forType(BeforeSaveCallback.class));
-
-		assertThat(entityCallbacks).hasSize(1).element(0).isInstanceOf(MyOtherCallback.class);
-	}
-
-	@Test
-	public void shouldSupportCallbackTypes() {
-
-		SimpleEntityCallbacks callbacks = new SimpleEntityCallbacks();
-
-		assertThat(callbacks.supportsEvent(MyBeforeSaveCallback.class, ResolvableType.forClass(Person.class))).isTrue();
-		assertThat(callbacks.supportsEvent(MyBeforeSaveCallback.class, ResolvableType.forClass(Child.class))).isTrue();
-		assertThat(callbacks.supportsEvent(BeforeSaveCallback.class, ResolvableType.forClass(PersonDocument.class)))
-				.isTrue();
-
-		assertThat(callbacks.supportsEvent(MyBeforeSaveCallback.class, ResolvableType.forClass(Object.class))).isFalse();
-		assertThat(callbacks.supportsEvent(MyBeforeSaveCallback.class, ResolvableType.forClass(User.class))).isFalse();
-	}
-
-	@Test
-	public void shouldSupportInstanceCallbackTypes() {
-
-		SimpleEntityCallbacks callbacks = new SimpleEntityCallbacks();
-
-		MyBeforeSaveCallback callback = new MyBeforeSaveCallback();
-
-		assertThat(callbacks.supportsEvent(callback, ResolvableType.forClass(Person.class),
-				ResolvableType.forClass(BeforeSaveCallback.class))).isTrue();
-		assertThat(callbacks.supportsEvent(callback, ResolvableType.forClass(Child.class),
-				ResolvableType.forClass(BeforeSaveCallback.class))).isTrue();
-		assertThat(callbacks.supportsEvent(callback, ResolvableType.forClass(PersonDocument.class),
-				ResolvableType.forClass(BeforeSaveCallback.class))).isTrue();
-		assertThat(callbacks.supportsEvent(callback, ResolvableType.forClass(PersonDocument.class),
-				ResolvableType.forClass(BeforeSaveCallback.class))).isTrue();
-
-		assertThat(callbacks.supportsEvent(callback, ResolvableType.forClass(User.class),
-				ResolvableType.forClass(BeforeSaveCallback.class))).isFalse();
-	}
-
-	@Configuration
-	static class MyConfig {
-
-		@Bean
-		MyBeforeSaveCallback callback() {
-			return new MyBeforeSaveCallback();
-		}
-
-		@Bean
-		@Lazy
-		Object namedCallback() {
-			return new MyOtherCallback();
-		}
-	}
-
-	@Configuration
-	static class LambdaConfig {
-
-		@Bean
-		BeforeSaveCallback<User> userCallback() {
-			return object -> object;
-		}
-
-		@Bean
-		BeforeSaveCallback<Person> personCallback() {
-			return object -> {
-				object.setSsn(object.getFirstName().length());
-				return object;
-			};
-		}
-	}
-
-	static class MyBeforeSaveCallback implements BeforeSaveCallback<Person> {
-
-		@Override
-		public Person onBeforeSave(Person object) {
-			object.setSsn(object.getFirstName().length());
-			return object;
-		}
-	}
-
-	static class MyOtherCallback implements BeforeSaveCallback<Person> {
-
-		@Override
-		public Person onBeforeSave(Person object) {
-			return object;
-		}
-	}
-
-	static class User {}
-}