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: spring-projects#488

Author: mp911de

src/main/java/org/springframework/data/redis/core/ReactiveRedisTemplate.java

@@ -235,7 +235,7 @@ public Flux<? extends Message<String, V>> listenTo(Topic... topics) {
 return container
 .receive(Arrays.asList(topics), getSerializationContext().getStringSerializationPair(),
 getSerializationContext().getValueSerializationPair()) //
-.doFinally((signalType) -> container.destroyLater().subscribeOn(Schedulers.elastic()));
+.doFinally((signalType) -> container.destroyLater().subscribe());
 }
 
 // -------------------------------------------------------------------------

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

@@ -18,7 +18,6 @@
 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 void destroy() {
  * @return the {@link Mono} signalling container termination.
  */
 public Mono<Void> destroyLater() {
+return Mono.defer(this::doDestroy);
+}
+
+private Mono<Void> doDestroy() {
+
+ReactiveRedisConnection connection = this.connection;
 
 if (connection != null) {
 
@@ -116,12 +121,8 @@ public Mono<Void> destroyLater() {
 }
 }
 
-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 @@ boolean hasRegistration() {
 
 /**
  * Unregister a subscriber and decrement subscriber count.
- * 
+ *
  * @return {@literal true} if this was the last unregistered subscriber.
  */
 boolean unregister() {

src/test/java/org/springframework/data/redis/core/ReactiveRedisTemplateUnitTests.java

@@ -15,15 +15,21 @@
  */
 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;
 
 /**
@@ -52,4 +58,31 @@ public void closeShouldUseAsyncRelease() {
 verify(connectionMock).closeLater();
 verifyNoMoreInteractions(connectionMock);
 }
+
+@Test // DATAREDIS-999
+public void listenToShouldSubscribeToChannel() {
+
+AtomicBoolean closed = new AtomicBoolean();
+when(connectionFactoryMock.getReactiveConnection()).thenReturn(connectionMock);
+when(connectionMock.closeLater()).thenReturn(Mono.<Void> 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<String, String> template = new ReactiveRedisTemplate<>(connectionFactoryMock,
+RedisSerializationContext.string());
+
+template.listenToChannel("channel") //
+.as(StepVerifier::create) //
+.thenAwait() //
+.thenCancel() //
+.verify();
+
+assertThat(closed).isTrue();
+}
 }