DATAREDIS-491 - Add configuration option to tune KeyspaceEventMessageListener usage.

We now allow more fine grained setup for usage of Redis keyspace events. This can be done programmatically via the RedisKeyValueAdapter or using `@EnableRedisRepositories.enableKeyspaceEvents`.

Original pull request: #193.

Author: christophstrobl

src/main/java/org/springframework/data/redis/core/RedisKeyValueAdapter.java

@@ -21,12 +21,16 @@
 import java.util.Map;
 import java.util.Map.Entry;
 import java.util.Set;
+import java.util.concurrent.atomic.AtomicReference;
 
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.springframework.beans.BeansException;
+import org.springframework.beans.factory.DisposableBean;
+import org.springframework.beans.factory.InitializingBean;
 import org.springframework.context.ApplicationContext;
 import org.springframework.context.ApplicationContextAware;
+import org.springframework.context.ApplicationEventPublisher;
 import org.springframework.context.ApplicationListener;
 import org.springframework.core.convert.ConversionService;
 import org.springframework.core.convert.ConverterNotFoundException;
@@ -50,22 +54,23 @@
 import org.springframework.data.redis.util.ByteUtils;
 import org.springframework.data.util.CloseableIterator;
 import org.springframework.util.Assert;
+import org.springframework.util.ObjectUtils;
 
 /**
  * Redis specific {@link KeyValueAdapter} implementation. Uses binary codec to read/write data from/to Redis. Objects
  * are stored in a Redis Hash using the value of {@link RedisHash}, the {@link KeyspaceConfiguration} or just
  * {@link Class#getName()} as a prefix. <br />
  * <strong>Example</strong>
- * 
+ *
  * <pre>
  * <code>
  * &#64;RedisHash("persons")
  * class Person {
  *   &#64;Id String id;
  *   String name;
  * }
- * 
- * 
+ *
+ *
  *         prefix              ID
  *           |                 |
  *           V                 V
@@ -76,29 +81,33 @@
  * 4) Rand al'Thor
  * </code>
  * </pre>
- * 
+ *
  * <br />
  * The {@link KeyValueAdapter} is <strong>not</strong> intended to store simple types such as {@link String} values.
  * Please use {@link RedisTemplate} for this purpose.
- * 
+ *
  * @author Christoph Strobl
  * @author Mark Paluch
  * @since 1.7
  */
 public class RedisKeyValueAdapter extends AbstractKeyValueAdapter
