Spring/spring-data-redis
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

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

Browse Source 
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
This commit is contained in:
Mark Paluch
2018-05-09 08:44:20 +02:00
committed by Christoph Strobl
parent 9f01270dd5
commit 96d888b8fa
3 changed files with 19 additions and 11 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -741,7 +741,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);
@@ -756,7 +756,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) {

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

@@ -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<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 @@ class LettuceSubscription extends AbstractSubscription {
pubsub.sync().punsubscribe(new byte[0]);
}
pubsub.removeListener(this.listener);
pubsub.close();
connectionProvider.release(pubsub);
}
protected void doPsubscribe(byte[]... patterns) {

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

@@ -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<byte[], byte[]> 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]);