Add option to disable Lettuce connection sharing

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

@@ -82,6 +82,7 @@ public class LettuceConnection implements RedisConnection {
 	private boolean isClosed = false;
 	private boolean isMulti = false;
 	private boolean isPipelined = false;
+	private boolean closeNativeConnection = true;
 	private List<Command<?, ?, ?>> ppline;
 	private RedisClient client;
 	private volatile LettuceSubscription subscription;
@@ -91,15 +92,31 @@ public class LettuceConnection implements RedisConnection {
 	 *
 	 * @param connection underlying Lettuce async connection
 	 * @param timeout the connection timeout (in milliseconds)
+	 * @param client The Lettuce client
 	 */
 	public LettuceConnection(com.lambdaworks.redis.RedisAsyncConnection<byte[], byte[]> connection, long timeout,
 			RedisClient client) {
+		this(connection, timeout, client, true);
+	}
+
+	/**
+	 * Instantiates a new lettuce connection.
+	 *
+	 * @param connection The underlying Lettuce async connection
+	 * @param timeout The connection timeout (in milliseconds)
+	 * @param client The Lettuce client
+	 * @param closeNativeConnection True if native connection should be called on {@link #close()}. This should
+	 * be set to false if the native connection is shared.
+	 */
+	public LettuceConnection(com.lambdaworks.redis.RedisAsyncConnection<byte[], byte[]> connection, long timeout,
+			RedisClient client, boolean closeNativeConnection) {
 		Assert.notNull(connection, "a valid connection is required");
 
 		this.asyncConn = connection;
 		this.timeout = timeout;
 		this.con = new com.lambdaworks.redis.RedisConnection<byte[], byte[]>(asyncConn);
 		this.client = client;
+		this.closeNativeConnection = closeNativeConnection;
 	}
 
 	protected DataAccessException convertLettuceAccessException(Exception ex) {
@@ -161,6 +178,14 @@ public class LettuceConnection implements RedisConnection {
 			txConn = null;
 		}
 
+		if(closeNativeConnection) {
+			try {
+				asyncConn.close();
+			} catch (RuntimeException ex) {
+				throw convertLettuceAccessException(ex);
+			}
+		}
+
 		if (subscription != null) {
 			if(subscription.isAlive()) {
 				subscription.doClose();

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

@@ -37,7 +37,7 @@ import com.lambdaworks.redis.RedisException;
  * <p>
  * This factory creates a new {@link LettuceConnection} on each call to
  * {@link #getConnection()}. Multiple {@link LettuceConnection}s share a single
- * thread-safe native connection.
+ * thread-safe native connection by default.
  *
  * <p>
  * The shared native connection is never closed by {@link LettuceConnection},
@@ -57,6 +57,7 @@ public class LettuceConnectionFactory implements InitializingBean, DisposableBea
 	private RedisClient client;
 	private long timeout = TimeUnit.MILLISECONDS.convert(60, TimeUnit.SECONDS);
 	private boolean validateConnection = false;
+	private boolean shareNativeConnection = true;
 	private RedisAsyncConnection<byte[], byte[]> connection;
 	private int dbIndex = 0;
 
@@ -79,8 +80,6 @@ public class LettuceConnectionFactory implements InitializingBean, DisposableBea
 	public void afterPropertiesSet() {
 		client = new RedisClient(hostName, port);
 		client.setDefaultTimeout(timeout, TimeUnit.MILLISECONDS);
-		// open a single connection to be shared for non-blocking and non-tx ops
-		initConnection();
 	}
 
 	public void destroy() {
@@ -88,15 +87,11 @@ public class LettuceConnectionFactory implements InitializingBean, DisposableBea
 	}
 
 	public RedisConnection getConnection() {
-		if (validateConnection) {
-			try {
-				new com.lambdaworks.redis.RedisConnection<byte[], byte[]>(connection).ping();
-			} catch (RedisException e) {
-				log.warn("Validation of shared connection failed. Creating a new connection.");
-				initConnection();
-			}
+		RedisAsyncConnection<byte[], byte[]> nativeConnection = getNativeConnection();
+		if (dbIndex > 0) {
+			nativeConnection.select(dbIndex);
 		}
-		return new LettuceConnection(connection, timeout, client);
+		return new LettuceConnection(nativeConnection, timeout, client, !shareNativeConnection);
 	}
 
 	public DataAccessException translateExceptionIfPossible(RuntimeException ex) {
@@ -175,12 +170,13 @@ public class LettuceConnectionFactory implements InitializingBean, DisposableBea
 	 * validation fails.
 	 * <p>
 	 * Lettuce will automatically reconnect until close is called, which should
-	 * never happen through {@link LettuceConnection}, therefore the default is
-	 * false.
+	 * never happen through {@link LettuceConnection} if a shared native
+	 * connection is used, therefore the default is false.
 	 * <p>
 	 * Setting this to true will result in a round-trip call to the server on
-	 * each new connection, so this setting should only be used if there is code
-	 * that is actively closing the native Lettuce connection.
+	 * each new connection, so this setting should only be used if connection
+	 * sharing is enabled and there is code that is actively closing the native
+	 * Lettuce connection.
 	 *
 	 * @param validateConnection
 	 *            enable connection validation
@@ -189,6 +185,28 @@ public class LettuceConnectionFactory implements InitializingBean, DisposableBea
 		this.validateConnection = validateConnection;
 	}
 
+	/**
+	 * Indicates if multiple {@link LettuceConnection}s should share a single
+	 * native connection.
+	 *
+	 * @return native connection shared
+	 */
+	public boolean getShareNativeConnection() {
+		return shareNativeConnection;
+	}
+
+	/**
+	 * Enables multiple {@link LettuceConnection}s to share a single native
+	 * connection. If set to false, every operation on {@link LettuceConnection}
+	 * will open and close a socket.
+	 *
+	 * @param shareNativeConnection
+	 *            enable connection sharing
+	 */
+	public void setShareNativeConnection(boolean shareNativeConnection) {
+		this.shareNativeConnection = shareNativeConnection;
+	}
+
 	/**
 	 * Returns the index of the database.
 	 *
@@ -210,10 +228,22 @@ public class LettuceConnectionFactory implements InitializingBean, DisposableBea
 		this.dbIndex = index;
 	}
 
-	protected void initConnection() {
-		connection = client.connectAsync(LettuceUtils.CODEC);
-		if (dbIndex > 0) {
-			connection.select(dbIndex);
+	protected RedisAsyncConnection<byte[], byte[]> getNativeConnection() {
+		if(shareNativeConnection) {
+			if(connection == null) {
+				connection = client.connectAsync(LettuceUtils.CODEC);
+			}
+			if (validateConnection) {
+				try {
+					new com.lambdaworks.redis.RedisConnection<byte[], byte[]>(connection).ping();
+				} catch (RedisException e) {
+					log.warn("Validation of shared connection failed. Creating a new connection.");
+					connection = client.connectAsync(LettuceUtils.CODEC);
+				}
+			}
+			return connection;
+		}else {
+			return client.connectAsync(LettuceUtils.CODEC);
 		}
 	}
 }