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

improve conn handling when all subs are removed

Browse Source 
DATAREDIS-107
RMLC properly closes
This commit is contained in:
Costin Leau
2013-01-25 10:57:04 +02:00
parent ecf30d868b
commit e3946af60c
4 changed files with 75 additions and 63 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -34,6 +34,7 @@ import org.springframework.beans.factory.InitializingBean;
import org.springframework.context.SmartLifecycle;
import org.springframework.core.task.SimpleAsyncTaskExecutor;
import org.springframework.core.task.TaskExecutor;
import org.springframework.dao.DataAccessException;
import org.springframework.data.redis.connection.Message;
import org.springframework.data.redis.connection.MessageListener;
import org.springframework.data.redis.connection.RedisConnection;
@@ -63,6 +64,7 @@ import org.springframework.util.ErrorHandler;
* Adding and removing listeners at the same time has undefined results. It is strongly recommended to synchronize/order these
* methods accordingly.
*
*
* @author Costin Leau
*/
public class RedisMessageListenerContainer implements InitializingBean, DisposableBean, BeanNameAware, SmartLifecycle {
@@ -207,18 +209,7 @@ public class RedisMessageListenerContainer implements InitializingBean, Disposab
public void stop() {
if (isRunning()) {
running = false;
synchronized (monitor) {
boolean shouldWait = listening;
subscriptionTask.cancel();
listening = false;
if (shouldWait) {
try {
monitor.wait(initWait);
} catch (InterruptedException ex) {
// stop waiting
}
}
}
subscriptionTask.cancel();
}
if (logger.isDebugEnabled()) {
@@ -554,7 +545,6 @@ public class RedisMessageListenerContainer implements InitializingBean, Disposab
// check stop listening case
if (listener == null && CollectionUtils.isEmpty(topics)) {
subscriptionTask.cancel();
logger.debug("Stopped listening for Redis messages");
return;
}
@@ -593,8 +583,14 @@ public class RedisMessageListenerContainer implements InitializingBean, Disposab
}
}
// double check whether there are still subscriptions available otherwise cancel the connection
// as most drivers forfeit the connection on unsubscribe
if (listenerTopics.isEmpty()) {
subscriptionTask.cancel();
}
// check the current listening state
if (listening) {
else if (listening) {
subscriptionTask.unsubscribeChannel(channelsToRemove.toArray(new byte[channelsToRemove.size()][]));
subscriptionTask.unsubscribePattern(patternsToRemove.toArray(new byte[patternsToRemove.size()][]));
}
@@ -717,22 +713,10 @@ public class RedisMessageListenerContainer implements InitializingBean, Disposab
// this block is executed once the subscription has ended
// meaning cleanup is required
listening = false;
if (connection != null) {
synchronized (localMonitor) {
if (connection != null) {
connection.close();
connection = null;
}
}
}
// done with the thread, app can be destroyed
synchronized (monitor) {
monitor.notify();
}
}
}
@@ -755,6 +739,12 @@ public class RedisMessageListenerContainer implements InitializingBean, Disposab
if (!listening) {
return;
}
boolean shouldWait = listening;
listening = false;
if (logger.isTraceEnabled()) {
logger.trace("Cancelling Redis subscription...");
}
if (connection != null) {
synchronized (localMonitor) {
if (connection != null) {
@@ -763,6 +753,24 @@ public class RedisMessageListenerContainer implements InitializingBean, Disposab
sub.pUnsubscribe();
sub.unsubscribe();
}
}
}
}
}
private void cleanUpConnection() {
listening = false;
if (connection != null) {
synchronized (localMonitor) {
if (connection != null) {
RedisConnection con = connection;
connection = null;
try {
con.close();
} catch (DataAccessException ex) {
logger.trace("Closing connection threw", ex);
}
}
}
}
@@ -866,5 +874,4 @@ public class RedisMessageListenerContainer implements InitializingBean, Disposab
});
}
}
}

41
src/test/java/org/springframework/data/redis/listener/PubSubResubscribeTests.java
View File

