DATAREDIS-720 - Reuse shared connection across reactive connection wrappers.

We now reuse a shared Lettuce connection across our reactive connection wrappers if LettuceConnectionFactory is configured to provide a shared connection instance. Reactive connections can either operate with a shared connection or with a connection provider only.

We also now execute blocking Redis commands (BLPOP, BRPOP, BRPOPLPUSH) on dedicated connection.

Original Pull Request: #290

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

@@ -20,12 +20,15 @@ import io.lettuce.core.ClientOptions;
 import io.lettuce.core.RedisClient;
 import io.lettuce.core.RedisException;
 import io.lettuce.core.RedisURI;
+import io.lettuce.core.api.StatefulConnection;
 import io.lettuce.core.api.StatefulRedisConnection;
 import io.lettuce.core.cluster.ClusterClientOptions;
 import io.lettuce.core.cluster.RedisClusterClient;
+import io.lettuce.core.cluster.api.StatefulRedisClusterConnection;
 import io.lettuce.core.codec.RedisCodec;
 import io.lettuce.core.resource.ClientResources;
 
+import java.nio.ByteBuffer;
 import java.time.Duration;
 import java.util.ArrayList;
 import java.util.List;
@@ -88,6 +91,7 @@ public class LettuceConnectionFactory
 	private boolean validateConnection = false;
 	private boolean shareNativeConnection = true;
 	private @Nullable StatefulRedisConnection<byte[], byte[]> connection;
+	private @Nullable StatefulConnection<ByteBuffer, ByteBuffer> reactiveConnection;
 	private @Nullable LettucePool pool;
 	/** Synchronization monitor for the shared Connection */
 	private final Object connectionMonitor = new Object();
