DATAKV-25

+ add error handler setter on the container
+ improve docs

spring-data-redis/src/main/java/org/springframework/data/keyvalue/redis/config/RedisListenerContainerParser.java

@@ -63,6 +63,12 @@ class RedisListenerContainerParser extends AbstractSimpleBeanDefinitionParser {
 				builder.addPropertyReference(propertyName, attribute.getValue());
 			}
 		}
+
+		String phase = element.getAttribute("phase");
+		if (StringUtils.hasText(phase)) {
+			builder.addPropertyValue("phase", phase);
+		}
+
 		postProcess(builder, element);
 
 		// parse nested listeners
@@ -81,6 +87,11 @@ class RedisListenerContainerParser extends AbstractSimpleBeanDefinitionParser {
 		}
 	}
 
+	@Override
+	protected boolean isEligibleAttribute(String attributeName) {
+		return (!"phase".equals(attributeName));
+	}
+
 	/**
 	 * Parses a listener definition. Returns the listener bean reference definition (as the array first entry) and its associated topics (also as bean definitions).
 	 * 

spring-data-redis/src/main/java/org/springframework/data/keyvalue/redis/listener/RedisMessageListenerContainer.java

@@ -44,6 +44,7 @@ import org.springframework.data.keyvalue.redis.serializer.StringRedisSerializer;
 import org.springframework.scheduling.SchedulingAwareRunnable;
 import org.springframework.util.ClassUtils;
 import org.springframework.util.CollectionUtils;
+import org.springframework.util.ErrorHandler;
 
 /**
  * Container providing asynchronous behaviour for Redis message listeners.
@@ -60,7 +61,10 @@ import org.springframework.util.CollectionUtils;
  */
 public class RedisMessageListenerContainer implements InitializingBean, DisposableBean, BeanNameAware, SmartLifecycle {
 
-	private static final Log log = LogFactory.getLog(RedisMessageListenerContainer.class);
+	/** Logger available to subclasses */
+	protected final Log logger = LogFactory.getLog(getClass());
+
+
 
 	/**
 	 * Default thread name prefix: "RedisListeningContainer-".
@@ -78,6 +82,8 @@ public class RedisMessageListenerContainer implements InitializingBean, Disposab
 
 	private String beanName;
 
+	private ErrorHandler errorHandler;
+
 
 	private final Object monitor = new Object();
 	// whether the container is running (or not)
@@ -115,8 +121,9 @@ public class RedisMessageListenerContainer implements InitializingBean, Disposab
 			subscriptionExecutor = taskExecutor;
 		}
 
-		start();
 		initialized = true;
+
+		start();
 	}
 
 	/**
@@ -140,8 +147,8 @@ public class RedisMessageListenerContainer implements InitializingBean, Disposab
 			if (taskExecutor instanceof DisposableBean) {
 				((DisposableBean) taskExecutor).destroy();
 
-				if (log.isDebugEnabled()) {
-					log.debug("Stopped internally-managed task executor");
+				if (logger.isDebugEnabled()) {
+					logger.debug("Stopped internally-managed task executor");
 				}
 			}
 		}
@@ -185,8 +192,8 @@ public class RedisMessageListenerContainer implements InitializingBean, Disposab
 				}
 			}
 
-			if (log.isDebugEnabled()) {
-				log.debug("Started RedisMessageListenerContainer");
+			if (logger.isDebugEnabled()) {
+				logger.debug("Started RedisMessageListenerContainer");
 			}
 		}
 	}
@@ -207,8 +214,74 @@ public class RedisMessageListenerContainer implements InitializingBean, Disposab
 			}
 		}
 
-		if (log.isDebugEnabled()) {
-			log.debug("Stopped RedisMessageListenerContainer");
+		if (logger.isDebugEnabled()) {
+			logger.debug("Stopped RedisMessageListenerContainer");
+		}
+	}
+
+
+	/**
+	 * Process a message received from the provider.
+	 * 
+	 * @param message
+	 * @param pattern
+	 */
+	protected void processMessage(MessageListener listener, Message message, byte[] pattern) {
+		executeListener(listener, message, pattern);
+	}
+
+
+	/**
+	 * Execute the specified listener.
+	 * 
+	 * @see #handleListenerException
+	 */
+	protected void executeListener(MessageListener listener, Message message, byte[] pattern) {
+		try {
+			listener.onMessage(message, pattern);
+		} catch (Throwable ex) {
+			handleListenerException(ex);
+		}
+	}
+
+	/**
+	 * Return whether this container is currently active,
+	 * that is, whether it has been set up but not shut down yet.
+	 */
+	public final boolean isActive() {
+		return initialized;
+	}
+
+	/**
+	 * Handle the given exception that arose during listener execution.
+	 * <p>The default implementation logs the exception at error level.
+	 * This can be overridden in subclasses.
+	 * @param ex the exception to handle
+	 */
+	protected void handleListenerException(Throwable ex) {
+		if (isActive()) {
+			// Regular case: failed while active.
+			// Invoke ErrorHandler if available.
+			invokeErrorHandler(ex);
+		}
+		else {
+			// Rare case: listener thread failed after container shutdown.
+			// Log at debug level, to avoid spamming the shutdown logger.
+			logger.debug("Listener exception after container shutdown", ex);
+		}
+	}
+
+	/**
+	 * Invoke the registered ErrorHandler, if any. Log at error level otherwise.
+	 * @param ex the uncaught error that arose during message processing.
+	 * @see #setErrorHandler
+	 */
+	protected void invokeErrorHandler(Throwable ex) {
+		if (this.errorHandler != null) {
+			this.errorHandler.handleError(ex);
+		}
+		else if (logger.isWarnEnabled()) {
+			logger.warn("Execution of JMS message listener failed, and no ErrorHandler has been set.", ex);
 		}
 	}
 
@@ -270,6 +343,15 @@ public class RedisMessageListenerContainer implements InitializingBean, Disposab
 		this.serializer = serializer;
 	}
 
