Fix corrupt subscription connections return to pool

with Redis 2.6

Redis 2.6 responds to unsubscribe requests even if
not subscribed. This results in unread data on the
InputStream on Subscription close if extra 
unsubscribe requests are issued. This causes 
problems if connection is reused 
(as in the case of Jedis pool).

- Only unsubscribe on Subscription close if still
subscribed

-  Use dedicated connection for RJC subscriptions,
as RJC makes extra unsubscribe calls on subscriber
close

src/main/java/org/springframework/data/redis/connection/jedis/JedisConnection.java

@@ -2437,10 +2437,4 @@ public class JedisConnection implements RedisConnection {
 			throw convertJedisAccessException(ex);
 		}
 	}
-
-	private void checkSubscription() {
-		if (isSubscribed()) {
-			throw new RedisSubscribedConnectionException("Cannot execute command - connection is subscribed");
-		}
-	}
 }

src/main/java/org/springframework/data/redis/connection/jedis/JedisSubscription.java

@@ -36,8 +36,12 @@ class JedisSubscription extends AbstractSubscription {
 
 	
 	protected void doClose() {
-		jedisPubSub.unsubscribe();
-		jedisPubSub.punsubscribe();
+		if(!getChannels().isEmpty()) {
+			jedisPubSub.unsubscribe();
+		}
+		if(!getPatterns().isEmpty()) {
+			jedisPubSub.punsubscribe();
+		}
 	}
 
 	

src/main/java/org/springframework/data/redis/connection/lettuce/LettuceSubscription.java

@@ -40,8 +40,12 @@ class LettuceSubscription extends AbstractSubscription {
 	}
 
 	protected void doClose() {
-		pubsub.unsubscribe(new byte[0]);
-		pubsub.punsubscribe(new byte[0]);
+		if(!getChannels().isEmpty()) {
+			pubsub.unsubscribe(new byte[0]);
+		}
+		if(!getPatterns().isEmpty()) {
+			pubsub.punsubscribe(new byte[0]);
+		}
 		pubsub.removeListener(this.listener);
 		pubsub.close();
 	}

src/main/java/org/springframework/data/redis/connection/rjc/RjcConnection.java

@@ -59,11 +59,38 @@ public class RjcConnection implements RedisConnection {
 	private volatile RjcSubscription subscription;
 	private volatile RedisNodeSubscriber subscriber;
 
+	/**
+	 * Constructs a new RjcConnection with a {@link RedisNodeSubscriber} that
+	 * re-uses the passed connection. Not recommended for pooled connections, as
+	 * {@link RedisNodeSubscriber} may leave the Reply InputStream in an
+	 * inconsistent state due to extra unsubscribe calls in close()
+	 *
+	 * @param connection
+	 *            The connection to use
+	 * @param dbIndex
+	 *            The index of the database to use
+	 */
 	public RjcConnection(org.idevlab.rjc.ds.RedisConnection connection, int dbIndex) {
+		this(connection, dbIndex, new RedisNodeSubscriber(new SingleDataSource(
+				new CloseSuppressingRjcConnection(connection))));
+	}
+
+	/**
+	 * Constructs a new RjcConnection with a dedicated
+	 * {@link RedisNodeSubscriber}
+	 *
+	 * @param connection
+	 *            The connection to use
+	 * @param dbIndex
+	 *            The index of the database to use
+	 * @param subscriber
+	 *            The connection to use for subscribe calls
+	 */
+	public RjcConnection(org.idevlab.rjc.ds.RedisConnection connection, int dbIndex,
+			RedisNodeSubscriber subscriber) {
 		SingleDataSource connectionDataSource = new SingleDataSource(connection);
 		session = new SessionFactoryImpl(connectionDataSource).create();
-		subscriber = new RedisNodeSubscriber();
-		subscriber.setDataSource(new SingleDataSource(new CloseSuppressingRjcConnection(connection)));
+		this.subscriber = subscriber;
 		client = new Client(connection);
 		this.connection = connection;
 
@@ -2153,10 +2180,4 @@ public class RjcConnection implements RedisConnection {
 			throw convertRjcAccessException(ex);
 		}
 	}
-
-	private void checkSubscription() {
-		if (isSubscribed()) {
-			throw new RedisSubscribedConnectionException("Cannot execute command - connection is subscribed");
-		}
-	}
 }

src/main/java/org/springframework/data/redis/connection/rjc/RjcConnectionFactory.java

@@ -20,6 +20,7 @@ import org.apache.commons.logging.LogFactory;
 import org.idevlab.rjc.ds.DataSource;
 import org.idevlab.rjc.ds.PoolableDataSource;
 import org.idevlab.rjc.ds.SimpleDataSource;
+import org.idevlab.rjc.message.RedisNodeSubscriber;
 import org.idevlab.rjc.protocol.Protocol;
 import org.springframework.beans.factory.DisposableBean;
 import org.springframework.beans.factory.InitializingBean;
@@ -45,6 +46,7 @@ public class RjcConnectionFactory implements InitializingBean, DisposableBean, R
 	private boolean usePool = true;
 	private int dbIndex = 0;
 	private DataSource dataSource;
+	private DataSource subscriptionDataSource;
 
 
 	/**
@@ -71,6 +73,7 @@ public class RjcConnectionFactory implements InitializingBean, DisposableBean, R
 		else {
 			dataSource = new SimpleDataSource(hostName, port, timeout, password);
 		}
+		subscriptionDataSource = new SimpleDataSource(hostName, port, timeout, password);
 	}
 
 	public void destroy() {
@@ -86,7 +89,8 @@ public class RjcConnectionFactory implements InitializingBean, DisposableBean, R
 
 	
 	public RedisConnection getConnection() {
-		return postProcessConnection(new RjcConnection(dataSource.getConnection(), dbIndex));
+		return postProcessConnection(new RjcConnection(dataSource.getConnection(), dbIndex,
+				new RedisNodeSubscriber(subscriptionDataSource)));
 	}
 
 	/**

src/main/java/org/springframework/data/redis/connection/rjc/RjcSubscription.java

@@ -37,7 +37,9 @@ class RjcSubscription extends AbstractSubscription {
 
 	
 	protected void doClose() {
-		subscriber.close();
+		if(!getChannels().isEmpty() || !getPatterns().isEmpty()) {
+			subscriber.close();
+		}
 	}
 
 	
@@ -57,6 +59,6 @@ class RjcSubscription extends AbstractSubscription {
 
 	
 	protected void doUnsubscribe(boolean all, byte[]... channels) {
-		subscriber.punsubscribe(RjcUtils.decodeMultiple(channels));
+		subscriber.unsubscribe(RjcUtils.decodeMultiple(channels));
 	}
 }

src/main/java/org/springframework/data/redis/connection/srp/SrpSubscription.java

@@ -40,8 +40,12 @@ class SrpSubscription extends AbstractSubscription {
 	}
 
 	protected void doClose() {
-		client.unsubscribe((Object[]) null);
-		client.punsubscribe((Object[]) null);
+		if(!getChannels().isEmpty()) {
+			client.unsubscribe((Object[]) null);
+		}
+		if(!getPatterns().isEmpty()) {
+			client.punsubscribe((Object[]) null);
+		}
 		client.removeListener(this.listener);
 	}
 

src/main/java/org/springframework/data/redis/connection/util/AbstractSubscription.java

@@ -163,8 +163,9 @@ public abstract class AbstractSubscription implements Subscription {
 		// shortcut for unsubscribing all patterns
 		if (ObjectUtils.isEmpty(patts)) {
 			if (!this.patterns.isEmpty()) {
-				patts = getPatterns().toArray(new byte[this.patterns.size()][]);
 				synchronized (this.patterns) {
+					patts = getPatterns().toArray(new byte[this.patterns.size()][]);
+					doPUnsubscribe(true, patts);
 					this.patterns.clear();
 				}
 			}
@@ -174,14 +175,13 @@ public abstract class AbstractSubscription implements Subscription {
 			}
 		}
 		else {
+			doPUnsubscribe(false, patts);
 			synchronized (this.patterns) {
 				remove(this.patterns, patts);
 			}
 		}
 
-		if (isWorking()) {
-			doPUnsubscribe(this.patterns.isEmpty(), patts);
-		}
+		closeIfUnsubscribed();
 	}
 
 	
@@ -193,8 +193,9 @@ public abstract class AbstractSubscription implements Subscription {
 		// shortcut for unsubscribing all channels
 		if (ObjectUtils.isEmpty(chans)) {
 			if (!this.channels.isEmpty()) {
-				chans = getChannels().toArray(new byte[this.channels.size()][]);
 				synchronized (this.channels) {
+					chans = getChannels().toArray(new byte[this.channels.size()][]);
+					doUnsubscribe(true, chans);
 					this.channels.clear();
 				}
 			}
@@ -204,14 +205,13 @@ public abstract class AbstractSubscription implements Subscription {
 			}
 		}
 		else {
+			doUnsubscribe(false, chans);
 			synchronized (this.channels) {
 				remove(this.channels, chans);
 			}
 		}
 
-		if (isWorking()) {
-			doUnsubscribe(this.channels.isEmpty(), chans);
-		}
+		closeIfUnsubscribed();
 	}
 
 	
@@ -225,12 +225,11 @@ public abstract class AbstractSubscription implements Subscription {
 		}
 	}
 
-	private boolean isWorking() {
+	private void closeIfUnsubscribed() {
 		if (channels.isEmpty() && patterns.isEmpty()) {
 			alive.set(false);
 			doClose();
 		}
-		return isAlive();
 	}