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

Fix SubscriptionTask leaving open connections

Browse Source 
DATAREDIS-124

- Close connection when SubscriptionTask is canceled and subscribing thread is unblocked

- Ensure non-blocking Lettuce and SRP only attempt
to close subscriptions if active

- Cleanup unecessary synchronization between
SubscriptionTask methods, now that connection is
no longer set to null

- Cleanup unused monitor notify when SubscriptionTask thread ends
This commit is contained in:
Jennifer Hickey
2013-04-03 16:07:57 -07:00
parent 2291edae67
commit 6f1739a681
4 changed files with 196 additions and 56 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

@@ -147,7 +147,9 @@ public class LettuceConnection implements RedisConnection {
}
if (subscription != null) {
subscription.doClose();
if(subscription.isAlive()) {
subscription.doClose();
}
subscription = null;
}
}

4
src/main/java/org/springframework/data/redis/connection/srp/SrpConnection.java
View File

@@ -141,7 +141,9 @@ public class SrpConnection implements RedisConnection {
queue.remove(this);
if (subscription != null) {
subscription.doClose();
if(subscription.isAlive()) {
subscription.doClose();
}
subscription = null;
}

98
src/main/java/org/springframework/data/redis/listener/RedisMessageListenerContainer.java
View File

@@ -664,17 +664,15 @@ public class RedisMessageListenerContainer implements InitializingBean, Disposab
// wait 3 rounds for subscription to be initialized
for (int i = 0; i < ROUNDS && !done; i++) {
if (connection != null) {
synchronized (localMonitor) {
if (connection != null && connection.isSubscribed()) {
if (connection.isSubscribed()) {
done = true;
connection.getSubscription().pSubscribe(unwrap(patternMapping.keySet()));
}
else {
try {
Thread.sleep(WAIT);
} catch (InterruptedException ex) {
done = true;
connection.getSubscription().pSubscribe(unwrap(patternMapping.keySet()));
}
else {
try {
Thread.sleep(WAIT);
} catch (InterruptedException ex) {
done = true;
}
}
}
}
@@ -683,8 +681,9 @@ public class RedisMessageListenerContainer implements InitializingBean, Disposab
}
private volatile RedisConnection connection;
private boolean subscriptionDone = false;
private final Object localMonitor = new Object();
private long subscriptionWait = TimeUnit.SECONDS.toMillis(5);
public boolean isLongLived() {
return true;
@@ -698,7 +697,7 @@ public class RedisMessageListenerContainer implements InitializingBean, Disposab
throw new IllegalStateException("Retrieved connection is already subscribed; aborting listening");
}
// NB: each Xsubscribe call blocks
// NB: some drivers' Xsubscribe calls block
synchronized (monitor) {
monitor.notify();
}
@@ -715,14 +714,12 @@ public class RedisMessageListenerContainer implements InitializingBean, Disposab
else {
connection.pSubscribe(new DispatchMessageListener(), unwrap(patternMapping.keySet()));
}
} finally {
// this block is executed once the subscription has ended
// meaning cleanup is required
// done with the thread, app can be destroyed
synchronized (monitor) {
monitor.notify();
// this block is executed once the subscription thread has ended, this may or may not mean
// the connection has been unsubscribed, depending on driver
synchronized (localMonitor) {
subscriptionDone = true;
localMonitor.notify();
}
}
}
@@ -752,14 +749,23 @@ public class RedisMessageListenerContainer implements InitializingBean, Disposab
logger.trace("Cancelling Redis subscription...");
}
if (connection != null) {
synchronized (localMonitor) {
if (connection != null) {
Subscription sub = connection.getSubscription();
if (sub != null) {
sub.pUnsubscribe();
sub.unsubscribe();
Subscription sub = connection.getSubscription();
if (sub != null) {
logger.trace("Unsubscribing from all channels");
sub.pUnsubscribe();
sub.unsubscribe();
synchronized(localMonitor) {
if(!subscriptionDone) {
try {
localMonitor.wait(subscriptionWait);
} catch (InterruptedException e) {
// Stop waiting
}
}
if(subscriptionDone) {
logger.trace("Closing connection");
connection.close();
}
}
}
}
@@ -768,13 +774,9 @@ public class RedisMessageListenerContainer implements InitializingBean, Disposab
void subscribeChannel(byte[]... channels) {
if (channels != null && channels.length > 0) {
if (connection != null) {
synchronized (localMonitor) {
if (connection != null) {
Subscription sub = connection.getSubscription();
if (sub != null) {
sub.subscribe(channels);
}
}
Subscription sub = connection.getSubscription();
if (sub != null) {
sub.subscribe(channels);
}
}
}
@@ -783,13 +785,9 @@ public class RedisMessageListenerContainer implements InitializingBean, Disposab
void subscribePattern(byte[]... patterns) {
if (patterns != null && patterns.length > 0) {
if (connection != null) {
synchronized (localMonitor) {
if (connection != null) {
Subscription sub = connection.getSubscription();
if (sub != null) {
sub.pSubscribe(patterns);
}
}
Subscription sub = connection.getSubscription();
if (sub != null) {
sub.pSubscribe(patterns);
}
}
}
@@ -798,13 +796,9 @@ public class RedisMessageListenerContainer implements InitializingBean, Disposab
void unsubscribeChannel(byte[]... channels) {
if (channels != null && channels.length > 0) {
if (connection != null) {
synchronized (localMonitor) {
if (connection != null) {
Subscription sub = connection.getSubscription();
if (sub != null) {
sub.unsubscribe(channels);
}
}
Subscription sub = connection.getSubscription();
if (sub != null) {
sub.unsubscribe(channels);
}
}
}
@@ -813,13 +807,9 @@ public class RedisMessageListenerContainer implements InitializingBean, Disposab
void unsubscribePattern(byte[]... patterns) {
if (patterns != null && patterns.length > 0) {
if (connection != null) {
synchronized (localMonitor) {
if (connection != null) {
Subscription sub = connection.getSubscription();
if (sub != null) {
sub.pUnsubscribe(patterns);
}
}
Subscription sub = connection.getSubscription();
if (sub != null) {
sub.pUnsubscribe(patterns);
}
}
}

146
src/test/java/org/springframework/data/redis/listener/SubscriptionConnectionTests.java Normal file
View File

@@ -0,0 +1,146 @@
/*
* Copyright 2011-2013 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.data.redis.listener;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
import org.junit.After;
import org.junit.Before;
import org.junit.Test;
import org.springframework.core.task.SimpleAsyncTaskExecutor;
import org.springframework.core.task.SyncTaskExecutor;
import org.springframework.data.redis.SettingsUtils;
import org.springframework.data.redis.connection.MessageListener;
import org.springframework.data.redis.connection.RedisConnection;
import org.springframework.data.redis.connection.jedis.JedisConnectionFactory;
import org.springframework.data.redis.listener.adapter.MessageListenerAdapter;
/**
* Integration tests confirming that {@link RedisMessageListenerContainer}
* closes connections after unsubscribing
*
* @author Jennifer Hickey
*
*/
public class SubscriptionConnectionTests {
private static final String CHANNEL = "pubsub::test";
private JedisConnectionFactory connectionFactory;
private List<RedisMessageListenerContainer> containers = new ArrayList<RedisMessageListenerContainer>();
private final Object handler = new Object() {
@SuppressWarnings("unused")
public void handleMessage(String message) {
System.out.println(message);
}
};
@Before
public void setUp() {
connectionFactory = new JedisConnectionFactory();
connectionFactory.setUsePool(true);
connectionFactory.setPort(SettingsUtils.getPort());
connectionFactory.setHostName(SettingsUtils.getHost());
connectionFactory.setDatabase(2);
connectionFactory.afterPropertiesSet();
}
@After
public void tearDown() throws Exception {
for (RedisMessageListenerContainer container : containers) {
if (container.isActive()) {
container.destroy();
}
}
connectionFactory.destroy();
}
@Test
public void testStopMessageListenerContainers() throws Exception {
// Grab all 8 connections from the pool. They should be released on
// container stop
for (int i = 0; i < 8; i++) {
RedisMessageListenerContainer container = new RedisMessageListenerContainer();
container.setConnectionFactory(connectionFactory);
container.setBeanName("container" + i);
container.addMessageListener(new MessageListenerAdapter(handler),
Arrays.asList(new ChannelTopic(CHANNEL)));
container.setTaskExecutor(new SyncTaskExecutor());
container.setSubscriptionExecutor(new SimpleAsyncTaskExecutor());
container.afterPropertiesSet();
container.start();
container.stop();
containers.add(container);
}
// verify we can now get a connection from the pool
RedisConnection connection = connectionFactory.getConnection();
connection.close();
}
@Test
public void testRemoveLastListener() {
// Grab all 8 connections from the pool
MessageListener listener = new MessageListenerAdapter(handler);
for (int i = 0; i < 8; i++) {
RedisMessageListenerContainer container = new RedisMessageListenerContainer();
container.setConnectionFactory(connectionFactory);
container.setBeanName("container" + i);
container.addMessageListener(listener, Arrays.asList(new ChannelTopic(CHANNEL)));
container.setTaskExecutor(new SyncTaskExecutor());
container.setSubscriptionExecutor(new SimpleAsyncTaskExecutor());
container.afterPropertiesSet();
container.start();
containers.add(container);
}
// Removing the sole listener from the container should free up a
// connection
containers.get(0).removeMessageListener(listener);
// verify we can now get a connection from the pool
RedisConnection connection = connectionFactory.getConnection();
connection.close();
}
@Test
public void testStopListening() {
// Grab all 8 connections from the pool.
MessageListener listener = new MessageListenerAdapter(handler);
for (int i = 0; i < 8; i++) {
RedisMessageListenerContainer container = new RedisMessageListenerContainer();
container.setConnectionFactory(connectionFactory);
container.setBeanName("container" + i);
container.addMessageListener(listener, Arrays.asList(new ChannelTopic(CHANNEL)));
container.setTaskExecutor(new SyncTaskExecutor());
container.setSubscriptionExecutor(new SimpleAsyncTaskExecutor());
container.afterPropertiesSet();
container.start();
containers.add(container);
}
// Unsubscribe all listeners from all topics, freeing up a connection
containers.get(0).removeMessageListener(null, Arrays.asList(new Topic[] {}));
// verify we can now get a connection from the pool
RedisConnection connection = connectionFactory.getConnection();
connection.close();
}
}