DATAREDIS-612 - Polishing.

Introduce dedicated methods utilizing the ReactiveMessageListenerContainer to create a stream of Messages via ReactiveRedisTemplate.

redisTemplate.listenToChannel("channel1", "channel2").doOnNext(msg -> {
    // message processing ...
}).subscribe();

Add close-/destroyLater methods avoiding blocking shutdown of connections and containers.

Original Pull Request: #295

src/main/java/org/springframework/data/redis/connection/ReactivePubSubCommands.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2017 the original author or authors.
+ * Copyright 2018 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.
@@ -27,9 +27,10 @@ import org.springframework.data.redis.connection.ReactiveSubscription.ChannelMes
  * Redis <a href="https://redis.io/commands/#pubsub">Pub/Sub</a> commands executed using reactive infrastructure.
  *
  * @author Mark Paluch
+ * @author Christoph Strobl
  * @since 2.1
  */
-public interface ReactiveRedisPubSubCommands {
+public interface ReactivePubSubCommands {
 
 	/**
 	 * Creates a subscription for this connection. Connections can have multiple {@link ReactiveSubscription}s.
@@ -63,7 +64,7 @@ public interface ReactiveRedisPubSubCommands {
 	 * Subscribes the connection to the given {@code channels}. Once subscribed, a connection enters listening mode and
 	 * can only subscribe to other channels or unsubscribe. No other commands are accepted until the connection is
 	 * unsubscribed.
-	 * <p/>
+	 * <p />
 	 * Note that cancellation of the {@link Flux} will unsubscribe from {@code channels}.
 	 *
 	 * @param channels channel names, must not be {@literal null}.
@@ -82,4 +83,5 @@ public interface ReactiveRedisPubSubCommands {
 	 * @see <a href="http://redis.io/commands/psubscribe">Redis Documentation: PSUBSCRIBE</a>
 	 */
 	Mono<Void> pSubscribe(ByteBuffer... patterns);
+
 }

src/main/java/org/springframework/data/redis/connection/ReactiveRedisConnection.java

@@ -50,8 +50,21 @@ import org.springframework.util.Assert;
  */
 public interface ReactiveRedisConnection extends Closeable {
 
+	/*
+	 * (non-Javadoc)
+	 * @see java.io.Closeable#close()
+	 */
 	@Override
-	void close();
+	default void close() {
+		closeLater().block();
+	}
+
+	/**
+	 * Asynchronously close the connection and release associated resources.
+	 *
+	 * @return the {@link Mono} signaling when done.
+	 */
+	Mono<Void> closeLater();
 
 	/**
 	 * Get {@link ReactiveKeyCommands}.
@@ -117,12 +130,12 @@ public interface ReactiveRedisConnection extends Closeable {
 	ReactiveHyperLogLogCommands hyperLogLogCommands();
 
 	/**
-	 * Get {@link ReactiveRedisPubSubCommands}.
+	 * Get {@link ReactivePubSubCommands}.
 	 *
 	 * @return never {@literal null}.
 	 * @since 2.1
 	 */
-	ReactiveRedisPubSubCommands pubSubCommands();
+	ReactivePubSubCommands pubSubCommands();
 
 	/**
 	 * Get {@link ReactiveScriptingCommands}.

src/main/java/org/springframework/data/redis/connection/ReactiveSubscription.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2017 the original author or authors.
+ * Copyright 2018 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.
@@ -20,19 +20,22 @@ import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
 
 import java.nio.ByteBuffer;
-import java.util.Collection;
+import java.util.Set;
+
+import org.springframework.util.Assert;
 
 /**
- * Subscription for Redis channels using reactive infrastructure. A {@link ReactiveSubscription} allows subscription to
+ * Subscription for Redis channels using reactive infrastructure. A {@link ReactiveSubscription} allows subscribing to
  * {@link #subscribe(ByteBuffer...) channels} and {@link #pSubscribe(ByteBuffer...) patterns}. It provides access to the
  * {@link ChannelMessage} {@link #receive() stream} that emits only messages for channels and patterns registered in
  * this {@link ReactiveSubscription}.
- * <p/>
+ * <p />
  * A reactive Redis connection can have multiple subscriptions. If two or more subscriptions subscribe to the same
  * target (channel/pattern) and one unsubscribes, then the other one will no longer receive messages for the target due
  * to how Redis handled Pub/Sub subscription.
  * 
  * @author Mark Paluch
+ * @author Christoph Strobl
  * @since 2.1
  */
 public interface ReactiveSubscription {
@@ -54,8 +57,8 @@ public interface ReactiveSubscription {
 	Mono<Void> pSubscribe(ByteBuffer... patterns);
 
 	/**
-	 * Cancels the current subscription for all {@link #getChannels() channels} given by name.
-	 * 
+	 * Cancels the current subscription for all {@link #getChannels() channels}.
+	 *
 	 * @return empty {@link Mono} that completes once the channel subscriptions are unregistered.
 	 */
 	Mono<Void> unsubscribe();
@@ -64,13 +67,13 @@ public interface ReactiveSubscription {
 	 * Cancels the current subscription for all given channels.
 	 *
 	 * @param channels channel names. Must not be empty.
-	 * @return empty {@link Mono} that completes once the channel subscription is unregistered.
+	 * @return empty {@link Mono} that completes once the channel subscriptions are unregistered.
 	 */
 	Mono<Void> unsubscribe(ByteBuffer... channels);
 
 	/**
 	 * Cancels the subscription for all channels matched by {@link #getPatterns()} patterns}.
-	 * 
+	 *
 	 * @return empty {@link Mono} that completes once the patterns subscriptions are unregistered.
 	 */
 	Mono<Void> pUnsubscribe();
@@ -79,91 +82,136 @@ public interface ReactiveSubscription {
 	 * Cancels the subscription for all channels matching the given patterns.
 	 *
 	 * @param patterns must not be empty.
-	 * @return empty {@link Mono} that completes once the patterns subscription is unregistered.
+	 * @return empty {@link Mono} that completes once the patterns subscriptions are unregistered.
 	 */
 	Mono<Void> pUnsubscribe(ByteBuffer... patterns);
 
 	/**
 	 * Returns the (named) channels for this subscription.
 	 *
-	 * @return collection of named channels
+	 * @return {@link Set} of named channels.
 	 */
-	Collection<ByteBuffer> getChannels();
+	Set<ByteBuffer> getChannels();
 
 	/**
 	 * Returns the channel patters for this subscription.
 	 *
-	 * @return collection of channel patterns
+	 * @return {@link Set} of channel patterns.
 	 */
-	Collection<ByteBuffer> getPatterns();
+	Set<ByteBuffer> getPatterns();
 
 	/**
-	 * Retrieve the message stream emitting {@link ChannelMessage} and {@link PatternMessage}. The resulting message
-	 * stream contains only messages for subscribed and registered {@link #getChannels() channels} and
-	 * {@link #getPatterns() patterns}.
-	 * <p/>
+	 * Retrieve the message stream emitting {@link Message messages}. The resulting message stream contains only messages
+	 * for subscribed and registered {@link #getChannels() channels} and {@link #getPatterns() patterns}.
+	 * <p />
 	 * Stream publishing uses {@link reactor.core.publisher.ConnectableFlux} turning the stream into a hot sequence.
 	 * Emission is paused if there is no demand. Messages received in that time are buffered. This stream terminates
-	 * either if all subscribers unsubscribe or if this {@link Subscription} is {@link #terminate() is terminated}.
-	 * 
-	 * @return the message stream.
+	 * either if all subscribers unsubscribe or if this {@link Subscription} is {@link #cancel() is terminated}.
+	 *
+	 * @return {@link Flux} emitting the {@link Message} stream.
 	 */
-	Flux<ChannelMessage<ByteBuffer, ByteBuffer>> receive();
+	Flux<Message<ByteBuffer, ByteBuffer>> receive();
 
 	/**
 	 * Unsubscribe from all {@link #getChannels() channels} and {@link #getPatterns() patterns} and request termination of
 	 * all active {@link #receive() message streams}. Active streams will terminate with a
 	 * {@link java.util.concurrent.CancellationException}.
-	 * 
+	 *
 	 * @return a {@link Mono} that completes once termination is finished.
 	 */
-	Mono<Void> terminate();
+	Mono<Void> cancel();
+
+	/**
+	 * {@link Message} represents a Redis channel message within Redis pub/sub.
+	 *
+	 * @param <C> channel representation type.
+	 * @param <M> message representation type.
+	 * @author Christoph Strobl
+	 * @since 2.1
+	 */
+	interface Message<C, M> {
+
+		/**
+		 * Get the channel the message published to.
+		 *
+		 * @return never {@literal null}.
+		 */
+		C getChannel();
+
+		/**
+		 * Get the actual message body.
+		 *
+		 * @return never {@literal null}.
+		 */
+		M getMessage();
+	}
 
 	/**
 	 * Value object for a Redis channel message.
-	 * 
+	 *
 	 * @param <C> type of how the channel name is represented.
-	 * @param <B> type of how the message is represented.
+	 * @param <M> type of how the message is represented.
 	 * @author Mark Paluch
+	 * @author Christoph Strobl
 	 * @since 2.1
 	 */
 	@EqualsAndHashCode
-	class ChannelMessage<C, B> {
+	class ChannelMessage<C, M> implements Message<C, M> {
 
 		private final C channel;
-		private final B message;
+		private final M message;
 
 		/**
 		 * Create a new {@link ChannelMessage}.
-		 * 
+		 *
 		 * @param channel must not be {@literal null}.
 		 * @param message must not be {@literal null}.
 		 */
-		public ChannelMessage(C channel, B message) {
+		public ChannelMessage(C channel, M message) {
+
+			Assert.notNull(channel, "Channel must not be null!");
+			Assert.notNull(message, "Message must not be null!");
+
 			this.channel = channel;
 			this.message = message;
 		}
 
+		/*
+		 * (non-Javadoc)
+		 * @see org.springframework.data.redis.connection.ReactiveSubscription.Message#getChannel()
+		 */
+		@Override
 		public C getChannel() {
 			return channel;
 		}
 
-		public B getMessage() {
+		/*
+		 * (non-Javadoc)
+		 * @see org.springframework.data.redis.connection.ReactiveSubscription.Message#getMessage()
+		 */
+		@Override
+		public M getMessage() {
 			return message;
 		}
+
+		@Override
+		public String toString() {
+			return "ChannelMessage {" + "channel=" + channel + ", message=" + message + '}';
+		}
 	}
 
 	/**
 	 * Value object for a Redis channel message received from a pattern subscription.
-	 * 
-	 * @param <C> type of how the pattern is represented.
+	 *
+	 * @param <P> type of how the pattern is represented.
 	 * @param <C> type of how the channel name is represented.
-	 * @param <B> type of how the message is represented.
+	 * @param <M> type of how the message is represented.
 	 * @author Mark Paluch
+	 * @author Christoph Strobl
 	 * @since 2.1
 	 */
 	@EqualsAndHashCode(callSuper = true)
-	class PatternMessage<P, C, B> extends ChannelMessage<C, B> {
+	class PatternMessage<P, C, M> extends ChannelMessage<C, M> {
 
 		private final P pattern;
 
@@ -174,14 +222,26 @@ public interface ReactiveSubscription {
 		 * @param channel must not be {@literal null}.
 		 * @param message must not be {@literal null}.
 		 */
-		public PatternMessage(P pattern, C channel, B message) {
+		public PatternMessage(P pattern, C channel, M message) {
 
 			super(channel, message);
+
+			Assert.notNull(pattern, "Pattern must not be null!");
 			this.pattern = pattern;
 		}
 
+		/**
+		 * Get the pattern that matched the channel.
+		 *
+		 * @return never {@literal null}.
+		 */
 		public P getPattern() {
 			return pattern;
 		}
+
+		@Override
+		public String toString() {
+			return "PatternMessage{" + "channel=" + getChannel() + ", pattern=" + pattern + ", message=" + getMessage() + '}';
+		}
 	}
 }

src/main/java/org/springframework/data/redis/connection/lettuce/LettuceReactivePubSubCommands.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2017 the original author or authors.
+ * Copyright 2018 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.
@@ -25,69 +25,73 @@ import java.nio.ByteBuffer;
 import java.util.function.Function;
 
 import org.reactivestreams.Publisher;
-import org.springframework.data.redis.connection.ReactiveRedisPubSubCommands;
+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.util.Assert;
 
 /**
  * @author Mark Paluch
+ * @author Christoph Strobl
  * @since 2.1
  */
 @RequiredArgsConstructor
-class LettuceReactivePubSubCommands implements ReactiveRedisPubSubCommands {
+class LettuceReactivePubSubCommands implements ReactivePubSubCommands {
 
 	private final @NonNull LettuceReactiveRedisConnection connection;
 
 	/*
 	 * (non-Javadoc)
-	 * @see org.springframework.data.redis.connection.ReactiveRedisPubSubCommands#createSubscription()
+	 * @see org.springframework.data.redis.connection.ReactivePubSubCommands#createSubscription()
 	 */
 	@Override
 	public Mono<ReactiveSubscription> createSubscription() {
+
 		return connection.getPubSubConnection()
-				.map(c -> new LettuceReactiveSubscription(c.reactive(), connection.translateException()));
+				.map(pubSubConnection -> new LettuceReactiveSubscription(pubSubConnection.reactive(),
+						connection.translateException()));
 	}
 
 	/*
 	 * (non-Javadoc)
-	 * @see org.springframework.data.redis.connection.ReactiveRedisPubSubCommands#publish(org.reactivestreams.Publisher)
+	 * @see org.springframework.data.redis.connection.ReactivePubSubCommands#publish(org.reactivestreams.Publisher)
 	 */
 	@Override
 	public Flux<Long> publish(Publisher<ChannelMessage<ByteBuffer, ByteBuffer>> messageStream) {
 
 		Assert.notNull(messageStream, "ChannelMessage stream must not be null!");
 
-		return connection.getCommands().flatMapMany(
-				c -> Flux.from(messageStream).flatMap(message -> c.publish(message.getChannel(), message.getMessage())));
+		return connection.getCommands().flatMapMany(commands -> Flux.from(messageStream)
+				.flatMap(message -> commands.publish(message.getChannel(), message.getMessage())));
 	}
 
 	/*
 	 * (non-Javadoc)
-	 * @see org.springframework.data.redis.connection.ReactiveRedisPubSubCommands#subscribe(java.nio.ByteBuffer[])
+	 * @see org.springframework.data.redis.connection.ReactivePubSubCommands#subscribe(java.nio.ByteBuffer[])
 	 */
 	@Override
 	public Mono<Void> subscribe(ByteBuffer... channels) {
 
 		Assert.notNull(channels, "Channels must not be null!");
 
-		return doWithPubSub(c -> c.subscribe(channels));
+		return doWithPubSub(commands -> commands.subscribe(channels));
 	}
 
 	/*
 	 * (non-Javadoc)
-	 * @see org.springframework.data.redis.connection.ReactiveRedisPubSubCommands#pSubscribe(java.nio.ByteBuffer[])
+	 * @see org.springframework.data.redis.connection.ReactivePubSubCommands#pSubscribe(java.nio.ByteBuffer[])
 	 */
 	@Override
 	public Mono<Void> pSubscribe(ByteBuffer... patterns) {
 
 		Assert.notNull(patterns, "Patterns must not be null!");
 
-		return doWithPubSub(c -> c.psubscribe(patterns));
+		return doWithPubSub(commands -> commands.psubscribe(patterns));
 	}
 
 	private <T> Mono<T> doWithPubSub(Function<RedisPubSubReactiveCommands<ByteBuffer, ByteBuffer>, Mono<T>> function) {
-		return connection.getPubSubConnection().flatMap(c -> function.apply(c.reactive()))
+
+		return connection.getPubSubConnection().flatMap(pubSubConnection -> function.apply(pubSubConnection.reactive()))
 				.onErrorMap(connection.translateException());
 	}
 }

src/main/java/org/springframework/data/redis/connection/lettuce/LettuceReactiveRedisConnection.java

@@ -175,7 +175,7 @@ class LettuceReactiveRedisConnection implements ReactiveRedisConnection {
 	 * @see org.springframework.data.redis.connection.ReactiveRedisConnection#pubSubCommands()
 	 */
 	@Override
-	public ReactiveRedisPubSubCommands pubSubCommands() {
+	public ReactivePubSubCommands pubSubCommands() {
 		return new LettuceReactivePubSubCommands(this);
 	}
 
@@ -225,11 +225,10 @@ class LettuceReactiveRedisConnection implements ReactiveRedisConnection {
 
 	/*
 	 * (non-Javadoc)
-	 * @see java.io.Closeable#close()
+	 * @see org.springframework.data.redis.connection.ReactiveRedisConnection#closeLater()
 	 */
-	@Override
-	public void close() {
-		dedicatedConnection.close();
+	public Mono<Void> closeLater() {
+		return Mono.fromRunnable(() -> dedicatedConnection.close());
 	}
 
 	protected Mono<? extends StatefulConnection<ByteBuffer, ByteBuffer>> getConnection() {

src/main/java/org/springframework/data/redis/connection/lettuce/LettuceReactiveSubscription.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2017 the original author or authors.
+ * Copyright 2018 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.
@@ -24,7 +24,6 @@ import reactor.core.publisher.Mono;
 
 import java.nio.ByteBuffer;
 import java.util.Arrays;
-import java.util.Collection;
 import java.util.Collections;
 import java.util.List;
 import java.util.Set;
@@ -37,11 +36,13 @@ import java.util.function.Supplier;
 import org.springframework.data.redis.connection.ReactiveSubscription;
 import org.springframework.lang.Nullable;
 import org.springframework.util.Assert;
+import org.springframework.util.ObjectUtils;
 
 /**
  * Lettuce-specific implementation of {@link ReactiveSubscription}.
  *
  * @author Mark Paluch
+ * @author Christoph Strobl
  * @since 2.1
  */
 class LettuceReactiveSubscription implements ReactiveSubscription {
@@ -91,7 +92,7 @@ class LettuceReactiveSubscription implements ReactiveSubscription {
 	 */
 	@Override
 	public Mono<Void> unsubscribe() {
-		return unsubscribe(channelState.getTargets().toArray(new ByteBuffer[0]));
+		return unsubscribe(channelState.getTargets().toArray(new ByteBuffer[channelState.getTargets().size()]));
 	}
 
 	/*
@@ -104,7 +105,7 @@ class LettuceReactiveSubscription implements ReactiveSubscription {
 		Assert.notNull(channels, "Channels must not be null!");
 		Assert.noNullElements(channels, "Channels must not contain null elements!");
 
-		return channels.length == 0 ? Mono.empty() : channelState.unsubscribe(channels, commands::unsubscribe);
+		return ObjectUtils.isEmpty(channels) ? Mono.empty() : channelState.unsubscribe(channels, commands::unsubscribe);
 	}
 
 	/*
@@ -113,7 +114,7 @@ class LettuceReactiveSubscription implements ReactiveSubscription {
 	 */
 	@Override
 	public Mono<Void> pUnsubscribe() {
-		return pUnsubscribe(patternState.getTargets().toArray(new ByteBuffer[0]));
+		return pUnsubscribe(patternState.getTargets().toArray(new ByteBuffer[patternState.getTargets().size()]));
 	}
 
 	/*
@@ -126,7 +127,7 @@ class LettuceReactiveSubscription implements ReactiveSubscription {
 		Assert.notNull(patterns, "Patterns must not be null!");
 		Assert.noNullElements(patterns, "Patterns must not contain null elements!");
 
-		return patterns.length == 0 ? Mono.empty() : patternState.unsubscribe(patterns, commands::punsubscribe);
+		return ObjectUtils.isEmpty(patterns) ? Mono.empty() : patternState.unsubscribe(patterns, commands::punsubscribe);
 	}
 
 	/*
@@ -134,8 +135,8 @@ class LettuceReactiveSubscription implements ReactiveSubscription {
 	 * @see org.springframework.data.redis.connection.ReactiveSubscription#getChannels()
 	 */
 	@Override
-	public Collection<ByteBuffer> getChannels() {
-		return Collections.unmodifiableCollection(channelState.getTargets());
+	public Set<ByteBuffer> getChannels() {
+		return channelState.getTargets();
 	}
 
 	/*
@@ -143,8 +144,8 @@ class LettuceReactiveSubscription implements ReactiveSubscription {
 	 * @see org.springframework.data.redis.connection.ReactiveSubscription#getPatterns()
 	 */
 	@Override
-	public Collection<ByteBuffer> getPatterns() {
-		return Collections.unmodifiableCollection(patternState.getTargets());
+	public Set<ByteBuffer> getPatterns() {
+		return patternState.getTargets();
 	}
 
 	/*
@@ -152,15 +153,15 @@ class LettuceReactiveSubscription implements ReactiveSubscription {
 	 * @see org.springframework.data.redis.connection.ReactiveSubscription#receive()
 	 */
 	@Override
-	public Flux<ChannelMessage<ByteBuffer, ByteBuffer>> receive() {
+	public Flux<Message<ByteBuffer, ByteBuffer>> receive() {
 
-		Flux<ChannelMessage<ByteBuffer, ByteBuffer>> channelMessages = channelState.receive(() -> commands.observeChannels() //
-				.filter(m -> channelState.getTargets().contains(m.getChannel())) //
-				.map(m -> new ChannelMessage<>(m.getChannel(), m.getMessage())));
+		Flux<Message<ByteBuffer, ByteBuffer>> channelMessages = channelState.receive(() -> commands.observeChannels() //
+				.filter(message -> channelState.getTargets().contains(message.getChannel())) //
+				.map(message -> new ChannelMessage<>(message.getChannel(), message.getMessage())));
 
-		Flux<ChannelMessage<ByteBuffer, ByteBuffer>> patternMessages = patternState.receive(() -> commands.observePatterns() //
-				.filter(m -> patternState.getTargets().contains(m.getPattern())) //
-				.map(m -> new PatternMessage<>(m.getPattern(), m.getChannel(), m.getMessage())));
+		Flux<Message<ByteBuffer, ByteBuffer>> patternMessages = patternState.receive(() -> commands.observePatterns() //
+				.filter(message -> patternState.getTargets().contains(message.getPattern())) //
+				.map(message -> new PatternMessage<>(message.getPattern(), message.getChannel(), message.getMessage())));
 
 		return channelMessages.mergeWith(patternMessages);
 	}
@@ -170,7 +171,7 @@ class LettuceReactiveSubscription implements ReactiveSubscription {
 	 * @see org.springframework.data.redis.connection.ReactiveSubscription#terminate()
 	 */
 	@Override
-	public Mono<Void> terminate() {
+	public Mono<Void> cancel() {
 
 		return unsubscribe().then(pUnsubscribe()).then(Mono.defer(() -> {
 
@@ -204,9 +205,8 @@ class LettuceReactiveSubscription implements ReactiveSubscription {
 		 */
 		Mono<Void> subscribe(ByteBuffer[] targets, Function<ByteBuffer[], Mono<Void>> subscribeFunction) {
 
-			return subscribeFunction.apply(targets).doOnSuccess((v) -> {
-				this.targets.addAll(Arrays.asList(targets));
-			}).onErrorMap(exceptionTranslator);
+			return subscribeFunction.apply(targets).doOnSuccess((discard) -> this.targets.addAll(Arrays.asList(targets)))
+					.onErrorMap(exceptionTranslator);
 		}
 
 		/**
@@ -223,14 +223,14 @@ class LettuceReactiveSubscription implements ReactiveSubscription {
 
 				List<ByteBuffer> targetCollection = Arrays.asList(targets);
 
-				return unsubscribeFunction.apply(targets).doOnSuccess((v) -> {
+				return unsubscribeFunction.apply(targets).doOnSuccess((discard) -> {
 					this.targets.removeAll(targetCollection);
 				}).onErrorMap(exceptionTranslator);
 			});
 		}
 
-		Collection<ByteBuffer> getTargets() {
-			return targets;
+		Set<ByteBuffer> getTargets() {
+			return Collections.unmodifiableSet(targets);
 		}
 
 		/**
@@ -254,13 +254,14 @@ class LettuceReactiveSubscription implements ReactiveSubscription {
 			}
 
 			ConnectableFlux<T> connectableFlux = connectFunction.get().onErrorMap(exceptionTranslator).publish();
-			Flux<T> fluxToUse = connectableFlux.doOnSubscribe(s -> {
+			Flux<T> fluxToUse = connectableFlux.doOnSubscribe(subscription -> {
 
 				if (subscribers.incrementAndGet() == 1) {
 					disposable = connectableFlux.connect();
 				}
-			}).doFinally(s -> {
+			}).doFinally(signalType -> {
 
+				// TODO: do new need to care about what happened?
 				if (subscribers.decrementAndGet() == 0) {
 
 					this.flux.compareAndSet(connectableFlux, null);

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

@@ -20,16 +20,22 @@ import reactor.core.publisher.Mono;
 
 import java.time.Duration;
 import java.time.Instant;
+import java.util.Arrays;
 import java.util.Collections;
 import java.util.List;
 
 import org.reactivestreams.Publisher;
 import org.springframework.data.redis.connection.DataType;
+import org.springframework.data.redis.connection.ReactiveSubscription.Message;
 import org.springframework.data.redis.core.script.RedisScript;
+import org.springframework.data.redis.listener.ChannelTopic;
+import org.springframework.data.redis.listener.PatternTopic;
+import org.springframework.data.redis.listener.Topic;
 import org.springframework.data.redis.serializer.RedisElementReader;
 import org.springframework.data.redis.serializer.RedisElementWriter;
 import org.springframework.data.redis.serializer.RedisSerializationContext;
 import org.springframework.data.redis.serializer.RedisSerializer;
+import org.springframework.util.Assert;
 
 /**
  * Interface that specified a basic set of Redis operations, implemented by {@link ReactiveRedisTemplate}. Not often
@@ -56,17 +62,59 @@ public interface ReactiveRedisOperations<K, V> {
 	 */
 	<T> Flux<T> execute(ReactiveRedisCallback<T> action);
 
+	// -------------------------------------------------------------------------
+	// Methods dealing with Redis Pub/Sub
+	// -------------------------------------------------------------------------
+
 	/**
 	 * Publishes the given message to the given channel.
 	 *
-	 * @param destination the channel to publish to, must not be {@literal null} or empty.
-	 * @param message message to publish.
+	 * @param destination the channel to publish to, must not be {@literal null} nor empty.
+	 * @param message message to publish. Must not be {@literal null}.
 	 * @return the number of clients that received the message
 	 * @since 2.1
 	 * @see <a href="http://redis.io/commands/publish">Redis Documentation: PUBLISH</a>
 	 */
 	Mono<Long> convertAndSend(String destination, V message);
 
+	/**
+	 * Subscribe to the given Redis {@code channels} and emit {@link Message messages} received for those.
+	 *
+	 * @param channels must not be {@literal null}.
+	 * @return a hot sequence of {@link Message messages}.
+	 * @since 2.1
+	 */
+	default Flux<? extends Message<String, V>> listenToChannel(String... channels) {
+
+		Assert.notNull(channels, "Channels must not be null!");
+
+		return listenTo(Arrays.stream(channels).map(ChannelTopic::of).toArray(ChannelTopic[]::new));
+	}
+
+	/**
+	 * Subscribe to the Redis channels matching the given {@code pattern} and emit {@link Message messages} received for
+	 * those.
+	 *
+	 * @param patterns must not be {@literal null}.
+	 * @return a hot sequence of {@link Message messages}.
+	 * @since 2.1
+	 */
+	default Flux<? extends Message<String, V>> listenToPattern(String... patterns) {
+
+		Assert.notNull(patterns, "Patterns must not be null!");
+		return listenTo(Arrays.stream(patterns).map(PatternTopic::of).toArray(PatternTopic[]::new));
+	}
+
+	/**
+	 * Subscribe to the Redis channels for the given {@link Topic topics} and emit {@link Message messages} received for
+	 * those.
+	 *
+	 * @param topics must not be {@literal null}.
+	 * @return a hot sequence of {@link Message messages}.
+	 * @since 2.1
+	 */
+	Flux<? extends Message<String, V>> listenTo(Topic... topics);
+
 	// -------------------------------------------------------------------------
 	// Methods dealing with Redis Keys
 	// -------------------------------------------------------------------------

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

@@ -17,11 +17,13 @@ package org.springframework.data.redis.core;
 
 import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
+import reactor.core.scheduler.Schedulers;
 
 import java.lang.reflect.Proxy;
 import java.nio.ByteBuffer;
 import java.time.Duration;
 import java.time.Instant;
+import java.util.Arrays;
 import java.util.List;
 
 import org.reactivestreams.Publisher;
@@ -30,9 +32,12 @@ import org.springframework.data.redis.connection.ReactiveRedisConnection;
 import org.springframework.data.redis.connection.ReactiveRedisConnection.CommandResponse;
 import org.springframework.data.redis.connection.ReactiveRedisConnection.KeyCommand;
 import org.springframework.data.redis.connection.ReactiveRedisConnectionFactory;
+import org.springframework.data.redis.connection.ReactiveSubscription.Message;
 import org.springframework.data.redis.core.script.DefaultReactiveScriptExecutor;
 import org.springframework.data.redis.core.script.ReactiveScriptExecutor;
 import org.springframework.data.redis.core.script.RedisScript;
+import org.springframework.data.redis.listener.ReactiveRedisMessageListenerContainer;
+import org.springframework.data.redis.listener.Topic;
 import org.springframework.data.redis.serializer.RedisElementReader;
 import org.springframework.data.redis.serializer.RedisElementWriter;
 import org.springframework.data.redis.serializer.RedisSerializationContext;
@@ -193,16 +198,36 @@ public class ReactiveRedisTemplate<K, V> implements ReactiveRedisOperations<K, V
 		return Flux.from(postProcessResult(result, connToUse, false)).doFinally(signal -> conn.close());
 	}
 
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.redis.core.ReactiveRedisOperations#convertAndSend(java.lang.String, java.lang.Object)
+	 */
 	@Override
 	public Mono<Long> convertAndSend(String destination, V message) {
 
 		Assert.hasText(destination, "Destination channel must not be empty!");
+		Assert.notNull(message, "Message must not be null!");
 
 		return createMono(connection -> connection.pubSubCommands().publish(
 				getSerializationContext().getStringSerializationPair().write(destination),
 				getSerializationContext().getValueSerializationPair().write(message)));
 	}
 
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.redis.core.ReactiveRedisOperations#listenTo(org.springframework.data.redis.listener.Topic[])
+	 */
+	@Override
+	public Flux<? extends Message<String, V>> listenTo(Topic... topics) {
+
+		ReactiveRedisMessageListenerContainer container = new ReactiveRedisMessageListenerContainer(getConnectionFactory());
+
+		return container
+				.receive(Arrays.asList(topics), getSerializationContext().getStringSerializationPair(),
+						getSerializationContext().getValueSerializationPair()) //
+				.doFinally((signalType) -> container.destroyLater().subscribeOn(Schedulers.elastic()));
+	}
+
 	// -------------------------------------------------------------------------
 	// Methods dealing with Redis keys
 	// -------------------------------------------------------------------------

src/main/java/org/springframework/data/redis/core/convert/MappingRedisConverter.java

@@ -21,8 +21,17 @@ import lombok.Getter;
 import lombok.RequiredArgsConstructor;
 
 import java.lang.reflect.Array;
-import java.util.*;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.Comparator;
+import java.util.HashMap;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map;
 import java.util.Map.Entry;
+import java.util.Optional;
+import java.util.Set;
 import java.util.regex.Matcher;
 import java.util.regex.Pattern;
 
@@ -35,7 +44,6 @@ import org.springframework.core.convert.support.GenericConversionService;
 import org.springframework.data.convert.CustomConversions;
 import org.springframework.data.convert.EntityInstantiator;
 import org.springframework.data.convert.EntityInstantiators;
-import org.springframework.data.keyvalue.core.mapping.KeySpaceResolver;
 import org.springframework.data.mapping.AssociationHandler;
 import org.springframework.data.mapping.MappingException;
 import org.springframework.data.mapping.PersistentPropertyAccessor;
@@ -450,9 +458,10 @@ public class MappingRedisConverter implements RedisConverter, InitializingBean {
 			targetProperty = getTargetPropertyOrNullForPath(path.replaceAll("\\.\\[.*\\]", ""), update.getTarget());
 
 			TypeInformation<?> ti = targetProperty == null ? ClassTypeInformation.OBJECT
-					: (targetProperty.isMap() ? (targetProperty.getTypeInformation().getMapValueType() != null
-							? targetProperty.getTypeInformation().getRequiredMapValueType()
-							: ClassTypeInformation.OBJECT) : targetProperty.getTypeInformation().getActualType());
+					: (targetProperty.isMap()
+							? (targetProperty.getTypeInformation().getMapValueType() != null
+									? targetProperty.getTypeInformation().getRequiredMapValueType() : ClassTypeInformation.OBJECT)
+							: targetProperty.getTypeInformation().getActualType());
 
 			writeInternal(entity.getKeySpace(), pUpdate.getPropertyPath(), pUpdate.getValue(), ti, sink);
 			return;

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

@@ -44,7 +44,7 @@ public class ChannelTopic implements Topic {
 
 	/**
 	 * Create a new {@link ChannelTopic} for channel subscriptions.
-	 * 
+	 *
 	 * @param name the channel name, must not be {@literal null} or empty.
 	 * @return the {@link ChannelTopic} for {@code channelName}.
 	 * @since 2.1

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

@@ -24,6 +24,7 @@ import org.springframework.util.Assert;
  *
  * @author Costin Leau
  * @author Mark Paluch
+ * @author Christoph Strobl
  */
 @EqualsAndHashCode
 public class PatternTopic implements Topic {
@@ -49,7 +50,7 @@ public class PatternTopic implements Topic {
 	 * @return the {@link PatternTopic} for {@code pattern}.
 	 * @since 2.1
 	 */
-	static PatternTopic of(String pattern) {
+	public static PatternTopic of(String pattern) {
 		return new PatternTopic(pattern);
 	}
 

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2017 the original author or authors.
+ * Copyright 2018 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.
@@ -18,12 +18,12 @@ 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;
 import java.util.Collection;
 import java.util.HashMap;
-import java.util.HashSet;
 import java.util.List;
 import java.util.Map;
 import java.util.Set;
@@ -34,16 +34,19 @@ import java.util.stream.StreamSupport;
 
 import org.springframework.beans.factory.DisposableBean;
 import org.springframework.dao.InvalidDataAccessApiUsageException;
+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.connection.ReactiveSubscription.ChannelMessage;
+import org.springframework.data.redis.connection.ReactiveSubscription.Message;
 import org.springframework.data.redis.connection.ReactiveSubscription.PatternMessage;
 import org.springframework.data.redis.serializer.RedisElementReader;
 import org.springframework.data.redis.serializer.RedisSerializationContext.SerializationPair;
 import org.springframework.data.redis.serializer.RedisSerializer;
 import org.springframework.lang.Nullable;
 import org.springframework.util.Assert;
+import org.springframework.util.ObjectUtils;
 
 /**
  * Container providing a stream of {@link ChannelMessage} for messages received via Redis Pub/Sub listeners. The stream
@@ -55,11 +58,12 @@ import org.springframework.util.Assert;
  * operations. Using reactive infrastructure allows usage of a single connection due to channel multiplexing.
  * <p />
  * This class is thread-safe and allows subscription by multiple concurrent threads.
- * 
+ *
  * @author Mark Paluch
+ * @author Christoph Strobl
  * @since 2.1
  * @see ReactiveSubscription
- * @see org.springframework.data.redis.connection.ReactiveRedisPubSubCommands
+ * @see ReactivePubSubCommands
  */
 public class ReactiveRedisMessageListenerContainer implements DisposableBean {
 
@@ -71,7 +75,7 @@ public class ReactiveRedisMessageListenerContainer implements DisposableBean {
 
 	/**
 	 * Create a new {@link ReactiveRedisMessageListenerContainer} given {@link ReactiveRedisConnectionFactory}.
-	 * 
+	 *
 	 * @param connectionFactory must not be {@literal null}.
 	 */
 	public ReactiveRedisMessageListenerContainer(ReactiveRedisConnectionFactory connectionFactory) {
@@ -86,8 +90,13 @@ public class ReactiveRedisMessageListenerContainer implements DisposableBean {
 	 */
 	@Override
 	public void destroy() {
+		destroyLater().block();
+	}
 
-		ReactiveRedisConnection connection = this.connection;
+	/**
+	 * @return the {@link Mono} signalling container termination.
+	 */
+	public Mono<Void> destroyLater() {
 
 		if (connection != null) {
 
@@ -97,55 +106,49 @@ public class ReactiveRedisMessageListenerContainer implements DisposableBean {
 				Map<ReactiveSubscription, Subscribers> local = new HashMap<>(subscriptions);
 				List<Mono<Void>> monos = local.keySet().stream() //
 						.peek(subscriptions::remove) //
-						.map(ReactiveSubscription::terminate) //
+						.map(ReactiveSubscription::cancel) //
 						.collect(Collectors.toList());
 
 				if (terminationSignals == null) {
-					terminationSignals = Flux.merge(monos);
+					terminationSignals = Flux.concat(monos);
 				} else {
-					terminationSignals = terminationSignals.mergeWith(Flux.merge(monos));
+					terminationSignals = terminationSignals.mergeWith(Flux.concat(monos));
 				}
 			}
 
 			if (terminationSignals != null) {
-				terminationSignals.blockLast();
+				return terminationSignals.collectList()
+						.doFinally(signalType -> connection.closeLater().subscribeOn(Schedulers.immediate()))
+						.flatMap(all -> Mono.empty());
 			}
-
-			connection.close();
 			this.connection = null;
 		}
+
+		return Mono.empty();
 	}
 
 	/**
 	 * Return the currently active {@link ReactiveSubscription subscriptions}.
-	 * 
+	 *
 	 * @return {@link Set} of active {@link ReactiveSubscription}
 	 */
 	public Collection<ReactiveSubscription> getActiveSubscriptions() {
 
-		Set<ReactiveSubscription> subscriptions = new HashSet<>(this.subscriptions.size(), 1);
-
-		this.subscriptions.forEach((subscription, subscribers) -> {
-
-			if (subscribers.hasRegistration()) {
-				subscriptions.add(subscription);
-			}
-		});
-
-		return subscriptions;
+		return subscriptions.entrySet().stream().filter(entry -> entry.getValue().hasRegistration())
+				.map(entry -> entry.getKey()).collect(Collectors.toList());
 	}
 
 	/**
 	 * Subscribe to one or more {@link ChannelTopic}s and receive a stream of {@link ChannelMessage}. Messages and channel
 	 * names are treated as {@link String}. The message stream subscribes lazily to the Redis channels and unsubscribes if
 	 * the {@link org.reactivestreams.Subscription} is {@link org.reactivestreams.Subscription#cancel() cancelled}.
-	 * 
+	 *
 	 * @param channelTopics the channels to subscribe.
 	 * @return the message stream.
 	 * @throws InvalidDataAccessApiUsageException if {@code patternTopics} is empty.
 	 * @see #receive(Iterable, SerializationPair, SerializationPair)
 	 */
-	public Flux<ChannelMessage<String, String>> receive(ChannelTopic... channelTopics) {
+	public Flux<Message<String, String>> receive(ChannelTopic... channelTopics) {
 
 		Assert.notNull(channelTopics, "ChannelTopics must not be null!");
 		Assert.noNullElements(channelTopics, "ChannelTopics must not contain null elements!");
@@ -158,8 +161,8 @@ public class ReactiveRedisMessageListenerContainer implements DisposableBean {
 	 * and channel names are treated as {@link String}. The message stream subscribes lazily to the Redis channels and
 	 * unsubscribes if the {@link org.reactivestreams.Subscription} is {@link org.reactivestreams.Subscription#cancel()
 	 * cancelled}.
-	 * 
-	 * @param channelTopics the channels to subscribe.
+	 *
+	 * @param patternTopics the channels to subscribe.
 	 * @return the message stream.
 	 * @throws InvalidDataAccessApiUsageException if {@code patternTopics} is empty.
 	 * @see #receive(Iterable, SerializationPair, SerializationPair)
@@ -180,39 +183,35 @@ public class ReactiveRedisMessageListenerContainer implements DisposableBean {
 	 * given {@code channelSerializer} and {@code messageSerializer}. The message stream subscribes lazily to the Redis
 	 * channels and unsubscribes if the {@link org.reactivestreams.Subscription} is
 	 * {@link org.reactivestreams.Subscription#cancel() cancelled}.
-	 * 
+	 *
 	 * @param topics the channels to subscribe.
 	 * @return the message stream.
 	 * @see #receive(Iterable, SerializationPair, SerializationPair)
 	 * @throws InvalidDataAccessApiUsageException if {@code topics} is empty.
 	 */
-	public <C, B> Flux<ChannelMessage<C, B>> receive(Iterable<? extends Topic> topics,
-			SerializationPair<C> channelSerializer, SerializationPair<B> messageSerializer) {
+	public <C, B> Flux<Message<C, B>> receive(Iterable<? extends Topic> topics, SerializationPair<C> channelSerializer,
+			SerializationPair<B> messageSerializer) {
 
 		Assert.notNull(topics, "Topics must not be null!");
 
-		ReactiveRedisConnection connection = this.connection;
-		if (connection == null) {
-			throw new IllegalStateException("ReactiveRedisMessageListenerContainer is already disposed!");
-		}
-
-		Mono<ReactiveSubscription> subscription = connection.pubSubCommands().createSubscription();
+		verifyConnection();
 
 		ByteBuffer[] patterns = getTargets(topics, PatternTopic.class);
 		ByteBuffer[] channels = getTargets(topics, ChannelTopic.class);
 
-		if (patterns.length == 0 && channels.length == 0) {
-			throw new InvalidDataAccessApiUsageException("No channels or patterns to subscribe");
+		if (ObjectUtils.isEmpty(patterns) && ObjectUtils.isEmpty(channels)) {
+			throw new InvalidDataAccessApiUsageException("No channels or patterns to subscribe to.");
 		}
 
-		return doReceive(channelSerializer, messageSerializer, subscription, patterns, channels);
+		return doReceive(channelSerializer, messageSerializer, connection.pubSubCommands().createSubscription(), patterns,
+				channels);
 	}
 
-	private <C, B> Flux<ChannelMessage<C, B>> doReceive(SerializationPair<C> channelSerializer,
+	private <C, B> Flux<Message<C, B>> doReceive(SerializationPair<C> channelSerializer,
 			SerializationPair<B> messageSerializer, Mono<ReactiveSubscription> subscription, ByteBuffer[] patterns,
 			ByteBuffer[] channels) {
 
-		Flux<ChannelMessage<ByteBuffer, ByteBuffer>> messageStream = subscription.flatMapMany(it -> {
+		Flux<Message<ByteBuffer, ByteBuffer>> messageStream = subscription.flatMapMany(it -> {
 
 			Mono<Void> subscribe = subscribe(patterns, channels, it);
 
@@ -238,15 +237,16 @@ public class ReactiveRedisMessageListenerContainer implements DisposableBean {
 
 	private static Mono<Void> subscribe(ByteBuffer[] patterns, ByteBuffer[] channels, ReactiveSubscription it) {
 
-		Assert.isTrue(channels.length != 0 || patterns.length != 0, "Must provide either channels or patterns!");
+		Assert.isTrue(!ObjectUtils.isEmpty(channels) || !ObjectUtils.isEmpty(patterns),
+				"Must provide either channels or patterns!");
 
 		Mono<Void> subscribe = null;
 
-		if (patterns.length != 0) {
+		if (!ObjectUtils.isEmpty(patterns)) {
 			subscribe = it.pSubscribe(patterns);
 		}
 
-		if (channels.length != 0) {
+		if (!ObjectUtils.isEmpty(channels)) {
 
 			Mono<Void> channelsSubscribe = it.subscribe(channels);
 
@@ -260,6 +260,17 @@ public class ReactiveRedisMessageListenerContainer implements DisposableBean {
 		return subscribe;
 	}
 
+	private boolean isActive() {
+		return connection != null;
+	}
+
+	private void verifyConnection() {
+
+		if (!isActive()) {
+			throw new IllegalStateException("ReactiveRedisMessageListenerContainer is already disposed!");
+		}
+	}
+
 	private Subscribers getSubscribers(ReactiveSubscription it) {
 		return subscriptions.computeIfAbsent(it, key -> new Subscribers());
 	}
@@ -274,8 +285,8 @@ public class ReactiveRedisMessageListenerContainer implements DisposableBean {
 	}
 
 	@SuppressWarnings("unchecked")
-	private <C, B> ChannelMessage<C, B> readMessage(RedisElementReader<C> channelSerializer,
-			RedisElementReader<B> messageSerializer, ChannelMessage<ByteBuffer, ByteBuffer> message) {
+	private <C, B> Message<C, B> readMessage(RedisElementReader<C> channelSerializer,
+			RedisElementReader<B> messageSerializer, Message<ByteBuffer, ByteBuffer> message) {
 
 		if (message instanceof PatternMessage) {
 
@@ -306,7 +317,7 @@ public class ReactiveRedisMessageListenerContainer implements DisposableBean {
 
 	/**
 	 * Object to track subscriber count and to determine the last unsubscribed subscriber.
-	 * 
+	 *
 	 * @author Mark Paluch
 	 */
 	static class Subscribers {

src/main/java/org/springframework/data/redis/serializer/DefaultRedisElementReader.java

@@ -19,6 +19,7 @@ import lombok.RequiredArgsConstructor;
 
 import java.nio.ByteBuffer;
 
+import org.springframework.data.redis.util.ByteUtils;
 import org.springframework.lang.Nullable;
 
 /**
@@ -44,9 +45,7 @@ class DefaultRedisElementReader<T> implements RedisElementReader<T> {
 			return (T) buffer;
 		}
 
-		byte[] bytes = new byte[buffer.slice().remaining()];
-		buffer.get(bytes);
-
-		return serializer.deserialize(bytes);
+		return serializer.deserialize(ByteUtils.extractBytes(buffer));
 	}
+
 }

src/main/java/org/springframework/data/redis/util/ByteUtils.java

@@ -232,4 +232,21 @@ public final class ByteUtils {
 
 		return charset.encode(theString);
 	}
+
+	/**
+	 * Extract/Transfer bytes from the given {@link ByteBuffer} into an array by duplicating the buffer and fetching its
+	 * content.
+	 *
+	 * @param buffer must not be {@literal null}.
+	 * @return the extracted bytes.
+	 * @since 2.1
+	 */
+	public static byte[] extractBytes(ByteBuffer buffer) {
+
+		ByteBuffer duplicate = buffer.duplicate();
+		byte[] bytes = new byte[duplicate.remaining()];
+		duplicate.get(bytes);
+
+		return bytes;
+	}
 }