From ee8c17a4e9011a48cbefcf5a9990c990dcd2961e Mon Sep 17 00:00:00 2001 From: Christoph Strobl Date: Tue, 12 Jun 2018 10:23:04 +0200 Subject: [PATCH] DATAREDIS-840 - Polishing. Introduce dedicated Subscription.close() to unsubscribe() and punsubscribe(). Original Pull Request: #346 --- .../data/redis/connection/Subscription.java | 7 ++ .../connection/util/AbstractSubscription.java | 9 ++ .../RedisMessageListenerContainer.java | 17 +-- .../RedisMessageListenerContainerTests.java | 67 +++-------- ...edisMessageListenerContainerUnitTests.java | 113 ++++++++++++++++++ 5 files changed, 153 insertions(+), 60 deletions(-) create mode 100644 src/test/java/org/springframework/data/redis/listener/RedisMessageListenerContainerUnitTests.java 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 91e380be1..89da3ee7f 100644 --- a/src/main/java/org/springframework/data/redis/connection/Subscription.java +++ b/src/main/java/org/springframework/data/redis/connection/Subscription.java @@ -90,4 +90,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 9fe074689..51a67e008 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 @@ -93,6 +93,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 4936da89f..6ac8eb3bb 100644 --- a/src/main/java/org/springframework/data/redis/listener/RedisMessageListenerContainer.java +++ b/src/main/java/org/springframework/data/redis/listener/RedisMessageListenerContainer.java @@ -725,9 +725,11 @@ public class RedisMessageListenerContainer implements InitializingBean, Disposab } public void run() { + synchronized (localMonitor) { subscriptionTaskRunning = true; } + try { connection = connectionFactory.getConnection(); if (connection.isSubscribed()) { @@ -835,21 +837,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()) { @@ -857,8 +859,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); } @@ -882,6 +883,7 @@ public class RedisMessageListenerContainer implements InitializingBean, Disposab } void closeConnection() { + if (connection != null) { logger.trace("Closing connection"); try { @@ -894,6 +896,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 e3157d539..fc875e192 100644 --- a/src/test/java/org/springframework/data/redis/listener/RedisMessageListenerContainerTests.java +++ b/src/test/java/org/springframework/data/redis/listener/RedisMessageListenerContainerTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2016-2017 the original author or authors. + * Copyright 2016-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. @@ -28,15 +28,12 @@ import org.junit.Test; import org.mockito.invocation.InvocationOnMock; import org.mockito.stubbing.Answer; import org.springframework.core.task.SyncTaskExecutor; -import org.springframework.data.redis.connection.MessageListener; -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.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 @@ -51,33 +48,35 @@ 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); + + connectionFactory = new JedisConnectionFactory(); + connectionFactory.setPort(SettingsUtils.getPort()); + connectionFactory.setHostName(SettingsUtils.getHost()); + connectionFactory.setDatabase(2); + 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 @@ -103,42 +102,4 @@ public class RedisMessageListenerContainerTests { 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(new Answer() { - - @Override - public Object answer(InvocationOnMock invocation) throws Throwable { - - Runnable r = invocation.getArgumentAt(0, Runnable.class); - new Thread(r).start(); - return null; - } - }).when(executorMock).execute(any(Runnable.class)); - - doAnswer(new Answer() { - - @Override - public Object answer(InvocationOnMock invocation) throws Throwable { - - when(connectionMock.isSubscribed()).thenReturn(true); - return null; - } - }).when(connectionMock).subscribe(any(MessageListener.class), any(byte[][].class)); - - 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..d39dffe38 --- /dev/null +++ b/src/test/java/org/springframework/data/redis/listener/RedisMessageListenerContainerUnitTests.java @@ -0,0 +1,113 @@ +/* + * 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.mockito.invocation.InvocationOnMock; +import org.mockito.stubbing.Answer; +import org.springframework.core.task.SyncTaskExecutor; +import org.springframework.data.redis.connection.MessageListener; +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(new Answer() { + + @Override + public Object answer(InvocationOnMock invocation) throws Throwable { + + Runnable r = invocation.getArgumentAt(0, Runnable.class); + new Thread(r).start(); + return null; + } + }).when(executorMock).execute(any(Runnable.class)); + + doAnswer(new Answer() { + + @Override + public Object answer(InvocationOnMock invocation) throws Throwable { + + when(connectionMock.isSubscribed()).thenReturn(true); + return null; + } + }).when(connectionMock).subscribe(any(MessageListener.class), any(byte[][].class)); + + container.addMessageListener(adapter, new ChannelTopic("a")); + container.start(); + + when(connectionMock.getSubscription()).thenReturn(subscriptionMock); + + container.stop(); + + assertThat(container.isRunning(), is(false)); + verify(subscriptionMock).close(); + } +}