From a2bc45cfa2c4556897788b34614086a701be5c56 Mon Sep 17 00:00:00 2001 From: Mark Paluch Date: Wed, 9 May 2018 08:44:20 +0200 Subject: [PATCH] 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: #341 --- .../connection/lettuce/LettuceConnection.java | 4 ++-- .../lettuce/LettuceSubscription.java | 8 ++++++-- .../lettuce/LettuceSubscriptionTests.java | 18 +++++++++++------- 3 files changed, 19 insertions(+), 11 deletions(-) diff --git a/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceConnection.java b/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceConnection.java index 2e2f08375..7297b48d1 100644 --- a/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceConnection.java +++ b/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceConnection.java @@ -752,7 +752,7 @@ public class LettuceConnection extends AbstractRedisConnection { 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 class LettuceConnection extends AbstractRedisConnection { throw new UnsupportedOperationException(); } try { - subscription = new LettuceSubscription(listener, switchToPubSub()); + subscription = new LettuceSubscription(listener, switchToPubSub(), connectionProvider); subscription.subscribe(channels); } catch (Exception ex) { diff --git a/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceSubscription.java b/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceSubscription.java index a16837863..36a5bb244 100644 --- a/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceSubscription.java +++ b/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceSubscription.java @@ -25,16 +25,19 @@ import org.springframework.data.redis.connection.util.AbstractSubscription; * Message subscription on top of Lettuce. * * @author Costin Leau + * @author Mark Paluch */ class LettuceSubscription extends AbstractSubscription { final StatefulRedisPubSubConnection pubsub; private LettuceMessageListener listener; + private final LettuceConnectionProvider connectionProvider; - LettuceSubscription(MessageListener listener, StatefulRedisPubSubConnection pubsubConnection) { + LettuceSubscription(MessageListener listener, StatefulRedisPubSubConnection pubsubConnection, LettuceConnectionProvider connectionProvider) { super(listener); this.pubsub = pubsubConnection; this.listener = new LettuceMessageListener(listener); + this.connectionProvider = connectionProvider; pubsub.addListener(this.listener); } @@ -47,7 +50,8 @@ class LettuceSubscription extends AbstractSubscription { pubsub.sync().punsubscribe(new byte[0]); } pubsub.removeListener(this.listener); - pubsub.close(); + + connectionProvider.release(pubsub); } protected void doPsubscribe(byte[]... patterns) { diff --git a/src/test/java/org/springframework/data/redis/connection/lettuce/LettuceSubscriptionTests.java b/src/test/java/org/springframework/data/redis/connection/lettuce/LettuceSubscriptionTests.java index 65eb898c9..241e4b018 100644 --- a/src/test/java/org/springframework/data/redis/connection/lettuce/LettuceSubscriptionTests.java +++ b/src/test/java/org/springframework/data/redis/connection/lettuce/LettuceSubscriptionTests.java @@ -34,6 +34,7 @@ import org.springframework.data.redis.connection.RedisInvalidSubscriptionExcepti * * @author Jennifer Hickey * @author Christoph Strobl + * @author Mark Paluch */ public class LettuceSubscriptionTests { @@ -45,6 +46,8 @@ public class LettuceSubscriptionTests { private RedisPubSubCommands asyncCommands; + private LettuceConnectionProvider connectionProvider; + @SuppressWarnings("unchecked") @Before public void setUp() { @@ -52,9 +55,10 @@ public class LettuceSubscriptionTests { 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 class LettuceSubscriptionTests { 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 class LettuceSubscriptionTests { 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 class LettuceSubscriptionTests { 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 class LettuceSubscriptionTests { 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 class LettuceSubscriptionTests { 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 class LettuceSubscriptionTests { 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]);