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 2b8da5633..4b511d62e 100644 --- a/src/main/java/org/springframework/data/redis/listener/RedisMessageListenerContainer.java +++ b/src/main/java/org/springframework/data/redis/listener/RedisMessageListenerContainer.java @@ -596,7 +596,7 @@ public class RedisMessageListenerContainer implements InitializingBean, Disposab * @param listener message listener * @param topics message listener topics */ - public void removeMessageListener(MessageListener listener, Collection topics) { + public void removeMessageListener(@Nullable MessageListener listener, Collection topics) { removeListener(listener, topics); } @@ -610,7 +610,7 @@ public class RedisMessageListenerContainer implements InitializingBean, Disposab * @param listener message listener * @param topic message topic */ - public void removeMessageListener(MessageListener listener, Topic topic) { + public void removeMessageListener(@Nullable MessageListener listener, Topic topic) { removeMessageListener(listener, Collections.singleton(topic)); } @@ -621,6 +621,8 @@ public class RedisMessageListenerContainer implements InitializingBean, Disposab * @param listener message listener */ public void removeMessageListener(MessageListener listener) { + + Assert.notNull(listener, "MessageListener must not be null"); removeMessageListener(listener, Collections.emptySet()); } @@ -717,12 +719,11 @@ public class RedisMessageListenerContainer implements InitializingBean, Disposab } } - private void removeListener(MessageListener listener, Collection topics) { + private void removeListener(@Nullable MessageListener listener, Collection topics) { - Assert.notNull(listener, "MessageListener must not be null"); Assert.notNull(topics, "Topics must not be null"); - if (listenerTopics.get(listener) == null) { + if (listener != null && listenerTopics.get(listener) == null) { // Listener not subscribed return; } @@ -1197,7 +1198,11 @@ public class RedisMessageListenerContainer implements InitializingBean, Disposab } if (!channels.isEmpty()) { - connection.subscribe(synchronizingMessageListener, channels.toArray(new byte[0][])); + if (patterns.isEmpty()) { + connection.subscribe(synchronizingMessageListener, channels.toArray(new byte[0][])); + } else { + subscribeChannel(channels.toArray(new byte[0][])); + } } } 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 41e3d546f..d60b8c745 100644 --- a/src/test/java/org/springframework/data/redis/listener/PubSubResubscribeTests.java +++ b/src/test/java/org/springframework/data/redis/listener/PubSubResubscribeTests.java @@ -210,12 +210,10 @@ public class PubSubResubscribeTests { container.stop(); String uniqueChannel = "random-" + UUID.randomUUID(); - PubSubAwaitUtil.runAndAwaitPatternSubscription(template.getConnectionFactory(), () -> { - container.addMessageListener(adapter, - Arrays.asList(new Topic[] { new ChannelTopic(uniqueChannel), new PatternTopic("s*") })); - container.start(); - }); + container.addMessageListener(adapter, + Arrays.asList(new Topic[] { new ChannelTopic(uniqueChannel), new PatternTopic("s*") })); + container.start(); // timing: There's currently no other way to synchronize // than to hope the subscribe/unsubscribe are executed within the time.