MessageListenerContainer must not be initialized twice.

Error when afterPropertiesSet is called more than once. Update documentation and error messages.
Original Pull Request: #2256

Author: christophstrobl

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

@@ -163,64 +163,11 @@ public class RedisMessageListenerContainer implements InitializingBean, Disposab
 
 	private volatile CompletableFuture<Void> unsubscribeFuture = new CompletableFuture<>();
 
-	/**
-	 * Container listening state.
-	 */
-	static class State {
-
-		private final boolean prepareListening;
-		private final boolean listening;
-
-		private State(boolean prepareListening, boolean listening) {
-			this.prepareListening = prepareListening;
-			this.listening = listening;
-		}
-
-		/**
-		 * Initial state. Next state is {@link #prepareListening()}.
-		 */
-		public static State notListening() {
-			return new State(false, false);
-		}
-
-		/**
-		 * Prepare listening after {@link #notListening()}. Next states are either {@link #notListening()} upon failure or
-		 * {@link #listening()}.
-		 */
-		public static State prepareListening() {
-			return new State(true, false);
-		}
-
-		/**
-		 * Active listening state after {@link #prepareListening()}. Next is {@link #prepareUnsubscribe()}.
-		 */
-		public static State listening() {
-			return new State(true, true);
-		}
-
-		/**
-		 * Prepare unsubscribe after {@link #listening()}. Next state is {@link #notListening()}.
-		 */
-		public static State prepareUnsubscribe() {
-			return new State(false, true);
-		}
-
-		private boolean isListenerActivated() {
-			return isListening() || isPrepareListening();
-		}
-
-		public boolean isListening() {
-			return listening;
-		}
-
-		public boolean isPrepareListening() {
-			return prepareListening;
-		}
-	}
-
 	@Override
 	public void afterPropertiesSet() {
 
+		Assert.state(!afterPropertiesSet, "Container already initialized.");
+
 		if (this.connectionFactory == null) {
 			throw new IllegalArgumentException("RedisConnectionFactory is not set");
 		}
@@ -234,6 +181,7 @@ public void afterPropertiesSet() {
 			subscriptionExecutor = taskExecutor;
 		}
 
+
 		this.subscriber = createSubscriber(connectionFactory, this.subscriptionExecutor);
 
 		afterPropertiesSet = true;
@@ -268,7 +216,7 @@ public void destroy() throws Exception {
 				((DisposableBean) taskExecutor).destroy();
 
 				if (logger.isDebugEnabled()) {
-					logger.debug("Stopped internally-managed task executor");
+					logger.debug("Stopped internally-managed task executor.");
 				}
 			}
 		}
@@ -323,7 +271,7 @@ private void lazyListen() {
 		} catch (ExecutionException e) {
 			throw new CompletionException(e.getCause());
 		} catch (TimeoutException e) {
-			throw new IllegalStateException("Subscription registration timeout exceeded", e);
+			throw new IllegalStateException("Subscription registration timeout exceeded.", e);
 		}
 	}
 
