Revise implementation of RedisMessageListenerContainer.

RedisMessageListenerContainer is now reimplemented using non-blocking synchronization guards and a state management to simplify its maintenances. Additionally, listener registration and subscription setup through the start() method awaits until the listener subscription is confirmed by the Redis server. The synchronization removes potential race conditions that could happen by concurrent access to blocking Redis connectors in which the registration state was guessed and not awaited.

Resolves: #964
Original Pull Request: #2256

src/main/asciidoc/reference/redis-messaging.adoc

@@ -47,7 +47,7 @@ Due to its blocking nature, low-level subscription is not attractive, as it requ
 
 `RedisMessageListenerContainer` acts as a message listener container. It is used to receive messages from a Redis channel and drive the `MessageListener` instances that are injected into it. The listener container is responsible for all threading of message reception and dispatches into the listener for processing. A message listener container is the intermediary between an MDP and a messaging provider and takes care of registering to receive messages, resource acquisition and release, exception conversion, and the like. This lets you as an application developer write the (possibly complex) business logic associated with receiving a message (and reacting to it) and delegates boilerplate Redis infrastructure concerns to the framework.
 
- A `MessageListener` can additionally implement `SubscriptionListener` to receive notifications upon subscription/unsubscribe confirmation. Listening to subscription notifications can be useful when synchronizing invocations.
+A `MessageListener` can additionally implement `SubscriptionListener` to receive notifications upon subscription/unsubscribe confirmation. Listening to subscription notifications can be useful when synchronizing invocations.
 
 Furthermore, to minimize the application footprint, `RedisMessageListenerContainer` lets one connection and one thread be shared by multiple listeners even though they do not share a subscription. Thus, no matter how many listeners or channels an application tracks, the runtime cost remains the same throughout its lifetime. Moreover, the container allows runtime configuration changes so that you can add or remove listeners while an application is running without the need for a restart. Additionally, the container uses a lazy subscription approach, using a `RedisConnection` only when needed. If all the listeners are unsubscribed, cleanup is automatically performed, and the thread is released.
 

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

