DATAREDIS-667 - Introduce LettuceConnectionProvider.

We now create and release Lettuce connections using LettuceConnectionProvider. Connection providers encapsulate the underlying client and expose only creation and release methods. A connection provider can provide connections for Standalone or Cluster connections or wrap a LettucePool.

Previously we used RedisClient and RedisClusterClient directly which required context propagation (pooling/non-pooling) conditional code to obtain the appropriate connection type.

Original Pull Request: #262

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

@@ -0,0 +1,64 @@
+/*
+ * 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.api.StatefulConnection;
+import io.lettuce.core.cluster.RedisClusterClient;
+import io.lettuce.core.cluster.api.StatefulRedisClusterConnection;
+import io.lettuce.core.codec.RedisCodec;
+import io.lettuce.core.pubsub.StatefulRedisPubSubConnection;
+
+/**
+ * Connection provider for Cluster connections.
+ *
+ * @author Mark Paluch
+ * @since 2.0
+ */
+class ClusterConnectionProvider implements LettuceConnectionProvider {
+
+	private final RedisClusterClient client;
+	private final RedisCodec<?, ?> codec;
+
+	ClusterConnectionProvider(RedisClusterClient client, RedisCodec<?, ?> codec) {
+
+		this.client = client;
+		this.codec = codec;
+	}
+
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.redis.connection.lettuce.LettuceConnectionProvider#getConnection(java.lang.Class)
+	 */
+	@SuppressWarnings("rawtypes")
+	@Override
+	public StatefulConnection<?, ?> getConnection(Class<? extends StatefulConnection> connectionType) {
+
+		if (connectionType.equals(StatefulRedisPubSubConnection.class)) {
+			return client.connectPubSub(codec);
+		}
+
+		if (StatefulRedisClusterConnection.class.isAssignableFrom(connectionType)
+				|| connectionType.equals(StatefulConnection.class)) {
+			return client.connect(codec);
+		}
+
+		throw new UnsupportedOperationException("Connection type " + connectionType + " not supported!");
+	}
+
+	public RedisClusterClient getClient() {
+		return client;
+	}
+}

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

@@ -17,15 +17,12 @@ package org.springframework.data.redis.connection.lettuce;
 
 import io.lettuce.core.RedisException;
 import io.lettuce.core.RedisURI;
-import io.lettuce.core.api.StatefulConnection;
 import io.lettuce.core.api.sync.BaseRedisCommands;
 import io.lettuce.core.cluster.RedisClusterClient;
 import io.lettuce.core.cluster.SlotHash;
 import io.lettuce.core.cluster.api.StatefulRedisClusterConnection;
 import io.lettuce.core.cluster.api.sync.RedisClusterCommands;
-import io.lettuce.core.codec.ByteArrayCodec;
-import io.lettuce.core.codec.RedisCodec;
-import io.lettuce.core.pubsub.StatefulRedisPubSubConnection;
+import lombok.RequiredArgsConstructor;
 
 import java.time.Duration;
 import java.util.ArrayList;
@@ -60,11 +57,9 @@ public class LettuceClusterConnection extends LettuceConnection implements Defau
 
 	static final ExceptionTranslationStrategy exceptionConverter = new PassThroughExceptionTranslationStrategy(
 			new LettuceExceptionConverter());
-	static final RedisCodec<byte[], byte[]> CODEC = ByteArrayCodec.INSTANCE;
 
 	private final Log log = LogFactory.getLog(getClass());
 
-	private final RedisClusterClient clusterClient;
 	private ClusterCommandExecutor clusterCommandExecutor;
 	private ClusterTopologyProvider topologyProvider;
 	private boolean disposeClusterCommandExecutorOnClose;
@@ -77,12 +72,7 @@ public class LettuceClusterConnection extends LettuceConnection implements Defau
 	 * @param clusterClient must not be {@literal null}.
 	 */
 	public LettuceClusterConnection(RedisClusterClient clusterClient) {
-
-		this(clusterClient, RedisURI.DEFAULT_TIMEOUT_DURATION,
-				new ClusterCommandExecutor(new LettuceClusterTopologyProvider(clusterClient),
-						new LettuceClusterNodeResourceProvider(clusterClient), exceptionConverter));
-
-		this.disposeClusterCommandExecutorOnClose = true;
+		this(new ClusterConnectionProvider(clusterClient, CODEC));
 	}
 
 	/**
@@ -93,7 +83,7 @@ public class LettuceClusterConnection extends LettuceConnection implements Defau
 	 * @param executor must not be {@literal null}.
 	 */
 	public LettuceClusterConnection(RedisClusterClient clusterClient, ClusterCommandExecutor executor) {
-		this(clusterClient, RedisURI.DEFAULT_TIMEOUT_DURATION, executor);
+		this(clusterClient, executor, RedisURI.DEFAULT_TIMEOUT_DURATION);
 	}
 
 	/**
@@ -105,16 +95,64 @@ public class LettuceClusterConnection extends LettuceConnection implements Defau
 	 * @param executor must not be {@literal null}.
 	 * @since 2.0
 	 */
