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 51f8bda64..7df5ac58f 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 @@ -147,7 +147,9 @@ public class LettuceConnection implements RedisConnection { } if (subscription != null) { - subscription.doClose(); + if(subscription.isAlive()) { + subscription.doClose(); + } subscription = null; } } diff --git a/src/main/java/org/springframework/data/redis/connection/srp/SrpConnection.java b/src/main/java/org/springframework/data/redis/connection/srp/SrpConnection.java index 86a112486..1055615db 100644 --- a/src/main/java/org/springframework/data/redis/connection/srp/SrpConnection.java +++ b/src/main/java/org/springframework/data/redis/connection/srp/SrpConnection.java @@ -141,7 +141,9 @@ public class SrpConnection implements RedisConnection { queue.remove(this); if (subscription != null) { - subscription.doClose(); + if(subscription.isAlive()) { + subscription.doClose(); + } subscription = null; } 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 b8b4f42bf..4cd6b4fa5 100644 --- a/src/main/java/org/springframework/data/redis/listener/RedisMessageListenerContainer.java +++ b/src/main/java/org/springframework/data/redis/listener/RedisMessageListenerContainer.java @@ -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); } } } diff --git a/src/test/java/org/springframework/data/redis/listener/SubscriptionConnectionTests.java b/src/test/java/org/springframework/data/redis/listener/SubscriptionConnectionTests.java new file mode 100644 index 000000000..0b91c5d89 --- /dev/null +++ b/src/test/java/org/springframework/data/redis/listener/SubscriptionConnectionTests.java @@ -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 containers = new ArrayList(); + + 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(); + } + +}