diff --git a/src/main/java/org/springframework/data/redis/connection/lettuce/DefaultLettuceClientConfiguration.java b/src/main/java/org/springframework/data/redis/connection/lettuce/DefaultLettuceClientConfiguration.java index 80abb5d06..3745fa341 100644 --- a/src/main/java/org/springframework/data/redis/connection/lettuce/DefaultLettuceClientConfiguration.java +++ b/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() */ diff --git a/src/main/java/org/springframework/data/redis/connection/lettuce/DefaultLettucePoolingClientConfiguration.java b/src/main/java/org/springframework/data/redis/connection/lettuce/DefaultLettucePoolingClientConfiguration.java new file mode 100644 index 000000000..eef977d71 --- /dev/null +++ b/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; + } +} diff --git a/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceClientConfiguration.java b/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceClientConfiguration.java index b06075df9..48238b02c 100644 --- a/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceClientConfiguration.java +++ b/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) diff --git a/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceConnectionFactory.java b/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceConnectionFactory.java index c9c6023db..3cd34befc 100644 --- a/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceConnectionFactory.java +++ b/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; } diff --git a/src/main/java/org/springframework/data/redis/connection/lettuce/LettucePoolingClientConfiguration.java b/src/main/java/org/springframework/data/redis/connection/lettuce/LettucePoolingClientConfiguration.java new file mode 100644 index 000000000..4e4be1e32 --- /dev/null +++ b/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: + *
+ *
SSL
+ *
no
+ *
Pooling
+ *
yes
+ *
Peer Verification
+ *
yes
+ *
Start TLS
+ *
no
+ *
Client Options
+ *
none
+ *
Client Resources
+ *
none
+ *
Connect Timeout
+ *
60 Seconds
+ *
Shutdown Timeout
+ *
2 Seconds
+ *
+ * + * @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); + } + } +} diff --git a/src/main/java/org/springframework/data/redis/connection/lettuce/LettucePoolingConnectionProvider.java b/src/main/java/org/springframework/data/redis/connection/lettuce/LettucePoolingConnectionProvider.java new file mode 100644 index 000000000..dcdbe05d4 --- /dev/null +++ b/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. + *

