DATAGEODE-236 - Adding RegionEntry event processing and more testing

Author: Udo Kohlmeyer

src/main/java/org/springframework/data/gemfire/config/annotation/AsCacheListener.java

@@ -36,12 +36,18 @@
 public @interface AsCacheListener {
 
 	/**
-	 * An array of {@link CacheListenerEventType} that control what events need to be observed
+	 * An array of {@link CacheListenerEventType} that control what region CRUD events need to be observed
+	 * {@link CacheListenerEventType} and {@link RegionCacheListenerEventType} cannot be set on the same method. As they
+	 * are mutually exclusive and require that the implementing method uses {@link org.apache.geode.cache.RegionEvent} or
+	 * {@link org.apache.geode.cache.EntryEvent}
 	 */
 	CacheListenerEventType[] eventTypes() default {};
 
 	/**
-	 * An array of {@link CacheListenerEventType} that control what events need to be observed
+	 * An array of {@link RegionCacheListenerEventType} that control what region events need to be observed
+	 * {@link CacheListenerEventType} and {@link RegionCacheListenerEventType} cannot be set on the same method. As they
+	 * are mutually exclusive and require that the implementing method uses {@link org.apache.geode.cache.RegionEvent} or
+	 * {@link org.apache.geode.cache.EntryEvent}
 	 */
 	RegionCacheListenerEventType[] regionEventTypes() default {};
 

src/main/java/org/springframework/data/gemfire/config/support/CacheListenerPostProcessor.java

@@ -14,9 +14,23 @@
  * limitations under the License.
  *
  */
+
 package org.springframework.data.gemfire.config.support;
 
+import static java.util.Arrays.*;
+import static org.springframework.data.gemfire.util.ArrayUtils.*;
+import static org.springframework.data.gemfire.util.RuntimeExceptionFactory.*;
+
+import java.lang.annotation.Annotation;
+import java.lang.reflect.Method;
+import java.lang.reflect.Modifier;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Optional;
+
+import org.apache.geode.cache.EntryEvent;
 import org.apache.geode.cache.Region;
+import org.apache.geode.cache.RegionEvent;
 import org.springframework.beans.BeansException;
 import org.springframework.beans.factory.BeanFactory;
 import org.springframework.beans.factory.BeanFactoryAware;
@@ -34,17 +48,9 @@
 import org.springframework.util.ObjectUtils;
 import org.springframework.util.ReflectionUtils;
 
-import java.lang.annotation.Annotation;
-import java.lang.reflect.Method;
-import java.lang.reflect.Modifier;
-import java.util.ArrayList;
-import java.util.List;
-import java.util.Optional;
-
-import static java.util.Arrays.*;
-import static org.springframework.data.gemfire.util.ArrayUtils.*;
-import static org.springframework.data.gemfire.util.RuntimeExceptionFactory.*;
-
+/**
+ *
+ */
 @Configuration public class CacheListenerPostProcessor implements BeanPostProcessor, BeanFactoryAware {
 
 	private ConfigurableListableBeanFactory beanFactory;
@@ -65,44 +71,52 @@ private void registerAnyDeclaredCacheListenerAnnotatedMethods(Object bean) {
 			if (cacheListenerAnnotation != null) {
 
 				Assert.isTrue(Modifier.isPublic(method.getModifiers()), String
-						.format("The bean [%s] method [%s] annotated with [%s] must be public", bean.getClass().getName(),
-								method.getName(), AsCacheListener.class.getName()));
+					.format("The bean [%s] method [%s] annotated with [%s] must be public", bean.getClass().getName(),
+						method.getName(), AsCacheListener.class.getName()));
 
 				AnnotationAttributes cacheListenerAttributes = resolveAnnotationAttributes(cacheListenerAnnotation);
 
-				CacheListenerEventType[] eventTypes = (CacheListenerEventType[]) cacheListenerAttributes.get("eventTypes");
+				CacheListenerEventType[] eventTypes = (CacheListenerEventType[]) cacheListenerAttributes
+					.get("eventTypes");
 				RegionCacheListenerEventType[] regionEventTypes = (RegionCacheListenerEventType[]) cacheListenerAttributes
-						.get("regionEventTypes");
+					.get("regionEventTypes");
 
 				if (regionEventTypes.length > 0 && eventTypes.length > 0) {
 					throw new IllegalArgumentException(String
-							.format("Populating both event and regionEvent types is not allowed for bean [%s] method [%s]",
-									bean.getClass().getName(), method.getName()));
+						.format("Populating both event and regionEvent types is not allowed for bean [%s] method [%s]",
+							bean.getClass().getName(), method.getName()));
 				}
 
-				String[] regions = getRegionsForEventRegistration(cacheListenerAttributes.getStringArray("regions"),
-						getBeanFactory());
-
-				if (eventTypes.length > 0) {
-					registerCacheListenerEvents(bean, method, regions, eventTypes);
-				}
-				if (regionEventTypes.length > 0) {
-					registerRegionCacheListenerEvents(bean, method, regions, regionEventTypes);
-				}
+				registerEventHandlerToRegion(bean, method, cacheListenerAttributes, eventTypes, regionEventTypes);
 			}
 		});
 	}
 
