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 e95841f76..e81bcea99 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 @@ -20,14 +20,21 @@ import reactor.core.publisher.Flux; import reactor.core.publisher.Mono; import java.nio.ByteBuffer; +import java.util.ArrayList; +import java.util.List; +import java.util.Map; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.atomic.AtomicLongFieldUpdater; +import java.util.function.BiFunction; 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.data.redis.connection.util.ByteArrayWrapper; +import org.springframework.data.redis.util.ByteUtils; import org.springframework.util.Assert; /** @@ -39,16 +46,27 @@ class LettuceReactivePubSubCommands implements ReactivePubSubCommands { private final LettuceReactiveRedisConnection connection; + private final Map channels = new ConcurrentHashMap<>(); + + private final Map patterns = new ConcurrentHashMap<>(); + LettuceReactivePubSubCommands(LettuceReactiveRedisConnection connection) { this.connection = connection; } + public Map getChannels() { + return channels; + } + + public Map getPatterns() { + return patterns; + } + @Override public Mono createSubscription(SubscriptionListener listener) { - return connection.getPubSubConnection() - .map(pubSubConnection -> new LettuceReactiveSubscription(listener, pubSubConnection, - connection.translateException())); + return connection.getPubSubConnection().map(pubSubConnection -> new LettuceReactiveSubscription(listener, + pubSubConnection, this, connection.translateException())); } @Override @@ -65,20 +83,157 @@ class LettuceReactivePubSubCommands implements ReactivePubSubCommands { Assert.notNull(channels, "Channels must not be null"); + Target.trackSubscriptions(channels, this.channels); // track usage but do not limit what to subscribe to + return doWithPubSub(commands -> commands.subscribe(channels)); } + public Mono unsubscribe(ByteBuffer... channels) { + + Assert.notNull(patterns, "Patterns must not be null"); + + ByteBuffer[] actualUnsubscribe = Target.trackUnsubscriptions(channels, this.channels); + + if (actualUnsubscribe.length == 0 && channels.length != 0) { + return Mono.empty(); + } + + return doWithPubSub(commands -> commands.unsubscribe(actualUnsubscribe)); + } + @Override public Mono pSubscribe(ByteBuffer... patterns) { Assert.notNull(patterns, "Patterns must not be null"); + Target.trackSubscriptions(patterns, this.patterns); // track usage but do not limit what to subscribe to + return doWithPubSub(commands -> commands.psubscribe(patterns)); } + public Mono pUnsubscribe(ByteBuffer... patterns) { + + Assert.notNull(patterns, "Patterns must not be null"); + + ByteBuffer[] actualUnsubscribe = Target.trackUnsubscriptions(patterns, this.patterns); + + if (actualUnsubscribe.length == 0 && patterns.length != 0) { + return Mono.empty(); + } + + return doWithPubSub(commands -> commands.punsubscribe(actualUnsubscribe)); + } + private Mono doWithPubSub(Function, Mono> function) { return connection.getPubSubConnection().flatMap(pubSubConnection -> function.apply(pubSubConnection.reactive())) .onErrorMap(connection.translateException()); } + + static class Target { + + private static final AtomicLongFieldUpdater SUBSCRIBERS = AtomicLongFieldUpdater.newUpdater(Target.class, + "subscribers"); + + private final byte[] raw; + + private volatile long subscribers; + + Target(byte[] raw) { + this.raw = raw; + } + + /** + * Record the subscriptions to {@code targets} and store these in {@code targetMap}. + * + * @param targets + * @param targetMap + */ + public static void trackSubscriptions(ByteBuffer[] targets, Map targetMap) { + doWithTargets(targets, targetMap, Target::allocate); + } + + /** + * Record the un-subscriptions to {@code targets} and store these in {@code targetMap}. Returns the targets to + * actually unsubscribe from if there are no subscribers to a particular target. + * + * @param targets + * @param targetMap + */ + public static ByteBuffer[] trackUnsubscriptions(ByteBuffer[] targets, Map targetMap) { + return doWithTargets(targets, targetMap, Target::deallocate); + } + + static ByteBuffer[] doWithTargets(ByteBuffer[] targets, Map targetMap, + BiFunction, Boolean> f) { + + List toSubscribe = new ArrayList<>(targets.length); + + synchronized (targetMap) { + for (ByteBuffer target : targets) { + if (f.apply(target, targetMap)) { + toSubscribe.add(target); + } + } + } + + return toSubscribe.toArray(new ByteBuffer[0]); + } + + boolean increment() { + return SUBSCRIBERS.incrementAndGet(this) == 1; + } + + boolean decrement() { + + long l = SUBSCRIBERS.get(this); + + if (l > 0) { + if (SUBSCRIBERS.compareAndSet(this, l, l - 1)) { + return l == 1; // return true if this was the last subscriber + } + } + + return false; + } + + static boolean allocate(ByteBuffer buffer, Map targets) { + + byte[] raw = ByteUtils.getBytes(buffer); + + ByteArrayWrapper wrapper = new ByteArrayWrapper(raw); + Target targetToUse = targets.get(wrapper); + + if (targetToUse == null) { + targetToUse = new Target(raw); + targets.put(wrapper, targetToUse); + } + + return targetToUse.increment(); + } + + static boolean deallocate(ByteBuffer buffer, Map targets) { + + byte[] raw = ByteUtils.getBytes(buffer); + + ByteArrayWrapper wrapper = new ByteArrayWrapper(raw); + Target targetToUse = targets.get(wrapper); + + if (targetToUse == null) { + return false; + } + + if (targetToUse.decrement()) { + targets.remove(wrapper); + return true; + } + + return false; + } + + @Override + public String toString() { + return String.format("%s: Subscribers: %s", new String(raw), SUBSCRIBERS.get(this)); + } + } } diff --git a/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceReactiveRedisConnection.java b/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceReactiveRedisConnection.java index 26ef883da..dfd83f049 100644 --- a/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceReactiveRedisConnection.java +++ b/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceReactiveRedisConnection.java @@ -51,6 +51,8 @@ class LettuceReactiveRedisConnection implements ReactiveRedisConnection { private final AsyncConnect> dedicatedConnection; private final AsyncConnect> pubSubConnection; + private final LettuceReactivePubSubCommands pubSub = new LettuceReactivePubSubCommands(this); + private @Nullable Mono> sharedConnection; /** @@ -137,7 +139,7 @@ class LettuceReactiveRedisConnection implements ReactiveRedisConnection { @Override public ReactivePubSubCommands pubSubCommands() { - return new LettuceReactivePubSubCommands(this); + return pubSub; } @Override 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 d75d5892e..c259e8992 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 @@ -23,18 +23,17 @@ import reactor.core.publisher.Flux; import reactor.core.publisher.Mono; import java.nio.ByteBuffer; -import java.util.Arrays; -import java.util.Collections; -import java.util.List; import java.util.Set; import java.util.concurrent.ConcurrentSkipListSet; import java.util.concurrent.atomic.AtomicLong; import java.util.concurrent.atomic.AtomicReference; import java.util.function.Function; import java.util.function.Supplier; +import java.util.stream.Collectors; import org.springframework.data.redis.connection.ReactiveSubscription; import org.springframework.data.redis.connection.SubscriptionListener; +import org.springframework.data.redis.connection.util.ByteArrayWrapper; import org.springframework.lang.Nullable; import org.springframework.util.Assert; import org.springframework.util.ObjectUtils; @@ -50,19 +49,22 @@ class LettuceReactiveSubscription implements ReactiveSubscription { private final LettuceByteBufferPubSubListenerWrapper listener; private final StatefulRedisPubSubConnection connection; - private final RedisPubSubReactiveCommands commands; + + private final RedisPubSubReactiveCommands reactive; + private final LettuceReactivePubSubCommands commands; private final State patternState; private final State channelState; LettuceReactiveSubscription(SubscriptionListener subscriptionListener, - StatefulRedisPubSubConnection connection, + StatefulRedisPubSubConnection connection, LettuceReactivePubSubCommands commands, Function exceptionTranslator) { this.listener = new LettuceByteBufferPubSubListenerWrapper( new LettuceMessageListener((messages, pattern) -> {}, subscriptionListener)); this.connection = connection; - this.commands = connection.reactive(); + this.reactive = connection.reactive(); + this.commands = commands; connection.addListener(listener); this.patternState = new State(exceptionTranslator); @@ -84,7 +86,7 @@ class LettuceReactiveSubscription implements ReactiveSubscription { Assert.notNull(patterns, "Patterns must not be null"); Assert.noNullElements(patterns, "Patterns must not contain null elements"); - return patternState.subscribe(patterns, commands::psubscribe); + return patternState.subscribe(patterns, commands::pSubscribe); } @Override @@ -112,7 +114,7 @@ class LettuceReactiveSubscription implements ReactiveSubscription { Assert.notNull(patterns, "Patterns must not be null"); Assert.noNullElements(patterns, "Patterns must not contain null elements"); - return ObjectUtils.isEmpty(patterns) ? Mono.empty() : patternState.unsubscribe(patterns, commands::punsubscribe); + return ObjectUtils.isEmpty(patterns) ? Mono.empty() : patternState.unsubscribe(patterns, commands::pUnsubscribe); } @Override @@ -128,12 +130,12 @@ class LettuceReactiveSubscription implements ReactiveSubscription { @Override public Flux> receive() { - Flux> channelMessages = channelState.receive(() -> commands.observeChannels() // - .filter(message -> channelState.getTargets().contains(message.getChannel())) // + Flux> channelMessages = channelState.receive(() -> reactive.observeChannels() // + .filter(message -> channelState.contains(message.getChannel())) // .map(message -> new ChannelMessage<>(message.getChannel(), message.getMessage()))); - Flux> patternMessages = patternState.receive(() -> commands.observePatterns() // - .filter(message -> patternState.getTargets().contains(message.getPattern())) // + Flux> patternMessages = patternState.receive(() -> reactive.observePatterns() // + .filter(message -> patternState.contains(message.getPattern())) // .map(message -> new PatternMessage<>(message.getPattern(), message.getChannel(), message.getMessage()))); return channelMessages.mergeWith(patternMessages); @@ -149,7 +151,7 @@ class LettuceReactiveSubscription implements ReactiveSubscription { // 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(() -> { + return reactive.ping().then(Mono.fromRunnable(() -> { connection.removeListener(listener); })); })); @@ -162,7 +164,7 @@ class LettuceReactiveSubscription implements ReactiveSubscription { */ static class State { - private final Set targets = new ConcurrentSkipListSet<>(); + private final Set targets = new ConcurrentSkipListSet<>(); private final AtomicLong subscribers = new AtomicLong(); private final AtomicReference> flux = new AtomicReference<>(); private final Function exceptionTranslator; @@ -182,8 +184,12 @@ class LettuceReactiveSubscription implements ReactiveSubscription { */ Mono subscribe(ByteBuffer[] targets, Function> subscribeFunction) { - return subscribeFunction.apply(targets).doOnSuccess((discard) -> this.targets.addAll(Arrays.asList(targets))) - .onErrorMap(exceptionTranslator); + return subscribeFunction.apply(targets).doOnSuccess((discard) -> { + + for (ByteBuffer target : targets) { + this.targets.add(getWrapper(target)); + } + }).onErrorMap(exceptionTranslator); } /** @@ -198,16 +204,18 @@ class LettuceReactiveSubscription implements ReactiveSubscription { return Mono.defer(() -> { - List targetCollection = Arrays.asList(targets); - return unsubscribeFunction.apply(targets).doOnSuccess((discard) -> { - this.targets.removeAll(targetCollection); + + for (ByteBuffer byteBuffer : targets) { + this.targets.remove(getWrapper(byteBuffer)); + } }).onErrorMap(exceptionTranslator); }); } Set getTargets() { - return Collections.unmodifiableSet(targets); + return targets.stream().map(ByteArrayWrapper::getArray).map(ByteBuffer::wrap) + .collect(Collectors.toUnmodifiableSet()); } /** @@ -263,5 +271,13 @@ class LettuceReactiveSubscription implements ReactiveSubscription { disposable.dispose(); } } + + public boolean contains(ByteBuffer target) { + return this.targets.contains(getWrapper(target)); + } + + private static ByteArrayWrapper getWrapper(ByteBuffer byteBuffer) { + return new ByteArrayWrapper(byteBuffer); + } } } diff --git a/src/main/java/org/springframework/data/redis/connection/util/ByteArrayWrapper.java b/src/main/java/org/springframework/data/redis/connection/util/ByteArrayWrapper.java index 9dbde9f54..6b92a3977 100644 --- a/src/main/java/org/springframework/data/redis/connection/util/ByteArrayWrapper.java +++ b/src/main/java/org/springframework/data/redis/connection/util/ByteArrayWrapper.java @@ -15,8 +15,10 @@ */ package org.springframework.data.redis.connection.util; +import java.nio.ByteBuffer; import java.util.Arrays; +import org.springframework.data.redis.util.ByteUtils; import org.springframework.lang.Nullable; /** @@ -24,11 +26,15 @@ import org.springframework.lang.Nullable; * * @author Costin Leau */ -public class ByteArrayWrapper { +public class ByteArrayWrapper implements Comparable { private final byte[] array; private final int hashCode; + public ByteArrayWrapper(ByteBuffer buffer) { + this(ByteUtils.getBytes(buffer.asReadOnlyBuffer())); + } + public ByteArrayWrapper(byte[] array) { this.array = array; this.hashCode = Arrays.hashCode(array); @@ -54,4 +60,14 @@ public class ByteArrayWrapper { public byte[] getArray() { return array; } + + @Override + public String toString() { + return new String(array); + } + + @Override + public int compareTo(ByteArrayWrapper o) { + return Arrays.compare(this.array, o.array); + } } diff --git a/src/test/java/org/springframework/data/redis/connection/lettuce/LettuceReactivePubSubCommandsUnitTests.java b/src/test/java/org/springframework/data/redis/connection/lettuce/LettuceReactivePubSubCommandsUnitTests.java new file mode 100644 index 000000000..e947b1cac --- /dev/null +++ b/src/test/java/org/springframework/data/redis/connection/lettuce/LettuceReactivePubSubCommandsUnitTests.java @@ -0,0 +1,201 @@ +/* + * Copyright 2022 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * https://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.springframework.data.redis.connection.lettuce; + +import static org.assertj.core.api.Assertions.*; +import static org.mockito.Mockito.*; + +import io.lettuce.core.pubsub.StatefulRedisPubSubConnection; +import io.lettuce.core.pubsub.api.reactive.RedisPubSubReactiveCommands; +import reactor.core.publisher.Mono; +import reactor.test.StepVerifier; + +import java.nio.ByteBuffer; + +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.mockito.Mock; +import org.mockito.junit.jupiter.MockitoSettings; +import org.mockito.quality.Strictness; + +/** + * Unit tests for {@link LettuceReactivePubSubCommands}. + * + * @author Mark Paluch + */ +@MockitoSettings(strictness = Strictness.LENIENT) +class LettuceReactivePubSubCommandsUnitTests { + + LettuceReactivePubSubCommands sut; + + @Mock LettuceReactiveRedisConnection connection; + + @Mock StatefulRedisPubSubConnection lettuceConnection; + @Mock RedisPubSubReactiveCommands reactiveCommands; + + @SuppressWarnings("unchecked") + @BeforeEach + void setUp() { + + when(connection.getPubSubConnection()).thenReturn(Mono.just(lettuceConnection)); + when(lettuceConnection.reactive()).thenReturn(reactiveCommands); + sut = new LettuceReactivePubSubCommands(connection); + } + + @Test // GH-2386 + void shouldSubscribeChannelMultipleTimes() { + + when(reactiveCommands.subscribe(any())).thenReturn(Mono.empty()); + + sut.subscribe(wrap("channel")) // + .as(StepVerifier::create) // + .verifyComplete(); + + sut.subscribe(wrap("channel")) // + .as(StepVerifier::create) // + .verifyComplete(); + + verify(reactiveCommands, times(2)).subscribe(wrap("channel")); + assertThat(sut.getChannels()).hasSize(1); + assertThat(sut.getPatterns()).isEmpty(); + } + + @Test // GH-2386 + void shouldNotUnsubscribeChannelIfUsedMultipleTimes() { + + when(reactiveCommands.subscribe(any())).thenReturn(Mono.empty()); + when(reactiveCommands.unsubscribe(any())).thenReturn(Mono.empty()); + + sut.subscribe(wrap("channel")) // + .as(StepVerifier::create) // + .verifyComplete(); + + sut.subscribe(wrap("channel")) // + .as(StepVerifier::create) // + .verifyComplete(); + + sut.unsubscribe(wrap("channel")) // + .as(StepVerifier::create) // + .verifyComplete(); + + verify(reactiveCommands, times(2)).subscribe(wrap("channel")); + verifyNoMoreInteractions(reactiveCommands); + assertThat(sut.getChannels()).hasSize(1); + assertThat(sut.getPatterns()).isEmpty(); + } + + @Test // GH-2386 + void shouldUnsubscribeChannelIfNotUsedAnymore() { + + when(reactiveCommands.subscribe(any())).thenReturn(Mono.empty()); + when(reactiveCommands.unsubscribe(any())).thenReturn(Mono.empty()); + + sut.subscribe(wrap("channel")) // + .as(StepVerifier::create) // + .verifyComplete(); + + sut.subscribe(wrap("channel")) // + .as(StepVerifier::create) // + .verifyComplete(); + + sut.unsubscribe(wrap("channel")) // + .as(StepVerifier::create) // + .verifyComplete(); + + sut.unsubscribe(wrap("channel")) // + .as(StepVerifier::create) // + .verifyComplete(); + + verify(reactiveCommands, times(2)).subscribe(wrap("channel")); + verify(reactiveCommands, times(1)).unsubscribe(wrap("channel")); + assertThat(sut.getChannels()).isEmpty(); + assertThat(sut.getPatterns()).isEmpty(); + } + + @Test // GH-2386 + void shouldSubscribePatternMultipleTimes() { + + when(reactiveCommands.psubscribe(any())).thenReturn(Mono.empty()); + + sut.pSubscribe(wrap("pattern")) // + .as(StepVerifier::create) // + .verifyComplete(); + + sut.pSubscribe(wrap("pattern")) // + .as(StepVerifier::create) // + .verifyComplete(); + + verify(reactiveCommands, times(2)).psubscribe(wrap("pattern")); + assertThat(sut.getChannels()).isEmpty(); + assertThat(sut.getPatterns()).hasSize(1); + } + + @Test // GH-2386 + void shouldNotUnsubscribePatternIfUsedMultipleTimes() { + + when(reactiveCommands.psubscribe(any())).thenReturn(Mono.empty()); + when(reactiveCommands.punsubscribe(any())).thenReturn(Mono.empty()); + + sut.pSubscribe(wrap("pattern")) // + .as(StepVerifier::create) // + .verifyComplete(); + + sut.pSubscribe(wrap("pattern")) // + .as(StepVerifier::create) // + .verifyComplete(); + + sut.pUnsubscribe(wrap("pattern")) // + .as(StepVerifier::create) // + .verifyComplete(); + + verify(reactiveCommands, times(2)).psubscribe(wrap("pattern")); + verifyNoMoreInteractions(reactiveCommands); + assertThat(sut.getChannels()).isEmpty(); + assertThat(sut.getPatterns()).hasSize(1); + } + + @Test // GH-2386 + void shouldUnsubscribePatternIfNotUsedAnymore() { + + when(reactiveCommands.psubscribe(any())).thenReturn(Mono.empty()); + when(reactiveCommands.punsubscribe(any())).thenReturn(Mono.empty()); + + sut.pSubscribe(wrap("pattern")) // + .as(StepVerifier::create) // + .verifyComplete(); + + sut.pSubscribe(wrap("pattern")) // + .as(StepVerifier::create) // + .verifyComplete(); + + sut.pUnsubscribe(wrap("pattern")) // + .as(StepVerifier::create) // + .verifyComplete(); + + sut.pUnsubscribe(wrap("pattern")) // + .as(StepVerifier::create) // + .verifyComplete(); + + verify(reactiveCommands, times(2)).psubscribe(wrap("pattern")); + verify(reactiveCommands, times(1)).punsubscribe(wrap("pattern")); + assertThat(sut.getChannels()).isEmpty(); + assertThat(sut.getPatterns()).isEmpty(); + } + + private static ByteBuffer wrap(String content) { + return ByteBuffer.wrap(content.getBytes()); + } +} 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 21a35bd8c..74a4ec36d 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 @@ -37,7 +37,6 @@ import org.junit.jupiter.api.Test; import org.junit.jupiter.api.extension.ExtendWith; import org.mockito.Mock; 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; @@ -57,17 +56,19 @@ class LettuceReactiveSubscriptionUnitTests { @Mock StatefulRedisPubSubConnection connectionMock; @Mock RedisPubSubReactiveCommands commandsMock; + @Mock LettuceReactivePubSubCommands pubSubMock; + @BeforeEach void before() { when(connectionMock.reactive()).thenReturn(commandsMock); - subscription = new LettuceReactiveSubscription(mock(SubscriptionListener.class), connectionMock, + subscription = new LettuceReactiveSubscription(mock(SubscriptionListener.class), connectionMock, pubSubMock, e -> new RedisSystemException(e.getMessage(), e)); } @Test // DATAREDIS-612 void shouldSubscribeChannels() { - when(commandsMock.subscribe(any())).thenReturn(Mono.empty()); + when(pubSubMock.subscribe(any())).thenReturn(Mono.empty()); Mono subscribe = subscription.subscribe(getByteBuffer("foo"), getByteBuffer("bar")); @@ -82,7 +83,7 @@ class LettuceReactiveSubscriptionUnitTests { @Test // DATAREDIS-612 void shouldSubscribeChannelsShouldFail() { - when(commandsMock.subscribe(any())).thenReturn(Mono.error(new RedisConnectionException("Foo"))); + when(pubSubMock.subscribe(any())).thenReturn(Mono.error(new RedisConnectionException("Foo"))); Mono subscribe = subscription.subscribe(getByteBuffer("foo"), getByteBuffer("bar")); @@ -92,7 +93,7 @@ class LettuceReactiveSubscriptionUnitTests { @Test // DATAREDIS-612 void shouldSubscribePatterns() { - when(commandsMock.psubscribe(any())).thenReturn(Mono.empty()); + when(pubSubMock.pSubscribe(any())).thenReturn(Mono.empty()); Mono subscribe = subscription.pSubscribe(getByteBuffer("foo"), getByteBuffer("bar")); @@ -107,33 +108,33 @@ class LettuceReactiveSubscriptionUnitTests { @Test // DATAREDIS-612 void shouldUnsubscribeChannels() { - when(commandsMock.subscribe(any())).thenReturn(Mono.empty()); - when(commandsMock.unsubscribe(any())).thenReturn(Mono.empty()); + when(pubSubMock.subscribe(any())).thenReturn(Mono.empty()); + when(pubSubMock.unsubscribe(any())).thenReturn(Mono.empty()); subscription.subscribe(getByteBuffer("foo"), getByteBuffer("bar")).as(StepVerifier::create).verifyComplete(); subscription.unsubscribe().as(StepVerifier::create).verifyComplete(); assertThat(subscription.getChannels()).isEmpty(); - verify(commandsMock).unsubscribe(any()); + verify(pubSubMock).unsubscribe(any()); } @Test // DATAREDIS-612 void shouldUnsubscribePatterns() { - when(commandsMock.psubscribe(any())).thenReturn(Mono.empty()); - when(commandsMock.punsubscribe(any())).thenReturn(Mono.empty()); + when(pubSubMock.pSubscribe(any())).thenReturn(Mono.empty()); + when(pubSubMock.pUnsubscribe(any())).thenReturn(Mono.empty()); subscription.pSubscribe(getByteBuffer("foo"), getByteBuffer("bar")).as(StepVerifier::create).verifyComplete(); subscription.pUnsubscribe().as(StepVerifier::create).verifyComplete(); assertThat(subscription.getPatterns()).isEmpty(); - verify(commandsMock).punsubscribe(any()); + verify(pubSubMock).pUnsubscribe(any()); } @Test // DATAREDIS-612 void shouldEmitChannelMessage() { - when(commandsMock.subscribe(any())).thenReturn(Mono.empty()); + when(pubSubMock.subscribe(any())).thenReturn(Mono.empty()); subscription.subscribe(getByteBuffer("foo"), getByteBuffer("bar")).as(StepVerifier::create).verifyComplete(); Sinks.Many> sink = Sinks.many().unicast() @@ -153,7 +154,7 @@ class LettuceReactiveSubscriptionUnitTests { @Test // DATAREDIS-612 void shouldEmitPatternMessage() { - when(commandsMock.psubscribe(any())).thenReturn(Mono.empty()); + when(pubSubMock.pSubscribe(any())).thenReturn(Mono.empty()); subscription.pSubscribe(getByteBuffer("foo*"), getByteBuffer("bar*")).as(StepVerifier::create).verifyComplete(); Sinks.Many> sink = Sinks.many().unicast() @@ -175,7 +176,7 @@ class LettuceReactiveSubscriptionUnitTests { @Test // DATAREDIS-612 void shouldEmitError() { - when(commandsMock.subscribe(any())).thenReturn(Mono.empty()); + when(pubSubMock.subscribe(any())).thenReturn(Mono.empty()); subscription.subscribe(getByteBuffer("foo"), getByteBuffer("bar")).as(StepVerifier::create).verifyComplete(); Sinks.Many> sink = Sinks.many().unicast() @@ -192,8 +193,8 @@ class LettuceReactiveSubscriptionUnitTests { @Test // DATAREDIS-612 void shouldTerminateActiveSubscriptions() { - when(commandsMock.psubscribe(any())).thenReturn(Mono.empty()); - when(commandsMock.punsubscribe(any())).thenReturn(Mono.empty()); + when(pubSubMock.pSubscribe(any())).thenReturn(Mono.empty()); + when(pubSubMock.pUnsubscribe(any())).thenReturn(Mono.empty()); subscription.pSubscribe(getByteBuffer("foo*")).as(StepVerifier::create).verifyComplete(); when(commandsMock.observeChannels()).thenReturn(Flux.never()); @@ -212,7 +213,7 @@ class LettuceReactiveSubscriptionUnitTests { Sinks.Many> sink = Sinks.many().unicast() .onBackpressureBuffer(); - when(commandsMock.psubscribe(any())).thenReturn(Mono.empty()); + when(pubSubMock.pSubscribe(any())).thenReturn(Mono.empty()); subscription.pSubscribe(getByteBuffer("foo*")).as(StepVerifier::create).verifyComplete(); when(commandsMock.observeChannels()).thenReturn(Flux.never()); 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 efa559280..8d517f6f1 100644 --- a/src/test/java/org/springframework/data/redis/listener/ReactiveRedisMessageListenerContainerIntegrationTests.java +++ b/src/test/java/org/springframework/data/redis/listener/ReactiveRedisMessageListenerContainerIntegrationTests.java @@ -18,12 +18,17 @@ package org.springframework.data.redis.listener; import static org.assertj.core.api.Assertions.*; import reactor.core.Disposable; +import reactor.core.publisher.Flux; import reactor.test.StepVerifier; import java.nio.ByteBuffer; import java.time.Duration; +import java.util.ArrayList; +import java.util.Arrays; import java.util.Collection; import java.util.Collections; +import java.util.List; +import java.util.Queue; import java.util.concurrent.BlockingQueue; import java.util.concurrent.CompletableFuture; import java.util.concurrent.LinkedBlockingDeque; @@ -47,6 +52,8 @@ 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; +import org.springframework.data.redis.serializer.RedisSerializationContext.SerializationPair; +import org.springframework.data.redis.serializer.RedisSerializer; import org.springframework.data.redis.test.extension.parametrized.MethodSource; import org.springframework.data.redis.test.extension.parametrized.ParameterizedRedisTest; import org.springframework.lang.Nullable; @@ -302,6 +309,49 @@ public class ReactiveRedisMessageListenerContainerIntegrationTests { .verify(); } + @ParameterizedRedisTest // GH-2386 + void multipleListenShouldTrackSubscriptions() throws Exception { + + ReactiveRedisMessageListenerContainer container = new ReactiveRedisMessageListenerContainer(connectionFactory); + + Flux> c1 = container.receiveLater(ChannelTopic.of(CHANNEL1)) + .block(); + Flux> c1p1 = container + .receiveLater(Arrays.asList(ChannelTopic.of(CHANNEL1), PatternTopic.of(PATTERN1)), + SerializationPair.fromSerializer(RedisSerializer.string()), + SerializationPair.fromSerializer(RedisSerializer.string())) + .block(); + + BlockingQueue> c1Collector = new LinkedBlockingDeque<>(); + BlockingQueue> c2Collector = new LinkedBlockingDeque<>(); + + Disposable c1Subscription = c1.doOnNext(c1Collector::add).subscribe(); + Disposable c2Subscription = c1p1.doOnNext(c2Collector::add).subscribe(); + + doPublish(CHANNEL1.getBytes(), MESSAGE.getBytes()); + + assertThat(c1Collector.poll(5, TimeUnit.SECONDS)).isNotNull(); + assertThat(c2Collector.poll(5, TimeUnit.SECONDS)).isNotNull(); + c1Collector.clear(); + c2Collector.clear(); + + c2Subscription.dispose(); + + Thread.sleep(200); + + doPublish(CHANNEL1.getBytes(), MESSAGE.getBytes()); + + assertThat(c1Collector.poll(5, TimeUnit.SECONDS)).isNotNull(); + assertThat(c2Collector.poll(100, TimeUnit.MILLISECONDS)).isNull(); + + c1Subscription.dispose(); + + doPublish(CHANNEL1.getBytes(), MESSAGE.getBytes()); + + assertThat(c1Collector.poll(100, TimeUnit.MILLISECONDS)).isNull(); + assertThat(c2Collector.poll(100, TimeUnit.MILLISECONDS)).isNull(); + } + private void doPublish(byte[] channel, byte[] message) { reactiveConnection.pubSubCommands().publish(ByteBuffer.wrap(channel), ByteBuffer.wrap(message)).subscribe(); }