diff --git a/src/main/java/org/springframework/data/redis/connection/Subscription.java b/src/main/java/org/springframework/data/redis/connection/Subscription.java index 41de27d07..a9568dee3 100644 --- a/src/main/java/org/springframework/data/redis/connection/Subscription.java +++ b/src/main/java/org/springframework/data/redis/connection/Subscription.java @@ -91,4 +91,11 @@ public interface Subscription { * @return true if the subscription still applies, false otherwise. */ boolean isAlive(); + + /** + * Shutdown the subscription and free any resources held. + * + * @since 1.8.12 + */ + void close(); } diff --git a/src/main/java/org/springframework/data/redis/connection/util/AbstractSubscription.java b/src/main/java/org/springframework/data/redis/connection/util/AbstractSubscription.java index d15723e21..35c781851 100644 --- a/src/main/java/org/springframework/data/redis/connection/util/AbstractSubscription.java +++ b/src/main/java/org/springframework/data/redis/connection/util/AbstractSubscription.java @@ -97,6 +97,15 @@ public abstract class AbstractSubscription implements Subscription { */ protected abstract void doPUnsubscribe(boolean all, byte[]... patterns); + /* + * (non-Javadoc) + * @see org.springframework.data.redis.connection.Subscription#close() + */ + @Override + public void close() { + doClose(); + } + /** * Shutdown the subscription and free any resources held. */ 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 cc96116e8..e01dc0ba1 100644 --- a/src/main/java/org/springframework/data/redis/listener/RedisMessageListenerContainer.java +++ b/src/main/java/org/springframework/data/redis/listener/RedisMessageListenerContainer.java @@ -729,9 +729,11 @@ public class RedisMessageListenerContainer implements InitializingBean, Disposab } public void run() { + synchronized (localMonitor) { subscriptionTaskRunning = true; } + try { connection = connectionFactory.getConnection(); if (connection.isSubscribed()) { @@ -839,21 +841,21 @@ public class RedisMessageListenerContainer implements InitializingBean, Disposab } void cancel() { - if (!listening) { + + if (!listening || connection == null) { return; } + listening = false; if (logger.isTraceEnabled()) { logger.trace("Cancelling Redis subscription..."); } - if (connection == null) { - return; - } - Subscription sub = connection.getSubscription(); + if (sub != null) { + synchronized (localMonitor) { if (logger.isTraceEnabled()) { @@ -861,8 +863,7 @@ public class RedisMessageListenerContainer implements InitializingBean, Disposab } try { - sub.pUnsubscribe(); - sub.unsubscribe(); + sub.close(); } catch (Exception e) { logger.warn("Unable to unsubscribe from subscriptions", e); } @@ -886,6 +887,7 @@ public class RedisMessageListenerContainer implements InitializingBean, Disposab } void closeConnection() { + if (connection != null) { logger.trace("Closing connection"); try { @@ -898,6 +900,7 @@ public class RedisMessageListenerContainer implements InitializingBean, Disposab } void subscribeChannel(byte[]... channels) { + if (channels != null && channels.length > 0) { if (connection != null) { synchronized (localMonitor) { diff --git a/src/test/java/org/springframework/data/redis/listener/RedisMessageListenerContainerTests.java b/src/test/java/org/springframework/data/redis/listener/RedisMessageListenerContainerTests.java index 04600d455..b10caf56e 100644 --- a/src/test/java/org/springframework/data/redis/listener/RedisMessageListenerContainerTests.java +++ b/src/test/java/org/springframework/data/redis/listener/RedisMessageListenerContainerTests.java @@ -25,14 +25,13 @@ import org.junit.After; import org.junit.Before; import org.junit.Test; import org.springframework.core.task.SyncTaskExecutor; -import org.springframework.data.redis.connection.RedisConnection; -import org.springframework.data.redis.connection.RedisConnectionFactory; -import org.springframework.data.redis.connection.Subscription; -import org.springframework.data.redis.connection.lettuce.LettuceConnectionFactory; +import org.springframework.data.redis.SettingsUtils; +import org.springframework.data.redis.connection.RedisStandaloneConfiguration; +import org.springframework.data.redis.connection.jedis.JedisConnectionFactory; import org.springframework.data.redis.listener.adapter.MessageListenerAdapter; /** - * Unit tests for {@link RedisMessageListenerContainer}. + * Integration tests for {@link RedisMessageListenerContainer}. * * @author Mark Paluch * @author Christoph Strobl @@ -47,39 +46,43 @@ public class RedisMessageListenerContainerTests { private final MessageListenerAdapter adapter = new MessageListenerAdapter(handler); + private JedisConnectionFactory connectionFactory; private RedisMessageListenerContainer container; - private RedisConnectionFactory connectionFactoryMock; - private RedisConnection connectionMock; - private Subscription subscriptionMock; private Executor executorMock; @Before public void setUp() { executorMock = mock(Executor.class); - connectionFactoryMock = mock(LettuceConnectionFactory.class); - connectionMock = mock(RedisConnection.class); - subscriptionMock = mock(Subscription.class); + + RedisStandaloneConfiguration configuration = new RedisStandaloneConfiguration(); + configuration.setPort(SettingsUtils.getPort()); + configuration.setHostName(SettingsUtils.getHost()); + configuration.setDatabase(2); + + connectionFactory = new JedisConnectionFactory(configuration); + connectionFactory.afterPropertiesSet(); container = new RedisMessageListenerContainer(); - container.setConnectionFactory(connectionFactoryMock); + container.setConnectionFactory(connectionFactory); container.setBeanName("container"); container.setTaskExecutor(new SyncTaskExecutor()); container.setSubscriptionExecutor(executorMock); - container.setMaxSubscriptionRegistrationWaitingTime(1); container.afterPropertiesSet(); } @After public void tearDown() throws Exception { + container.destroy(); + connectionFactory.destroy(); } @Test // DATAREDIS-415 public void interruptAtStart() { - Thread main = Thread.currentThread(); + final Thread main = Thread.currentThread(); // interrupt thread once Executor.execute is called doAnswer(invocationOnMock -> { @@ -93,36 +96,7 @@ public class RedisMessageListenerContainerTests { // reset the interrupted flag to not destroy the teardown assertThat(Thread.interrupted(), is(true)); - assertThat(container.isRunning(), is(false)); - } - - @Test // DATAREDIS-840 - public void containerShouldStopGracefullyOnUnsubscribeErrors() { - - when(connectionFactoryMock.getConnection()).thenReturn(connectionMock); - doThrow(new IllegalStateException()).when(subscriptionMock).pUnsubscribe(); - - doAnswer(it -> { - - Runnable r = it.getArgument(0); - new Thread(r).start(); - return null; - }).when(executorMock).execute(any()); - - doAnswer(it -> { - - when(connectionMock.isSubscribed()).thenReturn(true); - return null; - }).when(connectionMock).subscribe(any(), any()); - - container.addMessageListener(adapter, new ChannelTopic("a")); - container.start(); - - when(connectionMock.getSubscription()).thenReturn(subscriptionMock); - - container.stop(); assertThat(container.isRunning(), is(false)); - verify(subscriptionMock).pUnsubscribe(); } } diff --git a/src/test/java/org/springframework/data/redis/listener/RedisMessageListenerContainerUnitTests.java b/src/test/java/org/springframework/data/redis/listener/RedisMessageListenerContainerUnitTests.java new file mode 100644 index 000000000..bfcbdf00a --- /dev/null +++ b/src/test/java/org/springframework/data/redis/listener/RedisMessageListenerContainerUnitTests.java @@ -0,0 +1,102 @@ +/* + * Copyright 2018 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 static org.hamcrest.core.Is.*; +import static org.junit.Assert.*; +import static org.mockito.Mockito.*; + +import java.util.concurrent.Executor; + +import org.junit.Before; +import org.junit.Test; +import org.springframework.core.task.SyncTaskExecutor; +import org.springframework.data.redis.connection.RedisConnection; +import org.springframework.data.redis.connection.RedisConnectionFactory; +import org.springframework.data.redis.connection.Subscription; +import org.springframework.data.redis.connection.lettuce.LettuceConnectionFactory; +import org.springframework.data.redis.listener.adapter.MessageListenerAdapter; + +/** + * Unit tests for {@link RedisMessageListenerContainer}. + * + * @author Mark Paluch + * @author Christoph Strobl + */ +public class RedisMessageListenerContainerUnitTests { + + private final Object handler = new Object() { + + @SuppressWarnings("unused") + public void handleMessage(Object message) {} + }; + + private final MessageListenerAdapter adapter = new MessageListenerAdapter(handler); + + private RedisMessageListenerContainer container; + + private RedisConnectionFactory connectionFactoryMock; + private RedisConnection connectionMock; + private Subscription subscriptionMock; + private Executor executorMock; + + @Before + public void setUp() { + + executorMock = mock(Executor.class); + connectionFactoryMock = mock(LettuceConnectionFactory.class); + connectionMock = mock(RedisConnection.class); + subscriptionMock = mock(Subscription.class); + + container = new RedisMessageListenerContainer(); + container.setConnectionFactory(connectionFactoryMock); + container.setBeanName("container"); + container.setTaskExecutor(new SyncTaskExecutor()); + container.setSubscriptionExecutor(executorMock); + container.setMaxSubscriptionRegistrationWaitingTime(1); + container.afterPropertiesSet(); + } + + @Test // DATAREDIS-840 + public void containerShouldStopGracefullyOnUnsubscribeErrors() { + + when(connectionFactoryMock.getConnection()).thenReturn(connectionMock); + doThrow(new IllegalStateException()).when(subscriptionMock).pUnsubscribe(); + + doAnswer(it -> { + + Runnable r = it.getArgument(0); + new Thread(r).start(); + return null; + }).when(executorMock).execute(any()); + + doAnswer(it -> { + + when(connectionMock.isSubscribed()).thenReturn(true); + return null; + }).when(connectionMock).subscribe(any(), any()); + + container.addMessageListener(adapter, new ChannelTopic("a")); + container.start(); + + when(connectionMock.getSubscription()).thenReturn(subscriptionMock); + + container.stop(); + + assertThat(container.isRunning(), is(false)); + verify(subscriptionMock).close(); + } +}