-	public LettuceClusterConnection(RedisClusterClient clusterClient, Duration timeout, ClusterCommandExecutor executor) {
+	public LettuceClusterConnection(RedisClusterClient clusterClient, ClusterCommandExecutor executor, Duration timeout) {
+		this(new ClusterConnectionProvider(clusterClient, CODEC), executor, timeout);
+	}
 
-		super(null, timeout.toMillis(), clusterClient, null, 0);
+	/**
+	 * Creates new {@link LettuceClusterConnection} using {@link LettuceConnectionProvider} running commands across the
+	 * cluster via given {@link ClusterCommandExecutor}.
+	 *
+	 * @param connectionProvider must not be {@literal null}.
+	 * @since 2.0
+	 */
+	public LettuceClusterConnection(LettuceConnectionProvider connectionProvider) {
+
+		super(null, connectionProvider, RedisURI.DEFAULT_TIMEOUT_DURATION.toMillis(), 0);
+
+		topologyProvider = new LettuceClusterTopologyProvider(getClient());
+		clusterCommandExecutor = new ClusterCommandExecutor(topologyProvider,
+				new LettuceClusterNodeResourceProvider(getConnectionProvider()), exceptionConverter);
+		disposeClusterCommandExecutorOnClose = true;
+	}
+
+	/**
+	 * Creates new {@link LettuceClusterConnection} using {@link LettuceConnectionProvider} running commands across the
+	 * cluster via given {@link ClusterCommandExecutor}.
+	 *
+	 * @param connectionProvider must not be {@literal null}.
+	 * @param executor must not be {@literal null}.
+	 * @since 2.0
+	 */
+	public LettuceClusterConnection(LettuceConnectionProvider connectionProvider, ClusterCommandExecutor executor) {
+		this(connectionProvider, executor, RedisURI.DEFAULT_TIMEOUT_DURATION);
+	}
+
+	/**
+	 * Creates new {@link LettuceClusterConnection} using {@link LettuceConnectionProvider} running commands across the
+	 * cluster via given {@link ClusterCommandExecutor}.
+	 *
+	 * @param connectionProvider must not be {@literal null}.
+	 * @param executor must not be {@literal null}.
+	 * @since 2.0
+	 */
+	public LettuceClusterConnection(LettuceConnectionProvider connectionProvider, ClusterCommandExecutor executor,
+			Duration timeout) {
+
+		super(null, connectionProvider, timeout.toMillis(), 0);
 
-		Assert.notNull(clusterClient, "RedisClusterClient must not be null.");
 		Assert.notNull(executor, "ClusterCommandExecutor must not be null.");
 
-		this.clusterClient = clusterClient;
-		this.topologyProvider = new LettuceClusterTopologyProvider(clusterClient);
+		this.topologyProvider = new LettuceClusterTopologyProvider(getClient());
 		this.clusterCommandExecutor = executor;
+		this.disposeClusterCommandExecutorOnClose = false;
+	}
+
+	/**
+	 * @return access to {@link RedisClusterClient} for non-connection access.
+	 */
+	private RedisClusterClient getClient() {
+		return ((ClusterConnectionProvider) getConnectionProvider()).getClient();
 	}
 
 	/*
@@ -235,7 +273,7 @@ public class LettuceClusterConnection extends LettuceConnection implements Defau
 	@Override
 	public RedisClusterNode clusterGetNodeForSlot(int slot) {
 
-		return LettuceConverters.toRedisClusterNode(clusterClient.getPartitions().getPartitionBySlot(slot));
+		return LettuceConverters.toRedisClusterNode(getClient().getPartitions().getPartitionBySlot(slot));
 	}
 
 	/*
@@ -459,15 +497,6 @@ public class LettuceClusterConnection extends LettuceConnection implements Defau
 		}
 	}
 
-	/*
-	 * (non-Javadoc)
-	 * @see org.springframework.data.redis.connection.lettuce.LettuceConnection#getAsyncDedicatedConnection()
-	 */
-	@Override
-	protected StatefulConnection<byte[], byte[]> doGetAsyncDedicatedConnection() {
-		return clusterClient.connect(CODEC);
-	}
-
 	// --> cluster node stuff
 
 	/*
@@ -476,7 +505,7 @@ public class LettuceClusterConnection extends LettuceConnection implements Defau
 	 */
 	@Override
 	public List<RedisClusterNode> clusterGetNodes() {
-		return LettuceConverters.partitionsToClusterNodes(clusterClient.getPartitions());
+		return LettuceConverters.partitionsToClusterNodes(getClient().getPartitions());
 	}
 
 	/*
@@ -527,19 +556,6 @@ public class LettuceClusterConnection extends LettuceConnection implements Defau
 		return result;
 	}
 
-	/*
-	 * (non-Javadoc)
-	 * @see org.springframework.data.redis.connection.lettuce.LettuceConnection#switchToPubSub()
-	 */
-	@Override
-	protected StatefulRedisPubSubConnection<byte[], byte[]> switchToPubSub() {
-
-		close();
-
-		// open a pubsub one
-		return clusterClient.connectPubSub(CODEC);
-	}
-
 	public ClusterCommandExecutor getClusterCommandExecutor() {
 		return clusterCommandExecutor;
 	}
@@ -588,16 +604,12 @@ public class LettuceClusterConnection extends LettuceConnection implements Defau
 	 * @author Christoph Strobl
 	 * @since 1.7
 	 */
+	@RequiredArgsConstructor
 	static class LettuceClusterNodeResourceProvider implements ClusterNodeResourceProvider, DisposableBean {
 
-		private final RedisClusterClient client;
+		private final LettuceConnectionProvider connectionProvider;
 		private volatile StatefulRedisClusterConnection<byte[], byte[]> connection;
 
-		public LettuceClusterNodeResourceProvider(RedisClusterClient client) {
-
-			this.client = client;
-		}
-
 		@Override
 		@SuppressWarnings("unchecked")
 		public RedisClusterCommands<byte[], byte[]> getResourceForSpecificNode(RedisClusterNode node) {
@@ -607,7 +619,7 @@ public class LettuceClusterConnection extends LettuceConnection implements Defau
 			if (connection == null) {
 				synchronized (this) {
 					if (connection == null) {
-						this.connection = client.connect(CODEC);
+						this.connection = connectionProvider.getConnection();
 					}
 				}
 			}
@@ -631,7 +643,7 @@ public class LettuceClusterConnection extends LettuceConnection implements Defau
 		@Override
 		public void destroy() throws Exception {
 			if (connection != null) {
-				connection.close();
+				connectionProvider.release(connection);
 			}
 		}
 	}

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

@@ -40,6 +40,7 @@ import io.lettuce.core.protocol.CommandArgs;
 import io.lettuce.core.protocol.CommandType;
 import io.lettuce.core.pubsub.StatefulRedisPubSubConnection;
 import io.lettuce.core.sentinel.api.StatefulRedisSentinelConnection;
+import lombok.RequiredArgsConstructor;
 
 import java.lang.reflect.Constructor;
 import java.util.ArrayList;
@@ -60,9 +61,9 @@ import org.springframework.dao.InvalidDataAccessApiUsageException;
 import org.springframework.dao.QueryTimeoutException;
 import org.springframework.data.redis.ExceptionTranslationStrategy;
 import org.springframework.data.redis.FallbackExceptionTranslationStrategy;
-import org.springframework.data.redis.RedisConnectionFailureException;
 import org.springframework.data.redis.connection.*;
 import org.springframework.data.redis.connection.convert.TransactionResultConverter;
+import org.springframework.data.redis.connection.lettuce.LettuceConnectionProvider.TargetAware;
 import org.springframework.data.redis.core.RedisCommand;
 import org.springframework.data.redis.core.ScanOptions;
 import org.springframework.util.Assert;
@@ -92,6 +93,7 @@ public class LettuceConnection extends AbstractRedisConnection {
 	private final int defaultDbIndex;
 	private int dbIndex;
 
+	private final LettuceConnectionProvider connectionProvider;
 	private final StatefulConnection<byte[], byte[]> asyncSharedConn;
 	private StatefulConnection<byte[], byte[]> asyncDedicatedConn;
 
@@ -103,11 +105,8 @@ public class LettuceConnection extends AbstractRedisConnection {
 	private boolean isPipelined = false;
 	private List<LettuceResult> ppline;
 	private Queue<FutureResult<?>> txResults = new LinkedList<>();
-	private AbstractRedisClient client;
 	private volatile LettuceSubscription subscription;
-	private LettucePool pool;
 	/** flag indicating whether the connection needs to be dropped or not */
-	private boolean broken = false;
 	private boolean convertPipelineAndTxResults = true;
 
 	@SuppressWarnings("rawtypes")
@@ -258,7 +257,7 @@ public class LettuceConnection extends AbstractRedisConnection {
 
 	/**
 	 * @param sharedConnection A native connection that is shared with other {@link LettuceConnection}s. Should not be
-	 *          used for transactions or blocking operations
+	 *          used for transactions or blocking operations.
 	 * @param timeout The connection timeout (in milliseconds)
 	 * @param client The {@link RedisClient} to use when making pub/sub connections.
 	 * @param pool The connection pool to use for blocking and tx operations.
@@ -268,22 +267,40 @@ public class LettuceConnection extends AbstractRedisConnection {
 	public LettuceConnection(StatefulRedisConnection<byte[], byte[]> sharedConnection, long timeout,
 			AbstractRedisClient client, LettucePool pool, int defaultDbIndex) {
 
+		if (pool != null) {
+			this.connectionProvider = new LettucePoolConnectionProvider(pool);
+		} else {
+			this.connectionProvider = new StandaloneConnectionProvider((RedisClient) client, CODEC);
+		}
+
 		this.asyncSharedConn = sharedConnection;
 		this.timeout = timeout;
-		this.client = client;
-		this.pool = pool;
+		this.defaultDbIndex = defaultDbIndex;
+		this.dbIndex = this.defaultDbIndex;
+	}
+
+	/**
+	 * @param sharedConnection A native connection that is shared with other {@link LettuceConnection}s. Should not be
+	 *          used for transactions or blocking operations.
+	 * @param connectionProvider connection provider to obtain and release native connections.
+	 * @param timeout The connection timeout (in milliseconds)
+	 * @param defaultDbIndex The db index to use along with {@link RedisClient} when establishing a dedicated connection.
+	 * @since 2.0
+	 */
+	public LettuceConnection(StatefulRedisConnection<byte[], byte[]> sharedConnection,
+			LettuceConnectionProvider connectionProvider, long timeout, int defaultDbIndex) {
+
+		Assert.notNull(connectionProvider, "LettuceConnectionProvider must not be null.");
+
+		this.asyncSharedConn = sharedConnection;
+		this.connectionProvider = connectionProvider;
+		this.timeout = timeout;
 		this.defaultDbIndex = defaultDbIndex;
 		this.dbIndex = this.defaultDbIndex;
 	}
 
 	protected DataAccessException convertLettuceAccessException(Exception ex) {
-
-		DataAccessException exception = EXCEPTION_TRANSLATION.translate(ex);
-
-		if (exception instanceof RedisConnectionFailureException) {
-			broken = true;
-		}
-		return exception;
+		return EXCEPTION_TRANSLATION.translate(ex);
 	}
 
 	/*
@@ -435,34 +452,22 @@ public class LettuceConnection extends AbstractRedisConnection {
 		}
 	}
 
-	private void returnDedicatedAsyncConnection() {
-
-		if (pool != null) {
-
-			if (!broken) {
-				pool.returnResource(this.asyncDedicatedConn);
-			} else {
-				pool.returnBrokenResource(this.asyncDedicatedConn);
-			}
-			this.asyncDedicatedConn = null;
-
-		} else {
-
-			try {
-				asyncDedicatedConn.close();
-			} catch (RuntimeException ex) {
-				throw convertLettuceAccessException(ex);
-			}
-		}
-	}
-
 	public void close() throws DataAccessException {
+
 		super.close();
 
+		if (isClosed) {
+			return;
+		}
+
 		isClosed = true;
 
 		if (asyncDedicatedConn != null) {
-			returnDedicatedAsyncConnection();
+			try {
+				connectionProvider.release(asyncDedicatedConn);
+			} catch (RuntimeException ex) {
+				throw convertLettuceAccessException(ex);
+			}
 		}
 
 		if (subscription != null) {
@@ -803,10 +808,11 @@ public class LettuceConnection extends AbstractRedisConnection {
 	 *
 	 * @return never {@literal null}.
 	 */
+	@SuppressWarnings("unchecked")
 	protected StatefulRedisPubSubConnection<byte[], byte[]> switchToPubSub() {
 
 		close();
-		return ((RedisClient) client).connectPubSub(CODEC);
+		return (StatefulRedisPubSubConnection) connectionProvider.getConnection(StatefulRedisPubSubConnection.class);
 	}
 
 	void pipeline(LettuceResult result) {
@@ -852,15 +858,13 @@ public class LettuceConnection extends AbstractRedisConnection {
 	}
 
 	protected RedisClusterAsyncCommands<byte[], byte[]> getAsyncDedicatedConnection() {
+
 		if (asyncDedicatedConn == null) {
 
 			asyncDedicatedConn = doGetAsyncDedicatedConnection();
 
-			if (this.pool == null) {
-
-				if (asyncDedicatedConn instanceof StatefulRedisConnection) {
-					((StatefulRedisConnection<byte[], byte[]>) asyncDedicatedConn).sync().select(dbIndex);
-				}
+			if (asyncDedicatedConn instanceof StatefulRedisConnection) {
+				((StatefulRedisConnection<byte[], byte[]>) asyncDedicatedConn).sync().select(dbIndex);
 			}
 		}
 
@@ -875,26 +879,14 @@ public class LettuceConnection extends AbstractRedisConnection {
 				String.format("%s is not a supported connection type.", asyncDedicatedConn.getClass().getName()));
 	}
 
-	protected StatefulConnection<byte[], byte[]> doGetAsyncDedicatedConnection() {
-
-		if (this.pool != null) {
-			return pool.getResource();
-		} else {
-			return ((RedisClient) client).connect(CODEC);
-		}
-	}
-
 	RedisClusterCommands<byte[], byte[]> getDedicatedConnection() {
 
 		if (asyncDedicatedConn == null) {
 
 			asyncDedicatedConn = doGetAsyncDedicatedConnection();
 
-			if (this.pool == null) {
-
-				if (asyncDedicatedConn instanceof StatefulRedisConnection) {
-					((StatefulRedisConnection<byte[], byte[]>) asyncDedicatedConn).sync().select(dbIndex);
-				}
+			if (asyncDedicatedConn instanceof StatefulRedisConnection) {
+				((StatefulRedisConnection<byte[], byte[]>) asyncDedicatedConn).sync().select(dbIndex);
 			}
 		}
 
@@ -909,6 +901,10 @@ public class LettuceConnection extends AbstractRedisConnection {
 				String.format("%s is not a supported connection type.", asyncDedicatedConn.getClass().getName()));
 	}
 
+	protected StatefulConnection<byte[], byte[]> doGetAsyncDedicatedConnection() {
+		return connectionProvider.getConnection();
+	}
+
 	io.lettuce.core.ScanCursor getScanCursor(long cursorId) {
 		return io.lettuce.core.ScanCursor.of(Long.toString(cursorId));
 	}
@@ -957,15 +953,15 @@ public class LettuceConnection extends AbstractRedisConnection {
 			return false;
 		}
 
-		StatefulRedisConnection<String, String> connection = null;
+		StatefulRedisSentinelConnection<String, String> connection = null;
 		try {
-			connection = ((RedisClient) client).connect(getRedisURI(node));
+			connection = getConnection(node);
 			return connection.sync().ping().equalsIgnoreCase("pong");
 		} catch (Exception e) {
 			return false;
 		} finally {
 			if (connection != null) {
-				connection.close();
+				connectionProvider.release(connection);
 			}
 		}
 	}
@@ -980,11 +976,21 @@ public class LettuceConnection extends AbstractRedisConnection {
 	 */
 	@Override
 	protected RedisSentinelConnection getSentinelConnection(RedisNode sentinel) {
-		StatefulRedisSentinelConnection<String, String> connection = ((RedisClient) client)
-				.connectSentinel(getRedisURI(sentinel));
+
+		StatefulRedisSentinelConnection<String, String> connection = getConnection(sentinel);
 		return new LettuceSentinelConnection(connection);
 	}
 
+	@SuppressWarnings("unchecked")
+	private StatefulRedisSentinelConnection<String, String> getConnection(RedisNode sentinel) {
+		return (StatefulRedisSentinelConnection) ((TargetAware) connectionProvider)
+				.getConnection(StatefulRedisSentinelConnection.class, getRedisURI(sentinel));
+	}
+
+	LettuceConnectionProvider getConnectionProvider() {
+		return connectionProvider;
+	}
+
 	/**
 	 * {@link TypeHints} provide {@link CommandOutput} information for a given {@link CommandType}.
 	 *
@@ -1185,4 +1191,27 @@ public class LettuceConnection extends AbstractRedisConnection {
 			return BeanUtils.instantiateClass(constructor, CODEC);
 		}
 	}
+
+	@RequiredArgsConstructor
+	private class LettucePoolConnectionProvider implements LettuceConnectionProvider {
+
+		private final LettucePool pool;
+
+		@Override
+		@SuppressWarnings("unchecked")
+		public StatefulConnection<?, ?> getConnection(Class<? extends StatefulConnection> connectionType) {
+			return pool.getResource();
+		}
+
+		@Override
+		@SuppressWarnings("unchecked")
+		public void release(StatefulConnection<?, ?> connection) {
+
+			if (connection.isOpen()) {
+				pool.returnResource((StatefulConnection<byte[], byte[]>) connection);
+			} else {
+				pool.returnBrokenResource((StatefulConnection<byte[], byte[]>) connection);
+			}
+		}
+	}
 }

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

@@ -23,6 +23,7 @@ import io.lettuce.core.RedisURI;
 import io.lettuce.core.api.StatefulRedisConnection;
 import io.lettuce.core.cluster.ClusterClientOptions;
 import io.lettuce.core.cluster.RedisClusterClient;
+import io.lettuce.core.codec.RedisCodec;
 import io.lettuce.core.resource.ClientResources;
 
 import java.time.Duration;
@@ -82,6 +83,8 @@ public class LettuceConnectionFactory
 	private final LettuceClientConfiguration clientConfiguration;
 
 	private AbstractRedisClient client;
+	private LettuceConnectionProvider connectionProvider;
+	private LettuceConnectionProvider reactiveConnectionProvider;
 	private boolean validateConnection = false;
 	private boolean shareNativeConnection = true;
 	private StatefulRedisConnection<byte[], byte[]> connection;
@@ -213,7 +216,18 @@ public class LettuceConnectionFactory
 	 * @see org.springframework.beans.factory.InitializingBean#afterPropertiesSet()
 	 */
 	public void afterPropertiesSet() {
-		this.client = createRedisClient();
+
+		this.client = createClient();
+
+		this.connectionProvider = createConnectionProvider(client, LettuceConnection.CODEC);
+		this.reactiveConnectionProvider = createConnectionProvider(client, LettuceReactiveRedisConnection.CODEC);
+
+		if (isClusterAware()) {
+			this.clusterCommandExecutor = new ClusterCommandExecutor(
+					new LettuceClusterTopologyProvider((RedisClusterClient) client),
+					new LettuceClusterConnection.LettuceClusterNodeResourceProvider(this.connectionProvider),
+					EXCEPTION_TRANSLATION);
+		}
 	}
 
 	/*
@@ -224,6 +238,17 @@ public class LettuceConnectionFactory
 
 		resetConnection();
 
+		if (connectionProvider instanceof DisposableBean) {
+			try {
+				((DisposableBean) connectionProvider).destroy();
+			} catch (Exception e) {
+
+				if (log.isWarnEnabled()) {
+					log.warn(connectionProvider + " did not shut down gracefully.", e);
+				}
+			}
+		}
+
 		try {
 			Duration timeout = clientConfiguration.getShutdownTimeout();
 			client.shutdown(timeout.toMillis(), timeout.toMillis(), TimeUnit.MILLISECONDS);
@@ -255,8 +280,14 @@ public class LettuceConnectionFactory
 			return getClusterConnection();
 		}
 
-		LettuceConnection connection = new LettuceConnection(getSharedConnection(), getTimeout(), client, pool,
-				getDatabase());
+		LettuceConnection connection;
+		if (pool != null) {
+
+			connection = new LettuceConnection(getSharedConnection(), getTimeout(), null, pool, getDatabase());
+		} else {
+			connection = new LettuceConnection(getSharedConnection(), connectionProvider, getTimeout(), getDatabase());
+		}
+
 		connection.setConvertPipelineAndTxResults(convertPipelineAndTxResults);
 		return connection;
 	}
@@ -272,8 +303,8 @@ public class LettuceConnectionFactory
 			throw new InvalidDataAccessApiUsageException("Cluster is not configured!");
 		}
 
-		return new LettuceClusterConnection((RedisClusterClient) client, clientConfiguration.getCommandTimeout(),
-				clusterCommandExecutor);
+		return new LettuceClusterConnection(connectionProvider, clusterCommandExecutor,
+				clientConfiguration.getCommandTimeout());
 	}
 
 	/*
@@ -282,7 +313,7 @@ public class LettuceConnectionFactory
 	 */
 	@Override
 	public LettuceReactiveRedisConnection getReactiveConnection() {
-		return new LettuceReactiveRedisConnection(client);
+		return new LettuceReactiveRedisConnection(reactiveConnectionProvider);
 	}
 
 	/*
@@ -295,7 +326,7 @@ public class LettuceConnectionFactory
 			throw new InvalidDataAccessApiUsageException("Cluster is not configured!");
 		}
 
-		return new LettuceReactiveRedisClusterConnection((RedisClusterClient) client);
+		return new LettuceReactiveRedisClusterConnection(reactiveConnectionProvider);
 	}
 
 	public void initConnection() {
@@ -314,7 +345,7 @@ public class LettuceConnectionFactory
 	public void resetConnection() {
 		synchronized (this.connectionMonitor) {
 			if (this.connection != null) {
-				this.connection.close();
+				this.connectionProvider.release(this.connection);
 			}
 			this.connection = null;
 		}
@@ -338,6 +369,8 @@ public class LettuceConnectionFactory
 			}
 
 			if (!valid) {
+
+				connectionProvider.release(connection);
 				log.warn("Validation of shared connection failed. Creating a new connection.");
 				initConnection();
 			}
@@ -729,9 +762,9 @@ public class LettuceConnectionFactory
 	protected StatefulRedisConnection<byte[], byte[]> createLettuceConnector() {
 		try {
 
-			StatefulRedisConnection<byte[], byte[]> connection = null;
-			if (client instanceof RedisClient) {
-				connection = ((RedisClient) client).connect(LettuceConnection.CODEC);
+			StatefulRedisConnection<byte[], byte[]> connection;
+			if (!isClusterAware()) {
+				connection = connectionProvider.getConnection();
 				if (getDatabase() > 0) {
 					connection.sync().select(getDatabase());
 				}
@@ -744,7 +777,16 @@ public class LettuceConnectionFactory
 		}
 	}
 
-	private AbstractRedisClient createRedisClient() {
+	private LettuceConnectionProvider createConnectionProvider(AbstractRedisClient client, RedisCodec<?, ?> codec) {
+
+		if (isClusterAware()) {
+			return new ClusterConnectionProvider((RedisClusterClient) client, codec);
+		}
+
+		return new StandaloneConnectionProvider((RedisClient) client, codec);
+	}
+
+	private AbstractRedisClient createClient() {
 
 		if (isRedisSentinelAware()) {
 
@@ -768,9 +810,6 @@ public class LettuceConnectionFactory
 					.map(clientResources -> RedisClusterClient.create(clientResources, initialUris)) //
 					.orElseGet(() -> RedisClusterClient.create(initialUris));
 
-			this.clusterCommandExecutor = new ClusterCommandExecutor(new LettuceClusterTopologyProvider(clusterClient),
-					new LettuceClusterConnection.LettuceClusterNodeResourceProvider(clusterClient), EXCEPTION_TRANSLATION);
-
 			clientConfiguration.getClientOptions() //
 					.filter(clientOptions -> clientOptions instanceof ClusterClientOptions) //
 					.ifPresent(clientOptions -> clusterClient.setOptions((ClusterClientOptions) clientOptions));
@@ -778,15 +817,12 @@ public class LettuceConnectionFactory
 			return clusterClient;
 		}
 
-		if (pool != null) {
-			return pool.getClient();
-		}
-
 		RedisURI uri = createRedisURIAndApplySettings(getHostName(), getPort());
 		RedisClient redisClient = clientConfiguration.getClientResources() //
 				.map(clientResources -> RedisClient.create(clientResources, uri)) //
 				.orElseGet(() -> RedisClient.create(uri));
 		clientConfiguration.getClientOptions().ifPresent(redisClient::setOptions);
+
 		return redisClient;
 	}
 
@@ -817,10 +853,11 @@ public class LettuceConnectionFactory
 	@Override
 	public RedisSentinelConnection getSentinelConnection() {
 
-		if (!(client instanceof RedisClient)) {
-			throw new InvalidDataAccessResourceUsageException("Unable to connect to sentinels using " + client.getClass());
+		if (!(connectionProvider instanceof StandaloneConnectionProvider)) {
+			throw new InvalidDataAccessResourceUsageException(
+					"Unable to connect to sentinels using " + connectionProvider.getClass());
 		}
-		return new LettuceSentinelConnection(((RedisClient) client).connectSentinel());
+		return new LettuceSentinelConnection(connectionProvider);
 	}
 
 	private MutableLettuceClientConfiguration getMutableConfiguration() {

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

@@ -0,0 +1,88 @@
+/*
+ * 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.RedisURI;
+import io.lettuce.core.api.StatefulConnection;
+
+/**
+ * Defines a provider for Lettuce connections.
+ * <p />
+ * This interface is typically used to encapsulate a native factory which returns a {@link StatefulConnection
+ * connection} of on each invocation.
+ * <p/>
+ * Connection providers may create a new connection on each invocation or return pooled instances. Each obtained
+ * connection must be released through its connection provider to allow disposal or release back to the pool.
+ * <p/>
+ * Connection providers are usually associated with a {@link io.lettuce.core.codec.RedisCodec} to create connections
+ * with an appropriate codec.
+ *
+ * @author Mark Paluch
+ * @since 2.0
+ * @see StatefulConnection
+ */
+@FunctionalInterface
+public interface LettuceConnectionProvider {
+
+	/**
+	 * Request a connection given {@code connectionType}. Providing a connection type allows specialization to provide a
+	 * more specific connection type.
+	 *
+	 * @param connectionType must not be {@literal null}.
+	 * @return the requested connection. Must be {@link #release(StatefulConnection) released} if the connection is no
+	 *         longer in use.
+	 */
+	@SuppressWarnings("rawtypes")
+	StatefulConnection<?, ?> getConnection(Class<? extends StatefulConnection> connectionType);
+
+	/**
+	 * Request a connection.
+	 *
+	 * @return the requested connection. Must be {@link #release(StatefulConnection) released} if the connection is no
+	 *         longer in use.
+	 */
+	@SuppressWarnings("unchecked")
+	default <T> T getConnection() {
+		return (T) getConnection(StatefulConnection.class);
+	}
+
+	/**
+	 * Release the {@link StatefulConnection connection}. Closes connection {@link StatefulConnection#close()} by default.
+	 * Implementations may choose whether they override this method and return the connection to a pool.
+	 *
+	 * @param connection must not be {@literal null}.
+	 */
+	default void release(StatefulConnection<?, ?> connection) {
+		connection.close();
+	}
+
+	/**
+	 * Extension to {@link LettuceConnectionProvider} for providers that allow connection creation to specific nodes.
+	 */
+	interface TargetAware {
+
+		/**
+		 * Request a connection given {@code connectionType} for a specific {@link RedisURI}. Providing a connection type
+		 * allows specialization to provide a more specific connection type.
+		 *
+		 * @param connectionType must not be {@literal null}.
+		 * @param redisURI must not be {@literal null}.
+		 * @return the requested connection.
+		 */
+		@SuppressWarnings("rawtypes")
+		StatefulConnection<?, ?> getConnection(Class<? extends StatefulConnection> connectionType, RedisURI redisURI);
+	}
+}

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

@@ -52,11 +52,12 @@ class LettuceReactiveRedisClusterConnection extends LettuceReactiveRedisConnecti
 	 * @throws org.springframework.dao.InvalidDataAccessResourceUsageException when {@code client} is not suitable for
 	 *           cluster environment.
 	 */
-	LettuceReactiveRedisClusterConnection(RedisClusterClient client) {
+	LettuceReactiveRedisClusterConnection(LettuceConnectionProvider connectionProvider) {
 
-		super(client);
+		super(connectionProvider);
 
-		this.topologyProvider = new LettuceClusterTopologyProvider(client);
+		this.topologyProvider = new LettuceClusterTopologyProvider(
+				((ClusterConnectionProvider) connectionProvider).getClient());
 	}
 
 	/*

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

@@ -15,12 +15,9 @@
  */
 package org.springframework.data.redis.connection.lettuce;
 
-import io.lettuce.core.AbstractRedisClient;
-import io.lettuce.core.RedisClient;
 import io.lettuce.core.api.StatefulConnection;
 import io.lettuce.core.api.StatefulRedisConnection;
 import io.lettuce.core.api.reactive.BaseRedisReactiveCommands;
-import io.lettuce.core.cluster.RedisClusterClient;
 import io.lettuce.core.cluster.api.StatefulRedisClusterConnection;
 import io.lettuce.core.cluster.api.reactive.RedisClusterReactiveCommands;
 import io.lettuce.core.codec.RedisCodec;
@@ -43,29 +40,25 @@ import org.springframework.util.Assert;
  */
 class LettuceReactiveRedisConnection implements ReactiveRedisConnection {
 
-	private static final RedisCodec<ByteBuffer, ByteBuffer> CODEC = ByteBufferCodec.INSTANCE;
+	static final RedisCodec<ByteBuffer, ByteBuffer> CODEC = ByteBufferCodec.INSTANCE;
+
+	private final LettuceConnectionProvider connectionProvider;
 
 	private StatefulConnection<ByteBuffer, ByteBuffer> connection;
 
 	/**
 	 * Creates new {@link LettuceReactiveRedisConnection}.
 	 *
-	 * @param client 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.
 	 */
-	LettuceReactiveRedisConnection(AbstractRedisClient client) {
+	LettuceReactiveRedisConnection(LettuceConnectionProvider connectionProvider) {
 
-		Assert.notNull(client, "RedisClient must not be null!");
+		Assert.notNull(connectionProvider, "LettuceConnectionProvider must not be null!");
 
-		if (client instanceof RedisClient) {
-			connection = ((RedisClient) client).connect(CODEC);
-		} else if (client instanceof RedisClusterClient) {
-			connection = ((RedisClusterClient) client).connect(CODEC);
-		} else {
-			throw new InvalidDataAccessResourceUsageException(
-					String.format("Cannot use client of type %s", client.getClass()));
-		}
+		this.connectionProvider = connectionProvider;
+		this.connection = connectionProvider.getConnection();
 	}
 
 	/*
@@ -189,7 +182,11 @@ class LettuceReactiveRedisConnection implements ReactiveRedisConnection {
 	 */
 	@Override
 	public void close() {
-		connection.close();
+
+		synchronized (connectionProvider) {
+			connectionProvider.release(connection);
+			connection = null;
+		}
 	}
 
 	protected StatefulConnection<ByteBuffer, ByteBuffer> getConnection() {

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

@@ -17,6 +17,7 @@ package org.springframework.data.redis.connection.lettuce;
 
 import io.lettuce.core.RedisClient;
 import io.lettuce.core.RedisURI.Builder;
+import io.lettuce.core.api.StatefulConnection;
 import io.lettuce.core.resource.ClientResources;
 import io.lettuce.core.sentinel.api.StatefulRedisSentinelConnection;
 import io.lettuce.core.sentinel.api.sync.RedisSentinelCommands;
@@ -42,12 +43,12 @@ public class LettuceSentinelConnection implements RedisSentinelConnection {
 	private static final ExceptionTranslationStrategy EXCEPTION_TRANSLATION = new FallbackExceptionTranslationStrategy(
 			LettuceConverters.exceptionConverter());
 
-	private RedisClient redisClient;
+	private final LettuceConnectionProvider provider;
 	private StatefulRedisSentinelConnection<String, String> connection;
 
 	/**
 	 * Creates a {@link LettuceSentinelConnection} with a dedicated client for a supplied {@link RedisNode}.
-	 * 
+	 *
 	 * @param sentinel The sentinel to connect to.
 	 */
 	public LettuceSentinelConnection(RedisNode sentinel) {
@@ -56,7 +57,7 @@ public class LettuceSentinelConnection implements RedisSentinelConnection {
 
 	/**
 	 * Creates a {@link LettuceSentinelConnection} with a client for the supplied {@code host} and {@code port}.
-	 * 
+	 *
 	 * @param host must not be {@literal null}.
 	 * @param port sentinel port.
 	 */
@@ -64,14 +65,14 @@ public class LettuceSentinelConnection implements RedisSentinelConnection {
 
 		Assert.notNull(host, "Cannot create LettuceSentinelConnection using 'null' as host.");
 
-		redisClient = RedisClient.create(Builder.redis(host, port).build());
+		this.provider = new DedicatedClientConnectionProvider(host, port);
 		init();
 	}
 
 	/**
 	 * Creates a {@link LettuceSentinelConnection} with a client for the supplied {@code host} and {@code port} and reuse
 	 * existing {@link ClientResources}.
-	 * 
+	 *
 	 * @param host must not be {@literal null}.
 	 * @param port sentinel port.
 	 * @param clientResources must not be {@literal null}.
@@ -81,31 +82,45 @@ public class LettuceSentinelConnection implements RedisSentinelConnection {
 		Assert.notNull(clientResources, "Cannot create LettuceSentinelConnection using 'null' as ClientResources.");
 		Assert.notNull(host, "Cannot create LettuceSentinelConnection using 'null' as host.");
 
-		redisClient = RedisClient.create(clientResources, Builder.redis(host, port).build());
+		this.provider = new DedicatedClientConnectionProvider(host, port, clientResources);
 		init();
 	}
 
 	/**
 	 * Creates a {@link LettuceSentinelConnection} using a supplied {@link RedisClient}.
-	 * 
-	 * @param redisClient
+	 *
+	 * @param redisClient must not be {@literal null}.
 	 */
 	public LettuceSentinelConnection(RedisClient redisClient) {
 
 		Assert.notNull(redisClient, "Cannot create LettuceSentinelConnection using 'null' as client.");
-		this.redisClient = redisClient;
+		this.provider = t -> redisClient.connectSentinel();
 		init();
 	}
 
 	/**
 	 * Creates a {@link LettuceSentinelConnection} using a supplied redis connection.
-	 * 
+	 *
 	 * @param connection native Lettuce connection, must not be {@literal null}
 	 */
 	protected LettuceSentinelConnection(StatefulRedisSentinelConnection<String, String> connection) {
 
 		Assert.notNull(connection, "Cannot create LettuceSentinelConnection using 'null' as connection.");
-		this.connection = connection;
+		this.provider = t -> connection;
+		init();
+	}
+
+	/**
+	 * Creates a {@link LettuceSentinelConnection} using a {@link LettuceConnectionProvider}.
+	 *
+	 * @param connectionProvider must not be {@literal null}.
+	 * @since 2.0
+	 */
+	public LettuceSentinelConnection(LettuceConnectionProvider connectionProvider) {
+
+		Assert.notNull(connectionProvider, "LettuceConnectionProvider must not be null!");
+		this.provider = connectionProvider;
+		init();
 	}
 
 	/*
@@ -202,17 +217,14 @@ public class LettuceSentinelConnection implements RedisSentinelConnection {
 	 */
 	@Override
 	public void close() throws IOException {
-		connection.close();
-		connection = null;
-
-		if (redisClient != null) {
-			redisClient.shutdown();
-		}
+		provider.release(connection);
 	}
 
+	@SuppressWarnings({ "unchecked", "rawtypes" })
 	private void init() {
+
 		if (connection == null) {
-			connection = connectSentinel();
+			connection = (StatefulRedisSentinelConnection) provider.getConnection(StatefulRedisSentinelConnection.class);
 		}
 	}
 
@@ -220,12 +232,52 @@ public class LettuceSentinelConnection implements RedisSentinelConnection {
 		return connection.sync();
 	}
 
-	private StatefulRedisSentinelConnection<String, String> connectSentinel() {
-		return redisClient.connectSentinel();
-	}
-
 	@Override
 	public boolean isOpen() {
 		return connection != null && connection.isOpen();
 	}
+
+	/**
+	 * {@link LettuceConnectionProvider} for a dedicated client instance.
+	 */
+	private static class DedicatedClientConnectionProvider implements LettuceConnectionProvider {
+
+		private final RedisClient redisClient;
+
+		DedicatedClientConnectionProvider(String host, int port) {
+
+			Assert.notNull(host, "Cannot create LettuceSentinelConnection using 'null' as host.");
+
+			redisClient = RedisClient.create(Builder.redis(host, port).build());
+		}
+
+		DedicatedClientConnectionProvider(String host, int port, ClientResources clientResources) {
+
+			Assert.notNull(clientResources, "Cannot create LettuceSentinelConnection using 'null' as ClientResources.");
+			Assert.notNull(host, "Cannot create LettuceSentinelConnection using 'null' as host.");
+
+			redisClient = RedisClient.create(clientResources, Builder.redis(host, port).build());
+		}
+
+		/*
+		 * (non-Javadoc)
+		 * @see org.springframework.data.redis.connection.lettuce.LettuceConnectionProvider#getConnection(java.lang.Class)
+		 */
+		@SuppressWarnings("rawtypes")
+		@Override
+		public StatefulConnection<?, ?> getConnection(Class<? extends StatefulConnection> connectionType) {
+			return redisClient.connectSentinel();
+		}
+
+		/*
+		 * (non-Javadoc)
+		 * @see org.springframework.data.redis.connection.lettuce.LettuceConnectionProvider#release(io.lettuce.core.api.StatefulConnection)
+		 */
+		@Override
+		public void release(StatefulConnection<?, ?> connection) {
+
+			connection.close();
+			redisClient.shutdown();
+		}
+	}
 }

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

@@ -0,0 +1,87 @@
+/*
+ * 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.RedisClient;
+import io.lettuce.core.RedisURI;
+import io.lettuce.core.api.StatefulConnection;
+import io.lettuce.core.codec.RedisCodec;
+import io.lettuce.core.pubsub.StatefulRedisPubSubConnection;
+import io.lettuce.core.sentinel.api.StatefulRedisSentinelConnection;
+
+import org.springframework.data.redis.connection.lettuce.LettuceConnectionProvider.TargetAware;
+
+/**
+ * @author Mark Paluch
+ * @since 2.0
+ */
+class StandaloneConnectionProvider implements LettuceConnectionProvider, TargetAware {
+
+	private final RedisClient client;
+	private final RedisCodec<?, ?> codec;
+
+	StandaloneConnectionProvider(RedisClient client, RedisCodec<?, ?> codec) {
+
+		this.client = client;
+		this.codec = codec;
+	}
+
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.redis.connection.lettuce.LettuceConnectionProvider#getConnection(java.lang.Class)
+	 */
+	@SuppressWarnings("rawtypes")
+	@Override
+	public StatefulConnection<?, ?> getConnection(Class<? extends StatefulConnection> connectionType) {
+
+		if (connectionType.equals(StatefulRedisSentinelConnection.class)) {
+			return client.connectSentinel();
+		}
+
+		if (connectionType.equals(StatefulRedisPubSubConnection.class)) {
+			return client.connectPubSub(codec);
+		}
+
+		if (StatefulConnection.class.isAssignableFrom(connectionType)) {
+			return client.connect(codec);
+		}
+
+		throw new UnsupportedOperationException("Connection type " + connectionType + " not supported!");
+	}
+
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.redis.connection.lettuce.LettuceConnectionProvider.TargetAware#getConnection(java.lang.Class, io.lettuce.core.RedisURI)
+	 */
+	@SuppressWarnings("rawtypes")
+	@Override
+	public StatefulConnection<?, ?> getConnection(Class<? extends StatefulConnection> connectionType, RedisURI redisURI) {
+
+		if (connectionType.equals(StatefulRedisSentinelConnection.class)) {
+			return client.connectSentinel(redisURI);
+		}
+
+		if (connectionType.equals(StatefulRedisPubSubConnection.class)) {
+			return client.connectPubSub(codec, redisURI);
+		}
+
+		if (StatefulConnection.class.isAssignableFrom(connectionType)) {
+			return client.connect(codec, redisURI);
+		}
+
+		throw new UnsupportedOperationException("Connection type " + connectionType + " not supported!");
+	}
+}