Spring/spring-data-redis
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Polishing.

Browse Source 
Organize source code and cleanup compiler warnings.

See #2594
Original pull request: #2669
This commit is contained in:
John Blum
2023-08-08 16:16:15 -07:00
committed by Mark Paluch
parent 76c1830c57
commit 2eaf1746da
8 changed files with 1311 additions and 1157 deletions
Download Patch File Download Diff File Expand all files Collapse all files

96
src/main/java/org/springframework/data/redis/connection/ClusterCommandExecutor.java
View File

@@ -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 <T> NodeResult<T> executeCommandOnArbitraryNode(ClusterCommandCallback<?, T> cmd) {
public <T> NodeResult<T> executeCommandOnArbitraryNode(ClusterCommandCallback<?, T> commandCallback) {
Assert.notNull(commandCallback, "ClusterCommandCallback must not be null");
Assert.notNull(cmd, "ClusterCommandCallback must not be null");
List<RedisClusterNode> 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 <S, T> NodeResult<T> executeCommandOnSingleNode(ClusterCommandCallback<S, T> 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 <S, T> MultiNodeResult<T> executeCommandOnAllNodes(final ClusterCommandCallback<S, T> 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 <S, T> MultiNodeResult<T> executeCommandAsyncOnNodes(ClusterCommandCallback<S, T> callback,
@@ -202,6 +214,7 @@ public class ClusterCommandExecutor implements DisposableBean {
}
Map<NodeExecution, Future<NodeResult<T>>> 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<T> result = new MultiNodeResult<>();
Map<RedisClusterNode, Throwable> exceptions = new HashMap<>();
MultiNodeResult<T> result = new MultiNodeResult<>();
Set<String> 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 <S, T> MultiNodeResult<T> executeMultiKeyCommand(MultiKeyClusterCommandCallback<S, T> cmd,
Iterable<byte[]> keys) {
Map<RedisClusterNode, PositionalKeys> 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<NodeExecution, Future<NodeResult<T>>> futures = new LinkedHashMap<>();
for (Entry<RedisClusterNode, PositionalKeys> 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();

24
src/main/java/org/springframework/data/redis/connection/RedisConnectionFactory.java
View File

@@ -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.
* <p>
* 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.
* <p>
* 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.
*

261
src/main/java/org/springframework/data/redis/connection/jedis/JedisConnection.java
View File

@@ -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 <a href="https://github.com/redis/jedis">Jedis</a> library.
* <p>
@@ -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<Jedis> pool;
private final JedisClientConfig sentinelConfig;
private final Log LOGGER = LogFactory.getLog(getClass());
private List<JedisResult> pipelinedResults = new ArrayList<>();
private final @Nullable Pool<Jedis> pool;
private Queue<FutureResult<Response<?>>> 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 <code>JedisConnection</code> 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 <code>JedisConnection</code> 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<Jedis> pool, int dbIndex) {
this(jedis, pool, dbIndex, null);
}
/**
* Constructs a new <code>JedisConnection</code> 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<Jedis> 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 <code>JedisConnection</code> 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<Jedis> 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<Jedis, Object> directFunction,
Function<ResponseCommands, Response<Object>> pipelineFunction, Converter<Object, Object> 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<Object> 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<Object> closePipeline() {
if (pipeline != null) {
try {
return convertPipelineResults();
@@ -390,14 +396,19 @@ public class JedisConnection extends AbstractRedisConnection {
pipelinedResults.clear();
}
}
return Collections.emptyList();
}
private List<Object> convertPipelineResults() {
List<Object> 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<Object> 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;
}
<T> JedisResult<T, T> newJedisResult(Response<T> 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;
}
}

338
src/main/java/org/springframework/data/redis/connection/jedis/JedisConnectionFactory.java
View File

@@ -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> 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<Jedis> 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> state = new AtomicReference<>(State.CREATED);
private JedisClientConfig clientConfig = DefaultJedisClientConfig.builder().build();
private @Nullable Pool<Jedis> 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<Jedis> createPool() {
@@ -417,6 +436,7 @@ public class JedisConnectionFactory
GenericObjectPoolConfig<Jedis> 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> 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;
}
}
}

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

File diff suppressed because it is too large Load Diff

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

@@ -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> state = new AtomicReference<>(State.CREATED);
private @Nullable AbstractRedisClient client;
private @Nullable LettuceConnectionProvider connectionProvider;
private @Nullable LettuceConnectionProvider reactiveConnectionProvider;
private @Nullable SharedConnection<byte[]> connection;
private @Nullable SharedConnection<ByteBuffer> 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> 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<byte[]> connection;
private @Nullable SharedConnection<ByteBuffer> 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<byte[], byte[]> 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<byte[], byte[]> getSharedConnection() {
return shareNativeConnection && !isClusterAware()
? (StatefulRedisConnection) getOrCreateSharedConnection().getConnection()
? (StatefulRedisConnection<byte[], byte[]>) getOrCreateSharedConnection().getConnection()
: null;
}
@@ -1138,8 +1148,9 @@ public class LettuceConnectionFactory implements RedisConnectionFactory, Reactiv
*/
@Nullable
protected StatefulRedisClusterConnection<byte[], byte[]> getSharedClusterConnection() {
return shareNativeConnection && isClusterAware()
? (StatefulRedisClusterConnection) getOrCreateSharedConnection().getConnection()
? (StatefulRedisClusterConnection<byte[], byte[]>) 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<RedisURI> 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<RedisURI> 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> 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> 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);
}
}
}