@@ -0,0 +1,184 @@
+/*
+ * Copyright 2022 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.springframework.data.redis.listener;
+
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.List;
+import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.CopyOnWriteArrayList;
+import java.util.concurrent.atomic.AtomicIntegerFieldUpdater;
+import java.util.function.BiFunction;
+import java.util.stream.Collectors;
+
+import org.springframework.data.redis.connection.Message;
+import org.springframework.data.redis.connection.MessageListener;
+import org.springframework.data.redis.connection.SubscriptionListener;
+import org.springframework.data.redis.connection.util.ByteArrayWrapper;
+import org.springframework.lang.Nullable;
+
+/**
+ * Synchronizing {@link MessageListener} and {@link SubscriptionListener} that allows notifying a {@link Runnable}
+ * (through {@link SubscriptionSynchronizion}) upon completing subscriptions to channels or patterns.
+ *
+ * @author Mark Paluch
+ * @since 3.0
+ */
+class SynchronizingMessageListener implements MessageListener, SubscriptionListener {
+
+	private final MessageListener messageListener;
+	private final SubscriptionListener subscriptionListener;
+	private final List<SubscriptionSynchronizion> synchronizations = new CopyOnWriteArrayList<>();
+
+	public SynchronizingMessageListener(MessageListener messageListener, SubscriptionListener subscriptionListener) {
+		this.messageListener = messageListener;
+		this.subscriptionListener = subscriptionListener;
+	}
+
+	/**
+	 * Register a {@link SubscriptionSynchronizion}.
+	 *
+	 * @param synchronization must not be {@literal null}.
+	 */
+	public void addSynchronization(SubscriptionSynchronizion synchronization) {
+		this.synchronizations.add(synchronization);
+	}
+
+	@Override
+	public void onMessage(Message message, @Nullable byte[] pattern) {
+		messageListener.onMessage(message, pattern);
+	}
+
+	@Override
+	public void onChannelSubscribed(byte[] channel, long count) {
+
+		subscriptionListener.onChannelSubscribed(channel, count);
+		handleSubscription(channel, SubscriptionSynchronizion::onChannelSubscribed);
+	}
+
+	@Override
+	public void onChannelUnsubscribed(byte[] channel, long count) {
+		subscriptionListener.onChannelUnsubscribed(channel, count);
+	}
+
+	@Override
+	public void onPatternSubscribed(byte[] pattern, long count) {
+
+		subscriptionListener.onPatternSubscribed(pattern, count);
+		handleSubscription(pattern, SubscriptionSynchronizion::onPatternSubscribed);
+	}
+
+	@Override
+	public void onPatternUnsubscribed(byte[] pattern, long count) {
+		subscriptionListener.onPatternUnsubscribed(pattern, count);
+	}
+
+	void handleSubscription(byte[] topic,
+			BiFunction<SubscriptionSynchronizion, ByteArrayWrapper, Boolean> synchronizerCallback) {
+
+		if (synchronizations.isEmpty()) {
+			return;
+		}
+
+		ByteArrayWrapper binaryChannel = new ByteArrayWrapper(topic);
+		List<SubscriptionSynchronizion> finalized = new ArrayList<>(synchronizations.size());
+
+		for (SubscriptionSynchronizion synchronizer : synchronizations) {
+
+			if (synchronizerCallback.apply(synchronizer, binaryChannel)) {
+				finalized.add(synchronizer);
+			}
+		}
+
+		synchronizations.removeAll(finalized);
+	}
+
+	/**
+	 * Synchronization to await subscriptions for channels and patterns.
+	 */
+	static class SubscriptionSynchronizion {
+
+		private static final AtomicIntegerFieldUpdater<SubscriptionSynchronizion> DONE = AtomicIntegerFieldUpdater
+				.newUpdater(SubscriptionSynchronizion.class, "done");
+
+		private static final int NOT_DONE = 0;
+		private static final int DONE_DONE = 0;
+
+		private volatile int done = NOT_DONE;
+		private final Set<ByteArrayWrapper> remainingPatterns;
+		private final Set<ByteArrayWrapper> remainingChannels;
+
+		private final Runnable doneCallback;
+
+		public SubscriptionSynchronizion(Collection<byte[]> remainingPatterns, Collection<byte[]> remainingChannels,
+				Runnable doneCallback) {
+
+			if (remainingPatterns.isEmpty()) {
+				this.remainingPatterns = Collections.emptySet();
+			} else {
+				this.remainingPatterns = ConcurrentHashMap.newKeySet(remainingPatterns.size());
+				this.remainingPatterns
+						.addAll(remainingPatterns.stream().map(ByteArrayWrapper::new).collect(Collectors.toList()));
+			}
+
+			if (remainingChannels.isEmpty()) {
+				this.remainingChannels = Collections.emptySet();
+			} else {
+				this.remainingChannels = ConcurrentHashMap.newKeySet(remainingChannels.size());
+				this.remainingChannels
+						.addAll(remainingChannels.stream().map(ByteArrayWrapper::new).collect(Collectors.toList()));
+			}
+
+			this.doneCallback = doneCallback;
+		}
+
+		boolean onChannelSubscribed(ByteArrayWrapper channel) {
+
+			if (DONE.get(this) == NOT_DONE) {
+				remainingChannels.remove(channel);
+				return postSubscribe();
+			}
+
+			return false;
+		}
+
+		boolean onPatternSubscribed(ByteArrayWrapper pattern) {
+
+			if (DONE.get(this) == NOT_DONE) {
+				remainingPatterns.remove(pattern);
+				return postSubscribe();
+			}
+
+			return false;
+		}
+
+		/**
+		 * @return whether the synchronization is finished and can be removed.
+		 */
+		private boolean postSubscribe() {
+
+			if (remainingChannels.isEmpty() && remainingPatterns.isEmpty() && DONE.compareAndSet(this, NOT_DONE, DONE_DONE)) {
+				this.doneCallback.run();
+
+				return true;
+			}
+
+			return false;
+		}
+	}
+}

src/test/java/org/springframework/data/redis/core/RedisKeyValueAdapterUnitTests.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2016-2021 the original author or authors.
+ * Copyright 2016-2022 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -13,7 +13,6 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
-
 package org.springframework.data.redis.core;
 
 import static org.assertj.core.api.Assertions.*;
@@ -38,6 +37,7 @@ import org.mockito.quality.Strictness;
 
 import org.springframework.data.annotation.Id;
 import org.springframework.data.redis.connection.RedisConnection;