@@ -17,11 +17,14 @@
package org.springframework.data.redis.listener;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertNull;
import static org.junit.Assert.assertTrue;
import java.util.ArrayList;
import java.util.LinkedHashSet;
import java.util.List;
import java.util.Set;
import java.util.concurrent.BlockingDeque;
import java.util.concurrent.LinkedBlockingDeque;
import java.util.concurrent.TimeUnit;
@@ -30,6 +33,7 @@ import org.junit.After;
import org.junit.AfterClass;
import org.junit.Before;
import org.junit.Test;
import org.springframework.beans.factory.DisposableBean;
import org.springframework.core.task.SimpleAsyncTaskExecutor;
import org.springframework.core.task.SyncTaskExecutor;
import org.springframework.data.redis.ConnectionFactoryTracker;
@@ -104,6 +108,8 @@ public class PubSubResubscribeTests {
@After
public void tearDown() throws Exception {
container.destroy();
((DisposableBean) factory).destroy();
Thread.sleep(1000);
}
@@ -150,4 +156,37 @@ public class PubSubResubscribeTests {
assertTrue(msgs.contains(payload1));
assertTrue(msgs.contains(payload2));
}
}
@Test
public void testContainerChannelResubscribe() throws Exception {
String payload1 = "do";
String payload2 = "re mi";
String anotherPayload1 = "od";
String anotherPayload2 = "mi er";
String ANOTHER_CHANNEL = "pubsub::test::extra";
// bind listener on another channel
container.removeMessageListener(null, new ChannelTopic(CHANNEL));
container.addMessageListener(adapter, new ChannelTopic(ANOTHER_CHANNEL));
assertEquals(ZERO, template.convertAndSend(CHANNEL, payload1));
assertEquals(ZERO, template.convertAndSend(CHANNEL, payload2));
assertEquals(ONE, template.convertAndSend(ANOTHER_CHANNEL, anotherPayload1));
assertEquals(ONE, template.convertAndSend(ANOTHER_CHANNEL, anotherPayload2));
Set<String> set = new LinkedHashSet<String>();
set.add(bag.poll(1, TimeUnit.SECONDS));
set.add(bag.poll(1, TimeUnit.SECONDS));
System.out.println(set);
assertFalse(set.contains(payload1));
assertFalse(set.contains(payload2));
assertTrue(set.contains(anotherPayload1));
assertTrue(set.contains(anotherPayload2));
}
}

2
src/test/java/org/springframework/data/redis/listener/PubSubTestParams.java
View File

@@ -39,7 +39,7 @@ public class PubSubTestParams {
ObjectFactory<Person> personFactory = new PersonObjectFactory();
JedisConnectionFactory jedisConnFactory = new JedisConnectionFactory();
jedisConnFactory.setUsePool(true);
jedisConnFactory.setUsePool(false);
jedisConnFactory.setPort(SettingsUtils.getPort());
jedisConnFactory.setHostName(SettingsUtils.getHost());
jedisConnFactory.setDatabase(2);

34
src/test/java/org/springframework/data/redis/listener/PubSubTests.java
View File

@@ -159,38 +159,4 @@ public class PubSubTests<T> {
assertFalse(set.contains(payload1));
assertFalse(set.contains(payload2));
}
@Test
public void testContainerChannelResubscribe() throws Exception {
String payload1 = "do";
String payload2 = "re mi";
String anotherPayload1 = "od";
String anotherPayload2 = "mi er";
String ANOTHER_CHANNEL = "pubsub::test::extra";
// bind listener on another channel
container.addMessageListener(adapter, new ChannelTopic(ANOTHER_CHANNEL));
container.removeMessageListener(null, new ChannelTopic(CHANNEL));
assertEquals(ZERO, template.convertAndSend(CHANNEL, payload1));
assertEquals(ZERO, template.convertAndSend(CHANNEL, payload2));
assertEquals(ONE, template.convertAndSend(ANOTHER_CHANNEL, anotherPayload1));
assertEquals(ONE, template.convertAndSend(ANOTHER_CHANNEL, anotherPayload2));
Set<String> set = new LinkedHashSet<String>();
set.add(bag.poll(1, TimeUnit.SECONDS));
set.add(bag.poll(1, TimeUnit.SECONDS));
System.out.println(set);
assertFalse(set.contains(payload1));
assertFalse(set.contains(payload2));
assertTrue(set.contains(anotherPayload1));
assertTrue(set.contains(anotherPayload2));
}
}