diff --git a/src/main/java/org/springframework/data/redis/connection/ReactivePubSubCommands.java b/src/main/java/org/springframework/data/redis/connection/ReactivePubSubCommands.java index bf2b86017..38ce433d9 100644 --- a/src/main/java/org/springframework/data/redis/connection/ReactivePubSubCommands.java +++ b/src/main/java/org/springframework/data/redis/connection/ReactivePubSubCommands.java @@ -37,7 +37,18 @@ public interface ReactivePubSubCommands { * * @return the subscription. */ - Mono createSubscription(); + default Mono createSubscription() { + return createSubscription(SubscriptionListener.EMPTY); + } + + /** + * Creates a subscription for this connection. Connections can have multiple {@link ReactiveSubscription}s. + * + * @param subscriptionListener the subscription listener to listen for subscription confirmations. + * @return the subscription. + * @since 2.6 + */ + Mono createSubscription(SubscriptionListener subscriptionListener); /** * Publishes the given {@code message} to the given {@code channel}. diff --git a/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceByteBufferPubSubListenerWrapper.java b/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceByteBufferPubSubListenerWrapper.java new file mode 100644 index 000000000..03814c046 --- /dev/null +++ b/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceByteBufferPubSubListenerWrapper.java @@ -0,0 +1,112 @@ +/* + * 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.lettuce; + +import io.lettuce.core.pubsub.RedisPubSubListener; + +import java.nio.ByteBuffer; + +import org.springframework.lang.Nullable; +import org.springframework.util.Assert; + +/** + * Wrapper around {@link RedisPubSubListener} that converts {@link ByteBuffer} into {@code byte[]}. + * + * @author Mark Paluch + * @since 2.6 + */ +class LettuceByteBufferPubSubListenerWrapper implements RedisPubSubListener { + + private final RedisPubSubListener delegate; + + LettuceByteBufferPubSubListenerWrapper(RedisPubSubListener delegate) { + + Assert.notNull(delegate, "RedisPubSubListener must not be null!"); + + this.delegate = delegate; + } + + /* + * (non-Javadoc) + * @see io.lettuce.core.pubsub.RedisPubSubListener#message(java.lang.Object, java.lang.Object) + */ + public void message(ByteBuffer channel, ByteBuffer message) { + delegate.message(getBytes(channel), getBytes(message)); + } + + /* + * (non-Javadoc) + * @see io.lettuce.core.pubsub.RedisPubSubListener#message(java.lang.Object, java.lang.Object, java.lang.Object) + */ + public void message(ByteBuffer pattern, ByteBuffer channel, ByteBuffer message) { + delegate.message(getBytes(channel), getBytes(message), getBytes(pattern)); + } + + /* + * (non-Javadoc) + * @see io.lettuce.core.pubsub.RedisPubSubListener#subscribed(java.lang.Object, long) + */ + public void subscribed(ByteBuffer channel, long count) { + delegate.subscribed(getBytes(channel), count); + } + + /* + * (non-Javadoc) + * @see io.lettuce.core.pubsub.RedisPubSubListener#psubscribed(java.lang.Object, long) + */ + public void psubscribed(ByteBuffer pattern, long count) { + delegate.psubscribed(getBytes(pattern), count); + } + + /* + * (non-Javadoc) + * @see io.lettuce.core.pubsub.RedisPubSubListener#unsubscribed(java.lang.Object, long) + */ + public void unsubscribed(ByteBuffer channel, long count) { + delegate.unsubscribed(getBytes(channel), count); + } + + /* + * (non-Javadoc) + * @see io.lettuce.core.pubsub.RedisPubSubListener#punsubscribed(java.lang.Object, long) + */ + public void punsubscribed(ByteBuffer pattern, long count) { + delegate.punsubscribed(getBytes(pattern), count); + } + + /** + * Extract a byte array from {@link ByteBuffer} without consuming it. + * + * @param byteBuffer must not be {@literal null}. + * @return + */ + private static byte[] getBytes(@Nullable ByteBuffer byteBuffer) { + + if (byteBuffer == null) { + return new byte[0]; + } + + if (byteBuffer.hasArray()) { + return byteBuffer.array(); + } + + byteBuffer.mark(); + byte[] bytes = new byte[byteBuffer.remaining()]; + byteBuffer.get(bytes); + byteBuffer.reset(); + return bytes; + } +} 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 67ba6e3bc..65bee4a65 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 @@ -33,12 +33,13 @@ class LettuceMessageListener implements RedisPubSubListener { private final MessageListener listener; private final SubscriptionListener subscriptionListener; - LettuceMessageListener(MessageListener listener) { + LettuceMessageListener(MessageListener listener, SubscriptionListener subscriptionListener) { + Assert.notNull(listener, "MessageListener must not be null!"); + Assert.notNull(subscriptionListener, "SubscriptionListener must not be null!"); this.listener = listener; - this.subscriptionListener = listener instanceof SubscriptionListener ? (SubscriptionListener) listener - : SubscriptionListener.EMPTY; + this.subscriptionListener = subscriptionListener; } /* diff --git a/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceReactivePubSubCommands.java b/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceReactivePubSubCommands.java index 45b8106c9..e10c53d72 100644 --- a/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceReactivePubSubCommands.java +++ b/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceReactivePubSubCommands.java @@ -23,9 +23,11 @@ import java.nio.ByteBuffer; import java.util.function.Function; import org.reactivestreams.Publisher; + import org.springframework.data.redis.connection.ReactivePubSubCommands; import org.springframework.data.redis.connection.ReactiveSubscription; import org.springframework.data.redis.connection.ReactiveSubscription.ChannelMessage; +import org.springframework.data.redis.connection.SubscriptionListener; import org.springframework.util.Assert; /** @@ -43,13 +45,13 @@ class LettuceReactivePubSubCommands implements ReactivePubSubCommands { /* * (non-Javadoc) - * @see org.springframework.data.redis.connection.ReactivePubSubCommands#createSubscription() + * @see org.springframework.data.redis.connection.ReactivePubSubCommands#createSubscription(org.springframework.data.redis.connection.SubscriptionListener) */ @Override - public Mono createSubscription() { + public Mono createSubscription(SubscriptionListener listener) { return connection.getPubSubConnection() - .map(pubSubConnection -> new LettuceReactiveSubscription(pubSubConnection.reactive(), + .map(pubSubConnection -> new LettuceReactiveSubscription(listener, pubSubConnection, connection.translateException())); } diff --git a/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceReactiveSubscription.java b/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceReactiveSubscription.java index d4dc56038..12c782138 100644 --- a/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceReactiveSubscription.java +++ b/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceReactiveSubscription.java @@ -15,6 +15,7 @@ */ package org.springframework.data.redis.connection.lettuce; +import io.lettuce.core.pubsub.StatefulRedisPubSubConnection; import io.lettuce.core.pubsub.api.reactive.RedisPubSubReactiveCommands; import reactor.core.Disposable; import reactor.core.publisher.ConnectableFlux; @@ -33,6 +34,7 @@ import java.util.function.Function; import java.util.function.Supplier; import org.springframework.data.redis.connection.ReactiveSubscription; +import org.springframework.data.redis.connection.SubscriptionListener; import org.springframework.lang.Nullable; import org.springframework.util.Assert; import org.springframework.util.ObjectUtils; @@ -46,15 +48,23 @@ import org.springframework.util.ObjectUtils; */ class LettuceReactiveSubscription implements ReactiveSubscription { + private final LettuceByteBufferPubSubListenerWrapper listener; + private final StatefulRedisPubSubConnection connection; private final RedisPubSubReactiveCommands commands; private final State patternState; private final State channelState; - LettuceReactiveSubscription(RedisPubSubReactiveCommands commands, + LettuceReactiveSubscription(SubscriptionListener subscriptionListener, + StatefulRedisPubSubConnection connection, Function exceptionTranslator) { - this.commands = commands; + this.listener = new LettuceByteBufferPubSubListenerWrapper( + new LettuceMessageListener((messages, pattern) -> {}, subscriptionListener)); + this.connection = connection; + this.commands = connection.reactive(); + connection.addListener(listener); + this.patternState = new State(exceptionTranslator); this.channelState = new State(exceptionTranslator); } @@ -176,7 +186,12 @@ class LettuceReactiveSubscription implements ReactiveSubscription { channelState.terminate(); patternState.terminate(); - return Mono.empty(); + + // 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. + return commands.ping().then(Mono.fromRunnable(() -> { + connection.removeListener(listener); + })); })); } 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 05d977c22..b8dfd3727 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 @@ -24,6 +24,7 @@ import java.util.List; import java.util.concurrent.CompletableFuture; import org.springframework.data.redis.connection.MessageListener; +import org.springframework.data.redis.connection.SubscriptionListener; import org.springframework.data.redis.connection.util.AbstractSubscription; /** @@ -58,7 +59,8 @@ public class LettuceSubscription extends AbstractSubscription { super(listener); this.connection = pubsubConnection; - this.listener = new LettuceMessageListener(listener); + this.listener = new LettuceMessageListener(listener, + listener instanceof SubscriptionListener ? (SubscriptionListener) listener : SubscriptionListener.EMPTY); this.connectionProvider = connectionProvider; this.pubsub = connection.sync(); this.pubSubAsync = connection.async(); diff --git a/src/main/java/org/springframework/data/redis/listener/ReactiveRedisMessageListenerContainer.java b/src/main/java/org/springframework/data/redis/listener/ReactiveRedisMessageListenerContainer.java index 78a6d0fdc..eeb763052 100644 --- a/src/main/java/org/springframework/data/redis/listener/ReactiveRedisMessageListenerContainer.java +++ b/src/main/java/org/springframework/data/redis/listener/ReactiveRedisMessageListenerContainer.java @@ -40,6 +40,7 @@ import org.springframework.data.redis.connection.ReactiveSubscription; import org.springframework.data.redis.connection.ReactiveSubscription.ChannelMessage; import org.springframework.data.redis.connection.ReactiveSubscription.Message; import org.springframework.data.redis.connection.ReactiveSubscription.PatternMessage; +import org.springframework.data.redis.connection.SubscriptionListener; import org.springframework.data.redis.serializer.RedisElementReader; import org.springframework.data.redis.serializer.RedisSerializationContext.SerializationPair; import org.springframework.data.redis.serializer.RedisSerializer; @@ -178,21 +179,65 @@ public class ReactiveRedisMessageListenerContainer implements DisposableBean { } /** - * Subscribe to one or more {@link Topic}s and receive a stream of {@link ChannelMessage} The stream may contain + * Subscribe to one or more {@link Topic}s and receive a stream of {@link ChannelMessage}. The stream may contain * {@link PatternMessage} if subscribed to patterns. Messages, and channel names are serialized/deserialized using the * given {@code channelSerializer} and {@code messageSerializer}. The message stream subscribes lazily to the Redis * channels and unsubscribes if the {@link org.reactivestreams.Subscription} is * {@link org.reactivestreams.Subscription#cancel() cancelled}. * - * @param topics the channels to subscribe. + * @param topics the channels/patterns to subscribe. + * @param subscriptionListener listener to receive subscription/unsubscription notifications. + * @return the message stream. + * @throws InvalidDataAccessApiUsageException if {@code patternTopics} is empty. + * @see #receive(Iterable, SerializationPair, SerializationPair) + * @since 2.6 + */ + public Flux> receive(Iterable topics, + SubscriptionListener subscriptionListener) { + return receive(topics, stringSerializationPair, stringSerializationPair, subscriptionListener); + } + + /** + * Subscribe to one or more {@link Topic}s and receive a stream of {@link ChannelMessage}. The stream may contain + * {@link PatternMessage} if subscribed to patterns. Messages, and channel names are serialized/deserialized using the + * given {@code channelSerializer} and {@code messageSerializer}. The message stream subscribes lazily to the Redis + * channels and unsubscribes if the {@link org.reactivestreams.Subscription} is + * {@link org.reactivestreams.Subscription#cancel() cancelled}. + * + * @param topics the channels/patterns to subscribe. * @return the message stream. * @see #receive(Iterable, SerializationPair, SerializationPair) * @throws InvalidDataAccessApiUsageException if {@code topics} is empty. */ public Flux> receive(Iterable topics, SerializationPair channelSerializer, SerializationPair messageSerializer) { + return receive(topics, channelSerializer, messageSerializer, SubscriptionListener.EMPTY); + } + + /** + * Subscribe to one or more {@link Topic}s and receive a stream of {@link ChannelMessage}. The stream may contain + * {@link PatternMessage} if subscribed to patterns. Messages, and channel names are serialized/deserialized using the + * given {@code channelSerializer} and {@code messageSerializer}. The message stream subscribes lazily to the Redis + * channels and unsubscribes if the {@link org.reactivestreams.Subscription} is + * {@link org.reactivestreams.Subscription#cancel() cancelled}. {@link SubscriptionListener} is notified upon + * subscription/unsubscription and can be used for synchronization. + * + * @param topics the channels to subscribe. + * @param channelSerializer + * @param messageSerializer + * @param subscriptionListener listener to receive subscription/unsubscription notifications. + * @return the message stream. + * @see #receive(Iterable, SerializationPair, SerializationPair) + * @throws InvalidDataAccessApiUsageException if {@code topics} is empty. + * @since 2.6 + */ + public Flux> receive(Iterable topics, SerializationPair channelSerializer, + SerializationPair messageSerializer, SubscriptionListener subscriptionListener) { Assert.notNull(topics, "Topics must not be null!"); + Assert.notNull(channelSerializer, "Channel serializer must not be null!"); + Assert.notNull(messageSerializer, "Message serializer must not be null!"); + Assert.notNull(subscriptionListener, "SubscriptionListener must not be null!"); verifyConnection(); @@ -203,7 +248,8 @@ public class ReactiveRedisMessageListenerContainer implements DisposableBean { throw new InvalidDataAccessApiUsageException("No channels or patterns to subscribe to."); } - return doReceive(channelSerializer, messageSerializer, connection.pubSubCommands().createSubscription(), patterns, + return doReceive(channelSerializer, messageSerializer, + connection.pubSubCommands().createSubscription(subscriptionListener), patterns, channels); } @@ -226,7 +272,7 @@ public class ReactiveRedisMessageListenerContainer implements DisposableBean { Subscribers subscribers = getSubscribers(it); if (subscribers.unregister()) { subscriptions.remove(it); - it.unsubscribe().subscribe(v -> terminalProcessor.onComplete(), terminalProcessor::onError); + it.cancel().subscribe(v -> terminalProcessor.onComplete(), terminalProcessor::onError); } }).mergeWith(terminalProcessor); }); diff --git a/src/test/java/org/springframework/data/redis/connection/lettuce/LettuceReactiveSubscriptionUnitTests.java b/src/test/java/org/springframework/data/redis/connection/lettuce/LettuceReactiveSubscriptionUnitTests.java index 1ee50d8a0..8efb4d30c 100644 --- a/src/test/java/org/springframework/data/redis/connection/lettuce/LettuceReactiveSubscriptionUnitTests.java +++ b/src/test/java/org/springframework/data/redis/connection/lettuce/LettuceReactiveSubscriptionUnitTests.java @@ -21,6 +21,7 @@ import static org.mockito.Mockito.*; import static org.springframework.data.redis.util.ByteUtils.*; import io.lettuce.core.RedisConnectionException; +import io.lettuce.core.pubsub.StatefulRedisPubSubConnection; import io.lettuce.core.pubsub.api.reactive.RedisPubSubReactiveCommands; import reactor.core.Disposable; import reactor.core.publisher.DirectProcessor; @@ -40,6 +41,7 @@ import org.mockito.junit.jupiter.MockitoExtension; import org.springframework.data.redis.RedisSystemException; import org.springframework.data.redis.connection.ReactiveSubscription.Message; import org.springframework.data.redis.connection.ReactiveSubscription.PatternMessage; +import org.springframework.data.redis.connection.SubscriptionListener; /** * Unit tests for {@link LettuceReactiveSubscription}. @@ -52,11 +54,14 @@ class LettuceReactiveSubscriptionUnitTests { private LettuceReactiveSubscription subscription; + @Mock StatefulRedisPubSubConnection connectionMock; @Mock RedisPubSubReactiveCommands commandsMock; @BeforeEach void before() { - subscription = new LettuceReactiveSubscription(commandsMock, e -> new RedisSystemException(e.getMessage(), e)); + when(connectionMock.reactive()).thenReturn(commandsMock); + subscription = new LettuceReactiveSubscription(mock(SubscriptionListener.class), connectionMock, + e -> new RedisSystemException(e.getMessage(), e)); } @Test // DATAREDIS-612 diff --git a/src/test/java/org/springframework/data/redis/core/ReactiveRedisTemplateUnitTests.java b/src/test/java/org/springframework/data/redis/core/ReactiveRedisTemplateUnitTests.java index c3ae3d4e9..70ed2d3d9 100644 --- a/src/test/java/org/springframework/data/redis/core/ReactiveRedisTemplateUnitTests.java +++ b/src/test/java/org/springframework/data/redis/core/ReactiveRedisTemplateUnitTests.java @@ -71,7 +71,7 @@ class ReactiveRedisTemplateUnitTests { when(connectionMock.pubSubCommands()).thenReturn(pubSubCommands); when(pubSubCommands.subscribe(any())).thenReturn(Mono.empty()); - when(pubSubCommands.createSubscription()).thenReturn(Mono.just(subscription)); + when(pubSubCommands.createSubscription(any())).thenReturn(Mono.just(subscription)); when(subscription.receive()).thenReturn(Flux.create(sink -> {})); ReactiveRedisTemplate template = new ReactiveRedisTemplate<>(connectionFactoryMock, diff --git a/src/test/java/org/springframework/data/redis/listener/ReactiveRedisMessageListenerContainerIntegrationTests.java b/src/test/java/org/springframework/data/redis/listener/ReactiveRedisMessageListenerContainerIntegrationTests.java index 56ea8bebe..a3030ee03 100644 --- a/src/test/java/org/springframework/data/redis/listener/ReactiveRedisMessageListenerContainerIntegrationTests.java +++ b/src/test/java/org/springframework/data/redis/listener/ReactiveRedisMessageListenerContainerIntegrationTests.java @@ -22,19 +22,25 @@ import reactor.test.StepVerifier; import java.time.Duration; import java.util.Collection; +import java.util.Collections; import java.util.concurrent.BlockingQueue; +import java.util.concurrent.CompletableFuture; import java.util.concurrent.LinkedBlockingDeque; import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicReference; import java.util.function.Supplier; import org.awaitility.Awaitility; import org.junit.jupiter.api.AfterEach; import org.junit.jupiter.api.BeforeEach; +import org.springframework.data.redis.connection.Message; +import org.springframework.data.redis.connection.MessageListener; import org.springframework.data.redis.connection.ReactiveSubscription; import org.springframework.data.redis.connection.ReactiveSubscription.ChannelMessage; import org.springframework.data.redis.connection.ReactiveSubscription.PatternMessage; import org.springframework.data.redis.connection.RedisConnection; +import org.springframework.data.redis.connection.SubscriptionListener; import org.springframework.data.redis.connection.lettuce.LettuceConnectionFactory; import org.springframework.data.redis.core.ReactiveRedisTemplate; import org.springframework.data.redis.serializer.RedisSerializationContext; @@ -101,6 +107,53 @@ public class ReactiveRedisMessageListenerContainerIntegrationTests { container.destroy(); } + @ParameterizedRedisTest // GH-1622 + void receiveChannelShouldNotifySubscriptionListener() throws Exception { + + ReactiveRedisMessageListenerContainer container = new ReactiveRedisMessageListenerContainer(connectionFactory); + + AtomicReference onSubscribe = new AtomicReference<>(); + AtomicReference onUnsubscribe = new AtomicReference<>(); + CompletableFuture subscribe = new CompletableFuture<>(); + CompletableFuture unsubscribe = new CompletableFuture<>(); + + CompositeListener listener = new CompositeListener() { + @Override + public void onMessage(Message message, @Nullable byte[] pattern) { + + } + + @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.receive(Collections.singletonList(ChannelTopic.of(CHANNEL1)), listener).as(StepVerifier::create) // + .then(awaitSubscription(container::getActiveSubscriptions)) + .then(() -> connection.publish(CHANNEL1.getBytes(), MESSAGE.getBytes())) // + .assertNext(c -> { + + assertThat(c.getChannel()).isEqualTo(CHANNEL1); + assertThat(c.getMessage()).isEqualTo(MESSAGE); + }) // + .thenCancel().verify(); + + unsubscribe.get(10, TimeUnit.SECONDS); + + assertThat(onSubscribe).hasValue(CHANNEL1); + assertThat(onUnsubscribe).hasValue(CHANNEL1); + + container.destroy(); + } + @ParameterizedRedisTest // DATAREDIS-612 void shouldReceivePatternMessages() { @@ -120,6 +173,56 @@ public class ReactiveRedisMessageListenerContainerIntegrationTests { container.destroy(); } + @ParameterizedRedisTest // GH-1622 + void receivePatternShouldNotifySubscriptionListener() throws Exception { + + ReactiveRedisMessageListenerContainer container = new ReactiveRedisMessageListenerContainer(connectionFactory); + + 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.receive(Collections.singletonList(PatternTopic.of(PATTERN1)), listener) // + .cast(PatternMessage.class) // + .as(StepVerifier::create) // + .then(awaitSubscription(container::getActiveSubscriptions)) + .then(() -> connection.publish(CHANNEL1.getBytes(), MESSAGE.getBytes())) // + .assertNext(c -> { + + assertThat(c.getPattern()).isEqualTo(PATTERN1); + assertThat(c.getChannel()).isEqualTo(CHANNEL1); + assertThat(c.getMessage()).isEqualTo(MESSAGE); + }) // + .thenCancel().verify(); + + punsubscribe.get(10, TimeUnit.SECONDS); + + assertThat(onPsubscribe).hasValue(PATTERN1); + assertThat(onPunsubscribe).hasValue(PATTERN1); + + container.destroy(); + } + @ParameterizedRedisTest // DATAREDIS-612 void shouldPublishAndReceiveMessage() throws InterruptedException { @@ -191,4 +294,8 @@ public class ReactiveRedisMessageListenerContainerIntegrationTests { Awaitility.await().until(() -> !activeSubscriptions.get().isEmpty()); }; } + + interface CompositeListener extends MessageListener, SubscriptionListener { + + } } diff --git a/src/test/java/org/springframework/data/redis/listener/ReactiveRedisMessageListenerContainerUnitTests.java b/src/test/java/org/springframework/data/redis/listener/ReactiveRedisMessageListenerContainerUnitTests.java index 96f0bbebb..e98053f71 100644 --- a/src/test/java/org/springframework/data/redis/listener/ReactiveRedisMessageListenerContainerUnitTests.java +++ b/src/test/java/org/springframework/data/redis/listener/ReactiveRedisMessageListenerContainerUnitTests.java @@ -67,7 +67,7 @@ class ReactiveRedisMessageListenerContainerUnitTests { when(connectionFactoryMock.getReactiveConnection()).thenReturn(connectionMock); when(connectionMock.pubSubCommands()).thenReturn(commandsMock); when(connectionMock.closeLater()).thenReturn(Mono.empty()); - when(commandsMock.createSubscription()).thenReturn(Mono.just(subscriptionMock)); + when(commandsMock.createSubscription(any())).thenReturn(Mono.just(subscriptionMock)); when(subscriptionMock.subscribe(any())).thenReturn(Mono.empty()); when(subscriptionMock.pSubscribe(any())).thenReturn(Mono.empty()); when(subscriptionMock.unsubscribe()).thenReturn(Mono.empty()); @@ -182,7 +182,7 @@ class ReactiveRedisMessageListenerContainerUnitTests { assertThat(container.getActiveSubscriptions()).isEmpty(); } - @Test // DATAREDIS-612 + @Test // DATAREDIS-612, GH-1622 void shouldRegisterSubscriptionMultipleSubscribers() { reset(subscriptionMock); @@ -203,11 +203,11 @@ class ReactiveRedisMessageListenerContainerUnitTests { second.dispose(); - verify(subscriptionMock).unsubscribe(); + verify(subscriptionMock).cancel(); assertThat(container.getActiveSubscriptions()).isEmpty(); } - @Test // DATAREDIS-612 + @Test // DATAREDIS-612, GH-1622 void shouldUnsubscribeOnCancel() { when(subscriptionMock.receive()).thenReturn(DirectProcessor.create()); @@ -221,7 +221,7 @@ class ReactiveRedisMessageListenerContainerUnitTests { }).thenCancel().verify(); - verify(subscriptionMock).unsubscribe(); + verify(subscriptionMock).cancel(); } @Test // DATAREDIS-612