+	/**
+	 * Set an ErrorHandler to be invoked in case of any uncaught exceptions thrown
+	 * while processing a Message. By default there will be <b>no</b> ErrorHandler
+	 * so that error-level logging is the only result.
+	 */
+	public void setErrorHandler(ErrorHandler errorHandler) {
+		this.errorHandler = errorHandler;
+	}
+
 	/**
 	 * Attaches the given listeners (and their topics) to the container.
 	 * 
@@ -332,7 +414,7 @@ public class RedisMessageListenerContainer implements InitializingBean, Disposab
 	 * Method inspecting whether listening for messages (and thus using a thread) is actually needed and triggering it.
 	 */
 	private void lazyListen() {
-		boolean debug = log.isDebugEnabled();
+		boolean debug = logger.isDebugEnabled();
 		boolean started = false;
 
 		if (isRunning()) {
@@ -348,10 +430,10 @@ public class RedisMessageListenerContainer implements InitializingBean, Disposab
 				}
 				if (debug) {
 					if (started) {
-						log.debug("Started listening for Redis messages");
+						logger.debug("Started listening for Redis messages");
 					}
 					else {
-						log.debug("Postpone listening for Redis messages until actual listeners are added");
+						logger.debug("Postpone listening for Redis messages until actual listeners are added");
 					}
 				}
 			}
@@ -362,7 +444,7 @@ public class RedisMessageListenerContainer implements InitializingBean, Disposab
 		List<byte[]> channels = new ArrayList<byte[]>(topics.size());
 		List<byte[]> patterns = new ArrayList<byte[]>(topics.size());
 
-		boolean trace = log.isTraceEnabled();
+		boolean trace = logger.isTraceEnabled();
 
 		for (Topic topic : topics) {
 
@@ -378,7 +460,7 @@ public class RedisMessageListenerContainer implements InitializingBean, Disposab
 				channels.add(holder.array);
 
 				if (trace)
-					log.trace("Adding listener '" + listener + "' on channel '" + topic.getTopic() + "'");
+					logger.trace("Adding listener '" + listener + "' on channel '" + topic.getTopic() + "'");
 			}
 
 			else if (topic instanceof PatternTopic) {
@@ -391,7 +473,7 @@ public class RedisMessageListenerContainer implements InitializingBean, Disposab
 				patterns.add(holder.array);
 
 				if (trace)
-					log.trace("Adding listener '" + listener + "' for pattern '" + topic.getTopic() + "'");
+					logger.trace("Adding listener '" + listener + "' for pattern '" + topic.getTopic() + "'");
 			}
 
 			else {
@@ -406,6 +488,7 @@ public class RedisMessageListenerContainer implements InitializingBean, Disposab
 		}
 	}
 
