DATAREDIS-667 - Align Lettuce connection-pooling with LettuceClientConfiguration.

We now support Lettuce connection pooling by configuring a specialized client configuration via LettucePoolingClientConfiguration.builder(). Pooling settings can be configured through the builder and used with LettuceConnectionFactory.

Connection pooling is supported with Standalone and Sentinel-managed connections.

LettucePoolingClientConfiguration.builder()
   .poolConfig(poolConfig)
   .and() // allows configuration of further settings.

Original Pull Request: #262

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

@@ -57,6 +57,14 @@ class DefaultLettuceClientConfiguration implements LettuceClientConfiguration {
 		return useSsl;
 	}
 
+	/* (non-Javadoc)
+	 * @see org.springframework.data.redis.connection.lettuce.LettuceClientConfiguration#isUsePooling()
+	 */
+	@Override
+	public boolean isUsePooling() {
+		return false;
+	}
+
 	/* (non-Javadoc)
 	 * @see org.springframework.data.redis.connection.lettuce.LettuceClientConfiguration#isVerifyPeer()
 	 */

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

@@ -0,0 +1,61 @@
+/*
+ * 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.ClientOptions;
+import io.lettuce.core.resource.ClientResources;
+
+import java.time.Duration;
+
+import org.apache.commons.pool2.impl.GenericObjectPoolConfig;
+
+/**
+ * Default implementation of {@literal LettucePoolingClientConfiguration}.
+ *
+ * @author Mark Paluch
+ * @since 2.0
+ */
+class DefaultLettucePoolingClientConfiguration extends DefaultLettuceClientConfiguration
+		implements LettucePoolingClientConfiguration {
+
+	private final GenericObjectPoolConfig poolConfig;
+
+	DefaultLettucePoolingClientConfiguration(boolean useSsl, boolean verifyPeer, boolean startTls,
+			ClientResources clientResources, ClientOptions clientOptions, Duration timeout, Duration shutdownTimeout,
+			GenericObjectPoolConfig poolConfig) {
+
+		super(useSsl, verifyPeer, startTls, clientResources, clientOptions, timeout, shutdownTimeout);
+
+		this.poolConfig = poolConfig;
+	}
+
+	/* (non-Javadoc)
+	 * @see org.springframework.data.redis.connection.lettuce.LettuceClientConfiguration#isUsePooling()
+	 */
+	@Override
+	public boolean isUsePooling() {
+		return true;
+	}
+
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.redis.connection.lettuce.LettucePoolingClientConfiguration#getPoolConfig()
+	 */
+	@Override
+	public GenericObjectPoolConfig getPoolConfig() {
+		return poolConfig;
+	}
+}

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

