INT-3196: RedisQueueMDE: Add RecoveryInterval

Catch `RedisSystemException` in the `ListenerTask` and sleep current Thread
 with `recoveryInterval` if the Endpoint is `active` before the next `restart()`

JIRA: https://jira.springsource.org/browse/INT-3196

RedisConnFailure & RedisSys Exceptions recovery

Introduce `RedisIntegrationEvent`s

* Catch all `Exception`s on `this.boundListOperations.rightPop`
* Log them and send within `RedisExceptionEvent`
* Mark some fields with `volatile`

spring-integration-redis/src/main/java/org/springframework/integration/redis/config/RedisQueueInboundChannelAdapterParser.java

@@ -18,11 +18,10 @@ package org.springframework.integration.redis.config;
 
 import org.w3c.dom.Element;
 
-import org.springframework.beans.factory.BeanDefinitionStoreException;
 import org.springframework.beans.factory.support.AbstractBeanDefinition;
 import org.springframework.beans.factory.support.BeanDefinitionBuilder;
-import org.springframework.beans.factory.xml.AbstractSingleBeanDefinitionParser;
 import org.springframework.beans.factory.xml.ParserContext;
+import org.springframework.integration.config.xml.AbstractChannelAdapterParser;
 import org.springframework.integration.config.xml.IntegrationNamespaceUtils;
 import org.springframework.integration.redis.inbound.RedisQueueMessageDrivenEndpoint;
 import org.springframework.util.StringUtils;
@@ -33,29 +32,11 @@ import org.springframework.util.StringUtils;
  * @author Artem Bilan
  * @since 3.0
  */
-public class RedisQueueInboundChannelAdapterParser extends AbstractSingleBeanDefinitionParser {
+public class RedisQueueInboundChannelAdapterParser extends AbstractChannelAdapterParser {
 
 	@Override
-	protected Class<?> getBeanClass(Element element) {
-		return RedisQueueMessageDrivenEndpoint.class;
-	}
-
-	@Override
-	protected final String resolveId(Element element, AbstractBeanDefinition definition, ParserContext parserContext)
-			throws BeanDefinitionStoreException {
-		String id = element.getAttribute("id");
-		if (!element.hasAttribute("channel")) {
-			// the created channel will get the 'id', so the adapter's bean name includes a suffix
-			id = id + ".adapter";
-		}
-		else if (!StringUtils.hasText(id)) {
-			id = parserContext.getReaderContext().generateBeanName(definition);
-		}
-		return id;
-	}
-
-	@Override
-	protected void doParse(Element element, ParserContext parserContext, BeanDefinitionBuilder builder) {
+	protected AbstractBeanDefinition doParse(Element element, ParserContext parserContext, String channelName) {
+		BeanDefinitionBuilder builder = BeanDefinitionBuilder.genericBeanDefinition(RedisQueueMessageDrivenEndpoint.class);
 		builder.addConstructorArgValue(element.getAttribute("queue"));
 
 		String connectionFactory = element.getAttribute("connection-factory");
@@ -69,15 +50,9 @@ public class RedisQueueInboundChannelAdapterParser extends AbstractSingleBeanDef
 		IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "error-channel");
 		IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "expect-message");
 		IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "receive-timeout");
-		IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, IntegrationNamespaceUtils.AUTO_STARTUP);
-		IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, IntegrationNamespaceUtils.PHASE);
-
-		String channelName = element.getAttribute("channel");
-		if (!StringUtils.hasText(channelName)) {
-			channelName = IntegrationNamespaceUtils.createDirectChannel(element, parserContext);
-		}
 		builder.addPropertyReference("outputChannel", channelName);
 
+		return builder.getBeanDefinition();
 	}
 
 }

spring-integration-redis/src/main/java/org/springframework/integration/redis/event/RedisExceptionEvent.java

