Add support for subscription ready/unsubscription done events.

Original Pull Request: #2052

src/main/java/org/springframework/data/redis/connection/MessageListener.java

@@ -18,11 +18,14 @@ package org.springframework.data.redis.connection;
 import org.springframework.lang.Nullable;
 
 /**
- * Listener of messages published in Redis.
+ * Listener of messages published in Redis. A MessageListener can implement {@link SubscriptionListener} to receive
+ * notifications for subscription states.
  *
  * @author Costin Leau
  * @author Christoph Strobl
+ * @see SubscriptionListener
  */
+@FunctionalInterface
 public interface MessageListener {
 
 	/**

src/main/java/org/springframework/data/redis/connection/SubscriptionListener.java

@@ -0,0 +1,66 @@
+/*
+ * Copyright 2021 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.connection;
+
+/**
+ * Listener for subscription notifications.
+ * <p/>
+ * Subscription notifications are reported by Redis as confirmation for subscribe and unsubscribe operations for
+ * channels and patterns.
+ *
+ * @author Mark Paluch
+ * @since 2.6
+ */
+public interface SubscriptionListener {
+
+	/**
+	 * Empty {@link SubscriptionListener}.
+	 */
+	SubscriptionListener EMPTY = new SubscriptionListener() {};
+
+	/**
+	 * Notification when Redis has confirmed a channel subscription.
+	 *
+	 * @param channel name of the channel.
+	 * @param count subscriber count.
+	 */
+	default void onChannelSubscribed(byte[] channel, long count) {}
+
+	/**
+	 * Notification when Redis has confirmed a channel un-subscription.
+	 *
+	 * @param channel name of the channel.
+	 * @param count subscriber count.
+	 */
+	default void onChannelUnsubscribed(byte[] channel, long count) {}
+
+	/**
+	 * Notification when Redis has confirmed a pattern subscription.
+	 *
+	 * @param pattern the pattern.
+	 * @param count subscriber count.
+	 */
+	default void onPatternSubscribed(byte[] pattern, long count) {}
+
+	/**
+	 * Notification when Redis has confirmed a pattern un-subscription.
+	 *
+	 * @param pattern the pattern.
+	 * @param count subscriber count.
+	 */
+	default void onPatternUnsubscribed(byte[] pattern, long count) {}
+
+}

src/main/java/org/springframework/data/redis/connection/jedis/JedisClusterConnection.java

@@ -450,7 +450,7 @@ public class JedisClusterConnection implements DefaultedRedisClusterConnection {
 					"Connection already subscribed; use the connection Subscription to cancel or add new channels");
 		}
 		try {
-			BinaryJedisPubSub jedisPubSub = new JedisMessageListener(listener);
+			JedisMessageListener jedisPubSub = new JedisMessageListener(listener);
 			subscription = new JedisSubscription(listener, jedisPubSub, channels, null);
 			cluster.subscribe(jedisPubSub, channels);
 		} catch (Exception ex) {
@@ -466,7 +466,7 @@ public class JedisClusterConnection implements DefaultedRedisClusterConnection {
 					"Connection already subscribed; use the connection Subscription to cancel or add new channels");
 		}
 		try {
-			BinaryJedisPubSub jedisPubSub = new JedisMessageListener(listener);
+			JedisMessageListener jedisPubSub = new JedisMessageListener(listener);
 			subscription = new JedisSubscription(listener, jedisPubSub, null, patterns);
 			cluster.psubscribe(jedisPubSub, patterns);
 		} catch (Exception ex) {

src/main/java/org/springframework/data/redis/connection/jedis/JedisConnection.java

@@ -715,7 +715,7 @@ public class JedisConnection extends AbstractRedisConnection {
 
 		doWithJedis(it -> {
 
-			BinaryJedisPubSub jedisPubSub = new JedisMessageListener(listener);
+			JedisMessageListener jedisPubSub = new JedisMessageListener(listener);
 
 			subscription = new JedisSubscription(listener, jedisPubSub, null, patterns);
 			it.psubscribe(jedisPubSub, patterns);
@@ -740,7 +740,7 @@ public class JedisConnection extends AbstractRedisConnection {
 
 		doWithJedis(it -> {
 
-			BinaryJedisPubSub jedisPubSub = new JedisMessageListener(listener);
+			JedisMessageListener jedisPubSub = new JedisMessageListener(listener);
 
 			subscription = new JedisSubscription(listener, jedisPubSub, channels, null);
 			it.subscribe(jedisPubSub, channels);

src/main/java/org/springframework/data/redis/connection/jedis/JedisMessageListener.java

@@ -15,24 +15,29 @@
  */
 package org.springframework.data.redis.connection.jedis;
 
+import redis.clients.jedis.BinaryJedisPubSub;
+
 import org.springframework.data.redis.connection.DefaultMessage;
 import org.springframework.data.redis.connection.MessageListener;
+import org.springframework.data.redis.connection.SubscriptionListener;
 import org.springframework.util.Assert;
 
-import redis.clients.jedis.BinaryJedisPubSub;
-
 /**
  * MessageListener adapter on top of Jedis.
  *
  * @author Costin Leau
+ * @author Mark Paluch
  */
 class JedisMessageListener extends BinaryJedisPubSub {
 
 	private final MessageListener listener;
+	private final SubscriptionListener subscriptionListener;
 
 	JedisMessageListener(MessageListener listener) {
-		Assert.notNull(listener, "message listener is required");
+		Assert.notNull(listener, "MessageListener is required");
 		this.listener = listener;
+		this.subscriptionListener = listener instanceof SubscriptionListener ? (SubscriptionListener) listener
+				: SubscriptionListener.EMPTY;
 	}
 
 	public void onMessage(byte[] channel, byte[] message) {
@@ -44,18 +49,18 @@ class JedisMessageListener extends BinaryJedisPubSub {
 	}
 
 	public void onPSubscribe(byte[] pattern, int subscribedChannels) {
-		// no-op
+		subscriptionListener.onPatternSubscribed(pattern, subscribedChannels);
 	}
 
 	public void onPUnsubscribe(byte[] pattern, int subscribedChannels) {
-		// no-op
+		subscriptionListener.onPatternUnsubscribed(pattern, subscribedChannels);
 	}
 
 	public void onSubscribe(byte[] channel, int subscribedChannels) {
-		// no-op
+		subscriptionListener.onChannelSubscribed(channel, subscribedChannels);
 	}
 
 	public void onUnsubscribe(byte[] channel, int subscribedChannels) {
-		// no-op
+		subscriptionListener.onChannelUnsubscribed(channel, subscribedChannels);
 	}
 }

src/main/java/org/springframework/data/redis/connection/jedis/JedisSubscription.java

@@ -30,7 +30,7 @@ class JedisSubscription extends AbstractSubscription {
 
 	private final BinaryJedisPubSub jedisPubSub;
 
-	JedisSubscription(MessageListener listener, BinaryJedisPubSub jedisPubSub, @Nullable byte[][] channels,
+	JedisSubscription(MessageListener listener, JedisMessageListener jedisPubSub, @Nullable byte[][] channels,
 			@Nullable byte[][] patterns) {
 		super(listener, channels, patterns);
 		this.jedisPubSub = jedisPubSub;

src/main/java/org/springframework/data/redis/connection/lettuce/LettuceMessageListener.java

@@ -19,20 +19,26 @@ import io.lettuce.core.pubsub.RedisPubSubListener;
 
 import org.springframework.data.redis.connection.DefaultMessage;
 import org.springframework.data.redis.connection.MessageListener;
+import org.springframework.data.redis.connection.SubscriptionListener;
 import org.springframework.util.Assert;
 
 /**
  * MessageListener wrapper around Lettuce {@link RedisPubSubListener}.
  *
  * @author Costin Leau
+ * @author Mark Paluch
  */
 class LettuceMessageListener implements RedisPubSubListener<byte[], byte[]> {
 
 	private final MessageListener listener;
+	private final SubscriptionListener subscriptionListener;
 
 	LettuceMessageListener(MessageListener listener) {
 		Assert.notNull(listener, "MessageListener must not be null!");
+
 		this.listener = listener;
+		this.subscriptionListener = listener instanceof SubscriptionListener ? (SubscriptionListener) listener
+				: SubscriptionListener.EMPTY;
 	}
 
 	/*
@@ -55,23 +61,31 @@ class LettuceMessageListener implements RedisPubSubListener<byte[], byte[]> {
 	 * (non-Javadoc)
 	 * @see io.lettuce.core.pubsub.RedisPubSubListener#subscribed(java.lang.Object, long)
 	 */
-	public void subscribed(byte[] channel, long count) {}
+	public void subscribed(byte[] channel, long count) {
+		subscriptionListener.onChannelSubscribed(channel, count);
+	}
 
 	/*
 	 * (non-Javadoc)
 	 * @see io.lettuce.core.pubsub.RedisPubSubListener#psubscribed(java.lang.Object, long)
 	 */
-	public void psubscribed(byte[] pattern, long count) {}
+	public void psubscribed(byte[] pattern, long count) {
+		subscriptionListener.onPatternSubscribed(pattern, count);
+	}
 
 	/*
 	 * (non-Javadoc)
 	 * @see io.lettuce.core.pubsub.RedisPubSubListener#unsubscribed(java.lang.Object, long)
 	 */
-	public void unsubscribed(byte[] channel, long count) {}
+	public void unsubscribed(byte[] channel, long count) {
+		subscriptionListener.onChannelUnsubscribed(channel, count);
+	}
 
 	/*
 	 * (non-Javadoc)
 	 * @see io.lettuce.core.pubsub.RedisPubSubListener#punsubscribed(java.lang.Object, long)
 	 */
-	public void punsubscribed(byte[] pattern, long count) {}
+	public void punsubscribed(byte[] pattern, long count) {
+		subscriptionListener.onPatternUnsubscribed(pattern, count);
+	}
 }

src/main/java/org/springframework/data/redis/connection/lettuce/LettuceSubscription.java

@@ -16,8 +16,13 @@
 package org.springframework.data.redis.connection.lettuce;
 
 import io.lettuce.core.pubsub.StatefulRedisPubSubConnection;
+import io.lettuce.core.pubsub.api.async.RedisPubSubAsyncCommands;
 import io.lettuce.core.pubsub.api.sync.RedisPubSubCommands;
 
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.CompletableFuture;
+
 import org.springframework.data.redis.connection.MessageListener;
 import org.springframework.data.redis.connection.util.AbstractSubscription;
 
@@ -37,6 +42,7 @@ public class LettuceSubscription extends AbstractSubscription {
 	private final LettuceMessageListener listener;
 	private final LettuceConnectionProvider connectionProvider;
 	private final RedisPubSubCommands<byte[], byte[]> pubsub;
+	private final RedisPubSubAsyncCommands<byte[], byte[]> pubSubAsync;
 
 	/**
 	 * Creates a new {@link LettuceSubscription} given {@link MessageListener}, {@link StatefulRedisPubSubConnection}, and
@@ -55,6 +61,7 @@ public class LettuceSubscription extends AbstractSubscription {
 		this.listener = new LettuceMessageListener(listener);
 		this.connectionProvider = connectionProvider;
 		this.pubsub = connection.sync();
+		this.pubSubAsync = connection.async();
 
 		this.connection.addListener(this.listener);
 	}
@@ -70,15 +77,29 @@ public class LettuceSubscription extends AbstractSubscription {
 	@Override
 	protected void doClose() {
 
+		List<CompletableFuture<?>> futures = new ArrayList<>();
+
 		if (!getChannels().isEmpty()) {
-			doUnsubscribe(true);
+			futures.add(pubSubAsync.unsubscribe().toCompletableFuture());
 		}
 
 		if (!getPatterns().isEmpty()) {
-			doPUnsubscribe(true);
+			futures.add(pubSubAsync.punsubscribe().toCompletableFuture());
+		}
+
+		if (!futures.isEmpty()) {
+
+			// this is to ensure completion of the futures and result processing. Since we're unsubscribing first, we expect
+			// that we receive pub/sub confirmations before the PING response.
+			futures.add(pubSubAsync.ping().toCompletableFuture());
+
+			CompletableFuture.allOf(futures.toArray(new CompletableFuture[0])).whenComplete((v, t) -> {
+				connection.removeListener(listener);
+			});
+		} else {
+			connection.removeListener(listener);
 		}
 
-		connection.removeListener(this.listener);
 		connectionProvider.release(connection);
 	}
 

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

@@ -28,6 +28,7 @@ import java.util.concurrent.TimeUnit;
 
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
+
 import org.springframework.beans.factory.BeanNameAware;
 import org.springframework.beans.factory.DisposableBean;
 import org.springframework.beans.factory.InitializingBean;
@@ -41,6 +42,7 @@ import org.springframework.data.redis.connection.MessageListener;
 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.SubscriptionListener;
 import org.springframework.data.redis.connection.util.ByteArrayWrapper;
 import org.springframework.data.redis.serializer.RedisSerializer;
 import org.springframework.data.redis.serializer.StringRedisSerializer;
@@ -62,13 +64,16 @@ import org.springframework.util.ErrorHandler;
  * configured).
  * <p>
  * Adding and removing listeners at the same time has undefined results. It is strongly recommended to synchronize/order
- * these methods accordingly.
+ * these methods accordingly. {@link MessageListener Listeners} that wish to receive subscription/unsubscription
+ * callbacks in response to subscribe/unsubscribe commands can implement {@link SubscriptionListener}.
  *
  * @author Costin Leau
  * @author Jennifer Hickey
  * @author Way Joke
  * @author Thomas Darimont
  * @author Mark Paluch
+ * @see MessageListener
+ * @see SubscriptionListener
  */
 public class RedisMessageListenerContainer implements InitializingBean, DisposableBean, BeanNameAware, SmartLifecycle {
 
@@ -133,6 +138,7 @@ public class RedisMessageListenerContainer implements InitializingBean, Disposab
 
 	private long maxSubscriptionRegistrationWaitingTime = DEFAULT_SUBSCRIPTION_REGISTRATION_WAIT_TIME;
 
+	@Override
 	public void afterPropertiesSet() {
 		if (taskExecutor == null) {
 			manageExecutor = true;
@@ -958,7 +964,7 @@ public class RedisMessageListenerContainer implements InitializingBean, Disposab
 	 *
 	 * @author Costin Leau
 	 */
-	private class DispatchMessageListener implements MessageListener {
+	private class DispatchMessageListener implements MessageListener, SubscriptionListener {
 
 		@Override
 		public void onMessage(Message message, @Nullable byte[] pattern) {
@@ -977,6 +983,56 @@ public class RedisMessageListenerContainer implements InitializingBean, Disposab
 				dispatchMessage(listeners, message, pattern);
 			}
 		}
+
+		@Override
+		public void onChannelSubscribed(byte[] channel, long count) {
+			dispatchSubscriptionNotification(
+					channelMapping.getOrDefault(new ByteArrayWrapper(channel), Collections.emptyList()), channel, count,
+					SubscriptionListener::onChannelSubscribed);
+		}
+
+		@Override
+		public void onChannelUnsubscribed(byte[] channel, long count) {
+			dispatchSubscriptionNotification(
+					channelMapping.getOrDefault(new ByteArrayWrapper(channel), Collections.emptyList()), channel, count,
+					SubscriptionListener::onChannelUnsubscribed);
+		}
+
+		@Override
+		public void onPatternSubscribed(byte[] pattern, long count) {
+			dispatchSubscriptionNotification(
+					patternMapping.getOrDefault(new ByteArrayWrapper(pattern), Collections.emptyList()), pattern, count,
+					SubscriptionListener::onPatternSubscribed);
+		}
+
+		@Override
+		public void onPatternUnsubscribed(byte[] pattern, long count) {
+			dispatchSubscriptionNotification(
+					patternMapping.getOrDefault(new ByteArrayWrapper(pattern), Collections.emptyList()), pattern, count,
+					SubscriptionListener::onPatternUnsubscribed);
+		}
+	}
+
+	private void dispatchSubscriptionNotification(Collection<MessageListener> listeners, byte[] pattern, long count,
+			SubscriptionConsumer listenerConsumer) {
+
+		if (!CollectionUtils.isEmpty(listeners)) {
+			byte[] source = pattern.clone();
+
+			for (MessageListener messageListener : listeners) {
+				if (messageListener instanceof SubscriptionListener) {
+					taskExecutor.execute(() -> listenerConsumer.accept((SubscriptionListener) messageListener, source, count));
+				}
+			}
+		}
+	}
+
+	/**
+	 * 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) {

src/test/java/org/springframework/data/redis/connection/jedis/JedisSubscriptionUnitTests.java

@@ -39,7 +39,7 @@ import org.springframework.data.redis.connection.RedisInvalidSubscriptionExcepti
 @ExtendWith(MockitoExtension.class)
 class JedisSubscriptionUnitTests {
 
-	@Mock BinaryJedisPubSub jedisPubSub;
+	@Mock JedisMessageListener jedisPubSub;
 
 	@Mock MessageListener listener;
 

src/test/java/org/springframework/data/redis/connection/lettuce/LettuceSubscriptionUnitTests.java

@@ -18,10 +18,13 @@ package org.springframework.data.redis.connection.lettuce;
 import static org.assertj.core.api.Assertions.*;
 import static org.mockito.Mockito.*;
 
+import io.lettuce.core.RedisFuture;
 import io.lettuce.core.pubsub.StatefulRedisPubSubConnection;
+import io.lettuce.core.pubsub.api.async.RedisPubSubAsyncCommands;
 import io.lettuce.core.pubsub.api.sync.RedisPubSubCommands;
 
 import java.util.Collection;
+import java.util.concurrent.CompletableFuture;
 
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
@@ -42,7 +45,9 @@ class LettuceSubscriptionUnitTests {
 
 	private StatefulRedisPubSubConnection<byte[], byte[]> pubsub;
 
-	private RedisPubSubCommands<byte[], byte[]> asyncCommands;
+	private RedisPubSubCommands<byte[], byte[]> syncCommands;
+
+	private RedisPubSubAsyncCommands<byte[], byte[]> asyncCommands;
 
 	private LettuceConnectionProvider connectionProvider;
 
@@ -51,10 +56,12 @@ class LettuceSubscriptionUnitTests {
 	void setUp() {
 
 		pubsub = mock(StatefulRedisPubSubConnection.class);
-		asyncCommands = mock(RedisPubSubCommands.class);
+		syncCommands = mock(RedisPubSubCommands.class);
+		asyncCommands = mock(RedisPubSubAsyncCommands.class);
 		connectionProvider = mock(LettuceConnectionProvider.class);
 
-		when(pubsub.sync()).thenReturn(asyncCommands);
+		when(pubsub.sync()).thenReturn(syncCommands);
+		when(pubsub.async()).thenReturn(asyncCommands);
 		subscription = new LettuceSubscription(mock(MessageListener.class), pubsub, connectionProvider);
 	}
 
@@ -64,8 +71,8 @@ class LettuceSubscriptionUnitTests {
 		subscription.subscribe(new byte[][] { "a".getBytes() });
 		subscription.unsubscribe();
 
-		verify(asyncCommands).unsubscribe();
-		verify(asyncCommands, never()).punsubscribe();
+		verify(syncCommands).unsubscribe();
+		verify(syncCommands, never()).punsubscribe();
 		verify(connectionProvider).release(pubsub);
 		verify(pubsub).removeListener(any(LettuceMessageListener.class));
 
@@ -81,8 +88,8 @@ class LettuceSubscriptionUnitTests {
 		subscription.pSubscribe(new byte[][] { "s*".getBytes() });
 		subscription.unsubscribe();
 
-		verify(asyncCommands).unsubscribe();
-		verify(asyncCommands, never()).punsubscribe();
+		verify(syncCommands).unsubscribe();
+		verify(syncCommands, never()).punsubscribe();
 
 		assertThat(subscription.isAlive()).isTrue();
 		assertThat(subscription.getChannels()).isEmpty();
@@ -100,9 +107,9 @@ class LettuceSubscriptionUnitTests {
 		subscription.subscribe(channel);
 		subscription.unsubscribe(channel);
 
-		verify(asyncCommands).unsubscribe(channel);
-		verify(asyncCommands, never()).unsubscribe();
-		verify(asyncCommands, never()).punsubscribe();
+		verify(syncCommands).unsubscribe(channel);
+		verify(syncCommands, never()).unsubscribe();
+		verify(syncCommands, never()).punsubscribe();
 		verify(connectionProvider).release(pubsub);
 		verify(pubsub).removeListener(any(LettuceMessageListener.class));
 
@@ -119,9 +126,9 @@ class LettuceSubscriptionUnitTests {
 		subscription.subscribe(channels);
 		subscription.unsubscribe(new byte[][] { "a".getBytes() });
 
-		verify(asyncCommands).unsubscribe(new byte[][] { "a".getBytes() });
-		verify(asyncCommands, never()).unsubscribe();
-		verify(asyncCommands, never()).punsubscribe();
+		verify(syncCommands).unsubscribe(new byte[][] { "a".getBytes() });
+		verify(syncCommands, never()).unsubscribe();
+		verify(syncCommands, never()).punsubscribe();
 
 		assertThat(subscription.isAlive()).isTrue();
 
@@ -140,9 +147,9 @@ class LettuceSubscriptionUnitTests {
 		subscription.pSubscribe(new byte[][] { "s*".getBytes() });
 		subscription.unsubscribe(channel);
 
-		verify(asyncCommands).unsubscribe(channel);
-		verify(asyncCommands, never()).unsubscribe();
-		verify(asyncCommands, never()).punsubscribe();
+		verify(syncCommands).unsubscribe(channel);
+		verify(syncCommands, never()).unsubscribe();
+		verify(syncCommands, never()).punsubscribe();
 
 		assertThat(subscription.isAlive()).isTrue();
 		assertThat(subscription.getChannels()).isEmpty();
@@ -161,9 +168,9 @@ class LettuceSubscriptionUnitTests {
 		subscription.pSubscribe(new byte[][] { "s*".getBytes() });
 		subscription.unsubscribe(channel);
 
-		verify(asyncCommands).unsubscribe(channel);
-		verify(asyncCommands, never()).unsubscribe();
-		verify(asyncCommands, never()).punsubscribe();
+		verify(syncCommands).unsubscribe(channel);
+		verify(syncCommands, never()).unsubscribe();
+		verify(syncCommands, never()).punsubscribe();
 		assertThat(subscription.isAlive()).isTrue();
 
 		Collection<byte[]> channels = subscription.getChannels();
@@ -181,8 +188,8 @@ class LettuceSubscriptionUnitTests {
 		subscription.pSubscribe(new byte[][] { "s*".getBytes() });
 		subscription.unsubscribe();
 
-		verify(asyncCommands, never()).unsubscribe();
-		verify(asyncCommands, never()).punsubscribe();
+		verify(syncCommands, never()).unsubscribe();
+		verify(syncCommands, never()).punsubscribe();
 
 		assertThat(subscription.isAlive()).isTrue();
 		assertThat(subscription.getChannels()).isEmpty();
@@ -204,8 +211,8 @@ class LettuceSubscriptionUnitTests {
 		assertThat(subscription.isAlive()).isFalse();
 
 		subscription.unsubscribe();
-		verify(asyncCommands).unsubscribe();
-		verify(asyncCommands, never()).punsubscribe();
+		verify(syncCommands).unsubscribe();
+		verify(syncCommands, never()).punsubscribe();
 	}
 
 	@Test
@@ -226,8 +233,8 @@ class LettuceSubscriptionUnitTests {
 		subscription.pSubscribe(new byte[][] { "a*".getBytes() });
 		subscription.pUnsubscribe();
 
-		verify(asyncCommands, never()).unsubscribe();
-		verify(asyncCommands).punsubscribe();
+		verify(syncCommands, never()).unsubscribe();
+		verify(syncCommands).punsubscribe();
 		verify(connectionProvider).release(pubsub);
 		verify(pubsub).removeListener(any(LettuceMessageListener.class));
 
@@ -243,8 +250,8 @@ class LettuceSubscriptionUnitTests {
 		subscription.pSubscribe(new byte[][] { "s*".getBytes() });
 		subscription.pUnsubscribe();
 
-		verify(asyncCommands, never()).unsubscribe();
-		verify(asyncCommands).punsubscribe();
+		verify(syncCommands, never()).unsubscribe();
+		verify(syncCommands).punsubscribe();
 
 		assertThat(subscription.isAlive()).isTrue();
 		assertThat(subscription.getPatterns()).isEmpty();
@@ -262,9 +269,9 @@ class LettuceSubscriptionUnitTests {
 		subscription.pSubscribe(pattern);
 		subscription.pUnsubscribe(pattern);
 
-		verify(asyncCommands, never()).unsubscribe();
-		verify(asyncCommands, never()).punsubscribe();
-		verify(asyncCommands).punsubscribe(pattern);
+		verify(syncCommands, never()).unsubscribe();
+		verify(syncCommands, never()).punsubscribe();
+		verify(syncCommands).punsubscribe(pattern);
 		verify(connectionProvider).release(pubsub);
 		verify(pubsub).removeListener(any(LettuceMessageListener.class));
 
@@ -280,9 +287,9 @@ class LettuceSubscriptionUnitTests {
 		subscription.pSubscribe(patterns);
 		subscription.pUnsubscribe(new byte[][] { "a*".getBytes() });
 
-		verify(asyncCommands).punsubscribe(new byte[][] { "a*".getBytes() });
-		verify(asyncCommands, never()).unsubscribe();
-		verify(asyncCommands, never()).punsubscribe();
+		verify(syncCommands).punsubscribe(new byte[][] { "a*".getBytes() });
+		verify(syncCommands, never()).unsubscribe();
+		verify(syncCommands, never()).punsubscribe();
 
 		assertThat(subscription.isAlive()).isTrue();
 
@@ -301,9 +308,9 @@ class LettuceSubscriptionUnitTests {
 		subscription.pSubscribe(pattern);
 		subscription.pUnsubscribe(pattern);
 
-		verify(asyncCommands).punsubscribe(pattern);
-		verify(asyncCommands, never()).unsubscribe();
-		verify(asyncCommands, never()).punsubscribe();
+		verify(syncCommands).punsubscribe(pattern);
+		verify(syncCommands, never()).unsubscribe();
+		verify(syncCommands, never()).punsubscribe();
 
 		assertThat(subscription.isAlive()).isTrue();
 		assertThat(subscription.getPatterns()).isEmpty();
@@ -322,9 +329,9 @@ class LettuceSubscriptionUnitTests {
 		subscription.subscribe(new byte[][] { "a".getBytes() });
 		subscription.pUnsubscribe(pattern);
 
-		verify(asyncCommands, never()).unsubscribe();
-		verify(asyncCommands, never()).punsubscribe();
-		verify(asyncCommands).punsubscribe(pattern);
+		verify(syncCommands, never()).unsubscribe();
+		verify(syncCommands, never()).punsubscribe();
+		verify(syncCommands).punsubscribe(pattern);
 
 		assertThat(subscription.isAlive()).isTrue();
 
@@ -343,8 +350,8 @@ class LettuceSubscriptionUnitTests {
 		subscription.subscribe(new byte[][] { "s".getBytes() });
 		subscription.pUnsubscribe();
 
-		verify(asyncCommands, never()).unsubscribe();
-		verify(asyncCommands, never()).punsubscribe();
+		verify(syncCommands, never()).unsubscribe();
+		verify(syncCommands, never()).punsubscribe();
 		assertThat(subscription.isAlive()).isTrue();
 		assertThat(subscription.getPatterns()).isEmpty();
 
@@ -365,8 +372,8 @@ class LettuceSubscriptionUnitTests {
 
 		verify(connectionProvider).release(pubsub);
 		verify(pubsub).removeListener(any(LettuceMessageListener.class));
-		verify(asyncCommands).unsubscribe();
-		verify(asyncCommands, never()).punsubscribe();
+		verify(syncCommands).unsubscribe();
+		verify(syncCommands, never()).punsubscribe();
 	}
 
 	@Test
@@ -386,27 +393,41 @@ class LettuceSubscriptionUnitTests {
 
 		subscription.doClose();
 
-		verify(asyncCommands, never()).unsubscribe();
-		verify(asyncCommands, never()).punsubscribe();
+		verify(syncCommands, never()).unsubscribe();
+		verify(syncCommands, never()).punsubscribe();
 	}
 
 	@Test
 	void testDoCloseSubscribedChannels() {
 
+		RedisFuture<Void> future = mock(RedisFuture.class);
+		when(future.toCompletableFuture()).thenReturn(CompletableFuture.completedFuture(null));
+
+		when(asyncCommands.unsubscribe()).thenReturn(future);
+		when(asyncCommands.ping()).thenReturn((RedisFuture) future);
+
 		subscription.subscribe(new byte[][] { "a".getBytes() });
 		subscription.doClose();
 
+		verify(asyncCommands).ping();
 		verify(asyncCommands).unsubscribe();
-		verify(asyncCommands, never()).punsubscribe();
+		verifyNoMoreInteractions(asyncCommands);
 	}
 
 	@Test
 	void testDoCloseSubscribedPatterns() {
 
+		RedisFuture<Void> future = mock(RedisFuture.class);
+		when(future.toCompletableFuture()).thenReturn(CompletableFuture.completedFuture(null));
+
+		when(asyncCommands.punsubscribe()).thenReturn(future);
+		when(asyncCommands.ping()).thenReturn((RedisFuture) future);
+
 		subscription.pSubscribe(new byte[][] { "a*".getBytes() });
 		subscription.doClose();
 
-		verify(asyncCommands, never()).unsubscribe();
+		verify(asyncCommands).ping();
 		verify(asyncCommands).punsubscribe();
+		verifyNoMoreInteractions(asyncCommands);
 	}
 }

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

@@ -16,87 +16,202 @@
 package org.springframework.data.redis.listener;
 
 import static org.assertj.core.api.Assertions.*;
-import static org.mockito.Mockito.*;
 
-import java.util.concurrent.Executor;
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.AtomicReference;
 
 import org.junit.jupiter.api.AfterEach;
 import org.junit.jupiter.api.BeforeEach;
-import org.junit.jupiter.api.Test;
 
-import org.springframework.core.task.SyncTaskExecutor;
-import org.springframework.data.redis.SettingsUtils;
-import org.springframework.data.redis.connection.RedisStandaloneConfiguration;
+import org.springframework.data.redis.connection.Message;
+import org.springframework.data.redis.connection.MessageListener;
+import org.springframework.data.redis.connection.RedisConnection;
+import org.springframework.data.redis.connection.RedisConnectionFactory;
+import org.springframework.data.redis.connection.SubscriptionListener;
 import org.springframework.data.redis.connection.jedis.JedisConnectionFactory;
-import org.springframework.data.redis.listener.adapter.MessageListenerAdapter;
+import org.springframework.data.redis.connection.jedis.extension.JedisConnectionFactoryExtension;
+import org.springframework.data.redis.connection.lettuce.LettuceConnectionFactory;
+import org.springframework.data.redis.connection.lettuce.extension.LettuceConnectionFactoryExtension;
+import org.springframework.data.redis.test.extension.RedisStanalone;
+import org.springframework.data.redis.test.extension.parametrized.MethodSource;
+import org.springframework.data.redis.test.extension.parametrized.ParameterizedRedisTest;
+import org.springframework.lang.Nullable;
 
 /**
  * Integration tests for {@link RedisMessageListenerContainer}.
  *
  * @author Mark Paluch
- * @author Christoph Strobl
  */
+@MethodSource("testParams")
 class RedisMessageListenerContainerIntegrationTests {
 
-	private final Object handler = new Object() {
-
-		@SuppressWarnings("unused")
-		public void handleMessage(Object message) {}
-	};
-
-	private final MessageListenerAdapter adapter = new MessageListenerAdapter(handler);
-
-	private JedisConnectionFactory connectionFactory;
+	private RedisConnectionFactory connectionFactory;
 	private RedisMessageListenerContainer container;
 
-	private Executor executorMock;
+	public RedisMessageListenerContainerIntegrationTests(RedisConnectionFactory connectionFactory) {
+		this.connectionFactory = connectionFactory;
+	}
 
 	@BeforeEach
 	void setUp() {
 
-		executorMock = mock(Executor.class);
-
-		RedisStandaloneConfiguration configuration = new RedisStandaloneConfiguration();
-		configuration.setPort(SettingsUtils.getPort());
-		configuration.setHostName(SettingsUtils.getHost());
-		configuration.setDatabase(2);
-
-		connectionFactory = new JedisConnectionFactory(configuration);
-		connectionFactory.afterPropertiesSet();
-
 		container = new RedisMessageListenerContainer();
 		container.setConnectionFactory(connectionFactory);
 		container.setBeanName("container");
-		container.setTaskExecutor(new SyncTaskExecutor());
-		container.setSubscriptionExecutor(executorMock);
 		container.afterPropertiesSet();
 	}
 
+	public static Collection<Object[]> testParams() {
+
+		// Jedis
+		JedisConnectionFactory jedisConnFactory = JedisConnectionFactoryExtension
+				.getConnectionFactory(RedisStanalone.class);
+
+		// Lettuce
+		LettuceConnectionFactory lettuceConnFactory = LettuceConnectionFactoryExtension
+				.getConnectionFactory(RedisStanalone.class);
+
+		return Arrays.asList(new Object[][] { { jedisConnFactory }, { lettuceConnFactory } });
+	}
+
 	@AfterEach
 	void tearDown() throws Exception {
-
 		container.destroy();
-		connectionFactory.destroy();
 	}
 
-	@Test // DATAREDIS-415
-	void interruptAtStart() {
+	@ParameterizedRedisTest
+	void notifiesChannelSubscriptionState() throws Exception {
 
-		final Thread main = Thread.currentThread();
+		AtomicReference<String> onSubscribe = new AtomicReference<>();
+		AtomicReference<String> onUnsubscribe = new AtomicReference<>();
+		CompletableFuture<Void> subscribe = new CompletableFuture<>();
+		CompletableFuture<Void> unsubscribe = new CompletableFuture<>();
 
-		// interrupt thread once Executor.execute is called
-		doAnswer(invocationOnMock -> {
+		CompositeListener listener = new CompositeListener() {
+			@Override
+			public void onMessage(Message message, @Nullable byte[] pattern) {
 
-			main.interrupt();
-			return null;
-		}).when(executorMock).execute(any(Runnable.class));
+			}
 
-		container.addMessageListener(adapter, new ChannelTopic("a"));
+			@Override
+			public void onChannelSubscribed(byte[] channel, long count) {
+				onSubscribe.set(new String(channel));
+				subscribe.complete(null);
+			}
+
+			@Override
+			public void onChannelUnsubscribed(byte[] channel, long count) {
+				onUnsubscribe.set(new String(channel));
+				unsubscribe.complete(null);
+			}
+		};
+
+		container.addMessageListener(listener, new ChannelTopic("a"));
 		container.start();
 
-		// reset the interrupted flag to not destroy the teardown
-		assertThat(Thread.interrupted()).isTrue();
+		subscribe.get(10, TimeUnit.SECONDS);
+
+		container.destroy();
+
+		unsubscribe.get(10, TimeUnit.SECONDS);
+
+		assertThat(onSubscribe).hasValue("a");
+		assertThat(onUnsubscribe).hasValue("a");
+	}
+
+	@ParameterizedRedisTest
+	void notifiesPatternSubscriptionState() throws Exception {
+
+		AtomicReference<String> onPsubscribe = new AtomicReference<>();
+		AtomicReference<String> onPunsubscribe = new AtomicReference<>();
+		CompletableFuture<Void> psubscribe = new CompletableFuture<>();
+		CompletableFuture<Void> punsubscribe = new CompletableFuture<>();
+
+		CompositeListener listener = new CompositeListener() {
+			@Override
+			public void onMessage(Message message, @Nullable byte[] pattern) {
+
+			}
+
+			@Override
+			public void onPatternSubscribed(byte[] pattern, long count) {
+				onPsubscribe.set(new String(pattern));
+				psubscribe.complete(null);
+			}
+
+			@Override
+			public void onPatternUnsubscribed(byte[] pattern, long count) {
+				onPunsubscribe.set(new String(pattern));
+				punsubscribe.complete(null);
+			}
+		};
+
+		container.addMessageListener(listener, new PatternTopic("a"));
+		container.start();
+
+		psubscribe.get(10, TimeUnit.SECONDS);
+
+		container.destroy();
+
+		punsubscribe.get(10, TimeUnit.SECONDS);
+
+		assertThat(onPsubscribe).hasValue("a");
+		assertThat(onPunsubscribe).hasValue("a");
+	}
+
+	@ParameterizedRedisTest
+	void repeatedSubscribeShouldNotifyOnlyOnce() 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.addMessageListener(listener1, new PatternTopic("a"));
+		container.addMessageListener(listener2, new PatternTopic("a"));
+
+		container.start();
+
+		try (RedisConnection connection = connectionFactory.getConnection()) {
+			connection.publish("a".getBytes(), "hello".getBytes());
+		}
+
+		received.await(2, TimeUnit.SECONDS);
+		container.destroy();
+
+		assertThat(subscriptions1).hasValue(1);
+		assertThat(subscriptions2).hasValue(1);
+	}
+
+	interface CompositeListener extends MessageListener, SubscriptionListener {
 
-		assertThat(container.isRunning()).isFalse();
 	}
 }

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

@@ -0,0 +1,103 @@
+/*
+ * Copyright 2016-2021 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 static org.assertj.core.api.Assertions.*;
+import static org.mockito.Mockito.*;
+
+import java.util.concurrent.Executor;
+
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+import org.springframework.core.task.SyncTaskExecutor;
+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;
+
+/**
+ * Integration tests for {@link RedisMessageListenerContainer}.
+ *
+ * @author Mark Paluch
+ * @author Christoph Strobl
+ */
+class RedisMessageListenerContainerInterruptIntegrationTests {
+
+	private final Object handler = new Object() {
+
+		@SuppressWarnings("unused")
+		public void handleMessage(Object message) {}
+	};
+
+	private final MessageListenerAdapter adapter = new MessageListenerAdapter(handler);
+
+	private JedisConnectionFactory connectionFactory;
+	private RedisMessageListenerContainer container;
+
+	private Executor executorMock;
+
+	@BeforeEach
+	void setUp() {
+
+		executorMock = mock(Executor.class);
+
+		RedisStandaloneConfiguration configuration = new RedisStandaloneConfiguration();
+		configuration.setPort(SettingsUtils.getPort());
+		configuration.setHostName(SettingsUtils.getHost());
+		configuration.setDatabase(2);
+
+		connectionFactory = new JedisConnectionFactory(configuration);
+		connectionFactory.afterPropertiesSet();
+
+		container = new RedisMessageListenerContainer();
+		container.setConnectionFactory(connectionFactory);
+		container.setBeanName("container");
+		container.setTaskExecutor(new SyncTaskExecutor());
+		container.setSubscriptionExecutor(executorMock);
+		container.afterPropertiesSet();
+	}
+
+	@AfterEach
+	void tearDown() throws Exception {
+
+		container.destroy();
+		connectionFactory.destroy();
+	}
+
+	@Test // DATAREDIS-415
+	void interruptAtStart() {
+
+		final Thread main = Thread.currentThread();
+
+		// interrupt thread once Executor.execute is called
+		doAnswer(invocationOnMock -> {
+
+			main.interrupt();
+			return null;
+		}).when(executorMock).execute(any(Runnable.class));
+
+		container.addMessageListener(adapter, new ChannelTopic("a"));
+		container.start();
+
+		// reset the interrupted flag to not destroy the teardown
+		assertThat(Thread.interrupted()).isTrue();
+
+		assertThat(container.isRunning()).isFalse();
+	}
+}