Resubscribe message listener on connection failure

DATAREDIS-231

src/main/java/org/springframework/data/redis/listener/RedisMessageListenerContainer.java

@@ -34,6 +34,7 @@ import org.springframework.beans.factory.InitializingBean;
 import org.springframework.context.SmartLifecycle;
 import org.springframework.core.task.SimpleAsyncTaskExecutor;
 import org.springframework.core.task.TaskExecutor;
+import org.springframework.data.redis.RedisConnectionFailureException;
 import org.springframework.data.redis.connection.Message;
 import org.springframework.data.redis.connection.MessageListener;
 import org.springframework.data.redis.connection.RedisConnection;
@@ -65,6 +66,7 @@ import org.springframework.util.ErrorHandler;
  * 
  * 
  * @author Costin Leau
+ * @author Way Joke
  */
 public class RedisMessageListenerContainer implements InitializingBean, DisposableBean, BeanNameAware, SmartLifecycle {
 
@@ -116,8 +118,13 @@ public class RedisMessageListenerContainer implements InitializingBean, Disposab
 	private final SubscriptionTask subscriptionTask = new SubscriptionTask();
 
 	private volatile RedisSerializer<String> serializer = new StringRedisSerializer();
-
-
+	
+	// will recovery redis listening after recovery tnterval setting
+	private long recoveryInterval = 3000L;
+	// will recovery redis listening until reach the limit ( -1 will always try again) 
+	private int multipleRetries = -1;
+	// the retried times
+	private int numberOfRetries = 0;
 
 	public void afterPropertiesSet() {
 		if (taskExecutor == null) {
@@ -632,7 +639,45 @@ public class RedisMessageListenerContainer implements InitializingBean, Disposab
 		}
 	}
 
+	/**
+	 * Handle subscription task exception, will recovery redis listening on RedisConnectionFailureExceptions.
+	 * @param t Throwable exception
+	 */
+	protected void refreshConnectionUntilSuccessful(Throwable t) {
+		// we need to specifically deal with RedisConnectionFailureExceptions 
+		// so we aren't attempting to reconnect if other types of failures occur.
+		if (t instanceof RedisConnectionFailureException) {
 
+			if (multipleRetries > 0 && numberOfRetries > multipleRetries) {
+				logger.error("Retry count has reached the upper limit " + multipleRetries);
+				return;
+			}
+
+			subscriptionExecutor.execute(new Runnable() {
+
+				public void run() {
+					if (multipleRetries > 0) {
+						numberOfRetries++;
+					}
+					try {
+						stop();
+					} catch (Throwable ex) {
+						// as we know the stop will throw JedisConnectionException on jedis,
+						// catch Throwable will not need to import any jedis classes.
+						// The exception will be ignored.
+					}
+					logger.error("Restarting subscription task after " + recoveryInterval + "ms.");
+					try {
+						Thread.sleep(recoveryInterval);
+					} catch (InterruptedException e) {
+						return;
+					}
+					start();
+				}
+			});
+		}
+	}
+	
 	/**
 	 * Runnable used for Redis subscription. Implemented as a dedicated class to provide as many hints
 	 * as possible to the underlying thread pool.
@@ -721,6 +766,9 @@ public class RedisMessageListenerContainer implements InitializingBean, Disposab
 				else {
 					connection.pSubscribe(new DispatchMessageListener(), unwrap(patternMapping.keySet()));
 				}
+			} catch (Throwable t) {				
+				logger.error("SubscriptionTask abort with exception:", t);
+				refreshConnectionUntilSuccessful(t);
 			} finally {
 				// this block is executed once the subscription thread has ended, this may or may not mean
 				// the connection has been unsubscribed, depending on driver
@@ -870,4 +918,22 @@ public class RedisMessageListenerContainer implements InitializingBean, Disposab
 			});
 		}
 	}
+
+	/**
+	 * Set the recovery interval value(ms), must be greater than zero
+	 * @param recoveryInterval 
+	 */
+	public void setRecoveryInterval(long recoveryInterval) {
+		if (recoveryInterval > 0) {
+			this.recoveryInterval = recoveryInterval;
+		}
+	}
+
+	/**
+	 * Set the multiple retries times, less than zero on behalf of the infinite.
+	 * @param multipleRetries 
+	 */
+	public void setMultipleRetries(int multipleRetries) {
+		this.multipleRetries = multipleRetries;
+	}
 }