diff --git a/src/main/java/org/springframework/data/redis/connection/ReactiveRedisConnection.java b/src/main/java/org/springframework/data/redis/connection/ReactiveRedisConnection.java
index 3ca1c3672..b5d0f017c 100644
--- a/src/main/java/org/springframework/data/redis/connection/ReactiveRedisConnection.java
+++ b/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}.
 	 *
diff --git a/src/main/java/org/springframework/data/redis/connection/ReactiveRedisPubSubCommands.java b/src/main/java/org/springframework/data/redis/connection/ReactiveRedisPubSubCommands.java
new file mode 100644
index 000000000..7bdd7817c
--- /dev/null
+++ b/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);
+}
diff --git a/src/main/java/org/springframework/data/redis/connection/ReactiveSubscription.java b/src/main/java/org/springframework/data/redis/connection/ReactiveSubscription.java
new file mode 100644
index 000000000..855d0facd
--- /dev/null
+++ b/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;
+		}
+	}
+}
diff --git a/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceReactivePubSubCommands.java b/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceReactivePubSubCommands.java
new file mode 100644
index 000000000..4cab537f5
--- /dev/null
+++ b/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());
+	}
+}
diff --git a/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceReactiveRedisConnection.java b/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceReactiveRedisConnection.java
index d01986e8e..15426997e 100644
--- a/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceReactiveRedisConnection.java
+++ b/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);
diff --git a/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceReactiveSubscription.java b/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceReactiveSubscription.java
new file mode 100644
index 000000000..87c117f01
--- /dev/null
+++ b/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();
+			}
+		}
+	}
+}
diff --git a/src/main/java/org/springframework/data/redis/core/ReactiveRedisOperations.java b/src/main/java/org/springframework/data/redis/core/ReactiveRedisOperations.java
index e92ad2de5..4fdefbd6a 100644
--- a/src/main/java/org/springframework/data/redis/core/ReactiveRedisOperations.java
+++ b/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
 	// -------------------------------------------------------------------------
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 d7bc2b308..12b1750c2 100644
--- a/src/main/java/org/springframework/data/redis/core/ReactiveRedisTemplate.java
+++ b/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
 	// -------------------------------------------------------------------------
diff --git a/src/main/java/org/springframework/data/redis/listener/ReactiveRedisMessageListenerContainer.java b/src/main/java/org/springframework/data/redis/listener/ReactiveRedisMessageListenerContainer.java
new file mode 100644
index 000000000..b49d0260b
--- /dev/null
+++ b/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;
+		}
+	}
+}
diff --git a/src/main/java/org/springframework/data/redis/util/ByteUtils.java b/src/main/java/org/springframework/data/redis/util/ByteUtils.java
index 7b1783b54..2e24adb96 100644
--- a/src/main/java/org/springframework/data/redis/util/ByteUtils.java
+++ b/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);
+	}
 }
