From 01c82714a057bc657cf96ef97f6179f818c85d4b Mon Sep 17 00:00:00 2001 From: Mark Paluch Date: Mon, 28 Oct 2019 14:25:40 +0100 Subject: [PATCH] DATAREDIS-1053 - Subscribe to cleanup publishers when canceling a listenTo() stream. Cleanup publishers are now subscribed to instead of setting the subscribeOn Scheduler to ensure connections get closed after canceling the stream. Original Pull Request: #488 --- .../redis/core/ReactiveRedisTemplate.java | 2 +- ...ReactiveRedisMessageListenerContainer.java | 15 ++-- .../core/ReactiveRedisTemplateUnitTests.java | 71 +++++++++++++++++++ 3 files changed, 80 insertions(+), 8 deletions(-) create mode 100644 src/test/java/org/springframework/data/redis/core/ReactiveRedisTemplateUnitTests.java diff --git a/src/main/java/org/springframework/data/redis/core/ReactiveRedisTemplate.java b/src/main/java/org/springframework/data/redis/core/ReactiveRedisTemplate.java index 6ef6c43ca..043e30ec0 100644 --- a/src/main/java/org/springframework/data/redis/core/ReactiveRedisTemplate.java +++ b/src/main/java/org/springframework/data/redis/core/ReactiveRedisTemplate.java @@ -240,7 +240,7 @@ public class ReactiveRedisTemplate implements ReactiveRedisOperations container.destroyLater().subscribeOn(Schedulers.elastic())); + .doFinally((signalType) -> container.destroyLater().subscribe()); } // ------------------------------------------------------------------------- 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 b26b5e758..98a4cd5fe 100644 --- a/src/main/java/org/springframework/data/redis/listener/ReactiveRedisMessageListenerContainer.java +++ b/src/main/java/org/springframework/data/redis/listener/ReactiveRedisMessageListenerContainer.java @@ -18,7 +18,6 @@ package org.springframework.data.redis.listener; import reactor.core.publisher.Flux; import reactor.core.publisher.Mono; import reactor.core.publisher.MonoProcessor; -import reactor.core.scheduler.Schedulers; import java.nio.ByteBuffer; import java.util.Arrays; @@ -97,6 +96,12 @@ public class ReactiveRedisMessageListenerContainer implements DisposableBean { * @return the {@link Mono} signalling container termination. */ public Mono destroyLater() { + return Mono.defer(this::doDestroy); + } + + private Mono doDestroy() { + + ReactiveRedisConnection connection = this.connection; if (connection != null) { @@ -116,12 +121,8 @@ public class ReactiveRedisMessageListenerContainer implements DisposableBean { } } - if (terminationSignals != null) { - return terminationSignals.collectList() - .doFinally(signalType -> connection.closeLater().subscribeOn(Schedulers.immediate())) - .flatMap(all -> Mono.empty()); - } this.connection = null; + return terminationSignals != null ? terminationSignals.then(connection.closeLater()) : connection.closeLater(); } return Mono.empty(); @@ -344,7 +345,7 @@ public class ReactiveRedisMessageListenerContainer implements DisposableBean { /** * Unregister a subscriber and decrement subscriber count. - * + * * @return {@literal true} if this was the last unregistered subscriber. */ boolean unregister() { diff --git a/src/test/java/org/springframework/data/redis/core/ReactiveRedisTemplateUnitTests.java b/src/test/java/org/springframework/data/redis/core/ReactiveRedisTemplateUnitTests.java new file mode 100644 index 000000000..d782e39f0 --- /dev/null +++ b/src/test/java/org/springframework/data/redis/core/ReactiveRedisTemplateUnitTests.java @@ -0,0 +1,71 @@ +/* + * Copyright 2019 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.core; + +import static org.assertj.core.api.Assertions.*; +import static org.mockito.Mockito.*; + +import reactor.core.publisher.Flux; +import reactor.core.publisher.Mono; +import reactor.test.StepVerifier; + +import java.util.concurrent.atomic.AtomicBoolean; + +import org.junit.Test; + +import org.springframework.data.redis.connection.ReactivePubSubCommands; +import org.springframework.data.redis.connection.ReactiveRedisConnection; +import org.springframework.data.redis.connection.ReactiveRedisConnectionFactory; +import org.springframework.data.redis.connection.ReactiveSubscription; +import org.springframework.data.redis.serializer.RedisSerializationContext; + +/** + * Unit tests for {@link ReactiveRedisTemplate}. + * + * @author Mark Paluch + */ +public class ReactiveRedisTemplateUnitTests { + + ReactiveRedisConnectionFactory connectionFactoryMock = mock(ReactiveRedisConnectionFactory.class); + ReactiveRedisConnection connectionMock = mock(ReactiveRedisConnection.class); + + @Test // DATAREDIS-999 + public void listenToShouldSubscribeToChannel() { + + AtomicBoolean closed = new AtomicBoolean(); + when(connectionFactoryMock.getReactiveConnection()).thenReturn(connectionMock); + when(connectionMock.closeLater()).thenReturn(Mono. empty().doOnSubscribe(ignore -> closed.set(true))); + + ReactivePubSubCommands pubSubCommands = mock(ReactivePubSubCommands.class); + ReactiveSubscription subscription = mock(ReactiveSubscription.class); + + when(connectionMock.pubSubCommands()).thenReturn(pubSubCommands); + when(pubSubCommands.subscribe(any())).thenReturn(Mono.empty()); + when(pubSubCommands.createSubscription()).thenReturn(Mono.just(subscription)); + when(subscription.receive()).thenReturn(Flux.create(sink -> {})); + + ReactiveRedisTemplate template = new ReactiveRedisTemplate<>(connectionFactoryMock, + RedisSerializationContext.string()); + + template.listenToChannel("channel") // + .as(StepVerifier::create) // + .thenAwait() // + .thenCancel() // + .verify(); + + assertThat(closed).isTrue(); + } +}