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 814062fea..e91d5954e 100644
--- a/src/main/java/org/springframework/data/redis/core/ReactiveRedisTemplate.java
+++ b/src/main/java/org/springframework/data/redis/core/ReactiveRedisTemplate.java
@@ -235,7 +235,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());
 	}
 
 	// -------------------------------------------------------------------------
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<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() {
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 9b958ab46..94a8c98bc 100644
--- a/src/test/java/org/springframework/data/redis/core/ReactiveRedisTemplateUnitTests.java
+++ b/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 class ReactiveRedisTemplateUnitTests {
 		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();
+	}
 }