diff --git a/src/test/java/org/springframework/data/redis/connection/lettuce/LettuceReactiveSubscriptionUnitTests.java b/src/test/java/org/springframework/data/redis/connection/lettuce/LettuceReactiveSubscriptionUnitTests.java
new file mode 100644
index 000000000..7b7fe89b6
--- /dev/null
+++ b/src/test/java/org/springframework/data/redis/connection/lettuce/LettuceReactiveSubscriptionUnitTests.java
@@ -0,0 +1,233 @@
+/*
+ * 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 static org.assertj.core.api.Assertions.*;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.*;
+import static org.springframework.data.redis.util.ByteUtils.*;
+
+import io.lettuce.core.RedisConnectionException;
+import io.lettuce.core.pubsub.api.reactive.RedisPubSubReactiveCommands;
+import reactor.core.Disposable;
+import reactor.core.publisher.DirectProcessor;
+import reactor.core.publisher.Flux;
+import reactor.core.publisher.Mono;
+import reactor.test.StepVerifier;
+
+import java.nio.ByteBuffer;
+import java.util.concurrent.CancellationException;
+
+import org.junit.Before;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.mockito.Mock;
+import org.mockito.junit.MockitoJUnitRunner;
+import org.springframework.data.redis.RedisSystemException;
+import org.springframework.data.redis.connection.ReactiveSubscription.ChannelMessage;
+import org.springframework.data.redis.connection.ReactiveSubscription.PatternMessage;
+
+/**
+ * Unit tests for {@link LettuceReactiveSubscription}.
+ * 
+ * @author Mark Paluch
+ */
+@RunWith(MockitoJUnitRunner.class)
+public class LettuceReactiveSubscriptionUnitTests {
+
+	LettuceReactiveSubscription subscription;
+
+	@Mock RedisPubSubReactiveCommands<ByteBuffer, ByteBuffer> commandsMock;
+
+	@Before
+	public void before() {
+		subscription = new LettuceReactiveSubscription(commandsMock, e -> new RedisSystemException(e.getMessage(), e));
+	}
+
+	@Test // DATAREDIS-612
+	public void shouldSubscribeChannels() {
+
+		when(commandsMock.subscribe(any())).thenReturn(Mono.empty());
+
+		Mono<Void> subscribe = subscription.subscribe(getByteBuffer("foo"), getByteBuffer("bar"));
+
+		assertThat(subscription.getChannels()).isEmpty();
+
+		StepVerifier.create(subscribe).verifyComplete();
+
+		assertThat(subscription.getChannels()).containsOnly(getByteBuffer("foo"), getByteBuffer("bar"));
+		assertThat(subscription.getPatterns()).isEmpty();
+	}
+
+	@Test // DATAREDIS-612
+	public void shouldSubscribeChannelsShouldFail() {
+
+		when(commandsMock.subscribe(any())).thenReturn(Mono.error(new RedisConnectionException("Foo")));
+
+		Mono<Void> subscribe = subscription.subscribe(getByteBuffer("foo"), getByteBuffer("bar"));
+
+		StepVerifier.create(subscribe).expectError(RedisSystemException.class).verify();
+	}
+
+	@Test // DATAREDIS-612
+	public void shouldSubscribePatterns() {
+
+		when(commandsMock.psubscribe(any())).thenReturn(Mono.empty());
+
+		Mono<Void> subscribe = subscription.pSubscribe(getByteBuffer("foo"), getByteBuffer("bar"));
+
+		assertThat(subscription.getPatterns()).isEmpty();
+
+		StepVerifier.create(subscribe).verifyComplete();
+
+		assertThat(subscription.getPatterns()).containsOnly(getByteBuffer("foo"), getByteBuffer("bar"));
+		assertThat(subscription.getChannels()).isEmpty();
+	}
+
+	@Test // DATAREDIS-612
+	public void shouldUnsubscribeChannels() {
+
+		when(commandsMock.subscribe(any())).thenReturn(Mono.empty());
+		when(commandsMock.unsubscribe(any())).thenReturn(Mono.empty());
+		StepVerifier.create(subscription.subscribe(getByteBuffer("foo"), getByteBuffer("bar"))).verifyComplete();
+
+		StepVerifier.create(subscription.unsubscribe()).verifyComplete();
+
+		assertThat(subscription.getChannels()).isEmpty();
+		verify(commandsMock).unsubscribe(any());
+	}
+
+	@Test // DATAREDIS-612
+	public void shouldUnsubscribePatterns() {
+
+		when(commandsMock.psubscribe(any())).thenReturn(Mono.empty());
+		when(commandsMock.punsubscribe(any())).thenReturn(Mono.empty());
+		StepVerifier.create(subscription.pSubscribe(getByteBuffer("foo"), getByteBuffer("bar"))).verifyComplete();
+
+		StepVerifier.create(subscription.pUnsubscribe()).verifyComplete();
+
+		assertThat(subscription.getPatterns()).isEmpty();
+		verify(commandsMock).punsubscribe(any());
+	}
+
+	@Test // DATAREDIS-612
+	public void shouldEmitChannelMessage() {
+
+		when(commandsMock.subscribe(any())).thenReturn(Mono.empty());
+		StepVerifier.create(subscription.subscribe(getByteBuffer("foo"), getByteBuffer("bar"))).verifyComplete();
+
+		DirectProcessor<io.lettuce.core.pubsub.api.reactive.ChannelMessage<ByteBuffer, ByteBuffer>> emitter = DirectProcessor
+				.create();
+		when(commandsMock.observeChannels()).thenReturn(emitter);
+		when(commandsMock.observePatterns()).thenReturn(Flux.empty());
+
+		StepVerifier.create(subscription.receive()).then(() -> {
+
+			emitter.onNext(createChannelMessage("other", "body"));
+			emitter.onNext(createChannelMessage("foo", "body"));
+		}).assertNext(msg -> {
+			assertThat(msg.getChannel()).isEqualTo(getByteBuffer("foo"));
+		}).thenCancel().verify();
+	}
+
+	@Test // DATAREDIS-612
+	public void shouldEmitPatternMessage() {
+
+		when(commandsMock.psubscribe(any())).thenReturn(Mono.empty());
+		StepVerifier.create(subscription.pSubscribe(getByteBuffer("foo*"), getByteBuffer("bar*"))).verifyComplete();
+
+		DirectProcessor<io.lettuce.core.pubsub.api.reactive.PatternMessage<ByteBuffer, ByteBuffer>> emitter = DirectProcessor
+				.create();
+		when(commandsMock.observeChannels()).thenReturn(Flux.empty());
+		when(commandsMock.observePatterns()).thenReturn(emitter);
+
+		StepVerifier.create(subscription.receive()).then(() -> {
+
+			emitter.onNext(createPatternMessage("other*", "channel", "body"));
+			emitter.onNext(createPatternMessage("foo*", "foo", "body"));
+		}).assertNext(msg -> {
+
+			assertThat(((PatternMessage) msg).getPattern()).isEqualTo(getByteBuffer("foo*"));
+			assertThat(msg.getChannel()).isEqualTo(getByteBuffer("foo"));
+		}).thenCancel().verify();
+	}
+
+	@Test // DATAREDIS-612
+	public void shouldEmitError() {
+
+		when(commandsMock.subscribe(any())).thenReturn(Mono.empty());
+		StepVerifier.create(subscription.subscribe(getByteBuffer("foo"), getByteBuffer("bar"))).verifyComplete();
+
+		DirectProcessor<io.lettuce.core.pubsub.api.reactive.ChannelMessage<ByteBuffer, ByteBuffer>> emitter = DirectProcessor
+				.create();
+		when(commandsMock.observeChannels()).thenReturn(emitter);
+		when(commandsMock.observePatterns()).thenReturn(Flux.empty());
+
+		StepVerifier.create(subscription.receive()).then(() -> {
+
+			emitter.onError(new RedisConnectionException("foo"));
+		}).expectError(RedisSystemException.class).verify();
+	}
+
+	@Test // DATAREDIS-612
+	public void shouldTerminateActiveSubscriptions() {
+
+		when(commandsMock.psubscribe(any())).thenReturn(Mono.empty());
+		when(commandsMock.punsubscribe(any())).thenReturn(Mono.empty());
+		StepVerifier.create(subscription.pSubscribe(getByteBuffer("foo*"))).verifyComplete();
+
+		when(commandsMock.observeChannels()).thenReturn(Flux.never());
+		when(commandsMock.observePatterns()).thenReturn(Flux.never());
+
+		StepVerifier.create(subscription.receive()).then(() -> {
+			subscription.terminate().subscribe();
+		}).expectError(CancellationException.class).verify();
+
+		assertThat(subscription.getPatterns()).isEmpty();
+	}
+
+	@Test // DATAREDIS-612
+	public void cancelledSubscriptionShouldUnregisterDownstream() {
+
+		DirectProcessor<io.lettuce.core.pubsub.api.reactive.PatternMessage<ByteBuffer, ByteBuffer>> emitter = DirectProcessor
+				.create();
+
+		when(commandsMock.psubscribe(any())).thenReturn(Mono.empty());
+		StepVerifier.create(subscription.pSubscribe(getByteBuffer("foo*"))).verifyComplete();
+
+		when(commandsMock.observeChannels()).thenReturn(Flux.never());
+		when(commandsMock.observePatterns()).thenReturn(emitter);
+
+		Flux<ChannelMessage<ByteBuffer, ByteBuffer>> receive = subscription.receive();
+		Disposable subscribe = receive.subscribe();
+
+		assertThat(emitter.downstreamCount()).isEqualTo(1);
+
+		subscribe.dispose();
+		assertThat(emitter.downstreamCount()).isEqualTo(0);
+	}
+
+	private static io.lettuce.core.pubsub.api.reactive.ChannelMessage<ByteBuffer, ByteBuffer> createChannelMessage(
+			String channel, String body) {
+		return new io.lettuce.core.pubsub.api.reactive.ChannelMessage<>(getByteBuffer(channel), getByteBuffer(body));
+	}
+
+	private static io.lettuce.core.pubsub.api.reactive.PatternMessage<ByteBuffer, ByteBuffer> createPatternMessage(
+			String pattern, String channel, String body) {
+		return new io.lettuce.core.pubsub.api.reactive.PatternMessage<>(getByteBuffer(pattern), getByteBuffer(channel),
+				getByteBuffer(body));
+	}
+}
diff --git a/src/test/java/org/springframework/data/redis/listener/ReactiveOperationsTestParams.java b/src/test/java/org/springframework/data/redis/listener/ReactiveOperationsTestParams.java
new file mode 100644
index 000000000..dae2efeee
--- /dev/null
+++ b/src/test/java/org/springframework/data/redis/listener/ReactiveOperationsTestParams.java
@@ -0,0 +1,103 @@
+/*
+ * 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 static org.springframework.data.redis.connection.ClusterTestVariables.*;
+
+import java.time.Duration;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.List;
+
+import org.junit.runners.model.Statement;
+import org.springframework.data.redis.SettingsUtils;
+import org.springframework.data.redis.connection.RedisClusterConfiguration;
+import org.springframework.data.redis.connection.RedisClusterNode;
+import org.springframework.data.redis.connection.RedisStandaloneConfiguration;
+import org.springframework.data.redis.connection.lettuce.LettuceClientConfiguration;
+import org.springframework.data.redis.connection.lettuce.LettuceConnectionFactory;
+import org.springframework.data.redis.connection.lettuce.LettucePoolingClientConfiguration;
+import org.springframework.data.redis.connection.lettuce.LettuceTestClientResources;
+import org.springframework.data.redis.test.util.RedisClusterRule;
+
+/**
+ * Parameters for testing implementations of {@link ReactiveRedisMessageListenerContainer}
+ *
+ * @author Mark Paluch
+ */
+class ReactiveOperationsTestParams {
+
+	public static Collection<Object[]> testParams() {
+
+		LettuceClientConfiguration clientConfiguration = LettuceClientConfiguration.builder() //
+				.shutdownTimeout(Duration.ZERO) //
+				.clientResources(LettuceTestClientResources.getSharedClientResources()) //
+				.build();
+
+		LettucePoolingClientConfiguration poolingConfiguration = LettucePoolingClientConfiguration.builder() //
+				.shutdownTimeout(Duration.ZERO) //
+				.clientResources(LettuceTestClientResources.getSharedClientResources()) //
+				.build();
+
+		RedisStandaloneConfiguration standaloneConfiguration = new RedisStandaloneConfiguration(SettingsUtils.getHost(),
+				SettingsUtils.getPort());
+
+		LettuceConnectionFactory lettuceConnectionFactory = new LettuceConnectionFactory(standaloneConfiguration,
+				clientConfiguration);
+		lettuceConnectionFactory.afterPropertiesSet();
+
+		LettuceConnectionFactory poolingConnectionFactory = new LettuceConnectionFactory(standaloneConfiguration,
+				poolingConfiguration);
+		poolingConnectionFactory.afterPropertiesSet();
+
+		List<Object[]> list = Arrays.asList(new Object[][] { //
+				{ lettuceConnectionFactory, "Standalone" }, //
+				{ poolingConnectionFactory, "Pooled" }, //
+		});
+
+		if (clusterAvailable()) {
+
+			RedisClusterConfiguration clusterConfiguration = new RedisClusterConfiguration();
+			clusterConfiguration.addClusterNode(new RedisClusterNode(CLUSTER_HOST, MASTER_NODE_1_PORT));
+
+			LettuceConnectionFactory lettuceClusterConnectionFactory = new LettuceConnectionFactory(clusterConfiguration,
+					clientConfiguration);
+			lettuceClusterConnectionFactory.afterPropertiesSet();
+
+			list = new ArrayList<>(list);
+			list.add(new Object[] { lettuceClusterConnectionFactory, "Cluster" });
+		}
+
+		return list;
+	}
+
+	private static boolean clusterAvailable() {
+
+		try {
+			new RedisClusterRule().apply(new Statement() {
+				@Override
+				public void evaluate() {
+
+				}
+			}, null).evaluate();
+		} catch (Throwable throwable) {
+			return false;
+		}
+		return true;
+	}
+
+}
diff --git a/src/test/java/org/springframework/data/redis/listener/ReactiveRedisMessageListenerContainerIntegrationTests.java b/src/test/java/org/springframework/data/redis/listener/ReactiveRedisMessageListenerContainerIntegrationTests.java
new file mode 100644
index 000000000..4c1cf0e23
--- /dev/null
+++ b/src/test/java/org/springframework/data/redis/listener/ReactiveRedisMessageListenerContainerIntegrationTests.java
@@ -0,0 +1,172 @@
+/*
+ * 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 static org.assertj.core.api.Assertions.*;
+
+import reactor.core.Disposable;
+import reactor.test.StepVerifier;
+
+import java.util.Collection;
+import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.LinkedBlockingDeque;
+import java.util.concurrent.TimeUnit;
+import java.util.function.Supplier;
+
+import org.junit.After;
+import org.junit.AfterClass;
+import org.junit.Before;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.junit.runners.Parameterized;
+import org.junit.runners.Parameterized.Parameters;
+import org.springframework.data.redis.ConnectionFactoryTracker;
+import org.springframework.data.redis.connection.ReactiveSubscription;
+import org.springframework.data.redis.connection.ReactiveSubscription.PatternMessage;
+import org.springframework.data.redis.connection.RedisConnection;
+import org.springframework.data.redis.connection.lettuce.LettuceConnectionFactory;
+import org.springframework.data.redis.core.ReactiveRedisTemplate;
+import org.springframework.data.redis.serializer.RedisSerializationContext;
+import org.springframework.lang.Nullable;
+
+/**
+ * Integration tests for {@link ReactiveRedisMessageListenerContainer} via Lettuce.
+ * 
+ * @author Mark Paluch
+ */
+@RunWith(Parameterized.class)
+public class ReactiveRedisMessageListenerContainerIntegrationTests {
+
+	static final String CHANNEL1 = "my-channel";
+	static final String PATTERN1 = "my-chan*";
+	public static final String MESSAGE = "hello world";
+
+	private final LettuceConnectionFactory connectionFactory;
+	private @Nullable RedisConnection connection;
+
+	@Parameters(name = "{1}")
+	public static Collection<Object[]> testParams() {
+		return ReactiveOperationsTestParams.testParams();
+	}
+
+	@AfterClass
+	public static void cleanUp() {
+		ConnectionFactoryTracker.cleanUp();
+	}
+
+	/**
+	 * @param connectionFactory
+	 * @param label parameterized test label, no further use besides that.
+	 */
+	public ReactiveRedisMessageListenerContainerIntegrationTests(LettuceConnectionFactory connectionFactory,
+			String label) {
+
+		this.connectionFactory = connectionFactory;
+		ConnectionFactoryTracker.add(connectionFactory);
+	}
+
+	@Before
+	public void before() {
+		connection = connectionFactory.getConnection();
+	}
+
+	@After
+	public void tearDown() {
+
+		if (connection != null) {
+			connection.close();
+		}
+	}
+
+	@Test // DATAREDIS-612
+	public void shouldReceiveChannelMessages() {
+
+		ReactiveRedisMessageListenerContainer container = new ReactiveRedisMessageListenerContainer(connectionFactory);
+
+		StepVerifier.create(container.receive(new ChannelTopic(CHANNEL1))) //
+				.then(awaitSubscription(container::getActiveSubscriptions))
+				.then(() -> connection.publish(CHANNEL1.getBytes(), MESSAGE.getBytes())) //
+				.assertNext(c -> {
+
+					assertThat(c.getChannel()).isEqualTo(CHANNEL1);
+					assertThat(c.getMessage()).isEqualTo(MESSAGE);
+				}) //
+				.thenCancel().verify();
+
+		container.destroy();
+	}
+
+	@Test // DATAREDIS-612
+	public void shouldReceivePatternMessages() {
+
+		ReactiveRedisMessageListenerContainer container = new ReactiveRedisMessageListenerContainer(connectionFactory);
+
+		StepVerifier.create(container.receive(new PatternTopic(PATTERN1))) //
+				.then(awaitSubscription(container::getActiveSubscriptions))
+				.then(() -> connection.publish(CHANNEL1.getBytes(), MESSAGE.getBytes())) //
+				.assertNext(c -> {
+
+					assertThat(c.getPattern()).isEqualTo(PATTERN1);
+					assertThat(c.getChannel()).isEqualTo(CHANNEL1);
+					assertThat(c.getMessage()).isEqualTo(MESSAGE);
+				}) //
+				.thenCancel().verify();
+
+		container.destroy();
+	}
+
+	@Test // DATAREDIS-612
+	public void shouldPublishAndReceiveMessage() throws InterruptedException {
+
+		ReactiveRedisMessageListenerContainer container = new ReactiveRedisMessageListenerContainer(connectionFactory);
+		ReactiveRedisTemplate<String, String> template = new ReactiveRedisTemplate<>(connectionFactory,
+				RedisSerializationContext.string());
+
+		BlockingQueue<PatternMessage<String, String, String>> messages = new LinkedBlockingDeque<>();
+		Disposable subscription = container.receive(new PatternTopic(PATTERN1)).doOnNext(messages::add).subscribe();
+
+		StepVerifier.create(template.convertAndSend(CHANNEL1, MESSAGE), 0) //
+				.then(awaitSubscription(container::getActiveSubscriptions)) //
+				.thenRequest(1).expectNextCount(1) //
+				.verifyComplete();
+
+		PatternMessage<String, String, String> message = messages.poll(1, TimeUnit.SECONDS);
+
+		assertThat(message).isNotNull();
+		assertThat(message.getPattern()).isEqualTo(PATTERN1);
+		assertThat(message.getChannel()).isEqualTo(CHANNEL1);
+		assertThat(message.getMessage()).isEqualTo(MESSAGE);
+
+		subscription.dispose();
+		container.destroy();
+	}
+
+	private static Runnable awaitSubscription(Supplier<Collection<ReactiveSubscription>> activeSubscriptions) {
+
+		return () -> {
+
+			try {
+
+				while (activeSubscriptions.get().isEmpty()) {
+					Thread.sleep(10);
+				}
+
+			} catch (InterruptedException e) {
+				return;
+			}
+		};
+	}
+}
diff --git a/src/test/java/org/springframework/data/redis/listener/ReactiveRedisMessageListenerContainerUnitTests.java b/src/test/java/org/springframework/data/redis/listener/ReactiveRedisMessageListenerContainerUnitTests.java
new file mode 100644
index 000000000..a6e51676f
--- /dev/null
+++ b/src/test/java/org/springframework/data/redis/listener/ReactiveRedisMessageListenerContainerUnitTests.java
@@ -0,0 +1,270 @@
+/*
+ * 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 static org.assertj.core.api.Assertions.*;
+import static org.mockito.Mockito.*;
+import static org.springframework.data.redis.util.ByteUtils.*;
+
+import reactor.core.Disposable;
+import reactor.core.publisher.DirectProcessor;
+import reactor.core.publisher.Flux;
+import reactor.core.publisher.Mono;
+import reactor.core.publisher.MonoProcessor;
+import reactor.test.StepVerifier;
+
+import java.nio.ByteBuffer;
+import java.util.concurrent.CancellationException;
+
+import org.junit.Before;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.mockito.Mock;
+import org.mockito.junit.MockitoJUnitRunner;
+import org.springframework.data.redis.connection.ReactiveRedisConnection;
+import org.springframework.data.redis.connection.ReactiveRedisConnectionFactory;
+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.data.redis.connection.ReactiveSubscription.PatternMessage;
+
+/**
+ * Unit tests for {@link ReactiveRedisMessageListenerContainer}.
+ *
+ * @author Mark Paluch
+ */
+@RunWith(MockitoJUnitRunner.class)
+public class ReactiveRedisMessageListenerContainerUnitTests {
+
+	ReactiveRedisMessageListenerContainer container;
+
+	@Mock ReactiveRedisConnectionFactory connectionFactoryMock;
+	@Mock ReactiveRedisConnection connectionMock;
+	@Mock ReactiveRedisPubSubCommands commandsMock;
+	@Mock ReactiveSubscription subscriptionMock;
+
+	@Before
+	public void before() {
+
+		when(connectionFactoryMock.getReactiveConnection()).thenReturn(connectionMock);
+		when(connectionMock.pubSubCommands()).thenReturn(commandsMock);
+		when(commandsMock.createSubscription()).thenReturn(Mono.just(subscriptionMock));
+		when(subscriptionMock.subscribe(any())).thenReturn(Mono.empty());
+		when(subscriptionMock.pSubscribe(any())).thenReturn(Mono.empty());
+		when(subscriptionMock.unsubscribe()).thenReturn(Mono.empty());
+	}
+
+	@Test // DATAREDIS-612
+	public void shouldSubscribeToPattern() {
+
+		when(subscriptionMock.receive()).thenReturn(Flux.never());
+
+		container = createContainer();
+
+		StepVerifier.create(container.receive(new PatternTopic("foo*"))).thenAwait().thenCancel().verify();
+
+		verify(subscriptionMock).pSubscribe(getByteBuffer("foo*"));
+	}
+
+	@Test // DATAREDIS-612
+	public void shouldSubscribeToMultiplePatterns() {
+
+		when(subscriptionMock.receive()).thenReturn(Flux.never());
+		container = createContainer();
+
+		StepVerifier.create(container.receive(new PatternTopic("foo*"), new PatternTopic("bar*"))).thenRequest(1)
+				.thenAwait().thenCancel().verify();
+
+		verify(subscriptionMock).pSubscribe(getByteBuffer("foo*"), getByteBuffer("bar*"));
+	}
+
+	@Test // DATAREDIS-612
+	public void shouldSubscribeToChannel() {
+
+		when(subscriptionMock.receive()).thenReturn(Flux.never());
+		container = createContainer();
+
+		StepVerifier.create(container.receive(new ChannelTopic("foo"))).thenAwait().thenCancel().verify();
+
+		verify(subscriptionMock).subscribe(getByteBuffer("foo"));
+	}
+
+	@Test // DATAREDIS-612
+	public void shouldSubscribeToMultipleChannels() {
+
+		when(subscriptionMock.receive()).thenReturn(Flux.never());
+		container = createContainer();
+
+		StepVerifier.create(container.receive(new ChannelTopic("foo"), new ChannelTopic("bar"))).thenAwait().thenCancel()
+				.verify();
+
+		verify(subscriptionMock).subscribe(getByteBuffer("foo"), getByteBuffer("bar"));
+	}
+
+	@Test // DATAREDIS-612
+	public void shouldEmitChannelMessage() {
+
+		DirectProcessor<ChannelMessage<ByteBuffer, ByteBuffer>> processor = DirectProcessor.create();
+
+		when(subscriptionMock.receive()).thenReturn(processor);
+		container = createContainer();
+
+		Flux<ChannelMessage<String, String>> messageStream = container.receive(new ChannelTopic("foo"));
+
+		StepVerifier.create(messageStream).then(() -> {
+			processor.onNext(createChannelMessage("foo", "message"));
+		}).assertNext(msg -> {
+
+			assertThat(msg.getChannel()).isEqualTo("foo");
+			assertThat(msg.getMessage()).isEqualTo("message");
+		}).thenCancel().verify();
+	}
+
+	@Test // DATAREDIS-612
+	public void shouldEmitPatternMessage() {
+
+		DirectProcessor<ChannelMessage<ByteBuffer, ByteBuffer>> processor = DirectProcessor.create();
+
+		when(subscriptionMock.receive()).thenReturn(processor);
+		container = createContainer();
+
+		Flux<PatternMessage<String, String, String>> messageStream = container.receive(new PatternTopic("foo*"));
+
+		StepVerifier.create(messageStream).then(() -> {
+			processor.onNext(createPatternMessage("foo*", "foo", "message"));
+		}).assertNext(msg -> {
+
+			assertThat(msg.getPattern()).isEqualTo("foo*");
+			assertThat(msg.getChannel()).isEqualTo("foo");
+			assertThat(msg.getMessage()).isEqualTo("message");
+		}).thenCancel().verify();
+	}
+
+	@Test // DATAREDIS-612
+	public void shouldRegisterSubscription() {
+
+		MonoProcessor<Void> subscribeMono = MonoProcessor.create();
+
+		reset(subscriptionMock);
+		when(subscriptionMock.subscribe(any())).thenReturn(subscribeMono);
+		when(subscriptionMock.unsubscribe()).thenReturn(Mono.empty());
+		when(subscriptionMock.receive()).thenReturn(DirectProcessor.create());
+		container = createContainer();
+
+		Flux<ChannelMessage<String, String>> messageStream = container.receive(new ChannelTopic("foo*"));
+
+		Disposable subscription = messageStream.subscribe();
+
+		assertThat(container.getActiveSubscriptions()).isEmpty();
+		subscribeMono.onComplete();
+		assertThat(container.getActiveSubscriptions()).isNotEmpty();
+		subscription.dispose();
+		assertThat(container.getActiveSubscriptions()).isEmpty();
+	}
+
+	@Test // DATAREDIS-612
+	public void shouldRegisterSubscriptionMultipleSubscribers() {
+
+		reset(subscriptionMock);
+		when(subscriptionMock.subscribe(any())).thenReturn(Mono.empty());
+		when(subscriptionMock.unsubscribe()).thenReturn(Mono.empty());
+		when(subscriptionMock.receive()).thenReturn(DirectProcessor.create());
+		container = createContainer();
+
+		Flux<ChannelMessage<String, String>> messageStream = container.receive(new ChannelTopic("foo*"));
+
+		Disposable first = messageStream.subscribe();
+		Disposable second = messageStream.subscribe();
+
+		first.dispose();
+
+		verify(subscriptionMock, never()).unsubscribe();
+		assertThat(container.getActiveSubscriptions()).isNotEmpty();
+
+		second.dispose();
+
+		verify(subscriptionMock).unsubscribe();
+		assertThat(container.getActiveSubscriptions()).isEmpty();
+	}
+
+	@Test // DATAREDIS-612
+	public void shouldUnsubscribeOnCancel() {
+
+		when(subscriptionMock.receive()).thenReturn(DirectProcessor.create());
+		container = createContainer();
+
+		Flux<PatternMessage<String, String, String>> messageStream = container.receive(new PatternTopic("foo*"));
+
+		StepVerifier.create(messageStream).then(() -> {
+
+			// Then required to trigger cancel.
+
+		}).thenCancel().verify();
+
+		verify(subscriptionMock).unsubscribe();
+	}
+
+	@Test // DATAREDIS-612
+	public void shouldTerminateSubscriptionsOnShutdown() {
+
+		DirectProcessor<ChannelMessage<ByteBuffer, ByteBuffer>> processor = DirectProcessor.create();
+
+		when(subscriptionMock.receive()).thenReturn(processor);
+		when(subscriptionMock.terminate()).thenReturn(Mono.defer(() -> {
+
+			processor.onError(new CancellationException());
+			return Mono.empty();
+		}));
+		container = createContainer();
+
+		Flux<PatternMessage<String, String, String>> messageStream = container.receive(new PatternTopic("foo*"));
+
+		StepVerifier.create(messageStream).then(() -> {
+			container.destroy();
+		}).verifyError(CancellationException.class);
+	}
+
+	@Test // DATAREDIS-612
+	public void shouldCleanupDownstream() {
+
+		DirectProcessor<ChannelMessage<ByteBuffer, ByteBuffer>> processor = DirectProcessor.create();
+
+		when(subscriptionMock.receive()).thenReturn(processor);
+		container = createContainer();
+
+		Flux<PatternMessage<String, String, String>> messageStream = container.receive(new PatternTopic("foo*"));
+
+		StepVerifier.create(messageStream).then(() -> {
+			assertThat(processor.hasDownstreams()).isTrue();
+			processor.onNext(createPatternMessage("foo*", "foo", "message"));
+		}).expectNextCount(1).thenCancel().verify();
+
+		assertThat(processor.hasDownstreams()).isFalse();
+	}
+
+	private ReactiveRedisMessageListenerContainer createContainer() {
+		return new ReactiveRedisMessageListenerContainer(connectionFactoryMock);
+	}
+
+	private static ChannelMessage<ByteBuffer, ByteBuffer> createChannelMessage(String channel, String body) {
+		return new ChannelMessage<>(getByteBuffer(channel), getByteBuffer(body));
+	}
+
+	private static PatternMessage<ByteBuffer, ByteBuffer, ByteBuffer> createPatternMessage(String pattern, String channel,
+			String body) {
+		return new PatternMessage<>(getByteBuffer(pattern), getByteBuffer(channel), getByteBuffer(body));
+	}
+}