@@ -0,0 +1,30 @@
+/*
+ * Copyright 2013 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.integration.redis.event;
+
+/**
+ * @author Artem Bilan
+ * @since 3.0
+ */
+@SuppressWarnings("serial")
+public class RedisExceptionEvent extends RedisIntegrationEvent {
+
+	public RedisExceptionEvent(Object source, Throwable cause) {
+		super(source, cause);
+	}
+
+}

spring-integration-redis/src/main/java/org/springframework/integration/redis/event/RedisIntegrationEvent.java

@@ -0,0 +1,37 @@
+/*
+ * Copyright 2013 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.integration.redis.event;
+
+import org.springframework.integration.event.IntegrationEvent;
+
+/**
+ * @author Artem Bilan
+ * @since 3.0
+ *
+ */
+@SuppressWarnings("serial")
+public abstract class RedisIntegrationEvent extends IntegrationEvent {
+
+	public RedisIntegrationEvent(Object source) {
+		super(source);
+	}
+
+	public RedisIntegrationEvent(Object source, Throwable cause) {
+		super(source, cause);
+	}
+
+}

spring-integration-redis/src/main/java/org/springframework/integration/redis/event/package-info.java

@@ -0,0 +1,4 @@
+/**
+ * Events generated by the redis module
+ */
+package org.springframework.integration.redis.event;

spring-integration-redis/src/main/java/org/springframework/integration/redis/inbound/RedisQueueMessageDrivenEndpoint.java

@@ -18,8 +18,9 @@ package org.springframework.integration.redis.inbound;
 import java.util.concurrent.Executor;
 import java.util.concurrent.TimeUnit;
 
+import org.springframework.context.ApplicationEventPublisher;
+import org.springframework.context.ApplicationEventPublisherAware;
 import org.springframework.core.task.SimpleAsyncTaskExecutor;
-import org.springframework.data.redis.RedisSystemException;
 import org.springframework.data.redis.connection.RedisConnectionFactory;
 import org.springframework.data.redis.core.BoundListOperations;
 import org.springframework.data.redis.core.RedisTemplate;
@@ -31,6 +32,7 @@ import org.springframework.integration.MessageChannel;
 import org.springframework.integration.MessagingException;
 import org.springframework.integration.channel.MessagePublishingErrorHandler;
 import org.springframework.integration.endpoint.MessageProducerSupport;
+import org.springframework.integration.redis.event.RedisExceptionEvent;
 import org.springframework.integration.support.MessageBuilder;
 import org.springframework.integration.support.channel.BeanFactoryChannelResolver;
 import org.springframework.integration.util.ErrorHandlingTaskExecutor;
@@ -46,13 +48,17 @@ import org.springframework.util.Assert;
  * @since 3.0
  */
 @ManagedResource