+import org.springframework.data.redis.connection.SubscriptionListener;
 import org.springframework.data.redis.connection.jedis.JedisConnectionFactory;
 import org.springframework.data.redis.core.RedisKeyValueAdapter.EnableKeyspaceEvents;
 import org.springframework.data.redis.core.convert.Bucket;
@@ -72,6 +72,22 @@ class RedisKeyValueAdapterUnitTests {
 		template.setConnectionFactory(jedisConnectionFactoryMock);
 		template.afterPropertiesSet();
 
+		doAnswer(it -> {
+
+			SubscriptionListener listener = it.getArgument(0);
+			listener.onChannelSubscribed(it.getArgument(1), 0);
+
+			return null;
+		}).when(redisConnectionMock).subscribe(any(), any());
+
+		doAnswer(it -> {
+
+			SubscriptionListener listener = it.getArgument(0);
+			listener.onPatternSubscribed(it.getArgument(1), 0);
+
+			return null;
+		}).when(redisConnectionMock).pSubscribe(any(), any());
+
 		when(jedisConnectionFactoryMock.getConnection()).thenReturn(redisConnectionMock);
 
 		Properties keyspaceEventsConfig = new Properties();

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

@@ -33,7 +33,6 @@ import org.junit.jupiter.api.BeforeEach;
 
 import org.springframework.core.task.SimpleAsyncTaskExecutor;
 import org.springframework.core.task.SyncTaskExecutor;
-import org.springframework.data.redis.SettingsUtils;
 import org.springframework.data.redis.connection.RedisConnectionFactory;
 import org.springframework.data.redis.connection.jedis.JedisConnectionFactory;
 import org.springframework.data.redis.connection.jedis.extension.JedisConnectionFactoryExtension;
@@ -77,9 +76,6 @@ public class PubSubResubscribeTests {
 
 	public static Collection<Object[]> testParams() {
 
-		int port = SettingsUtils.getPort();
-		String host = SettingsUtils.getHost();
-
 		List<RedisConnectionFactory> factories = new ArrayList<>(3);
 
 		// Jedis

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2011-2021 the original author or authors.
+ * Copyright 2011-2022 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -34,7 +34,6 @@ import org.junit.jupiter.api.BeforeEach;
 import org.springframework.core.task.SimpleAsyncTaskExecutor;
 import org.springframework.core.task.SyncTaskExecutor;
 import org.springframework.data.redis.ObjectFactory;
-import org.springframework.data.redis.connection.ConnectionUtils;
 import org.springframework.data.redis.connection.RedisConnectionFactory;
 import org.springframework.data.redis.connection.jedis.JedisConnectionFactory;
 import org.springframework.data.redis.connection.lettuce.LettuceConnectionFactory;
@@ -171,18 +170,14 @@ public class PubSubTests<T> {
 	}
 
 	@SuppressWarnings("unchecked")
-	@ParameterizedRedisTest // DATAREDIS-251
+	@ParameterizedRedisTest // DATAREDIS-251, GH-964
 	void testStartListenersToNoSpecificChannelTest() throws InterruptedException {
 
 		assumeThat(isClusterAware(template.getConnectionFactory())).isFalse();
-		assumeThat(ConnectionUtils.isJedis(template.getConnectionFactory())).isTrue();
 
-		PubSubAwaitUtil.runAndAwaitPatternSubscription(template.getRequiredConnectionFactory(), () -> {
-
-			container.removeMessageListener(adapter, new ChannelTopic(CHANNEL));
-			container.addMessageListener(adapter, Collections.singletonList(new PatternTopic(CHANNEL + "*")));
-			container.start();
-		});
+		container.removeMessageListener(adapter, new ChannelTopic(CHANNEL));
+		container.addMessageListener(adapter, Collections.singletonList(new PatternTopic(CHANNEL + "*")));
+		container.start();
 
 		T payload = getT();
 

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2016-2021 the original author or authors.
+ * Copyright 2016-2022 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -18,6 +18,7 @@ package org.springframework.data.redis.listener;
 import static org.assertj.core.api.Assertions.*;
 import static org.mockito.Mockito.*;
 
+import java.util.concurrent.CompletionException;
 import java.util.concurrent.Executor;
 
 import org.junit.jupiter.api.AfterEach;
@@ -25,11 +26,12 @@ import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
 
 import org.springframework.core.task.SyncTaskExecutor;
+import org.springframework.data.redis.RedisConnectionFailureException;
 import org.springframework.data.redis.SettingsUtils;
 import org.springframework.data.redis.connection.RedisStandaloneConfiguration;
 import org.springframework.data.redis.connection.jedis.JedisConnectionFactory;
 import org.springframework.data.redis.listener.adapter.MessageListenerAdapter;
-import org.springframework.data.redis.test.extension.parametrized.MethodSource;
+import org.springframework.util.backoff.FixedBackOff;
 
 /**
  * Integration tests for {@link RedisMessageListenerContainer}.
@@ -37,7 +39,7 @@ import org.springframework.data.redis.test.extension.parametrized.MethodSource;
  * @author Mark Paluch
  * @author Christoph Strobl
  */
-class RedisMessageListenerContainerInterruptIntegrationTests {
+class RedisMessageListenerContainerFailureIntegrationTests {
 
 	private final Object handler = new Object() {
 
@@ -80,24 +82,44 @@ class RedisMessageListenerContainerInterruptIntegrationTests {
 		connectionFactory.destroy();
 	}
 
-	@Test // DATAREDIS-415
+	@Test // DATAREDIS-415, GH-964
 	void interruptAtStart() {
 
-		final Thread main = Thread.currentThread();
+		Thread main = Thread.currentThread();
 
 		// interrupt thread once Executor.execute is called
 		doAnswer(invocationOnMock -> {
 
 			main.interrupt();
-			return null;
+			throw new InterruptedException();
 		}).when(executorMock).execute(any(Runnable.class));
 
 		container.addMessageListener(adapter, new ChannelTopic("a"));
-		container.start();
+		assertThatThrownBy(() -> container.start()).isInstanceOf(CompletionException.class)
+				.hasRootCauseInstanceOf(InterruptedException.class);
 
 		// reset the interrupted flag to not destroy the teardown
-		assertThat(Thread.interrupted()).isTrue();
+		Thread.interrupted();
 
-		assertThat(container.isRunning()).isFalse();
+		assertThat(container.isRunning()).isTrue();
+		assertThat(container.isListening()).isFalse();
+	}
+
+	@Test // GH-964
+	void connectionFailureAndRetry() {
+
+		// interrupt thread once Executor.execute is called
+		doAnswer(invocationOnMock -> {
+
+			throw new RedisConnectionFailureException("I want to break free!");
+		}).when(executorMock).execute(any(Runnable.class));
+
+		container.setRecoveryBackoff(new FixedBackOff(1, 5));
+		container.addMessageListener(adapter, new ChannelTopic("a"));
+		assertThatThrownBy(() -> container.start()).isInstanceOf(CompletionException.class)
+				.hasRootCauseInstanceOf(RedisConnectionFailureException.class);
+
+		assertThat(container.isRunning()).isTrue();
+		assertThat(container.isListening()).isFalse();
 	}
 }

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2016-2021 the original author or authors.
+ * Copyright 2016-2022 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -26,6 +26,7 @@ import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicInteger;
 import java.util.concurrent.atomic.AtomicReference;
 
+import org.hamcrest.Matchers;
 import org.junit.jupiter.api.AfterEach;
 import org.junit.jupiter.api.BeforeEach;
 
@@ -213,6 +214,143 @@ class RedisMessageListenerContainerIntegrationTests {
 		assertThat(subscriptions1.get() + subscriptions2.get()).isGreaterThan(0);
 	}
 
+	@ParameterizedRedisTest // GH-964
+	void subscribeAfterStart() throws Exception {
+
+		AtomicInteger subscriptions1 = new AtomicInteger();
+		AtomicInteger subscriptions2 = new AtomicInteger();
+		CountDownLatch received = new CountDownLatch(2);
+
+		CompositeListener listener1 = new CompositeListener() {
+			@Override
+			public void onMessage(Message message, @Nullable byte[] pattern) {
+				received.countDown();
+			}
+
+			@Override
+			public void onPatternSubscribed(byte[] pattern, long count) {
+				subscriptions1.incrementAndGet();
+			}
+		};
+
+		CompositeListener listener2 = new CompositeListener() {
+			@Override
+			public void onMessage(Message message, @Nullable byte[] pattern) {
+				received.countDown();
+			}
+
+			@Override
+			public void onPatternSubscribed(byte[] pattern, long count) {
+				subscriptions2.incrementAndGet();
+			}
+		};
+
+		container.start();
+
+		container.addMessageListener(listener1, new PatternTopic("a"));
+		container.addMessageListener(listener2, new PatternTopic("a"));
+
+		try (RedisConnection connection = connectionFactory.getConnection()) {
+			connection.publish("a".getBytes(), "hello".getBytes());
+		}
+
+		assertThat(received.await(2, TimeUnit.SECONDS)).isTrue();
+		container.destroy();
+
+		await().until(() -> subscriptions1.get() > 0 || subscriptions2.get() > 0);
+
+		assertThat(subscriptions1.get() + subscriptions2.get()).isGreaterThan(0);
+	}
+
+	@ParameterizedRedisTest // GH-964
+	void multipleStarts() throws Exception {
+
+		AtomicInteger subscriptions = new AtomicInteger();
+		CountDownLatch received = new CountDownLatch(1);
+
+		CompositeListener listener1 = new CompositeListener() {
+			@Override
+			public void onMessage(Message message, @Nullable byte[] pattern) {
+				received.countDown();
+			}
+
+			@Override
+			public void onPatternSubscribed(byte[] pattern, long count) {
+				subscriptions.incrementAndGet();
+			}
+		};
+
+		container.start();
+		container.addMessageListener(listener1, new PatternTopic("a"));
+		container.stop();
+		container.start();
+
+		// Listeners run on a listener executor and they can be notified later
+		await().untilAtomic(subscriptions, Matchers.is(2));
+		assertThat(subscriptions.get()).isEqualTo(2);
+
+		try (RedisConnection connection = connectionFactory.getConnection()) {
+			connection.publish("a".getBytes(), "hello".getBytes());
+		}
+
+		assertThat(received.await(2, TimeUnit.SECONDS)).isTrue();
+		container.destroy();
+	}
+
+	@ParameterizedRedisTest // GH-964
+	void shouldRegisterChannelsAndTopics() throws Exception {
+
+		AtomicInteger subscriptions = new AtomicInteger();
+		CountDownLatch received = new CountDownLatch(2);
+
+		CompositeListener patternListener = new CompositeListener() {
+			@Override
+			public void onMessage(Message message, @Nullable byte[] pattern) {
+				if (message.toString().contains("pattern")) {
+					received.countDown();
+				}
+			}
+
+			@Override
+			public void onPatternSubscribed(byte[] pattern, long count) {
+				subscriptions.incrementAndGet();
+			}
+		};
+
+		CompositeListener channelListener = new CompositeListener() {
+			@Override
+			public void onMessage(Message message, @Nullable byte[] pattern) {
+				if (message.toString().contains("channel")) {
+					received.countDown();
+				}
+			}
+
+			@Override
+			public void onChannelSubscribed(byte[] channel, long count) {
+				subscriptions.incrementAndGet();
+			}
+		};
+
+		container.start();
+
+		container.addMessageListener(patternListener, new PatternTopic("a-pattern-0"));
+		container.addMessageListener(patternListener, new PatternTopic("a-pattern-1"));
+		container.addMessageListener(channelListener, new ChannelTopic("a-channel-0"));
+		container.addMessageListener(channelListener, new ChannelTopic("a-channel-1"));
+
+		// Listeners run on a listener executor and they can be notified later
+		await().untilAtomic(subscriptions, Matchers.is(4));
+		assertThat(subscriptions.get()).isEqualTo(4);
+
+		try (RedisConnection connection = connectionFactory.getConnection()) {
+			connection.publish("a-pattern-1".getBytes(), "pattern".getBytes());
+			connection.publish("a-channel-0".getBytes(), "channel".getBytes());
+		}
+
+		assertThat(received.await(2, TimeUnit.SECONDS)).isTrue();
+		container.destroy();
+	}
+
 	interface CompositeListener extends MessageListener, SubscriptionListener {
 
 	}

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2018-2021 the original author or authors.
+ * Copyright 2018-2022 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -18,6 +18,7 @@ package org.springframework.data.redis.listener;
 import static org.assertj.core.api.Assertions.*;
 import static org.mockito.Mockito.*;
 
+import java.nio.charset.StandardCharsets;
 import java.util.concurrent.Executor;
 
 import org.junit.jupiter.api.BeforeEach;
@@ -27,7 +28,8 @@ import org.springframework.core.task.SyncTaskExecutor;
 import org.springframework.data.redis.connection.RedisConnection;
 import org.springframework.data.redis.connection.RedisConnectionFactory;
 import org.springframework.data.redis.connection.Subscription;
-import org.springframework.data.redis.connection.lettuce.LettuceConnectionFactory;
+import org.springframework.data.redis.connection.SubscriptionListener;
+import org.springframework.data.redis.connection.jedis.JedisConnectionFactory;
 import org.springframework.data.redis.listener.adapter.MessageListenerAdapter;
 
 /**
@@ -57,7 +59,7 @@ class RedisMessageListenerContainerUnitTests {
 	void setUp() {
 
 		executorMock = mock(Executor.class);
-		connectionFactoryMock = mock(LettuceConnectionFactory.class);
+		connectionFactoryMock = mock(JedisConnectionFactory.class);
 		connectionMock = mock(RedisConnection.class);
 		subscriptionMock = mock(Subscription.class);
 
@@ -85,7 +87,11 @@ class RedisMessageListenerContainerUnitTests {
 
 		doAnswer(it -> {
 
+			SubscriptionListener listener = it.getArgument(0);
 			when(connectionMock.isSubscribed()).thenReturn(true);
+
+			listener.onChannelSubscribed("a".getBytes(StandardCharsets.UTF_8), 0);
+
 			return null;
 		}).when(connectionMock).subscribe(any(), any());
 

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2011-2021 the original author or authors.
+ * Copyright 2011-2022 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -27,7 +27,6 @@ import org.junit.jupiter.api.AfterEach;
 
 import org.springframework.core.task.SimpleAsyncTaskExecutor;
 import org.springframework.core.task.SyncTaskExecutor;
-import org.springframework.data.redis.SettingsUtils;
 import org.springframework.data.redis.connection.MessageListener;
 import org.springframework.data.redis.connection.RedisConnection;
 import org.springframework.data.redis.connection.RedisConnectionFactory;
@@ -70,8 +69,6 @@ public class SubscriptionConnectionTests {
 	}
 
 	public static Collection<Object[]> testParams() {
-		int port = SettingsUtils.getPort();
-		String host = SettingsUtils.getHost();
 
 		// Jedis
 		JedisConnectionFactory jedisConnFactory = JedisConnectionFactoryExtension
@@ -93,8 +90,9 @@ public class SubscriptionConnectionTests {
 		}
 	}
 
-	@ParameterizedRedisTest
+	@ParameterizedRedisTest // GH-964
 	void testStopMessageListenerContainers() throws Exception {
+
 		// Grab all 8 connections from the pool. They should be released on
 		// container stop
 		for (int i = 0; i < 8; i++) {
@@ -108,12 +106,6 @@ public class SubscriptionConnectionTests {
 			container.afterPropertiesSet();
 			container.start();
 
-			if (connectionFactory instanceof JedisConnectionFactory) {
-				// Need to sleep shortly as jedis cannot deal propery with multiple repsonses within one connection
-				// see https://github.com/xetorthio/jedis/issues/186
-				Thread.sleep(100);
-			}
-
 			container.stop();
 			containers.add(container);
 		}
@@ -125,6 +117,7 @@ public class SubscriptionConnectionTests {
 
 	@ParameterizedRedisTest
 	void testRemoveLastListener() throws Exception {
+
 		// Grab all 8 connections from the pool
 		MessageListener listener = new MessageListenerAdapter(handler);
 		for (int i = 0; i < 8; i++) {
@@ -150,6 +143,7 @@ public class SubscriptionConnectionTests {
 
 	@ParameterizedRedisTest
 	void testStopListening() throws InterruptedException {
+
 		// Grab all 8 connections from the pool.
 		MessageListener listener = new MessageListenerAdapter(handler);
 		for (int i = 0; i < 8; i++) {
@@ -165,7 +159,7 @@ public class SubscriptionConnectionTests {
 		}
 
 		// Unsubscribe all listeners from all topics, freeing up a connection
-		containers.get(0).removeMessageListener(null, Arrays.asList(new Topic[] {}));
+		containers.get(0).stop();
 
 		// verify we can now get a connection from the pool
 		RedisConnection connection = connectionFactory.getConnection();