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

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

@@ -240,7 +240,7 @@ public class ReactiveRedisTemplate<K, V> implements ReactiveRedisOperations<K, V
 		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 @@ 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<Void> destroyLater() {
+		return Mono.defer(this::doDestroy);
+	}
+
+	private Mono<Void> 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() {

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.<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();
+	}
+}