+	private void registerEventHandlerToRegion(Object bean, Method method, AnnotationAttributes cacheListenerAttributes,
+		CacheListenerEventType[] eventTypes, RegionCacheListenerEventType[] regionEventTypes) {
+		String[] regions = getRegionsForEventRegistration(cacheListenerAttributes.getStringArray("regions"),
+			getBeanFactory());
+
+		if (eventTypes.length > 0) {
+			EventProcessorUtils.validateCacheListenerMethodParameters(method,EntryEvent.class);
+			registerCacheListenerEvents(bean, method, regions, eventTypes);
+		}
+		if (regionEventTypes.length > 0) {
+			EventProcessorUtils.validateCacheListenerMethodParameters(method, RegionEvent.class);
+			registerRegionCacheListenerEvents(bean, method, regions, regionEventTypes);
+		}
+	}
+
 	private void registerCacheListenerEvents(Object bean, Method method, String[] regions,
-			CacheListenerEventType[] eventTypes) {
+		CacheListenerEventType[] eventTypes) {
 
 		EventProcessorUtils.registerCacheListenerOntoRegions(regions, method, bean, eventTypes, getBeanFactory());
 	}
 
 	private void registerRegionCacheListenerEvents(Object bean, Method method, String[] regions,
-			RegionCacheListenerEventType[] regionEventTypes) {
+		RegionCacheListenerEventType[] regionEventTypes) {
 		EventProcessorUtils
-				.registerCacheListenerForRegionEventsOntoRegions(regions, method, bean, regionEventTypes, getBeanFactory());
+			.registerCacheListenerForRegionEventsOntoRegions(regions, method, bean, regionEventTypes, getBeanFactory());
 	}
 
 	private String[] getRegionsForEventRegistration(String[] regions, ConfigurableListableBeanFactory beanFactory) {
@@ -118,7 +132,8 @@ private String[] getRegionsForEventRegistration(String[] regions, ConfigurableLi
 			String[] tempRegions = new String[regionNames.size()];
 			regionNames.toArray(tempRegions);
 			return tempRegions;
-		} else {
+		}
+		else {
 			return regions;
 		}
 	}
@@ -128,33 +143,33 @@ private AnnotationAttributes resolveAnnotationAttributes(Annotation annotation)
 		return AnnotationAttributes.fromMap(AnnotationUtils.getAnnotationAttributes(annotation, false, true));
 	}
 
+	/**
+	 * Returns a reference to the containing Spring {@link BeanFactory}.
+	 *
+	 * @return a reference to the containing Spring {@link BeanFactory}.
+	 * @throws IllegalStateException if the {@link BeanFactory} was not configured.
+	 * @see org.springframework.beans.factory.BeanFactory
+	 */
+	protected ConfigurableListableBeanFactory getBeanFactory() {
+		return Optional.ofNullable(this.beanFactory)
+			.orElseThrow(() -> newIllegalStateException("BeanFactory was not properly configured"));
+	}
+
 	/**
 	 * Sets a reference to the configured Spring {@link BeanFactory}.
 	 *
 	 * @param beanFactory configured Spring {@link BeanFactory}.
 	 * @throws IllegalArgumentException if the given {@link BeanFactory} is not an instance of
-	 *                                  {@link ConfigurableListableBeanFactory}.
+	 * {@link ConfigurableListableBeanFactory}.
 	 * @see org.springframework.beans.factory.BeanFactoryAware
 	 * @see org.springframework.beans.factory.BeanFactory
 	 */
 	@Override @SuppressWarnings("all") public final void setBeanFactory(BeanFactory beanFactory) throws BeansException {
 
 		Assert.isInstanceOf(ConfigurableListableBeanFactory.class, beanFactory, String
-				.format("BeanFactory [%1$s] must be an instance of %2$s", ObjectUtils.nullSafeClassName(beanFactory),
-						ConfigurableListableBeanFactory.class.getSimpleName()));
+			.format("BeanFactory [%1$s] must be an instance of %2$s", ObjectUtils.nullSafeClassName(beanFactory),
+				ConfigurableListableBeanFactory.class.getSimpleName()));
 
 		this.beanFactory = (ConfigurableListableBeanFactory) beanFactory;
 	}