-public class RedisQueueMessageDrivenEndpoint extends MessageProducerSupport {
+public class RedisQueueMessageDrivenEndpoint extends MessageProducerSupport implements ApplicationEventPublisherAware {
 
 	public static final long DEFAULT_RECEIVE_TIMEOUT = 1000;
 
+	public static final long DEFAULT_RECOVERY_INTERVAL = 5000;
+
 	private final BoundListOperations<String, byte[]> boundListOperations;
 
-	private MessageChannel errorChannel;
+	private volatile ApplicationEventPublisher applicationEventPublisher;
+
+	private volatile MessageChannel errorChannel;
 
 	private volatile Executor taskExecutor;
 
@@ -62,6 +68,8 @@ public class RedisQueueMessageDrivenEndpoint extends MessageProducerSupport {
 
 	private volatile long receiveTimeout = DEFAULT_RECEIVE_TIMEOUT;
 
+	private volatile long recoveryInterval = DEFAULT_RECOVERY_INTERVAL;
+
 	private volatile boolean active;
 
 	private volatile boolean listening;
@@ -81,6 +89,11 @@ public class RedisQueueMessageDrivenEndpoint extends MessageProducerSupport {
 		this.boundListOperations = template.boundListOps(queueName);
 	}
 
+	@Override
+	public void setApplicationEventPublisher(ApplicationEventPublisher applicationEventPublisher) {
+		this.applicationEventPublisher = applicationEventPublisher;
+	}
+
 	public void setSerializer(RedisSerializer<?> serializer) {
 		this.serializer = serializer;
 	}
@@ -129,6 +142,10 @@ public class RedisQueueMessageDrivenEndpoint extends MessageProducerSupport {
 		this.errorChannel = errorChannel;
 	}
 
+	public void setRecoveryInterval(long recoveryInterval) {
+		this.recoveryInterval = recoveryInterval;
+	}
+
 	@Override
 	protected void onInit() {
 		super.onInit();
@@ -160,13 +177,12 @@ public class RedisQueueMessageDrivenEndpoint extends MessageProducerSupport {
 		try {
 			value = this.boundListOperations.rightPop(this.receiveTimeout, TimeUnit.MILLISECONDS);
 		}
-		catch (RedisSystemException e) {
-			if (this.active) {
-				throw e;
-			}
-			else {
-				logger.error(e);
-			}
+		catch (Exception e) {
+			logger.error("Failed to execute listening task. Will attempt to resubmit in " + this.recoveryInterval + " milliseconds.", e);
+			this.listening = false;
+			this.sleepBeforeRecoveryAttempt();
+			this.publishException(e);
+			return;
 		}
 
 		if (value != null) {
@@ -200,6 +216,32 @@ public class RedisQueueMessageDrivenEndpoint extends MessageProducerSupport {
 		}
 	}
 
+	/**
+	 * Sleep according to the specified recovery interval.
+	 * Called between recovery attempts.
+	 */
+	private void sleepBeforeRecoveryAttempt() {
+		if (this.recoveryInterval > 0) {
+			try {
+				Thread.sleep(this.recoveryInterval);
+			}
+			catch (InterruptedException e) {
+				logger.debug("Thread interrupted while sleeping the recovery interval");
+			}
+		}
+	}
+
+	private void publishException(Exception e) {
+		if (this.applicationEventPublisher != null) {
+			this.applicationEventPublisher.publishEvent(new RedisExceptionEvent(this, e));
+		}
+		else {
+			if (logger.isDebugEnabled()) {
+				logger.debug("No application event publisher for exception: " + e.getMessage());
+			}
+		}
+	}
+
 	private void restart() {
 		this.taskExecutor.execute(new ListenerTask());
 	}

spring-integration-redis/src/main/java/org/springframework/integration/redis/outbound/RedisQueueOutboundChannelAdapter.java

@@ -43,7 +43,7 @@ public class RedisQueueOutboundChannelAdapter extends AbstractMessageHandler imp
 
 	private final Expression queueNameExpression;
 
-	private EvaluationContext evaluationContext;
+	private volatile EvaluationContext evaluationContext;
 
 	private volatile boolean extractPayload = true;
 

spring-integration-redis/src/test/java/org/springframework/integration/redis/inbound/RedisQueueMessageDrivenEndpointTests.java

@@ -18,9 +18,13 @@ package org.springframework.integration.redis.inbound;
 
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertSame;
 import static org.junit.Assert.assertThat;
 
+import java.util.ArrayList;
+import java.util.Arrays;
 import java.util.Date;
+import java.util.List;
 import java.util.UUID;
 
 import org.hamcrest.Matchers;
@@ -29,7 +33,13 @@ import org.junit.runner.RunWith;
 import org.mockito.Mockito;
 
 import org.springframework.beans.factory.BeanFactory;
+import org.springframework.beans.factory.DisposableBean;
+import org.springframework.beans.factory.InitializingBean;
 import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.context.ApplicationEvent;
+import org.springframework.context.ApplicationEventPublisher;
+import org.springframework.data.redis.RedisConnectionFailureException;
+import org.springframework.data.redis.RedisSystemException;
 import org.springframework.data.redis.connection.RedisConnectionFactory;
 import org.springframework.data.redis.core.RedisTemplate;
 import org.springframework.data.redis.core.StringRedisTemplate;
@@ -40,7 +50,9 @@ import org.springframework.integration.MessageChannel;
 import org.springframework.integration.MessagingException;
 import org.springframework.integration.channel.QueueChannel;
 import org.springframework.integration.core.PollableChannel;
+import org.springframework.integration.event.IntegrationEvent;
 import org.springframework.integration.message.ErrorMessage;
+import org.springframework.integration.redis.event.RedisExceptionEvent;
 import org.springframework.integration.redis.rules.RedisAvailable;
 import org.springframework.integration.redis.rules.RedisAvailableTests;
 import org.springframework.integration.support.MessageBuilder;
@@ -68,7 +80,6 @@ public class RedisQueueMessageDrivenEndpointTests extends RedisAvailableTests {
 	@Autowired
 	private PollableChannel symmetricalOutputChannel;
 
-
 	@Test
 	@RedisAvailable
 	@SuppressWarnings("unchecked")
@@ -158,7 +169,6 @@ public class RedisQueueMessageDrivenEndpointTests extends RedisAvailableTests {
 		assertThat(((Exception) receive.getPayload()).getCause().getMessage(),
 				Matchers.containsString("java.lang.String cannot be cast to org.springframework.integration.Message"));
 
-
 		endpoint.stop();
 	}
 
@@ -194,4 +204,60 @@ public class RedisQueueMessageDrivenEndpointTests extends RedisAvailableTests {
 		assertEquals(payload, receive.getPayload());
 	}
 
+	@Test
+	@RedisAvailable
+	@SuppressWarnings("unchecked")
+	public void testInt3196Recovery() throws Exception {
+		String queueName = "test.si.Int3196Recovery";
+		QueueChannel channel = new QueueChannel();
+
+		final List<ApplicationEvent> exceptionEvents = new ArrayList<ApplicationEvent>();
+
+		RedisQueueMessageDrivenEndpoint endpoint = new RedisQueueMessageDrivenEndpoint(queueName, this.connectionFactory);
+		endpoint.setBeanFactory(Mockito.mock(BeanFactory.class));
+		endpoint.setApplicationEventPublisher(new ApplicationEventPublisher() {
+
+			@Override
+			public void publishEvent(ApplicationEvent event) {
+				exceptionEvents.add(event);
+			}
+		});
+		endpoint.setOutputChannel(channel);
+		endpoint.setReceiveTimeout(100);
+		endpoint.setRecoveryInterval(200);
+		endpoint.afterPropertiesSet();
+		endpoint.start();
+
+		((DisposableBean) this.connectionFactory).destroy();
+
+		Thread.sleep(300);
+
+		assertThat(exceptionEvents.size(), Matchers.greaterThan(0));
+		for (ApplicationEvent exceptionEvent : exceptionEvents) {
+			assertThat(exceptionEvent, Matchers.instanceOf(RedisExceptionEvent.class));
+			assertSame(endpoint, exceptionEvent.getSource());
+			assertThat(((IntegrationEvent) exceptionEvent).getCause().getClass(),
+					Matchers.isIn(Arrays.<Class<? extends Throwable>> asList(RedisSystemException.class, RedisConnectionFailureException.class)));
+		}
+
+		((InitializingBean) this.connectionFactory).afterPropertiesSet();
+
+		RedisTemplate<String, Object> redisTemplate = new RedisTemplate<String, Object>();
+		redisTemplate.setConnectionFactory(this.getConnectionFactoryForTest());
+		redisTemplate.setEnableDefaultSerializer(false);
+		redisTemplate.setKeySerializer(new StringRedisSerializer());
+		redisTemplate.setValueSerializer(new JdkSerializationRedisSerializer());
+		redisTemplate.afterPropertiesSet();
+
+		String payload = "testing";
+
+		redisTemplate.boundListOps(queueName).leftPush(payload);
+
+		Message<?> receive = channel.receive(1000);
+		assertNotNull(receive);
+		assertEquals(payload, receive.getPayload());
+
+		endpoint.stop();
+	}
+
 }