diff --git a/README.adoc b/README.adoc index 0979a3fb7..bcb2c0ea2 100644 --- a/README.adoc +++ b/README.adoc @@ -9,7 +9,7 @@ This modules provides integration with the https://redis.io/[Redis] store. == Features -* Connection package as low-level abstraction across multiple Redis drivers (https://github.com/lettuce-io/lettuce-core[Lettuce] and https://github.com/xetorthio/jedis[Jedis]). +* Connection package as low-level abstraction across multiple Redis drivers (https://github.com/lettuce-io/lettuce-core[Lettuce] and https://github.com/redis/jedis[Jedis]). * Exception translation to Spring’s portable Data Access exception hierarchy for Redis driver exceptions * https://docs.spring.io/spring-data/data-redis/docs/current/reference/html/#redis:template[`RedisTemplate`] that provides a high level abstraction for performing various Redis operations, exception translation and serialization support. * Pubsub support (such as a MessageListenerContainer for message-driven POJOs). diff --git a/src/main/asciidoc/reference/redis-repositories.adoc b/src/main/asciidoc/reference/redis-repositories.adoc index c29b7e3e9..b4097dfa8 100644 --- a/src/main/asciidoc/reference/redis-repositories.adoc +++ b/src/main/asciidoc/reference/redis-repositories.adoc @@ -845,6 +845,7 @@ class RedisOperationsProducer { LettuceConnectionFactory connectionFactory = new LettuceConnectionFactory(new RedisStandaloneConfiguration()); connectionFactory.afterPropertiesSet(); + connectionFactory.start(); return connectionFactory; } diff --git a/src/main/java/org/springframework/data/redis/connection/RedisConnectionFactory.java b/src/main/java/org/springframework/data/redis/connection/RedisConnectionFactory.java index 5a14fd8e6..1251399e0 100644 --- a/src/main/java/org/springframework/data/redis/connection/RedisConnectionFactory.java +++ b/src/main/java/org/springframework/data/redis/connection/RedisConnectionFactory.java @@ -30,8 +30,8 @@ public interface RedisConnectionFactory extends PersistenceExceptionTranslator { * Returns a suitable {@link RedisConnection connection} for interacting with Redis. * * @return {@link RedisConnection connection} for interacting with Redis. - * @throws IllegalStateException if the connection factory requires initialization and the factory has not yet - * been initialized. + * @throws IllegalStateException if the connection factory requires initialization and the factory has not yet been + * initialized. */ RedisConnection getConnection(); @@ -39,8 +39,8 @@ public interface RedisConnectionFactory extends PersistenceExceptionTranslator { * Returns a suitable {@link RedisClusterConnection connection} for interacting with Redis Cluster. * * @return a {@link RedisClusterConnection connection} for interacting with Redis Cluster. - * @throws IllegalStateException if the connection factory requires initialization and the factory has not yet - * been initialized. + * @throws IllegalStateException if the connection factory requires initialization and the factory has not yet been + * initialized. * @since 1.7 */ RedisClusterConnection getClusterConnection(); @@ -48,12 +48,12 @@ public interface RedisConnectionFactory extends PersistenceExceptionTranslator { /** * Specifies if pipelined results should be converted to the expected data type. *

