Use proper subscribe methods and fix nullability assumptions.

See: #964
Original Pull Request: #2256

Author: mp911de

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

@@ -596,7 +596,7 @@ public void addMessageListener(MessageListener listener, Topic topic) {
 	 * @param listener message listener
 	 * @param topics message listener topics
 	 */
-	public void removeMessageListener(MessageListener listener, Collection<? extends Topic> topics) {
+	public void removeMessageListener(@Nullable MessageListener listener, Collection<? extends Topic> topics) {
 		removeListener(listener, topics);
 	}
 
@@ -610,7 +610,7 @@ public void removeMessageListener(MessageListener listener, Collection<? extends
 	 * @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 void removeMessageListener(MessageListener listener, Topic topic) {
 	 * @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 @@ else if (topic instanceof PatternTopic) {
 		}
 	}
 
-	private void removeListener(MessageListener listener, Collection<? extends Topic> topics) {
+	private void removeListener(@Nullable MessageListener listener, Collection<? extends Topic> 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 @@ void doSubscribe(RedisConnection connection, Collection<byte[]> patterns, Collec
 			}
 
 			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][]));
+				}
 			}
 		}
 

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

@@ -210,12 +210,10 @@ void testInitializeContainerWithMultipleTopicsIncludingPattern() throws Exceptio
 		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.