DATAREDIS-612 - Send and receive Pub/Sub messages using reactive connections.

We now provide sending and receiving of Redis Pub/Sub messages with reactive connections. Messages can be sent with ReactiveRedisTemplate and received as infinite stream through ReactiveRedisMessageListenerContainer. ReactiveRedisMessageListenerContainer uses a single connection to multiplex different subscriptions.

Reactive Pub/Sub supports Standalone, Sentinel and Redis Cluster setups.

ReactiveRedisConnectionFactory factory = …;

ReactiveRedisTemplate<String, String> template = new ReactiveRedisTemplate<>(factory, RedisSerializationContext.string());
Mono<Long> publish = template.convertAndSend("hello!", "world");

ReactiveRedisMessageListenerContainer container = new ReactiveRedisMessageListenerContainer(factory);
Flux<ChannelMessage<String, String>> channelMessages = container.receive(new ChannelTopic("foo"));
Flux<PatternMessage<String, String, String>> patternMessages = container.receive(new PatternTopic("foo"));

//
container.destroy();

Original Pull Request: #295

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

@@ -116,6 +116,14 @@ public interface ReactiveRedisConnection extends Closeable {
 	 */
 	ReactiveHyperLogLogCommands hyperLogLogCommands();
 
+	/**
+	 * Get {@link ReactiveRedisPubSubCommands}.
+	 *
+	 * @return never {@literal null}.
+	 * @since 2.1
+	 */
+	ReactiveRedisPubSubCommands pubSubCommands();
+
 	/**
 	 * Get {@link ReactiveScriptingCommands}.
 	 *

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

@@ -0,0 +1,85 @@
+/*
+ * Copyright 2017 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
+ *
+ *      http://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.connection;
+
+import reactor.core.publisher.Flux;
+import reactor.core.publisher.Mono;
+
+import java.nio.ByteBuffer;
+
+import org.reactivestreams.Publisher;
+import org.springframework.data.redis.connection.ReactiveSubscription.ChannelMessage;
+
+/**
+ * Redis <a href="https://redis.io/commands/#pubsub">Pub/Sub</a> commands executed using reactive infrastructure.
+ *
+ * @author Mark Paluch
+ * @since 2.1
+ */
+public interface ReactiveRedisPubSubCommands {
+
+	/**
+	 * Creates a subscription for this connection. Connections can have multiple {@link ReactiveSubscription}s.
+	 *
+	 * @return the subscription.
+	 */
+	Mono<ReactiveSubscription> createSubscription();
+
+	/**
+	 * Publishes the given {@code message} to the given {@code channel}.
+	 *
+	 * @param channel the channel to publish to. Must not be {@literal null}.
+	 * @param message message to publish. Must not be {@literal null}.
+	 * @return the number of clients that received the message.
+	 * @see <a href="http://redis.io/commands/publish">Redis Documentation: PUBLISH</a>
+	 */
+	default Mono<Long> publish(ByteBuffer channel, ByteBuffer message) {
+		return publish(Mono.just(new ChannelMessage<>(channel, message))).next();
+	}
+
+	/**
+	 * Publishes the given messages to the {@link ChannelMessage#getChannel() appropriate channels}.
+	 *
+	 * @param messageStream the messages to publish to. Must not be {@literal null}.
+	 * @return the number of clients that received the message.
+	 * @see <a href="http://redis.io/commands/publish">Redis Documentation: PUBLISH</a>
+	 */
+	Flux<Long> publish(Publisher<ChannelMessage<ByteBuffer, ByteBuffer>> messageStream);
+
+	/**
+	 * 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/>
+	 * Note that cancellation of the {@link Flux} will unsubscribe from {@code channels}.
+	 *
+	 * @param channels channel names, must not be {@literal null}.
+	 * @see <a href="http://redis.io/commands/subscribe">Redis Documentation: SUBSCRIBE</a>
+	 */
+	Mono<Void> subscribe(ByteBuffer... channels);
+
+	/**
+	 * Subscribes the connection to all channels matching the given {@code patterns}. 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 />
+	 * Note that cancellation of the {@link Flux} will unsubscribe from {@code patterns}.
+	 *
+	 * @param patterns channel name patterns, must not be {@literal null}.
+	 * @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/ReactiveSubscription.java

@@ -0,0 +1,187 @@
+/*
+ * Copyright 2017 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
+ *
+ *      http://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.connection;
+
+import lombok.EqualsAndHashCode;
+import reactor.core.publisher.Flux;
+import reactor.core.publisher.Mono;
+
+import java.nio.ByteBuffer;
+import java.util.Collection;
+
+/**
+ * Subscription for Redis channels using reactive infrastructure. A {@link ReactiveSubscription} allows subscription 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/>
+ * 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
+ * @since 2.1
+ */
+public interface ReactiveSubscription {
+
+	/**
+	 * Subscribes to the {@code channels} and adds these to the current subscription.
+	 *
+	 * @param channels channel names. Must not be empty.
+	 * @return empty {@link Mono} that completes once the channel subscription is registered.
+	 */
+	Mono<Void> subscribe(ByteBuffer... channels);
+
+	/**
+	 * Subscribes to the channel {@code patterns} and adds these to the current subscription.
+	 *
+	 * @param patterns channel patterns. Must not be empty.
+	 * @return empty {@link Mono} that completes once the pattern subscription is registered.
+	 */
+	Mono<Void> pSubscribe(ByteBuffer... patterns);
+
+	/**
+	 * Cancels the current subscription for all {@link #getChannels() channels} given by name.
+	 * 
+	 * @return empty {@link Mono} that completes once the channel subscriptions are unregistered.
+	 */
+	Mono<Void> unsubscribe();
+
+	/**
+	 * 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.
+	 */
+	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();
+
+	/**
+	 * 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.
+	 */
+	Mono<Void> pUnsubscribe(ByteBuffer... patterns);
+
+	/**
+	 * Returns the (named) channels for this subscription.
+	 *
+	 * @return collection of named channels
+	 */
+	Collection<ByteBuffer> getChannels();
+
+	/**
+	 * Returns the channel patters for this subscription.
+	 *
+	 * @return collection of channel patterns
+	 */
+	Collection<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/>
+	 * 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.
+	 */
+	Flux<ChannelMessage<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();
+
+	/**
+	 * 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.
+	 * @author Mark Paluch
+	 * @since 2.1
+	 */
+	@EqualsAndHashCode
+	class ChannelMessage<C, B> {
+
+		private final C channel;
+		private final B 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) {
+			this.channel = channel;
+			this.message = message;
+		}
+
+		public C getChannel() {
+			return channel;
+		}
+
+		public B getMessage() {
+			return message;
+		}
+	}
+
+	/**
+	 * Value object for a Redis channel message received from a pattern subscription.
+	 * 
+	 * @param <C> 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.
+	 * @author Mark Paluch
+	 * @since 2.1
+	 */
+	@EqualsAndHashCode(callSuper = true)
+	class PatternMessage<P, C, B> extends ChannelMessage<C, B> {
+
+		private final P pattern;
+
+		/**
+		 * Create a new {@link PatternMessage}.
+		 * 
+		 * @param pattern must not be {@literal null}.
+		 * @param channel must not be {@literal null}.
+		 * @param message must not be {@literal null}.
+		 */
+		public PatternMessage(P pattern, C channel, B message) {
+
+			super(channel, message);
+			this.pattern = pattern;
+		}
+
+		public P getPattern() {
+			return pattern;
+		}
+	}
+}

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

@@ -0,0 +1,93 @@
+/*
+ * Copyright 2017 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
+ *
+ *      http://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.connection.lettuce;
+
+import io.lettuce.core.pubsub.api.reactive.RedisPubSubReactiveCommands;
+import lombok.NonNull;
+import lombok.RequiredArgsConstructor;
+import reactor.core.publisher.Flux;
+import reactor.core.publisher.Mono;
+
+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.ReactiveSubscription;
+import org.springframework.data.redis.connection.ReactiveSubscription.ChannelMessage;
+import org.springframework.util.Assert;
+
+/**
+ * @author Mark Paluch
+ * @since 2.1
+ */
+@RequiredArgsConstructor
+class LettuceReactivePubSubCommands implements ReactiveRedisPubSubCommands {
+
+	private final @NonNull LettuceReactiveRedisConnection connection;
+
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.redis.connection.ReactiveRedisPubSubCommands#createSubscription()
+	 */
+	@Override
+	public Mono<ReactiveSubscription> createSubscription() {
+		return connection.getPubSubConnection()
+				.map(c -> new LettuceReactiveSubscription(c.reactive(), connection.translateException()));
+	}
+
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.redis.connection.ReactiveRedisPubSubCommands#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())));
+	}
+
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.redis.connection.ReactiveRedisPubSubCommands#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));
+	}
+
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.redis.connection.ReactiveRedisPubSubCommands#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));
+	}
+
+	private <T> Mono<T> doWithPubSub(Function<RedisPubSubReactiveCommands<ByteBuffer, ByteBuffer>, Mono<T>> function) {
+		return connection.getPubSubConnection().flatMap(c -> function.apply(c.reactive()))
+				.onErrorMap(connection.translateException());
+	}
+}

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