-		implements ApplicationContextAware, ApplicationListener<RedisKeyspaceEvent> {
+		implements InitializingBean, ApplicationContextAware, ApplicationListener<RedisKeyspaceEvent> {
 
 	private static final Logger LOGGER = LoggerFactory.getLogger(RedisKeyValueAdapter.class);
 
 	private RedisOperations<?, ?> redisOps;
 	private RedisConverter converter;
 	private RedisMessageListenerContainer messageListenerContainer;
-	private KeyExpirationEventMessageListener expirationListener;
+	private AtomicReference<KeyExpirationEventMessageListener> expirationListener = new AtomicReference<KeyExpirationEventMessageListener>(
+			null);
+	private ApplicationEventPublisher eventPublisher;
+
+	private EnableKeyspaceEvents enableKeyspaceEvents = EnableKeyspaceEvents.ON_STARTUP;
 
 	/**
 	 * Creates new {@link RedisKeyValueAdapter} with default {@link RedisMappingContext} and default
 	 * {@link CustomConversions}.
-	 * 
+	 *
 	 * @param redisOps must not be {@literal null}.
 	 */
 	public RedisKeyValueAdapter(RedisOperations<?, ?> redisOps) {
@@ -107,7 +116,7 @@ public RedisKeyValueAdapter(RedisOperations<?, ?> redisOps) {
 
 	/**
 	 * Creates new {@link RedisKeyValueAdapter} with default {@link CustomConversions}.
-	 * 
+	 *
 	 * @param redisOps must not be {@literal null}.
 	 * @param mappingContext must not be {@literal null}.
 	 */
@@ -117,7 +126,7 @@ public RedisKeyValueAdapter(RedisOperations<?, ?> redisOps, RedisMappingContext
 
 	/**
 	 * Creates new {@link RedisKeyValueAdapter}.
-	 * 
+	 *
 	 * @param redisOps must not be {@literal null}.
 	 * @param mappingContext must not be {@literal null}.
 	 * @param customConversions can be {@literal null}.
@@ -138,12 +147,12 @@ public RedisKeyValueAdapter(RedisOperations<?, ?> redisOps, RedisMappingContext
 		converter = mappingConverter;
 		this.redisOps = redisOps;
 
-		initKeyExpirationListener();
+		intiMessageListenerContainer();
 	}
 
 	/**
 	 * Creates new {@link RedisKeyValueAdapter} with specific {@link RedisConverter}.
-	 * 
+	 *
 	 * @param redisOps must not be {@literal null}.
 	 * @param mappingContext must not be {@literal null}.
 	 */
@@ -156,7 +165,7 @@ public RedisKeyValueAdapter(RedisOperations<?, ?> redisOps, RedisConverter redis
 		converter = redisConverter;
 		this.redisOps = redisOps;
 
-		initKeyExpirationListener();
+		intiMessageListenerContainer();
 	}
 
 	/**
@@ -175,6 +184,14 @@ public Object put(final Serializable id, final Object item, final Serializable k
 			converter.write(item, rdo);
 		}
 
+		if (ObjectUtils.nullSafeEquals(EnableKeyspaceEvents.ON_DEMAND, enableKeyspaceEvents)
+				&& this.expirationListener.get() == null) {
+
+			if (rdo.getTimeToLive() != null && rdo.getTimeToLive().longValue() > 0) {
+				initKeyExpirationListener();
+			}
+		}
+
 		if (rdo.getId() == null) {
 
 			rdo.setId(converter.getConversionService().convert(id, String.class));
@@ -397,7 +414,7 @@ public Long doInRedis(RedisConnection connection) throws DataAccessException {
 
 	/**
 	 * Execute {@link RedisCallback} via underlying {@link RedisOperations}.
-	 * 
+	 *
 	 * @param callback must not be {@literal null}.
 	 * @see RedisOperations#execute(RedisCallback)
 	 * @return
@@ -408,7 +425,7 @@ public <T> T execute(RedisCallback<T> callback) {
 
 	/**
 	 * Get the {@link RedisConverter} in use.
-	 * 
+	 *
 	 * @return never {@literal null}.
 	 */
 	public RedisConverter getConverter() {
@@ -430,7 +447,7 @@ public byte[] createKey(String keyspace, String id) {
 
 	/**
 	 * Convert given source to binary representation using the underlying {@link ConversionService}.
-	 * 
+	 *
 	 * @param source
 	 * @return
 	 * @throws ConverterNotFoundException
@@ -444,13 +461,38 @@ public byte[] toBytes(Object source) {
 		return converter.getConversionService().convert(source, byte[].class);
 	}
 
+	/**
+	 * Configure usage of {@link KeyExpirationEventMessageListener}.
+	 *
+	 * @param enableKeyspaceEvents
+	 * @since 1.8
+	 */
+	public void setEnableKeyspaceEvents(EnableKeyspaceEvents enableKeyspaceEvents) {
+		this.enableKeyspaceEvents = enableKeyspaceEvents;
+	}
+
+	/**
+	 * @see org.springframework.beans.factory.InitializingBean#afterPropertiesSet()
+	 * @since 1.8
+	 */
+	@Override
+	public void afterPropertiesSet() {
+
+		if (ObjectUtils.nullSafeEquals(EnableKeyspaceEvents.ON_STARTUP, this.enableKeyspaceEvents)) {
+			initKeyExpirationListener();
+		}
+	}
+
 	/*
 	 * (non-Javadoc)
 	 * @see org.springframework.beans.factory.DisposableBean#destroy()
 	 */
 	public void destroy() throws Exception {
 
-		this.expirationListener.destroy();
+		if (this.expirationListener.get() != null) {
+			this.expirationListener.get().destroy();
+		}
+
 		this.messageListenerContainer.destroy();
 	}
 
@@ -490,26 +532,39 @@ public Void doInRedis(RedisConnection connection) throws DataAccessException {
 	 */
 	@Override
 	public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
-		this.expirationListener.setApplicationEventPublisher(applicationContext);
+		this.eventPublisher = applicationContext;
 	}
 
-	private void initKeyExpirationListener() {
+	private void intiMessageListenerContainer() {
 
 		this.messageListenerContainer = new RedisMessageListenerContainer();
 		messageListenerContainer.setConnectionFactory(((RedisTemplate<?, ?>) redisOps).getConnectionFactory());
 		messageListenerContainer.afterPropertiesSet();
 		messageListenerContainer.start();
+	}
+
+	private void initKeyExpirationListener() {
+
+		if (this.expirationListener.get() == null) {
+
+			MappingExpirationListener listener = new MappingExpirationListener(this.messageListenerContainer, this.redisOps,
+					this.converter);
+
+			if (this.eventPublisher != null) {
+				listener.setApplicationEventPublisher(this.eventPublisher);
+			}
 
-		this.expirationListener = new MappingExpirationListener(this.messageListenerContainer, this.redisOps,
-				this.converter);
-		this.expirationListener.init();
+			if (this.expirationListener.compareAndSet(null, listener)) {
+				listener.init();
+			}
+		}
 	}
 
 	/**
 	 * {@link MessageListener} implementation used to capture Redis keypspace notifications. Tries to read a previously
 	 * created phantom key {@code keyspace:id:phantom} to provide the expired object as part of the published
 	 * {@link RedisKeyExpiredEvent}.
-	 * 
+	 *
 	 * @author Christoph Strobl
 	 * @since 1.7
 	 */
@@ -520,7 +575,7 @@ static class MappingExpirationListener extends KeyExpirationEventMessageListener
 
 		/**
 		 * Creates new {@link MappingExpirationListener}.
-		 * 
+		 *
 		 * @param listenerContainer
 		 * @param ops
 		 * @param converter
@@ -582,4 +637,26 @@ private boolean isKeyExpirationMessage(Message message) {
 		}
 	}
 
+	/**
+	 * @author Christoph Strobl
+	 * @since 1.8
+	 */
+	public static enum EnableKeyspaceEvents {
+
+		/**
+		 * Initializes the {@link KeyExpirationEventMessageListener} on startup.
+		 */
+		ON_STARTUP,
+
+		/**
+		 * Initializes the {@link KeyExpirationEventMessageListener} on first insert having expiration time set.
+		 */
+		ON_DEMAND,
+
+		/**
+		 * Turn {@link KeyExpirationEventMessageListener} usage off. No expiration events will be received.
+		 */
+		OFF
+	}
+
 }

src/main/java/org/springframework/data/redis/repository/configuration/EnableRedisRepositories.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2015 the original author or authors.
+ * Copyright 2015-2016 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.
@@ -27,9 +27,11 @@
 import org.springframework.context.annotation.Import;
 import org.springframework.data.keyvalue.core.KeyValueOperations;
 import org.springframework.data.keyvalue.repository.config.QueryCreatorType;
+import org.springframework.data.redis.core.RedisKeyValueAdapter.EnableKeyspaceEvents;
 import org.springframework.data.redis.core.RedisOperations;
 import org.springframework.data.redis.core.convert.KeyspaceConfiguration;
 import org.springframework.data.redis.core.index.IndexConfiguration;
+import org.springframework.data.redis.listener.KeyExpirationEventMessageListener;
 import org.springframework.data.redis.repository.query.RedisQueryCreator;
 import org.springframework.data.redis.repository.support.RedisRepositoryFactoryBean;
 import org.springframework.data.repository.config.DefaultRepositoryBaseClass;
@@ -53,7 +55,8 @@
 
 	/**
 	 * Alias for the {@link #basePackages()} attribute. Allows for more concise annotation declarations e.g.:
-	 * {@code @EnableRedisRepositories("org.my.pkg")} instead of {@code @EnableRedisRepositories(basePackages="org.my.pkg")}.
+	 * {@code @EnableRedisRepositories("org.my.pkg")} instead of
+	 * {@code @EnableRedisRepositories(basePackages="org.my.pkg")}.
 	 */
 	String[] value() default {};
 
@@ -154,4 +157,12 @@
 	 */
 	Class<? extends KeyspaceConfiguration> keyspaceConfiguration() default KeyspaceConfiguration.class;
 
+	/**
+	 * Configure usage of {@link KeyExpirationEventMessageListener}.
+	 *
+	 * @return
+	 * @since 1.8
+	 */
+	EnableKeyspaceEvents enableKeyspaceEvents() default EnableKeyspaceEvents.ON_DEMAND;
+
 }

src/main/java/org/springframework/data/redis/repository/configuration/RedisRepositoryConfigurationExtension.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2015 the original author or authors.
+ * Copyright 2015-2016 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.
@@ -122,6 +122,11 @@ public void registerBeansForRoot(BeanDefinitionRegistry registry, RepositoryConf
 				new RuntimeBeanReference(REDIS_CONVERTER_BEAN_NAME));
 
 		redisKeyValueAdapterDefinition.setConstructorArgumentValues(constructorArgumentValuesForRedisKeyValueAdapter);
+
+		DirectFieldAccessor dfa = new DirectFieldAccessor(configurationSource);
+		AnnotationAttributes aa = (AnnotationAttributes) dfa.getPropertyValue("attributes");
+		redisKeyValueAdapterDefinition.setAttribute("enableKeyspaceEvents", aa.getEnum("enableKeyspaceEvents"));
+
 		registerIfNotAlreadyRegistered(redisKeyValueAdapterDefinition, registry, REDIS_ADAPTER_BEAN_NAME,
 				configurationSource);
 

src/test/java/org/springframework/data/redis/core/RedisKeyValueAdapterTests.java

@@ -69,6 +69,7 @@ public void setUp() {
 		mappingContext.afterPropertiesSet();
 
 		adapter = new RedisKeyValueAdapter(template, mappingContext);
+		adapter.afterPropertiesSet();
 
 		template.execute(new RedisCallback<Void>() {