properly close the underlying pubsub connection on (p)unsubscribe all

DATAREDIS-113

Author: Costin Leau

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

@@ -65,7 +65,6 @@ public class LettuceConnection implements RedisConnection {
 
 private final com.lambdaworks.redis.RedisAsyncConnection<byte[], byte[]> asyncConn;
 private final com.lambdaworks.redis.RedisConnection<byte[], byte[]> con;
-private RedisPubSubConnection<byte[], byte[]> pubsub;
 private final RedisCodec<byte[], byte[]> codec = LettuceUtils.CODEC;
 private final long timeout;
 
@@ -157,7 +156,7 @@ public boolean isClosed() {
 }
 
 public RedisAsyncConnection<byte[], byte[]> getNativeConnection() {
-return (pubsub != null ? pubsub : asyncConn);
+return (subscription != null ? subscription.pubsub : asyncConn);
 }
 
 

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

@@ -28,7 +28,7 @@
  */
 class LettuceSubscription extends AbstractSubscription {
 
-private final RedisPubSubConnection<byte[], byte[]> pubsub;
+final RedisPubSubConnection<byte[], byte[]> pubsub;
 private LettuceMessageListener listener;
 
 LettuceSubscription(MessageListener listener, RedisPubSubConnection<byte[], byte[]> pubsubConnection) {
@@ -43,6 +43,7 @@ protected void doClose() {
 pubsub.unsubscribe(new byte[0]);
 pubsub.punsubscribe(new byte[0]);
 pubsub.removeListener(this.listener);
+pubsub.close();
 }