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

DATAREDIS-348 - Switch to mp911de/lettuce 3.0.

Browse Source 
Updated lettuce redis client to latest snapshot, adopt changed API and use API for commands which were emulated using LUA script

Implemente Sentinel support using lettuce library, adopt finer grained exceptions and aligned exception behavior to match other libraries. Added shutdown-timeout property in order to control shutdown duration of the netty framework which reduces test run duration. Documentation updated as well. Contains also a fix to use psetex instead of setex (typo).

Switch to Jedis for Redis Version discovery in tests. Polished documentation. Use LettuceConverter and native time() method.

Original pull request: #144.
Related pull request: #104.
This commit is contained in:
Mark Paluch
2014-09-12 22:28:24 +02:00
committed by Oliver Gierke
parent 4cd6b82e7b
commit 7426b27688
23 changed files with 1237 additions and 200 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -44,7 +44,8 @@ public interface RedisConnectionFactory extends PersistenceExceptionTranslator {
boolean getConvertPipelineAndTxResults();
/**
* @return
* Provides a suitable connection for interacting with Redis Sentinel.
* @return connection for interacting with Redis Sentinel.
* @since 1.4
*/
RedisSentinelConnection getSentinelConnection();

2
src/main/java/org/springframework/data/redis/connection/lettuce/AuthenticatingRedisClient.java
View File

@@ -25,7 +25,9 @@ import com.lambdaworks.redis.pubsub.RedisPubSubConnection;
* Extension of {@link RedisClient} that calls auth on all new connections using the supplied credentials
*
* @author Jennifer Hickey
* @deprecated Use password in RedisURI
*/
@Deprecated
public class AuthenticatingRedisClient extends RedisClient {
private String password;

51
src/main/java/org/springframework/data/redis/connection/lettuce/DefaultLettucePool.java
View File

@@ -17,6 +17,7 @@ package org.springframework.data.redis.connection.lettuce;
import java.util.concurrent.TimeUnit;
import com.lambdaworks.redis.RedisURI;
import org.apache.commons.pool2.BasePooledObjectFactory;
import org.apache.commons.pool2.PooledObject;
import org.apache.commons.pool2.impl.DefaultPooledObject;
@@ -24,6 +25,7 @@ import org.apache.commons.pool2.impl.GenericObjectPool;
import org.apache.commons.pool2.impl.GenericObjectPoolConfig;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.data.redis.connection.PoolException;
import org.springframework.data.redis.connection.RedisSentinelConfiguration;
import org.springframework.util.Assert;
import com.lambdaworks.redis.RedisAsyncConnection;
@@ -34,6 +36,7 @@ import com.lambdaworks.redis.RedisClient;
*
* @author Jennifer Hickey
* @author Christoph Strobl
* @author Mark Paluch
*/
public class DefaultLettucePool implements LettucePool, InitializingBean {
@@ -46,6 +49,7 @@ public class DefaultLettucePool implements LettucePool, InitializingBean {
private int port = 6379;
private String password;
private long timeout = TimeUnit.MILLISECONDS.convert(60, TimeUnit.SECONDS);
private RedisSentinelConfiguration sentinelConfiguration;
/**
* Constructs a new <code>DefaultLettucePool</code> instance with default settings.
@@ -63,6 +67,16 @@ public class DefaultLettucePool implements LettucePool, InitializingBean {
this.port = port;
}
/**
* Uses the {@link RedisSentinelConfiguration} and {@link RedisClient} defaults for configuring the connection pool
* based on sentinels.
*
* @param sentinelConfiguration The Sentinel configuration
*/
public DefaultLettucePool(RedisSentinelConfiguration sentinelConfiguration) {
this.sentinelConfiguration = sentinelConfiguration;
}
/**
* Uses the {@link RedisClient} defaults for configuring the connection pool
*
@@ -76,14 +90,43 @@ public class DefaultLettucePool implements LettucePool, InitializingBean {
this.poolConfig = poolConfig;
}
/**
* @return true when {@link RedisSentinelConfiguration} is present.
* @since 1.5
*/
public boolean isRedisSentinelAware() {
return sentinelConfiguration != null;
}
@SuppressWarnings({ "rawtypes" })
public void afterPropertiesSet() {
this.client = password != null ? new AuthenticatingRedisClient(hostName, port, password) : new RedisClient(
hostName, port);
this.client = new RedisClient(getRedisURI());
client.setDefaultTimeout(timeout, TimeUnit.MILLISECONDS);
this.internalPool = new GenericObjectPool<RedisAsyncConnection>(new LettuceFactory(client, dbIndex), poolConfig);
}
/**
*
* @return a RedisURI pointing either to a single Redis host or containing a set of sentinels.
*/
private RedisURI getRedisURI() {
if(isRedisSentinelAware()) {
return LettuceConverters.sentinelConfigurationToRedisURI(sentinelConfiguration);
}
return createSimpleHostRedisURI();
}
private RedisURI createSimpleHostRedisURI() {
RedisURI.Builder builder = RedisURI.Builder.redis(hostName, port);
if(password != null) {
builder.withPassword(password);
}
builder.withTimeout(timeout, TimeUnit.MILLISECONDS);
return builder.build();
}
@SuppressWarnings("unchecked")
public RedisAsyncConnection<byte[], byte[]> getResource() {
try {
@@ -118,6 +161,10 @@ public class DefaultLettucePool implements LettucePool, InitializingBean {
}
}
/**
*
* @return The Redis client
*/
public RedisClient getClient() {
return client;
}

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

@@ -17,10 +17,8 @@ package org.springframework.data.redis.connection.lettuce;
import static com.lambdaworks.redis.protocol.CommandType.*;
import java.io.ByteArrayOutputStream;
import java.io.IOException;
import java.io.OutputStream;
import java.lang.reflect.Constructor;
import java.lang.reflect.Method;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
@@ -33,6 +31,7 @@ import java.util.Properties;
import java.util.Queue;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
@@ -48,7 +47,9 @@ import org.springframework.data.redis.connection.AbstractRedisConnection;
import org.springframework.data.redis.connection.DataType;
import org.springframework.data.redis.connection.FutureResult;
import org.springframework.data.redis.connection.MessageListener;
import org.springframework.data.redis.connection.RedisNode;
import org.springframework.data.redis.connection.RedisPipelineException;
import org.springframework.data.redis.connection.RedisSentinelConnection;
import org.springframework.data.redis.connection.RedisSubscribedConnectionException;
import org.springframework.data.redis.connection.ReturnType;
import org.springframework.data.redis.connection.SortParameters;
@@ -65,12 +66,26 @@ import org.springframework.data.redis.core.types.RedisClientInfo;
import org.springframework.util.Assert;
import org.springframework.util.ClassUtils;
import org.springframework.util.ObjectUtils;
import org.springframework.util.ReflectionUtils;
import com.lambdaworks.redis.AbstractRedisClient;
import com.lambdaworks.redis.KeyScanCursor;
import com.lambdaworks.redis.LettuceFutures;
import com.lambdaworks.redis.MapScanCursor;
import com.lambdaworks.redis.RedisAsyncConnection;
import com.lambdaworks.redis.RedisAsyncConnectionImpl;
import com.lambdaworks.redis.RedisChannelHandler;
import com.lambdaworks.redis.RedisClient;
import com.lambdaworks.redis.RedisClusterConnection;
import com.lambdaworks.redis.RedisConnection;
import com.lambdaworks.redis.RedisException;
import com.lambdaworks.redis.ScriptOutputType;
import com.lambdaworks.redis.RedisSentinelAsyncConnection;
import com.lambdaworks.redis.RedisURI;
import com.lambdaworks.redis.ScanArgs;
import com.lambdaworks.redis.ScoredValue;
import com.lambdaworks.redis.ScoredValueScanCursor;
import com.lambdaworks.redis.SortArgs;
import com.lambdaworks.redis.ValueScanCursor;
import com.lambdaworks.redis.ZStoreArgs;
import com.lambdaworks.redis.codec.RedisCodec;
import com.lambdaworks.redis.output.BooleanOutput;
@@ -93,22 +108,31 @@ import com.lambdaworks.redis.protocol.CommandType;
import com.lambdaworks.redis.pubsub.RedisPubSubConnection;
/**
* {@code RedisConnection} implementation on top of <a href="https://github.com/wg/lettuce">Lettuce</a> Redis client.
* {@code RedisConnection} implementation on top of <a href="https://github.com/mp911de/lettuce">Lettuce</a> Redis
* client.
*
* @author Costin Leau
* @author Jennifer Hickey
* @author Christoph Strobl
* @author Thomas Darimont
* @author David Liu
* @author Mark Paluch
*/
public class LettuceConnection extends AbstractRedisConnection {
static final RedisCodec<byte[], byte[]> CODEC = new BytesRedisCodec();
private static final Method SYNC_HANDLER;
private static final ExceptionTranslationStrategy EXCEPTION_TRANSLATION = new FallbackExceptionTranslationStrategy(
LettuceConverters.exceptionConverter());
static final RedisCodec<byte[], byte[]> CODEC = new BytesRedisCodec();
private static final TypeHints typeHints = new TypeHints();
static {
SYNC_HANDLER = ReflectionUtils.findMethod(AbstractRedisClient.class, "syncHandler", RedisChannelHandler.class,
Class[].class);
ReflectionUtils.makeAccessible(SYNC_HANDLER);
}
private final com.lambdaworks.redis.RedisAsyncConnection<byte[], byte[]> asyncSharedConn;
private final com.lambdaworks.redis.RedisConnection<byte[], byte[]> sharedConn;
private com.lambdaworks.redis.RedisAsyncConnection<byte[], byte[]> asyncDedicatedConn;
@@ -142,10 +166,14 @@ public class LettuceConnection extends AbstractRedisConnection {
@SuppressWarnings("unchecked")
@Override
public Object get() {
if (convertPipelineAndTxResults && converter != null) {
return converter.convert(resultHolder.get());
try {
if (convertPipelineAndTxResults && converter != null) {
return converter.convert(resultHolder.get());
}
return resultHolder.get();
} catch (ExecutionException e) {
throw EXCEPTION_TRANSLATION.translate(e);
}
return resultHolder.get();
}
}
@@ -267,9 +295,8 @@ public class LettuceConnection extends AbstractRedisConnection {
RedisClient client, LettucePool pool) {
this.asyncSharedConn = sharedConnection;
this.timeout = timeout;
this.sharedConn = sharedConnection != null ? new com.lambdaworks.redis.RedisConnection<byte[], byte[]>(
asyncSharedConn) : null;
this.client = client;
this.sharedConn = sharedConnection != null ? syncConnection(asyncSharedConn) : null;
this.pool = pool;
}
@@ -284,11 +311,11 @@ public class LettuceConnection extends AbstractRedisConnection {
}
@SuppressWarnings({ "rawtypes", "unchecked" })
private Object await(Command cmd) {
if (isMulti && cmd.type != MULTI) {
private Object await(com.lambdaworks.redis.protocol.RedisCommand cmd) {
if (isMulti && cmd instanceof Command && ((Command) cmd).isMulti()) {
return null;
}
return getAsyncConnection().await(cmd, timeout, TimeUnit.MILLISECONDS);
return LettuceFutures.await(cmd, timeout, TimeUnit.MILLISECONDS);
}
@Override
@@ -311,26 +338,30 @@ public class LettuceConnection extends AbstractRedisConnection {
Assert.hasText(command, "a valid command needs to be specified");
try {
String name = command.trim().toUpperCase();
CommandType cmd = CommandType.valueOf(name);
CommandType commandType = CommandType.valueOf(name);
validateCommandIfRunningInTransactionMode(cmd, args);
validateCommandIfRunningInTransactionMode(commandType, args);
CommandArgs<byte[], byte[]> cmdArg = new CommandArgs<byte[], byte[]>(CODEC);
if (!ObjectUtils.isEmpty(args)) {
cmdArg.addKeys(args);
}
CommandOutput expectedOutput = commandOutputTypeHint != null ? commandOutputTypeHint : typeHints.getTypeHint(cmd);
RedisAsyncConnectionImpl<byte[], byte[]> connectionImpl = (RedisAsyncConnectionImpl<byte[], byte[]>) getAsyncConnection();
CommandOutput expectedOutput = commandOutputTypeHint != null ? commandOutputTypeHint : typeHints
.getTypeHint(commandType);
Command cmd = new Command(commandType, expectedOutput, cmdArg);
if (isPipelined()) {
pipeline(new LettuceResult(getAsyncConnection().dispatch(cmd, expectedOutput, cmdArg)));
pipeline(new LettuceResult(connectionImpl.dispatch(cmd)));
return null;
} else if (isQueueing()) {
transaction(new LettuceTxResult(getAsyncConnection().dispatch(cmd, expectedOutput, cmdArg)));
transaction(new LettuceTxResult(connectionImpl.dispatch(cmd)));
return null;
} else {
return await(getAsyncConnection().dispatch(cmd, expectedOutput, cmdArg));
return await(connectionImpl.dispatch(cmd));
}
} catch (RedisException ex) {
throw convertLettuceAccessException(ex);
@@ -405,7 +436,8 @@ public class LettuceConnection extends AbstractRedisConnection {
for (LettuceResult result : ppline) {
futures.add(result.getResultHolder());
}
boolean done = getAsyncConnection().awaitAll(futures.toArray(new Command[futures.size()]));
boolean done = LettuceFutures.awaitAll(timeout, TimeUnit.MILLISECONDS,
futures.toArray(new Command[futures.size()]));
List<Object> results = new ArrayList<Object>(futures.size());
Exception problem = null;
@@ -1257,48 +1289,27 @@ public class LettuceConnection extends AbstractRedisConnection {
}
/**
* {@code pSetEx} is not directly supported and therefore emulated via {@literal lua script}.
*
* @since 1.3
* @see org.springframework.data.redis.connection.RedisStringCommands#pSetEx(byte[], long, byte[])
*/
@Override
public void pSetEx(byte[] key, long milliseconds, byte[] value) {
byte[] script = createRedisScriptForPSetEx(key, milliseconds, value);
byte[][] emptyArgs = new byte[0][0];
try {
if (isPipelined()) {
pipeline(new LettuceStatusResult(getAsyncConnection().eval(script, ScriptOutputType.STATUS, emptyArgs,
emptyArgs)));
pipeline(new LettuceStatusResult(getAsyncConnection().psetex(key, milliseconds, value)));
return;
}
if (isQueueing()) {
transaction(new LettuceTxStatusResult(getConnection().eval(script, ScriptOutputType.STATUS, emptyArgs,
emptyArgs)));
transaction(new LettuceTxStatusResult(getConnection().psetex(key, milliseconds, value)));
return;
}
this.eval(script, ReturnType.STATUS, 0);
getConnection().psetex(key, milliseconds, value);
} catch (Exception ex) {
throw convertLettuceAccessException(ex);
}
}
private byte[] createRedisScriptForPSetEx(byte[] key, long milliseconds, byte[] value) {
StringBuilder sb = new StringBuilder("return redis.call('PSETEX'");
sb.append(",'");
sb.append(new String(key));
sb.append("',");
sb.append(milliseconds);
sb.append(",'");
sb.append(new String(value));
sb.append("')");
return sb.toString().getBytes();
}
public Boolean setNX(byte[] key, byte[] value) {
try {
if (isPipelined()) {
@@ -1989,9 +2000,6 @@ public class LettuceConnection extends AbstractRedisConnection {
}
public List<byte[]> sRandMember(byte[] key, long count) {
if (count < 0) {
throw new UnsupportedOperationException("sRandMember with a negative count is not supported");
}
try {
if (isPipelined()) {
pipeline(new LettuceResult(getAsyncConnection().srandmember(key, count),
@@ -2912,18 +2920,20 @@ public class LettuceConnection extends AbstractRedisConnection {
try {
byte[][] keys = extractScriptKeys(numKeys, keysAndArgs);
byte[][] args = extractScriptArgs(numKeys, keysAndArgs);
String convertedScript = LettuceConverters.toString(script);
if (isPipelined()) {
pipeline(new LettuceResult(getAsyncConnection().eval(script, LettuceConverters.toScriptOutputType(returnType),
keys, args), new LettuceEvalResultsConverter<T>(returnType)));
pipeline(new LettuceResult(getAsyncConnection().eval(convertedScript,
LettuceConverters.toScriptOutputType(returnType), keys, args), new LettuceEvalResultsConverter<T>(
returnType)));
return null;
}
if (isQueueing()) {
transaction(new LettuceTxResult(getConnection().eval(script, LettuceConverters.toScriptOutputType(returnType),
keys, args), new LettuceEvalResultsConverter<T>(returnType)));
transaction(new LettuceTxResult(getConnection().eval(convertedScript,
LettuceConverters.toScriptOutputType(returnType), keys, args), new LettuceEvalResultsConverter<T>(
returnType)));
return null;
}
return new LettuceEvalResultsConverter<T>(returnType).convert(getConnection().eval(script,
return new LettuceEvalResultsConverter<T>(returnType).convert(getConnection().eval(convertedScript,
LettuceConverters.toScriptOutputType(returnType), keys, args));
} catch (Exception ex) {
throw convertLettuceAccessException(ex);
@@ -3024,19 +3034,19 @@ public class LettuceConnection extends AbstractRedisConnection {
*/
@Override
public Long time() {
try {
/*
* We have to emulate the time command via eval script since the current driver version doesn't
* support the time command natively.
* see https://github.com/wg/lettuce/issues/19
*/
List<byte[]> result = (List<byte[]>) eval("return redis.call('TIME')".getBytes(), ReturnType.MULTI, 0);
List<byte[]> result = getConnection().time();
Assert.notEmpty(result, "Received invalid result from server. Expected 2 items in collection.");
Assert.isTrue(result.size() == 2, "Received invalid nr of arguments from redis server. Expected 2 received "
+ result.size());
Assert.notEmpty(result, "Received invalid result from server. Expected 2 items in collection.");
Assert.isTrue(result.size() == 2, "Received invalid nr of arguments from redis server. Expected 2 received "
+ result.size());
return Converters.toTimeMillis(new String(result.get(0)), new String(result.get(1)));
return Converters.toTimeMillis(new String(result.get(0)), new String(result.get(1)));
} catch (Exception ex) {
throw convertLettuceAccessException(ex);
}
}
@Override
@@ -3188,13 +3198,25 @@ public class LettuceConnection extends AbstractRedisConnection {
throw new UnsupportedOperationException("'SCAN' cannot be called in pipeline / transaction mode.");
}
String params = " ," + cursorId + options.toOptionString();
String script = "return redis.call('SCAN'" + params + ")";
com.lambdaworks.redis.ScanCursor scanCursor = getScanCursor(cursorId);
ScanArgs scanArgs = getScanArgs(options);
List<?> result = eval(script.getBytes(), ReturnType.MULTI, 0);
String nextCursorId = LettuceConverters.bytesToString().convert((byte[]) result.get(0));
if (isPipelined()) {
pipeline(new LettuceResult(getAsyncConnection().scan(scanCursor, scanArgs)));
return null;
}
return new ScanIteration<byte[]>(Long.valueOf(nextCursorId), ((List<byte[]>) result.get(1)));
if (isQueueing()) {
transaction(new LettuceTxResult(getAsyncConnection().scan(scanCursor, scanArgs)));
return null;
}
KeyScanCursor<byte[]> keyScanCursor = getConnection().scan(scanCursor, scanArgs);
String nextCursorId = keyScanCursor.getCursor();
List<byte[]> keys = keyScanCursor.getKeys();
return new ScanIteration<byte[]>(Long.valueOf(nextCursorId), (keys));
}
}.open();
@@ -3226,51 +3248,19 @@ public class LettuceConnection extends AbstractRedisConnection {
throw new UnsupportedOperationException("'HSCAN' cannot be called in pipeline / transaction mode.");
}
byte[] script = createBinaryLuaScriptForScan("HSCAN", key, cursorId, options);
List<?> result = eval(script, ReturnType.MULTI, 0);
String nextCursorId = LettuceConverters.bytesToString().convert((byte[]) result.get(0));
com.lambdaworks.redis.ScanCursor scanCursor = getScanCursor(cursorId);
ScanArgs scanArgs = getScanArgs(options);
Map<byte[], byte[]> values = failsafeReadScanValues(result, LettuceConverters.bytesListToMapConverter());
MapScanCursor<byte[], byte[]> mapScanCursor = getConnection().hscan(key, scanCursor, scanArgs);
String nextCursorId = mapScanCursor.getCursor();
Map<byte[], byte[]> values = mapScanCursor.getMap();
return new ScanIteration<Entry<byte[], byte[]>>(Long.valueOf(nextCursorId), values.entrySet());
}
}.open();
}
private byte[] createBinaryLuaScriptForScan(String command, byte[] key, long cursorId, ScanOptions options) {
ByteArrayOutputStream outputStream = new ByteArrayOutputStream();
try {
outputStream.write(("return redis.call('" + command + "'").getBytes("UTF-8"));
outputStream.write(", '".getBytes("UTF-8"));
writeFilteredKey(key, outputStream);
outputStream.write("', ".getBytes("UTF-8"));
outputStream.write(Long.toString(cursorId).getBytes("UTF-8"));
outputStream.write(options.toOptionString().getBytes("UTF-8"));
outputStream.write(")".getBytes("UTF-8"));
} catch (IOException e) {
e.printStackTrace();
}
return outputStream.toByteArray();
}
private void writeFilteredKey(byte[] key, OutputStream stream) {
byte toBeFiltered = (byte) '\r';
for (byte b : key) {
try {
if (toBeFiltered == b) {
stream.write("\\r".getBytes());
} else {
stream.write(b);
}
} catch (IOException e) {
throw new IllegalArgumentException("Cannot convert key to suitable redis key for lua", e);
}
}
}
/*
* (non-Javadoc)
* @see org.springframework.data.redis.connection.RedisSetCommands#sScan(byte[], org.springframework.data.redis.core.ScanOptions)
@@ -3298,11 +3288,13 @@ public class LettuceConnection extends AbstractRedisConnection {
throw new UnsupportedOperationException("'SSCAN' cannot be called in pipeline / transaction mode.");
}
byte[] script = createBinaryLuaScriptForScan("SSCAN", key, cursorId, options);
List<?> result = eval(script, ReturnType.MULTI, 0);
String nextCursorId = LettuceConverters.bytesToString().convert((byte[]) result.get(0));
com.lambdaworks.redis.ScanCursor scanCursor = getScanCursor(cursorId);
ScanArgs scanArgs = getScanArgs(options);
List<byte[]> values = failsafeReadScanValues(result, null);
ValueScanCursor<byte[]> valueScanCursor = getConnection().sscan(key, scanCursor, scanArgs);
String nextCursorId = valueScanCursor.getCursor();
List<byte[]> values = failsafeReadScanValues(valueScanCursor.getValues(), null);
return new ScanIteration<byte[]>(Long.valueOf(nextCursorId), values);
}
}.open();
@@ -3335,12 +3327,15 @@ public class LettuceConnection extends AbstractRedisConnection {
throw new UnsupportedOperationException("'ZSCAN' cannot be called in pipeline / transaction mode.");
}
byte[] script = createBinaryLuaScriptForScan("ZSCAN", key, cursorId, options);
List<?> result = eval(script, ReturnType.MULTI, 0);
com.lambdaworks.redis.ScanCursor scanCursor = getScanCursor(cursorId);
ScanArgs scanArgs = getScanArgs(options);
String nextCursorId = LettuceConverters.bytesToString().convert((byte[]) result.get(0));
ScoredValueScanCursor<byte[]> scoredValueScanCursor = getConnection().zscan(key, scanCursor, scanArgs);
String nextCursorId = scoredValueScanCursor.getCursor();
List<Tuple> values = failsafeReadScanValues(result, LettuceConverters.bytesListToTupleListConverter());
List<ScoredValue<byte[]>> result = scoredValueScanCursor.getValues();
List<Tuple> values = failsafeReadScanValues(result, LettuceConverters.scoredValuesToTupleList());
return new ScanIteration<Tuple>(Long.valueOf(nextCursorId), values);
}
}.open();
@@ -3350,7 +3345,7 @@ public class LettuceConnection extends AbstractRedisConnection {
private <T> T failsafeReadScanValues(List<?> source, @SuppressWarnings("rawtypes") Converter converter) {
try {
return (T) (converter != null ? converter.convert(source.get(1)) : source.get(1));
return (T) (converter != null ? converter.convert(source) : source);
} catch (IndexOutOfBoundsException e) {
// ignore this one
}
@@ -3425,11 +3420,21 @@ public class LettuceConnection extends AbstractRedisConnection {
private com.lambdaworks.redis.RedisConnection<byte[], byte[]> getDedicatedConnection() {
if (dedicatedConn == null) {
this.dedicatedConn = new com.lambdaworks.redis.RedisConnection<byte[], byte[]>(getAsyncDedicatedConnection());
this.dedicatedConn = syncConnection(getAsyncDedicatedConnection());
}
return dedicatedConn;
}
private com.lambdaworks.redis.RedisConnection<byte[], byte[]> syncConnection(
RedisAsyncConnection<byte[], byte[]> asyncConnection) {
try {
return (com.lambdaworks.redis.RedisConnection<byte[], byte[]>) SYNC_HANDLER.invoke(null, asyncConnection,
new Class[] { com.lambdaworks.redis.RedisConnection.class, RedisClusterConnection.class });
} catch (Exception ex) {
throw convertLettuceAccessException(ex);
}
}
private Future<Long> asyncBitOp(BitOperation op, byte[] destination, byte[]... keys) {
switch (op) {
case AND:
@@ -3480,6 +3485,23 @@ public class LettuceConnection extends AbstractRedisConnection {
return new byte[0][0];
}
private com.lambdaworks.redis.ScanCursor getScanCursor(long cursorId) {
return com.lambdaworks.redis.ScanCursor.of(Long.toString(cursorId));
}
private ScanArgs getScanArgs(ScanOptions options) {
if (options == null) {
return null;
}
ScanArgs scanArgs = ScanArgs.Builder.matches(options.getPattern());
if (options.getCount() != null) {
scanArgs.limit(options.getCount());
}
return scanArgs;
}
private ZStoreArgs zStoreArgs(Aggregate aggregate, int[] weights) {
ZStoreArgs args = new ZStoreArgs();
if (aggregate != null) {
@@ -3522,6 +3544,36 @@ public class LettuceConnection extends AbstractRedisConnection {
}
}
@Override
protected boolean isActive(RedisNode node) {
if (node == null) {
return false;
}
RedisConnection<String, String> connection = null;
try {
connection = client.connect(getRedisURI(node));
return connection.ping().equalsIgnoreCase("pong");
} catch (Exception e) {
return false;
} finally {
if (connection != null) {
connection.close();
}
}
}
private RedisURI getRedisURI(RedisNode node) {
return RedisURI.Builder.redis(node.getHost(), node.getPort()).build();
}
@Override
protected RedisSentinelConnection getSentinelConnection(RedisNode sentinel) {
RedisSentinelAsyncConnection<String, String> connection = client.connectSentinelAsync(getRedisURI(sentinel));
return new LettuceSentinelConnection(connection);
}
/**
* {@link TypeHints} provide {@link CommandOutput} information for a given {@link CommandType}.
*
@@ -3577,6 +3629,9 @@ public class LettuceConnection extends AbstractRedisConnection {
COMMAND_OUTPUT_TYPE_MAPPING.put(ZREMRANGEBYSCORE, IntegerOutput.class);
COMMAND_OUTPUT_TYPE_MAPPING.put(ZREVRANK, IntegerOutput.class);
COMMAND_OUTPUT_TYPE_MAPPING.put(ZUNIONSTORE, IntegerOutput.class);
COMMAND_OUTPUT_TYPE_MAPPING.put(PFCOUNT, IntegerOutput.class);
COMMAND_OUTPUT_TYPE_MAPPING.put(PFMERGE, IntegerOutput.class);
COMMAND_OUTPUT_TYPE_MAPPING.put(PFADD, IntegerOutput.class);
// DOUBLE
COMMAND_OUTPUT_TYPE_MAPPING.put(HINCRBYFLOAT, DoubleOutput.class);
@@ -3766,7 +3821,37 @@ public class LettuceConnection extends AbstractRedisConnection {
*/
@Override
public Long pfAdd(byte[] key, byte[]... values) {
throw new UnsupportedOperationException();
Assert.notEmpty(values, "PFADD requires at least one non 'null' value.");
Assert.noNullElements(values, "Values for PFADD must not contain 'null'.");
try {
if (isPipelined()) {
if (values.length == 1) {
pipeline(new LettuceResult(getAsyncConnection().pfadd(key, values[0])));
} else {
pipeline(new LettuceResult(getAsyncConnection().pfadd(key, values[0], LettuceConverters.subarray(values, 1))));
}
return null;
}
if (isQueueing()) {
if (values.length == 1) {
transaction(new LettuceTxResult(getConnection().pfadd(key, values[0])));
} else {
transaction(new LettuceTxResult(getConnection().pfadd(key, values[0], LettuceConverters.subarray(values, 1))));
}
return null;
}
if (values.length == 1) {
return getConnection().pfadd(key, values[0]);
}
return getConnection().pfadd(key, values[0], LettuceConverters.subarray(values, 1));
} catch (Exception ex) {
throw convertLettuceAccessException(ex);
}
}
/*
@@ -3775,7 +3860,36 @@ public class LettuceConnection extends AbstractRedisConnection {
*/
@Override
public Long pfCount(byte[]... keys) {
throw new UnsupportedOperationException();
Assert.notEmpty(keys, "PFCOUNT requires at least one non 'null' key.");
Assert.noNullElements(keys, "Keys for PFCOUNT must not contain 'null'.");
try {
if (isPipelined()) {
if (keys.length == 1) {
pipeline(new LettuceResult(getAsyncConnection().pfcount(keys[0])));
} else {
pipeline(new LettuceResult(getAsyncConnection().pfcount(keys[0], LettuceConverters.subarray(keys, 1))));
}
return null;
}
if (isQueueing()) {
if (keys.length == 1) {
transaction(new LettuceTxResult(getConnection().pfcount(keys[0])));
} else {
transaction(new LettuceTxResult(getConnection().pfcount(keys[0], LettuceConverters.subarray(keys, 1))));
}
return null;
}
if (keys.length == 1) {
return getConnection().pfcount(keys[0]);
}
return getConnection().pfcount(keys[0], LettuceConverters.subarray(keys, 1));
} catch (Exception ex) {
throw convertLettuceAccessException(ex);
}
}
/*
@@ -3784,7 +3898,36 @@ public class LettuceConnection extends AbstractRedisConnection {
*/
@Override
public void pfMerge(byte[] destinationKey, byte[]... sourceKeys) {
throw new UnsupportedOperationException();
Assert.notEmpty(sourceKeys, "PFMERGE requires at least one non 'null' source key.");
Assert.noNullElements(sourceKeys, "source key for PFMERGE must not contain 'null'.");
try {
if (isPipelined()) {
if (sourceKeys.length == 1) {
pipeline(new LettuceResult(getAsyncConnection().pfmerge(destinationKey, sourceKeys[0])));
} else {
pipeline(new LettuceResult(getAsyncConnection().pfmerge(destinationKey, sourceKeys[0],
LettuceConverters.subarray(sourceKeys, 1))));
}
}
if (isQueueing()) {
if (sourceKeys.length == 1) {
transaction(new LettuceTxResult(getConnection().pfmerge(destinationKey, sourceKeys[0])));
} else {
transaction(new LettuceTxResult(getConnection().pfmerge(destinationKey, sourceKeys[0],
LettuceConverters.subarray(sourceKeys, 1))));
}
}
if (sourceKeys.length == 1) {
getConnection().pfmerge(destinationKey, sourceKeys[0]);
} else {
getConnection().pfmerge(destinationKey, sourceKeys[0], LettuceConverters.subarray(sourceKeys, 1));
}
} catch (Exception ex) {
throw convertLettuceAccessException(ex);
}
}
/*
@@ -3793,7 +3936,7 @@ public class LettuceConnection extends AbstractRedisConnection {
*/
@Override
public Set<byte[]> zRangeByLex(byte[] key) {
throw new UnsupportedOperationException("ZRANGEBYLEX is no supported for lettuce.");
return zRangeByLex(key, Range.unbounded());
}
/*
@@ -3802,7 +3945,7 @@ public class LettuceConnection extends AbstractRedisConnection {
*/
@Override
public Set<byte[]> zRangeByLex(byte[] key, Range range) {
throw new UnsupportedOperationException("ZRANGEBYLEX is no supported for lettuce.");
return zRangeByLex(key, range, null);
}
/*
@@ -3811,7 +3954,43 @@ public class LettuceConnection extends AbstractRedisConnection {
*/
@Override
public Set<byte[]> zRangeByLex(byte[] key, Range range, Limit limit) {
throw new UnsupportedOperationException("ZRANGEBYLEX is no supported for lettuce.");
Assert.notNull(range, "Range cannot be null for ZRANGEBYLEX.");
String min = LettuceConverters.boundaryToBytesForZRangeByLex(range.getMin(), LettuceConverters.MINUS_BYTES);
String max = LettuceConverters.boundaryToBytesForZRangeByLex(range.getMax(), LettuceConverters.PLUS_BYTES);
try {
if (isPipelined()) {
if (limit != null) {
pipeline(new LettuceResult(getAsyncConnection().zrangebylex(key, min, max, limit.getOffset(),
limit.getCount()), LettuceConverters.bytesListToBytesSet()));
} else {
pipeline(new LettuceResult(getAsyncConnection().zrangebylex(key, min, max),
LettuceConverters.bytesListToBytesSet()));
}
return null;
}
if (isQueueing()) {
if (limit != null) {
transaction(new LettuceTxResult(getConnection().zrangebylex(key, min, max, limit.getOffset(),
limit.getCount()), LettuceConverters.bytesListToBytesSet()));
} else {
transaction(new LettuceTxResult(getConnection().zrangebylex(key, min, max),
LettuceConverters.bytesListToBytesSet()));
}
return null;
}
if (limit != null) {
return LettuceConverters.bytesListToBytesSet().convert(
getConnection().zrangebylex(key, min, max, limit.getOffset(), limit.getCount()));
}
return LettuceConverters.bytesListToBytesSet().convert(getConnection().zrangebylex(key, min, max));
} catch (Exception ex) {
throw convertLettuceAccessException(ex);
}
}
}

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

@@ -18,6 +18,9 @@ package org.springframework.data.redis.connection.lettuce;
import java.util.concurrent.TimeUnit;
import com.lambdaworks.redis.LettuceFutures;
import com.lambdaworks.redis.RedisFuture;
import com.lambdaworks.redis.RedisURI;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.beans.factory.DisposableBean;
@@ -29,6 +32,7 @@ import org.springframework.data.redis.RedisConnectionFailureException;
import org.springframework.data.redis.connection.Pool;
import org.springframework.data.redis.connection.RedisConnection;
import org.springframework.data.redis.connection.RedisConnectionFactory;
import org.springframework.data.redis.connection.RedisSentinelConfiguration;
import org.springframework.data.redis.connection.RedisSentinelConnection;
import org.springframework.util.Assert;
@@ -37,7 +41,7 @@ import com.lambdaworks.redis.RedisClient;
import com.lambdaworks.redis.RedisException;
/**
* Connection factory creating <a href="http://github.com/wg/lettuce">Lettuce</a>-based connections.
* Connection factory creating <a href="http://github.com/mp911de/lettuce">Lettuce</a>-based connections.
* <p>
* This factory creates a new {@link LettuceConnection} on each call to {@link #getConnection()}. Multiple
* {@link LettuceConnection}s share a single thread-safe native connection by default.
@@ -54,6 +58,8 @@ import com.lambdaworks.redis.RedisException;
*/
public class LettuceConnectionFactory implements InitializingBean, DisposableBean, RedisConnectionFactory {
public static final String PING_REPLY = "PONG";
private static final ExceptionTranslationStrategy EXCEPTION_TRANSLATION = new PassThroughExceptionTranslationStrategy(
LettuceConverters.exceptionConverter());
@@ -63,6 +69,7 @@ public class LettuceConnectionFactory implements InitializingBean, DisposableBea
private int port = 6379;
private RedisClient client;
private long timeout = TimeUnit.MILLISECONDS.convert(60, TimeUnit.SECONDS);
private long shutdownTimeout = TimeUnit.MILLISECONDS.convert(2, TimeUnit.SECONDS);
private boolean validateConnection = false;
private boolean shareNativeConnection = true;
private RedisAsyncConnection<byte[], byte[]> connection;
@@ -72,6 +79,7 @@ public class LettuceConnectionFactory implements InitializingBean, DisposableBea
private final Object connectionMonitor = new Object();
private String password;
private boolean convertPipelineAndTxResults = true;
private RedisSentinelConfiguration sentinelConfiguration;
/**
* Constructs a new <code>LettuceConnectionFactory</code> instance with default settings.
@@ -86,6 +94,16 @@ public class LettuceConnectionFactory implements InitializingBean, DisposableBea
this.port = port;
}
/**
* Constructs a new {@link LettuceConnectionFactory} instance using the given {@link RedisSentinelConfiguration}
*
* @param sentinelConfiguration
* @since 1.5
*/
public LettuceConnectionFactory(RedisSentinelConfiguration sentinelConfiguration) {
this.sentinelConfiguration = sentinelConfiguration;
}
public LettuceConnectionFactory(LettucePool pool) {
this.pool = pool;
}
@@ -96,7 +114,7 @@ public class LettuceConnectionFactory implements InitializingBean, DisposableBea
public void destroy() {
resetConnection();
client.shutdown();
client.shutdown(shutdownTimeout, shutdownTimeout, TimeUnit.MILLISECONDS);
}
public RedisConnection getConnection() {
@@ -131,9 +149,22 @@ public class LettuceConnectionFactory implements InitializingBean, DisposableBea
*/
public void validateConnection() {
synchronized (this.connectionMonitor) {
try {
new com.lambdaworks.redis.RedisConnection<byte[], byte[]>(connection).ping();
} catch (RedisException e) {
boolean valid = false;
if (connection.isOpen()) {
try {
RedisFuture<String> ping = connection.ping();
LettuceFutures.awaitAll(timeout, TimeUnit.MILLISECONDS, ping);
if(PING_REPLY.equalsIgnoreCase(ping.get())) {
valid = true;
}
} catch (Exception e) {
log.debug("Validation failed", e);
}
}
if (!valid) {
log.warn("Validation of shared connection failed. Creating a new connection.");
initConnection();
}
@@ -280,6 +311,22 @@ public class LettuceConnectionFactory implements InitializingBean, DisposableBea
this.password = password;
}
/**
* Returns the shutdown timeout for shutting down the RedisClient (in milliseconds).
* @return shutdown timeout
*/
public long getShutdownTimeout() {
return shutdownTimeout;
}
/**
* Sets the shutdown timeout for shutting down the RedisClient (in milliseconds).
* @param shutdownTimeout the shutdown timeout
*/
public void setShutdownTimeout(long shutdownTimeout) {
this.shutdownTimeout = shutdownTimeout;
}
/**
* Specifies if pipelined results should be converted to the expected data type. If false, results of
* {@link LettuceConnection#closePipeline()} and {LettuceConnection#exec()} will be of the type returned by the
@@ -331,17 +378,39 @@ public class LettuceConnectionFactory implements InitializingBean, DisposableBea
}
private RedisClient createRedisClient() {
if(isRedisSentinelAware()) {
RedisURI redisURI = getSentinelRedisURI();
return new RedisClient(redisURI);
}
if (pool != null) {
return pool.getClient();
}
RedisClient client = password != null ? new AuthenticatingRedisClient(hostName, port, password) : new RedisClient(
hostName, port);
client.setDefaultTimeout(timeout, TimeUnit.MILLISECONDS);
RedisURI.Builder builder = RedisURI.Builder.redis(hostName, port);
if(password != null) {
builder.withPassword(password);
}
builder.withTimeout(timeout, TimeUnit.MILLISECONDS);
RedisClient client = new RedisClient(builder.build());
return client;
}
private RedisURI getSentinelRedisURI() {
return LettuceConverters.sentinelConfigurationToRedisURI(sentinelConfiguration);
}
/**
* @return true when {@link RedisSentinelConfiguration} is present.
* @since 1.5
*/
public boolean isRedisSentinelAware() {
return sentinelConfiguration != null;
}
@Override
public RedisSentinelConnection getSentinelConnection() {
throw new UnsupportedOperationException();
return new LettuceSentinelConnection(client.connectSentinelAsync());
}
}

187
src/main/java/org/springframework/data/redis/connection/lettuce/LettuceConverters.java
View File

@@ -15,6 +15,7 @@
*/
package org.springframework.data.redis.connection.lettuce;
import java.nio.ByteBuffer;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
@@ -30,6 +31,9 @@ import org.springframework.core.convert.converter.Converter;
import org.springframework.dao.DataAccessException;
import org.springframework.data.redis.connection.DefaultTuple;
import org.springframework.data.redis.connection.RedisListCommands.Position;
import org.springframework.data.redis.connection.RedisNode;
import org.springframework.data.redis.connection.RedisSentinelConfiguration;
import org.springframework.data.redis.connection.RedisServer;
import org.springframework.data.redis.connection.RedisZSetCommands.Range.Boundary;
import org.springframework.data.redis.connection.RedisZSetCommands.Tuple;
import org.springframework.data.redis.connection.ReturnType;
@@ -44,10 +48,11 @@ import org.springframework.util.CollectionUtils;
import org.springframework.util.StringUtils;
import com.lambdaworks.redis.KeyValue;
import com.lambdaworks.redis.RedisURI;
import com.lambdaworks.redis.ScoredValue;
import com.lambdaworks.redis.ScriptOutputType;
import com.lambdaworks.redis.SortArgs;
import com.lambdaworks.redis.protocol.Charsets;
import com.lambdaworks.redis.protocol.LettuceCharsets;
/**
* Lettuce type converters
@@ -55,35 +60,40 @@ import com.lambdaworks.redis.protocol.Charsets;
* @author Jennifer Hickey
* @author Christoph Strobl
* @author Thomas Darimont
* @author Mark Paluch
*/
abstract public class LettuceConverters extends Converters {
private static final Converter<Date, Long> DATE_TO_LONG;
private static final Converter<List<byte[]>, Set<byte[]>> BYTES_LIST_TO_BYTES_SET;
private static final Converter<byte[], String> BYTES_TO_STRING;
private static final Converter<String, byte[]> STRING_TO_BYTES;
private static final Converter<Set<byte[]>, List<byte[]>> BYTES_SET_TO_BYTES_LIST;
private static final Converter<KeyValue<byte[], byte[]>, List<byte[]>> KEY_VALUE_TO_BYTES_LIST;
private static final Converter<List<ScoredValue<byte[]>>, Set<Tuple>> SCORED_VALUES_TO_TUPLE_SET;
private static final Converter<List<ScoredValue<byte[]>>, List<Tuple>> SCORED_VALUES_TO_TUPLE_LIST;
private static final Converter<ScoredValue<byte[]>, Tuple> SCORED_VALUE_TO_TUPLE;
private static final Converter<Exception, DataAccessException> EXCEPTION_CONVERTER = new LettuceExceptionConverter();
private static final Converter<Long, Boolean> LONG_TO_BOOLEAN = new LongToBooleanConverter();
private static final Converter<List<byte[]>, Map<byte[], byte[]>> BYTES_LIST_TO_MAP;
private static final Converter<List<byte[]>, List<Tuple>> BYTES_LIST_TO_TUPLE_LIST_CONVERTER;
private static final Converter<String[], List<RedisClientInfo>> STRING_TO_LIST_OF_CLIENT_INFO = new StringToRedisClientInfoConverter();
public static final byte[] PLUS_BYTES;
public static final byte[] MINUS_BYTES;
public static final byte[] POSITIVE_INFINITY_BYTES;
public static final byte[] NEGATIVE_INFINITY_BYTES;
static {
DATE_TO_LONG = new Converter<Date, Long>() {
public Long convert(Date source) {
return source != null ? source.getTime() : null;
}
};
BYTES_LIST_TO_BYTES_SET = new Converter<List<byte[]>, Set<byte[]>>() {
public Set<byte[]> convert(List<byte[]> results) {
return results != null ? new LinkedHashSet<byte[]>(results) : null;
}
};
BYTES_TO_STRING = new Converter<byte[], String>() {
@@ -94,7 +104,16 @@ abstract public class LettuceConverters extends Converters {
}
return new String(source);
}
};
STRING_TO_BYTES = new Converter<String, byte[]>() {
@Override
public byte[] convert(String source) {
if (source == null) {
return null;
}
return source.getBytes();
}
};
BYTES_SET_TO_BYTES_LIST = new Converter<Set<byte[]>, List<byte[]>>() {
public List<byte[]> convert(Set<byte[]> results) {
@@ -130,7 +149,6 @@ abstract public class LettuceConverters extends Converters {
return target;
}
};
SCORED_VALUES_TO_TUPLE_SET = new Converter<List<ScoredValue<byte[]>>, Set<Tuple>>() {
public Set<Tuple> convert(List<ScoredValue<byte[]>> source) {
@@ -144,11 +162,23 @@ abstract public class LettuceConverters extends Converters {
return tuples;
}
};
SCORED_VALUES_TO_TUPLE_LIST = new Converter<List<ScoredValue<byte[]>>, List<Tuple>>() {
public List<Tuple> convert(List<ScoredValue<byte[]>> source) {
if (source == null) {
return null;
}
List<Tuple> tuples = new ArrayList<Tuple>(source.size());
for (ScoredValue<byte[]> value : source) {
tuples.add(LettuceConverters.toTuple(value));
}
return tuples;
}
};
SCORED_VALUE_TO_TUPLE = new Converter<ScoredValue<byte[]>, Tuple>() {
public Tuple convert(ScoredValue<byte[]> source) {
return source != null ? new DefaultTuple(source.value, Double.valueOf(source.score)) : null;
}
};
BYTES_LIST_TO_TUPLE_LIST_CONVERTER = new Converter<List<byte[]>, List<Tuple>>() {
@@ -169,6 +199,10 @@ abstract public class LettuceConverters extends Converters {
}
};
PLUS_BYTES = toBytes("+");
MINUS_BYTES = toBytes("-");
POSITIVE_INFINITY_BYTES = toBytes("+inf");
NEGATIVE_INFINITY_BYTES = toBytes("-inf");
}
public static List<Tuple> toTuple(List<byte[]> list) {
@@ -217,6 +251,10 @@ abstract public class LettuceConverters extends Converters {
return SCORED_VALUES_TO_TUPLE_SET;
}
public static Converter<List<ScoredValue<byte[]>>, List<Tuple>> scoredValuesToTupleList() {
return SCORED_VALUES_TO_TUPLE_LIST;
}
public static Converter<ScoredValue<byte[]>, Tuple> scoredValueToTuple() {
return SCORED_VALUE_TO_TUPLE;
}
@@ -301,7 +339,7 @@ abstract public class LettuceConverters extends Converters {
return args;
}
if (params.getByPattern() != null) {
args.by(new String(params.getByPattern(), Charsets.ASCII));
args.by(new String(params.getByPattern(), LettuceCharsets.ASCII));
}
if (params.getLimit() != null) {
args.limit(params.getLimit().getStart(), params.getLimit().getCount());
@@ -309,7 +347,7 @@ abstract public class LettuceConverters extends Converters {
if (params.getGetPattern() != null) {
byte[][] pattern = params.getGetPattern();
for (byte[] bs : pattern) {
args.get(new String(bs, Charsets.ASCII));
args.get(new String(bs, LettuceCharsets.ASCII));
}
}
if (params.getOrder() != null) {
@@ -330,6 +368,17 @@ abstract public class LettuceConverters extends Converters {
return stringToRedisClientListConverter().convert(clientList);
}
public static byte[][] subarray(byte[][] input, int index) {
if (input.length > index) {
byte[][] output = new byte[input.length - index][];
System.arraycopy(input, index, output, 0, output.length);
return output;
}
return null;
}
public static String boundaryToStringForZRange(Boundary boundary, String defaultValue) {
if (boundary == null || boundary.getValue() == null) {
@@ -352,4 +401,126 @@ abstract public class LettuceConverters extends Converters {
return prefix + value;
}
/**
* @param source List of Maps containing node details from SENTINEL SLAVES or SENTINEL MASTERS. May be empty or
* {@literal null}.
* @return List of {@link RedisServer}'s. List is empty if List of Maps is empty.
* @since 1.5
*/
public static List<RedisServer> toListOfRedisServer(List<Map<String, String>> source) {
if (CollectionUtils.isEmpty(source)) {
return Collections.emptyList();
}
List<RedisServer> sentinels = new ArrayList<RedisServer>();
for (Map<String, String> info : source) {
sentinels.add(RedisServer.newServerFrom(Converters.toProperties(info)));
}
return sentinels;
}
/**
* @param sentinelConfiguration the sentinel configuration containing one or more sentinels and a master name. Must
* not be {@literal null}
* @return A {@link RedisURI} containing Redis Sentinel addresses of {@link RedisSentinelConfiguration}
* @since 1.5
*/
public static RedisURI sentinelConfigurationToRedisURI(RedisSentinelConfiguration sentinelConfiguration) {
Assert.notNull(sentinelConfiguration, "RedisSentinelConfiguration is required");
Set<RedisNode> sentinels = sentinelConfiguration.getSentinels();
RedisURI.Builder builder = null;
for (RedisNode sentinel : sentinels) {
if (builder == null) {
builder = RedisURI.Builder.sentinel(sentinel.getHost(), sentinel.getPort(), sentinelConfiguration.getMaster()
.getName());
} else {
builder.withSentinel(sentinel.getHost(), sentinel.getPort());
}
}
return builder.build();
}
public static byte[] toBytes(String source) {
return STRING_TO_BYTES.convert(source);
}
public static byte[] toBytes(Integer source) {
return String.valueOf(source).getBytes();
}
public static byte[] toBytes(Long source) {
return String.valueOf(source).getBytes();
}
/**
* @param source
* @return
* @since 1.6
*/
public static byte[] toBytes(Double source) {
return toBytes(String.valueOf(source));
}
/**
* Converts a given {@link Boundary} to its binary representation suitable for {@literal ZRANGEBY*} commands, despite
* {@literal ZRANGEBYLEX}.
*
* @param boundary
* @param defaultValue
* @return
* @since 1.6
*/
public static String boundaryToBytesForZRange(Boundary boundary, byte[] defaultValue) {
if (boundary == null || boundary.getValue() == null) {
return toString(defaultValue);
}
return boundaryToBytes(boundary, new byte[] {}, toBytes("("));
}
/**
* Converts a given {@link Boundary} to its binary representation suitable for ZRANGEBYLEX command.
*
* @param boundary
* @return
* @since 1.6
*/
public static String boundaryToBytesForZRangeByLex(Boundary boundary, byte[] defaultValue) {
if (boundary == null || boundary.getValue() == null) {
return toString(defaultValue);
}
return boundaryToBytes(boundary, toBytes("["), toBytes("("));
}
private static String boundaryToBytes(Boundary boundary, byte[] inclPrefix, byte[] exclPrefix) {
byte[] prefix = boundary.isIncluding() ? inclPrefix : exclPrefix;
byte[] value = null;
if (boundary.getValue() instanceof byte[]) {
value = (byte[]) boundary.getValue();
} else if (boundary.getValue() instanceof Double) {
value = toBytes((Double) boundary.getValue());
} else if (boundary.getValue() instanceof Long) {
value = toBytes((Long) boundary.getValue());
} else if (boundary.getValue() instanceof Integer) {
value = toBytes((Integer) boundary.getValue());
} else if (boundary.getValue() instanceof String) {
value = toBytes((String) boundary.getValue());
} else {
throw new IllegalArgumentException(String.format("Cannot convert %s to binary format", boundary.getValue()));
}
ByteBuffer buffer = ByteBuffer.allocate(prefix.length + value.length);
buffer.put(prefix);
buffer.put(value);
return toString(buffer.array());
}
}

27
src/main/java/org/springframework/data/redis/connection/lettuce/LettuceExceptionConverter.java
View File

@@ -15,9 +15,14 @@
*/
package org.springframework.data.redis.connection.lettuce;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeoutException;
import org.jboss.netty.channel.ChannelException;
import com.lambdaworks.redis.RedisCommandInterruptedException;
import com.lambdaworks.redis.RedisCommandTimeoutException;
import com.lambdaworks.redis.RedisConnectionException;
import com.lambdaworks.redis.RedisException;
import io.netty.channel.ChannelException;
import org.springframework.core.convert.converter.Converter;
import org.springframework.dao.DataAccessException;
import org.springframework.dao.QueryTimeoutException;
@@ -37,6 +42,14 @@ public class LettuceExceptionConverter implements Converter<Exception, DataAcces
public DataAccessException convert(Exception ex) {
if (ex instanceof ExecutionException) {
if(ex.getCause() != ex && ex.getCause() instanceof Exception) {
return convert((Exception) ex.getCause());
}
return new RedisSystemException("Error in execution", ex);
}
if (ex instanceof DataAccessException) {
return (DataAccessException) ex;
}
@@ -45,18 +58,18 @@ public class LettuceExceptionConverter implements Converter<Exception, DataAcces
return new RedisSystemException("Redis command interrupted", ex);
}
if (ex instanceof RedisException) {
return new RedisSystemException("Redis exception", ex);
}
if (ex instanceof ChannelException) {
if (ex instanceof ChannelException || ex instanceof RedisConnectionException) {
return new RedisConnectionFailureException("Redis connection failed", ex);
}
if (ex instanceof TimeoutException) {
if (ex instanceof TimeoutException || ex instanceof RedisCommandTimeoutException) {
return new QueryTimeoutException("Redis command timed out", ex);
}
if (ex instanceof RedisException) {
return new RedisSystemException("Redis exception", ex);
}
return null;
}
}

202
src/main/java/org/springframework/data/redis/connection/lettuce/LettuceSentinelConnection.java Normal file
View File

@@ -0,0 +1,202 @@
/*
* Copyright 2014 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.data.redis.connection.lettuce;
import java.io.IOException;
import java.util.List;
import com.lambdaworks.redis.RedisClient;
import com.lambdaworks.redis.RedisSentinelAsyncConnection;
import org.springframework.data.redis.ExceptionTranslationStrategy;
import org.springframework.data.redis.FallbackExceptionTranslationStrategy;
import org.springframework.data.redis.connection.NamedNode;
import org.springframework.data.redis.connection.RedisNode;
import org.springframework.data.redis.connection.RedisSentinelConnection;
import org.springframework.data.redis.connection.RedisServer;
import org.springframework.util.Assert;
/**
* @author Christoph Strobl
* @author Mark Paluch
* @since 1.5
*/
public class LettuceSentinelConnection implements RedisSentinelConnection {
private static final ExceptionTranslationStrategy EXCEPTION_TRANSLATION = new FallbackExceptionTranslationStrategy(
LettuceConverters.exceptionConverter());
private RedisClient redisClient;
private RedisSentinelAsyncConnection<String, String> connection;
/**
* Creates a {@link LettuceSentinelConnection} with a dedicated client for a supplied {@link RedisNode}.
* @param sentinel The sentinel to connect to.
*/
public LettuceSentinelConnection(RedisNode sentinel) {
this(sentinel.getHost(), sentinel.getPort());
}
/**
* Creates a {@link LettuceSentinelConnection} with a client for the supplied {@code host} and {@code port}.
* @param host hostname
* @param port sentinel port
*/
public LettuceSentinelConnection(String host, int port) {
Assert.notNull(host, "Cannot created LettuceSentinelConnection using 'null' as host.");
redisClient = new RedisClient(host, port);
init();
}
/**
* Creates a {@link LettuceSentinelConnection} using a supplied {@link RedisClient}.
* @param redisClient
*/
public LettuceSentinelConnection(RedisClient redisClient) {
Assert.notNull(redisClient, "Cannot create LettuceSentinelConnection using 'null' as client.");
this.redisClient = redisClient;
init();
}
/**
* Creates a {@link LettuceSentinelConnection} using a supplied redis connection.
* @param connection native Lettuce connection, must not be {@literal null}
*/
protected LettuceSentinelConnection(RedisSentinelAsyncConnection<String, String> connection) {
Assert.notNull(connection, "Cannot create LettuceSentinelConnection using 'null' as connection.");
this.connection = connection;
}
/*
* (non-Javadoc)
* @see org.springframework.data.redis.connection.RedisSentinelCommands#failover(org.springframework.data.redis.connection.NamedNode)
*/
@Override
public void failover(NamedNode master) {
Assert.notNull(master, "Redis node master must not be 'null' for failover.");
Assert.hasText(master.getName(), "Redis master name must not be 'null' or empty for failover.");
connection.failover(master.getName());
}
/*
* (non-Javadoc)
* @see org.springframework.data.redis.connection.RedisSentinelCommands#masters()
*/
@Override
public List<RedisServer> masters() {
try {
return LettuceConverters.toListOfRedisServer(connection.masters().get());
} catch (Exception e) {
throw EXCEPTION_TRANSLATION.translate(e);
}
}
/*
* (non-Javadoc)
* @see org.springframework.data.redis.connection.RedisSentinelCommands#slaves(org.springframework.data.redis.connection.NamedNode)
*/
@Override
public List<RedisServer> slaves(NamedNode master) {
Assert.notNull(master, "Master node cannot be 'null' when loading slaves.");
return slaves(master.getName());
}
/**
* @param masterName
* @see org.springframework.data.redis.connection.RedisSentinelCommands#slaves(org.springframework.data.redis.connection.NamedNode)
* @return
*/
public List<RedisServer> slaves(String masterName) {
Assert.hasText(masterName, "Name of redis master cannot be 'null' or empty when loading slaves.");
try {
return LettuceConverters.toListOfRedisServer(connection.slaves(masterName).get());
} catch (Exception e) {
throw EXCEPTION_TRANSLATION.translate(e);
}
}
/*
* (non-Javadoc)
* @see org.springframework.data.redis.connection.RedisSentinelCommands#remove(org.springframework.data.redis.connection.NamedNode)
*/
@Override
public void remove(NamedNode master) {
Assert.notNull(master, "Master node cannot be 'null' when trying to remove.");
remove(master.getName());
}
/**
* @param masterName
* @see org.springframework.data.redis.connection.RedisSentinelCommands#remove(org.springframework.data.redis.connection.NamedNode)
*/
public void remove(String masterName) {
Assert.hasText(masterName, "Name of redis master cannot be 'null' or empty when trying to remove.");
connection.remove(masterName);
}
/*
* (non-Javadoc)
* @see org.springframework.data.redis.connection.RedisSentinelCommands#monitor(org.springframework.data.redis.connection.RedisServer)
*/
@Override
public void monitor(RedisServer server) {
Assert.notNull(server, "Cannot monitor 'null' server.");
Assert.hasText(server.getName(), "Name of server to monitor must not be 'null' or empty.");
Assert.hasText(server.getHost(), "Host must not be 'null' for server to monitor.");
Assert.notNull(server.getPort(), "Port must not be 'null' for server to monitor.");
Assert.notNull(server.getQuorum(), "Quorum must not be 'null' for server to monitor.");
connection.monitor(server.getName(), server.getHost(), server.getPort().intValue(), server.getQuorum()
.intValue());
}
/*
* (non-Javadoc)
* @see java.io.Closeable#close()
*/
@Override
public void close() throws IOException {
connection.close();
connection = null;
if(redisClient != null) {
redisClient.shutdown();
}
}
private void init() {
if (connection == null) {
connection = connectSentinel();
}
}
private RedisSentinelAsyncConnection<String, String> connectSentinel() {
return redisClient.connectSentinelAsync();
}
@Override
public boolean isOpen() {
return connection != null && connection.isOpen();
}
}

6
src/main/java/org/springframework/data/redis/connection/lettuce/LettuceUtils.java
View File

@@ -24,6 +24,7 @@ import java.util.List;
import java.util.Properties;
import java.util.Set;
import com.lambdaworks.redis.protocol.LettuceCharsets;
import org.springframework.dao.DataAccessException;
import org.springframework.data.redis.RedisSystemException;
import org.springframework.data.redis.connection.DefaultTuple;
@@ -43,7 +44,6 @@ import com.lambdaworks.redis.ScriptOutputType;
import com.lambdaworks.redis.SortArgs;
import com.lambdaworks.redis.ZStoreArgs;
import com.lambdaworks.redis.codec.RedisCodec;
import com.lambdaworks.redis.protocol.Charsets;
/**
* Helper class featuring methods for Lettuce connection handling, providing support for exception translation.
@@ -111,7 +111,7 @@ abstract class LettuceUtils {
}
if (params.getByPattern() != null) {
args.by(new String(params.getByPattern(), Charsets.ASCII));
args.by(new String(params.getByPattern(), LettuceCharsets.ASCII));
}
if (params.getLimit() != null) {
@@ -121,7 +121,7 @@ abstract class LettuceUtils {
if (params.getGetPattern() != null) {
byte[][] pattern = params.getGetPattern();
for (byte[] bs : pattern) {
args.get(new String(bs, Charsets.ASCII));
args.get(new String(bs, LettuceCharsets.ASCII));
}
}

2
src/main/java/org/springframework/data/redis/connection/lettuce/package-info.java
View File

@@ -1,5 +1,5 @@
/**
* Connection package for <a href="https://github.com/wg/lettuce">Lettuce</a> Redis client.
* Connection package for <a href="https://github.com/mp911de/lettuce">Lettuce</a> Redis client.
*/
package org.springframework.data.redis.connection.lettuce;