@@ -242,6 +246,7 @@ public class LettuceConnectionFactory
 	public void destroy() {
 
 		resetConnection();
+		resetReactiveConnection();
 
 		if (connectionProvider instanceof DisposableBean) {
 			try {
@@ -318,7 +323,10 @@ public class LettuceConnectionFactory
 	 */
 	@Override
 	public LettuceReactiveRedisConnection getReactiveConnection() {
-		return new LettuceReactiveRedisConnection(reactiveConnectionProvider);
+
+		return getShareNativeConnection()
+				? new LettuceReactiveRedisConnection(getSharedReactiveConnection(), reactiveConnectionProvider)
+				: new LettuceReactiveRedisConnection(reactiveConnectionProvider);
 	}
 
 	/*
@@ -327,11 +335,16 @@ public class LettuceConnectionFactory
 	 */
 	@Override
 	public LettuceReactiveRedisClusterConnection getReactiveClusterConnection() {
+
 		if (!isClusterAware()) {
 			throw new InvalidDataAccessApiUsageException("Cluster is not configured!");
 		}
 
-		return new LettuceReactiveRedisClusterConnection(reactiveConnectionProvider);
+		RedisClusterClient client = (RedisClusterClient) this.client;
+
+		return getShareNativeConnection()
+				? new LettuceReactiveRedisClusterConnection(getSharedReactiveConnection(), reactiveConnectionProvider, client)
+				: new LettuceReactiveRedisClusterConnection(reactiveConnectionProvider, client);
 	}
 
 	public void initConnection() {
@@ -344,11 +357,23 @@ public class LettuceConnectionFactory
 		}
 	}
 
+	public void initReactiveConnection() {
+
+		synchronized (this.connectionMonitor) {
+			if (this.reactiveConnection != null) {
+				resetReactiveConnection();
+			}
+			this.reactiveConnection = createReactiveLettuceConnector();
+		}
+	}
+
 	/**
 	 * Reset the underlying shared Connection, to be reinitialized on next access.
 	 */
 	public void resetConnection() {
+
 		synchronized (this.connectionMonitor) {
+
 			if (this.connection != null) {
 				this.connectionProvider.release(this.connection);
 			}
@@ -356,17 +381,49 @@ public class LettuceConnectionFactory
 		}
 	}
 
+	/**
+	 * Reset the underlying shared Connection, to be reinitialized on next access.
+	 *
+	 * @since 2.0.1
+	 */
+	public void resetReactiveConnection() {
+
+		synchronized (this.connectionMonitor) {
+
+			if (this.reactiveConnection != null) {
+				this.reactiveConnectionProvider.release(this.reactiveConnection);
+			}
+			this.reactiveConnection = null;
+		}
+	}
+
 	/**
 	 * Validate the shared Connection and reinitialize if invalid
 	 */
 	public void validateConnection() {
+
+		validateConnection(connection, connectionProvider, this::initConnection);
+
+		validateConnection(reactiveConnection, reactiveConnectionProvider, this::initReactiveConnection);
+	}
+
+	private void validateConnection(@Nullable StatefulConnection<?, ?> connection,
+			LettuceConnectionProvider connectionProvider, Runnable initConnection) {
+
 		synchronized (this.connectionMonitor) {
 
 			boolean valid = false;
 
-			if (connection.isOpen()) {
+			if (connection != null && connection.isOpen()) {
 				try {
-					connection.sync().ping();
+
+					if (connection instanceof StatefulRedisConnection) {
+						((StatefulRedisConnection) connection).sync().ping();
+					}
+
+					if (connection instanceof StatefulRedisClusterConnection) {
+						((StatefulRedisConnection) connection).sync().ping();
+					}
 					valid = true;
 				} catch (Exception e) {
 					log.debug("Validation failed", e);
@@ -375,9 +432,13 @@ public class LettuceConnectionFactory
 
 			if (!valid) {
 
-				connectionProvider.release(connection);
+				if (connection != null) {
+					connectionProvider.release(connection);
+				}
+
 				log.warn("Validation of shared connection failed. Creating a new connection.");
-				initConnection();
+
+				initConnection.run();
 			}
 		}
 	}
@@ -775,7 +836,12 @@ public class LettuceConnectionFactory
 		return clusterConfiguration != null;
 	}
 
+	/**
+	 * @return the shared connection using {@literal byte} array encoding for imperative API use. {@literal null} if
+	 *         {@link #getShareNativeConnection() connection sharing} is disabled.
+	 */
 	protected StatefulRedisConnection<byte[], byte[]> getSharedConnection() {
+
 		if (shareNativeConnection) {
 			synchronized (this.connectionMonitor) {
 				if (this.connection == null) {
@@ -791,6 +857,28 @@ public class LettuceConnectionFactory
 		}
 	}
 
+	/**
+	 * @return the shared connection using {@link ByteBuffer} encoding for reactive API use. {@literal null} if
+	 *         {@link #getShareNativeConnection() connection sharing} is disabled.
+	 * @since 2.0.1
+	 */
+	protected StatefulConnection<ByteBuffer, ByteBuffer> getSharedReactiveConnection() {
+
+		if (shareNativeConnection) {
+			synchronized (this.connectionMonitor) {
+				if (this.reactiveConnection == null) {
+					initReactiveConnection();
+				}
+				if (validateConnection) {
+					validateConnection();
+				}
+				return this.reactiveConnection;
+			}
+		} else {
+			return null;
+		}
+	}
+
 	protected StatefulRedisConnection<byte[], byte[]> createLettuceConnector() {
 		try {
 
@@ -809,6 +897,23 @@ public class LettuceConnectionFactory
 		}
 	}
 
+	protected StatefulConnection<ByteBuffer, ByteBuffer> createReactiveLettuceConnector() {
+
+		try {
+
+			StatefulConnection<ByteBuffer, ByteBuffer> connection;
+			connection = reactiveConnectionProvider.getConnection(StatefulConnection.class);
+
+			if (connection instanceof StatefulRedisConnection && getDatabase() > 0) {
+				((StatefulRedisConnection) connection).sync().select(getDatabase());
+			}
+
+			return connection;
+		} catch (RedisException e) {
+			throw new RedisConnectionFailureException("Unable to connect to Redis on " + getHostName() + ":" + getPort(), e);
+		}
+	}
+
 	private LettuceConnectionProvider createConnectionProvider(AbstractRedisClient client, RedisCodec<?, ?> codec) {
 
 		LettuceConnectionProvider connectionProvider = doConnectionProvider(client, codec);

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

@@ -241,7 +241,7 @@ class LettuceReactiveListCommands implements ReactiveListCommands {
 	@Override
 	public Flux<PopResponse> bPop(Publisher<BPopCommand> commands) {
 
-		return connection.execute(cmd -> Flux.from(commands).concatMap(command -> {
+		return connection.executeDedicated(cmd -> Flux.from(commands).concatMap(command -> {
 
 			Assert.notNull(command.getKeys(), "Keys must not be null!");
 			Assert.notNull(command.getDirection(), "Direction must not be null!");
@@ -281,7 +281,7 @@ class LettuceReactiveListCommands implements ReactiveListCommands {
 	@Override
 	public Flux<ByteBufferResponse<BRPopLPushCommand>> bRPopLPush(Publisher<BRPopLPushCommand> commands) {
 
-		return connection.execute(cmd -> Flux.from(commands).concatMap(command -> {
+		return connection.executeDedicated(cmd -> Flux.from(commands).concatMap(command -> {
 
 			Assert.notNull(command.getKey(), "Key must not be null!");
 			Assert.notNull(command.getDestination(), "Destination key must not be null!");

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

@@ -15,6 +15,7 @@
  */
 package org.springframework.data.redis.connection.lettuce;
 
+import io.lettuce.core.api.StatefulConnection;
 import io.lettuce.core.api.reactive.BaseRedisReactiveCommands;
 import io.lettuce.core.api.reactive.RedisReactiveCommands;
 import io.lettuce.core.cluster.RedisClusterClient;
@@ -45,19 +46,37 @@ class LettuceReactiveRedisClusterConnection extends LettuceReactiveRedisConnecti
 	private final ClusterTopologyProvider topologyProvider;
 
 	/**
-	 * Creates new {@link LettuceReactiveRedisClusterConnection}.
+	 * Creates new {@link LettuceReactiveRedisClusterConnection} given {@link LettuceConnectionProvider} and
+	 * {@link RedisClusterClient}.
 	 *
+	 * @param connectionProvider must not be {@literal null}.
 	 * @param client must not be {@literal null}.
 	 * @throws IllegalArgumentException when {@code client} is {@literal null}.
-	 * @throws org.springframework.dao.InvalidDataAccessResourceUsageException when {@code client} is not suitable for
-	 *           cluster environment.
 	 */
-	LettuceReactiveRedisClusterConnection(LettuceConnectionProvider connectionProvider) {
+	LettuceReactiveRedisClusterConnection(LettuceConnectionProvider connectionProvider, RedisClusterClient client) {
 
 		super(connectionProvider);
 
-		this.topologyProvider = new LettuceClusterTopologyProvider(
-				((ClusterConnectionProvider) connectionProvider).getClient());
+		this.topologyProvider = new LettuceClusterTopologyProvider(client);
+	}
+
+	/**
+	 * Creates new {@link LettuceReactiveRedisClusterConnection} given a shared {@link StatefulConnection connection},
+	 * {@link LettuceConnectionProvider} and {@link RedisClusterClient}.
+	 *
+	 * @param sharedConnection must not be {@literal null}.
+	 * @param connectionProvider must not be {@literal null}.
+	 * @param client must not be {@literal null}.
+	 * @throws IllegalArgumentException when {@code client} is {@literal null}.
+	 * @since 2.0.1
+	 */
+	@SuppressWarnings("unchecked")
+	LettuceReactiveRedisClusterConnection(StatefulConnection<ByteBuffer, ByteBuffer> sharedConnection,
+			LettuceConnectionProvider connectionProvider, RedisClusterClient client) {
+
+		super(sharedConnection, connectionProvider);
+
+		this.topologyProvider = new LettuceClusterTopologyProvider(client);
 	}
 
 	/*
@@ -183,7 +202,7 @@ class LettuceReactiveRedisClusterConnection extends LettuceReactiveRedisConnecti
 			return Flux.error(e);
 		}
 
-		return Flux.defer(() -> callback.doWithCommands(getCommands(node))).onErrorMap(translateException());
+		return getCommands(node).flatMapMany(callback::doWithCommands).onErrorMap(translateException());
 	}
 
 	/*
@@ -192,33 +211,27 @@ class LettuceReactiveRedisClusterConnection extends LettuceReactiveRedisConnecti
 	 */
 	@SuppressWarnings({ "unchecked", "rawtypes" })
 	@Override
-	protected StatefulRedisClusterConnection<ByteBuffer, ByteBuffer> getConnection() {
-
-		Assert.isInstanceOf(StatefulRedisClusterConnection.class, super.getConnection(),
-				"Connection needs to be instance of StatefulRedisClusterConnection");
-
-		return (StatefulRedisClusterConnection) super.getConnection();
+	protected Mono<StatefulRedisClusterConnection<ByteBuffer, ByteBuffer>> getConnection() {
+		return (Mono) super.getConnection();
 	}
 
 	/*
 	 * (non-Javadoc)
 	 * @see org.springframework.data.redis.connection.lettuce.LettuceReactiveRedisConnection#getCommands()
 	 */
-	protected RedisClusterReactiveCommands<ByteBuffer, ByteBuffer> getCommands() {
-		return getConnection().reactive();
+	protected Mono<RedisClusterReactiveCommands<ByteBuffer, ByteBuffer>> getCommands() {
+		return getConnection().map(StatefulRedisClusterConnection::reactive);
 	}
 
 	@SuppressWarnings({ "unchecked", "rawtypes" })
-	protected RedisReactiveCommands<ByteBuffer, ByteBuffer> getCommands(RedisNode node) {
-
-		if (!(getConnection() instanceof StatefulRedisClusterConnection)) {
-			throw new IllegalArgumentException("o.O connection needs to be cluster compatible " + getConnection());
-		}
+	protected Mono<RedisReactiveCommands<ByteBuffer, ByteBuffer>> getCommands(RedisNode node) {
 
 		if (StringUtils.hasText(node.getId())) {
-			return ((StatefulRedisClusterConnection) getConnection()).getConnection(node.getId()).reactive();
+			return getConnection().cast(StatefulRedisClusterConnection.class)
+					.map(it -> it.getConnection(node.getId()).reactive());
 		}
 
-		return ((StatefulRedisClusterConnection) getConnection()).getConnection(node.getHost(), node.getPort()).reactive();
+		return getConnection().cast(StatefulRedisClusterConnection.class)
+				.map(it -> it.getConnection(node.getHost(), node.getPort()).reactive());
 	}
 }

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

@@ -23,8 +23,11 @@ import io.lettuce.core.cluster.api.reactive.RedisClusterReactiveCommands;
 import io.lettuce.core.codec.RedisCodec;
 import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
+import reactor.core.scheduler.Schedulers;
 
 import java.nio.ByteBuffer;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.atomic.AtomicReference;
 import java.util.function.Function;
 
 import org.reactivestreams.Publisher;
@@ -43,9 +46,9 @@ class LettuceReactiveRedisConnection implements ReactiveRedisConnection {
 
 	static final RedisCodec<ByteBuffer, ByteBuffer> CODEC = ByteBufferCodec.INSTANCE;
 
-	private final LettuceConnectionProvider connectionProvider;
+	private final AsyncConnect dedicatedConnection;
 
-	private @Nullable StatefulConnection<ByteBuffer, ByteBuffer> connection;
+	private @Nullable Mono<StatefulConnection<ByteBuffer, ByteBuffer>> sharedConnection;
 
 	/**
 	 * Creates new {@link LettuceReactiveRedisConnection}.
@@ -59,8 +62,27 @@ class LettuceReactiveRedisConnection implements ReactiveRedisConnection {
 
 		Assert.notNull(connectionProvider, "LettuceConnectionProvider must not be null!");
 
-		this.connectionProvider = connectionProvider;
-		this.connection = connectionProvider.getConnection(StatefulConnection.class);
+		this.dedicatedConnection = new AsyncConnect(connectionProvider);
+	}
+
+	/**
+	 * Creates new {@link LettuceReactiveRedisConnection} given a shared {@link StatefulConnection connection}.
+	 *
+	 * @param sharedConnection must not be {@literal null}.
+	 * @param connectionProvider must not be {@literal null}.
+	 * @throws IllegalArgumentException when {@code client} is {@literal null}.
+	 * @throws InvalidDataAccessResourceUsageException when {@code client} is not suitable for connection.
+	 * @since 2.0.1
+	 */
+	@SuppressWarnings("unchecked")
+	LettuceReactiveRedisConnection(StatefulConnection<ByteBuffer, ByteBuffer> sharedConnection,
+			LettuceConnectionProvider connectionProvider) {
+
+		Assert.notNull(sharedConnection, "Shared StatefulConnection must not be null!");
+		Assert.notNull(connectionProvider, "LettuceConnectionProvider must not be null!");
+
+		this.dedicatedConnection = new AsyncConnect(connectionProvider);
+		this.sharedConnection = Mono.just(sharedConnection);
 	}
 
 	/*
@@ -176,33 +198,55 @@ class LettuceReactiveRedisConnection implements ReactiveRedisConnection {
 	 * @return
 	 */
 	public <T> Flux<T> execute(LettuceReactiveCallback<T> callback) {
-		return Flux.defer(() -> callback.doWithCommands(getCommands())).onErrorMap(translateException());
+		return getCommands().flatMapMany(callback::doWithCommands).onErrorMap(translateException());
 	}
 
-	/* (non-Javadoc)
+	/**
+	 * @param callback
+	 * @return
+	 * @since 2.0.1
+	 */
+	public <T> Flux<T> executeDedicated(LettuceReactiveCallback<T> callback) {
+		return getDedicatedCommands().flatMapMany(callback::doWithCommands).onErrorMap(translateException());
+	}
+
+	/*
+	 * (non-Javadoc)
 	 * @see java.io.Closeable#close()
 	 */
 	@Override
 	public void close() {
-
-		if (connection != null) {
-			synchronized (connectionProvider) {
-				connectionProvider.release(connection);
-				connection = null;
-			}
-		}
+		dedicatedConnection.close();
 	}
 
-	protected StatefulConnection<ByteBuffer, ByteBuffer> getConnection() {
+	protected Mono<? extends StatefulConnection<ByteBuffer, ByteBuffer>> getConnection() {
 
-		if (connection != null) {
-			return connection;
+		if (sharedConnection != null) {
+			return sharedConnection;
 		}
 
-		throw new IllegalStateException("Connection is closed");
+		return getDedicatedConnection();
 	}
 
-	protected RedisClusterReactiveCommands<ByteBuffer, ByteBuffer> getCommands() {
+	protected Mono<StatefulConnection<ByteBuffer, ByteBuffer>> getDedicatedConnection() {
+		return dedicatedConnection.getConnection().onErrorMap(translateException());
+	}
+
+	protected Mono<? extends RedisClusterReactiveCommands<ByteBuffer, ByteBuffer>> getCommands() {
+
+		if (sharedConnection != null) {
+			return sharedConnection.map(LettuceReactiveRedisConnection::getRedisClusterReactiveCommands);
+		}
+
+		return getDedicatedCommands();
+	}
+
+	protected Mono<? extends RedisClusterReactiveCommands<ByteBuffer, ByteBuffer>> getDedicatedCommands() {
+		return dedicatedConnection.getConnection().map(LettuceReactiveRedisConnection::getRedisClusterReactiveCommands);
+	}
+
+	private static RedisClusterReactiveCommands<ByteBuffer, ByteBuffer> getRedisClusterReactiveCommands(
+			StatefulConnection<ByteBuffer, ByteBuffer> connection) {
 
 		if (connection instanceof StatefulRedisConnection) {
 			return ((StatefulRedisConnection<ByteBuffer, ByteBuffer>) connection).reactive();
@@ -210,7 +254,7 @@ class LettuceReactiveRedisConnection implements ReactiveRedisConnection {
 			return ((StatefulRedisClusterConnection<ByteBuffer, ByteBuffer>) connection).reactive();
 		}
 
-		throw new RuntimeException("o.O unknown connection type " + connection);
+		throw new IllegalStateException("o.O unknown connection type " + connection);
 	}
 
 	<T> Function<Throwable, Throwable> translateException() {
@@ -260,4 +304,94 @@ class LettuceReactiveRedisConnection implements ReactiveRedisConnection {
 			return value.duplicate();
 		}
 	}
+
+	/**
+	 * Asynchronous connection utility. This utility has a lifecycle controlled by {@link #getConnection()} and
+	 * {@link #close()} methods:
+	 * <ol>
+	 * <li>Initial state: Not connected. Calling {@link #getConnection()} will transition to the next state.</li>
+	 * <li>Connection requested: First caller to {@link #getConnection()} initiates an asynchronous connect. Connection is
+	 * accessible through the resulting {@link Mono}.</li>
+	 * <li>Closing: Call to {@link #close()} initiates connection closing.</li>
+	 * <li>Closed: Connection is closed.</li>
+	 * </ol>
+	 *
+	 * @author Mark Paluch
+	 * @since 2.0.1
+	 */
+	static class AsyncConnect {
+
+		private final Mono<StatefulConnection<ByteBuffer, ByteBuffer>> connectionPublisher;
+		private final LettuceConnectionProvider connectionProvider;
+
+		private AtomicReference<State> state = new AtomicReference<>(State.INITIAL);
+		private volatile @Nullable CompletableFuture<StatefulConnection<ByteBuffer, ByteBuffer>> connection;
+
+		@SuppressWarnings("unchecked")
+		AsyncConnect(LettuceConnectionProvider connectionProvider) {
+
+			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)));
+
+			this.connectionPublisher = defer.subscribeOn(Schedulers.elastic());
+		}
+
+		/**
+		 * Obtain a connection publisher. This method connects asynchronously by requesting a connection from
+		 * {@link LettuceConnectionProvider} with non-blocking synchronization if called concurrently.
+		 *
+		 * @return
+		 */
+		Mono<StatefulConnection<ByteBuffer, ByteBuffer>> getConnection() {
+
+			if (state.get() == State.CLOSED) {
+				throw new IllegalStateException("Connection is closed!");
+			}
+
+			CompletableFuture<StatefulConnection<ByteBuffer, ByteBuffer>> connection = this.connection;
+
+			if (connection != null) {
+				return Mono.fromCompletionStage(connection);
+			}
+
+			if (state.compareAndSet(State.INITIAL, State.CONNECTION_REQUESTED)) {
+				this.connection = connectionPublisher.toFuture();
+			}
+
+			for (;;) {
+
+				connection = this.connection;
+				if (connection != null) {
+					return Mono.fromCompletionStage(connection);
+				}
+
+				Thread thread = Thread.currentThread();
+				if (thread.isInterrupted()) {
+					thread.interrupt();
+					return Mono.error(new InterruptedException());
+				}
+			}
+		}
+
+		/**
+		 * Close connection (blocking call).
+		 */
+		void close() {
+
+			if (state.compareAndSet(State.CONNECTION_REQUESTED, State.CLOSING)) {
+				getConnection().doOnSuccess(connectionProvider::release).doOnSuccess(it -> state.set(State.CLOSED)).block();
+			}
+
+			state.compareAndSet(State.INITIAL, State.CLOSED);
+		}
+
+		enum State {
+			INITIAL, CONNECTION_REQUESTED, CLOSING, CLOSED;
+		}
+	}
 }