Polishing.

- Add integration test to verify all removeMessageListener variants
- Add unit tests for negative tests on removeMesssageListener invalid inputs
- Clarify javadocs

Original Pull Request #3237

Author: onobc

src/main/java/org/springframework/data/redis/listener/RedisMessageListenerContainer.java

@@ -585,12 +585,9 @@ public void addMessageListener(MessageListener listener, Topic topic) {
 	/**
 	 * Removes a message listener from the given topics. If the container is running, the listener stops receiving
 	 * (matching) messages as soon as possible.
-	 * <p>
-	 * Note that this method obeys the Redis (p)unsubscribe semantics - meaning an empty/null collection will remove
-	 * listener from all channels.
 	 *
 	 * @param listener message listener.
-	 * @param topics message listener topics.
+	 * @param topics message topics to remove listener from or {@literal empty} to remove the listener from all topics.
 	 */
 	public void removeMessageListener(MessageListener listener, Collection<? extends Topic> topics) {
 		removeListener(listener, topics);
@@ -599,27 +596,22 @@ public void removeMessageListener(MessageListener listener, Collection<? extends
 	/**
 	 * Removes a message listener from the given topic. If the container is running, the listener stops receiving
 	 * (matching) messages as soon as possible.
-	 * <p>
-	 * Note that this method obeys the Redis (p)unsubscribe semantics - meaning an empty/null collection will remove
-	 * listener from all channels.
 	 *
 	 * @param listener message listener.
 	 * @param topic message topic.
 	 */
 	public void removeMessageListener(MessageListener listener, Topic topic) {
+		Assert.notNull(topic, "Topic must not be null");
 		removeMessageListener(listener, Collections.singleton(topic));
 	}
 
 	/**
-	 * Removes the given message listener completely (from all topics). If the container is running, the listener stops
+	 * Removes the given message listener from all topics. If the container is running, the listener stops
 	 * receiving (matching) messages as soon as possible.
 	 *
 	 * @param listener message listener.
 	 */
 	public void removeMessageListener(MessageListener listener) {
-
-		Assert.notNull(listener, "MessageListener must not be null");
-
 		removeMessageListener(listener, Collections.emptySet());
 	}
 

src/test/java/org/springframework/data/redis/listener/PubSubResubscribeTests.java

@@ -178,8 +178,9 @@ void testContainerChannelResubscribe() {
 		String ANOTHER_CHANNEL = "pubsub::test::extra";
 
 		// bind listener on another channel
+		container.addMessageListener(adapter, new ChannelTopic(CHANNEL));
 		container.addMessageListener(adapter, new ChannelTopic(ANOTHER_CHANNEL));
-		container.removeMessageListener(null, new ChannelTopic(CHANNEL));
+		container.removeMessageListener(adapter, new ChannelTopic(CHANNEL));
 
 		// Listener removed from channel
 		template.convertAndSend(CHANNEL, payload1);

src/test/java/org/springframework/data/redis/listener/RedisMessageListenerContainerIntegrationTests.java

@@ -18,8 +18,12 @@
 import static org.assertj.core.api.Assertions.*;
 import static org.awaitility.Awaitility.*;
 
+import java.time.Duration;
 import java.util.Arrays;
 import java.util.Collection;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Set;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.TimeUnit;
@@ -354,6 +358,102 @@ public void onChannelSubscribed(byte[] channel, long count) {
 		container.destroy();
 	}
 
+	/**
+	 * Tests the variants of {@link RedisMessageListenerContainer#removeMessageListener} to ensure the removal stops
+	 * message delivery and that the pattern and channel listeners do not interfere with one another upon removal.
+	 */
+	@Test // GH-3237
+	void shouldRemoveListenerFromChannelsAndTopics() throws Exception {
+
+		AtomicInteger subscriptions = new AtomicInteger();
+		Map<String, String> patternsMessages = new HashMap<>();
+		Map<String, String> channelsMessages = new HashMap<>();
+
+		MessageListener patternListener = new CompositeListener() {
+			@Override
+			public void onMessage(Message message, byte @Nullable [] pattern) {
+				patternsMessages.put(new String(pattern), message.toString());
+			}
+
+			@Override
+			public void onPatternSubscribed(byte[] pattern, long count) {
+				subscriptions.incrementAndGet();
+			}
+
+		};
+
+		CompositeListener channelListener = new CompositeListener() {
+			@Override
+			public void onMessage(Message message, byte @Nullable [] pattern) {
+				channelsMessages.put(new String(pattern), message.toString());
+			}
+
+			@Override
+			public void onChannelSubscribed(byte[] channel, long count) {
+				subscriptions.incrementAndGet();
+			}
+
+		};
+
+		try {
+			container.start();
+			container.addMessageListener(patternListener, new PatternTopic("a-pattern-0"));
+			container.addMessageListener(patternListener, new PatternTopic("a-pattern-1"));
+			container.addMessageListener(patternListener, new PatternTopic("a-pattern-2"));
+			container.addMessageListener(channelListener, new ChannelTopic("a-channel-0"));
+			container.addMessageListener(channelListener, new ChannelTopic("a-channel-1"));
+
+			// Wait for the subscriptions to register
+			await().untilAtomic(subscriptions, Matchers.is(5));
+
+			try (RedisConnection connection = connectionFactory.getConnection()) {
+
+				// remove patternListener from single topic
+				container.removeMessageListener(patternListener, new PatternTopic("a-pattern-0"));
+
+				// send each topic a msg - removed listener should not get msg
+				connection.publish("a-pattern-0".getBytes(), "pattern100".getBytes());
+				connection.publish("a-pattern-1".getBytes(), "pattern101".getBytes());
+				connection.publish("a-pattern-2".getBytes(), "pattern102".getBytes());
+				await().atMost(Duration.ofSeconds(15)).untilAsserted(() -> {
+					assertThat(patternsMessages).doesNotContain(entry("a-pattern-0", "pattern100"))
+							.contains(entry("a-pattern-1", "pattern101"), entry("a-pattern-2", "pattern102"));
+				});
+				patternsMessages.clear();
+
+				// remove pattern listener from multiple (2 remaining topics)
+				container.removeMessageListener(patternListener,
+						Set.of(new PatternTopic("a-pattern-1"), new PatternTopic("a-pattern-2")));
+
+				// no listener should receive msgs
+				connection.publish("a-pattern-0".getBytes(), "pattern100".getBytes());
+				connection.publish("a-pattern-1".getBytes(), "pattern101".getBytes());
+				connection.publish("a-pattern-2".getBytes(), "pattern102".getBytes());
+				await().atMost(Duration.ofSeconds(15)).untilAsserted(() -> assertThat(patternsMessages).isEmpty());
+
+				// sanity check sending msgs to channels (removal of pattern listeners did not disrupt)
+				connection.publish("a-channel-0".getBytes(), "channel100".getBytes());
+				connection.publish("a-channel-1".getBytes(), "channel101".getBytes());
+				await().atMost(Duration.ofSeconds(15)).untilAsserted(() -> {
+					assertThat(channelsMessages).containsExactly(entry("a-channel-0", "channel100"),
+							entry("a-channel-1", "channel101"));
+				});
+				channelsMessages.clear();
+
+				// remove channel listener from all channel topics
+				container.removeMessageListener(channelListener);
+
+				// should receive no msgs
+				connection.publish("a-channel-0".getBytes(), "channel100".getBytes());
+				connection.publish("a-channel-1".getBytes(), "channel101".getBytes());
+				await().atMost(Duration.ofSeconds(15)).untilAsserted(() -> assertThat(channelsMessages).isEmpty());
+			}
+		}
+		finally {
+			container.destroy();
+		}
+	}
+
 	interface CompositeListener extends MessageListener, SubscriptionListener {
 
 	}

src/test/java/org/springframework/data/redis/listener/RedisMessageListenerContainerUnitTests.java

@@ -20,6 +20,7 @@
 
 import java.nio.charset.StandardCharsets;
 import java.util.Collections;
+import java.util.Set;
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.Executor;
 import java.util.concurrent.atomic.AtomicBoolean;
@@ -225,19 +226,29 @@ void failsOnDuplicateInit() {
 		assertThatIllegalStateException().isThrownBy(() -> container.afterPropertiesSet());
 	}
 
-	@Test // GH-3009
-	void shouldRemoveAllListenersWhenListenerIsNull() {
+	@Test // GH-3237
+	void removeListenerBySingleTopicShouldFailWhenTopicIsNull() {
+		assertThatIllegalArgumentException().isThrownBy(() -> container.removeMessageListener(adapter, (Topic) null));
+	}
 
-		MessageListener listener1 = mock(MessageListener.class);
-		MessageListener listener2 = mock(MessageListener.class);
-		Topic topic = new ChannelTopic("topic1");
+	@Test // GH-3237
+	void removeListenerBySingleTopicShouldFailWhenListenerIsNull() {
+		assertThatIllegalArgumentException().isThrownBy(() -> container.removeMessageListener(null, new ChannelTopic("a")));
+	}
 
-		container.addMessageListener(listener1, Collections.singletonList(topic));
-		container.addMessageListener(listener2, Collections.singletonList(topic));
+	@Test // GH-3237
+	void removeListenerBySetShouldFailWhenListenerIsNull() {
+		assertThatIllegalArgumentException().isThrownBy(() -> container.removeMessageListener(null, Collections.emptySet()));
+	}
 
-		container.removeMessageListener(null, Collections.singletonList(topic));
+	@Test // GH-3237
+	void removeListenerBySetShouldFailWhenSetIsNull() {
+		assertThatIllegalArgumentException().isThrownBy(() -> container.removeMessageListener(adapter, (Set)null));
+	}
 
-		assertThatNoException().isThrownBy(() -> container.removeMessageListener(null, Collections.singletonList(topic)));
+	@Test // GH-3237
+	void removeListenerFromAllTopicsShouldFailWhenListenerIsNull() {
+		assertThatIllegalArgumentException().isThrownBy(() -> container.removeMessageListener(null));
 	}
 
 	@Test // GH-3208