diff --git a/src/main/java/org/springframework/data/redis/connection/ClusterCommandExecutor.java b/src/main/java/org/springframework/data/redis/connection/ClusterCommandExecutor.java index ad38c7050..26d9cfcba 100644 --- a/src/main/java/org/springframework/data/redis/connection/ClusterCommandExecutor.java +++ b/src/main/java/org/springframework/data/redis/connection/ClusterCommandExecutor.java @@ -32,6 +32,7 @@ import org.springframework.data.redis.ExceptionTranslationStrategy; import org.springframework.data.redis.TooManyClusterRedirectionsException; import org.springframework.data.redis.connection.util.ByteArraySet; import org.springframework.data.redis.connection.util.ByteArrayWrapper; +import org.springframework.lang.NonNull; import org.springframework.lang.Nullable; import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor; import org.springframework.util.Assert; @@ -48,12 +49,18 @@ import org.springframework.util.ObjectUtils; */ public class ClusterCommandExecutor implements DisposableBean { - private AsyncTaskExecutor executor; - private final ClusterTopologyProvider topologyProvider; - private final ClusterNodeResourceProvider resourceProvider; - private final ExceptionTranslationStrategy exceptionTranslationStrategy; + protected static final AsyncTaskExecutor DEFAULT_TASK_EXECUTOR = new SimpleAsyncTaskExecutor(); + private int maxRedirects = 5; + private final AsyncTaskExecutor executor; + + private final ClusterNodeResourceProvider resourceProvider; + + private final ClusterTopologyProvider topologyProvider; + + private final ExceptionTranslationStrategy exceptionTranslationStrategy; + /** * Create a new instance of {@link ClusterCommandExecutor}. * @@ -64,13 +71,7 @@ public class ClusterCommandExecutor implements DisposableBean { public ClusterCommandExecutor(ClusterTopologyProvider topologyProvider, ClusterNodeResourceProvider resourceProvider, ExceptionTranslationStrategy exceptionTranslation) { - Assert.notNull(topologyProvider, "ClusterTopologyProvider must not be null"); - Assert.notNull(resourceProvider, "ClusterNodeResourceProvider must not be null"); - Assert.notNull(exceptionTranslation, "ExceptionTranslationStrategy must not be null"); - - this.topologyProvider = topologyProvider; - this.resourceProvider = resourceProvider; - this.exceptionTranslationStrategy = exceptionTranslation; + this(topologyProvider, resourceProvider, exceptionTranslation, DEFAULT_TASK_EXECUTOR); } /** @@ -82,27 +83,33 @@ public class ClusterCommandExecutor implements DisposableBean { public ClusterCommandExecutor(ClusterTopologyProvider topologyProvider, ClusterNodeResourceProvider resourceProvider, ExceptionTranslationStrategy exceptionTranslation, @Nullable AsyncTaskExecutor executor) { - this(topologyProvider, resourceProvider, exceptionTranslation); - this.executor = executor; + Assert.notNull(topologyProvider, "ClusterTopologyProvider must not be null"); + Assert.notNull(resourceProvider, "ClusterNodeResourceProvider must not be null"); + Assert.notNull(exceptionTranslation, "ExceptionTranslationStrategy must not be null"); + + this.topologyProvider = topologyProvider; + this.resourceProvider = resourceProvider; + this.exceptionTranslationStrategy = exceptionTranslation; + this.executor = resolveTaskExecutor(executor); } - { - if (executor == null) { - this.executor = new SimpleAsyncTaskExecutor(); - } + private @NonNull AsyncTaskExecutor resolveTaskExecutor(@Nullable AsyncTaskExecutor taskExecutor) { + return taskExecutor != null ? taskExecutor : DEFAULT_TASK_EXECUTOR; } /** * Run {@link ClusterCommandCallback} on a random node. * - * @param cmd must not be {@literal null}. + * @param commandCallback must not be {@literal null}. * @return never {@literal null}. */ - public NodeResult executeCommandOnArbitraryNode(ClusterCommandCallback cmd) { + public NodeResult executeCommandOnArbitraryNode(ClusterCommandCallback commandCallback) { + + Assert.notNull(commandCallback, "ClusterCommandCallback must not be null"); - Assert.notNull(cmd, "ClusterCommandCallback must not be null"); List nodes = new ArrayList<>(getClusterTopology().getActiveNodes()); - return executeCommandOnSingleNode(cmd, nodes.get(new Random().nextInt(nodes.size()))); + + return executeCommandOnSingleNode(commandCallback, nodes.get(new Random().nextInt(nodes.size()))); } /** @@ -110,8 +117,8 @@ public class ClusterCommandExecutor implements DisposableBean { * * @param cmd must not be {@literal null}. * @param node must not be {@literal null}. + * @return the {@link NodeResult} from the single, targeted {@link RedisClusterNode}. * @throws IllegalArgumentException in case no resource can be acquired for given node. - * @return */ public NodeResult executeCommandOnSingleNode(ClusterCommandCallback cmd, RedisClusterNode node) { return executeCommandOnSingleNode(cmd, node, 0); @@ -132,19 +139,21 @@ public class ClusterCommandExecutor implements DisposableBean { RedisClusterNode nodeToUse = lookupNode(node); S client = this.resourceProvider.getResourceForSpecificNode(nodeToUse); + Assert.notNull(client, "Could not acquire resource for node; Is your cluster info up to date"); try { return new NodeResult<>(node, cmd.doInCluster(client)); - } catch (RuntimeException ex) { + } catch (RuntimeException cause) { - RuntimeException translatedException = convertToDataAccessException(ex); - if (translatedException instanceof ClusterRedirectException) { - ClusterRedirectException cre = (ClusterRedirectException) translatedException; - return executeCommandOnSingleNode(cmd, - topologyProvider.getTopology().lookup(cre.getTargetHost(), cre.getTargetPort()), redirectCount + 1); + RuntimeException translatedException = convertToDataAccessException(cause); + + if (translatedException instanceof ClusterRedirectException clusterRedirectException) { + return executeCommandOnSingleNode(cmd, topologyProvider.getTopology() + .lookup(clusterRedirectException.getTargetHost(), clusterRedirectException.getTargetPort()), + redirectCount + 1); } else { - throw translatedException != null ? translatedException : ex; + throw translatedException != null ? translatedException : cause; } } finally { this.resourceProvider.returnResourceForSpecificNode(nodeToUse, client); @@ -159,10 +168,11 @@ public class ClusterCommandExecutor implements DisposableBean { * @throws IllegalArgumentException in case the node could not be resolved to a topology-known node */ private RedisClusterNode lookupNode(RedisClusterNode node) { + try { return topologyProvider.getTopology().lookup(node); - } catch (ClusterStateFailureException e) { - throw new IllegalArgumentException(String.format("Node %s is unknown to cluster", node), e); + } catch (ClusterStateFailureException cause) { + throw new IllegalArgumentException(String.format("Node %s is unknown to cluster", node), cause); } } @@ -171,7 +181,8 @@ public class ClusterCommandExecutor implements DisposableBean { * * @param cmd must not be {@literal null}. * @return never {@literal null}. - * @throws ClusterCommandExecutionFailureException + * @throws ClusterCommandExecutionFailureException if a failure occurs while executing the given + * {@link ClusterCommandCallback command} on any given {@link RedisClusterNode node}. */ public MultiNodeResult executeCommandOnAllNodes(final ClusterCommandCallback cmd) { return executeCommandAsyncOnNodes(cmd, getClusterTopology().getActiveMasterNodes()); @@ -181,7 +192,8 @@ public class ClusterCommandExecutor implements DisposableBean { * @param callback must not be {@literal null}. * @param nodes must not be {@literal null}. * @return never {@literal null}. - * @throws ClusterCommandExecutionFailureException + * @throws ClusterCommandExecutionFailureException if a failure occurs while executing the given + * {@link ClusterCommandCallback command} on any given {@link RedisClusterNode node}. * @throws IllegalArgumentException in case the node could not be resolved to a topology-known node */ public MultiNodeResult executeCommandAsyncOnNodes(ClusterCommandCallback callback, @@ -202,6 +214,7 @@ public class ClusterCommandExecutor implements DisposableBean { } Map>> futures = new LinkedHashMap<>(); + for (RedisClusterNode node : resolvedRedisClusterNodes) { futures.put(new NodeExecution(node), executor.submit(() -> executeCommandOnSingleNode(callback, node))); } @@ -213,10 +226,10 @@ public class ClusterCommandExecutor implements DisposableBean { boolean done = false; - MultiNodeResult result = new MultiNodeResult<>(); Map exceptions = new HashMap<>(); - + MultiNodeResult result = new MultiNodeResult<>(); Set saveGuard = new HashSet<>(); + while (!done) { done = true; @@ -242,6 +255,7 @@ public class ClusterCommandExecutor implements DisposableBean { } catch (ExecutionException e) { RuntimeException ex = convertToDataAccessException((Exception) e.getCause()); + exceptions.put(execution.getNode(), ex != null ? ex : e.getCause()); } catch (InterruptedException e) { @@ -253,6 +267,7 @@ public class ClusterCommandExecutor implements DisposableBean { } } } + try { Thread.sleep(10); } catch (InterruptedException e) { @@ -273,14 +288,15 @@ public class ClusterCommandExecutor implements DisposableBean { * * @param cmd must not be {@literal null}. * @return never {@literal null}. - * @throws ClusterCommandExecutionFailureException + * @throws ClusterCommandExecutionFailureException if a failure occurs while executing the given + * {@link MultiKeyClusterCommandCallback command}. */ public MultiNodeResult executeMultiKeyCommand(MultiKeyClusterCommandCallback cmd, Iterable keys) { Map nodeKeyMap = new HashMap<>(); - int index = 0; + for (byte[] key : keys) { for (RedisClusterNode node : getClusterTopology().getKeyServingNodes(key)) { nodeKeyMap.computeIfAbsent(node, val -> PositionalKeys.empty()).append(PositionalKey.of(key, index++)); @@ -288,6 +304,7 @@ public class ClusterCommandExecutor implements DisposableBean { } Map>> futures = new LinkedHashMap<>(); + for (Entry entry : nodeKeyMap.entrySet()) { if (entry.getKey().isMaster()) { @@ -309,6 +326,7 @@ public class ClusterCommandExecutor implements DisposableBean { Assert.notNull(key, "Keys for execution must not be null"); S client = this.resourceProvider.getResourceForSpecificNode(node); + Assert.notNull(client, "Could not acquire resource for node; Is your cluster info up to date"); try { @@ -479,7 +497,9 @@ public class ClusterCommandExecutor implements DisposableBean { } /** - * @return + * Returns the key as an array of bytes. + * + * @return the key as an array of bytes. */ public byte[] getKey() { return key.getArray(); 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 1251399e0..96b7ce01f 100644 --- a/src/main/java/org/springframework/data/redis/connection/RedisConnectionFactory.java +++ b/src/main/java/org/springframework/data/redis/connection/RedisConnectionFactory.java @@ -26,6 +26,18 @@ import org.springframework.dao.support.PersistenceExceptionTranslator; */ 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. + * + * @return {@code true} to convert pipeline and transaction results; {@code false} otherwise. + */ + boolean getConvertPipelineAndTxResults(); + /** * Returns a suitable {@link RedisConnection connection} for interacting with Redis. * @@ -45,18 +57,6 @@ public interface RedisConnectionFactory extends PersistenceExceptionTranslator { */ RedisClusterConnection getClusterConnection(); - /** - * 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. - * - * @return {@code true} to convert pipeline and transaction results; {@code false} otherwise. - */ - boolean getConvertPipelineAndTxResults(); - /** * Returns a suitable {@link RedisSentinelConnection connection} for interacting with Redis Sentinel. * diff --git a/src/main/java/org/springframework/data/redis/connection/jedis/JedisConnection.java b/src/main/java/org/springframework/data/redis/connection/jedis/JedisConnection.java index 0c23bb171..eaa1b5ba4 100644 --- a/src/main/java/org/springframework/data/redis/connection/jedis/JedisConnection.java +++ b/src/main/java/org/springframework/data/redis/connection/jedis/JedisConnection.java @@ -15,21 +15,6 @@ */ package org.springframework.data.redis.connection.jedis; -import redis.clients.jedis.BuilderFactory; -import redis.clients.jedis.CommandArguments; -import redis.clients.jedis.CommandObject; -import redis.clients.jedis.DefaultJedisClientConfig; -import redis.clients.jedis.HostAndPort; -import redis.clients.jedis.Jedis; -import redis.clients.jedis.JedisClientConfig; -import redis.clients.jedis.Pipeline; -import redis.clients.jedis.Response; -import redis.clients.jedis.Transaction; -import redis.clients.jedis.commands.ProtocolCommand; -import redis.clients.jedis.commands.ServerCommands; -import redis.clients.jedis.exceptions.JedisDataException; -import redis.clients.jedis.util.Pool; - import java.util.ArrayList; import java.util.Collections; import java.util.LinkedList; @@ -47,7 +32,25 @@ import org.springframework.dao.InvalidDataAccessApiUsageException; import org.springframework.data.redis.ExceptionTranslationStrategy; import org.springframework.data.redis.FallbackExceptionTranslationStrategy; import org.springframework.data.redis.RedisSystemException; -import org.springframework.data.redis.connection.*; +import org.springframework.data.redis.connection.AbstractRedisConnection; +import org.springframework.data.redis.connection.FutureResult; +import org.springframework.data.redis.connection.MessageListener; +import org.springframework.data.redis.connection.RedisCommands; +import org.springframework.data.redis.connection.RedisGeoCommands; +import org.springframework.data.redis.connection.RedisHashCommands; +import org.springframework.data.redis.connection.RedisHyperLogLogCommands; +import org.springframework.data.redis.connection.RedisKeyCommands; +import org.springframework.data.redis.connection.RedisListCommands; +import org.springframework.data.redis.connection.RedisNode; +import org.springframework.data.redis.connection.RedisPipelineException; +import org.springframework.data.redis.connection.RedisScriptingCommands; +import org.springframework.data.redis.connection.RedisServerCommands; +import org.springframework.data.redis.connection.RedisSetCommands; +import org.springframework.data.redis.connection.RedisStreamCommands; +import org.springframework.data.redis.connection.RedisStringCommands; +import org.springframework.data.redis.connection.RedisSubscribedConnectionException; +import org.springframework.data.redis.connection.RedisZSetCommands; +import org.springframework.data.redis.connection.Subscription; import org.springframework.data.redis.connection.convert.TransactionResultConverter; import org.springframework.data.redis.connection.jedis.JedisInvoker.ResponseCommands; import org.springframework.data.redis.connection.jedis.JedisResult.JedisResultBuilder; @@ -56,6 +59,24 @@ import org.springframework.lang.Nullable; import org.springframework.util.Assert; import org.springframework.util.CollectionUtils; +import org.apache.commons.logging.Log; +import org.apache.commons.logging.LogFactory; + +import redis.clients.jedis.BuilderFactory; +import redis.clients.jedis.CommandArguments; +import redis.clients.jedis.CommandObject; +import redis.clients.jedis.DefaultJedisClientConfig; +import redis.clients.jedis.HostAndPort; +import redis.clients.jedis.Jedis; +import redis.clients.jedis.JedisClientConfig; +import redis.clients.jedis.Pipeline; +import redis.clients.jedis.Response; +import redis.clients.jedis.Transaction; +import redis.clients.jedis.commands.ProtocolCommand; +import redis.clients.jedis.commands.ServerCommands; +import redis.clients.jedis.exceptions.JedisDataException; +import redis.clients.jedis.util.Pool; + /** * {@code RedisConnection} implementation on top of Jedis library. *

@@ -78,18 +99,23 @@ import org.springframework.util.CollectionUtils; */ public class JedisConnection extends AbstractRedisConnection { - private final Log LOGGER = LogFactory.getLog(getClass()); + private static final ExceptionTranslationStrategy EXCEPTION_TRANSLATION = + new FallbackExceptionTranslationStrategy(JedisExceptionConverter.INSTANCE); - private static final ExceptionTranslationStrategy EXCEPTION_TRANSLATION = new FallbackExceptionTranslationStrategy( - JedisExceptionConverter.INSTANCE); + private boolean convertPipelineAndTxResults = true; private final Jedis jedis; + private final JedisClientConfig sentinelConfig; + private final JedisInvoker invoker = new JedisInvoker((directFunction, pipelineFunction, converter, nullDefault) -> doInvoke(false, directFunction, pipelineFunction, converter, nullDefault)); + private final JedisInvoker statusInvoker = new JedisInvoker((directFunction, pipelineFunction, converter, nullDefault) -> doInvoke(true, directFunction, pipelineFunction, converter, nullDefault)); + private volatile @Nullable JedisSubscription subscription; + private final JedisGeoCommands geoCommands = new JedisGeoCommands(this); private final JedisHashCommands hashCommands = new JedisHashCommands(this); private final JedisHyperLogLogCommands hllCommands = new JedisHyperLogLogCommands(this); @@ -102,62 +128,58 @@ public class JedisConnection extends AbstractRedisConnection { private final JedisStringCommands stringCommands = new JedisStringCommands(this); private final JedisZSetCommands zSetCommands = new JedisZSetCommands(this); - private final @Nullable Pool pool; - private final JedisClientConfig sentinelConfig; + private final Log LOGGER = LogFactory.getLog(getClass()); private List pipelinedResults = new ArrayList<>(); + + private final @Nullable Pool pool; + private Queue>> txResults = new LinkedList<>(); - private volatile @Nullable JedisSubscription subscription; - private volatile @Nullable Transaction transaction; private volatile @Nullable Pipeline pipeline; - private boolean convertPipelineAndTxResults = true; + private volatile @Nullable Transaction transaction; /** - * Constructs a new JedisConnection instance. + * Constructs a new {@link JedisConnection}. * - * @param jedis Jedis entity + * @param jedis {@link Jedis} client. */ public JedisConnection(Jedis jedis) { this(jedis, null, 0); } /** - * Constructs a new JedisConnection instance backed by a jedis pool. + * Constructs a new <{@link JedisConnection} backed by a Jedis {@link Pool}. * - * @param jedis - * @param pool can be null, if no pool is used - * @param dbIndex + * @param jedis {@link Jedis} client. + * @param pool {@link Pool} of Redis connections; can be null, if no pool is used. + * @param dbIndex {@link Integer index} of the Redis database to use. */ public JedisConnection(Jedis jedis, Pool pool, int dbIndex) { this(jedis, pool, dbIndex, null); } /** - * Constructs a new JedisConnection instance backed by a jedis pool. + * Constructs a new <{@link JedisConnection} backed by a Jedis {@link Pool}. * - * @param jedis - * @param pool can be null, if no pool is used - * @param dbIndex - * @param clientName the client name, can be {@literal null}. + * @param jedis {@link Jedis} client. + * @param pool {@link Pool} of Redis connections; can be null, if no pool is used. + * @param dbIndex {@link Integer index} of the Redis database to use. + * @param clientName {@link String name} given to this client; can be {@literal null}. * @since 1.8 */ protected JedisConnection(Jedis jedis, @Nullable Pool pool, int dbIndex, @Nullable String clientName) { this(jedis, pool, createConfig(dbIndex, clientName), createConfig(dbIndex, clientName)); } - private static DefaultJedisClientConfig createConfig(int dbIndex, @Nullable String clientName) { - return DefaultJedisClientConfig.builder().database(dbIndex).clientName(clientName).build(); - } - /** - * Constructs a new JedisConnection instance backed by a jedis pool. + * Constructs a new <{@link JedisConnection} backed by a Jedis {@link Pool}. * - * @param jedis - * @param pool can be null, if no pool is used - * @param nodeConfig node configuration - * @param sentinelConfig sentinel configuration + * @param jedis {@link Jedis} client. + * @param pool {@link Pool} of Redis connections; can be null, if no pool is used. + * @param nodeConfig {@literal Redis Node} configuration + * @param sentinelConfig {@literal Redis Sentinel} configuration * @since 2.5 */ protected JedisConnection(Jedis jedis, @Nullable Pool pool, JedisClientConfig nodeConfig, @@ -173,13 +195,17 @@ public class JedisConnection extends AbstractRedisConnection { if (nodeConfig.getDatabase() != jedis.getDB()) { try { select(nodeConfig.getDatabase()); - } catch (DataAccessException ex) { + } catch (DataAccessException cause) { close(); - throw ex; + throw cause; } } } + private static DefaultJedisClientConfig createConfig(int dbIndex, @Nullable String clientName) { + return DefaultJedisClientConfig.builder().database(dbIndex).clientName(clientName).build(); + } + @Nullable private Object doInvoke(boolean status, Function directFunction, Function> pipelineFunction, Converter converter, @@ -211,9 +237,9 @@ public class JedisConnection extends AbstractRedisConnection { }); } - protected DataAccessException convertJedisAccessException(Exception ex) { - DataAccessException exception = EXCEPTION_TRANSLATION.translate(ex); - return exception != null ? exception : new RedisSystemException(ex.getMessage(), ex); + protected DataAccessException convertJedisAccessException(Exception cause) { + DataAccessException exception = EXCEPTION_TRANSLATION.translate(cause); + return exception != null ? exception : new RedisSystemException(cause.getMessage(), cause); } @Override @@ -290,6 +316,7 @@ public class JedisConnection extends AbstractRedisConnection { CommandArguments arguments = new CommandArguments(protocolCommand).addObjects(args); CommandObject commandObject = new CommandObject<>(arguments, BuilderFactory.RAW_OBJECT); + if (isPipelined()) { pipeline(newJedisResult(getRequiredPipeline().executeCommand(commandObject))); } else { @@ -308,64 +335,42 @@ public class JedisConnection extends AbstractRedisConnection { super.close(); JedisSubscription subscription = this.subscription; - try { - if (subscription != null) { - subscription.close(); - } - } catch (Exception ex) { - LOGGER.debug("Cannot terminate subscription", ex); - } finally { + + if (subscription != null) { + doExceptionThrowingOperationSafely(subscription::close, "Cannot terminate subscription"); this.subscription = null; } - // return the connection to the pool - if (pool != null) { + Jedis jedis = getJedis(); + + // Return connection to the pool + if (this.pool != null) { jedis.close(); - return; } - - // else close the connection normally (doing the try/catch dance) - - try { - jedis.quit(); - } catch (Exception ex) { - LOGGER.debug("Failed to QUIT during close", ex); + else { + doExceptionThrowingOperationSafely(jedis::quit, "Failed to quit during close"); + doExceptionThrowingOperationSafely(jedis::disconnect, "Failed to disconnect during close"); } - - try { - jedis.disconnect(); - } catch (Exception ex) { - LOGGER.debug("Failed to disconnect during close", ex); - } - } - - private Exception handleCloseException(@Nullable Exception exceptionToThrow, Exception cause) { - - if (exceptionToThrow == null) { - return cause; - } - - return exceptionToThrow; } @Override public Jedis getNativeConnection() { - return jedis; + return this.jedis; } @Override public boolean isClosed() { - return doWithJedis(it -> !it.isConnected()); + return !Boolean.TRUE.equals(doWithJedis(Jedis::isConnected)); } @Override public boolean isQueueing() { - return transaction != null; + return this.transaction != null; } @Override public boolean isPipelined() { - return pipeline != null; + return this.pipeline != null; } @Override @@ -382,6 +387,7 @@ public class JedisConnection extends AbstractRedisConnection { @Override public List closePipeline() { + if (pipeline != null) { try { return convertPipelineResults(); @@ -390,14 +396,19 @@ public class JedisConnection extends AbstractRedisConnection { pipelinedResults.clear(); } } + return Collections.emptyList(); } private List convertPipelineResults() { + List results = new ArrayList<>(); + getRequiredPipeline().sync(); + Exception cause = null; - for (JedisResult result : pipelinedResults) { + + for (JedisResult result : pipelinedResults) { try { Object data = result.get(); @@ -418,13 +429,16 @@ public class JedisConnection extends AbstractRedisConnection { results.add(e); } } + if (cause != null) { throw new RedisPipelineException(cause, results); } + return results; } - void pipeline(JedisResult result) { + void pipeline(JedisResult result) { + if (isQueueing()) { transaction(result); } else { @@ -441,7 +455,7 @@ public class JedisConnection extends AbstractRedisConnection { Assert.notNull(message, "Message must not be null"); - return invoke().just(j -> j.echo(message)); + return invoke().just(jedis -> jedis.echo(message)); } @Override @@ -451,6 +465,7 @@ public class JedisConnection extends AbstractRedisConnection { @Override public void discard() { + try { getRequiredTransaction().discard(); } catch (Exception ex) { @@ -463,8 +478,8 @@ public class JedisConnection extends AbstractRedisConnection { @Override public List exec() { - try { + try { if (transaction == null) { throw new InvalidDataAccessApiUsageException("No ongoing transaction; Did you forget to call multi"); } @@ -474,8 +489,8 @@ public class JedisConnection extends AbstractRedisConnection { return !CollectionUtils.isEmpty(results) ? new TransactionResultConverter<>(txResults, JedisExceptionConverter.INSTANCE).convert(results) : results; - } catch (Exception ex) { - throw convertJedisAccessException(ex); + } catch (Exception cause) { + throw convertJedisAccessException(cause); } finally { txResults.clear(); transaction = null; @@ -484,38 +499,34 @@ public class JedisConnection extends AbstractRedisConnection { @Nullable public Pipeline getPipeline() { - return pipeline; + return this.pipeline; } public Pipeline getRequiredPipeline() { Pipeline pipeline = getPipeline(); - if (pipeline == null) { - throw new IllegalStateException("Connection has no active pipeline"); - } + Assert.state(pipeline != null, "Connection has no active pipeline"); return pipeline; } @Nullable public Transaction getTransaction() { - return transaction; + return this.transaction; } public Transaction getRequiredTransaction() { Transaction transaction = getTransaction(); - if (transaction == null) { - throw new IllegalStateException("Connection has no active transaction"); - } + Assert.state(transaction != null, "Connection has no active transaction"); return transaction; } public Jedis getJedis() { - return jedis; + return this.jedis; } /** @@ -525,7 +536,7 @@ public class JedisConnection extends AbstractRedisConnection { * @since 2.5 */ JedisInvoker invoke() { - return invoker; + return this.invoker; } /** @@ -536,7 +547,7 @@ public class JedisConnection extends AbstractRedisConnection { * @since 2.5 */ JedisInvoker invokeStatus() { - return statusInvoker; + return this.statusInvoker; } JedisResult newJedisResult(Response response) { @@ -555,6 +566,7 @@ public class JedisConnection extends AbstractRedisConnection { @Override public void multi() { + if (isQueueing()) { return; } @@ -563,8 +575,8 @@ public class JedisConnection extends AbstractRedisConnection { throw new InvalidDataAccessApiUsageException("Cannot use Transaction while a pipeline is open"); } - doWithJedis(it -> { - this.transaction = it.multi(); + doWithJedis(jedis -> { + this.transaction = jedis.multi(); }); } @@ -580,13 +592,14 @@ public class JedisConnection extends AbstractRedisConnection { @Override public void watch(byte[]... keys) { + if (isQueueing()) { throw new InvalidDataAccessApiUsageException("WATCH is not supported when a transaction is active"); } - doWithJedis(it -> { + doWithJedis(jedis -> { for (byte[] key : keys) { - it.watch(key); + jedis.watch(key); } }); } @@ -597,17 +610,18 @@ public class JedisConnection extends AbstractRedisConnection { @Override public Long publish(byte[] channel, byte[] message) { - return invoke().just(j -> j.publish(channel, message)); + return invoke().just(jedis -> jedis.publish(channel, message)); } @Override public Subscription getSubscription() { - return subscription; + return this.subscription; } @Override public boolean isSubscribed() { - return (subscription != null && subscription.isAlive()); + Subscription subscription = getSubscription(); + return subscription != null && subscription.isAlive(); } @Override @@ -666,11 +680,12 @@ public class JedisConnection extends AbstractRedisConnection { protected boolean isActive(RedisNode node) { Jedis verification = null; + try { verification = getJedis(node); verification.connect(); return verification.ping().equalsIgnoreCase("pong"); - } catch (Exception e) { + } catch (Exception cause) { return false; } finally { if (verification != null) { @@ -694,9 +709,8 @@ public class JedisConnection extends AbstractRedisConnection { try { return callback.apply(getJedis()); - - } catch (Exception ex) { - throw convertJedisAccessException(ex); + } catch (Exception cause) { + throw convertJedisAccessException(cause); } } @@ -704,9 +718,26 @@ public class JedisConnection extends AbstractRedisConnection { try { callback.accept(getJedis()); - } catch (Exception ex) { - throw convertJedisAccessException(ex); + } catch (Exception cause) { + throw convertJedisAccessException(cause); } } + private void doExceptionThrowingOperationSafely(ExceptionThrowingOperation operation, String logMessage) { + + try { + operation.run(); + } + catch (Exception cause) { + if (LOGGER.isDebugEnabled()) { + LOGGER.debug(logMessage, cause); + } + } + } + + + @FunctionalInterface + private interface ExceptionThrowingOperation { + void run() throws Exception; + } } 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 abcaa1fd4..e620b6671 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 @@ -98,19 +98,32 @@ public class JedisConnectionFactory implements RedisConnectionFactory, InitializingBean, DisposableBean, SmartLifecycle { private static final Log log = LogFactory.getLog(JedisConnectionFactory.class); - private static final ExceptionTranslationStrategy EXCEPTION_TRANSLATION = new PassThroughExceptionTranslationStrategy( - JedisExceptionConverter.INSTANCE); - private final JedisClientConfiguration clientConfiguration; + private static final ExceptionTranslationStrategy EXCEPTION_TRANSLATION = + new PassThroughExceptionTranslationStrategy(JedisExceptionConverter.INSTANCE); - private RedisStandaloneConfiguration standaloneConfig = new RedisStandaloneConfiguration("localhost", - Protocol.DEFAULT_PORT); - - private @Nullable RedisConfiguration configuration; + private boolean convertPipelineAndTxResults = true; private int phase = 0; // in between min and max values - private boolean convertPipelineAndTxResults = true; + private final AtomicReference state = new AtomicReference<>(State.CREATED); + + private @Nullable ClusterCommandExecutor clusterCommandExecutor; + + private @Nullable ClusterTopologyProvider topologyProvider; + + private JedisClientConfig clientConfig = DefaultJedisClientConfig.builder().build(); + + private final JedisClientConfiguration clientConfiguration; + + private @Nullable JedisCluster cluster; + + private @Nullable Pool pool; + + private @Nullable RedisConfiguration configuration; + + private RedisStandaloneConfiguration standaloneConfig = + new RedisStandaloneConfiguration("localhost", Protocol.DEFAULT_PORT); /** * Lifecycle state of this factory. @@ -119,15 +132,6 @@ public class JedisConnectionFactory CREATED, STARTING, STARTED, STOPPING, STOPPED, DESTROYED; } - 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). */ @@ -138,14 +142,14 @@ public class JedisConnectionFactory /** * Constructs a new {@link JedisConnectionFactory} instance given {@link JedisClientConfiguration}. * - * @param clientConfig must not be {@literal null} + * @param clientConfiguration must not be {@literal null} * @since 2.0 */ - private JedisConnectionFactory(JedisClientConfiguration clientConfig) { + private JedisConnectionFactory(JedisClientConfiguration clientConfiguration) { - Assert.notNull(clientConfig, "JedisClientConfiguration must not be null"); + Assert.notNull(clientConfiguration, "JedisClientConfiguration must not be null"); - this.clientConfiguration = clientConfig; + this.clientConfiguration = clientConfiguration; } /** @@ -157,139 +161,143 @@ public class JedisConnectionFactory this((RedisSentinelConfiguration) null, poolConfig); } - /** - * Constructs a new {@link JedisConnectionFactory} instance using the given {@link JedisPoolConfig} applied to - * {@link JedisSentinelPool}. - * - * @param sentinelConfig must not be {@literal null}. - * @since 1.4 - */ - public JedisConnectionFactory(RedisSentinelConfiguration sentinelConfig) { - this(sentinelConfig, new MutableJedisClientConfiguration()); - } - - /** - * Constructs a new {@link JedisConnectionFactory} instance using the given {@link JedisPoolConfig} applied to - * {@link JedisSentinelPool}. - * - * @param sentinelConfig the sentinel configuration to use. - * @param poolConfig pool configuration. Defaulted to new instance if {@literal null}. - * @since 1.4 - */ - public JedisConnectionFactory(RedisSentinelConfiguration sentinelConfig, @Nullable JedisPoolConfig poolConfig) { - - this.configuration = sentinelConfig; - this.clientConfiguration = MutableJedisClientConfiguration - .create(poolConfig != null ? poolConfig : new JedisPoolConfig()); - } - /** * Constructs a new {@link JedisConnectionFactory} instance using the given {@link RedisClusterConfiguration} applied * to create a {@link JedisCluster}. * - * @param clusterConfig must not be {@literal null}. + * @param clusterConfiguration must not be {@literal null}. * @since 1.7 */ - public JedisConnectionFactory(RedisClusterConfiguration clusterConfig) { - this(clusterConfig, new MutableJedisClientConfiguration()); - } - - /** - * Constructs a new {@link JedisConnectionFactory} instance using the given {@link RedisClusterConfiguration} applied - * to create a {@link JedisCluster}. - * - * @param clusterConfig must not be {@literal null}. - * @since 1.7 - */ - public JedisConnectionFactory(RedisClusterConfiguration clusterConfig, JedisPoolConfig poolConfig) { - - Assert.notNull(clusterConfig, "RedisClusterConfiguration must not be null"); - - this.configuration = clusterConfig; - this.clientConfiguration = MutableJedisClientConfiguration.create(poolConfig); - } - - /** - * Constructs a new {@link JedisConnectionFactory} instance using the given {@link RedisStandaloneConfiguration}. - * - * @param standaloneConfig must not be {@literal null}. - * @since 2.0 - */ - public JedisConnectionFactory(RedisStandaloneConfiguration standaloneConfig) { - this(standaloneConfig, new MutableJedisClientConfiguration()); - } - - /** - * Constructs a new {@link JedisConnectionFactory} instance using the given {@link RedisStandaloneConfiguration} and - * {@link JedisClientConfiguration}. - * - * @param standaloneConfig must not be {@literal null}. - * @param clientConfig must not be {@literal null}. - * @since 2.0 - */ - public JedisConnectionFactory(RedisStandaloneConfiguration standaloneConfig, JedisClientConfiguration clientConfig) { - - this(clientConfig); - - Assert.notNull(standaloneConfig, "RedisStandaloneConfiguration must not be null"); - - this.standaloneConfig = standaloneConfig; - } - - /** - * Constructs a new {@link JedisConnectionFactory} instance using the given {@link RedisSentinelConfiguration} and - * {@link JedisClientConfiguration}. - * - * @param sentinelConfig must not be {@literal null}. - * @param clientConfig must not be {@literal null}. - * @since 2.0 - */ - public JedisConnectionFactory(RedisSentinelConfiguration sentinelConfig, JedisClientConfiguration clientConfig) { - - this(clientConfig); - - Assert.notNull(sentinelConfig, "RedisSentinelConfiguration must not be null"); - - this.configuration = sentinelConfig; + public JedisConnectionFactory(RedisClusterConfiguration clusterConfiguration) { + this(clusterConfiguration, new MutableJedisClientConfiguration()); } /** * Constructs a new {@link JedisConnectionFactory} instance using the given {@link RedisClusterConfiguration} and * {@link JedisClientConfiguration}. * - * @param clusterConfig must not be {@literal null}. - * @param clientConfig must not be {@literal null}. + * @param clusterConfiguration must not be {@literal null}. + * @param clientConfiguration must not be {@literal null}. * @since 2.0 */ - public JedisConnectionFactory(RedisClusterConfiguration clusterConfig, JedisClientConfiguration clientConfig) { + public JedisConnectionFactory(RedisClusterConfiguration clusterConfiguration, + JedisClientConfiguration clientConfiguration) { - this(clientConfig); + this(clientConfiguration); - Assert.notNull(clusterConfig, "RedisClusterConfiguration must not be null"); + Assert.notNull(clusterConfiguration, "RedisClusterConfiguration must not be null"); - this.configuration = clusterConfig; + this.configuration = clusterConfiguration; + } + + /** + * Constructs a new {@link JedisConnectionFactory} instance using the given {@link RedisClusterConfiguration} applied + * to create a {@link JedisCluster}. + * + * @param clusterConfiguration must not be {@literal null}. + * @since 1.7 + */ + public JedisConnectionFactory(RedisClusterConfiguration clusterConfiguration, JedisPoolConfig poolConfig) { + + Assert.notNull(clusterConfiguration, "RedisClusterConfiguration must not be null"); + + this.configuration = clusterConfiguration; + this.clientConfiguration = MutableJedisClientConfiguration.create(poolConfig); + } + + /** + * Constructs a new {@link JedisConnectionFactory} instance using the given {@link JedisPoolConfig} applied to + * {@link JedisSentinelPool}. + * + * @param sentinelConfiguration must not be {@literal null}. + * @since 1.4 + */ + public JedisConnectionFactory(RedisSentinelConfiguration sentinelConfiguration) { + this(sentinelConfiguration, new MutableJedisClientConfiguration()); + } + + /** + * Constructs a new {@link JedisConnectionFactory} instance using the given {@link RedisSentinelConfiguration} and + * {@link JedisClientConfiguration}. + * + * @param sentinelConfiguration must not be {@literal null}. + * @param clientConfiguration must not be {@literal null}. + * @since 2.0 + */ + public JedisConnectionFactory(RedisSentinelConfiguration sentinelConfiguration, + JedisClientConfiguration clientConfiguration) { + + this(clientConfiguration); + + Assert.notNull(sentinelConfiguration, "RedisSentinelConfiguration must not be null"); + + this.configuration = sentinelConfiguration; + } + + /** + * Constructs a new {@link JedisConnectionFactory} instance using the given {@link JedisPoolConfig} applied to + * {@link JedisSentinelPool}. + * + * @param sentinelConfiguration the sentinel configuration to use. + * @param poolConfig pool configuration. Defaulted to new instance if {@literal null}. + * @since 1.4 + */ + public JedisConnectionFactory(RedisSentinelConfiguration sentinelConfiguration, + @Nullable JedisPoolConfig poolConfig) { + + this.configuration = sentinelConfiguration; + this.clientConfiguration = MutableJedisClientConfiguration + .create(poolConfig != null ? poolConfig : new JedisPoolConfig()); + } + + /** + * Constructs a new {@link JedisConnectionFactory} instance using the given {@link RedisStandaloneConfiguration}. + * + * @param standaloneConfiguration must not be {@literal null}. + * @since 2.0 + */ + public JedisConnectionFactory(RedisStandaloneConfiguration standaloneConfiguration) { + this(standaloneConfiguration, new MutableJedisClientConfiguration()); + } + + /** + * Constructs a new {@link JedisConnectionFactory} instance using the given {@link RedisStandaloneConfiguration} and + * {@link JedisClientConfiguration}. + * + * @param standaloneConfiguration must not be {@literal null}. + * @param clientConfiguration must not be {@literal null}. + * @since 2.0 + */ + public JedisConnectionFactory(RedisStandaloneConfiguration standaloneConfiguration, + JedisClientConfiguration clientConfiguration) { + + this(clientConfiguration); + + Assert.notNull(standaloneConfiguration, "RedisStandaloneConfiguration must not be null"); + + this.standaloneConfig = standaloneConfiguration; + } + + @Nullable + protected ClusterCommandExecutor getClusterCommandExecutor() { + return this.clusterCommandExecutor; } @Override public void afterPropertiesSet() { - clientConfig = createClientConfig(getDatabase(), getRedisUsername(), getRedisPassword()); + this.clientConfig = createClientConfig(getDatabase(), getRedisUsername(), getRedisPassword()); if (isAutoStartup()) { start(); } } - JedisClientConfig createSentinelClientConfig(SentinelConfiguration sentinelConfiguration) { - return createClientConfig(0, sentinelConfiguration.getSentinelUsername(), - sentinelConfiguration.getSentinelPassword()); - } - private JedisClientConfig createClientConfig(int database, @Nullable String username, RedisPassword password) { DefaultJedisClientConfig.Builder builder = DefaultJedisClientConfig.builder(); - clientConfiguration.getClientName().ifPresent(builder::clientName); + this.clientConfiguration.getClientName().ifPresent(builder::clientName); builder.connectionTimeoutMillis(getConnectTimeout()); builder.socketTimeoutMillis(getReadTimeout()); @@ -304,21 +312,25 @@ public class JedisConnectionFactory builder.ssl(true); - clientConfiguration.getSslSocketFactory().ifPresent(builder::sslSocketFactory); - clientConfiguration.getHostnameVerifier().ifPresent(builder::hostnameVerifier); - clientConfiguration.getSslParameters().ifPresent(builder::sslParameters); + this.clientConfiguration.getSslSocketFactory().ifPresent(builder::sslSocketFactory); + this.clientConfiguration.getHostnameVerifier().ifPresent(builder::hostnameVerifier); + this.clientConfiguration.getSslParameters().ifPresent(builder::sslParameters); } return builder.build(); } + JedisClientConfig createSentinelClientConfig(SentinelConfiguration sentinelConfiguration) { + return createClientConfig(0, sentinelConfiguration.getSentinelUsername(), + sentinelConfiguration.getSentinelPassword()); + } + @Override public void start() { - State current = state - .getAndUpdate(state -> State.CREATED.equals(state) || State.STOPPED.equals(state) ? State.STARTING : state); + State current = this.state.getAndUpdate(state -> isCreatedOrStopped(state) ? State.STARTING : state); - if (State.CREATED.equals(current) || State.STOPPED.equals(current)) { + if (isCreatedOrStopped(current)) { if (getUsePool() && !isRedisClusterAware()) { this.pool = createPool(); @@ -333,17 +345,21 @@ public class JedisConnectionFactory EXCEPTION_TRANSLATION); } - state.set(State.STARTED); + this.state.set(State.STARTED); } } + private boolean isCreatedOrStopped(@Nullable State state) { + return State.CREATED.equals(state) || State.STOPPED.equals(state); + } + @Override public void stop() { - if (state.compareAndSet(State.STARTED, State.STOPPING)) { + if (this.state.compareAndSet(State.STARTED, State.STOPPING)) { if (getUsePool() && !isRedisClusterAware()) { - if (pool != null) { + if (this.pool != null) { try { this.pool.close(); this.pool = null; @@ -353,12 +369,14 @@ public class JedisConnectionFactory } } - if (this.clusterCommandExecutor != null) { + ClusterCommandExecutor clusterCommandExecutor = this.clusterCommandExecutor; + + if (clusterCommandExecutor != null) { try { - this.clusterCommandExecutor.destroy(); + clusterCommandExecutor.destroy(); this.clusterCommandExecutor = null; - } catch (Exception e) { - throw new RuntimeException(e); + } catch (Exception cause) { + throw new RuntimeException(cause); } } @@ -369,17 +387,18 @@ public class JedisConnectionFactory try { this.cluster.close(); this.cluster = null; - } catch (Exception ex) { - log.warn("Cannot properly close Jedis cluster", ex); + } catch (Exception cause) { + log.warn("Cannot properly close Jedis cluster", cause); } } - state.set(State.STOPPED); + + this.state.set(State.STOPPED); } } @Override public int getPhase() { - return phase; + return this.phase; } /** @@ -394,7 +413,7 @@ public class JedisConnectionFactory @Override public boolean isRunning() { - return State.STARTED.equals(state.get()); + return State.STARTED.equals(this.state.get()); } private Pool createPool() { @@ -417,6 +436,7 @@ public class JedisConnectionFactory GenericObjectPoolConfig poolConfig = getPoolConfig() != null ? getPoolConfig() : new JedisPoolConfig(); JedisClientConfig sentinelConfig = createSentinelClientConfig(config); + return new JedisSentinelPool(config.getMaster().getName(), convertToJedisSentinelSet(config.getSentinels()), poolConfig, this.clientConfig, sentinelConfig); } @@ -431,7 +451,7 @@ public class JedisConnectionFactory return new JedisPool(getPoolConfig(), new HostAndPort(getHostName(), getPort()), this.clientConfig); } - private JedisCluster createCluster() { + JedisCluster createCluster() { return createCluster((RedisClusterConfiguration) this.configuration, getPoolConfig()); } @@ -462,6 +482,7 @@ public class JedisConnectionFactory Assert.notNull(clusterConfig, "Cluster configuration must not be null"); Set hostAndPort = new HashSet<>(); + for (RedisNode node : clusterConfig.getClusterNodes()) { hostAndPort.add(new HostAndPort(node.getHost(), node.getPort())); } @@ -491,12 +512,15 @@ public class JedisConnectionFactory JedisClientConfig sentinelConfig = this.clientConfig; SentinelConfiguration sentinelConfiguration = getSentinelConfiguration(); + if (sentinelConfiguration != null) { sentinelConfig = createSentinelClientConfig(sentinelConfiguration); } - JedisConnection connection = (getUsePool() ? new JedisConnection(jedis, pool, this.clientConfig, sentinelConfig) - : new JedisConnection(jedis, null, this.clientConfig, sentinelConfig)); + JedisConnection connection = getUsePool() + ? new JedisConnection(jedis, this.pool, this.clientConfig, sentinelConfig) + : new JedisConnection(jedis, null, this.clientConfig, sentinelConfig); + connection.setConvertPipelineAndTxResults(convertPipelineAndTxResults); return postProcessConnection(connection); @@ -509,19 +533,21 @@ public class JedisConnectionFactory * @return Jedis instance ready for wrapping into a {@link RedisConnection}. */ protected Jedis fetchJedisConnector() { + try { - if (getUsePool() && pool != null) { - return pool.getResource(); + if (getUsePool() && this.pool != null) { + return this.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); + } catch (Exception cause) { + throw new RedisConnectionFailureException("Cannot get Jedis connection", cause); } } @@ -549,8 +575,10 @@ public class JedisConnectionFactory throw new InvalidDataAccessApiUsageException("Cluster is not configured"); } - return postProcessConnection( - new JedisClusterConnection(this.cluster, this.clusterCommandExecutor, this.topologyProvider)); + JedisClusterConnection clusterConnection = + new JedisClusterConnection(this.cluster, getClusterCommandExecutor(), this.topologyProvider); + + return postProcessConnection(clusterConnection); } /** @@ -703,13 +731,8 @@ public class JedisConnectionFactory * @return the use of connection pooling. */ public boolean getUsePool() { - // Jedis Sentinel cannot operate without a pool. - if (isRedisSentinelAware()) { - return true; - } - - return clientConfiguration.isUsePooling(); + return isRedisSentinelAware() || getClientConfiguration().isUsePooling(); } /** @@ -812,7 +835,7 @@ public class JedisConnectionFactory * @since 2.0 */ public JedisClientConfiguration getClientConfiguration() { - return clientConfiguration; + return this.clientConfiguration; } /** @@ -821,7 +844,7 @@ public class JedisConnectionFactory */ @Nullable public RedisStandaloneConfiguration getStandaloneConfiguration() { - return standaloneConfig; + return this.standaloneConfig; } /** @@ -1078,6 +1101,5 @@ public class JedisConnectionFactory public void setConnectTimeout(Duration connectTimeout) { this.connectTimeout = connectTimeout; } - } } diff --git a/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceConnection.java b/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceConnection.java index 9d50e0f05..30f176180 100644 --- a/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceConnection.java +++ b/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceConnection.java @@ -100,17 +100,45 @@ import org.springframework.util.ObjectUtils; */ public class LettuceConnection extends AbstractRedisConnection { - private final Log LOGGER = LogFactory.getLog(getClass()); + private static final ExceptionTranslationStrategy EXCEPTION_TRANSLATION = + new FallbackExceptionTranslationStrategy(LettuceExceptionConverter.INSTANCE); static final RedisCodec CODEC = ByteArrayCodec.INSTANCE; - private static final ExceptionTranslationStrategy EXCEPTION_TRANSLATION = new FallbackExceptionTranslationStrategy( - LettuceExceptionConverter.INSTANCE); private static final TypeHints typeHints = new TypeHints(); + private static class LettuceTransactionResultConverter extends TransactionResultConverter { + + public LettuceTransactionResultConverter(Queue> txResults, + Converter exceptionConverter) { + + super(txResults, exceptionConverter); + } + + @Override + public List convert(List execResults) { + // Lettuce Empty list means null (watched variable modified) + return execResults.isEmpty() ? null : super.convert(execResults); + } + } + + // refers only to main connection as pubsub happens on a different one + private boolean convertPipelineAndTxResults = true; + private boolean isClosed = false; + private boolean isMulti = false; + private boolean isPipelined = false; + + private int dbIndex; + private final int defaultDbIndex; + private final long timeout; + + private final LettuceConnectionProvider connectionProvider; + + private volatile @Nullable LettuceSubscription subscription; + private final LettuceGeoCommands geoCommands = new LettuceGeoCommands(this); private final LettuceHashCommands hashCommands = new LettuceHashCommands(this); - private final LettuceHyperLogLogCommands hllCommands = new LettuceHyperLogLogCommands(this); + private final LettuceHyperLogLogCommands hyperLogLogCommands = new LettuceHyperLogLogCommands(this); private final LettuceKeyCommands keyCommands = new LettuceKeyCommands(this); private final LettuceListCommands listCommands = new LettuceListCommands(this); private final LettuceScriptingCommands scriptingCommands = new LettuceScriptingCommands(this); @@ -120,66 +148,21 @@ public class LettuceConnection extends AbstractRedisConnection { private final LettuceStringCommands stringCommands = new LettuceStringCommands(this); private final LettuceZSetCommands zSetCommands = new LettuceZSetCommands(this); - private final int defaultDbIndex; - private int dbIndex; - - private final LettuceConnectionProvider connectionProvider; - private final @Nullable StatefulConnection asyncSharedConn; - private @Nullable StatefulConnection asyncDedicatedConn; - - private final long timeout; - - // refers only to main connection as pubsub happens on a different one - private boolean isClosed = false; - private boolean isMulti = false; - private boolean isPipelined = false; private @Nullable List> ppline; - private @Nullable PipeliningFlushState flushState; - private final Queue> txResults = new LinkedList<>(); - private volatile @Nullable LettuceSubscription subscription; - /** flag indicating whether the connection needs to be dropped or not */ - private boolean convertPipelineAndTxResults = true; + + private final Log LOGGER = LogFactory.getLog(getClass()); + private PipeliningFlushPolicy pipeliningFlushPolicy = PipeliningFlushPolicy.flushEachCommand(); - LettuceResult newLettuceResult(Future resultHolder) { - return newLettuceResult(resultHolder, (val) -> val); - } + private @Nullable PipeliningFlushState pipeliningFlushState; - LettuceResult newLettuceResult(Future resultHolder, Converter converter) { + private final Queue> txResults = new LinkedList<>(); - return LettuceResultBuilder. forResponse(resultHolder).mappedWith(converter) - .convertPipelineAndTxResults(convertPipelineAndTxResults).build(); - } - - LettuceResult newLettuceResult(Future resultHolder, Converter converter, - Supplier defaultValue) { - - return LettuceResultBuilder. forResponse(resultHolder).mappedWith(converter) - .convertPipelineAndTxResults(convertPipelineAndTxResults).defaultNullTo(defaultValue).build(); - } - - LettuceResult newLettuceStatusResult(Future resultHolder) { - return LettuceResultBuilder. forResponse(resultHolder).buildStatusResult(); - } - - private class LettuceTransactionResultConverter extends TransactionResultConverter { - public LettuceTransactionResultConverter(Queue> txResults, - Converter exceptionConverter) { - super(txResults, exceptionConverter); - } - - @Override - public List convert(List execResults) { - // Lettuce Empty list means null (watched variable modified) - if (execResults.isEmpty()) { - return null; - } - return super.convert(execResults); - } - } + private @Nullable StatefulConnection asyncDedicatedConnection; + private final @Nullable StatefulConnection asyncSharedConnection; /** - * Instantiates a new lettuce connection. + * Creates a new {@link LettuceConnection}. * * @param timeout The connection timeout (in milliseconds) * @param client The {@link RedisClient} to use when instantiating a native connection @@ -189,7 +172,7 @@ public class LettuceConnection extends AbstractRedisConnection { } /** - * Instantiates a new lettuce connection. + * Creates a new {@link LettuceConnection}. * * @param sharedConnection A native connection that is shared with other {@link LettuceConnection}s. Will not be used * for transactions or blocking operations @@ -202,8 +185,10 @@ public class LettuceConnection extends AbstractRedisConnection { } /** - * @param sharedConnection A native connection that is shared with other {@link LettuceConnection}s. Should not be - * used for transactions or blocking operations. + * Creates a new {@link LettuceConnection}. + * + * @param sharedConnection A native connection that is shared with other {@link LettuceConnection}s. + * Should not be used for transactions or blocking operations. * @param timeout The connection timeout (in milliseconds) * @param client The {@link RedisClient} to use when making pub/sub connections. * @param defaultDbIndex The db index to use along with {@link RedisClient} when establishing a dedicated connection. @@ -213,15 +198,17 @@ public class LettuceConnection extends AbstractRedisConnection { @Nullable AbstractRedisClient client, int defaultDbIndex) { this.connectionProvider = new StandaloneConnectionProvider((RedisClient) client, CODEC); - this.asyncSharedConn = sharedConnection; + this.asyncSharedConnection = sharedConnection; this.timeout = timeout; this.defaultDbIndex = defaultDbIndex; this.dbIndex = this.defaultDbIndex; } /** - * @param sharedConnection A native connection that is shared with other {@link LettuceConnection}s. Should not be - * used for transactions or blocking operations. + * Creates a new {@link LettuceConnection}. + * + * @param sharedConnection A native connection that is shared with other {@link LettuceConnection}s. + * Should not be used for transactions or blocking operations. * @param connectionProvider connection provider to obtain and release native connections. * @param timeout The connection timeout (in milliseconds) * @param defaultDbIndex The db index to use along with {@link RedisClient} when establishing a dedicated connection. @@ -229,12 +216,15 @@ public class LettuceConnection extends AbstractRedisConnection { */ public LettuceConnection(@Nullable StatefulRedisConnection sharedConnection, LettuceConnectionProvider connectionProvider, long timeout, int defaultDbIndex) { + this((StatefulConnection) sharedConnection, connectionProvider, timeout, defaultDbIndex); } /** - * @param sharedConnection A native connection that is shared with other {@link LettuceConnection}s. Should not be - * used for transactions or blocking operations. + * Creates a new {@link LettuceConnection}. + * + * @param sharedConnection A native connection that is shared with other {@link LettuceConnection}s. + * Should not be used for transactions or blocking operations. * @param connectionProvider connection provider to obtain and release native connections. * @param timeout The connection timeout (in milliseconds) * @param defaultDbIndex The db index to use along with {@link RedisClient} when establishing a dedicated connection. @@ -245,17 +235,13 @@ public class LettuceConnection extends AbstractRedisConnection { Assert.notNull(connectionProvider, "LettuceConnectionProvider must not be null"); - this.asyncSharedConn = sharedConnection; + this.asyncSharedConnection = sharedConnection; this.connectionProvider = connectionProvider; this.timeout = timeout; this.defaultDbIndex = defaultDbIndex; this.dbIndex = this.defaultDbIndex; } - protected DataAccessException convertLettuceAccessException(Exception ex) { - return EXCEPTION_TRANSLATION.translate(ex); - } - @Override public org.springframework.data.redis.connection.RedisCommands commands() { return this; @@ -263,57 +249,61 @@ public class LettuceConnection extends AbstractRedisConnection { @Override public RedisGeoCommands geoCommands() { - return geoCommands; + return this.geoCommands; } @Override public RedisHashCommands hashCommands() { - return hashCommands; + return this.hashCommands; } @Override public RedisHyperLogLogCommands hyperLogLogCommands() { - return hllCommands; + return this.hyperLogLogCommands; } @Override public RedisKeyCommands keyCommands() { - return keyCommands; + return this.keyCommands; } @Override public RedisListCommands listCommands() { - return listCommands; + return this.listCommands; } @Override public RedisScriptingCommands scriptingCommands() { - return scriptingCommands; + return this.scriptingCommands; } @Override public RedisSetCommands setCommands() { - return setCommands; + return this.setCommands; } @Override public RedisServerCommands serverCommands() { - return serverCommands; + return this.serverCommands; } @Override public RedisStreamCommands streamCommands() { - return streamCommands; + return this.streamCommands; } @Override public RedisStringCommands stringCommands() { - return stringCommands; + return this.stringCommands; } @Override public RedisZSetCommands zSetCommands() { - return zSetCommands; + return this.zSetCommands; + } + + protected DataAccessException convertLettuceAccessException(Exception cause) { + return EXCEPTION_TRANSLATION.translate(cause); } @Override @@ -334,420 +324,45 @@ public class LettuceConnection extends AbstractRedisConnection { @SuppressWarnings({ "rawtypes", "unchecked" }) public Object execute(String command, @Nullable CommandOutput commandOutputTypeHint, byte[]... args) { - Assert.hasText(command, "a valid command needs to be specified"); + Assert.hasText(command, () -> String.format("A valid command [%s] needs to be specified", command)); - String name = command.trim().toUpperCase(); - ProtocolKeyword commandType = getCommandType(name); + ProtocolKeyword commandType = getCommandType(command.trim().toUpperCase()); validateCommandIfRunningInTransactionMode(commandType, args); - CommandArgs cmdArg = new CommandArgs<>(CODEC); + CommandArgs commandArguments = new CommandArgs<>(CODEC); + if (!ObjectUtils.isEmpty(args)) { - cmdArg.addKeys(args); + commandArguments.addKeys(args); } CommandOutput expectedOutput = commandOutputTypeHint != null ? commandOutputTypeHint : typeHints.getTypeHint(commandType); - Command cmd = new Command(commandType, expectedOutput, cmdArg); - return invoke().just(RedisClusterAsyncCommands::dispatch, cmd.getType(), cmd.getOutput(), cmd.getArgs()); + Command redisCommand = new Command(commandType, expectedOutput, commandArguments); + + return invoke().just(RedisClusterAsyncCommands::dispatch, redisCommand.getType(), redisCommand.getOutput(), + redisCommand.getArgs()); } - @Override - public void close() { - - super.close(); - - if (isClosed) { - return; - } - - isClosed = true; - - try { - reset(); - } catch (RuntimeException e) { - LOGGER.debug("Failed to reset connection during close", e); - } - } - - private void reset() { - - if (asyncDedicatedConn != null) { - try { - if (customizedDatabaseIndex()) { - potentiallySelectDatabase(defaultDbIndex); - } - connectionProvider.release(asyncDedicatedConn); - asyncDedicatedConn = null; - } catch (RuntimeException ex) { - throw convertLettuceAccessException(ex); - } - } - - LettuceSubscription subscription = this.subscription; - if (subscription != null) { - if (subscription.isAlive()) { - subscription.doClose(); - } - this.subscription = null; - } - - this.dbIndex = defaultDbIndex; - } - - @Override - public boolean isClosed() { - return isClosed && !isSubscribed(); - } - - @Override - public RedisClusterAsyncCommands getNativeConnection() { - - LettuceSubscription subscription = this.subscription; - return (subscription != null && subscription.isAlive() ? subscription.getNativeConnection().async() - : getAsyncConnection()); - } - - @Override - public boolean isQueueing() { - return isMulti; - } - - @Override - public boolean isPipelined() { - return isPipelined; - } - - @Override - public void openPipeline() { - - if (!isPipelined) { - isPipelined = true; - ppline = new ArrayList<>(); - flushState = this.pipeliningFlushPolicy.newPipeline(); - flushState.onOpen(this.getOrCreateDedicatedConnection()); - } - } - - @Override - public List closePipeline() { - - if (!isPipelined) { - return Collections.emptyList(); - } - - flushState.onClose(this.getOrCreateDedicatedConnection()); - flushState = null; - isPipelined = false; - List> futures = new ArrayList<>(ppline.size()); - for (LettuceResult result : ppline) { - futures.add(result.getResultHolder()); - } - - try { - boolean done = LettuceFutures.awaitAll(timeout, TimeUnit.MILLISECONDS, - futures.toArray(new RedisFuture[futures.size()])); - - List results = new ArrayList<>(futures.size()); - - Exception problem = null; - - if (done) { - for (LettuceResult result : ppline) { - - if (result.getResultHolder().getOutput().hasError()) { - - Exception err = new InvalidDataAccessApiUsageException(result.getResultHolder().getOutput().getError()); - // remember only the first error - if (problem == null) { - problem = err; - } - results.add(err); - } else if (!result.isStatus()) { - - try { - results.add(result.conversionRequired() ? result.convert(result.get()) : result.get()); - } catch (DataAccessException e) { - if (problem == null) { - problem = e; - } - results.add(e); - } - } - } - } - ppline.clear(); - - if (problem != null) { - throw new RedisPipelineException(problem, results); - } - - if (done) { - return results; - } - - throw new RedisPipelineException(new QueryTimeoutException("Redis command timed out")); - } catch (Exception e) { - throw new RedisPipelineException(e); - } - } - - @Override - public byte[] echo(byte[] message) { - return invoke().just(RedisClusterAsyncCommands::echo, message); - } - - @Override - public String ping() { - return invoke().just(RedisClusterAsyncCommands::ping); - } - - @Override - public void discard() { - isMulti = false; - try { - if (isPipelined()) { - pipeline(newLettuceStatusResult(getAsyncDedicatedRedisCommands().discard())); - return; - } - getDedicatedRedisCommands().discard(); - } catch (Exception ex) { - throw convertLettuceAccessException(ex); - } finally { - txResults.clear(); - } - } - - @Override - @SuppressWarnings({ "unchecked", "rawtypes" }) - public List exec() { - - isMulti = false; - - try { - Converter exceptionConverter = this::convertLettuceAccessException; - - if (isPipelined()) { - RedisFuture exec = getAsyncDedicatedRedisCommands().exec(); - - LettuceTransactionResultConverter resultConverter = new LettuceTransactionResultConverter( - new LinkedList<>(txResults), exceptionConverter); - - pipeline(newLettuceResult(exec, - source -> resultConverter.convert(LettuceConverters.transactionResultUnwrapper().convert(source)))); - return null; - } - - TransactionResult transactionResult = getDedicatedRedisCommands().exec(); - List results = LettuceConverters.transactionResultUnwrapper().convert(transactionResult); - return convertPipelineAndTxResults - ? new LettuceTransactionResultConverter(txResults, exceptionConverter).convert(results) - : results; - } catch (Exception ex) { - throw convertLettuceAccessException(ex); - } finally { - txResults.clear(); - } - } - - @Override - public void multi() { - if (isQueueing()) { - return; - } - isMulti = true; - try { - if (isPipelined()) { - getAsyncDedicatedRedisCommands().multi(); - return; - } - getDedicatedRedisCommands().multi(); - } catch (Exception ex) { - throw convertLettuceAccessException(ex); - } - } - - @Override - public void select(int dbIndex) { - - if (asyncSharedConn != null) { - throw new InvalidDataAccessApiUsageException("Selecting a new database not supported due to shared connection;" - + " Use separate ConnectionFactorys to work with multiple databases"); - } - - this.dbIndex = dbIndex; - - invokeStatus().just(RedisClusterAsyncCommands::dispatch, CommandType.SELECT, - new StatusOutput<>(ByteArrayCodec.INSTANCE), new CommandArgs<>(ByteArrayCodec.INSTANCE).add(dbIndex)); - } - - @Override - public void unwatch() { - - try { - if (isPipelined()) { - pipeline(newLettuceStatusResult(getAsyncDedicatedRedisCommands().unwatch())); - return; - } - if (isQueueing()) { - transaction(newLettuceStatusResult(getAsyncDedicatedRedisCommands().unwatch())); - return; - } - getDedicatedRedisCommands().unwatch(); - } catch (Exception ex) { - throw convertLettuceAccessException(ex); - } - } - - @Override - public void watch(byte[]... keys) { - if (isQueueing()) { - throw new InvalidDataAccessApiUsageException("WATCH is not supported when a transaction is active"); - } - try { - if (isPipelined()) { - pipeline(newLettuceStatusResult(getAsyncDedicatedRedisCommands().watch(keys))); - return; - } - if (isQueueing()) { - transaction(new LettuceStatusResult(getAsyncDedicatedRedisCommands().watch(keys))); - return; - } - getDedicatedRedisCommands().watch(keys); - } catch (Exception ex) { - throw convertLettuceAccessException(ex); - } - } - - // - // Pub/Sub functionality - // - - @Override - public Long publish(byte[] channel, byte[] message) { - return invoke().just(RedisClusterAsyncCommands::publish, channel, message); - } - - @Override - public Subscription getSubscription() { - return subscription; - } - - @Override - public boolean isSubscribed() { - return (subscription != null && subscription.isAlive()); - } - - @Override - public void pSubscribe(MessageListener listener, byte[]... patterns) { - - checkSubscription(); + RedisClusterAsyncCommands getAsyncConnection() { if (isQueueing() || isPipelined()) { - throw new InvalidDataAccessApiUsageException("Transaction/Pipelining is not supported for Pub/Sub subscriptions"); + return getAsyncDedicatedConnection(); } - try { - subscription = initSubscription(listener); - subscription.pSubscribe(patterns); - } catch (Exception ex) { - throw convertLettuceAccessException(ex); - } - } + StatefulConnection sharedConnection = this.asyncSharedConnection; - @Override - public void subscribe(MessageListener listener, byte[]... channels) { - - checkSubscription(); - - if (isQueueing() || isPipelined()) { - throw new InvalidDataAccessApiUsageException("Transaction/Pipelining is not supported for Pub/Sub subscriptions"); + if (sharedConnection != null) { + if (sharedConnection instanceof StatefulRedisConnection statefulConnection) { + return statefulConnection.async(); + } + if (sharedConnection instanceof StatefulRedisClusterConnection statefulClusterConnection) { + return statefulClusterConnection.async(); + } } - try { - subscription = initSubscription(listener); - subscription.subscribe(channels); - } catch (Exception ex) { - throw convertLettuceAccessException(ex); - } - } - - @SuppressWarnings("unchecked") - T failsafeReadScanValues(List source, @SuppressWarnings("rawtypes") Converter converter) { - - try { - return (T) (converter != null ? converter.convert(source) : source); - } catch (IndexOutOfBoundsException e) { - // ignore this one - } - return null; - } - - /** - * Specifies if pipelined and transaction results should be converted to the expected data type. If false, results of - * {@link #closePipeline()} and {@link #exec()} will be of the type returned by the Lettuce driver - * - * @param convertPipelineAndTxResults Whether or not to convert pipeline and tx results - */ - public void setConvertPipelineAndTxResults(boolean convertPipelineAndTxResults) { - this.convertPipelineAndTxResults = convertPipelineAndTxResults; - } - - /** - * Configures the flushing policy when using pipelining. - * - * @param pipeliningFlushPolicy the flushing policy to control when commands get written to the Redis connection. - * @see PipeliningFlushPolicy#flushEachCommand() - * @see #openPipeline() - * @see StatefulRedisConnection#flushCommands() - * @since 2.3 - */ - public void setPipeliningFlushPolicy(PipeliningFlushPolicy pipeliningFlushPolicy) { - - Assert.notNull(pipeliningFlushPolicy, "PipeliningFlushingPolicy must not be null"); - - this.pipeliningFlushPolicy = pipeliningFlushPolicy; - } - - /** - * {@link #close()} the current connection and open a new pub/sub connection to the Redis server. - * - * @return never {@literal null}. - */ - @SuppressWarnings("unchecked") - protected StatefulRedisPubSubConnection switchToPubSub() { - - checkSubscription(); - reset(); - return connectionProvider.getConnection(StatefulRedisPubSubConnection.class); - } - - /** - * Customization hook to create a {@link LettuceSubscription}. - * - * @param listener the {@link MessageListener} to notify. - * @param connection Pub/Sub connection. - * @param connectionProvider the {@link LettuceConnectionProvider} for connection release. - * @return a {@link LettuceSubscription}. - * @since 2.2 - */ - protected LettuceSubscription doCreateSubscription(MessageListener listener, - StatefulRedisPubSubConnection connection, LettuceConnectionProvider connectionProvider) { - return new LettuceSubscription(listener, connection, connectionProvider); - } - - void pipeline(LettuceResult result) { - - if (flushState != null) { - flushState.onCommand(getOrCreateDedicatedConnection()); - } - - if (isQueueing()) { - transaction(result); - } else { - ppline.add(result); - } + return getAsyncDedicatedConnection(); } /** @@ -795,9 +410,10 @@ public class LettuceConnection extends AbstractRedisConnection { } else { pipeline(newLettuceResult(future.get(), converter, nullDefault)); } - } catch (Exception ex) { - throw convertLettuceAccessException(ex); + } catch (Exception cause) { + throw convertLettuceAccessException(cause); } + return null; }); } @@ -805,16 +421,17 @@ public class LettuceConnection extends AbstractRedisConnection { if (isQueueing()) { return new LettuceInvoker(connection, (future, converter, nullDefault) -> { + try { if (statusCommand) { transaction(newLettuceStatusResult(future.get())); } else { transaction(newLettuceResult(future.get(), converter, nullDefault)); } - - } catch (Exception ex) { - throw convertLettuceAccessException(ex); + } catch (Exception cause) { + throw convertLettuceAccessException(cause); } + return null; }); } @@ -825,37 +442,458 @@ public class LettuceConnection extends AbstractRedisConnection { Object result = await(future.get()); - if (result == null) { - return nullDefault.get(); - } - - return converter.convert(result); - } catch (Exception ex) { - throw convertLettuceAccessException(ex); + return result != null ? converter.convert(result) : nullDefault.get(); + } catch (Exception cause) { + throw convertLettuceAccessException(cause); } }); } - void transaction(FutureResult result) { - txResults.add(result); + LettuceResult newLettuceResult(Future resultHolder, Converter converter) { + + return LettuceResultBuilder.forResponse(resultHolder) + .mappedWith(converter) + .convertPipelineAndTxResults(this.convertPipelineAndTxResults) + .build(); } - RedisClusterAsyncCommands getAsyncConnection() { + LettuceResult newLettuceResult(Future resultHolder, Converter converter, + Supplier defaultValue) { + + return LettuceResultBuilder.forResponse(resultHolder) + .mappedWith(converter) + .convertPipelineAndTxResults(this.convertPipelineAndTxResults) + .defaultNullTo(defaultValue) + .build(); + } + + LettuceResult newLettuceStatusResult(Future resultHolder) { + return LettuceResultBuilder.forResponse(resultHolder).buildStatusResult(); + } + + void pipeline(LettuceResult result) { + + PipeliningFlushState pipeliningFlushState = this.pipeliningFlushState; + + if (pipeliningFlushState != null) { + pipeliningFlushState.onCommand(getOrCreateDedicatedConnection()); + } + + if (isQueueing()) { + transaction(result); + } else { + this.ppline.add(result); + } + } + + void transaction(FutureResult result) { + this.txResults.add(result); + } + + @Override + public void close() { + + super.close(); + + if (isClosed) { + return; + } + + isClosed = true; + + try { + reset(); + } catch (RuntimeException cause) { + LOGGER.debug("Failed to reset connection during close", cause); + } + } + + private void reset() { + + if (this.asyncDedicatedConnection != null) { + try { + if (customizedDatabaseIndex()) { + potentiallySelectDatabase(this.defaultDbIndex); + } + this.connectionProvider.release(this.asyncDedicatedConnection); + this.asyncDedicatedConnection = null; + } catch (RuntimeException cause) { + throw convertLettuceAccessException(cause); + } + } + + LettuceSubscription subscription = this.subscription; + + if (isAlive(subscription)) { + subscription.doClose(); + } + + this.subscription = null; + this.dbIndex = defaultDbIndex; + } + + @Override + public boolean isClosed() { + return this.isClosed && !isSubscribed(); + } + + @Override + public RedisClusterAsyncCommands getNativeConnection() { + + LettuceSubscription subscription = this.subscription; + + return isAlive(subscription) ? subscription.getNativeConnection().async() : getAsyncConnection(); + } + + private boolean isAlive(@Nullable LettuceSubscription subscription) { + return subscription != null && subscription.isAlive(); + } + + @Override + public boolean isQueueing() { + return this.isMulti; + } + + @Override + public boolean isPipelined() { + return this.isPipelined; + } + + @Override + public void openPipeline() { + + if (!isPipelined) { + isPipelined = true; + ppline = new ArrayList<>(); + pipeliningFlushState = this.pipeliningFlushPolicy.newPipeline(); + pipeliningFlushState.onOpen(this.getOrCreateDedicatedConnection()); + } + } + + @Override + public List closePipeline() { + + if (!isPipelined) { + return Collections.emptyList(); + } + + pipeliningFlushState.onClose(this.getOrCreateDedicatedConnection()); + pipeliningFlushState = null; + isPipelined = false; + + List> futures = new ArrayList<>(ppline.size()); + + for (LettuceResult result : ppline) { + futures.add(result.getResultHolder()); + } + + try { + + boolean done = LettuceFutures.awaitAll(timeout, TimeUnit.MILLISECONDS, futures.toArray(new RedisFuture[0])); + + List results = new ArrayList<>(futures.size()); + + Exception problem = null; + + if (done) { + for (LettuceResult result : ppline) { + + if (result.getResultHolder().getOutput().hasError()) { + + Exception exception = new InvalidDataAccessApiUsageException(result.getResultHolder() + .getOutput().getError()); + + // remember only the first error + if (problem == null) { + problem = exception; + } + + results.add(exception); + } else if (!result.isStatus()) { + + try { + results.add(result.conversionRequired() ? result.convert(result.get()) : result.get()); + } catch (DataAccessException cause) { + if (problem == null) { + problem = cause; + } + results.add(cause); + } + } + } + } + + ppline.clear(); + + if (problem != null) { + throw new RedisPipelineException(problem, results); + } + + if (done) { + return results; + } + + throw new RedisPipelineException(new QueryTimeoutException("Redis command timed out")); + } catch (Exception cause) { + throw new RedisPipelineException(cause); + } + } + + @Override + public byte[] echo(byte[] message) { + return invoke().just(RedisClusterAsyncCommands::echo, message); + } + + @Override + public String ping() { + return invoke().just(RedisClusterAsyncCommands::ping); + } + + @Override + public void discard() { + + isMulti = false; + + try { + if (isPipelined()) { + pipeline(newLettuceStatusResult(getAsyncDedicatedRedisCommands().discard())); + return; + } + getDedicatedRedisCommands().discard(); + } catch (Exception cause) { + throw convertLettuceAccessException(cause); + } finally { + txResults.clear(); + } + } + + @Override + @SuppressWarnings({ "unchecked", "rawtypes" }) + public List exec() { + + isMulti = false; + + try { + Converter exceptionConverter = this::convertLettuceAccessException; + + if (isPipelined()) { + RedisFuture exec = getAsyncDedicatedRedisCommands().exec(); + + LettuceTransactionResultConverter resultConverter = new LettuceTransactionResultConverter( + new LinkedList<>(txResults), exceptionConverter); + + pipeline(newLettuceResult(exec, source -> + resultConverter.convert(LettuceConverters.transactionResultUnwrapper().convert(source)))); + + return null; + } + + TransactionResult transactionResult = getDedicatedRedisCommands().exec(); + + List results = LettuceConverters.transactionResultUnwrapper().convert(transactionResult); + + return convertPipelineAndTxResults + ? new LettuceTransactionResultConverter(txResults, exceptionConverter).convert(results) + : results; + } catch (Exception cause) { + throw convertLettuceAccessException(cause); + } finally { + txResults.clear(); + } + } + + @Override + public void multi() { + + if (isQueueing()) { + return; + } + + isMulti = true; + + try { + if (isPipelined()) { + getAsyncDedicatedRedisCommands().multi(); + return; + } + getDedicatedRedisCommands().multi(); + } catch (Exception cause) { + throw convertLettuceAccessException(cause); + } + } + + @Override + public void select(int dbIndex) { + + if (asyncSharedConnection != null) { + throw new InvalidDataAccessApiUsageException("Selecting a new database not supported due to shared connection;" + + " Use separate ConnectionFactorys to work with multiple databases"); + } + + this.dbIndex = dbIndex; + + invokeStatus().just(RedisClusterAsyncCommands::dispatch, CommandType.SELECT, + new StatusOutput<>(ByteArrayCodec.INSTANCE), new CommandArgs<>(ByteArrayCodec.INSTANCE).add(dbIndex)); + } + + @Override + public void unwatch() { + + try { + if (isPipelined()) { + pipeline(newLettuceStatusResult(getAsyncDedicatedRedisCommands().unwatch())); + return; + } + if (isQueueing()) { + transaction(newLettuceStatusResult(getAsyncDedicatedRedisCommands().unwatch())); + return; + } + getDedicatedRedisCommands().unwatch(); + } catch (Exception cause) { + throw convertLettuceAccessException(cause); + } + } + + @Override + public void watch(byte[]... keys) { + + if (isQueueing()) { + throw new InvalidDataAccessApiUsageException("WATCH is not supported when a transaction is active"); + } + + try { + if (isPipelined()) { + pipeline(newLettuceStatusResult(getAsyncDedicatedRedisCommands().watch(keys))); + return; + } + if (isQueueing()) { + transaction(new LettuceStatusResult(getAsyncDedicatedRedisCommands().watch(keys))); + return; + } + getDedicatedRedisCommands().watch(keys); + } catch (Exception cause) { + throw convertLettuceAccessException(cause); + } + } + + // + // Pub/Sub functionality + // + + @Override + public Long publish(byte[] channel, byte[] message) { + return invoke().just(RedisClusterAsyncCommands::publish, channel, message); + } + + @Override + public Subscription getSubscription() { + return this.subscription; + } + + @Override + public boolean isSubscribed() { + Subscription subscription = getSubscription(); + return subscription != null && subscription.isAlive(); + } + + @Override + public void pSubscribe(MessageListener listener, byte[]... patterns) { + + checkSubscription(); if (isQueueing() || isPipelined()) { - return getAsyncDedicatedConnection(); + throw new InvalidDataAccessApiUsageException("Transaction/Pipelining is not supported for Pub/Sub subscriptions"); } - if (asyncSharedConn != null) { - - if (asyncSharedConn instanceof StatefulRedisConnection) { - return ((StatefulRedisConnection) asyncSharedConn).async(); - } - if (asyncSharedConn instanceof StatefulRedisClusterConnection) { - return ((StatefulRedisClusterConnection) asyncSharedConn).async(); - } + try { + this.subscription = initSubscription(listener); + this.subscription.pSubscribe(patterns); + } catch (Exception cause) { + throw convertLettuceAccessException(cause); } - return getAsyncDedicatedConnection(); + } + + @Override + public void subscribe(MessageListener listener, byte[]... channels) { + + checkSubscription(); + + if (isQueueing() || isPipelined()) { + throw new InvalidDataAccessApiUsageException("Transaction/Pipelining is not supported for Pub/Sub subscriptions"); + } + + try { + this.subscription = initSubscription(listener); + this.subscription.subscribe(channels); + } catch (Exception cause) { + throw convertLettuceAccessException(cause); + } + } + + @SuppressWarnings("unchecked") + T failsafeReadScanValues(List source, @SuppressWarnings("rawtypes") @Nullable Converter converter) { + + try { + return (T) (converter != null ? converter.convert(source) : source); + } catch (IndexOutOfBoundsException ignore) { + } + + return null; + } + + /** + * Specifies if pipelined and transaction results should be converted to the expected data type. If false, results of + * {@link #closePipeline()} and {@link #exec()} will be of the type returned by the Lettuce driver + * + * @param convertPipelineAndTxResults Whether or not to convert pipeline and tx results + */ + public void setConvertPipelineAndTxResults(boolean convertPipelineAndTxResults) { + this.convertPipelineAndTxResults = convertPipelineAndTxResults; + } + + /** + * Configures the flushing policy when using pipelining. + * + * @param pipeliningFlushPolicy the flushing policy to control when commands get written to the Redis connection. + * @see PipeliningFlushPolicy#flushEachCommand() + * @see #openPipeline() + * @see StatefulRedisConnection#flushCommands() + * @since 2.3 + */ + public void setPipeliningFlushPolicy(PipeliningFlushPolicy pipeliningFlushPolicy) { + + Assert.notNull(pipeliningFlushPolicy, "PipeliningFlushingPolicy must not be null"); + + this.pipeliningFlushPolicy = pipeliningFlushPolicy; + } + + /** + * {@link #close()} the current connection and open a new pub/sub connection to the Redis server. + * + * @return never {@literal null}. + */ + @SuppressWarnings("unchecked") + protected StatefulRedisPubSubConnection switchToPubSub() { + + checkSubscription(); + reset(); + + return this.connectionProvider.getConnection(StatefulRedisPubSubConnection.class); + } + + /** + * Customization hook to create a {@link LettuceSubscription}. + * + * @param listener the {@link MessageListener} to notify. + * @param connection Pub/Sub connection. + * @param connectionProvider the {@link LettuceConnectionProvider} for connection release. + * @return a {@link LettuceSubscription}. + * @since 2.2 + */ + protected LettuceSubscription doCreateSubscription(MessageListener listener, + StatefulRedisPubSubConnection connection, LettuceConnectionProvider connectionProvider) { + + return new LettuceSubscription(listener, connection, connectionProvider); } protected RedisClusterCommands getConnection() { @@ -864,13 +902,12 @@ public class LettuceConnection extends AbstractRedisConnection { return getDedicatedConnection(); } - if (asyncSharedConn != null) { - - if (asyncSharedConn instanceof StatefulRedisConnection) { - return ((StatefulRedisConnection) asyncSharedConn).sync(); + if (asyncSharedConnection != null) { + if (asyncSharedConnection instanceof StatefulRedisConnection statefulConnection) { + return statefulConnection.sync(); } - if (asyncSharedConn instanceof StatefulRedisClusterConnection) { - return ((StatefulRedisClusterConnection) asyncSharedConn).sync(); + if (asyncSharedConnection instanceof StatefulRedisClusterConnection statefulClusterConnection) { + return statefulClusterConnection.sync(); } } @@ -881,15 +918,16 @@ public class LettuceConnection extends AbstractRedisConnection { StatefulConnection connection = getOrCreateDedicatedConnection(); - if (connection instanceof StatefulRedisConnection) { - return ((StatefulRedisConnection) connection).sync(); + if (connection instanceof StatefulRedisConnection statefulConnection) { + return statefulConnection.sync(); } - if (connection instanceof StatefulRedisClusterConnection) { - return ((StatefulRedisClusterConnection) connection).sync(); + if (connection instanceof StatefulRedisClusterConnection statefulClusterConnection) { + return statefulClusterConnection.sync(); } - throw new IllegalStateException( - String.format("%s is not a supported connection type", connection.getClass().getName())); + String message = String.format("%s is not a supported connection type", connection.getClass().getName()); + + throw new IllegalStateException(message); } protected RedisClusterAsyncCommands getAsyncDedicatedConnection() { @@ -900,24 +938,25 @@ public class LettuceConnection extends AbstractRedisConnection { StatefulConnection connection = getOrCreateDedicatedConnection(); - if (connection instanceof StatefulRedisConnection) { - return ((StatefulRedisConnection) connection).async(); + if (connection instanceof StatefulRedisConnection statefulConnection) { + return statefulConnection.async(); } - if (asyncDedicatedConn instanceof StatefulRedisClusterConnection) { - return ((StatefulRedisClusterConnection) connection).async(); + if (asyncDedicatedConnection instanceof StatefulRedisClusterConnection statefulClusterConnection) { + return statefulClusterConnection.async(); } - throw new IllegalStateException( - String.format("%s is not a supported connection type", connection.getClass().getName())); + String message = String.format("%s is not a supported connection type", connection.getClass().getName()); + + throw new IllegalStateException(message); } @SuppressWarnings("unchecked") protected StatefulConnection doGetAsyncDedicatedConnection() { - StatefulConnection connection = connectionProvider.getConnection(StatefulConnection.class); + StatefulConnection connection = getConnectionProvider().getConnection(StatefulConnection.class); if (customizedDatabaseIndex()) { - potentiallySelectDatabase(dbIndex); + potentiallySelectDatabase(this.dbIndex); } return connection; @@ -927,14 +966,15 @@ public class LettuceConnection extends AbstractRedisConnection { protected boolean isActive(RedisNode node) { StatefulRedisSentinelConnection connection = null; + try { connection = getConnection(node); return connection.sync().ping().equalsIgnoreCase("pong"); - } catch (Exception e) { + } catch (Exception cause) { return false; } finally { if (connection != null) { - connectionProvider.release(connection); + getConnectionProvider().release(connection); } } } @@ -943,23 +983,24 @@ public class LettuceConnection extends AbstractRedisConnection { protected RedisSentinelConnection getSentinelConnection(RedisNode sentinel) { StatefulRedisSentinelConnection connection = getConnection(sentinel); + return new LettuceSentinelConnection(connection); } LettuceConnectionProvider getConnectionProvider() { - return connectionProvider; + return this.connectionProvider; } @SuppressWarnings("unchecked") private StatefulRedisSentinelConnection getConnection(RedisNode sentinel) { - return ((TargetAware) connectionProvider).getConnection(StatefulRedisSentinelConnection.class, + return ((TargetAware) getConnectionProvider()).getConnection(StatefulRedisSentinelConnection.class, getRedisURI(sentinel)); } @Nullable private T await(RedisFuture cmd) { - if (isMulti) { + if (this.isMulti) { return null; } @@ -972,24 +1013,23 @@ public class LettuceConnection extends AbstractRedisConnection { private StatefulConnection getOrCreateDedicatedConnection() { - if (asyncDedicatedConn == null) { - asyncDedicatedConn = doGetAsyncDedicatedConnection(); + if (this.asyncDedicatedConnection == null) { + this.asyncDedicatedConnection = doGetAsyncDedicatedConnection(); } - return asyncDedicatedConn; + return this.asyncDedicatedConnection; } - @SuppressWarnings("unchecked") private RedisCommands getDedicatedRedisCommands() { - return (RedisCommands) getDedicatedConnection(); + return (RedisCommands) getDedicatedConnection(); } - @SuppressWarnings("unchecked") private RedisAsyncCommands getAsyncDedicatedRedisCommands() { - return (RedisAsyncCommands) getAsyncDedicatedConnection(); + return (RedisAsyncCommands) getAsyncDedicatedConnection(); } private void checkSubscription() { + if (isSubscribed()) { throw new RedisSubscribedConnectionException( "Connection already subscribed; use the connection Subscription to cancel or add new channels"); @@ -1001,7 +1041,12 @@ public class LettuceConnection extends AbstractRedisConnection { } private RedisURI getRedisURI(RedisNode node) { - return RedisURI.Builder.redis(node.getHost(), node.getPort()).build(); + return RedisURI.Builder.redis(node.getHost(), getPort(node)).build(); + } + + private int getPort(RedisNode node) { + Integer port = node.getPort(); + return port != null ? port : RedisURI.DEFAULT_REDIS_PORT; } private boolean customizedDatabaseIndex() { @@ -1009,8 +1054,9 @@ public class LettuceConnection extends AbstractRedisConnection { } private void potentiallySelectDatabase(int dbIndex) { - if (asyncDedicatedConn instanceof StatefulRedisConnection) { - ((StatefulRedisConnection) asyncDedicatedConn).sync().select(dbIndex); + + if (asyncDedicatedConnection instanceof StatefulRedisConnection statefulConnection) { + statefulConnection.sync().select(dbIndex); } } @@ -1025,14 +1071,16 @@ public class LettuceConnection extends AbstractRedisConnection { } } - private void validateCommand(ProtocolKeyword cmd, @Nullable byte[]... args) { + private void validateCommand(ProtocolKeyword command, @Nullable byte[]... args) { + + RedisCommand redisCommand = RedisCommand.failsafeCommandLookup(command.name()); - RedisCommand redisCommand = RedisCommand.failsafeCommandLookup(cmd.name()); if (!RedisCommand.UNKNOWN.equals(redisCommand) && redisCommand.requiresArguments()) { try { redisCommand.validateArgumentCount(args != null ? args.length : 0); - } catch (IllegalArgumentException e) { - throw new InvalidDataAccessApiUsageException(String.format("Validation failed for %s command", cmd), e); + } catch (IllegalArgumentException cause) { + String message = String.format("Validation failed for %s command", command); + throw new InvalidDataAccessApiUsageException(message, cause); } } } @@ -1041,7 +1089,7 @@ public class LettuceConnection extends AbstractRedisConnection { try { return CommandType.valueOf(name); - } catch (IllegalArgumentException e) { + } catch (IllegalArgumentException cause) { return new CustomCommandType(name); } } @@ -1059,7 +1107,7 @@ public class LettuceConnection extends AbstractRedisConnection { @SuppressWarnings("rawtypes") // private static final Map, Constructor> CONSTRUCTORS = new ConcurrentHashMap<>(); - { + static { // INTEGER COMMAND_OUTPUT_TYPE_MAPPING.put(BITCOUNT, IntegerOutput.class); COMMAND_OUTPUT_TYPE_MAPPING.put(BITOP, IntegerOutput.class); @@ -1219,7 +1267,7 @@ public class LettuceConnection extends AbstractRedisConnection { /** * Returns the {@link CommandOutput} mapped for given {@link CommandType} or {@link ByteArrayOutput} as default. * - * @param type + * @param type {@link ProtocolKeyword} used to lookup the type hint. * @return {@link ByteArrayOutput} as default when no matching {@link CommandOutput} available. */ @SuppressWarnings("rawtypes") @@ -1230,28 +1278,33 @@ public class LettuceConnection extends AbstractRedisConnection { /** * Returns the {@link CommandOutput} mapped for given {@link CommandType} given {@link CommandOutput} as default. * - * @param type - * @return + * @param type {@link ProtocolKeyword} used to lookup the type hint. + * @return the {@link CommandOutput} mapped for given {@link CommandType} given {@link CommandOutput} as default. */ @SuppressWarnings("rawtypes") - public CommandOutput getTypeHint(ProtocolKeyword type, CommandOutput defaultType) { + public CommandOutput getTypeHint(@Nullable ProtocolKeyword type, CommandOutput defaultType) { if (type == null || !COMMAND_OUTPUT_TYPE_MAPPING.containsKey(type)) { return defaultType; } - CommandOutput outputType = instanciateCommandOutput(COMMAND_OUTPUT_TYPE_MAPPING.get(type)); + + CommandOutput outputType = instantiateCommandOutput(COMMAND_OUTPUT_TYPE_MAPPING.get(type)); + return outputType != null ? outputType : defaultType; } @SuppressWarnings({ "rawtypes", "unchecked" }) - private CommandOutput instanciateCommandOutput(Class type) { + private CommandOutput instantiateCommandOutput(Class type) { Assert.notNull(type, "Cannot create instance for 'null' type."); + Constructor constructor = CONSTRUCTORS.get(type); + if (constructor == null) { constructor = (Constructor) ClassUtils.getConstructorIfAvailable(type, RedisCodec.class); CONSTRUCTORS.put(type, constructor); } + return BeanUtils.instantiateClass(constructor, CODEC); } } @@ -1315,7 +1368,7 @@ public class LettuceConnection extends AbstractRedisConnection { /** * Callback if the pipeline gets opened. * - * @param connection + * @param connection Lettuce {@link StatefulConnection}. * @see #openPipeline() */ void onOpen(StatefulConnection connection); @@ -1323,7 +1376,7 @@ public class LettuceConnection extends AbstractRedisConnection { /** * Callback for each issued Redis command. * - * @param connection + * @param connection Lettuce {@link StatefulConnection}. * @see #pipeline(LettuceResult) */ void onCommand(StatefulConnection connection); @@ -1331,7 +1384,7 @@ public class LettuceConnection extends AbstractRedisConnection { /** * Callback if the pipeline gets closed. * - * @param connection + * @param connection Lettuce {@link StatefulConnection}. * @see #closePipeline() */ void onClose(StatefulConnection connection); @@ -1432,45 +1485,40 @@ public class LettuceConnection extends AbstractRedisConnection { /** * @since 2.3.8 */ - static class CustomCommandType implements ProtocolKeyword { - - private final String name; - - CustomCommandType(String name) { - this.name = name; - } + record CustomCommandType(String name) implements ProtocolKeyword { @Override public byte[] getBytes() { - return name.getBytes(StandardCharsets.US_ASCII); + return name().getBytes(StandardCharsets.US_ASCII); } @Override public String name() { - return name; + return this.name; } @Override - public boolean equals(@Nullable Object o) { + public boolean equals(@Nullable Object obj) { - if (this == o) { + if (this == obj) { return true; } - if (!(o instanceof CustomCommandType)) { + + if (!(obj instanceof CustomCommandType that)) { return false; } - CustomCommandType that = (CustomCommandType) o; - return ObjectUtils.nullSafeEquals(name, that.name); + + return ObjectUtils.nullSafeEquals(this.name(), that.name()); } @Override public int hashCode() { - return ObjectUtils.nullSafeHashCode(name); + return ObjectUtils.nullSafeHashCode(name()); } @Override public String toString() { - return name; + return name(); } } } 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 6bb2a3bd3..5883fee84 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 @@ -111,181 +111,8 @@ import org.springframework.util.StringUtils; public class LettuceConnectionFactory implements RedisConnectionFactory, ReactiveRedisConnectionFactory, InitializingBean, DisposableBean, SmartLifecycle { - private static final ExceptionTranslationStrategy EXCEPTION_TRANSLATION = new PassThroughExceptionTranslationStrategy( - LettuceExceptionConverter.INSTANCE); - - private final Log log = LogFactory.getLog(getClass()); - - /** Synchronization monitor for the shared Connection */ - private final Object connectionMonitor = new Object(); - - private final LettuceClientConfiguration clientConfiguration; - - private RedisStandaloneConfiguration standaloneConfig = new RedisStandaloneConfiguration("localhost", 6379); - - private @Nullable RedisConfiguration configuration; - private int phase = 0; // in between min and max values - - 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 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. - */ - public LettuceConnectionFactory() { - this(new MutableLettuceClientConfiguration()); - } - - /** - * Constructs a new {@link LettuceConnectionFactory} instance with default settings. - */ - public LettuceConnectionFactory(RedisStandaloneConfiguration configuration) { - this(configuration, new MutableLettuceClientConfiguration()); - } - - /** - * Constructs a new {@link LettuceConnectionFactory} instance given {@link LettuceClientConfiguration}. - * - * @param clientConfig must not be {@literal null} - * @since 2.0 - */ - private LettuceConnectionFactory(LettuceClientConfiguration clientConfig) { - - Assert.notNull(clientConfig, "LettuceClientConfiguration must not be null"); - - this.clientConfiguration = clientConfig; - this.configuration = this.standaloneConfig; - } - - /** - * Constructs a new {@link LettuceConnectionFactory} instance with default settings. - */ - public LettuceConnectionFactory(String host, int port) { - this(new RedisStandaloneConfiguration(host, port), new MutableLettuceClientConfiguration()); - } - - /** - * Constructs a new {@link LettuceConnectionFactory} instance using the given {@link RedisSocketConfiguration}. - * - * @param redisConfiguration must not be {@literal null}. - * @since 2.1 - */ - public LettuceConnectionFactory(RedisConfiguration redisConfiguration) { - this(redisConfiguration, new MutableLettuceClientConfiguration()); - } - - /** - * Constructs a new {@link LettuceConnectionFactory} instance using the given {@link RedisSentinelConfiguration}. - * - * @param sentinelConfiguration must not be {@literal null}. - * @since 1.6 - */ - public LettuceConnectionFactory(RedisSentinelConfiguration sentinelConfiguration) { - this(sentinelConfiguration, new MutableLettuceClientConfiguration()); - } - - /** - * Constructs a new {@link LettuceConnectionFactory} instance using the given {@link RedisClusterConfiguration} - * applied to create a {@link RedisClusterClient}. - * - * @param clusterConfiguration must not be {@literal null}. - * @since 1.7 - */ - public LettuceConnectionFactory(RedisClusterConfiguration clusterConfiguration) { - this(clusterConfiguration, new MutableLettuceClientConfiguration()); - } - - /** - * Constructs a new {@link LettuceConnectionFactory} instance using the given {@link RedisStandaloneConfiguration} and - * {@link LettuceClientConfiguration}. - * - * @param standaloneConfig must not be {@literal null}. - * @param clientConfig must not be {@literal null}. - * @since 2.0 - */ - public LettuceConnectionFactory(RedisStandaloneConfiguration standaloneConfig, - LettuceClientConfiguration clientConfig) { - - this(clientConfig); - - Assert.notNull(standaloneConfig, "RedisStandaloneConfiguration must not be null"); - - this.standaloneConfig = standaloneConfig; - this.configuration = this.standaloneConfig; - } - - /** - * Constructs a new {@link LettuceConnectionFactory} instance using the given - * {@link RedisStaticMasterReplicaConfiguration} and {@link LettuceClientConfiguration}. - * - * @param redisConfiguration must not be {@literal null}. - * @param clientConfig must not be {@literal null}. - * @since 2.1 - */ - public LettuceConnectionFactory(RedisConfiguration redisConfiguration, LettuceClientConfiguration clientConfig) { - - this(clientConfig); - - Assert.notNull(redisConfiguration, "RedisConfiguration must not be null"); - - this.configuration = redisConfiguration; - } - - /** - * Constructs a new {@link LettuceConnectionFactory} instance using the given {@link RedisSentinelConfiguration} and - * {@link LettuceClientConfiguration}. - * - * @param sentinelConfiguration must not be {@literal null}. - * @param clientConfig must not be {@literal null}. - * @since 2.0 - */ - public LettuceConnectionFactory(RedisSentinelConfiguration sentinelConfiguration, - LettuceClientConfiguration clientConfig) { - - this(clientConfig); - - Assert.notNull(sentinelConfiguration, "RedisSentinelConfiguration must not be null"); - - this.configuration = sentinelConfiguration; - } - - /** - * Constructs a new {@link LettuceConnectionFactory} instance using the given {@link RedisClusterConfiguration} and - * {@link LettuceClientConfiguration}. - * - * @param clusterConfiguration must not be {@literal null}. - * @param clientConfig must not be {@literal null}. - * @since 2.0 - */ - public LettuceConnectionFactory(RedisClusterConfiguration clusterConfiguration, - LettuceClientConfiguration clientConfig) { - - this(clientConfig); - - Assert.notNull(clusterConfiguration, "RedisClusterConfiguration must not be null"); - - this.configuration = clusterConfiguration; - } + private static final ExceptionTranslationStrategy EXCEPTION_TRANSLATION = + new PassThroughExceptionTranslationStrategy(LettuceExceptionConverter.INSTANCE); /** * Creates a {@link RedisConfiguration} based on a {@link String URI} according to the following: @@ -298,11 +125,12 @@ public class LettuceConnectionFactory implements RedisConnectionFactory, Reactiv * @param redisUri the connection URI in the format of a {@link RedisURI}. * @return an appropriate {@link RedisConfiguration} instance representing the Redis URI. * @since 2.5.3 + * @see #createRedisConfiguration(RedisURI) * @see RedisURI */ public static RedisConfiguration createRedisConfiguration(String redisUri) { - Assert.hasText(redisUri, "RedisURI must not be null and not empty"); + Assert.hasText(redisUri, "RedisURI must not be null or empty"); return createRedisConfiguration(RedisURI.create(redisUri)); } @@ -335,11 +163,190 @@ public class LettuceConnectionFactory implements RedisConnectionFactory, Reactiv return LettuceConverters.createRedisStandaloneConfiguration(redisUri); } + private boolean validateConnection = false; + private boolean shareNativeConnection = true; + private boolean eagerInitialization = false; + private boolean convertPipelineAndTxResults = true; + + private int phase = 0; // in between min and max values + + private @Nullable AbstractRedisClient client; + + private final AtomicReference state = new AtomicReference<>(State.CREATED); + + private @Nullable ClusterCommandExecutor clusterCommandExecutor; + + private final LettuceClientConfiguration clientConfiguration; + + private @Nullable LettuceConnectionProvider connectionProvider; + private @Nullable LettuceConnectionProvider reactiveConnectionProvider; + + private final Log log = LogFactory.getLog(getClass()); + + /** Synchronization monitor for the shared Connection */ + private final Object connectionMonitor = new Object(); + + private PipeliningFlushPolicy pipeliningFlushPolicy = PipeliningFlushPolicy.flushEachCommand(); + + private @Nullable RedisConfiguration configuration; + + private RedisStandaloneConfiguration standaloneConfig = + new RedisStandaloneConfiguration("localhost", 6379); + + private @Nullable SharedConnection connection; + private @Nullable SharedConnection reactiveConnection; + + /** + * Lifecycle state of this factory. + */ + enum State { + CREATED, STARTING, STARTED, STOPPING, STOPPED, DESTROYED; + } + + /** + * Constructs a new {@link LettuceConnectionFactory} instance with default settings. + */ + public LettuceConnectionFactory() { + this(new MutableLettuceClientConfiguration()); + } + + /** + * Constructs a new {@link LettuceConnectionFactory} instance with default settings. + */ + public LettuceConnectionFactory(String host, int port) { + this(new RedisStandaloneConfiguration(host, port), new MutableLettuceClientConfiguration()); + } + + /** + * Constructs a new {@link LettuceConnectionFactory} instance given {@link LettuceClientConfiguration}. + * + * @param clientConfiguration must not be {@literal null} + * @since 2.0 + */ + private LettuceConnectionFactory(LettuceClientConfiguration clientConfiguration) { + + Assert.notNull(clientConfiguration, "LettuceClientConfiguration must not be null"); + + this.clientConfiguration = clientConfiguration; + this.configuration = this.standaloneConfig; + } + + /** + * Constructs a new {@link LettuceConnectionFactory} instance using the given {@link RedisSocketConfiguration}. + * + * @param redisConfiguration must not be {@literal null}. + * @since 2.1 + */ + public LettuceConnectionFactory(RedisConfiguration redisConfiguration) { + this(redisConfiguration, new MutableLettuceClientConfiguration()); + } + + /** + * Constructs a new {@link LettuceConnectionFactory} instance using the given + * {@link RedisStaticMasterReplicaConfiguration} and {@link LettuceClientConfiguration}. + * + * @param redisConfiguration must not be {@literal null}. + * @param clientConfiguration must not be {@literal null}. + * @since 2.1 + */ + public LettuceConnectionFactory(RedisConfiguration redisConfiguration, + LettuceClientConfiguration clientConfiguration) { + + this(clientConfiguration); + + Assert.notNull(redisConfiguration, "RedisConfiguration must not be null"); + + this.configuration = redisConfiguration; + } + + /** + * Constructs a new {@link LettuceConnectionFactory} instance using the given {@link RedisClusterConfiguration} + * applied to create a {@link RedisClusterClient}. + * + * @param clusterConfiguration must not be {@literal null}. + * @since 1.7 + */ + public LettuceConnectionFactory(RedisClusterConfiguration clusterConfiguration) { + this(clusterConfiguration, new MutableLettuceClientConfiguration()); + } + + /** + * Constructs a new {@link LettuceConnectionFactory} instance using the given {@link RedisClusterConfiguration} and + * {@link LettuceClientConfiguration}. + * + * @param clusterConfiguration must not be {@literal null}. + * @param clientConfiguration must not be {@literal null}. + * @since 2.0 + */ + public LettuceConnectionFactory(RedisClusterConfiguration clusterConfiguration, + LettuceClientConfiguration clientConfiguration) { + + this(clientConfiguration); + + Assert.notNull(clusterConfiguration, "RedisClusterConfiguration must not be null"); + + this.configuration = clusterConfiguration; + } + + /** + * Constructs a new {@link LettuceConnectionFactory} instance using the given {@link RedisSentinelConfiguration}. + * + * @param sentinelConfiguration must not be {@literal null}. + * @since 1.6 + */ + public LettuceConnectionFactory(RedisSentinelConfiguration sentinelConfiguration) { + this(sentinelConfiguration, new MutableLettuceClientConfiguration()); + } + + /** + * Constructs a new {@link LettuceConnectionFactory} instance using the given {@link RedisSentinelConfiguration} and + * {@link LettuceClientConfiguration}. + * + * @param sentinelConfiguration must not be {@literal null}. + * @param clientConfiguration must not be {@literal null}. + * @since 2.0 + */ + public LettuceConnectionFactory(RedisSentinelConfiguration sentinelConfiguration, + LettuceClientConfiguration clientConfiguration) { + + this(clientConfiguration); + + Assert.notNull(sentinelConfiguration, "RedisSentinelConfiguration must not be null"); + + this.configuration = sentinelConfiguration; + } + + /** + * Constructs a new {@link LettuceConnectionFactory} instance with default settings. + */ + public LettuceConnectionFactory(RedisStandaloneConfiguration configuration) { + this(configuration, new MutableLettuceClientConfiguration()); + } + + /** + * Constructs a new {@link LettuceConnectionFactory} instance using the given {@link RedisStandaloneConfiguration} and + * {@link LettuceClientConfiguration}. + * + * @param standaloneConfiguration must not be {@literal null}. + * @param clientConfiguration must not be {@literal null}. + * @since 2.0 + */ + public LettuceConnectionFactory(RedisStandaloneConfiguration standaloneConfiguration, + LettuceClientConfiguration clientConfiguration) { + + this(clientConfiguration); + + Assert.notNull(standaloneConfiguration, "RedisStandaloneConfiguration must not be null"); + + this.standaloneConfig = standaloneConfiguration; + this.configuration = this.standaloneConfig; + } + @Override public void start() { - State current = state - .getAndUpdate(state -> State.CREATED.equals(state) || State.STOPPED.equals(state) ? State.STARTING : 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)) { @@ -350,7 +357,6 @@ public class LettuceConnectionFactory implements RedisConnectionFactory, Reactiv createConnectionProvider(client, LettuceReactiveRedisConnection.CODEC)); if (isClusterAware()) { - this.clusterCommandExecutor = new ClusterCommandExecutor( new LettuceClusterTopologyProvider((RedisClusterClient) client), new LettuceClusterConnection.LettuceClusterNodeResourceProvider(this.connectionProvider), @@ -386,10 +392,9 @@ public class LettuceConnectionFactory implements RedisConnectionFactory, Reactiv client.shutdown(quietPeriod.toMillis(), timeout.toMillis(), TimeUnit.MILLISECONDS); client = null; - } catch (Exception e) { - + } catch (Exception cause) { if (log.isWarnEnabled()) { - log.warn(ClassUtils.getShortName(client.getClass()) + " did not shut down gracefully.", e); + log.warn(ClassUtils.getShortName(client.getClass()) + " did not shut down gracefully.", cause); } } } @@ -430,26 +435,26 @@ public class LettuceConnectionFactory implements RedisConnectionFactory, Reactiv stop(); client = null; - if (clusterCommandExecutor != null) { + if (clusterCommandExecutor != null) { try { clusterCommandExecutor.destroy(); } catch (Exception ex) { log.warn("Cannot properly close cluster command executor", ex); } } + state.set(State.DESTROYED); } private void dispose(@Nullable LettuceConnectionProvider connectionProvider) { - if (connectionProvider instanceof DisposableBean) { + if (connectionProvider instanceof DisposableBean disposableBean) { try { - ((DisposableBean) connectionProvider).destroy(); - } catch (Exception e) { - + disposableBean.destroy(); + } catch (Exception cause) { if (log.isWarnEnabled()) { - log.warn(connectionProvider + " did not shut down gracefully.", e); + log.warn(connectionProvider + " did not shut down gracefully.", cause); } } } @@ -464,9 +469,11 @@ public class LettuceConnectionFactory implements RedisConnectionFactory, Reactiv return getClusterConnection(); } - LettuceConnection connection = doCreateLettuceConnection(getSharedConnection(), connectionProvider, getTimeout(), - getDatabase()); + LettuceConnection connection = doCreateLettuceConnection(getSharedConnection(), connectionProvider, + getTimeout(), getDatabase()); + connection.setConvertPipelineAndTxResults(convertPipelineAndTxResults); + return connection; } @@ -484,6 +491,7 @@ public class LettuceConnectionFactory implements RedisConnectionFactory, Reactiv StatefulRedisClusterConnection sharedConnection = getSharedClusterConnection(); LettuceClusterTopologyProvider topologyProvider = new LettuceClusterTopologyProvider(clusterClient); + return doCreateLettuceClusterConnection(sharedConnection, connectionProvider, topologyProvider, clusterCommandExecutor, clientConfiguration.getCommandTimeout()); } @@ -505,6 +513,7 @@ public class LettuceConnectionFactory implements RedisConnectionFactory, Reactiv long timeout, int database) { LettuceConnection connection = new LettuceConnection(sharedConnection, connectionProvider, timeout, database); + connection.setPipeliningFlushPolicy(this.pipeliningFlushPolicy); return connection; @@ -530,6 +539,7 @@ public class LettuceConnectionFactory implements RedisConnectionFactory, Reactiv LettuceClusterConnection connection = new LettuceClusterConnection(sharedConnection, connectionProvider, topologyProvider, clusterCommandExecutor, commandTimeout); + connection.setPipeliningFlushPolicy(this.pipeliningFlushPolicy); return connection; @@ -865,7 +875,6 @@ public class LettuceConnectionFactory implements RedisConnectionFactory, Reactiv Assert.isTrue(index >= 0, "invalid DB index (a positive index required)"); if (RedisConfiguration.isDatabaseIndexAware(configuration)) { - ((WithDatabaseIndex) configuration).setDatabase(index); return; } @@ -930,7 +939,8 @@ public class LettuceConnectionFactory implements RedisConnectionFactory, Reactiv AbstractRedisClient client = getNativeClient(); - Assert.state(client != null, "Client not yet initialized; Did you forget to call initialize the bean"); + Assert.state(client != null, + "Client not yet initialized; Did you forget to call initialize the bean"); return client; } @@ -965,7 +975,6 @@ public class LettuceConnectionFactory implements RedisConnectionFactory, Reactiv public void setPassword(String password) { if (RedisConfiguration.isAuthenticationAware(configuration)) { - ((WithPassword) configuration).setPassword(password); return; } @@ -1025,7 +1034,7 @@ public class LettuceConnectionFactory implements RedisConnectionFactory, Reactiv * @since 2.0 */ public LettuceClientConfiguration getClientConfiguration() { - return clientConfiguration; + return this.clientConfiguration; } /** @@ -1033,7 +1042,7 @@ public class LettuceConnectionFactory implements RedisConnectionFactory, Reactiv * @since 2.0 */ public RedisStandaloneConfiguration getStandaloneConfiguration() { - return standaloneConfig; + return this.standaloneConfig; } /** @@ -1042,7 +1051,7 @@ public class LettuceConnectionFactory implements RedisConnectionFactory, Reactiv */ @Nullable public RedisSocketConfiguration getSocketConfiguration() { - return isDomainSocketAware() ? (RedisSocketConfiguration) configuration : null; + return isDomainSocketAware() ? (RedisSocketConfiguration) this.configuration : null; } /** @@ -1051,7 +1060,7 @@ public class LettuceConnectionFactory implements RedisConnectionFactory, Reactiv */ @Nullable public RedisSentinelConfiguration getSentinelConfiguration() { - return isRedisSentinelAware() ? (RedisSentinelConfiguration) configuration : null; + return isRedisSentinelAware() ? (RedisSentinelConfiguration) this.configuration : null; } /** @@ -1060,7 +1069,7 @@ public class LettuceConnectionFactory implements RedisConnectionFactory, Reactiv */ @Nullable public RedisClusterConfiguration getClusterConfiguration() { - return isClusterAware() ? (RedisClusterConfiguration) configuration : null; + return isClusterAware() ? (RedisClusterConfiguration) this.configuration : null; } /** @@ -1125,8 +1134,9 @@ public class LettuceConnectionFactory implements RedisConnectionFactory, Reactiv */ @Nullable protected StatefulRedisConnection getSharedConnection() { + return shareNativeConnection && !isClusterAware() - ? (StatefulRedisConnection) getOrCreateSharedConnection().getConnection() + ? (StatefulRedisConnection) getOrCreateSharedConnection().getConnection() : null; } @@ -1138,8 +1148,9 @@ public class LettuceConnectionFactory implements RedisConnectionFactory, Reactiv */ @Nullable protected StatefulRedisClusterConnection getSharedClusterConnection() { + return shareNativeConnection && isClusterAware() - ? (StatefulRedisClusterConnection) getOrCreateSharedConnection().getConnection() + ? (StatefulRedisClusterConnection) getOrCreateSharedConnection().getConnection() : null; } @@ -1157,9 +1168,8 @@ public class LettuceConnectionFactory implements RedisConnectionFactory, Reactiv LettuceConnectionProvider connectionProvider = doCreateConnectionProvider(client, codec); - if (this.clientConfiguration instanceof LettucePoolingClientConfiguration) { - return new LettucePoolingConnectionProvider(connectionProvider, - (LettucePoolingClientConfiguration) this.clientConfiguration); + if (this.clientConfiguration instanceof LettucePoolingClientConfiguration poolingClientConfiguration) { + return new LettucePoolingConnectionProvider(connectionProvider, poolingClientConfiguration); } return connectionProvider; @@ -1181,7 +1191,7 @@ public class LettuceConnectionFactory implements RedisConnectionFactory, Reactiv return isStaticMasterReplicaAware() ? createStaticMasterReplicaConnectionProvider((RedisClient) client, codec) : isClusterAware() ? createClusterConnectionProvider((RedisClusterClient) client, codec) - : createStandaloneConnectionProvider((RedisClient) client, codec); + : createStandaloneConnectionProvider((RedisClient) client, codec); } @SuppressWarnings("all") @@ -1190,7 +1200,8 @@ public class LettuceConnectionFactory implements RedisConnectionFactory, Reactiv List nodes = ((RedisStaticMasterReplicaConfiguration) this.configuration).getNodes().stream() .map(it -> createRedisURIAndApplySettings(it.getHostName(), it.getPort())) - .peek(it -> it.setDatabase(getDatabase())).collect(Collectors.toList()); + .peek(it -> it.setDatabase(getDatabase())) + .collect(Collectors.toList()); return new StaticMasterReplicaConnectionProvider(client, codec, nodes, getClientConfiguration().getReadFrom().orElse(null)); @@ -1208,12 +1219,14 @@ public class LettuceConnectionFactory implements RedisConnectionFactory, Reactiv return isStaticMasterReplicaAware() ? createStaticMasterReplicaClient() : isRedisSentinelAware() ? createSentinelClient() - : isClusterAware() ? createClusterClient() : createBasicClient(); + : 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); @@ -1237,8 +1250,9 @@ public class LettuceConnectionFactory implements RedisConnectionFactory, Reactiv @SuppressWarnings("all") private RedisURI getSentinelRedisURI() { - RedisURI redisUri = LettuceConverters - .sentinelConfigurationToRedisURI((RedisSentinelConfiguration) this.configuration); + RedisSentinelConfiguration sentinelConfiguration = (RedisSentinelConfiguration) this.configuration; + + RedisURI redisUri = LettuceConverters.sentinelConfigurationToRedisURI(sentinelConfiguration); applyToAll(redisUri, it -> { @@ -1256,8 +1270,8 @@ public class LettuceConnectionFactory implements RedisConnectionFactory, Reactiv redisUri.setCredentialsProvider(factory.createCredentialsProvider(this.configuration)); - RedisCredentialsProvider sentinelCredentials = factory - .createSentinelCredentialsProvider((RedisSentinelConfiguration) this.configuration); + RedisCredentialsProvider sentinelCredentials = + factory.createSentinelCredentialsProvider((RedisSentinelConfiguration) this.configuration); redisUri.getSentinels().forEach(it -> it.setCredentialsProvider(sentinelCredentials)); }); @@ -1270,44 +1284,50 @@ public class LettuceConnectionFactory implements RedisConnectionFactory, Reactiv List initialUris = new ArrayList<>(); - ClusterConfiguration configuration = (ClusterConfiguration) this.configuration; + ClusterConfiguration clusterConfiguration = (ClusterConfiguration) this.configuration; - configuration.getClusterNodes().stream().map(node -> createRedisURIAndApplySettings(node.getHost(), node.getPort())) + clusterConfiguration.getClusterNodes().stream() + .map(node -> createRedisURIAndApplySettings(node.getHost(), node.getPort())) .forEach(initialUris::add); RedisClusterClient clusterClient = this.clientConfiguration.getClientResources() .map(clientResources -> RedisClusterClient.create(clientResources, initialUris)) .orElseGet(() -> RedisClusterClient.create(initialUris)); - clusterClient.setOptions(getClusterClientOptions(configuration)); + clusterClient.setOptions(getClusterClientOptions(clusterConfiguration)); return clusterClient; } - private ClusterClientOptions getClusterClientOptions(ClusterConfiguration configuration) { + private ClusterClientOptions getClusterClientOptions(ClusterConfiguration clusterConfiguration) { 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)); + Optional clusterClientOptions = clientOptions + .filter(ClusterClientOptions.class::isInstance) + .map(ClusterClientOptions.class::cast); - if (configuration.getMaxRedirects() != null) { - return clusterClientOptions.mutate().maxRedirects(configuration.getMaxRedirects()).build(); + ClusterClientOptions resolvedClusterClientOptions = clusterClientOptions.orElseGet(() -> clientOptions + .map(it -> ClusterClientOptions.builder(it).build()) + .orElseGet(ClusterClientOptions::create)); + + if (clusterConfiguration.getMaxRedirects() != null) { + return resolvedClusterClientOptions.mutate().maxRedirects(clusterConfiguration.getMaxRedirects()).build(); } - return clusterClientOptions; + return resolvedClusterClientOptions; } @SuppressWarnings("all") private RedisClient createBasicClient() { RedisURI uri = isDomainSocketAware() - ? createRedisSocketURIAndApplySettings(((DomainSocketConfiguration) this.configuration).getSocket()) + ? createRedisSocketURIAndApplySettings(getSocketConfiguration().getSocket()) : 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); @@ -1376,8 +1396,8 @@ 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 @@ -1401,6 +1421,12 @@ public class LettuceConnectionFactory implements RedisConnectionFactory, Reactiv return clientConfiguration.getCommandTimeout().toMillis(); } + private void logWarning(String message, Object... arguments) { + if (this.log.isWarnEnabled()) { + this.log.warn(String.format(message, arguments)); + } + } + /** * Wrapper for shared connections. Keeps track of the connection lifecycleThe wrapper is thread-safe as it * synchronizes concurrent calls by blocking. @@ -1474,9 +1500,10 @@ public class LettuceConnectionFactory implements RedisConnectionFactory, Reactiv if (connection instanceof StatefulRedisClusterConnection) { ((StatefulRedisClusterConnection) connection).sync().ping(); } + valid = true; - } catch (Exception e) { - log.debug("Validation failed", e); + } catch (Exception cause) { + log.debug("Validation failed", cause); } } @@ -1516,11 +1543,14 @@ public class LettuceConnectionFactory implements RedisConnectionFactory, Reactiv private boolean useSsl; private boolean verifyPeer = true; private boolean startTls; + private @Nullable ClientResources clientResources; - private @Nullable String clientName; + private Duration timeout = Duration.ofSeconds(RedisURI.DEFAULT_TIMEOUT); private Duration shutdownTimeout = Duration.ofMillis(100); + private @Nullable String clientName; + @Override public boolean isUseSsl() { return useSsl; @@ -1626,8 +1656,8 @@ public class LettuceConnectionFactory implements RedisConnectionFactory, Reactiv try { return delegate.getConnection(connectionType); - } catch (RuntimeException e) { - throw translateException(e); + } catch (RuntimeException cause) { + throw translateException(cause); } } @@ -1636,8 +1666,8 @@ public class LettuceConnectionFactory implements RedisConnectionFactory, Reactiv try { return ((TargetAware) delegate).getConnection(connectionType, redisURI); - } catch (RuntimeException e) { - throw translateException(e); + } catch (RuntimeException cause) { + throw translateException(cause); } } @@ -1689,15 +1719,14 @@ public class LettuceConnectionFactory implements RedisConnectionFactory, Reactiv @Override public void destroy() throws Exception { - if (delegate instanceof DisposableBean) { - ((DisposableBean) delegate).destroy(); + if (delegate instanceof DisposableBean disposableBean) { + disposableBean.destroy(); } } - private RuntimeException translateException(Throwable e) { - return e instanceof RedisConnectionFailureException ? (RedisConnectionFailureException) e - : new RedisConnectionFailureException("Unable to connect to Redis", e); + private RuntimeException translateException(Throwable cause) { + return cause instanceof RedisConnectionFailureException connectionFailure ? connectionFailure + : new RedisConnectionFailureException("Unable to connect to Redis", cause); } - } } diff --git a/src/test/java/org/springframework/data/redis/connection/lettuce/LettuceConnectionFactoryRedisURITests.java b/src/test/java/org/springframework/data/redis/connection/lettuce/LettuceConnectionFactoryRedisURITests.java deleted file mode 100644 index 4bdf03feb..000000000 --- a/src/test/java/org/springframework/data/redis/connection/lettuce/LettuceConnectionFactoryRedisURITests.java +++ /dev/null @@ -1,29 +0,0 @@ -/* - * Copyright 2015-2023 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 - * - * https://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - */ -package org.springframework.data.redis.connection.lettuce; - -import io.lettuce.core.RedisURI; - -/** - * Unit tests for the {@link LettuceConnectionFactory#createRedisConfiguration(RedisURI)} factory method. - * - * @author Chris Bono - */ -class LettuceConnectionFactoryRedisURITests { - - - -} 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 514110806..ae8a4e456 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 @@ -38,6 +38,7 @@ import io.lettuce.core.resource.ClientResources; import reactor.test.StepVerifier; import java.time.Duration; +import java.util.Arrays; import java.util.Collections; import java.util.Objects; import java.util.concurrent.CompletableFuture; @@ -64,6 +65,8 @@ import org.springframework.data.redis.connection.RedisStandaloneConfiguration; import org.springframework.data.redis.test.extension.LettuceTestClientResources; import org.springframework.test.util.ReflectionTestUtils; +import org.assertj.core.api.InstanceOfAssertFactories; + /** * Unit tests for {@link LettuceConnectionFactory}. * @@ -1125,6 +1128,7 @@ class LettuceConnectionFactoryUnitTests { @Test // GH-2116 void createRedisConfigurationRequiresRedisUri() { + assertThatIllegalArgumentException() .isThrownBy(() -> LettuceConnectionFactory.createRedisConfiguration((RedisURI) null)) .withMessage("RedisURI must not be null"); @@ -1236,6 +1240,35 @@ class LettuceConnectionFactoryUnitTests { assertThat(connectionFactory.isRunning()).isTrue(); } + @Test // GH-2594 + void createRedisConfigurationWithNullInvalidRedisUriString() { + + Arrays.asList(" ", "", null).forEach(redisUri -> + assertThatIllegalArgumentException() + .isThrownBy(() -> LettuceConnectionFactory.createRedisConfiguration(redisUri)) + .withMessage("RedisURI must not be null or empty") + .withNoCause()); + } + + @Test + public void createRedisConfigurationWithValidRedisUriString() { + + RedisConfiguration redisConfiguration = + LettuceConnectionFactory.createRedisConfiguration("redis://skullbox:6789"); + + assertThat(redisConfiguration).isInstanceOf(RedisStandaloneConfiguration.class); + + assertThat(redisConfiguration) + .asInstanceOf(InstanceOfAssertFactories.type(RedisStandaloneConfiguration.class)) + .extracting(RedisStandaloneConfiguration::getHostName) + .isEqualTo("skullbox"); + + assertThat(redisConfiguration) + .asInstanceOf(InstanceOfAssertFactories.type(RedisStandaloneConfiguration.class)) + .extracting(RedisStandaloneConfiguration::getPort) + .isEqualTo(6789); + } + static class CustomRedisConfiguration implements RedisConfiguration, WithHostAndPort { private String hostName;