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 015541ae8..0e49e8f46 100644 --- a/src/main/java/org/springframework/data/redis/listener/RedisMessageListenerContainer.java +++ b/src/main/java/org/springframework/data/redis/listener/RedisMessageListenerContainer.java @@ -847,27 +847,40 @@ public class RedisMessageListenerContainer implements InitializingBean, Disposab if (logger.isTraceEnabled()) { logger.trace("Cancelling Redis subscription..."); } - if (connection != null) { - Subscription sub = connection.getSubscription(); - if (sub != null) { - synchronized (localMonitor) { + + if (connection == null) { + return; + } + + Subscription sub = connection.getSubscription(); + if (sub != null) { + synchronized (localMonitor) { + + if (logger.isTraceEnabled()) { logger.trace("Unsubscribing from all channels"); + } + + try { sub.pUnsubscribe(); sub.unsubscribe(); - if (subscriptionTaskRunning) { - try { - localMonitor.wait(subscriptionWait); - } catch (InterruptedException e) { - // Stop waiting - Thread.currentThread().interrupt(); - } - } - if (!subscriptionTaskRunning) { - closeConnection(); - } else { - logger.warn("Unable to close connection. Subscription task still running"); + } catch (Exception e) { + logger.warn("Unable to unsubscribe from subscriptions", e); + } + + if (subscriptionTaskRunning) { + try { + localMonitor.wait(subscriptionWait); + } catch (InterruptedException e) { + // Stop waiting + Thread.currentThread().interrupt(); } } + + if (!subscriptionTaskRunning) { + closeConnection(); + } else { + logger.warn("Unable to close connection. Subscription task still running"); + } } } } 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 917735af5..66beddd49 100644 --- a/src/test/java/org/springframework/data/redis/listener/RedisMessageListenerContainerTests.java +++ b/src/test/java/org/springframework/data/redis/listener/RedisMessageListenerContainerTests.java @@ -26,8 +26,10 @@ import org.junit.After; import org.junit.Before; import org.junit.Test; import org.springframework.core.task.SyncTaskExecutor; -import org.springframework.data.redis.SettingsUtils; -import org.springframework.data.redis.connection.jedis.JedisConnectionFactory; +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; /** @@ -44,28 +46,27 @@ 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() throws Exception { executorMock = mock(Executor.class); - - connectionFactory = new JedisConnectionFactory(); - connectionFactory.setPort(SettingsUtils.getPort()); - connectionFactory.setHostName(SettingsUtils.getHost()); - connectionFactory.setDatabase(2); - - connectionFactory.afterPropertiesSet(); + connectionFactoryMock = mock(LettuceConnectionFactory.class); + connectionMock = mock(RedisConnection.class); + subscriptionMock = mock(Subscription.class); container = new RedisMessageListenerContainer(); - container.setConnectionFactory(connectionFactory); + container.setConnectionFactory(connectionFactoryMock); container.setBeanName("container"); container.setTaskExecutor(new SyncTaskExecutor()); container.setSubscriptionExecutor(executorMock); + container.setMaxSubscriptionRegistrationWaitingTime(1); container.afterPropertiesSet(); } @@ -73,7 +74,6 @@ public class RedisMessageListenerContainerTests { public void tearDown() throws Exception { container.destroy(); - connectionFactory.destroy(); } @Test // DATAREDIS-415 @@ -97,4 +97,33 @@ public class RedisMessageListenerContainerTests { 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(); + } }