@@ -333,7 +281,7 @@ private void lazyListen() {
 	private CompletableFuture<Void> lazyListen(BackOffExecution backOffExecution) {
 
 		if (!hasTopics()) {
-			logger.debug("Postpone listening for Redis messages until actual listeners are added");
+			logger.debug("Postpone listening for Redis messages until actual listeners are added.");
 			return CompletableFuture.completedFuture(null);
 		}
 
@@ -367,10 +315,10 @@ private boolean doSubscribe(BackOffExecution backOffExecution) {
 		listenFuture.whenComplete((unused, throwable) -> {
 
 			if (throwable == null) {
-				logger.debug("RedisMessageListenerContainer listeners registered successfully");
+				logger.debug("RedisMessageListenerContainer listeners registered successfully.");
 				this.state.set(State.listening());
 			} else {
-				logger.debug("Failed to start RedisMessageListenerContainer listeners", throwable);
+				logger.debug("Failed to start RedisMessageListenerContainer listeners.", throwable);
 				this.state.set(State.notListening());
 			}
 
@@ -382,7 +330,7 @@ private boolean doSubscribe(BackOffExecution backOffExecution) {
 			}
 		});
 
-		logger.debug("Subscribing to topics for RedisMessageListenerContainer");
+		logger.debug("Subscribing to topics for RedisMessageListenerContainer.");
 
 		return true;
 	}
@@ -422,7 +370,7 @@ public void stop(Runnable callback) {
 			stopListening();
 
 			if (logger.isDebugEnabled()) {
-				logger.debug("Stopped RedisMessageListenerContainer");
+				logger.debug("Stopped RedisMessageListenerContainer.");
 			}
 
 			callback.run();
@@ -459,7 +407,7 @@ private boolean doUnsubscribe() {
 			this.unsubscribeFuture = new CompletableFuture<>();
 
 			if (logger.isDebugEnabled()) {
-				logger.debug("Stopped RedisMessageListenerContainer");
+				logger.debug("Stopped listening.");
 			}
 
 			return true;
@@ -831,7 +779,7 @@ public void setRecoveryInterval(long recoveryInterval) {
 	 *
 	 * @see #handleSubscriptionException
 	 * @see #setRecoveryInterval(long)
-	 * @since 3.0
+	 * @since 2.7
 	 */
 	public void setRecoveryBackoff(BackOff recoveryInterval) {
 
@@ -928,8 +876,7 @@ protected void handleSubscriptionException(CompletableFuture<Void> future, BackO
 				long recoveryInterval = backOffExecution.nextBackOff();
 
 				if (recoveryInterval != BackOffExecution.STOP) {
-					logger.error(
-							"Connection failure occurred:" + ex + ". Restarting subscription task after " + recoveryInterval + " ms");
+					logger.error(String.format("Connection failure occurred: %s. Restarting subscription task after %s ms.", ex, recoveryInterval));
 				}
 
 				return recoveryInterval;
@@ -1011,14 +958,6 @@ private void dispatchSubscriptionNotification(Collection<MessageListener> listen
 		}
 	}
 
-	/**
-	 * Represents an operation that accepts three input arguments {@link SubscriptionListener},
-	 * {@code channel or pattern}, and {@code count} and returns no result.
-	 */
-	interface SubscriptionConsumer {
-		void accept(SubscriptionListener listener, byte[] channelOrPattern, long count);
-	}
-
 	private void dispatchMessage(Collection<MessageListener> listeners, Message message, @Nullable byte[] pattern) {
 
 		byte[] source = (pattern != null ? pattern.clone() : message.getChannel());
@@ -1057,6 +996,73 @@ private byte[] serialize(Topic topic) {
 		return serializer.serialize(topic.getTopic());
 	}
 
+	/**
+	 * Represents an operation that accepts three input arguments {@link SubscriptionListener},
+	 * {@code channel or pattern}, and {@code count} and returns no result.
+	 */
+	interface SubscriptionConsumer {
+		void accept(SubscriptionListener listener, byte[] channelOrPattern, long count);
+	}
+
+	/**
+	 * Container listening state.
+	 *
+	 * @author Mark Paluch
+	 * @since 2.7
+	 */
+	static class State {
+
+		private final boolean prepareListening;
+		private final boolean listening;
+
+		private State(boolean prepareListening, boolean listening) {
+
+			this.prepareListening = prepareListening;
+			this.listening = listening;
+		}
+
+		/**
+		 * Initial state. Next state is {@link #prepareListening()}.
+		 */
+		static State notListening() {
+			return new State(false, false);
+		}
+
+		/**
+		 * Prepare listening after {@link #notListening()}. Next states are either {@link #notListening()} upon failure or
+		 * {@link #listening()}.
+		 */
+		static State prepareListening() {
+			return new State(true, false);
+		}
+
+		/**
+		 * Active listening state after {@link #prepareListening()}. Next is {@link #prepareUnsubscribe()}.
+		 */
+		static State listening() {
+			return new State(true, true);
+		}
+
+		/**
+		 * Prepare unsubscribe after {@link #listening()}. Next state is {@link #notListening()}.
+		 */
+		static State prepareUnsubscribe() {
+			return new State(false, true);
+		}
+
+		private boolean isListenerActivated() {
+			return isListening() || isPrepareListening();
+		}
+
+		public boolean isListening() {
+			return listening;
+		}
+
+		public boolean isPrepareListening() {
+			return prepareListening;
+		}
+	}
+
 	/**
 	 * Actual message dispatcher/multiplexer.
 	 *
@@ -1119,7 +1125,7 @@ public void onPatternUnsubscribed(byte[] pattern, long count) {
 	 * actual listeners without blocking the event loop.
 	 *
 	 * @author Mark Paluch
-	 * @since 3.0
+	 * @since 2.7
 	 */
 	class Subscriber {
 
@@ -1324,7 +1330,7 @@ private void doWithSubscription(byte[][] data, BiConsumer<Subscription, byte[][]
 	 * Blocking variant of a subscriber for connectors that block within the (p)subscribe method.
 	 *
 	 * @author Mark Paluch
-	 * @since 3.0
+	 * @since 2.7
 	 */
 	class BlockingSubscriber extends Subscriber {
 

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

@@ -105,4 +105,9 @@ void containerShouldStopGracefullyOnUnsubscribeErrors() {
 		assertThat(container.isRunning()).isFalse();
 		verify(subscriptionMock).close();
 	}
+
+	@Test // GH-964
+	void failsOnDuplicateInit() {
+		assertThatExceptionOfType(IllegalStateException.class).isThrownBy(() -> container.afterPropertiesSet());
+	}
 }