+ * 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, GenericObjectPool>> poolRef = new ConcurrentHashMap<>( + 32); + private final Map, GenericObjectPool>> 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 connectionType) { + + GenericObjectPool> 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> 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(); + } +} diff --git a/src/test/java/org/springframework/data/redis/connection/lettuce/LettuceClientConfigurationUnitTests.java b/src/test/java/org/springframework/data/redis/connection/lettuce/LettuceClientConfigurationUnitTests.java index be6fc39b2..9fd3fc34c 100644 --- a/src/test/java/org/springframework/data/redis/connection/lettuce/LettuceClientConfigurationUnitTests.java +++ b/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(); diff --git a/src/test/java/org/springframework/data/redis/connection/lettuce/LettuceConnectionFactoryTests.java b/src/test/java/org/springframework/data/redis/connection/lettuce/LettuceConnectionFactoryTests.java index aff5e93ba..6bd267365 100644 --- a/src/test/java/org/springframework/data/redis/connection/lettuce/LettuceConnectionFactoryTests.java +++ b/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(); + } } diff --git a/src/test/java/org/springframework/data/redis/connection/lettuce/LettucePoolingClientConfigurationUnitTests.java b/src/test/java/org/springframework/data/redis/connection/lettuce/LettucePoolingClientConfigurationUnitTests.java new file mode 100644 index 000000000..68cfdeea0 --- /dev/null +++ b/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)); + } +} diff --git a/src/test/java/org/springframework/data/redis/connection/lettuce/LettuceReactiveCommandsTestsBase.java b/src/test/java/org/springframework/data/redis/connection/lettuce/LettuceReactiveCommandsTestsBase.java index 0849846b3..8521a4408 100644 --- a/src/test/java/org/springframework/data/redis/connection/lettuce/LettuceReactiveCommandsTestsBase.java +++ b/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 nativeCommands; - @Parameterized.Parameters(name = "{1}") + @Parameterized.Parameters(name = "{2}") public static List parameters() { - return Arrays.asList(new Object[] { LettuceRedisClientProvider.local(), "Standalone" }, - new Object[] { LettuceRedisClusterClientProvider.local(), "Cluster" }); + + LettuceRedisClientProvider standalone = LettuceRedisClientProvider.local(); + LettuceRedisClusterClientProvider cluster = LettuceRedisClusterClientProvider.local(); + + List 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()); } } diff --git a/src/test/java/org/springframework/data/redis/connection/lettuce/LettuceReactiveHyperLogLogCommandsTests.java b/src/test/java/org/springframework/data/redis/connection/lettuce/LettuceReactiveHyperLogLogCommandsTests.java index fd7b6c23b..520f7f558 100644 --- a/src/test/java/org/springframework/data/redis/connection/lettuce/LettuceReactiveHyperLogLogCommandsTests.java +++ b/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)); } - } diff --git a/src/test/java/org/springframework/data/redis/connection/lettuce/LettuceReactiveListCommandTests.java b/src/test/java/org/springframework/data/redis/connection/lettuce/LettuceReactiveListCommandTests.java index 212d00b03..32cde8c01 100644 --- a/src/test/java/org/springframework/data/redis/connection/lettuce/LettuceReactiveListCommandTests.java +++ b/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); diff --git a/src/test/java/org/springframework/data/redis/connection/lettuce/LettuceReactiveServerCommandsTests.java b/src/test/java/org/springframework/data/redis/connection/lettuce/LettuceReactiveServerCommandsTests.java index c82ee6081..773b025ab 100644 --- a/src/test/java/org/springframework/data/redis/connection/lettuce/LettuceReactiveServerCommandsTests.java +++ b/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(); diff --git a/src/test/java/org/springframework/data/redis/connection/lettuce/LettuceReactiveStringCommandsTests.java b/src/test/java/org/springframework/data/redis/connection/lettuce/LettuceReactiveStringCommandsTests.java index 0deb3a7ac..0f9698f86 100644 --- a/src/test/java/org/springframework/data/redis/connection/lettuce/LettuceReactiveStringCommandsTests.java +++ b/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 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, diff --git a/src/test/java/org/springframework/data/redis/connection/lettuce/LettuceReactiveZSetCommandsTests.java b/src/test/java/org/springframework/data/redis/connection/lettuce/LettuceReactiveZSetCommandsTests.java index 7d245a82f..3c52f36f6 100644 --- a/src/test/java/org/springframework/data/redis/connection/lettuce/LettuceReactiveZSetCommandsTests.java +++ b/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); diff --git a/src/test/java/org/springframework/data/redis/connection/lettuce/LettuceSentinelIntegrationTests.java b/src/test/java/org/springframework/data/redis/connection/lettuce/LettuceSentinelIntegrationTests.java index a4fa98b29..42576aeaf 100644 --- a/src/test/java/org/springframework/data/redis/connection/lettuce/LettuceSentinelIntegrationTests.java +++ b/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 parameters() { + + List 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 diff --git a/src/test/java/org/springframework/data/redis/core/ReactiveOperationsTestParams.java b/src/test/java/org/springframework/data/redis/core/ReactiveOperationsTestParams.java index 9d8dafd7f..816c689d8 100644 --- a/src/test/java/org/springframework/data/redis/core/ReactiveOperationsTestParams.java +++ b/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 objectTemplate = new ReactiveRedisTemplate<>(lettuceConnectionFactory, RedisSerializationContext.fromSerializer(jdkSerializationRedisSerializer)); + ReactiveRedisTemplate pooledObjectTemplate = new ReactiveRedisTemplate<>(poolingConnectionFactory, + RedisSerializationContext.fromSerializer(jdkSerializationRedisSerializer)); + StringRedisSerializer stringRedisSerializer = new StringRedisSerializer(); ReactiveRedisTemplate stringTemplate = new ReactiveRedisTemplate<>(lettuceConnectionFactory, RedisSerializationContext.fromSerializer(stringRedisSerializer)); @@ -129,6 +142,7 @@ abstract public class ReactiveOperationsTestParams { List 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" }, //