- * If {@literal false}, results of {@link RedisConnection#closePipeline()} and {@link RedisConnection#exec()} - * will be of the type returned by the underlying driver. This method is mostly for backwards compatibility - * with {@literal 1.0}. It is generally always a good idea to allow results to be converted and deserialized. - * In fact, this is now the default behavior. + * If {@literal false}, results of {@link RedisConnection#closePipeline()} and {@link RedisConnection#exec()} will be + * of the type returned by the underlying driver. This method is mostly for backwards compatibility with + * {@literal 1.0}. It is generally always a good idea to allow results to be converted and deserialized. In fact, this + * is now the default behavior. * - * @return a boolen indicating whether to convert pipeline and transaction results. + * @return {@code true} to convert pipeline and transaction results; {@code false} otherwise. */ boolean getConvertPipelineAndTxResults(); @@ -61,8 +61,8 @@ public interface RedisConnectionFactory extends PersistenceExceptionTranslator { * Returns a suitable {@link RedisSentinelConnection connection} for interacting with Redis Sentinel. * * @return a {@link RedisSentinelConnection connection} for interacting with Redis Sentinel. - * @throws IllegalStateException if the connection factory requires initialization and the factory has not yet - * been initialized. + * @throws IllegalStateException if the connection factory requires initialization and the factory has not yet been + * initialized. * @since 1.4 */ RedisSentinelConnection getSentinelConnection(); diff --git a/src/main/java/org/springframework/data/redis/connection/jedis/JedisConnectionFactory.java b/src/main/java/org/springframework/data/redis/connection/jedis/JedisConnectionFactory.java index 51741a008..e0b96cb41 100644 --- a/src/main/java/org/springframework/data/redis/connection/jedis/JedisConnectionFactory.java +++ b/src/main/java/org/springframework/data/redis/connection/jedis/JedisConnectionFactory.java @@ -64,7 +64,7 @@ import org.springframework.util.CollectionUtils; import org.springframework.util.ObjectUtils; /** - * Connection factory creating Jedis based connections. + * Connection factory creating Jedis based connections. *

* {@link JedisConnectionFactory} should be configured using an environmental configuration and the * {@link JedisClientConfiguration client configuration}. Jedis supports the following environmental configurations: @@ -74,8 +74,9 @@ import org.springframework.util.ObjectUtils; *

  • {@link RedisClusterConfiguration}
    • * *

    - * This connection factory must be {@link #afterPropertiesSet() initialized} prior to {@link #getConnection obtaining - * connections}. + * This connection factory must be {@link #afterPropertiesSet() initialized} and {@link SmartLifecycle#start() started} + * prior to {@link #getConnection obtaining connections}. You can {@link SmartLifecycle#stop()} and + * {@link SmartLifecycle#start() restart} this connection factory if needed. * * @author Costin Leau * @author Thomas Darimont @@ -86,30 +87,36 @@ import org.springframework.util.ObjectUtils; * @see JedisClientConfiguration * @see Jedis */ -public class JedisConnectionFactory implements RedisConnectionFactory, InitializingBean, DisposableBean, SmartLifecycle { +public class JedisConnectionFactory + implements RedisConnectionFactory, InitializingBean, DisposableBean, SmartLifecycle { private final static Log log = LogFactory.getLog(JedisConnectionFactory.class); private static final ExceptionTranslationStrategy EXCEPTION_TRANSLATION = new PassThroughExceptionTranslationStrategy( JedisExceptionConverter.INSTANCE); private final JedisClientConfiguration clientConfiguration; - private JedisClientConfig clientConfig = DefaultJedisClientConfig.builder().build(); - private @Nullable Pool pool; + private boolean convertPipelineAndTxResults = true; private RedisStandaloneConfiguration standaloneConfig = new RedisStandaloneConfiguration("localhost", Protocol.DEFAULT_PORT); private @Nullable RedisConfiguration configuration; - private @Nullable JedisCluster cluster; - private @Nullable ClusterTopologyProvider topologyProvider; - private @Nullable ClusterCommandExecutor clusterCommandExecutor; - + /** + * Lifecycle state of this factory. + */ enum State { CREATED, STARTING, STARTED, STOPPING, STOPPED, DESTROYED; } - private AtomicReference state = new AtomicReference<>(State.CREATED); + private final AtomicReference state = new AtomicReference<>(State.CREATED); + + private JedisClientConfig clientConfig = DefaultJedisClientConfig.builder().build(); + + private @Nullable Pool pool; + private @Nullable JedisCluster cluster; + private @Nullable ClusterTopologyProvider topologyProvider; + private @Nullable ClusterCommandExecutor clusterCommandExecutor; /** * Constructs a new {@link JedisConnectionFactory} instance with default settings (default connection pooling). @@ -253,115 +260,6 @@ public class JedisConnectionFactory implements RedisConnectionFactory, Initializ this.configuration = clusterConfig; } - /** - * Returns a Jedis instance to be used as a Redis connection. The instance can be newly created or retrieved from a - * pool. - * - * @return Jedis instance ready for wrapping into a {@link RedisConnection}. - */ - protected Jedis fetchJedisConnector() { - try { - - if (getUsePool() && pool != null) { - return pool.getResource(); - } - - Jedis jedis = createJedis(); - // force initialization (see Jedis issue #82) - jedis.connect(); - - return jedis; - } catch (Exception ex) { - throw new RedisConnectionFailureException("Cannot get Jedis connection", ex); - } - } - - private Jedis createJedis() { - return new Jedis(new HostAndPort(getHostName(), getPort()), this.clientConfig); - } - - /** - * Post process a newly retrieved connection. Useful for decorating or executing initialization commands on a new - * connection. This implementation simply returns the connection. - * - * @param connection the jedis connection. - * @return processed connection - */ - protected JedisConnection postProcessConnection(JedisConnection connection) { - return connection; - } - - @Override - public void start() { - - State current = state.getAndUpdate(state -> { - if (State.CREATED.equals(state) || State.STOPPED.equals(state)) { - return State.STARTING; - } - return state; - }); - - if (State.CREATED.equals(current) || State.STOPPED.equals(current)) { - - if (getUsePool() && !isRedisClusterAware()) { - this.pool = createPool(); - } - - if (isRedisClusterAware()) { - - this.cluster = createCluster(); - this.topologyProvider = createTopologyProvider(this.cluster); - this.clusterCommandExecutor = new ClusterCommandExecutor(this.topologyProvider, - new JedisClusterConnection.JedisClusterNodeResourceProvider(this.cluster, this.topologyProvider), - EXCEPTION_TRANSLATION); - } - - state.set(State.STARTED); - } - } - - @Override - public void stop() { - - if (state.compareAndSet(State.STARTED, State.STOPPING)) { - if (getUsePool() && !isRedisClusterAware()) { - if (pool != null) { - try { - this.pool.close(); - } catch (Exception ex) { - log.warn("Cannot properly close Jedis pool", ex); - } - this.pool = null; - } - } - - if(this.clusterCommandExecutor != null) { - try { - this.clusterCommandExecutor.destroy(); - } catch (Exception e) { - throw new RuntimeException(e); - } - } - - if (this.cluster != null) { - - this.topologyProvider = null; - - try { - cluster.close(); - } catch (Exception ex) { - log.warn("Cannot properly close Jedis cluster", ex); - } - } - state.set(State.STOPPED); - } - } - - @Override - public boolean isRunning() { - return State.STARTED.equals(state.get()); - } - @Override public void afterPropertiesSet() { clientConfig = createClientConfig(getDatabase(), getRedisUsername(), getRedisPassword()); @@ -399,6 +297,76 @@ public class JedisConnectionFactory implements RedisConnectionFactory, Initializ return builder.build(); } + @Override + public void start() { + + State current = state + .getAndUpdate(state -> State.CREATED.equals(state) || State.STOPPED.equals(state) ? State.STARTING : state); + + if (State.CREATED.equals(current) || State.STOPPED.equals(current)) { + + if (getUsePool() && !isRedisClusterAware()) { + this.pool = createPool(); + } + + if (isRedisClusterAware()) { + + this.cluster = createCluster(); + this.topologyProvider = createTopologyProvider(this.cluster); + this.clusterCommandExecutor = new ClusterCommandExecutor(this.topologyProvider, + new JedisClusterConnection.JedisClusterNodeResourceProvider(this.cluster, this.topologyProvider), + EXCEPTION_TRANSLATION); + } + + state.set(State.STARTED); + } + } + + @Override + public void stop() { + + if (state.compareAndSet(State.STARTED, State.STOPPING)) { + + if (getUsePool() && !isRedisClusterAware()) { + if (pool != null) { + try { + this.pool.close(); + this.pool = null; + } catch (Exception ex) { + log.warn("Cannot properly close Jedis pool", ex); + } + } + } + + if (this.clusterCommandExecutor != null) { + try { + this.clusterCommandExecutor.destroy(); + this.clusterCommandExecutor = null; + } catch (Exception e) { + throw new RuntimeException(e); + } + } + + if (this.cluster != null) { + + this.topologyProvider = null; + + try { + this.cluster.close(); + this.cluster = null; + } catch (Exception ex) { + log.warn("Cannot properly close Jedis cluster", ex); + } + } + state.set(State.STOPPED); + } + } + + @Override + public boolean isRunning() { + return State.STARTED.equals(state.get()); + } + private Pool createPool() { if (isRedisSentinelAware()) { @@ -473,12 +441,14 @@ public class JedisConnectionFactory implements RedisConnectionFactory, Initializ return new JedisCluster(hostAndPort, this.clientConfig, redirects, poolConfig); } + @Override public void destroy() { stop(); state.set(State.DESTROYED); } + @Override public RedisConnection getConnection() { assertInitialized(); @@ -498,9 +468,48 @@ public class JedisConnectionFactory implements RedisConnectionFactory, Initializ JedisConnection connection = (getUsePool() ? new JedisConnection(jedis, pool, this.clientConfig, sentinelConfig) : new JedisConnection(jedis, null, this.clientConfig, sentinelConfig)); connection.setConvertPipelineAndTxResults(convertPipelineAndTxResults); + return postProcessConnection(connection); } + /** + * Returns a Jedis instance to be used as a Redis connection. The instance can be newly created or retrieved from a + * pool. + * + * @return Jedis instance ready for wrapping into a {@link RedisConnection}. + */ + protected Jedis fetchJedisConnector() { + try { + + if (getUsePool() && pool != null) { + return pool.getResource(); + } + + Jedis jedis = createJedis(); + // force initialization (see Jedis issue #82) + jedis.connect(); + + return jedis; + } catch (Exception ex) { + throw new RedisConnectionFailureException("Cannot get Jedis connection", ex); + } + } + + private Jedis createJedis() { + return new Jedis(new HostAndPort(getHostName(), getPort()), this.clientConfig); + } + + /** + * Post process a newly retrieved connection. Useful for decorating or executing initialization commands on a new + * connection. This implementation simply returns the connection. + * + * @param connection the jedis connection. + * @return processed connection + */ + protected JedisConnection postProcessConnection(JedisConnection connection) { + return connection; + } + @Override public RedisClusterConnection getClusterConnection() { @@ -509,9 +518,24 @@ public class JedisConnectionFactory implements RedisConnectionFactory, Initializ if (!isRedisClusterAware()) { throw new InvalidDataAccessApiUsageException("Cluster is not configured"); } - return new JedisClusterConnection(this.cluster, this.clusterCommandExecutor, this.topologyProvider); + + return postProcessConnection( + new JedisClusterConnection(this.cluster, this.clusterCommandExecutor, this.topologyProvider)); } + /** + * Post process a newly retrieved connection. Useful for decorating or executing initialization commands on a new + * connection. This implementation simply returns the connection. + * + * @param connection the jedis connection. + * @return processed connection. + * @since 3.2 + */ + protected JedisClusterConnection postProcessConnection(JedisClusterConnection connection) { + return connection; + } + + @Override public DataAccessException translateExceptionIfPossible(RuntimeException ex) { return EXCEPTION_TRANSLATION.translate(ex); } @@ -790,22 +814,24 @@ public class JedisConnectionFactory implements RedisConnectionFactory, Initializ } /** - * Specifies if pipelined results should be converted to the expected data type. If false, results of + * Specifies if pipelined results should be converted to the expected data type. If {@code false}, results of * {@link JedisConnection#closePipeline()} and {@link JedisConnection#exec()} will be of the type returned by the * Jedis driver. * - * @return Whether or not to convert pipeline and tx results. + * @return {@code true} to convert pipeline and transaction results; {@code false} otherwise. */ + @Override public boolean getConvertPipelineAndTxResults() { return convertPipelineAndTxResults; } /** - * Specifies if pipelined results should be converted to the expected data type. If false, results of + * Specifies if pipelined results should be converted to the expected data type. If {@code false}, results of * {@link JedisConnection#closePipeline()} and {@link JedisConnection#exec()} will be of the type returned by the * Jedis driver. * - * @param convertPipelineAndTxResults Whether or not to convert pipeline and tx results. + * @param convertPipelineAndTxResults {@code true} to convert pipeline and transaction results; {@code false} + * otherwise. */ public void setConvertPipelineAndTxResults(boolean convertPipelineAndTxResults) { this.convertPipelineAndTxResults = convertPipelineAndTxResults; @@ -910,7 +936,8 @@ public class JedisConnectionFactory implements RedisConnectionFactory, Initializ } switch (current) { - case CREATED, STOPPED -> throw new IllegalStateException(String.format("JedisConnectionFactory has been %s. Use start() to initialize it", current)); + case CREATED, STOPPED -> throw new IllegalStateException( + String.format("JedisConnectionFactory has been %s. Use start() to initialize it", current)); case DESTROYED -> throw new IllegalStateException( "JedisConnectionFactory was destroyed and cannot be used anymore"); default -> throw new IllegalStateException(String.format("JedisConnectionFactory is %s", current)); diff --git a/src/main/java/org/springframework/data/redis/connection/jedis/package-info.java b/src/main/java/org/springframework/data/redis/connection/jedis/package-info.java index e3a02a282..c32ac1cab 100644 --- a/src/main/java/org/springframework/data/redis/connection/jedis/package-info.java +++ b/src/main/java/org/springframework/data/redis/connection/jedis/package-info.java @@ -1,5 +1,5 @@ /** - * Connection package for Jedis library. + * Connection package for Jedis library. */ @org.springframework.lang.NonNullApi @org.springframework.lang.NonNullFields 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 2a77891a2..4ffaeea41 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 @@ -15,53 +15,7 @@ */ package org.springframework.data.redis.connection.lettuce; -import static org.springframework.data.redis.connection.lettuce.LettuceConnection.CODEC; -import static org.springframework.data.redis.connection.lettuce.LettuceConnection.PipeliningFlushPolicy; - -import java.nio.ByteBuffer; -import java.time.Duration; -import java.util.ArrayList; -import java.util.List; -import java.util.Optional; -import java.util.concurrent.CompletableFuture; -import java.util.concurrent.CompletionStage; -import java.util.concurrent.TimeUnit; -import java.util.concurrent.atomic.AtomicReference; -import java.util.function.Consumer; -import java.util.stream.Collectors; - -import org.springframework.beans.factory.DisposableBean; -import org.springframework.beans.factory.InitializingBean; -import org.springframework.context.SmartLifecycle; -import org.springframework.dao.DataAccessException; -import org.springframework.dao.InvalidDataAccessApiUsageException; -import org.springframework.data.redis.ExceptionTranslationStrategy; -import org.springframework.data.redis.PassThroughExceptionTranslationStrategy; -import org.springframework.data.redis.RedisConnectionFailureException; -import org.springframework.data.redis.connection.ClusterCommandExecutor; -import org.springframework.data.redis.connection.ClusterTopologyProvider; -import org.springframework.data.redis.connection.ReactiveRedisConnectionFactory; -import org.springframework.data.redis.connection.RedisClusterConfiguration; -import org.springframework.data.redis.connection.RedisClusterConnection; -import org.springframework.data.redis.connection.RedisConfiguration; -import org.springframework.data.redis.connection.RedisConfiguration.ClusterConfiguration; -import org.springframework.data.redis.connection.RedisConfiguration.DomainSocketConfiguration; -import org.springframework.data.redis.connection.RedisConfiguration.WithDatabaseIndex; -import org.springframework.data.redis.connection.RedisConfiguration.WithPassword; -import org.springframework.data.redis.connection.RedisConnection; -import org.springframework.data.redis.connection.RedisConnectionFactory; -import org.springframework.data.redis.connection.RedisPassword; -import org.springframework.data.redis.connection.RedisSentinelConfiguration; -import org.springframework.data.redis.connection.RedisSentinelConnection; -import org.springframework.data.redis.connection.RedisSocketConfiguration; -import org.springframework.data.redis.connection.RedisStandaloneConfiguration; -import org.springframework.data.redis.connection.RedisStaticMasterReplicaConfiguration; -import org.springframework.data.util.Optionals; -import org.springframework.lang.Nullable; -import org.springframework.util.Assert; -import org.springframework.util.ClassUtils; -import org.springframework.util.ObjectUtils; -import org.springframework.util.StringUtils; +import static org.springframework.data.redis.connection.lettuce.LettuceConnection.*; import io.lettuce.core.AbstractRedisClient; import io.lettuce.core.ClientOptions; @@ -78,8 +32,39 @@ import io.lettuce.core.cluster.api.StatefulRedisClusterConnection; import io.lettuce.core.codec.RedisCodec; import io.lettuce.core.resource.ClientResources; +import java.nio.ByteBuffer; +import java.time.Duration; +import java.util.ArrayList; +import java.util.List; +import java.util.Optional; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.CompletionStage; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicReference; +import java.util.function.Consumer; +import java.util.stream.Collectors; + import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; +import org.springframework.beans.factory.DisposableBean; +import org.springframework.beans.factory.InitializingBean; +import org.springframework.context.SmartLifecycle; +import org.springframework.dao.DataAccessException; +import org.springframework.dao.InvalidDataAccessApiUsageException; +import org.springframework.data.redis.ExceptionTranslationStrategy; +import org.springframework.data.redis.PassThroughExceptionTranslationStrategy; +import org.springframework.data.redis.RedisConnectionFailureException; +import org.springframework.data.redis.connection.*; +import org.springframework.data.redis.connection.RedisConfiguration.ClusterConfiguration; +import org.springframework.data.redis.connection.RedisConfiguration.DomainSocketConfiguration; +import org.springframework.data.redis.connection.RedisConfiguration.WithDatabaseIndex; +import org.springframework.data.redis.connection.RedisConfiguration.WithPassword; +import org.springframework.data.util.Optionals; +import org.springframework.lang.Nullable; +import org.springframework.util.Assert; +import org.springframework.util.ClassUtils; +import org.springframework.util.ObjectUtils; +import org.springframework.util.StringUtils; /** * Connection factory creating Lettuce-based connections. @@ -104,8 +89,9 @@ import org.apache.commons.logging.LogFactory; *

  • {@link RedisClusterConfiguration}
    • * *

    - * This connection factory must be {@link #afterPropertiesSet() initialized} prior to {@link #getConnection obtaining - * connections}. + * This connection factory must be {@link #afterPropertiesSet() initialized} and {@link SmartLifecycle#start() started} + * prior to {@link #getConnection obtaining connections}. You can {@link SmartLifecycle#stop()} and + * {@link SmartLifecycle#start() restart} this connection factory if needed. * * @author Costin Leau * @author Jennifer Hickey @@ -125,32 +111,38 @@ public class LettuceConnectionFactory implements RedisConnectionFactory, Reactiv LettuceExceptionConverter.INSTANCE); private final Log log = LogFactory.getLog(getClass()); - private final LettuceClientConfiguration clientConfiguration; - private @Nullable AbstractRedisClient client; - private @Nullable LettuceConnectionProvider connectionProvider; - private @Nullable LettuceConnectionProvider reactiveConnectionProvider; - private boolean validateConnection = false; - private boolean shareNativeConnection = true; - private boolean eagerInitialization = false; - private @Nullable SharedConnection connection; - private @Nullable SharedConnection reactiveConnection; /** Synchronization monitor for the shared Connection */ private final Object connectionMonitor = new Object(); - private boolean convertPipelineAndTxResults = true; + + private final LettuceClientConfiguration clientConfiguration; private RedisStandaloneConfiguration standaloneConfig = new RedisStandaloneConfiguration("localhost", 6379); - private PipeliningFlushPolicy pipeliningFlushPolicy = PipeliningFlushPolicy.flushEachCommand(); private @Nullable RedisConfiguration configuration; - private @Nullable ClusterCommandExecutor clusterCommandExecutor; + private boolean validateConnection = false; + private boolean shareNativeConnection = true; + private boolean eagerInitialization = false; + private boolean convertPipelineAndTxResults = true; + private PipeliningFlushPolicy pipeliningFlushPolicy = PipeliningFlushPolicy.flushEachCommand(); + + /** + * Lifecycle state of this factory. + */ enum State { CREATED, STARTING, STARTED, STOPPING, STOPPED, DESTROYED; } - private AtomicReference state = new AtomicReference<>(State.CREATED); + private final AtomicReference state = new AtomicReference<>(State.CREATED); + + private @Nullable AbstractRedisClient client; + private @Nullable LettuceConnectionProvider connectionProvider; + private @Nullable LettuceConnectionProvider reactiveConnectionProvider; + private @Nullable SharedConnection connection; + private @Nullable SharedConnection reactiveConnection; + private @Nullable ClusterCommandExecutor clusterCommandExecutor; /** * Constructs a new {@link LettuceConnectionFactory} instance with default settings. @@ -341,12 +333,8 @@ public class LettuceConnectionFactory implements RedisConnectionFactory, Reactiv @Override public void start() { - State current = state.getAndUpdate(state -> { - if (State.CREATED.equals(state) || State.STOPPED.equals(state)) { - return State.STARTING; - } - return state; - }); + State current = state + .getAndUpdate(state -> State.CREATED.equals(state) || State.STOPPED.equals(state) ? State.STARTING : state); if (State.CREATED.equals(current) || State.STOPPED.equals(current)) { @@ -376,23 +364,33 @@ public class LettuceConnectionFactory implements RedisConnectionFactory, Reactiv public void stop() { if (state.compareAndSet(State.STARTED, State.STOPPING)) { - resetConnection(); - dispose(connectionProvider); - dispose(reactiveConnectionProvider); - try { - Duration quietPeriod = clientConfiguration.getShutdownQuietPeriod(); - Duration timeout = clientConfiguration.getShutdownTimeout(); - client.shutdown(quietPeriod.toMillis(), timeout.toMillis(), TimeUnit.MILLISECONDS); - state.set(State.STOPPED); - } catch (Exception e) { - if (log.isWarnEnabled()) { - log.warn((client != null ? ClassUtils.getShortName(client.getClass()) : "LettuceClient") - + " did not shut down gracefully.", e); + resetConnection(); + + dispose(connectionProvider); + connectionProvider = null; + + dispose(reactiveConnectionProvider); + reactiveConnectionProvider = null; + + if (client != null) { + + try { + Duration quietPeriod = clientConfiguration.getShutdownQuietPeriod(); + Duration timeout = clientConfiguration.getShutdownTimeout(); + + client.shutdown(quietPeriod.toMillis(), timeout.toMillis(), TimeUnit.MILLISECONDS); + client = null; + } catch (Exception e) { + + if (log.isWarnEnabled()) { + log.warn(ClassUtils.getShortName(client.getClass()) + " did not shut down gracefully.", e); + } } } - state.set(State.STOPPED); } + + state.set(State.STOPPED); } @Override @@ -421,7 +419,7 @@ public class LettuceConnectionFactory implements RedisConnectionFactory, Reactiv state.set(State.DESTROYED); } - private void dispose(LettuceConnectionProvider connectionProvider) { + private void dispose(@Nullable LettuceConnectionProvider connectionProvider) { if (connectionProvider instanceof DisposableBean) { try { @@ -435,9 +433,10 @@ public class LettuceConnectionFactory implements RedisConnectionFactory, Reactiv } } + @Override public RedisConnection getConnection() { - assertInitialized(); + assertStarted(); if (isClusterAware()) { return getClusterConnection(); @@ -452,7 +451,7 @@ public class LettuceConnectionFactory implements RedisConnectionFactory, Reactiv @Override public RedisClusterConnection getClusterConnection() { - assertInitialized(); + assertStarted(); if (!isClusterAware()) { throw new InvalidDataAccessApiUsageException("Cluster is not configured"); @@ -517,7 +516,7 @@ public class LettuceConnectionFactory implements RedisConnectionFactory, Reactiv @Override public LettuceReactiveRedisConnection getReactiveConnection() { - assertInitialized(); + assertStarted(); if (isClusterAware()) { return getReactiveClusterConnection(); @@ -531,7 +530,7 @@ public class LettuceConnectionFactory implements RedisConnectionFactory, Reactiv @Override public LettuceReactiveRedisClusterConnection getReactiveClusterConnection() { - assertInitialized(); + assertStarted(); if (!isClusterAware()) { throw new InvalidDataAccessApiUsageException("Cluster is not configured"); @@ -581,7 +580,7 @@ public class LettuceConnectionFactory implements RedisConnectionFactory, Reactiv */ public void validateConnection() { - assertInitialized(); + assertStarted(); getOrCreateSharedConnection().validateConnection(); getOrCreateSharedReactiveConnection().validateConnection(); @@ -611,6 +610,7 @@ public class LettuceConnectionFactory implements RedisConnectionFactory, Reactiv } } + @Override public DataAccessException translateExceptionIfPossible(RuntimeException ex) { return EXCEPTION_TRANSLATION.translate(ex); } @@ -888,7 +888,7 @@ public class LettuceConnectionFactory implements RedisConnectionFactory, Reactiv */ @Nullable public AbstractRedisClient getNativeClient() { - assertInitialized(); + assertStarted(); return this.client; } @@ -1042,22 +1042,24 @@ public class LettuceConnectionFactory implements RedisConnectionFactory, Reactiv } /** - * Specifies if pipelined results should be converted to the expected data type. If false, results of + * Specifies if pipelined results should be converted to the expected data type. If {@code false}, results of * {@link LettuceConnection#closePipeline()} and {LettuceConnection#exec()} will be of the type returned by the * Lettuce driver. * - * @return Whether or not to convert pipeline and tx results. + * @return {@code true} to convert pipeline and transaction results; {@code false} otherwise. */ + @Override public boolean getConvertPipelineAndTxResults() { return convertPipelineAndTxResults; } /** - * Specifies if pipelined and transaction results should be converted to the expected data type. If false, results of - * {@link LettuceConnection#closePipeline()} and {LettuceConnection#exec()} will be of the type returned by the - * Lettuce driver. + * Specifies if pipelined and transaction results should be converted to the expected data type. If {@code false}, + * results of {@link LettuceConnection#closePipeline()} and {LettuceConnection#exec()} will be of the type returned by + * the Lettuce driver. * - * @param convertPipelineAndTxResults Whether or not to convert pipeline and tx results. + * @param convertPipelineAndTxResults {@code true} to convert pipeline and transaction results; {@code false} + * otherwise. */ public void setConvertPipelineAndTxResults(boolean convertPipelineAndTxResults) { this.convertPipelineAndTxResults = convertPipelineAndTxResults; @@ -1156,8 +1158,8 @@ public class LettuceConnectionFactory implements RedisConnectionFactory, Reactiv protected LettuceConnectionProvider doCreateConnectionProvider(AbstractRedisClient client, RedisCodec codec) { return isStaticMasterReplicaAware() ? createStaticMasterReplicaConnectionProvider((RedisClient) client, codec) - : isClusterAware() ? createClusterConnectionProvider((RedisClusterClient) client, codec) - : createStandaloneConnectionProvider((RedisClient) client, codec); + : isClusterAware() ? createClusterConnectionProvider((RedisClusterClient) client, codec) + : createStandaloneConnectionProvider((RedisClient) client, codec); } @SuppressWarnings("all") @@ -1165,9 +1167,8 @@ public class LettuceConnectionFactory implements RedisConnectionFactory, Reactiv RedisCodec codec) { List nodes = ((RedisStaticMasterReplicaConfiguration) this.configuration).getNodes().stream() - .map(it -> createRedisURIAndApplySettings(it.getHostName(), it.getPort())) - .peek(it -> it.setDatabase(getDatabase())) - .collect(Collectors.toList()); + .map(it -> createRedisURIAndApplySettings(it.getHostName(), it.getPort())) + .peek(it -> it.setDatabase(getDatabase())).collect(Collectors.toList()); return new StaticMasterReplicaConnectionProvider(client, codec, nodes, getClientConfiguration().getReadFrom().orElse(null)); @@ -1184,15 +1185,13 @@ public class LettuceConnectionFactory implements RedisConnectionFactory, Reactiv protected AbstractRedisClient createClient() { return isStaticMasterReplicaAware() ? createStaticMasterReplicaClient() - : isRedisSentinelAware() ? createSentinelClient() - : isClusterAware() ? createClusterClient() - : createBasicClient(); + : isRedisSentinelAware() ? createSentinelClient() + : isClusterAware() ? createClusterClient() : createBasicClient(); } private RedisClient createStaticMasterReplicaClient() { - RedisClient redisClient = this.clientConfiguration.getClientResources() - .map(RedisClient::create) + RedisClient redisClient = this.clientConfiguration.getClientResources().map(RedisClient::create) .orElseGet(RedisClient::create); this.clientConfiguration.getClientOptions().ifPresent(redisClient::setOptions); @@ -1251,8 +1250,7 @@ public class LettuceConnectionFactory implements RedisConnectionFactory, Reactiv ClusterConfiguration configuration = (ClusterConfiguration) this.configuration; - configuration.getClusterNodes().stream() - .map(node -> createRedisURIAndApplySettings(node.getHost(), node.getPort())) + configuration.getClusterNodes().stream().map(node -> createRedisURIAndApplySettings(node.getHost(), node.getPort())) .forEach(initialUris::add); RedisClusterClient clusterClient = this.clientConfiguration.getClientResources() @@ -1268,12 +1266,9 @@ public class LettuceConnectionFactory implements RedisConnectionFactory, Reactiv Optional clientOptions = this.clientConfiguration.getClientOptions(); - ClusterClientOptions clusterClientOptions = clientOptions - .filter(ClusterClientOptions.class::isInstance) - .map(ClusterClientOptions.class::cast) - .orElseGet(() -> clientOptions - .map(it -> ClusterClientOptions.builder(it).build()) - .orElseGet(ClusterClientOptions::create)); + ClusterClientOptions clusterClientOptions = clientOptions.filter(ClusterClientOptions.class::isInstance) + .map(ClusterClientOptions.class::cast).orElseGet(() -> clientOptions + .map(it -> ClusterClientOptions.builder(it).build()).orElseGet(ClusterClientOptions::create)); if (configuration.getMaxRedirects() != null) { return clusterClientOptions.mutate().maxRedirects(configuration.getMaxRedirects()).build(); @@ -1290,15 +1285,14 @@ public class LettuceConnectionFactory implements RedisConnectionFactory, Reactiv : createRedisURIAndApplySettings(getHostName(), getPort()); RedisClient redisClient = this.clientConfiguration.getClientResources() - .map(clientResources -> RedisClient.create(clientResources, uri)) - .orElseGet(() -> RedisClient.create(uri)); + .map(clientResources -> RedisClient.create(clientResources, uri)).orElseGet(() -> RedisClient.create(uri)); this.clientConfiguration.getClientOptions().ifPresent(redisClient::setOptions); return redisClient; } - private void assertInitialized() { + private void assertStarted() { State current = state.get(); @@ -1307,7 +1301,8 @@ public class LettuceConnectionFactory implements RedisConnectionFactory, Reactiv } switch (current) { - case CREATED, STOPPED -> throw new IllegalStateException(String.format("LettuceConnectionFactory has been %s. Use start() to initialize it", current)); + case CREATED, STOPPED -> throw new IllegalStateException( + String.format("LettuceConnectionFactory has been %s. Use start() to initialize it", current)); case DESTROYED -> throw new IllegalStateException( "LettuceConnectionFactory was destroyed and cannot be used anymore"); default -> throw new IllegalStateException(String.format("LettuceConnectionFactory is %s", current)); @@ -1359,14 +1354,14 @@ public class LettuceConnectionFactory implements RedisConnectionFactory, Reactiv getRedisPassword().toOptional().ifPresent(builder::withPassword); } - clientConfiguration.getRedisCredentialsProviderFactory().ifPresent(factory -> - builder.withAuthentication(factory.createCredentialsProvider(this.configuration))); + clientConfiguration.getRedisCredentialsProviderFactory() + .ifPresent(factory -> builder.withAuthentication(factory.createCredentialsProvider(this.configuration))); } @Override public RedisSentinelConnection getSentinelConnection() { - assertInitialized(); + assertStarted(); return new LettuceSentinelConnection(connectionProvider); } diff --git a/src/test/java/org/springframework/data/redis/ConnectionFactoryTracker.java b/src/test/java/org/springframework/data/redis/ConnectionFactoryTracker.java index 489fab2c0..946097bd2 100644 --- a/src/test/java/org/springframework/data/redis/ConnectionFactoryTracker.java +++ b/src/test/java/org/springframework/data/redis/ConnectionFactoryTracker.java @@ -41,8 +41,8 @@ public abstract class ConnectionFactoryTracker { throw new UnsupportedOperationException("Cannot track managed resource"); } - if(factory instanceof SmartLifecycle smartLifecycle) { - if(!smartLifecycle.isRunning() && smartLifecycle.isAutoStartup()) { + if (factory instanceof SmartLifecycle smartLifecycle) { + if (!smartLifecycle.isRunning() && smartLifecycle.isAutoStartup()) { smartLifecycle.start(); } } diff --git a/src/test/java/org/springframework/data/redis/cache/RedisCacheTests.java b/src/test/java/org/springframework/data/redis/cache/RedisCacheTests.java index d531eef94..326fc78a3 100644 --- a/src/test/java/org/springframework/data/redis/cache/RedisCacheTests.java +++ b/src/test/java/org/springframework/data/redis/cache/RedisCacheTests.java @@ -15,10 +15,10 @@ */ package org.springframework.data.redis.cache; -import static org.assertj.core.api.Assertions.assertThat; -import static org.assertj.core.api.Assertions.assertThatExceptionOfType; -import static org.assertj.core.api.Assertions.assertThatIllegalStateException; -import static org.assertj.core.api.Assumptions.assumeThat; +import static org.assertj.core.api.Assertions.*; +import static org.assertj.core.api.Assumptions.*; + +import io.netty.util.concurrent.DefaultThreadFactory; import java.io.Serializable; import java.nio.charset.StandardCharsets; @@ -37,7 +37,6 @@ import java.util.function.Consumer; import java.util.stream.IntStream; import org.junit.jupiter.api.BeforeEach; - import org.springframework.cache.Cache.ValueWrapper; import org.springframework.cache.interceptor.SimpleKey; import org.springframework.cache.interceptor.SimpleKeyGenerator; @@ -51,8 +50,6 @@ import org.springframework.data.redis.test.extension.parametrized.MethodSource; import org.springframework.data.redis.test.extension.parametrized.ParameterizedRedisTest; import org.springframework.lang.Nullable; -import io.netty.util.concurrent.DefaultThreadFactory; - /** * Tests for {@link RedisCache} with {@link DefaultRedisCacheWriter} using different {@link RedisSerializer} and * {@link RedisConnectionFactory} pairs. @@ -443,7 +440,7 @@ public class RedisCacheTests { Object key = SimpleKeyGenerator .generateKey(Collections.singletonList(new InvalidKey(sample.getFirstname(), sample.getBirthdate()))); - assertThatExceptionOfType(IllegalStateException.class).isThrownBy(() -> cache.put(key, sample)); + assertThatIllegalStateException().isThrownBy(() -> cache.put(key, sample)); } @ParameterizedRedisTest // GH-2079 diff --git a/src/test/java/org/springframework/data/redis/connection/jedis/JedisConnectionFactoryIntegrationTests.java b/src/test/java/org/springframework/data/redis/connection/jedis/JedisConnectionFactoryIntegrationTests.java index 61aabedfa..acbc683e1 100644 --- a/src/test/java/org/springframework/data/redis/connection/jedis/JedisConnectionFactoryIntegrationTests.java +++ b/src/test/java/org/springframework/data/redis/connection/jedis/JedisConnectionFactoryIntegrationTests.java @@ -19,7 +19,6 @@ import static org.assertj.core.api.Assertions.*; import org.junit.jupiter.api.AfterEach; import org.junit.jupiter.api.Test; - import org.springframework.data.redis.SettingsUtils; import org.springframework.data.redis.connection.RedisConnection; import org.springframework.data.redis.connection.RedisStandaloneConfiguration; @@ -84,12 +83,14 @@ class JedisConnectionFactoryIntegrationTests { factory.stop(); assertThat(factory.isRunning()).isFalse(); - assertThatExceptionOfType(IllegalStateException.class).isThrownBy(() -> factory.getConnection()); + assertThatIllegalStateException().isThrownBy(() -> factory.getConnection()); factory.start(); assertThat(factory.isRunning()).isTrue(); try (RedisConnection connection = factory.getConnection()) { assertThat(connection.ping()).isEqualTo("PONG"); } + + factory.destroy(); } } diff --git a/src/test/java/org/springframework/data/redis/connection/jedis/JedisConnectionFactoryUnitTests.java b/src/test/java/org/springframework/data/redis/connection/jedis/JedisConnectionFactoryUnitTests.java index 1fdbe738a..f5cf14a83 100644 --- a/src/test/java/org/springframework/data/redis/connection/jedis/JedisConnectionFactoryUnitTests.java +++ b/src/test/java/org/springframework/data/redis/connection/jedis/JedisConnectionFactoryUnitTests.java @@ -35,7 +35,6 @@ import javax.net.ssl.SSLSocketFactory; import org.apache.commons.pool2.impl.GenericObjectPoolConfig; import org.junit.jupiter.api.Test; - import org.springframework.data.redis.connection.RedisClusterConfiguration; import org.springframework.data.redis.connection.RedisPassword; import org.springframework.data.redis.connection.RedisSentinelConfiguration; @@ -334,7 +333,7 @@ class JedisConnectionFactoryUnitTests { connectionFactory.afterPropertiesSet(); assertThat(connectionFactory.isRunning()).isFalse(); - assertThatExceptionOfType(IllegalStateException.class).isThrownBy(() -> connectionFactory.getConnection()); + assertThatIllegalStateException().isThrownBy(() -> connectionFactory.getConnection()); } private JedisConnectionFactory initSpyedConnectionFactory(RedisSentinelConfiguration sentinelConfig, 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 3f5d34840..7583a58e8 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 @@ -575,10 +575,13 @@ class LettuceConnectionFactoryTests { assertThat(factory.isRunning()).isTrue(); factory.stop(); + assertThat(factory.isRunning()).isFalse(); - assertThatExceptionOfType(IllegalStateException.class).isThrownBy(() -> factory.getConnection()); + assertThatIllegalStateException().isThrownBy(() -> factory.getConnection()); + factory.start(); assertThat(factory.isRunning()).isTrue(); + try (RedisConnection connection = factory.getConnection()) { assertThat(connection.ping()).isEqualTo("PONG"); } diff --git a/src/test/java/org/springframework/data/redis/connection/lettuce/LettuceConnectionFactoryUnitTests.java b/src/test/java/org/springframework/data/redis/connection/lettuce/LettuceConnectionFactoryUnitTests.java index 209e7cd57..e2e1058aa 100644 --- a/src/test/java/org/springframework/data/redis/connection/lettuce/LettuceConnectionFactoryUnitTests.java +++ b/src/test/java/org/springframework/data/redis/connection/lettuce/LettuceConnectionFactoryUnitTests.java @@ -15,20 +15,27 @@ */ package org.springframework.data.redis.connection.lettuce; -import static org.assertj.core.api.Assertions.assertThat; -import static org.assertj.core.api.Assertions.assertThatExceptionOfType; -import static org.assertj.core.api.Assertions.assertThatIllegalArgumentException; -import static org.assertj.core.api.Assertions.assertThatIllegalStateException; -import static org.mockito.Mockito.any; -import static org.mockito.Mockito.mock; -import static org.mockito.Mockito.times; -import static org.mockito.Mockito.verify; -import static org.mockito.Mockito.when; -import static org.mockito.Mockito.withSettings; -import static org.springframework.data.redis.connection.ClusterTestVariables.CLUSTER_NODE_1; -import static org.springframework.data.redis.connection.RedisConfiguration.WithHostAndPort; -import static org.springframework.data.redis.test.extension.LettuceTestClientResources.getSharedClientResources; -import static org.springframework.test.util.ReflectionTestUtils.getField; +import static org.assertj.core.api.Assertions.*; +import static org.mockito.Mockito.*; +import static org.springframework.data.redis.connection.ClusterTestVariables.*; +import static org.springframework.data.redis.connection.RedisConfiguration.*; +import static org.springframework.data.redis.test.extension.LettuceTestClientResources.*; +import static org.springframework.test.util.ReflectionTestUtils.*; + +import io.lettuce.core.AbstractRedisClient; +import io.lettuce.core.ClientOptions; +import io.lettuce.core.RedisClient; +import io.lettuce.core.RedisURI; +import io.lettuce.core.api.StatefulConnection; +import io.lettuce.core.api.StatefulRedisConnection; +import io.lettuce.core.cluster.ClusterClientOptions; +import io.lettuce.core.cluster.RedisClusterClient; +import io.lettuce.core.cluster.api.StatefulRedisClusterConnection; +import io.lettuce.core.cluster.api.sync.RedisAdvancedClusterCommands; +import io.lettuce.core.codec.ByteArrayCodec; +import io.lettuce.core.codec.RedisCodec; +import io.lettuce.core.resource.ClientResources; +import reactor.test.StepVerifier; import java.time.Duration; import java.util.Collections; @@ -40,7 +47,6 @@ import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.Disabled; import org.junit.jupiter.api.Test; import org.mockito.ArgumentMatchers; - import org.springframework.beans.DirectFieldAccessor; import org.springframework.beans.factory.DisposableBean; import org.springframework.data.redis.ConnectionFactoryTracker; @@ -58,22 +64,6 @@ import org.springframework.data.redis.connection.RedisStandaloneConfiguration; import org.springframework.data.redis.test.extension.LettuceTestClientResources; import org.springframework.test.util.ReflectionTestUtils; -import io.lettuce.core.AbstractRedisClient; -import io.lettuce.core.ClientOptions; -import io.lettuce.core.RedisClient; -import io.lettuce.core.RedisURI; -import io.lettuce.core.api.StatefulConnection; -import io.lettuce.core.api.StatefulRedisConnection; -import io.lettuce.core.cluster.ClusterClientOptions; -import io.lettuce.core.cluster.RedisClusterClient; -import io.lettuce.core.cluster.api.StatefulRedisClusterConnection; -import io.lettuce.core.cluster.api.sync.RedisAdvancedClusterCommands; -import io.lettuce.core.codec.ByteArrayCodec; -import io.lettuce.core.codec.RedisCodec; -import io.lettuce.core.resource.ClientResources; - -import reactor.test.StepVerifier; - /** * Unit tests for {@link LettuceConnectionFactory}. * @@ -1244,7 +1234,7 @@ class LettuceConnectionFactoryUnitTests { connectionFactory.afterPropertiesSet(); assertThat(connectionFactory.isRunning()).isFalse(); - assertThatExceptionOfType(IllegalStateException.class).isThrownBy(() -> connectionFactory.getConnection()); + assertThatIllegalStateException().isThrownBy(() -> connectionFactory.getConnection()); } static class CustomRedisConfiguration implements RedisConfiguration, WithHostAndPort { diff --git a/src/test/java/org/springframework/data/redis/listener/RedisMessageListenerContainerUnitTests.java b/src/test/java/org/springframework/data/redis/listener/RedisMessageListenerContainerUnitTests.java index af60d8096..ebb268dcc 100644 --- a/src/test/java/org/springframework/data/redis/listener/RedisMessageListenerContainerUnitTests.java +++ b/src/test/java/org/springframework/data/redis/listener/RedisMessageListenerContainerUnitTests.java @@ -23,7 +23,6 @@ import java.util.concurrent.Executor; import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.Test; - import org.springframework.core.task.SyncTaskExecutor; import org.springframework.data.redis.RedisConnectionFailureException; import org.springframework.data.redis.connection.RedisConnection; @@ -150,6 +149,6 @@ class RedisMessageListenerContainerUnitTests { @Test // GH-964 void failsOnDuplicateInit() { - assertThatExceptionOfType(IllegalStateException.class).isThrownBy(() -> container.afterPropertiesSet()); + assertThatIllegalStateException().isThrownBy(() -> container.afterPropertiesSet()); } }