Safely add and register the MessageListener to Topic mapping.

Given addListener(:MessageListener, :Collection<Topic>) could be called concurrently from the addMessageListener(:MessageListener, Collection<Topic>) method by multiple Threads, and the RedisMessageListenerContainer Javadoc specifically states that it is safe to call the addMessageListener(..) method conurrently without any external synchronization, and the registeration (or mapping) of listener to Topics is a componund action, then a race condition is possible.

Closes #2755

Author: jxblum

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

@@ -612,20 +612,17 @@ private void initMapping(Map<? extends MessageListener, Collection<? extends Top
 
 	private void addListener(MessageListener listener, Collection<? extends Topic> topics) {
 
-		Assert.notNull(listener, "a valid listener is required");
-		Assert.notEmpty(topics, "at least one topic is required");
+		Assert.notNull(listener, "A valid listener is required");
+		Assert.notEmpty(topics, "At least one topic is required");
 
 		List<byte[]> channels = new ArrayList<>(topics.size());
 		List<byte[]> patterns = new ArrayList<>(topics.size());
 
 		boolean trace = logger.isTraceEnabled();
 
-		// add listener mapping
-		Set<Topic> set = listenerTopics.get(listener);
-		if (set == null) {
-			set = new CopyOnWriteArraySet<>();
-			listenerTopics.put(listener, set);
-		}
+		// safely lookup or add MessageListener to Topic mapping
+		Set<Topic> set = listenerTopics.computeIfAbsent(listener, key -> new CopyOnWriteArraySet<>());
+
 		set.addAll(topics);
 
 		for (Topic topic : topics) {