@@ -21,6 +21,7 @@ import io.lettuce.core.api.reactive.BaseRedisReactiveCommands;
 import io.lettuce.core.cluster.api.StatefulRedisClusterConnection;
 import io.lettuce.core.cluster.api.reactive.RedisClusterReactiveCommands;
 import io.lettuce.core.codec.RedisCodec;
+import io.lettuce.core.pubsub.StatefulRedisPubSubConnection;
 import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
 import reactor.core.scheduler.Schedulers;
@@ -46,7 +47,8 @@ class LettuceReactiveRedisConnection implements ReactiveRedisConnection {
 
 	static final RedisCodec<ByteBuffer, ByteBuffer> CODEC = ByteBufferCodec.INSTANCE;
 
-	private final AsyncConnect dedicatedConnection;
+	private final AsyncConnect<StatefulConnection<ByteBuffer, ByteBuffer>> dedicatedConnection;
+	private final AsyncConnect<StatefulRedisPubSubConnection<ByteBuffer, ByteBuffer>> pubSubConnection;
 
 	private @Nullable Mono<StatefulConnection<ByteBuffer, ByteBuffer>> sharedConnection;
 
@@ -62,7 +64,8 @@ class LettuceReactiveRedisConnection implements ReactiveRedisConnection {
 
 		Assert.notNull(connectionProvider, "LettuceConnectionProvider must not be null!");
 
-		this.dedicatedConnection = new AsyncConnect(connectionProvider);
+		this.dedicatedConnection = new AsyncConnect(connectionProvider, StatefulConnection.class);
+		this.pubSubConnection = new AsyncConnect(connectionProvider, StatefulRedisPubSubConnection.class);
 	}
 
 	/**
@@ -81,7 +84,8 @@ class LettuceReactiveRedisConnection implements ReactiveRedisConnection {
 		Assert.notNull(sharedConnection, "Shared StatefulConnection must not be null!");
 		Assert.notNull(connectionProvider, "LettuceConnectionProvider must not be null!");
 
-		this.dedicatedConnection = new AsyncConnect(connectionProvider);
+		this.dedicatedConnection = new AsyncConnect(connectionProvider, StatefulConnection.class);
+		this.pubSubConnection = new AsyncConnect(connectionProvider, StatefulRedisPubSubConnection.class);
 		this.sharedConnection = Mono.just(sharedConnection);
 	}
 
@@ -166,6 +170,15 @@ class LettuceReactiveRedisConnection implements ReactiveRedisConnection {
 		return new LettuceReactiveHyperLogLogCommands(this);
 	}
 
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.redis.connection.ReactiveRedisConnection#pubSubCommands()
+	 */
+	@Override
+	public ReactiveRedisPubSubCommands pubSubCommands() {
+		return new LettuceReactivePubSubCommands(this);
+	}
+
 	/*
 	 * (non-Javadoc)
 	 * @see org.springframework.data.redis.connection.ReactiveRedisConnection#scriptingCommands()
@@ -232,6 +245,10 @@ class LettuceReactiveRedisConnection implements ReactiveRedisConnection {
 		return dedicatedConnection.getConnection().onErrorMap(translateException());
 	}
 
+	protected Mono<StatefulRedisPubSubConnection<ByteBuffer, ByteBuffer>> getPubSubConnection() {
+		return pubSubConnection.getConnection().onErrorMap(translateException());
+	}
+
 	protected Mono<? extends RedisClusterReactiveCommands<ByteBuffer, ByteBuffer>> getCommands() {
 
 		if (sharedConnection != null) {
@@ -320,24 +337,22 @@ class LettuceReactiveRedisConnection implements ReactiveRedisConnection {
 	 * @author Christoph Strobl
 	 * @since 2.0.1
 	 */
-	static class AsyncConnect {
+	static class AsyncConnect<T extends StatefulConnection<?, ?>> {
 
-		private final Mono<StatefulConnection<ByteBuffer, ByteBuffer>> connectionPublisher;
+		private final Mono<T> connectionPublisher;
 		private final LettuceConnectionProvider connectionProvider;
 
 		private AtomicReference<State> state = new AtomicReference<>(State.INITIAL);
-		private volatile @Nullable CompletableFuture<StatefulConnection<ByteBuffer, ByteBuffer>> connection;
+		private volatile @Nullable CompletableFuture<T> connection;
 
 		@SuppressWarnings("unchecked")
-		AsyncConnect(LettuceConnectionProvider connectionProvider) {
+		AsyncConnect(LettuceConnectionProvider connectionProvider, Class<? extends T> connectionType) {
 
 			Assert.notNull(connectionProvider, "LettuceConnectionProvider must not be null!");
 
 			this.connectionProvider = connectionProvider;
 
-			Mono<StatefulConnection<ByteBuffer, ByteBuffer>> defer = Mono
-					.defer(() -> Mono.<StatefulConnection<ByteBuffer, ByteBuffer>> just(
-							connectionProvider.getConnection(StatefulConnection.class)));
+			Mono<T> defer = Mono.defer(() -> Mono.<T> just(connectionProvider.getConnection(connectionType)));
 
 			this.connectionPublisher = defer.subscribeOn(Schedulers.elastic());
 		}
@@ -348,13 +363,13 @@ class LettuceReactiveRedisConnection implements ReactiveRedisConnection {
 		 *
 		 * @return never {@literal null}.
 		 */
-		Mono<StatefulConnection<ByteBuffer, ByteBuffer>> getConnection() {
+		Mono<T> getConnection() {
 
 			if (state.get() == State.CLOSED) {
 				throw new IllegalStateException("Unable to connect. Connection is closed!");
 			}
 
-			CompletableFuture<StatefulConnection<ByteBuffer, ByteBuffer>> connection = this.connection;
+			CompletableFuture<T> connection = this.connection;
 
 			if (connection != null) {
 				return Mono.fromCompletionStage(connection);

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

@@ -0,0 +1,289 @@
+/*
+ * Copyright 2017 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
+ *
+ *      http://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.connection.lettuce;
+
+import io.lettuce.core.pubsub.api.reactive.RedisPubSubReactiveCommands;
+import lombok.RequiredArgsConstructor;
+import reactor.core.Disposable;
+import reactor.core.publisher.ConnectableFlux;
+import reactor.core.publisher.Flux;
+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;
+import java.util.concurrent.ConcurrentSkipListSet;
+import java.util.concurrent.atomic.AtomicLong;
+import java.util.concurrent.atomic.AtomicReference;
+import java.util.function.Function;
+import java.util.function.Supplier;
+
+import org.springframework.data.redis.connection.ReactiveSubscription;
+import org.springframework.lang.Nullable;
+import org.springframework.util.Assert;
+
+/**
+ * Lettuce-specific implementation of {@link ReactiveSubscription}.
+ *
+ * @author Mark Paluch
+ * @since 2.1
+ */
+class LettuceReactiveSubscription implements ReactiveSubscription {
+
+	private final RedisPubSubReactiveCommands<ByteBuffer, ByteBuffer> commands;
+
+	private final State patternState;
+	private final State channelState;
+
+	LettuceReactiveSubscription(RedisPubSubReactiveCommands<ByteBuffer, ByteBuffer> commands,
+			Function<Throwable, Throwable> exceptionTranslator) {
+
+		this.commands = commands;
+		this.patternState = new State(exceptionTranslator);
+		this.channelState = new State(exceptionTranslator);
+	}
+
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.redis.connection.ReactiveSubscription#subscribe(java.nio.ByteBuffer[])
+	 */
+	@Override
+	public Mono<Void> subscribe(ByteBuffer... channels) {
+
+		Assert.notNull(channels, "Channels must not be null!");
+		Assert.noNullElements(channels, "Channels must not contain null elements!");
+
+		return channelState.subscribe(channels, commands::subscribe);
+	}
+
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.redis.connection.ReactiveSubscription#pSubscribe(java.nio.ByteBuffer[])
+	 */
+	@Override
+	public Mono<Void> pSubscribe(ByteBuffer... patterns) {
+
+		Assert.notNull(patterns, "Patterns must not be null!");
+		Assert.noNullElements(patterns, "Patterns must not contain null elements!");
+
+		return patternState.subscribe(patterns, commands::psubscribe);
+	}
+
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.redis.connection.ReactiveSubscription#unsubscribe()
+	 */
+	@Override
+	public Mono<Void> unsubscribe() {
+		return unsubscribe(channelState.getTargets().toArray(new ByteBuffer[0]));
+	}
+
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.redis.connection.ReactiveSubscription#unsubscribe(java.nio.ByteBuffer[])
+	 */
+	@Override
+	public Mono<Void> unsubscribe(ByteBuffer... channels) {
+
+		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);
+	}
+
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.redis.connection.ReactiveSubscription#pUnsubscribe()
+	 */
+	@Override
+	public Mono<Void> pUnsubscribe() {
+		return pUnsubscribe(patternState.getTargets().toArray(new ByteBuffer[0]));
+	}
+
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.redis.connection.ReactiveSubscription#pUnsubscribe(java.nio.ByteBuffer[])
+	 */
+	@Override
+	public Mono<Void> pUnsubscribe(ByteBuffer... patterns) {
+
+		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);
+	}
+
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.redis.connection.ReactiveSubscription#getChannels()
+	 */
+	@Override
+	public Collection<ByteBuffer> getChannels() {
+		return Collections.unmodifiableCollection(channelState.getTargets());
+	}
+
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.redis.connection.ReactiveSubscription#getPatterns()
+	 */
+	@Override
+	public Collection<ByteBuffer> getPatterns() {
+		return Collections.unmodifiableCollection(patternState.getTargets());
+	}
+
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.redis.connection.ReactiveSubscription#receive()
+	 */
+	@Override
+	public Flux<ChannelMessage<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<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())));
+
+		return channelMessages.mergeWith(patternMessages);
+	}
+
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.redis.connection.ReactiveSubscription#terminate()
+	 */
+	@Override
+	public Mono<Void> terminate() {
+
+		return unsubscribe().then(pUnsubscribe()).then(Mono.defer(() -> {
+
+			channelState.terminate();
+			patternState.terminate();
+			return Mono.empty();
+		}));
+	}
+
+	/**
+	 * Subscription state holder.
+	 * 
+	 * @author Mark Paluch
+	 */
+	@RequiredArgsConstructor
+	static class State {
+
+		private final Set<ByteBuffer> targets = new ConcurrentSkipListSet<>();
+		private final AtomicLong subscribers = new AtomicLong();
+		private final AtomicReference<Flux<?>> flux = new AtomicReference<>();
+		private final Function<Throwable, Throwable> exceptionTranslator;
+
+		private volatile @Nullable Disposable disposable;
+
+		/**
+		 * Subscribe to {@code targets} using subscribe {@link Function} and register {@code targets} after subscription.
+		 * 
+		 * @param targets
+		 * @param subscribeFunction
+		 * @return
+		 */
+		Mono<Void> subscribe(ByteBuffer[] targets, Function<ByteBuffer[], Mono<Void>> subscribeFunction) {
+
+			return subscribeFunction.apply(targets).doOnSuccess((v) -> {
+				this.targets.addAll(Arrays.asList(targets));
+			}).onErrorMap(exceptionTranslator);
+		}
+
+		/**
+		 * Unsubscribe from to {@code targets} using unsubscribe {@link Function} and register {@code targets} after
+		 * subscription.
+		 * 
+		 * @param targets
+		 * @param unsubscribeFunction
+		 * @return
+		 */
+		Mono<Void> unsubscribe(ByteBuffer[] targets, Function<ByteBuffer[], Mono<Void>> unsubscribeFunction) {
+
+			return Mono.defer(() -> {
+
+				List<ByteBuffer> targetCollection = Arrays.asList(targets);
+
+				return unsubscribeFunction.apply(targets).doOnSuccess((v) -> {
+					this.targets.removeAll(targetCollection);
+				}).onErrorMap(exceptionTranslator);
+			});
+		}
+
+		Collection<ByteBuffer> getTargets() {
+			return targets;
+		}
+
+		/**
+		 * Create a message stream from connect {@link Function}. Multiple calls to this method are lock-free synchronized.
+		 * The first successful caller creates the actual stream. Other concurrent callers that do not pass the
+		 * synchronization use the stream created by the first successful caller.
+		 * <p/>
+		 * The stream registers a disposal function upon subscription for external {@link #terminate() termination}.
+		 *
+		 * @param connectFunction
+		 * @param <T> message type.
+		 * @return
+		 */
+		@SuppressWarnings("unchecked")
+		<T> Flux<T> receive(Supplier<Flux<T>> connectFunction) {
+
+			Flux<?> fastPath = flux.get();
+
+			if (fastPath != null) {
+				return (Flux) fastPath;
+			}
+
+			ConnectableFlux<T> connectableFlux = connectFunction.get().onErrorMap(exceptionTranslator).publish();
+			Flux<T> fluxToUse = connectableFlux.doOnSubscribe(s -> {
+
+				if (subscribers.incrementAndGet() == 1) {
+					disposable = connectableFlux.connect();
+				}
+			}).doFinally(s -> {
+
+				if (subscribers.decrementAndGet() == 0) {
+
+					this.flux.compareAndSet(connectableFlux, null);
+					terminate();
+				}
+			});
+
+			if (this.flux.compareAndSet(null, fluxToUse)) {
+				return fluxToUse;
+			}
+
+			return (Flux) this.flux.get();
+		}
+
+		void terminate() {
+
+			this.flux.set(null);
+
+			Disposable disposable = this.disposable;
+
+			if (disposable != null && !disposable.isDisposed()) {
+				disposable.dispose();
+			}
+		}
+	}
+}

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

@@ -56,6 +56,17 @@ public interface ReactiveRedisOperations<K, V> {
 	 */
 	<T> Flux<T> execute(ReactiveRedisCallback<T> action);
 
+	/**
+	 * 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.
+	 * @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);
+
 	// -------------------------------------------------------------------------
 	// Methods dealing with Redis Keys
 	// -------------------------------------------------------------------------

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

@@ -193,6 +193,16 @@ public class ReactiveRedisTemplate<K, V> implements ReactiveRedisOperations<K, V
 		return Flux.from(postProcessResult(result, connToUse, false)).doFinally(signal -> conn.close());
 	}
 
+	@Override
+	public Mono<Long> convertAndSend(String destination, V message) {
+
+		Assert.hasText(destination, "Destination channel must not be empty!");
+
+		return createMono(connection -> connection.pubSubCommands().publish(
+				getSerializationContext().getStringSerializationPair().write(destination),
+				getSerializationContext().getValueSerializationPair().write(message)));
+	}
+
 	// -------------------------------------------------------------------------
 	// Methods dealing with Redis keys
 	// -------------------------------------------------------------------------

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

@@ -0,0 +1,354 @@
+/*
+ * Copyright 2017 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
+ *
+ *      http://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.listener;
+
+import reactor.core.publisher.Flux;
+import reactor.core.publisher.Mono;
+import reactor.core.publisher.MonoProcessor;
+
+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;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.atomic.AtomicLongFieldUpdater;
+import java.util.stream.Collectors;
+import java.util.stream.StreamSupport;
+
+import org.springframework.beans.factory.DisposableBean;
+import org.springframework.dao.InvalidDataAccessApiUsageException;
+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.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;
+
+/**
+ * Container providing a stream of {@link ChannelMessage} for messages received via Redis Pub/Sub listeners. The stream
+ * is infinite and registers Redis subscriptions. Handles the low level details of listening, converting and message
+ * dispatching.
+ * <p />
+ * Note the container allocates a single connection when it is created and releases the connection on
+ * {@link #destroy()}. Connections are allocated eagerly to not interfere with non-blocking use during application
+ * 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
+ * @since 2.1
+ * @see ReactiveSubscription
+ * @see org.springframework.data.redis.connection.ReactiveRedisPubSubCommands
+ */
+public class ReactiveRedisMessageListenerContainer implements DisposableBean {
+
+	private final SerializationPair<String> stringSerializationPair = SerializationPair
+			.fromSerializer(RedisSerializer.string());
+	private final Map<ReactiveSubscription, Subscribers> subscriptions = new ConcurrentHashMap<>();
+
+	private volatile @Nullable ReactiveRedisConnection connection;
+
+	/**
+	 * Create a new {@link ReactiveRedisMessageListenerContainer} given {@link ReactiveRedisConnectionFactory}.
+	 * 
+	 * @param connectionFactory must not be {@literal null}.
+	 */
+	public ReactiveRedisMessageListenerContainer(ReactiveRedisConnectionFactory connectionFactory) {
+
+		Assert.notNull(connectionFactory, "ReactiveRedisConnectionFactory must not be null!");
+		this.connection = connectionFactory.getReactiveConnection();
+	}
+
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.beans.factory.DisposableBean#destroy()
+	 */
+	@Override
+	public void destroy() {
+
+		ReactiveRedisConnection connection = this.connection;
+
+		if (connection != null) {
+
+			Flux<Void> terminationSignals = null;
+			while (!subscriptions.isEmpty()) {
+
+				Map<ReactiveSubscription, Subscribers> local = new HashMap<>(subscriptions);
+				List<Mono<Void>> monos = local.keySet().stream() //
+						.peek(subscriptions::remove) //
+						.map(ReactiveSubscription::terminate) //
+						.collect(Collectors.toList());
+
+				if (terminationSignals == null) {
+					terminationSignals = Flux.merge(monos);
+				} else {
+					terminationSignals = terminationSignals.mergeWith(Flux.merge(monos));
+				}
+			}
+
+			if (terminationSignals != null) {
+				terminationSignals.blockLast();
+			}
+
+			connection.close();
+			this.connection = null;
+		}
+	}
+
+	/**
+	 * 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;
+	}
+
+	/**
+	 * 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) {
+
+		Assert.notNull(channelTopics, "ChannelTopics must not be null!");
+		Assert.noNullElements(channelTopics, "ChannelTopics must not contain null elements!");
+
+		return receive(Arrays.asList(channelTopics), stringSerializationPair, stringSerializationPair);
+	}
+
+	/**
+	 * Subscribe to one or more {@link PatternTopic}s and receive a stream of {@link PatternMessage}. Messages, pattern,
+	 * 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)
+	 */
+	@SuppressWarnings("unchecked")
+	public Flux<PatternMessage<String, String, String>> receive(PatternTopic... patternTopics) {
+
+		Assert.notNull(patternTopics, "PatternTopic must not be null!");
+		Assert.noNullElements(patternTopics, "PatternTopic must not contain null elements!");
+
+		return receive(Arrays.asList(patternTopics), stringSerializationPair, stringSerializationPair)
+				.map(m -> (PatternMessage<String, String, String>) m);
+	}
+
+	/**
+	 * Subscribe to one or more {@link Topic}s and receive a stream of {@link ChannelMessage} The stream may contain
+	 * {@link PatternMessage} if subscribed to patterns. Messages, and channel names are serialized/deserialized using the
+	 * 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) {
+
+		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();
+
+		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");
+		}
+
+		return doReceive(channelSerializer, messageSerializer, subscription, patterns, channels);
+	}
+
+	private <C, B> Flux<ChannelMessage<C, B>> doReceive(SerializationPair<C> channelSerializer,
+			SerializationPair<B> messageSerializer, Mono<ReactiveSubscription> subscription, ByteBuffer[] patterns,
+			ByteBuffer[] channels) {
+
+		Flux<ChannelMessage<ByteBuffer, ByteBuffer>> messageStream = subscription.flatMapMany(it -> {
+
+			Mono<Void> subscribe = subscribe(patterns, channels, it);
+
+			MonoProcessor<ChannelMessage<ByteBuffer, ByteBuffer>> terminalProcessor = MonoProcessor.create();
+			return it.receive().mergeWith(subscribe.then(Mono.defer(() -> {
+
+				getSubscribers(it).registered();
+
+				return Mono.empty();
+			}))).doOnCancel(() -> {
+
+				Subscribers subscribers = getSubscribers(it);
+				if (subscribers.unregister()) {
+					subscriptions.remove(it);
+					it.unsubscribe().subscribe(v -> terminalProcessor.onComplete(), terminalProcessor::onError);
+				}
+			}).mergeWith(terminalProcessor);
+		});
+
+		return messageStream
+				.map(message -> readMessage(channelSerializer.getReader(), messageSerializer.getReader(), message));
+	}
+
+	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!");
+
+		Mono<Void> subscribe = null;
+
+		if (patterns.length != 0) {
+			subscribe = it.pSubscribe(patterns);
+		}
+
+		if (channels.length != 0) {
+
+			Mono<Void> channelsSubscribe = it.subscribe(channels);
+
+			if (subscribe == null) {
+				subscribe = channelsSubscribe;
+			} else {
+				subscribe = subscribe.and(channelsSubscribe);
+			}
+		}
+
+		return subscribe;
+	}
+
+	private Subscribers getSubscribers(ReactiveSubscription it) {
+		return subscriptions.computeIfAbsent(it, key -> new Subscribers());
+	}
+
+	private ByteBuffer[] getTargets(Iterable<? extends Topic> topics, Class<?> classFilter) {
+
+		return StreamSupport.stream(topics.spliterator(), false) //
+				.filter(classFilter::isInstance) //
+				.map(Topic::getTopic) //
+				.map(stringSerializationPair::write) //
+				.toArray(ByteBuffer[]::new);
+	}
+
+	@SuppressWarnings("unchecked")
+	private <C, B> ChannelMessage<C, B> readMessage(RedisElementReader<C> channelSerializer,
+			RedisElementReader<B> messageSerializer, ChannelMessage<ByteBuffer, ByteBuffer> message) {
+
+		if (message instanceof PatternMessage) {
+
+			PatternMessage<ByteBuffer, ByteBuffer, ByteBuffer> patternMessage = (PatternMessage) message;
+
+			String pattern = read(stringSerializationPair.getReader(), patternMessage.getPattern());
+			C channel = read(channelSerializer, patternMessage.getChannel());
+			B body = read(messageSerializer, patternMessage.getMessage());
+
+			return new PatternMessage<>(pattern, channel, body);
+		}
+
+		C channel = read(channelSerializer, message.getChannel());
+		B body = read(messageSerializer, message.getMessage());
+
+		return new ChannelMessage<>(channel, body);
+	}
+
+	private static <C> C read(RedisElementReader<C> reader, ByteBuffer buffer) {
+
+		try {
+			buffer.mark();
+			return reader.read(buffer);
+		} finally {
+			buffer.reset();
+		}
+	}
+
+	/**
+	 * Object to track subscriber count and to determine the last unsubscribed subscriber.
+	 * 
+	 * @author Mark Paluch
+	 */
+	static class Subscribers {
+
+		private static final AtomicLongFieldUpdater<Subscribers> SUBSCRIBERS = AtomicLongFieldUpdater
+				.newUpdater(Subscribers.class, "subscribers");
+
+		// accessed via SUBSCRIBERS
+		@SuppressWarnings("unused") private volatile long subscribers;
+
+		/**
+		 * Register a subscriber and increment subscriber count.
+		 */
+		void registered() {
+			SUBSCRIBERS.incrementAndGet(this);
+		}
+
+		/**
+		 * @return {@literal true} if at least one subscriber registered via {@link #registered()}.
+		 */
+		boolean hasRegistration() {
+			return SUBSCRIBERS.get(this) > 0;
+		}
+
+		/**
+		 * Unregister a subscriber and decrement subscriber count.
+		 * 
+		 * @return {@literal true} if this was the last unregistered subscriber.
+		 */
+		boolean unregister() {
+
+			long value = SUBSCRIBERS.get(this);
+
+			if (value <= 0) {
+				return false;
+			}
+
+			if (SUBSCRIBERS.compareAndSet(this, value, value - 1) && value == 1) {
+				return true;
+			}
+
+			return false;
+		}
+	}
+}

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

@@ -16,6 +16,8 @@
 package org.springframework.data.redis.util;
 
 import java.nio.ByteBuffer;
+import java.nio.charset.Charset;
+import java.nio.charset.StandardCharsets;
 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.List;
@@ -203,4 +205,31 @@ public final class ByteUtils {
 
 		return -1;
 	}
+
+	/**
+	 * Convert a {@link String} into a {@link ByteBuffer} using {@link java.nio.charset.StandardCharsets#UTF_8}.
+	 *
+	 * @param theString must not be {@literal null}.
+	 * @return
+	 * @since 2.1
+	 */
+	public static ByteBuffer getByteBuffer(String theString) {
+		return getByteBuffer(theString, StandardCharsets.UTF_8);
+	}
+
+	/**
+	 * Convert a {@link String} into a {@link ByteBuffer} using the given {@link Charset}.
+	 *
+	 * @param theString must not be {@literal null}.
+	 * @param charset must not be {@literal null}.
+	 * @return
+	 * @since 2.1
+	 */
+	public static ByteBuffer getByteBuffer(String theString, Charset charset) {
+
+		Assert.notNull(theString, "The String must not be null!");
+		Assert.notNull(charset, "The String must not be null!");
+
+		return charset.encode(theString);
+	}
 }