diff --git a/src/main/java/org/springframework/data/redis/connection/MessageListener.java b/src/main/java/org/springframework/data/redis/connection/MessageListener.java index ebc557543..069b42b1e 100644 --- a/src/main/java/org/springframework/data/redis/connection/MessageListener.java +++ b/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 { /** diff --git a/src/main/java/org/springframework/data/redis/connection/SubscriptionListener.java b/src/main/java/org/springframework/data/redis/connection/SubscriptionListener.java new file mode 100644 index 000000000..1f5f616a9 --- /dev/null +++ b/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. + *

+ * 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) {} + +} diff --git a/src/main/java/org/springframework/data/redis/connection/jedis/JedisClusterConnection.java b/src/main/java/org/springframework/data/redis/connection/jedis/JedisClusterConnection.java index 6bd6fcbfa..cde11d992 100644 --- a/src/main/java/org/springframework/data/redis/connection/jedis/JedisClusterConnection.java +++ b/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) { diff --git a/src/main/java/org/springframework/data/redis/connection/jedis/JedisConnection.java b/src/main/java/org/springframework/data/redis/connection/jedis/JedisConnection.java index 1524bff2d..9a8563915 100644 --- a/src/main/java/org/springframework/data/redis/connection/jedis/JedisConnection.java +++ b/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); diff --git a/src/main/java/org/springframework/data/redis/connection/jedis/JedisMessageListener.java b/src/main/java/org/springframework/data/redis/connection/jedis/JedisMessageListener.java index 6d8e06cf2..114d96749 100644 --- a/src/main/java/org/springframework/data/redis/connection/jedis/JedisMessageListener.java +++ b/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); } } diff --git a/src/main/java/org/springframework/data/redis/connection/jedis/JedisSubscription.java b/src/main/java/org/springframework/data/redis/connection/jedis/JedisSubscription.java index 915c4dba4..5358b9f95 100644 --- a/src/main/java/org/springframework/data/redis/connection/jedis/JedisSubscription.java +++ b/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; diff --git a/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceMessageListener.java b/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceMessageListener.java index dafe89c22..67ba6e3bc 100644 --- a/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceMessageListener.java +++ b/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 { 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 { * (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); + } } diff --git a/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceSubscription.java b/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceSubscription.java index c2fd9472b..05d977c22 100644 --- a/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceSubscription.java +++ b/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 pubsub; + private final RedisPubSubAsyncCommands 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> 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); } diff --git a/src/main/java/org/springframework/data/redis/listener/RedisMessageListenerContainer.java b/src/main/java/org/springframework/data/redis/listener/RedisMessageListenerContainer.java index 99dabda10..06de5aff8 100644 --- a/src/main/java/org/springframework/data/redis/listener/RedisMessageListenerContainer.java +++ b/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). *

* 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 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 listeners, Message message, @Nullable byte[] pattern) { diff --git a/src/test/java/org/springframework/data/redis/connection/jedis/JedisSubscriptionUnitTests.java b/src/test/java/org/springframework/data/redis/connection/jedis/JedisSubscriptionUnitTests.java index 27e9b71c4..beec82b75 100644 --- a/src/test/java/org/springframework/data/redis/connection/jedis/JedisSubscriptionUnitTests.java +++ b/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; diff --git a/src/test/java/org/springframework/data/redis/connection/lettuce/LettuceSubscriptionUnitTests.java b/src/test/java/org/springframework/data/redis/connection/lettuce/LettuceSubscriptionUnitTests.java index 51a513124..124cc9024 100644 --- a/src/test/java/org/springframework/data/redis/connection/lettuce/LettuceSubscriptionUnitTests.java +++ b/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 pubsub; - private RedisPubSubCommands asyncCommands; + private RedisPubSubCommands syncCommands; + + private RedisPubSubAsyncCommands 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 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 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 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); } } diff --git a/src/test/java/org/springframework/data/redis/listener/RedisMessageListenerContainerIntegrationTests.java b/src/test/java/org/springframework/data/redis/listener/RedisMessageListenerContainerIntegrationTests.java index 1d0b83f86..535424be3 100644 --- a/src/test/java/org/springframework/data/redis/listener/RedisMessageListenerContainerIntegrationTests.java +++ b/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 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 onSubscribe = new AtomicReference<>(); + AtomicReference onUnsubscribe = new AtomicReference<>(); + CompletableFuture subscribe = new CompletableFuture<>(); + CompletableFuture 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 onPsubscribe = new AtomicReference<>(); + AtomicReference onPunsubscribe = new AtomicReference<>(); + CompletableFuture psubscribe = new CompletableFuture<>(); + CompletableFuture 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(); } } diff --git a/src/test/java/org/springframework/data/redis/listener/RedisMessageListenerContainerInterruptIntegrationTests.java b/src/test/java/org/springframework/data/redis/listener/RedisMessageListenerContainerInterruptIntegrationTests.java new file mode 100644 index 000000000..ae347e1ba --- /dev/null +++ b/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(); + } +}