diff --git a/src/main/java/org/springframework/data/redis/listener/RedisMessageListenerContainer.java b/src/main/java/org/springframework/data/redis/listener/RedisMessageListenerContainer.java index 6eace700d..dbcde0a9d 100644 --- a/src/main/java/org/springframework/data/redis/listener/RedisMessageListenerContainer.java +++ b/src/main/java/org/springframework/data/redis/listener/RedisMessageListenerContainer.java @@ -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 }); } } - } \ No newline at end of file diff --git a/src/test/java/org/springframework/data/redis/listener/PubSubResubscribeTests.java b/src/test/java/org/springframework/data/redis/listener/PubSubResubscribeTests.java index 64bd8059f..aef8336c7 100644 --- a/src/test/java/org/springframework/data/redis/listener/PubSubResubscribeTests.java +++ b/src/test/java/org/springframework/data/redis/listener/PubSubResubscribeTests.java @@ -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 set = new LinkedHashSet(); + 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)); + } +} \ No newline at end of file diff --git a/src/test/java/org/springframework/data/redis/listener/PubSubTestParams.java b/src/test/java/org/springframework/data/redis/listener/PubSubTestParams.java index 9c643f26d..6badf6eb8 100644 --- a/src/test/java/org/springframework/data/redis/listener/PubSubTestParams.java +++ b/src/test/java/org/springframework/data/redis/listener/PubSubTestParams.java @@ -39,7 +39,7 @@ public class PubSubTestParams { ObjectFactory personFactory = new PersonObjectFactory(); JedisConnectionFactory jedisConnFactory = new JedisConnectionFactory(); - jedisConnFactory.setUsePool(true); + jedisConnFactory.setUsePool(false); jedisConnFactory.setPort(SettingsUtils.getPort()); jedisConnFactory.setHostName(SettingsUtils.getHost()); jedisConnFactory.setDatabase(2); diff --git a/src/test/java/org/springframework/data/redis/listener/PubSubTests.java b/src/test/java/org/springframework/data/redis/listener/PubSubTests.java index e6137254c..65d658b75 100644 --- a/src/test/java/org/springframework/data/redis/listener/PubSubTests.java +++ b/src/test/java/org/springframework/data/redis/listener/PubSubTests.java @@ -159,38 +159,4 @@ public class PubSubTests { 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 set = new LinkedHashSet(); - 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)); - } - } \ No newline at end of file