+
 	/**
 	 * Runnable used for Redis subscription. Implemented as a dedicated class to provide as many hints
 	 * as possible to the underlying thread pool.
@@ -639,7 +722,7 @@ public class RedisMessageListenerContainer implements InitializingBean, Disposab
 				taskExecutor.execute(new Runnable() {
 					@Override
 					public void run() {
-						messageListener.onMessage(message, null);
+						processMessage(messageListener, message, null);
 					}
 				});
 			}
@@ -650,7 +733,7 @@ public class RedisMessageListenerContainer implements InitializingBean, Disposab
 				taskExecutor.execute(new Runnable() {
 					@Override
 					public void run() {
-						messageListener.onMessage(message, pattern.clone());
+						processMessage(messageListener, message, pattern.clone());
 					}
 				});
 			}

spring-data-redis/src/main/resources/org/springframework/data/keyvalue/redis/config/spring-redis-1.0.xsd

@@ -84,19 +84,6 @@
 					</xsd:appinfo>
 				</xsd:annotation>
 			</xsd:attribute>
-			<xsd:attribute name="error-handler" type="xsd:string">
-				<xsd:annotation>
-					<xsd:documentation><![CDATA[
-	A reference to an ErrorHandler strategy for handling any uncaught Exceptions
-	that may occur during the execution of the MessageListener.
-					]]></xsd:documentation>
-					<xsd:appinfo>
-						<tool:annotation kind="ref">
-							<tool:expected-type type="org.springframework.util.ErrorHandler"/>
-						</tool:annotation>
-					</xsd:appinfo>
-				</xsd:annotation>
-			</xsd:attribute>
 			<xsd:attribute name="phase" type="xsd:string">
 				<xsd:annotation>
 					<xsd:documentation><![CDATA[

spring-data-redis/src/test/java/org/springframework/data/keyvalue/redis/config/NamespaceTest.java

@@ -17,6 +17,8 @@ package org.springframework.data.keyvalue.redis.config;
 
 import static org.junit.Assert.*;
 
+import java.util.concurrent.TimeUnit;
+
 import org.junit.After;
 import org.junit.Before;
 import org.junit.Test;
@@ -56,4 +58,15 @@ public class NamespaceTest {
 		template.convertAndSend("z1", "[Z]test");
 		//Thread.sleep(TimeUnit.SECONDS.toMillis(5));
 	}
+
+	@Test
+	public void testErrorHandler() throws Exception {
+		StubErrorHandler handler = ctx.getBean(StubErrorHandler.class);
+		
+		int index = handler.throwables.size();
+		StringRedisTemplate template = ctx.getBean(StringRedisTemplate.class);
+		template.convertAndSend("exception", "test1");
+		handler.throwables.pollLast(3, TimeUnit.SECONDS);
+		assertEquals(index + 1, handler.throwables.size());
+	}
 }

spring-data-redis/src/test/java/org/springframework/data/keyvalue/redis/config/StubErrorHandler.java

@@ -0,0 +1,35 @@
+/*
+ * Copyright 2011 the original author or authors.
+ * 
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ * 
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ * 
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.springframework.data.keyvalue.redis.config;
+
+import java.util.concurrent.BlockingDeque;
+import java.util.concurrent.LinkedBlockingDeque;
+
+import org.springframework.util.ErrorHandler;
+
+/**
+ * @author Costin Leau
+ */
+public class StubErrorHandler implements ErrorHandler {
+
+	public BlockingDeque<Throwable> throwables = new LinkedBlockingDeque<Throwable>();
+
+	@Override
+	public void handleError(Throwable t) {
+		throwables.add(t);
+	}
+
+}

spring-data-redis/src/test/java/org/springframework/data/keyvalue/redis/listener/adapter/ThrowableMessageListener.java

@@ -0,0 +1,31 @@
+/*
+ * Copyright 2011 the original author or authors.
+ * 
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ * 
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ * 
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.springframework.data.keyvalue.redis.listener.adapter;
+
+import org.springframework.data.keyvalue.redis.connection.Message;
+import org.springframework.data.keyvalue.redis.connection.MessageListener;
+
+/**
+ * 
+ * @author Costin Leau
+ */
+public class ThrowableMessageListener implements MessageListener {
+
+	@Override
+	public void onMessage(Message message, byte[] pattern) {
+		throw new IllegalStateException("throwing exception for message " + message);
+	}
+}

spring-data-redis/src/test/resources/org/springframework/data/keyvalue/redis/config/namespace.xml

@@ -12,17 +12,21 @@
 
 	<task:executor id="testTaskExecutor" />
 	 			
-	<redis:listener-container>
+	<redis:listener-container topic-serializer="serializer">
 		<!-- default handle method -->
 		<redis:listener ref="testBean1" channel="z1 z2" pattern="x*"/>
 		<!-- channel subscription only -->
 		<redis:listener ref="testBean1" method="anotherHandle" channel="x1" serializer="serializer"/>
+		<redis:listener ref="testBean2" channel="exception"/>
 	</redis:listener-container>
 	
 	<bean id="testBean1" class="org.springframework.data.keyvalue.redis.listener.adapter.RedisMDP"/>
+	<bean id="testBean2" class="org.springframework.data.keyvalue.redis.listener.adapter.ThrowableMessageListener"/>
 	
 	<bean id="serializer" class="org.springframework.data.keyvalue.redis.serializer.StringRedisSerializer"/>
 	
+	<bean id="handler" class="org.springframework.data.keyvalue.redis.config.StubErrorHandler"/>
+	
 	<bean id="redisTemplate" class="org.springframework.data.keyvalue.redis.core.StringRedisTemplate">
 		<property name="connectionFactory" ref="redisConnectionFactory"/>
 	</bean>