DATAREDIS-830 - Release Pub/Sub connection when closing LettuceSubscription.

We now release the native connection back to the connection provider when LettuceSubscription is closed. Previously, we just closed the connection which interfered with pooling as pooling connection providers still had a reference on the connection.

Original Pull Request: spring-projects#341

Author: mp911de

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

@@ -752,7 +752,7 @@ public void pSubscribe(MessageListener listener, byte[]... patterns) {
 			throw new UnsupportedOperationException();
 		}
 		try {
-			subscription = new LettuceSubscription(listener, switchToPubSub());
+			subscription = new LettuceSubscription(listener, switchToPubSub(), connectionProvider);
 			subscription.pSubscribe(patterns);
 		} catch (Exception ex) {
 			throw convertLettuceAccessException(ex);
@@ -767,7 +767,7 @@ public void subscribe(MessageListener listener, byte[]... channels) {
 			throw new UnsupportedOperationException();
 		}
 		try {
-			subscription = new LettuceSubscription(listener, switchToPubSub());
+			subscription = new LettuceSubscription(listener, switchToPubSub(), connectionProvider);
 			subscription.subscribe(channels);
 
 		} catch (Exception ex) {

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

@@ -25,16 +25,19 @@
  * Message subscription on top of Lettuce.
  *
  * @author Costin Leau
+ * @author Mark Paluch
  */
 class LettuceSubscription extends AbstractSubscription {
 
 	final StatefulRedisPubSubConnection<byte[], byte[]> pubsub;
 	private LettuceMessageListener listener;
+	private final LettuceConnectionProvider connectionProvider;
 
-	LettuceSubscription(MessageListener listener, StatefulRedisPubSubConnection<byte[], byte[]> pubsubConnection) {
+	LettuceSubscription(MessageListener listener, StatefulRedisPubSubConnection<byte[], byte[]> pubsubConnection, LettuceConnectionProvider connectionProvider) {
 		super(listener);
 		this.pubsub = pubsubConnection;
 		this.listener = new LettuceMessageListener(listener);
+		this.connectionProvider = connectionProvider;
 
 		pubsub.addListener(this.listener);
 	}
@@ -47,7 +50,8 @@ protected void doClose() {
 			pubsub.sync().punsubscribe(new byte[0]);
 		}
 		pubsub.removeListener(this.listener);
-		pubsub.close();
+		
+		connectionProvider.release(pubsub);
 	}
 
 	protected void doPsubscribe(byte[]... patterns) {

src/test/java/org/springframework/data/redis/connection/lettuce/LettuceSubscriptionTests.java

@@ -34,6 +34,7 @@
  *
  * @author Jennifer Hickey
  * @author Christoph Strobl
+ * @author Mark Paluch
  */
 public class LettuceSubscriptionTests {
 
@@ -45,16 +46,19 @@ public class LettuceSubscriptionTests {
 
 	private RedisPubSubCommands<byte[], byte[]> asyncCommands;
 
+	private LettuceConnectionProvider connectionProvider;
+
 	@SuppressWarnings("unchecked")
 	@Before
 	public void setUp() {
 
 		pubsub = Mockito.mock(StatefulRedisPubSubConnection.class);
 		listener = Mockito.mock(MessageListener.class);
 		asyncCommands = Mockito.mock(RedisPubSubCommands.class);
+		connectionProvider = Mockito.mock(LettuceConnectionProvider.class);
 
 		Mockito.when(pubsub.sync()).thenReturn(asyncCommands);
-		subscription = new LettuceSubscription(listener, pubsub);
+		subscription = new LettuceSubscription(listener, pubsub, connectionProvider);
 	}
 
 	@Test
@@ -64,7 +68,7 @@ public void testUnsubscribeAllAndClose() {
 		verify(asyncCommands, times(1)).unsubscribe(new byte[][] { "a".getBytes() });
 		verify(asyncCommands, never()).unsubscribe(new byte[0]);
 		verify(asyncCommands, never()).punsubscribe(new byte[0]);
-		verify(pubsub).close();
+		verify(connectionProvider).release(pubsub);
 		verify(pubsub).removeListener(any(LettuceMessageListener.class));
 		assertFalse(subscription.isAlive());
 		assertTrue(subscription.getChannels().isEmpty());
@@ -94,7 +98,7 @@ public void testUnsubscribeChannelAndClose() {
 		verify(asyncCommands, times(1)).unsubscribe(channel);
 		verify(asyncCommands, never()).unsubscribe(new byte[0]);
 		verify(asyncCommands, never()).punsubscribe(new byte[0]);
-		verify(pubsub).close();
+		verify(connectionProvider).release(pubsub);
 		verify(pubsub).removeListener(any(LettuceMessageListener.class));
 		assertFalse(subscription.isAlive());
 		assertTrue(subscription.getChannels().isEmpty());
@@ -167,7 +171,7 @@ public void testUnsubscribeAllNoChannels() {
 	public void testUnsubscribeNotAlive() {
 		subscription.subscribe(new byte[][] { "a".getBytes() });
 		subscription.unsubscribe();
-		verify(pubsub, times(1)).close();
+		verify(connectionProvider, times(1)).release(pubsub);
 		verify(pubsub, times(1)).removeListener(any(LettuceMessageListener.class));
 		assertFalse(subscription.isAlive());
 		subscription.unsubscribe();
@@ -192,7 +196,7 @@ public void testPUnsubscribeAllAndClose() {
 		verify(asyncCommands, never()).punsubscribe(new byte[0]);
 		verify(asyncCommands, times(1)).punsubscribe(new byte[][] { "a*".getBytes() });
 		assertFalse(subscription.isAlive());
-		verify(pubsub).close();
+		verify(connectionProvider).release(pubsub);
 		verify(pubsub).removeListener(any(LettuceMessageListener.class));
 		assertTrue(subscription.getChannels().isEmpty());
 		assertTrue(subscription.getPatterns().isEmpty());
@@ -221,7 +225,7 @@ public void testPUnsubscribeAndClose() {
 		verify(asyncCommands, never()).unsubscribe(new byte[0]);
 		verify(asyncCommands, never()).punsubscribe(new byte[0]);
 		verify(asyncCommands, times(1)).punsubscribe(pattern);
-		verify(pubsub).close();
+		verify(connectionProvider).release(pubsub);
 		verify(pubsub).removeListener(any(LettuceMessageListener.class));
 		assertFalse(subscription.isAlive());
 		assertTrue(subscription.getChannels().isEmpty());
@@ -296,7 +300,7 @@ public void testPUnsubscribeNotAlive() {
 		subscription.unsubscribe();
 		assertFalse(subscription.isAlive());
 		subscription.pUnsubscribe();
-		verify(pubsub, times(1)).close();
+		verify(connectionProvider, times(1)).release(pubsub);
 		verify(pubsub, times(1)).removeListener(any(LettuceMessageListener.class));
 		verify(asyncCommands, times(1)).unsubscribe(new byte[][] { "a".getBytes() });
 		verify(asyncCommands, never()).unsubscribe(new byte[0]);