@@ -54,6 +54,11 @@ public interface LettuceClientConfiguration {
 	 */
 	boolean isUseSsl();
 
+	/**
+	 * @return {@literal true} to use connection-pooling.
+	 */
+	boolean isUsePooling();
+
 	/**
 	 * @return {@literal true} to verify peers when using {@link #isUseSsl() SSL}.
 	 */
@@ -217,15 +222,15 @@ public interface LettuceClientConfiguration {
 	class DefaultLettuceClientConfigurationBuilder
 			implements LettuceClientConfigurationBuilder, LettuceSslClientConfigurationBuilder {
 
-		private boolean useSsl;
-		private boolean verifyPeer = true;
-		private boolean startTls;
-		private ClientResources clientResources;
-		private ClientOptions clientOptions;
-		private Duration timeout = Duration.ofSeconds(RedisURI.DEFAULT_TIMEOUT);
-		private Duration shutdownTimeout = Duration.ofMillis(100);
+		boolean useSsl;
+		boolean verifyPeer = true;
+		boolean startTls;
+		ClientResources clientResources;
+		ClientOptions clientOptions;
+		Duration timeout = Duration.ofSeconds(RedisURI.DEFAULT_TIMEOUT);
+		Duration shutdownTimeout = Duration.ofMillis(100);
 
-		private DefaultLettuceClientConfigurationBuilder() {}
+		DefaultLettuceClientConfigurationBuilder() {}
 
 		/*
 		 * (non-Javadoc)

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

@@ -38,7 +38,6 @@ import org.springframework.beans.factory.DisposableBean;
 import org.springframework.beans.factory.InitializingBean;
 import org.springframework.dao.DataAccessException;
 import org.springframework.dao.InvalidDataAccessApiUsageException;
-import org.springframework.dao.InvalidDataAccessResourceUsageException;
 import org.springframework.data.redis.ExceptionTranslationStrategy;
 import org.springframework.data.redis.PassThroughExceptionTranslationStrategy;
 import org.springframework.data.redis.RedisConnectionFailureException;
@@ -281,8 +280,8 @@ public class LettuceConnectionFactory
 		}
 
 		LettuceConnection connection;
-		if (pool != null) {
 
+		if (pool != null) {
 			connection = new LettuceConnection(getSharedConnection(), getTimeout(), null, pool, getDatabase());
 		} else {
 			connection = new LettuceConnection(getSharedConnection(), connectionProvider, getTimeout(), getDatabase());
@@ -779,6 +778,16 @@ public class LettuceConnectionFactory
 
 	private LettuceConnectionProvider createConnectionProvider(AbstractRedisClient client, RedisCodec<?, ?> codec) {
 
+		LettuceConnectionProvider connectionProvider = doConnectionProvider(client, codec);
+
+		if (this.clientConfiguration.isUsePooling()) {
+			return new LettucePoolingConnectionProvider(connectionProvider, this.clientConfiguration);
+		}
+
+		return connectionProvider;
+	}
+
+	private LettuceConnectionProvider doConnectionProvider(AbstractRedisClient client, RedisCodec<?, ?> codec) {
 		if (isClusterAware()) {
 			return new ClusterConnectionProvider((RedisClusterClient) client, codec);
 		}
@@ -852,11 +861,6 @@ public class LettuceConnectionFactory
 
 	@Override
 	public RedisSentinelConnection getSentinelConnection() {
-
-		if (!(connectionProvider instanceof StandaloneConnectionProvider)) {
-			throw new InvalidDataAccessResourceUsageException(
-					"Unable to connect to sentinels using " + connectionProvider.getClass());
-		}
 		return new LettuceSentinelConnection(connectionProvider);
 	}
 
@@ -895,6 +899,14 @@ public class LettuceConnectionFactory
 			return useSsl;
 		}
 
+		/* (non-Javadoc)
+		 * @see org.springframework.data.redis.connection.lettuce.LettuceClientConfiguration#isUsePooling()
+		 */
+		@Override
+		public boolean isUsePooling() {
+			return false;
+		}
+
 		public void setUseSsl(boolean useSsl) {
 			this.useSsl = useSsl;
 		}

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

@@ -0,0 +1,124 @@
+/*
+ * 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 org.apache.commons.pool2.impl.GenericObjectPoolConfig;
+import org.springframework.data.redis.connection.jedis.JedisClientConfiguration;
+import org.springframework.util.Assert;
+
+/**
+ * @author Mark Paluch
+ */
+public interface LettucePoolingClientConfiguration extends LettuceClientConfiguration {
+
+	/**
+	 * Creates a new {@link LettucePoolingClientConfigurationBuilder} to build {@link LettucePoolingClientConfiguration}
+	 * to be used with the jedis client.
+	 *
+	 * @return a new {@link LettucePoolingClientConfigurationBuilder} to build {@link LettucePoolingClientConfiguration}.
+	 */
+	static LettucePoolingClientConfigurationBuilder builder() {
+		return new DefaultLettucePoolingClientConfigurationBuilder();
+	}
+
+	/**
+	 * Creates a default {@link LettucePoolingClientConfiguration} with:
+	 * <dl>
+	 * <dt>SSL</dt>
+	 * <dd>no</dd>
+	 * <dt>Pooling</dt>
+	 * <dd>yes</dd>
+	 * <dt>Peer Verification</dt>
+	 * <dd>yes</dd>
+	 * <dt>Start TLS</dt>
+	 * <dd>no</dd>
+	 * <dt>Client Options</dt>
+	 * <dd>none</dd>
+	 * <dt>Client Resources</dt>
+	 * <dd>none</dd>
+	 * <dt>Connect Timeout</dt>
+	 * <dd>60 Seconds</dd>
+	 * <dt>Shutdown Timeout</dt>
+	 * <dd>2 Seconds</dd>
+	 * </dl>
+	 *
+	 * @return a {@link LettucePoolingClientConfiguration} with defaults.
+	 */
+	static LettucePoolingClientConfiguration defaultConfiguration() {
+		return builder().build();
+	}
+
+	/**
+	 * @return the optional {@link GenericObjectPoolConfig}.
+	 */
+	GenericObjectPoolConfig getPoolConfig();
+
+	/**
+	 * Builder for Pooling-related {@link LettuceClientConfiguration}.
+	 */
+	interface LettucePoolingClientConfigurationBuilder {
+
+		/**
+		 * @param poolConfig must not be {@literal null}.
+		 * @return {@literal this} builder.
+		 * @throws IllegalArgumentException if poolConfig is {@literal null}.
+		 */
+		LettucePoolingClientConfigurationBuilder poolConfig(GenericObjectPoolConfig poolConfig);
+
+		/**
+		 * Return to {@link LettuceClientConfigurationBuilder}.
+		 *
+		 * @return {@link LettuceClientConfigurationBuilder}.
+		 */
+		LettuceClientConfigurationBuilder and();
+
+		/**
+		 * Build the {@link JedisClientConfiguration} with the configuration applied from this builder.
+		 *
+		 * @return a new {@link JedisClientConfiguration} object.
+		 */
+		LettucePoolingClientConfiguration build();
+	}
+
+	/**
+	 * Default {@link LettuceClientConfigurationBuilder} implementation to build an immutable
+	 * {@link LettuceClientConfiguration}.
+	 */
+	class DefaultLettucePoolingClientConfigurationBuilder extends DefaultLettuceClientConfigurationBuilder
+			implements LettucePoolingClientConfigurationBuilder {
+
+		GenericObjectPoolConfig poolConfig = new GenericObjectPoolConfig();
+
+		@Override
+		public LettucePoolingClientConfigurationBuilder poolConfig(GenericObjectPoolConfig poolConfig) {
+
+			Assert.notNull(poolConfig, "GenericObjectPoolConfig must not be null!");
+
+			this.poolConfig = poolConfig;
+			return this;
+		}
+
+		/*
+		 * (non-Javadoc)
+		 * @see org.springframework.data.redis.connection.lettuce.LettuceClientConfiguration.LettuceClientConfigurationBuilder#build()
+		 */
+		@Override
+		public LettucePoolingClientConfiguration build() {
+			return new DefaultLettucePoolingClientConfiguration(useSsl, verifyPeer, startTls, clientResources, clientOptions,
+					timeout, shutdownTimeout, poolConfig);
+		}
+	}
+}

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

@@ -0,0 +1,119 @@
+/*
+ * 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.support.ConnectionPoolSupport;
+
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.apache.commons.pool2.impl.GenericObjectPool;
+import org.apache.commons.pool2.impl.GenericObjectPoolConfig;
+import org.springframework.beans.factory.DisposableBean;
+import org.springframework.data.redis.connection.PoolException;
+
+/**
+ * {@link LettuceConnectionProvider} with connection pooling support. This connection provider holds multiple pools (one
+ * per connection type) for contextualized connection allocation.
+ * <p />
+ * Each allocated connection is tracked and to be returned into the pool which created the connection. Instances of this
+ * class require {@link #destroy() disposal} to de-allocate lingering connections that were not returned to the pool and
+ * to close the pools.
+ *
+ * @author Mark Paluch
+ * @since 2.0
+ * @see #getConnection(Class)
+ */
+class LettucePoolingConnectionProvider implements LettuceConnectionProvider, DisposableBean {
+
+	private final static Log log = LogFactory.getLog(LettucePoolingConnectionProvider.class);
+
+	private final LettuceConnectionProvider connectionProvider;
+	private final GenericObjectPoolConfig poolConfig;
+	private final Map<StatefulConnection<?, ?>, GenericObjectPool<StatefulConnection<?, ?>>> poolRef = new ConcurrentHashMap<>(
+			32);
+	private final Map<Class<?>, GenericObjectPool<StatefulConnection<?, ?>>> pools = new ConcurrentHashMap<>(32);
+
+	LettucePoolingConnectionProvider(LettuceConnectionProvider connectionProvider,
+			LettuceClientConfiguration clientConfiguration) {
+
+		this.connectionProvider = connectionProvider;
+		this.poolConfig = ((LettucePoolingClientConfiguration) clientConfiguration).getPoolConfig();
+	}
+
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.redis.connection.lettuce.LettuceConnectionProvider#getConnection(java.lang.Class)
+	 */
+	@Override
+	@SuppressWarnings("rawtypes")
+	public StatefulConnection<?, ?> getConnection(Class<? extends StatefulConnection> connectionType) {
+
+		GenericObjectPool<StatefulConnection<?, ?>> pool = pools.computeIfAbsent(connectionType, poolType -> {
+			return ConnectionPoolSupport.createGenericObjectPool(() -> connectionProvider.getConnection(connectionType),
+					poolConfig, false);
+		});
+
+		try {
+			StatefulConnection<?, ?> connection = pool.borrowObject();
+
+			poolRef.put(connection, pool);
+
+			return connection;
+		} catch (Exception e) {
+			throw new PoolException("Could not get a resource from the pool", e);
+		}
+	}
+
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.redis.connection.lettuce.LettuceConnectionProvider#release(io.lettuce.core.api.StatefulConnection)
+	 */
+	@Override
+	public void release(StatefulConnection<?, ?> connection) {
+
+		GenericObjectPool<StatefulConnection<?, ?>> pool = poolRef.remove(connection);
+
+		if (pool == null) {
+			throw new PoolException("Returned connection " + connection
+					+ " was either previously returned or does not belong to this connection provider");
+		}
+
+		pool.returnObject(connection);
+	}
+
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.beans.factory.DisposableBean#destroy()
+	 */
+	@Override
+	public void destroy() throws Exception {
+
+		if (!poolRef.isEmpty()) {
+
+			log.warn("LettucePoolingConnectionProvider contains unreleased connections");
+
+			poolRef.forEach((connection, pool) -> pool.returnObject(connection));
+			poolRef.clear();
+		}
+
+		pools.forEach((type, pool) -> pool.close());
+		pools.clear();
+	}
+}

src/test/java/org/springframework/data/redis/connection/lettuce/LettuceClientConfigurationUnitTests.java

@@ -31,11 +31,12 @@ import org.junit.Test;
  */
 public class LettuceClientConfigurationUnitTests {
 
-	@Test // DATAREDIS-574
+	@Test // DATAREDIS-574, DATAREDIS-667
 	public void shouldCreateEmptyConfiguration() {
 
 		LettuceClientConfiguration configuration = LettuceClientConfiguration.defaultConfiguration();
 
+		assertThat(configuration.isUsePooling()).isFalse();
 		assertThat(configuration.isUseSsl()).isFalse();
 		assertThat(configuration.isVerifyPeer()).isTrue();
 		assertThat(configuration.isStartTls()).isFalse();
@@ -45,7 +46,7 @@ public class LettuceClientConfigurationUnitTests {
 		assertThat(configuration.getShutdownTimeout()).isEqualTo(Duration.ofMillis(100));
 	}
 
-	@Test // DATAREDIS-574
+	@Test // DATAREDIS-574, DATAREDIS-667
 	public void shouldConfigureAllProperties() {
 
 		ClientOptions clientOptions = ClientOptions.create();
@@ -61,6 +62,7 @@ public class LettuceClientConfigurationUnitTests {
 				.shutdownTimeout(Duration.ofHours(2)) //
 				.build();
 
+		assertThat(configuration.isUsePooling()).isFalse();
 		assertThat(configuration.isUseSsl()).isTrue();
 		assertThat(configuration.isVerifyPeer()).isFalse();
 		assertThat(configuration.isStartTls()).isTrue();

src/test/java/org/springframework/data/redis/connection/lettuce/LettuceConnectionFactoryTests.java

@@ -24,6 +24,9 @@ import io.lettuce.core.RedisException;
 import io.lettuce.core.api.async.RedisAsyncCommands;
 import io.lettuce.core.api.reactive.BaseRedisReactiveCommands;
 
+import java.time.Duration;
+
+import org.apache.commons.pool2.impl.GenericObjectPoolConfig;
 import org.junit.After;
 import org.junit.AfterClass;
 import org.junit.Before;
@@ -35,6 +38,7 @@ import org.springframework.data.redis.RedisSystemException;
 import org.springframework.data.redis.SettingsUtils;
 import org.springframework.data.redis.connection.DefaultStringRedisConnection;
 import org.springframework.data.redis.connection.RedisConnection;
+import org.springframework.data.redis.connection.RedisStandaloneConfiguration;
 import org.springframework.data.redis.connection.StringRedisConnection;
 
 /**
@@ -328,4 +332,33 @@ public class LettuceConnectionFactoryTests {
 
 		assertThat(factory.getReactiveConnection().execute(BaseRedisReactiveCommands::ping).blockFirst(), is("PONG"));
 	}
+
+	@Test // DATAREDIS-667
+	public void factoryCreatesPooledConnections() {
+
+		GenericObjectPoolConfig poolConfig = new GenericObjectPoolConfig();
+
+		LettuceClientConfiguration configuration = LettucePoolingClientConfiguration.builder().poolConfig(poolConfig).and()
+				.clientResources(LettuceTestClientResources.getSharedClientResources()).shutdownTimeout(Duration.ZERO).build();
+
+		LettuceConnectionFactory factory = new LettuceConnectionFactory(new RedisStandaloneConfiguration(), configuration);
+		factory.setShareNativeConnection(false);
+		factory.afterPropertiesSet();
+
+		ConnectionFactoryTracker.add(factory);
+
+		RedisConnection initial = factory.getConnection();
+		Object initialNativeConnection = initial.getNativeConnection();
+
+		initial.close();
+
+		RedisConnection subsequent = factory.getConnection();
+		Object subsequentNativeConnection = subsequent.getNativeConnection();
+
+		subsequent.close();
+
+		assertThat(initialNativeConnection, is(subsequentNativeConnection));
+
+		factory.destroy();
+	}
 }

src/test/java/org/springframework/data/redis/connection/lettuce/LettucePoolingClientConfigurationUnitTests.java

@@ -0,0 +1,80 @@
+/*
+ * 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 io.lettuce.core.ClientOptions;
+import io.lettuce.core.resource.ClientResources;
+
+import java.time.Duration;
+
+import org.apache.commons.pool2.impl.GenericObjectPoolConfig;
+import org.junit.Test;
+
+/**
+ * Unit tests for {@link LettucePoolingClientConfiguration}.
+ *
+ * @author Mark Paluch
+ */
+public class LettucePoolingClientConfigurationUnitTests {
+
+	@Test // DATAREDIS-667
+	public void shouldCreateEmptyConfiguration() {
+
+		LettucePoolingClientConfiguration configuration = LettucePoolingClientConfiguration.defaultConfiguration();
+
+		assertThat(configuration.isUsePooling()).isTrue();
+		assertThat(configuration.getPoolConfig()).isNotNull();
+		assertThat(configuration.isUseSsl()).isFalse();
+		assertThat(configuration.isVerifyPeer()).isTrue();
+		assertThat(configuration.isStartTls()).isFalse();
+		assertThat(configuration.getClientOptions()).isEmpty();
+		assertThat(configuration.getClientResources()).isEmpty();
+		assertThat(configuration.getCommandTimeout()).isEqualTo(Duration.ofSeconds(60));
+		assertThat(configuration.getShutdownTimeout()).isEqualTo(Duration.ofMillis(100));
+	}
+
+	@Test // DATAREDIS-667
+	public void shouldConfigureAllProperties() {
+
+		ClientOptions clientOptions = ClientOptions.create();
+		ClientResources sharedClientResources = LettuceTestClientResources.getSharedClientResources();
+		GenericObjectPoolConfig poolConfig = new GenericObjectPoolConfig();
+
+		LettucePoolingClientConfiguration configuration = (LettucePoolingClientConfiguration) LettucePoolingClientConfiguration
+				.builder() //
+				.poolConfig(poolConfig).and() //
+				.useSsl() //
+				.disablePeerVerification() //
+				.startTls().and() //
+				.clientOptions(clientOptions) //
+				.clientResources(sharedClientResources) //
+				.commandTimeout(Duration.ofMinutes(5)) //
+				.shutdownTimeout(Duration.ofHours(2)) //
+				.build();
+
+		assertThat(configuration.isUsePooling()).isTrue();
+		assertThat(configuration.getPoolConfig()).isEqualTo(poolConfig);
+		assertThat(configuration.isUseSsl()).isTrue();
+		assertThat(configuration.isVerifyPeer()).isFalse();
+		assertThat(configuration.isStartTls()).isTrue();
+		assertThat(configuration.getClientOptions()).contains(clientOptions);
+		assertThat(configuration.getClientResources()).contains(sharedClientResources);
+		assertThat(configuration.getCommandTimeout()).isEqualTo(Duration.ofMinutes(5));
+		assertThat(configuration.getShutdownTimeout()).isEqualTo(Duration.ofHours(2));
+	}
+}

src/test/java/org/springframework/data/redis/connection/lettuce/LettuceReactiveCommandsTestsBase.java

@@ -15,19 +15,16 @@
  */
 package org.springframework.data.redis.connection.lettuce;
 
-import static org.hamcrest.core.Is.*;
-import static org.junit.Assume.*;
-
-import io.lettuce.core.AbstractRedisClient;
-import io.lettuce.core.RedisClient;
+import io.lettuce.core.api.StatefulRedisConnection;
 import io.lettuce.core.api.sync.RedisCommands;
-import io.lettuce.core.cluster.RedisClusterClient;
+import io.lettuce.core.cluster.api.StatefulRedisClusterConnection;
 import io.lettuce.core.cluster.api.sync.RedisAdvancedClusterCommands;
 import io.lettuce.core.cluster.api.sync.RedisClusterCommands;
+import io.lettuce.core.codec.StringCodec;
 
 import java.nio.ByteBuffer;
 import java.nio.charset.Charset;
-import java.util.Arrays;
+import java.util.ArrayList;
 import java.util.List;
 
 import org.junit.After;
@@ -76,40 +73,55 @@ public abstract class LettuceReactiveCommandsTestsBase {
 	static final ByteBuffer SAME_SLOT_KEY_3_BBUFFER = ByteBuffer.wrap(SAME_SLOT_KEY_3_BYTES);
 	static final ByteBuffer VALUE_3_BBUFFER = ByteBuffer.wrap(VALUE_3_BYTES);
 
-	@Parameterized.Parameter(value = 0) public Object clientProvider;
-	@Parameterized.Parameter(value = 1) public Object displayName;
+	@Parameterized.Parameter(value = 0) public LettuceConnectionProvider connectionProvider;
+	@Parameterized.Parameter(value = 1) public LettuceConnectionProvider nativeConnectionProvider;
+	@Parameterized.Parameter(value = 2) public Object displayName;
 
 	LettuceReactiveRedisConnection connection;
 	RedisClusterCommands<String, String> nativeCommands;
 
-	@Parameterized.Parameters(name = "{1}")
+	@Parameterized.Parameters(name = "{2}")
 	public static List<Object[]> parameters() {
-		return Arrays.asList(new Object[] { LettuceRedisClientProvider.local(), "Standalone" },
-				new Object[] { LettuceRedisClusterClientProvider.local(), "Cluster" });
+
+		LettuceRedisClientProvider standalone = LettuceRedisClientProvider.local();
+		LettuceRedisClusterClientProvider cluster = LettuceRedisClusterClientProvider.local();
+
+		List<Object[]> parameters = new ArrayList<>();
+
+		StandaloneConnectionProvider standaloneProvider = new StandaloneConnectionProvider(standalone.getClient(),
+				LettuceReactiveRedisConnection.CODEC);
+		StandaloneConnectionProvider nativeConnectionProvider = new StandaloneConnectionProvider(standalone.getClient(),
+				StringCodec.UTF8);
+
+		parameters.add(new Object[] { standaloneProvider, nativeConnectionProvider, "Standalone" });
+		parameters.add(new Object[] {
+				new LettucePoolingConnectionProvider(standaloneProvider, LettucePoolingClientConfiguration.builder().build()),
+				nativeConnectionProvider, "Standalone/Pooled" });
+
+		if (cluster.test()) {
+
+			ClusterConnectionProvider clusterProvider = new ClusterConnectionProvider(cluster.getClient(),
+					LettuceReactiveRedisConnection.CODEC);
+			ClusterConnectionProvider nativeClusterConnectionProvider = new ClusterConnectionProvider(cluster.getClient(),
+					StringCodec.UTF8);
+
+			parameters.add(new Object[] { clusterProvider, nativeClusterConnectionProvider, "Cluster" });
+		}
+
+		return parameters;
 	}
 
 	@Before
 	public void setUp() {
 
-		AbstractRedisClient abstractRedisClient = null;
-		if (clientProvider instanceof LettuceRedisClientProvider) {
-			abstractRedisClient = ((LettuceRedisClientProvider) clientProvider).getClient();
-		} else if (clientProvider instanceof LettuceRedisClusterClientProvider) {
-			abstractRedisClient = ((LettuceRedisClusterClientProvider) clientProvider).getClient();
-			assumeThat(((LettuceRedisClusterClientProvider) clientProvider).test(), is(true));
+		if (nativeConnectionProvider instanceof StandaloneConnectionProvider) {
+			nativeCommands = ((StatefulRedisConnection) nativeConnectionProvider.getConnection()).sync();
+			this.connection = new LettuceReactiveRedisConnection(connectionProvider);
+
+		} else {
+			nativeCommands = ((StatefulRedisClusterConnection) nativeConnectionProvider.getConnection()).sync();
+			this.connection = new LettuceReactiveRedisClusterConnection(connectionProvider);
 		}
-
-		if (abstractRedisClient instanceof RedisClient) {
-			nativeCommands = ((RedisClient) abstractRedisClient).connect().sync();
-			connection = new LettuceReactiveRedisConnection(
-					new StandaloneConnectionProvider(((RedisClient) abstractRedisClient), LettuceReactiveRedisConnection.CODEC));
-
-		} else if (abstractRedisClient instanceof RedisClusterClient) {
-			nativeCommands = ((RedisClusterClient) abstractRedisClient).connect().sync();
-			connection = new LettuceReactiveRedisClusterConnection(new ClusterConnectionProvider(
-					((RedisClusterClient) abstractRedisClient), LettuceReactiveRedisConnection.CODEC));
-		}
-
 	}
 
 	@After
@@ -119,11 +131,11 @@ public abstract class LettuceReactiveCommandsTestsBase {
 			flushAll();
 
 			if (nativeCommands instanceof RedisCommands) {
-				((RedisCommands) nativeCommands).getStatefulConnection().close();
+				nativeConnectionProvider.release(((RedisCommands) nativeCommands).getStatefulConnection());
 			}
 
 			if (nativeCommands instanceof RedisAdvancedClusterCommands) {
-				((RedisAdvancedClusterCommands) nativeCommands).getStatefulConnection().close();
+				nativeConnectionProvider.release(((RedisAdvancedClusterCommands) nativeCommands).getStatefulConnection());
 			}
 		}
 

src/test/java/org/springframework/data/redis/connection/lettuce/LettuceReactiveHyperLogLogCommandsTests.java

@@ -26,6 +26,7 @@ import org.springframework.data.redis.test.util.LettuceRedisClientProvider;
 
 /**
  * @author Christoph Strobl
+ * @author Mark Paluch
  */
 public class LettuceReactiveHyperLogLogCommandsTests extends LettuceReactiveCommandsTestsBase {
 
@@ -56,7 +57,7 @@ public class LettuceReactiveHyperLogLogCommandsTests extends LettuceReactiveComm
 	@Test // DATAREDIS-525
 	public void pfCountWithMultipleKeysShouldReturnCorrectly() {
 
-		assumeThat(clientProvider instanceof LettuceRedisClientProvider, is(true));
+		assumeThat(connectionProvider instanceof StandaloneConnectionProvider, is(true));
 
 		nativeCommands.pfadd(KEY_1, new String[] { VALUE_1, VALUE_2 });
 		nativeCommands.pfadd(KEY_2, new String[] { VALUE_2, VALUE_3 });
@@ -67,7 +68,7 @@ public class LettuceReactiveHyperLogLogCommandsTests extends LettuceReactiveComm
 	@Test // DATAREDIS-525
 	public void pfMergeShouldWorkCorrectly() {
 
-		assumeThat(clientProvider instanceof LettuceRedisClientProvider, is(true));
+		assumeThat(connectionProvider instanceof StandaloneConnectionProvider, is(true));
 
 		nativeCommands.pfadd(KEY_1, new String[] { VALUE_1, VALUE_2 });
 		nativeCommands.pfadd(KEY_2, new String[] { VALUE_2, VALUE_3 });
@@ -78,5 +79,4 @@ public class LettuceReactiveHyperLogLogCommandsTests extends LettuceReactiveComm
 
 		assertThat(nativeCommands.pfcount(new String[] { KEY_3 }), is(3L));
 	}
-
 }

src/test/java/org/springframework/data/redis/connection/lettuce/LettuceReactiveListCommandTests.java

@@ -38,6 +38,7 @@ import reactor.core.publisher.Mono;
 
 /**
  * @author Christoph Strobl
+ * @author Mark Paluch
  */
 public class LettuceReactiveListCommandTests extends LettuceReactiveCommandsTestsBase {
 
@@ -203,7 +204,7 @@ public class LettuceReactiveListCommandTests extends LettuceReactiveCommandsTest
 	@Test // DATAREDIS-525
 	public void blPopShouldReturnFirstAvailable() {
 
-		assumeThat(clientProvider instanceof LettuceRedisClientProvider, is(true));
+		assumeThat(connectionProvider instanceof StandaloneConnectionProvider, is(true));
 
 		nativeCommands.rpush(KEY_1, VALUE_1, VALUE_2, VALUE_3);
 
@@ -216,7 +217,7 @@ public class LettuceReactiveListCommandTests extends LettuceReactiveCommandsTest
 	@Test // DATAREDIS-525
 	public void brPopShouldReturnLastAvailable() {
 
-		assumeThat(clientProvider instanceof LettuceRedisClientProvider, is(true));
+		assumeThat(connectionProvider instanceof StandaloneConnectionProvider, is(true));
 
 		nativeCommands.rpush(KEY_1, VALUE_1, VALUE_2, VALUE_3);
 
@@ -242,7 +243,7 @@ public class LettuceReactiveListCommandTests extends LettuceReactiveCommandsTest
 	@Test // DATAREDIS-525
 	public void brPopLPushShouldWorkCorrectly() {
 
-		assumeThat(clientProvider instanceof LettuceRedisClientProvider, is(true));
+		assumeThat(connectionProvider instanceof StandaloneConnectionProvider, is(true));
 
 		nativeCommands.rpush(KEY_1, VALUE_1, VALUE_2, VALUE_3);
 		nativeCommands.rpush(KEY_2, VALUE_1);

src/test/java/org/springframework/data/redis/connection/lettuce/LettuceReactiveServerCommandsTests.java

@@ -38,8 +38,11 @@ public class LettuceReactiveServerCommandsTests extends LettuceReactiveCommandsT
 		StepVerifier.create(connection.serverCommands().bgReWriteAof()).expectNextCount(1).verifyComplete();
 	}
 
-	@Test // DATAREDIS-659
+	@Test // DATAREDIS-659, DATAREDIS-667
 	public void bgSaveShouldRespondCorrectly() {
+
+		assumeTrue(connectionProvider instanceof StandaloneConnectionProvider);
+
 		StepVerifier.create(connection.serverCommands().bgSave()).expectNextCount(1).verifyComplete();
 	}
 
@@ -48,8 +51,11 @@ public class LettuceReactiveServerCommandsTests extends LettuceReactiveCommandsT
 		StepVerifier.create(connection.serverCommands().lastSave()).expectNextCount(1).verifyComplete();
 	}
 
-	@Test // DATAREDIS-659
+	@Test // DATAREDIS-659, DATAREDIS-667
 	public void saveShouldRespondCorrectly() {
+
+		assumeTrue(connectionProvider instanceof StandaloneConnectionProvider);
+
 		StepVerifier.create(connection.serverCommands().save()).expectNext("OK").verifyComplete();
 	}
 
@@ -61,9 +67,8 @@ public class LettuceReactiveServerCommandsTests extends LettuceReactiveCommandsT
 	@Test // DATAREDIS-659
 	public void flushDbShouldRespondCorrectly() {
 
-		StepVerifier
-				.create(connection.serverCommands().flushDb() //
-						.then(connection.stringCommands().set(KEY_1_BBUFFER, VALUE_1_BBUFFER))) //
+		StepVerifier.create(connection.serverCommands().flushDb() //
+				.then(connection.stringCommands().set(KEY_1_BBUFFER, VALUE_1_BBUFFER))) //
 				.expectNextCount(1) //
 				.verifyComplete();
 
@@ -77,9 +82,8 @@ public class LettuceReactiveServerCommandsTests extends LettuceReactiveCommandsT
 	@Test // DATAREDIS-659
 	public void flushAllShouldRespondCorrectly() {
 
-		StepVerifier
-				.create(connection.serverCommands().flushAll() //
-						.then(connection.stringCommands().set(KEY_1_BBUFFER, VALUE_1_BBUFFER))) //
+		StepVerifier.create(connection.serverCommands().flushAll() //
+				.then(connection.stringCommands().set(KEY_1_BBUFFER, VALUE_1_BBUFFER))) //
 				.expectNextCount(1) //
 				.verifyComplete();
 

src/test/java/org/springframework/data/redis/connection/lettuce/LettuceReactiveStringCommandsTests.java

@@ -236,7 +236,7 @@ public class LettuceReactiveStringCommandsTests extends LettuceReactiveCommandsT
 	@Test // DATAREDIS-525
 	public void mSetNXShouldAddMultipleKeyValuePairs() {
 
-		assumeThat(clientProvider instanceof LettuceRedisClientProvider, is(true));
+		assumeThat(connectionProvider instanceof StandaloneConnectionProvider, is(true));
 
 		Map<ByteBuffer, ByteBuffer> map = new LinkedHashMap<>();
 		map.put(KEY_1_BBUFFER, VALUE_1_BBUFFER);
@@ -251,7 +251,7 @@ public class LettuceReactiveStringCommandsTests extends LettuceReactiveCommandsT
 	@Test // DATAREDIS-525
 	public void mSetNXShouldNotAddMultipleKeyValuePairsWhenAlreadyExit() {
 
-		assumeThat(clientProvider instanceof LettuceRedisClientProvider, is(true));
+		assumeThat(connectionProvider instanceof StandaloneConnectionProvider, is(true));
 
 		nativeCommands.set(KEY_2, VALUE_2);
 
@@ -346,7 +346,7 @@ public class LettuceReactiveStringCommandsTests extends LettuceReactiveCommandsT
 	@Test // DATAREDIS-525
 	public void bitOpAndShouldWorkAsExpected() {
 
-		assumeThat(clientProvider instanceof LettuceRedisClientProvider, is(true));
+		assumeThat(connectionProvider instanceof StandaloneConnectionProvider, is(true));
 
 		nativeCommands.set(KEY_1, VALUE_1);
 		nativeCommands.set(KEY_2, VALUE_2);
@@ -363,7 +363,7 @@ public class LettuceReactiveStringCommandsTests extends LettuceReactiveCommandsT
 	@Test // DATAREDIS-525
 	public void bitOpOrShouldWorkAsExpected() {
 
-		assumeThat(clientProvider instanceof LettuceRedisClientProvider, is(true));
+		assumeThat(connectionProvider instanceof StandaloneConnectionProvider, is(true));
 
 		nativeCommands.set(KEY_1, VALUE_1);
 		nativeCommands.set(KEY_2, VALUE_2);
@@ -380,7 +380,7 @@ public class LettuceReactiveStringCommandsTests extends LettuceReactiveCommandsT
 	@Test // DATAREDIS-525
 	public void bitNotShouldThrowExceptionWhenMoreThanOnSourceKey() {
 
-		assumeThat(clientProvider instanceof LettuceRedisClientProvider, is(true));
+		assumeThat(connectionProvider instanceof StandaloneConnectionProvider, is(true));
 
 		StepVerifier
 				.create(connection.stringCommands().bitOp(Arrays.asList(KEY_1_BBUFFER, KEY_2_BBUFFER), BitOperation.NOT,

src/test/java/org/springframework/data/redis/connection/lettuce/LettuceReactiveZSetCommandsTests.java

@@ -415,7 +415,7 @@ public class LettuceReactiveZSetCommandsTests extends LettuceReactiveCommandsTes
 	@Test // DATAREDIS-525
 	public void zUnionStoreShouldWorkCorrectly() {
 
-		assumeThat(clientProvider instanceof LettuceRedisClientProvider, is(true));
+		assumeThat(connectionProvider instanceof StandaloneConnectionProvider, is(true));
 
 		nativeCommands.zadd(KEY_1, 1D, VALUE_1);
 		nativeCommands.zadd(KEY_1, 2D, VALUE_2);
@@ -432,7 +432,7 @@ public class LettuceReactiveZSetCommandsTests extends LettuceReactiveCommandsTes
 	@Test // DATAREDIS-525
 	public void zInterStoreShouldWorkCorrectly() {
 
-		assumeThat(clientProvider instanceof LettuceRedisClientProvider, is(true));
+		assumeThat(connectionProvider instanceof StandaloneConnectionProvider, is(true));
 
 		nativeCommands.zadd(KEY_1, 1D, VALUE_1);
 		nativeCommands.zadd(KEY_1, 2D, VALUE_2);

src/test/java/org/springframework/data/redis/connection/lettuce/LettuceSentinelIntegrationTests.java

@@ -20,15 +20,18 @@ import static org.hamcrest.core.Is.is;
 import static org.hamcrest.core.IsEqual.equalTo;
 import static org.junit.Assert.*;
 
+import java.util.ArrayList;
 import java.util.Collection;
 import java.util.List;
 
 import org.junit.After;
-import org.junit.Before;
+import org.junit.AfterClass;
 import org.junit.ClassRule;
-import org.junit.Ignore;
 import org.junit.Rule;
 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.AbstractConnectionIntegrationTests;
 import org.springframework.data.redis.connection.DefaultStringRedisConnection;
@@ -43,6 +46,7 @@ import org.springframework.data.redis.test.util.RedisSentinelRule;
  * @author Mark Paluch
  * @author Christoph Strobl
  */
+@RunWith(Parameterized.class)
 public class LettuceSentinelIntegrationTests extends AbstractConnectionIntegrationTests {
 
 	private static final String MASTER_NAME = "mymaster";
@@ -58,8 +62,14 @@ public class LettuceSentinelIntegrationTests extends AbstractConnectionIntegrati
 	public static @ClassRule RedisSentinelRule sentinelRule = RedisSentinelRule.forConfig(SENTINEL_CONFIG).oneActive();
 	public @Rule MinimumRedisVersionRule minimumVersionRule = new MinimumRedisVersionRule();
 
-	@Before
-	public void setUp() {
+	public LettuceSentinelIntegrationTests(LettuceConnectionFactory connectionFactory, String displayName) {
+		this.connectionFactory = connectionFactory;
+	}
+
+	@Parameters(name = "{1}")
+	public static List<Object[]> parameters() {
+
+		List<Object[]> parameters = new ArrayList<>();
 
 		LettuceConnectionFactory lettuceConnectionFactory = new LettuceConnectionFactory(SENTINEL_CONFIG);
 		lettuceConnectionFactory.setClientResources(LettuceTestClientResources.getSharedClientResources());
@@ -67,14 +77,37 @@ public class LettuceSentinelIntegrationTests extends AbstractConnectionIntegrati
 		lettuceConnectionFactory.setShutdownTimeout(0);
 		lettuceConnectionFactory.afterPropertiesSet();
 
-		connectionFactory = lettuceConnectionFactory;
-		super.setUp();
+		LettuceConnectionFactory pooledConnectionFactory = new LettuceConnectionFactory(SENTINEL_CONFIG,
+				LettucePoolingClientConfiguration.builder().build());
+		pooledConnectionFactory.setShareNativeConnection(false);
+		pooledConnectionFactory.afterPropertiesSet();
+
+		ConnectionFactoryTracker.add(lettuceConnectionFactory);
+		ConnectionFactoryTracker.add(pooledConnectionFactory);
+
+		parameters.add(new Object[] { lettuceConnectionFactory, "Sentinel" });
+		parameters.add(new Object[] { pooledConnectionFactory, "Sentinel/Pooled" });
+
+		return parameters;
 	}
 
 	@After
 	public void tearDown() {
-		super.tearDown();
-		((LettuceConnectionFactory) connectionFactory).destroy();
+
+		try {
+
+			// since we use more than one db we're required to flush them all
+			connection.flushAll();
+		} catch (Exception e) {
+			// Connection may be closed in certain cases, like after pub/sub
+			// tests
+		}
+		connection.close();
+	}
+
+	@AfterClass
+	public static void afterClass() {
+		ConnectionFactoryTracker.cleanUp();
 	}
 
 	@Test // DATAREDIS-348

src/test/java/org/springframework/data/redis/core/ReactiveOperationsTestParams.java

@@ -39,6 +39,7 @@ 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.serializer.GenericJackson2JsonRedisSerializer;
 import org.springframework.data.redis.serializer.GenericToStringSerializer;
@@ -65,6 +66,11 @@ abstract public class ReactiveOperationsTestParams {
 				.clientResources(LettuceTestClientResources.getSharedClientResources()) //
 				.build();
 
+		LettuceClientConfiguration poolingConfiguration = LettucePoolingClientConfiguration.builder() //
+				.and().shutdownTimeout(Duration.ZERO) //
+				.clientResources(LettuceTestClientResources.getSharedClientResources()) //
+				.build();
+
 		RedisStandaloneConfiguration standaloneConfiguration = new RedisStandaloneConfiguration(SettingsUtils.getHost(),
 				SettingsUtils.getPort());
 
@@ -87,10 +93,17 @@ abstract public class ReactiveOperationsTestParams {
 				clientConfiguration);
 		lettuceConnectionFactory.afterPropertiesSet();
 
+		LettuceConnectionFactory poolingConnectionFactory = new LettuceConnectionFactory(standaloneConfiguration,
+				poolingConfiguration);
+		poolingConnectionFactory.afterPropertiesSet();
+
 		JdkSerializationRedisSerializer jdkSerializationRedisSerializer = new JdkSerializationRedisSerializer();
 		ReactiveRedisTemplate<Object, Object> objectTemplate = new ReactiveRedisTemplate<>(lettuceConnectionFactory,
 				RedisSerializationContext.fromSerializer(jdkSerializationRedisSerializer));
 
+		ReactiveRedisTemplate<Object, Object> pooledObjectTemplate = new ReactiveRedisTemplate<>(poolingConnectionFactory,
+				RedisSerializationContext.fromSerializer(jdkSerializationRedisSerializer));
+
 		StringRedisSerializer stringRedisSerializer = new StringRedisSerializer();
 		ReactiveRedisTemplate<String, String> stringTemplate = new ReactiveRedisTemplate<>(lettuceConnectionFactory,
 				RedisSerializationContext.fromSerializer(stringRedisSerializer));
@@ -129,6 +142,7 @@ abstract public class ReactiveOperationsTestParams {
 		List<Object[]> list = Arrays.asList(new Object[][] { //
 				{ stringTemplate, stringFactory, stringFactory, stringRedisSerializer, "String" }, //
 				{ objectTemplate, personFactory, personFactory, jdkSerializationRedisSerializer, "Person/JDK" }, //
+				{ pooledObjectTemplate, personFactory, personFactory, jdkSerializationRedisSerializer, "Pooled/Person/JDK" }, //
 				{ longTemplate, stringFactory, longFactory, longToStringSerializer, "Long" }, //
 				{ doubleTemplate, stringFactory, doubleFactory, doubleToStringSerializer, "Double" }, //
 				{ rawTemplate, rawFactory, rawFactory, null, "raw" }, //