-
-	/**
-	 * Returns a reference to the containing Spring {@link BeanFactory}.
-	 *
-	 * @return a reference to the containing Spring {@link BeanFactory}.
-	 * @throws IllegalStateException if the {@link BeanFactory} was not configured.
-	 * @see org.springframework.beans.factory.BeanFactory
-	 */
-	protected ConfigurableListableBeanFactory getBeanFactory() {
-		return Optional.ofNullable(this.beanFactory)
-				.orElseThrow(() -> newIllegalStateException("BeanFactory was not properly configured"));
-	}
 }

src/main/java/org/springframework/data/gemfire/eventing/EventProcessorUtils.java

@@ -14,48 +14,61 @@
  * limitations under the License.
  *
  */
+
 package org.springframework.data.gemfire.eventing;
 
+import static java.util.Arrays.*;
+
+import java.lang.reflect.Method;
+import java.util.Optional;
+
 import org.apache.geode.cache.CacheListener;
+import org.apache.geode.cache.EntryEvent;
 import org.apache.geode.cache.Region;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
 import org.springframework.beans.factory.BeanFactory;
 import org.springframework.data.gemfire.eventing.config.CacheListenerEventType;
 import org.springframework.data.gemfire.eventing.config.PojoCacheListenerWrapper;
 import org.springframework.data.gemfire.eventing.config.PojoRegionEventCacheListenerWrapper;
 import org.springframework.data.gemfire.eventing.config.RegionCacheListenerEventType;
 
-import java.lang.reflect.Method;
-import java.util.Optional;
-
-import static java.util.Arrays.*;
-
 public class EventProcessorUtils {
 
-	private static transient Logger logger = LoggerFactory.getLogger(EventProcessorUtils.class);
-
 	public static void registerCacheListenerOntoRegions(String[] regions, Method method, Object bean,
-			CacheListenerEventType[] eventTypes, BeanFactory beanFactory) {
+		CacheListenerEventType[] eventTypes, BeanFactory beanFactory) {
 		PojoCacheListenerWrapper cacheListenerWrapper = new PojoCacheListenerWrapper(method, bean, eventTypes);
 
 		registerCacheListenerToRegions(regions, beanFactory, cacheListenerWrapper);
 	}
 
 	public static void registerCacheListenerForRegionEventsOntoRegions(String[] regions, Method method, Object bean,
-			RegionCacheListenerEventType[] regionEventTypes, BeanFactory beanFactory) {
+		RegionCacheListenerEventType[] regionEventTypes, BeanFactory beanFactory) {
 		PojoRegionEventCacheListenerWrapper regionEventCacheListenerWrapper = new PojoRegionEventCacheListenerWrapper(
-				method, bean, regionEventTypes);
+			method, bean, regionEventTypes);
 
 		registerCacheListenerToRegions(regions, beanFactory, regionEventCacheListenerWrapper);
 	}
 
 	private static void registerCacheListenerToRegions(String[] regions, BeanFactory beanFactory,
-			CacheListener cacheListener) {
+		CacheListener cacheListener) {
 		stream(regions).forEach(regionName -> {
 			Optional<Region> regionBeanOptional = Optional.of(beanFactory.getBean(regionName, Region.class));
 
 			regionBeanOptional.ifPresent(region -> region.getAttributesMutator().addCacheListener(cacheListener));
 		});
 	}
+
+	public static void validateCacheListenerMethodParameters(Method method, Class requireParameterType) {
+		Class<?>[] parameterTypes = method.getParameterTypes();
+		if (parameterTypes.length != 1) {
+			throw new IllegalArgumentException(String
+				.format("CacheListener method: %s does not currently support more than one parameter",
+					method.getName()));
+		}
+		if (!parameterTypes[0].isAssignableFrom(EntryEvent.class)) {
+			throw new IllegalArgumentException(String
+				.format("CacheListener method: %s requires an %s parameter type", method.getName(),
+					requireParameterType.getName()));
+		}
+	}
+
 }