From 7426b276888a23bf2c93509ca682c5b947d8ed59 Mon Sep 17 00:00:00 2001 From: Mark Paluch Date: Fri, 12 Sep 2014 22:28:24 +0200 Subject: [PATCH] DATAREDIS-348 - Switch to mp911de/lettuce 3.0. 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. --- build.gradle | 3 +- gradle.properties | 2 +- src/asciidoc/reference/redis.adoc | 16 +- .../connection/RedisConnectionFactory.java | 3 +- .../lettuce/AuthenticatingRedisClient.java | 2 + .../lettuce/DefaultLettucePool.java | 51 ++- .../connection/lettuce/LettuceConnection.java | 427 +++++++++++++----- .../lettuce/LettuceConnectionFactory.java | 87 +++- .../connection/lettuce/LettuceConverters.java | 187 +++++++- .../lettuce/LettuceExceptionConverter.java | 27 +- .../lettuce/LettuceSentinelConnection.java | 202 +++++++++ .../connection/lettuce/LettuceUtils.java | 6 +- .../connection/lettuce/package-info.java | 2 +- .../AbstractConnectionIntegrationTests.java | 20 +- .../LettuceConnectionFactoryTests.java | 5 + .../LettuceConnectionIntegrationTests.java | 33 +- ...uceConnectionPipelineIntegrationTests.java | 6 - ...ConnectionTransactionIntegrationTests.java | 7 - .../LettuceConnectionUnitTestSuite.java | 22 +- .../LettuceSentinelConnectionUnitTests.java | 217 +++++++++ .../LettuceSentinelIntegrationTests.java | 105 +++++ .../test/util/MinimumRedisVersionRule.java | 3 +- template.mf | 4 +- 23 files changed, 1237 insertions(+), 200 deletions(-) create mode 100644 src/main/java/org/springframework/data/redis/connection/lettuce/LettuceSentinelConnection.java create mode 100644 src/test/java/org/springframework/data/redis/connection/lettuce/LettuceSentinelConnectionUnitTests.java create mode 100644 src/test/java/org/springframework/data/redis/connection/lettuce/LettuceSentinelIntegrationTests.java diff --git a/build.gradle b/build.gradle index d04295178..605977a9d 100644 --- a/build.gradle +++ b/build.gradle @@ -19,6 +19,7 @@ group = 'org.springframework.data' repositories { maven { url "https://repo.spring.io/libs-snapshot" } maven { url "https://repo.spring.io/plugins-release" } + maven { url "https://oss.sonatype.org/content/repositories/snapshots/" } mavenCentral() jcenter() } @@ -84,7 +85,7 @@ dependencies { compile("com.github.spullara.redis:client:$srpVersion", optional) compile("org.jredis:jredis-core-api:$jredisVersion", optional) compile("org.jredis:jredis-core-ri:$jredisVersion", optional) - compile("com.lambdaworks:lettuce:$lettuceVersion", optional) + compile("biz.paluch.redis:lettuce:$lettuceVersion", optional) // Mappers compile("org.codehaus.jackson:jackson-mapper-asl:$jacksonVersion", optional) diff --git a/gradle.properties b/gradle.properties index 1201c0b95..c30c3e40d 100644 --- a/gradle.properties +++ b/gradle.properties @@ -10,5 +10,5 @@ srpVersion=0.7 jacksonVersion=1.8.8 fasterXmlJacksonVersion=2.5.1 fasterXmlJacksonDatabindVersion=2.5.1 -lettuceVersion=2.3.3 +lettuceVersion=3.2.Final mockitoVersion=1.10.19 diff --git a/src/asciidoc/reference/redis.adoc b/src/asciidoc/reference/redis.adoc index 9497808bc..db571b7fd 100644 --- a/src/asciidoc/reference/redis.adoc +++ b/src/asciidoc/reference/redis.adoc @@ -141,7 +141,7 @@ Needless to say, the configuration is quite similar to that of the other connect [[redis:connectors:lettuce]] === Configuring Lettuce connector -https://github.com/wg/lettuce[Lettuce] is the fourth open-source connector supported by Spring Data Redis through the `org.springframework.data.redis.connection.lettuce` package. +https://github.com/mp911de/lettuce[Lettuce] is the fourth open-source connector supported by Spring Data Redis through the `org.springframework.data.redis.connection.lettuce` package. Its configuration is probably easy to guess: @@ -165,7 +165,7 @@ There are also a few Lettuce-specific connection parameters that can be tweaked. For dealing with high available Redis there is support for http://redis.io/topics/sentinel[Redis Sentinel] using `RedisSentinelConfiguration`. -NOTE: Please note that currently only http://github.com/xetorthio/jedis[Jedis] supports Redis Sentinel. +NOTE: Please note that currently only http://github.com/xetorthio/jedis[Jedis] and lettuce http://github.com/mp911de/lettuce[Lettuce] support Redis Sentinel. [source,java] ---- @@ -173,7 +173,7 @@ NOTE: Please note that currently only http://github.com/xetorthio/jedis[Jedis] s public RedisConnectionFactory jedisConnectionFactory() { RedisSentinelConfiguration sentinelConfig = new RedisSentinelConfiguration() .master("mymaster") .sentinel("127.0.0.1", 26379) .sentinel("127.0.0.1", 26380); - return new JedisConnectionFactory(sentinelConfig); + return new JedisConnectionFactory(sentinelConfig); } ---- @@ -185,6 +185,16 @@ public RedisConnectionFactory jedisConnectionFactory() { - `spring.redis.sentinel.master`: name of the master node. - `spring.redis.sentinel.nodes`: Comma delimited list of host:port pairs. ==== +======= +[source,java] +---- +@Bean +public RedisConnectionFactory lettuceConnectionFactory() { + RedisSentinelConfiguration sentinelConfig = new RedisSentinelConfiguration() .master("mymaster") + .sentinel("127.0.0.1", 26379) .sentinel("127.0.0.1", 26380); + return new LettuceConnectionFactory(sentinelConfig); +} +---- Sometimes direct interaction with the one of the Sentinels is required. Using `RedisConnectionFactory.getSentinelConnection()` or `RedisConnection.getSentinelCommands()` gives you access to the first active Sentinel configured. diff --git a/src/main/java/org/springframework/data/redis/connection/RedisConnectionFactory.java b/src/main/java/org/springframework/data/redis/connection/RedisConnectionFactory.java index 1b286a862..a123273db 100644 --- a/src/main/java/org/springframework/data/redis/connection/RedisConnectionFactory.java +++ b/src/main/java/org/springframework/data/redis/connection/RedisConnectionFactory.java @@ -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(); diff --git a/src/main/java/org/springframework/data/redis/connection/lettuce/AuthenticatingRedisClient.java b/src/main/java/org/springframework/data/redis/connection/lettuce/AuthenticatingRedisClient.java index 4ec46c934..309a0795f 100644 --- a/src/main/java/org/springframework/data/redis/connection/lettuce/AuthenticatingRedisClient.java +++ b/src/main/java/org/springframework/data/redis/connection/lettuce/AuthenticatingRedisClient.java @@ -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; diff --git a/src/main/java/org/springframework/data/redis/connection/lettuce/DefaultLettucePool.java b/src/main/java/org/springframework/data/redis/connection/lettuce/DefaultLettucePool.java index 178ab2fe9..449a27fd5 100644 --- a/src/main/java/org/springframework/data/redis/connection/lettuce/DefaultLettucePool.java +++ b/src/main/java/org/springframework/data/redis/connection/lettuce/DefaultLettucePool.java @@ -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 DefaultLettucePool 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(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 getResource() { try { @@ -118,6 +161,10 @@ public class DefaultLettucePool implements LettucePool, InitializingBean { } } + /** + * + * @return The Redis client + */ public RedisClient getClient() { return client; } diff --git a/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceConnection.java b/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceConnection.java index 2843a9987..6a5fb5145 100644 --- a/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceConnection.java +++ b/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceConnection.java @@ -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 Lettuce Redis client. + * {@code RedisConnection} implementation on top of Lettuce 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 CODEC = new BytesRedisCodec(); + + private static final Method SYNC_HANDLER; private static final ExceptionTranslationStrategy EXCEPTION_TRANSLATION = new FallbackExceptionTranslationStrategy( LettuceConverters.exceptionConverter()); - - static final RedisCodec 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 asyncSharedConn; private final com.lambdaworks.redis.RedisConnection sharedConn; private com.lambdaworks.redis.RedisAsyncConnection 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( - 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 cmdArg = new CommandArgs(CODEC); if (!ObjectUtils.isEmpty(args)) { cmdArg.addKeys(args); } - CommandOutput expectedOutput = commandOutputTypeHint != null ? commandOutputTypeHint : typeHints.getTypeHint(cmd); + RedisAsyncConnectionImpl connectionImpl = (RedisAsyncConnectionImpl) 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 results = new ArrayList(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 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(returnType))); + pipeline(new LettuceResult(getAsyncConnection().eval(convertedScript, + LettuceConverters.toScriptOutputType(returnType), keys, args), new LettuceEvalResultsConverter( + returnType))); return null; } if (isQueueing()) { - transaction(new LettuceTxResult(getConnection().eval(script, LettuceConverters.toScriptOutputType(returnType), - keys, args), new LettuceEvalResultsConverter(returnType))); + transaction(new LettuceTxResult(getConnection().eval(convertedScript, + LettuceConverters.toScriptOutputType(returnType), keys, args), new LettuceEvalResultsConverter( + returnType))); return null; } - return new LettuceEvalResultsConverter(returnType).convert(getConnection().eval(script, + return new LettuceEvalResultsConverter(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 result = (List) eval("return redis.call('TIME')".getBytes(), ReturnType.MULTI, 0); + List 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(Long.valueOf(nextCursorId), ((List) result.get(1))); + if (isQueueing()) { + transaction(new LettuceTxResult(getAsyncConnection().scan(scanCursor, scanArgs))); + return null; + } + + KeyScanCursor keyScanCursor = getConnection().scan(scanCursor, scanArgs); + String nextCursorId = keyScanCursor.getCursor(); + + List keys = keyScanCursor.getKeys(); + + return new ScanIteration(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 values = failsafeReadScanValues(result, LettuceConverters.bytesListToMapConverter()); + MapScanCursor mapScanCursor = getConnection().hscan(key, scanCursor, scanArgs); + String nextCursorId = mapScanCursor.getCursor(); + + Map values = mapScanCursor.getMap(); return new ScanIteration>(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 values = failsafeReadScanValues(result, null); + ValueScanCursor valueScanCursor = getConnection().sscan(key, scanCursor, scanArgs); + String nextCursorId = valueScanCursor.getCursor(); + + List values = failsafeReadScanValues(valueScanCursor.getValues(), null); return new ScanIteration(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 scoredValueScanCursor = getConnection().zscan(key, scanCursor, scanArgs); + String nextCursorId = scoredValueScanCursor.getCursor(); - List values = failsafeReadScanValues(result, LettuceConverters.bytesListToTupleListConverter()); + List> result = scoredValueScanCursor.getValues(); + + List values = failsafeReadScanValues(result, LettuceConverters.scoredValuesToTupleList()); return new ScanIteration(Long.valueOf(nextCursorId), values); } }.open(); @@ -3350,7 +3345,7 @@ public class LettuceConnection extends AbstractRedisConnection { private 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 getDedicatedConnection() { if (dedicatedConn == null) { - this.dedicatedConn = new com.lambdaworks.redis.RedisConnection(getAsyncDedicatedConnection()); + this.dedicatedConn = syncConnection(getAsyncDedicatedConnection()); } return dedicatedConn; } + private com.lambdaworks.redis.RedisConnection syncConnection( + RedisAsyncConnection asyncConnection) { + try { + return (com.lambdaworks.redis.RedisConnection) SYNC_HANDLER.invoke(null, asyncConnection, + new Class[] { com.lambdaworks.redis.RedisConnection.class, RedisClusterConnection.class }); + } catch (Exception ex) { + throw convertLettuceAccessException(ex); + } + } + private Future 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 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 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 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 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 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); + } } } diff --git a/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceConnectionFactory.java b/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceConnectionFactory.java index 3b28c0bbb..73fe212f9 100644 --- a/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceConnectionFactory.java +++ b/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceConnectionFactory.java @@ -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 Lettuce-based connections. + * Connection factory creating Lettuce-based connections. *

* 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 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 LettuceConnectionFactory 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(connection).ping(); - } catch (RedisException e) { + + boolean valid = false; + + if (connection.isOpen()) { + try { + RedisFuture 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()); } } diff --git a/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceConverters.java b/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceConverters.java index a30a72c38..73360394f 100644 --- a/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceConverters.java +++ b/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceConverters.java @@ -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_TO_LONG; private static final Converter, Set> BYTES_LIST_TO_BYTES_SET; private static final Converter BYTES_TO_STRING; + private static final Converter STRING_TO_BYTES; private static final Converter, List> BYTES_SET_TO_BYTES_LIST; private static final Converter, List> KEY_VALUE_TO_BYTES_LIST; private static final Converter>, Set> SCORED_VALUES_TO_TUPLE_SET; + private static final Converter>, List> SCORED_VALUES_TO_TUPLE_LIST; private static final Converter, Tuple> SCORED_VALUE_TO_TUPLE; private static final Converter EXCEPTION_CONVERTER = new LettuceExceptionConverter(); private static final Converter LONG_TO_BOOLEAN = new LongToBooleanConverter(); private static final Converter, Map> BYTES_LIST_TO_MAP; private static final Converter, List> BYTES_LIST_TO_TUPLE_LIST_CONVERTER; - private static final Converter> 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() { public Long convert(Date source) { return source != null ? source.getTime() : null; } - }; BYTES_LIST_TO_BYTES_SET = new Converter, Set>() { public Set convert(List results) { return results != null ? new LinkedHashSet(results) : null; } - }; BYTES_TO_STRING = new Converter() { @@ -94,7 +104,16 @@ abstract public class LettuceConverters extends Converters { } return new String(source); } + }; + STRING_TO_BYTES = new Converter() { + @Override + public byte[] convert(String source) { + if (source == null) { + return null; + } + return source.getBytes(); + } }; BYTES_SET_TO_BYTES_LIST = new Converter, List>() { public List convert(Set results) { @@ -130,7 +149,6 @@ abstract public class LettuceConverters extends Converters { return target; } - }; SCORED_VALUES_TO_TUPLE_SET = new Converter>, Set>() { public Set convert(List> source) { @@ -144,11 +162,23 @@ abstract public class LettuceConverters extends Converters { return tuples; } }; + + SCORED_VALUES_TO_TUPLE_LIST = new Converter>, List>() { + public List convert(List> source) { + if (source == null) { + return null; + } + List tuples = new ArrayList(source.size()); + for (ScoredValue value : source) { + tuples.add(LettuceConverters.toTuple(value)); + } + return tuples; + } + }; SCORED_VALUE_TO_TUPLE = new Converter, Tuple>() { public Tuple convert(ScoredValue source) { return source != null ? new DefaultTuple(source.value, Double.valueOf(source.score)) : null; } - }; BYTES_LIST_TO_TUPLE_LIST_CONVERTER = new Converter, List>() { @@ -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 toTuple(List list) { @@ -217,6 +251,10 @@ abstract public class LettuceConverters extends Converters { return SCORED_VALUES_TO_TUPLE_SET; } + public static Converter>, List> scoredValuesToTupleList() { + return SCORED_VALUES_TO_TUPLE_LIST; + } + public static Converter, 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 toListOfRedisServer(List> source) { + + if (CollectionUtils.isEmpty(source)) { + return Collections.emptyList(); + } + + List sentinels = new ArrayList(); + for (Map 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 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()); + } + } diff --git a/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceExceptionConverter.java b/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceExceptionConverter.java index 51ab4d16a..e7ab27977 100644 --- a/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceExceptionConverter.java +++ b/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceExceptionConverter.java @@ -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 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 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 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 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 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 connectSentinel() { + return redisClient.connectSentinelAsync(); + } + + + @Override + public boolean isOpen() { + return connection != null && connection.isOpen(); + } + +} diff --git a/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceUtils.java b/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceUtils.java index 4e0ee11fd..8cc24fd45 100644 --- a/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceUtils.java +++ b/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceUtils.java @@ -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)); } } diff --git a/src/main/java/org/springframework/data/redis/connection/lettuce/package-info.java b/src/main/java/org/springframework/data/redis/connection/lettuce/package-info.java index c54723fa1..5f69d43da 100644 --- a/src/main/java/org/springframework/data/redis/connection/lettuce/package-info.java +++ b/src/main/java/org/springframework/data/redis/connection/lettuce/package-info.java @@ -1,5 +1,5 @@ /** - * Connection package for Lettuce Redis client. + * Connection package for Lettuce Redis client. */ package org.springframework.data.redis.connection.lettuce; diff --git a/src/test/java/org/springframework/data/redis/connection/AbstractConnectionIntegrationTests.java b/src/test/java/org/springframework/data/redis/connection/AbstractConnectionIntegrationTests.java index f07b91dd2..ab1ca031f 100644 --- a/src/test/java/org/springframework/data/redis/connection/AbstractConnectionIntegrationTests.java +++ b/src/test/java/org/springframework/data/redis/connection/AbstractConnectionIntegrationTests.java @@ -2074,8 +2074,8 @@ public abstract class AbstractConnectionIntegrationTests { @IfProfileValue(name = "redisVersion", value = "2.8.9+") public void pfAddShouldAddToNonExistingKeyCorrectly() { - if (!ConnectionUtils.isJedis(connectionFactory)) { - throw new AssumptionViolatedException("PFADD is only available for jedis"); + if (!ConnectionUtils.isJedis(connectionFactory) && !ConnectionUtils.isLettuce(connectionFactory)) { + throw new AssumptionViolatedException("PFADD is only available for jedis and lettuce"); } actual.add(connection.pfAdd("hll", "a", "b", "c")); @@ -2091,8 +2091,8 @@ public abstract class AbstractConnectionIntegrationTests { @IfProfileValue(name = "redisVersion", value = "2.8.9+") public void pfAddShouldReturnZeroWhenValueAlreadyExists() { - if (!ConnectionUtils.isJedis(connectionFactory)) { - throw new AssumptionViolatedException("PFADD is only available for jedis"); + if (!ConnectionUtils.isJedis(connectionFactory) && !ConnectionUtils.isLettuce(connectionFactory)) { + throw new AssumptionViolatedException("PFADD is only available for jedis and lettuce"); } actual.add(connection.pfAdd("hll", "a", "b", "c")); @@ -2112,8 +2112,8 @@ public abstract class AbstractConnectionIntegrationTests { @IfProfileValue(name = "redisVersion", value = "2.8.9+") public void pfCountShouldReturnCorrectly() { - if (!ConnectionUtils.isJedis(connectionFactory)) { - throw new AssumptionViolatedException("PFADD is only available for jedis"); + if (!ConnectionUtils.isJedis(connectionFactory) && !ConnectionUtils.isLettuce(connectionFactory)) { + throw new AssumptionViolatedException("PFADD is only available for jedis and lettuce"); } actual.add(connection.pfAdd("hll", "a", "b", "c")); @@ -2131,8 +2131,8 @@ public abstract class AbstractConnectionIntegrationTests { @IfProfileValue(name = "redisVersion", value = "2.8.9+") public void pfCountWithMultipleKeysShouldReturnCorrectly() { - if (!ConnectionUtils.isJedis(connectionFactory)) { - throw new AssumptionViolatedException("PFADD is only available for jedis"); + if (!ConnectionUtils.isJedis(connectionFactory) && !ConnectionUtils.isLettuce(connectionFactory)) { + throw new AssumptionViolatedException("PFADD is only available for jedis and lettuce"); } actual.add(connection.pfAdd("hll", "a", "b", "c")); @@ -2152,8 +2152,8 @@ public abstract class AbstractConnectionIntegrationTests { @IfProfileValue(name = "redisVersion", value = "2.8.9+") public void pfCountWithNullKeysShouldThrowIllegalArgumentException() { - if (!ConnectionUtils.isJedis(connectionFactory)) { - throw new AssumptionViolatedException("PFADD is only available for jedis"); + if (!ConnectionUtils.isJedis(connectionFactory) && !ConnectionUtils.isLettuce(connectionFactory)) { + throw new AssumptionViolatedException("PFADD is only available for jedis and lettuce"); } actual.add(connection.pfCount((String[]) null)); diff --git a/src/test/java/org/springframework/data/redis/connection/lettuce/LettuceConnectionFactoryTests.java b/src/test/java/org/springframework/data/redis/connection/lettuce/LettuceConnectionFactoryTests.java index e6759dff9..b28660182 100644 --- a/src/test/java/org/springframework/data/redis/connection/lettuce/LettuceConnectionFactoryTests.java +++ b/src/test/java/org/springframework/data/redis/connection/lettuce/LettuceConnectionFactoryTests.java @@ -46,6 +46,7 @@ public class LettuceConnectionFactoryTests { public void setUp() { factory = new LettuceConnectionFactory(SettingsUtils.getHost(), SettingsUtils.getPort()); factory.afterPropertiesSet(); + factory.setShutdownTimeout(0); connection = new DefaultStringRedisConnection(factory.getConnection()); } @@ -109,6 +110,7 @@ public class LettuceConnectionFactoryTests { @Test public void testSelectDb() { LettuceConnectionFactory factory2 = new LettuceConnectionFactory(SettingsUtils.getHost(), SettingsUtils.getPort()); + factory2.setShutdownTimeout(0); factory2.setDatabase(1); factory2.afterPropertiesSet(); StringRedisConnection connection2 = new DefaultStringRedisConnection(factory2.getConnection()); @@ -177,7 +179,9 @@ public class LettuceConnectionFactoryTests { newConnection.close(); } + @Test public void testGetConnectionException() { + factory.resetConnection(); factory.setHostName("fakeHost"); factory.afterPropertiesSet(); try { @@ -207,6 +211,7 @@ public class LettuceConnectionFactoryTests { DefaultLettucePool pool = new DefaultLettucePool(SettingsUtils.getHost(), SettingsUtils.getPort()); pool.afterPropertiesSet(); LettuceConnectionFactory factory2 = new LettuceConnectionFactory(pool); + factory2.setShutdownTimeout(0); factory2.afterPropertiesSet(); RedisConnection conn2 = factory2.getConnection(); conn2.close(); diff --git a/src/test/java/org/springframework/data/redis/connection/lettuce/LettuceConnectionIntegrationTests.java b/src/test/java/org/springframework/data/redis/connection/lettuce/LettuceConnectionIntegrationTests.java index 45745181f..622e79848 100644 --- a/src/test/java/org/springframework/data/redis/connection/lettuce/LettuceConnectionIntegrationTests.java +++ b/src/test/java/org/springframework/data/redis/connection/lettuce/LettuceConnectionIntegrationTests.java @@ -16,6 +16,7 @@ package org.springframework.data.redis.connection.lettuce; +import static org.hamcrest.CoreMatchers.*; import static org.junit.Assert.*; import static org.junit.Assume.*; import static org.springframework.data.redis.SpinBarrier.*; @@ -38,9 +39,12 @@ import org.springframework.data.redis.TestCondition; import org.springframework.data.redis.connection.AbstractConnectionIntegrationTests; import org.springframework.data.redis.connection.DefaultStringRedisConnection; import org.springframework.data.redis.connection.RedisConnection; +import org.springframework.data.redis.connection.RedisSentinelConfiguration; import org.springframework.data.redis.connection.ReturnType; import org.springframework.data.redis.connection.StringRedisConnection; +import org.springframework.data.redis.test.util.RedisSentinelRule; import org.springframework.data.redis.test.util.RelaxedJUnit4ClassRunner; +import org.springframework.data.redis.test.util.RequiresRedisSentinel; import org.springframework.test.annotation.IfProfileValue; import org.springframework.test.context.ContextConfiguration; @@ -133,6 +137,7 @@ public class LettuceConnectionIntegrationTests extends AbstractConnectionIntegra DefaultLettucePool pool = new DefaultLettucePool(SettingsUtils.getHost(), SettingsUtils.getPort()); pool.afterPropertiesSet(); LettuceConnectionFactory factory2 = new LettuceConnectionFactory(pool); + factory2.setShutdownTimeout(0); factory2.afterPropertiesSet(); RedisConnection connection = factory2.getConnection(); // Use the connection to make sure the channel is initialized, else nothing happens on close @@ -170,6 +175,7 @@ public class LettuceConnectionIntegrationTests extends AbstractConnectionIntegra @Test public void testCloseNonPooledConnectionNotShared() { LettuceConnectionFactory factory2 = new LettuceConnectionFactory(SettingsUtils.getHost(), SettingsUtils.getPort()); + factory2.setShutdownTimeout(0); factory2.setShareNativeConnection(false); factory2.afterPropertiesSet(); RedisConnection connection = factory2.getConnection(); @@ -190,6 +196,7 @@ public class LettuceConnectionIntegrationTests extends AbstractConnectionIntegra DefaultLettucePool pool = new DefaultLettucePool(SettingsUtils.getHost(), SettingsUtils.getPort()); pool.afterPropertiesSet(); LettuceConnectionFactory factory2 = new LettuceConnectionFactory(pool); + factory2.setShutdownTimeout(0); factory2.setShareNativeConnection(false); factory2.afterPropertiesSet(); RedisConnection connection = factory2.getConnection(); @@ -210,6 +217,7 @@ public class LettuceConnectionIntegrationTests extends AbstractConnectionIntegra DefaultLettucePool pool = new DefaultLettucePool(SettingsUtils.getHost(), SettingsUtils.getPort()); pool.afterPropertiesSet(); LettuceConnectionFactory factory2 = new LettuceConnectionFactory(pool); + factory2.setShutdownTimeout(0); factory2.setShareNativeConnection(false); factory2.afterPropertiesSet(); RedisConnection connection = factory2.getConnection(); @@ -224,12 +232,6 @@ public class LettuceConnectionIntegrationTests extends AbstractConnectionIntegra super.testSelect(); } - @Test(expected = UnsupportedOperationException.class) - @IfProfileValue(name = "redisVersion", value = "2.6+") - public void testSRandMemberCountNegative() { - super.testSRandMemberCountNegative(); - } - @Test @IfProfileValue(name = "runLongTests", value = "true") public void testScriptKill() throws Exception { @@ -241,6 +243,7 @@ public class LettuceConnectionIntegrationTests extends AbstractConnectionIntegra // Use a different factory to get a non-shared native conn for blocking script final LettuceConnectionFactory factory2 = new LettuceConnectionFactory(SettingsUtils.getHost(), SettingsUtils.getPort()); + factory2.setShutdownTimeout(0); factory2.afterPropertiesSet(); DefaultStringRedisConnection conn2 = new DefaultStringRedisConnection(factory2.getConnection()); try { @@ -269,6 +272,7 @@ public class LettuceConnectionIntegrationTests extends AbstractConnectionIntegra verifyResults(Arrays.asList(new Object[] { true })); // Lettuce does not support select when using shared conn, use a new conn factory LettuceConnectionFactory factory2 = new LettuceConnectionFactory(); + factory2.setShutdownTimeout(0); factory2.setDatabase(1); factory2.afterPropertiesSet(); StringRedisConnection conn2 = new DefaultStringRedisConnection(factory2.getConnection()); @@ -313,7 +317,7 @@ public class LettuceConnectionIntegrationTests extends AbstractConnectionIntegra assertEquals(Arrays.asList(new Object[] { "key1", "arg1" }), Arrays.asList(new Object[] { new String(scriptResults.get(0)), new String(scriptResults.get(1)) })); } - + /** * @see DATAREDIS-106 */ @@ -323,10 +327,21 @@ public class LettuceConnectionIntegrationTests extends AbstractConnectionIntegra connection.zAdd("myzset", 1, "one"); connection.zAdd("myzset", 2, "two"); connection.zAdd("myzset", 3, "three"); - + Set zRangeByScore = connection.zRangeByScore("myzset", "(1", "2"); - + assertEquals("two", new String(zRangeByScore.iterator().next())); } + /** + * @see DATAREDIS-348 + */ + @Test + @RequiresRedisSentinel(RedisSentinelRule.SentinelsAvailable.ONE_ACTIVE) + public void shouldReturnSentinelCommandsWhenWhenActiveSentinelFound() { + + ((LettuceConnection) byteConnection).setSentinelConfiguration(new RedisSentinelConfiguration().master("mymaster") + .sentinel("127.0.0.1", 26379).sentinel("127.0.0.1", 26380)); + assertThat(connection.getSentinelConnection(), notNullValue()); + } } diff --git a/src/test/java/org/springframework/data/redis/connection/lettuce/LettuceConnectionPipelineIntegrationTests.java b/src/test/java/org/springframework/data/redis/connection/lettuce/LettuceConnectionPipelineIntegrationTests.java index 51a50cd86..3f19493be 100644 --- a/src/test/java/org/springframework/data/redis/connection/lettuce/LettuceConnectionPipelineIntegrationTests.java +++ b/src/test/java/org/springframework/data/redis/connection/lettuce/LettuceConnectionPipelineIntegrationTests.java @@ -52,12 +52,6 @@ public class LettuceConnectionPipelineIntegrationTests extends AbstractConnectio super.testSelect(); } - @Test(expected = UnsupportedOperationException.class) - @IfProfileValue(name = "redisVersion", value = "2.6+") - public void testSRandMemberCountNegative() { - super.testSRandMemberCountNegative(); - } - @Test @IfProfileValue(name = "runLongTests", value = "true") public void testScriptKill() throws Exception { diff --git a/src/test/java/org/springframework/data/redis/connection/lettuce/LettuceConnectionTransactionIntegrationTests.java b/src/test/java/org/springframework/data/redis/connection/lettuce/LettuceConnectionTransactionIntegrationTests.java index d0592c334..d4eba631d 100644 --- a/src/test/java/org/springframework/data/redis/connection/lettuce/LettuceConnectionTransactionIntegrationTests.java +++ b/src/test/java/org/springframework/data/redis/connection/lettuce/LettuceConnectionTransactionIntegrationTests.java @@ -25,7 +25,6 @@ import org.springframework.data.redis.connection.AbstractConnectionTransactionIn import org.springframework.data.redis.connection.DefaultStringRedisConnection; import org.springframework.data.redis.connection.StringRedisConnection; import org.springframework.data.redis.test.util.RelaxedJUnit4ClassRunner; -import org.springframework.test.annotation.IfProfileValue; import org.springframework.test.context.ContextConfiguration; /** @@ -39,12 +38,6 @@ import org.springframework.test.context.ContextConfiguration; @ContextConfiguration("LettuceConnectionIntegrationTests-context.xml") public class LettuceConnectionTransactionIntegrationTests extends AbstractConnectionTransactionIntegrationTests { - @Test(expected = UnsupportedOperationException.class) - @IfProfileValue(name = "redisVersion", value = "2.6+") - public void testSRandMemberCountNegative() { - super.testSRandMemberCountNegative(); - } - @Test public void testMove() { connection.set("foo", "bar"); diff --git a/src/test/java/org/springframework/data/redis/connection/lettuce/LettuceConnectionUnitTestSuite.java b/src/test/java/org/springframework/data/redis/connection/lettuce/LettuceConnectionUnitTestSuite.java index ec0f1f934..707ee36a0 100644 --- a/src/test/java/org/springframework/data/redis/connection/lettuce/LettuceConnectionUnitTestSuite.java +++ b/src/test/java/org/springframework/data/redis/connection/lettuce/LettuceConnectionUnitTestSuite.java @@ -15,19 +15,23 @@ */ package org.springframework.data.redis.connection.lettuce; -import static org.mockito.Matchers.*; +import static org.mockito.Matchers.any; +import static org.mockito.Matchers.eq; import static org.mockito.Mockito.*; +import java.lang.reflect.InvocationTargetException; + import org.junit.Before; import org.junit.Test; import org.junit.runner.RunWith; import org.junit.runners.Suite; +import org.springframework.dao.InvalidDataAccessResourceUsageException; import org.springframework.data.redis.connection.AbstractConnectionUnitTestBase; import org.springframework.data.redis.connection.RedisServerCommands.ShutdownOption; import org.springframework.data.redis.connection.lettuce.LettuceConnectionUnitTestSuite.LettuceConnectionUnitTests; import org.springframework.data.redis.connection.lettuce.LettuceConnectionUnitTestSuite.LettucePipelineConnectionUnitTests; -import com.lambdaworks.redis.RedisAsyncConnection; +import com.lambdaworks.redis.RedisAsyncConnectionImpl; import com.lambdaworks.redis.RedisClient; import com.lambdaworks.redis.codec.RedisCodec; @@ -39,13 +43,13 @@ import com.lambdaworks.redis.codec.RedisCodec; public class LettuceConnectionUnitTestSuite { @SuppressWarnings("rawtypes") - public static class LettuceConnectionUnitTests extends AbstractConnectionUnitTestBase { + public static class LettuceConnectionUnitTests extends AbstractConnectionUnitTestBase { protected LettuceConnection connection; @SuppressWarnings({ "unchecked" }) @Before - public void setUp() { + public void setUp() throws InvocationTargetException, IllegalAccessException { RedisClient clientMock = mock(RedisClient.class); when(clientMock.connectAsync((RedisCodec) any())).thenReturn(getNativeRedisConnectionMock()); @@ -132,13 +136,21 @@ public class LettuceConnectionUnitTestSuite { connection.slaveOfNoOne(); verifyNativeConnectionInvocation().slaveofNoOne(); } + + /** + * @see DATAREDIS-348 + */ + @Test(expected = InvalidDataAccessResourceUsageException.class) + public void shouldThrowExceptionWhenAccessingRedisSentinelsCommandsWhenNoSentinelsConfigured() { + connection.getSentinelConnection(); + } } public static class LettucePipelineConnectionUnitTests extends LettuceConnectionUnitTests { @Override @Before - public void setUp() { + public void setUp() throws InvocationTargetException, IllegalAccessException { super.setUp(); this.connection.openPipeline(); } diff --git a/src/test/java/org/springframework/data/redis/connection/lettuce/LettuceSentinelConnectionUnitTests.java b/src/test/java/org/springframework/data/redis/connection/lettuce/LettuceSentinelConnectionUnitTests.java new file mode 100644 index 000000000..b821991b4 --- /dev/null +++ b/src/test/java/org/springframework/data/redis/connection/lettuce/LettuceSentinelConnectionUnitTests.java @@ -0,0 +1,217 @@ +/* + * 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 static org.mockito.Matchers.eq; +import static org.mockito.Mockito.*; + +import java.util.List; +import java.util.Map; + +import org.junit.Before; +import org.junit.Test; +import org.junit.runner.RunWith; +import org.mockito.Mock; +import org.mockito.runners.MockitoJUnitRunner; +import org.springframework.data.redis.connection.RedisNode; +import org.springframework.data.redis.connection.RedisNode.RedisNodeBuilder; +import org.springframework.data.redis.connection.RedisServer; +import org.springframework.data.redis.connection.jedis.JedisSentinelConnection; + +import redis.clients.jedis.Jedis; + +import com.lambdaworks.redis.RedisClient; +import com.lambdaworks.redis.RedisFuture; +import com.lambdaworks.redis.RedisSentinelAsyncConnection; + +/** + * @author Christoph Strobl + * @author Mark Paluch + */ +@RunWith(MockitoJUnitRunner.class) +public class LettuceSentinelConnectionUnitTests { + + public static final String MASTER_ID = "mymaster"; + private @Mock RedisClient redisClientMock; + + private @Mock RedisSentinelAsyncConnection connectionMock; + + private @Mock RedisFuture>> redisFutureMock; + + private LettuceSentinelConnection connection; + + @Before + public void setUp() { + + when(redisClientMock.connectSentinelAsync()).thenReturn(connectionMock); + this.connection = new LettuceSentinelConnection(redisClientMock); + } + + /** + * @see DATAREDIS-348 + */ + @Test + public void shouldConnectAfterCreation() { + verify(redisClientMock, times(1)).connectSentinelAsync(); + } + + /** + * @see DATAREDIS-348 + */ + @SuppressWarnings("resource") + @Test + public void shouldNotConnectIfAlreadyConnected() { + + Jedis yetAnotherJedisMock = mock(Jedis.class); + when(yetAnotherJedisMock.isConnected()).thenReturn(true); + + new JedisSentinelConnection(yetAnotherJedisMock); + + verify(yetAnotherJedisMock, never()).connect(); + } + + /** + * @see DATAREDIS-348 + */ + @Test + public void failoverShouldBeSentCorrectly() { + + connection.failover(new RedisNodeBuilder().withName(MASTER_ID).build()); + verify(connectionMock, times(1)).failover(eq(MASTER_ID)); + } + + /** + * @see DATAREDIS-348 + */ + @Test(expected = IllegalArgumentException.class) + public void failoverShouldThrowExceptionIfMasterNodeIsNull() { + connection.failover(null); + } + + /** + * @see DATAREDIS-348 + */ + @Test(expected = IllegalArgumentException.class) + public void failoverShouldThrowExceptionIfMasterNodeNameIsEmpty() { + connection.failover(new RedisNodeBuilder().build()); + } + + /** + * @see DATAREDIS-348 + */ + @Test + public void mastersShouldReadMastersCorrectly() { + + when(connectionMock.masters()).thenReturn(redisFutureMock); + connection.masters(); + verify(connectionMock, times(1)).masters(); + } + + /** + * @see DATAREDIS-348 + */ + @Test + public void shouldReadSlavesCorrectly() { + + when(connectionMock.slaves(MASTER_ID)).thenReturn(redisFutureMock); + connection.slaves(MASTER_ID); + verify(connectionMock, times(1)).slaves(eq(MASTER_ID)); + } + + /** + * @see DATAREDIS-348 + */ + @Test + public void shouldReadSlavesCorrectlyWhenGivenNamedNode() { + + when(connectionMock.slaves(MASTER_ID)).thenReturn(redisFutureMock); + connection.slaves(new RedisNodeBuilder().withName(MASTER_ID).build()); + verify(connectionMock, times(1)).slaves(eq(MASTER_ID)); + } + + /** + * @see DATAREDIS-348 + */ + @Test(expected = IllegalArgumentException.class) + public void readSlavesShouldThrowExceptionWhenGivenEmptyMasterName() { + connection.slaves(""); + } + + /** + * @see DATAREDIS-348 + */ + @Test(expected = IllegalArgumentException.class) + public void readSlavesShouldThrowExceptionWhenGivenNull() { + connection.slaves((RedisNode) null); + } + + /** + * @see DATAREDIS-348 + */ + @Test(expected = IllegalArgumentException.class) + public void readSlavesShouldThrowExceptionWhenNodeWithoutName() { + connection.slaves(new RedisNodeBuilder().build()); + } + + /** + * @see DATAREDIS-348 + */ + @Test + public void shouldRemoveMasterCorrectlyWhenGivenNamedNode() { + + connection.remove(new RedisNodeBuilder().withName(MASTER_ID).build()); + verify(connectionMock, times(1)).remove(eq(MASTER_ID)); + } + + /** + * @see DATAREDIS-348 + */ + @Test(expected = IllegalArgumentException.class) + public void removeShouldThrowExceptionWhenGivenEmptyMasterName() { + connection.remove(""); + } + + /** + * @see DATAREDIS-348 + */ + @Test(expected = IllegalArgumentException.class) + public void removeShouldThrowExceptionWhenGivenNull() { + connection.remove((RedisNode) null); + } + + /** + * @see DATAREDIS-348 + */ + @Test(expected = IllegalArgumentException.class) + public void removeShouldThrowExceptionWhenNodeWithoutName() { + connection.remove(new RedisNodeBuilder().build()); + } + + /** + * @see DATAREDIS-348 + */ + @Test + public void monitorShouldBeSentCorrectly() { + + RedisServer server = new RedisServer("127.0.0.1", 6382); + server.setName("anothermaster"); + server.setQuorum(3L); + + connection.monitor(server); + verify(connectionMock, times(1)).monitor(eq("anothermaster"), eq("127.0.0.1"), eq(6382), eq(3)); + } + +} diff --git a/src/test/java/org/springframework/data/redis/connection/lettuce/LettuceSentinelIntegrationTests.java b/src/test/java/org/springframework/data/redis/connection/lettuce/LettuceSentinelIntegrationTests.java new file mode 100644 index 000000000..8ec068115 --- /dev/null +++ b/src/test/java/org/springframework/data/redis/connection/lettuce/LettuceSentinelIntegrationTests.java @@ -0,0 +1,105 @@ +/* + * 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 static org.hamcrest.CoreMatchers.*; +import static org.junit.Assert.*; + +import java.util.Collection; +import java.util.List; + +import org.junit.*; +import org.springframework.dao.InvalidDataAccessApiUsageException; +import org.springframework.data.redis.connection.*; +import org.springframework.data.redis.connection.jedis.JedisConnectionFactory; +import org.springframework.data.redis.test.util.RedisSentinelRule; +import org.springframework.test.annotation.IfProfileValue; + +/** + * @author Christoph Strobl + * @author Thomas Darimont + * @author Mark Paluch + */ +public class LettuceSentinelIntegrationTests extends AbstractConnectionIntegrationTests { + + private static final String MASTER_NAME = "mymaster"; + private static final RedisServer SENTINEL_0 = new RedisServer("127.0.0.1", 26379); + private static final RedisServer SENTINEL_1 = new RedisServer("127.0.0.1", 26380); + + private static final RedisServer SLAVE_0 = new RedisServer("127.0.0.1", 6380); + private static final RedisServer SLAVE_1 = new RedisServer("127.0.0.1", 6381); + + private static final RedisSentinelConfiguration SENTINEL_CONFIG = new RedisSentinelConfiguration() // + .master(MASTER_NAME) + .sentinel(SENTINEL_0) + .sentinel(SENTINEL_1); + + public @Rule RedisSentinelRule sentinelRule = RedisSentinelRule.forConfig(SENTINEL_CONFIG).oneActive(); + + private static LettuceConnectionFactory lettuceConnectionFactory; + + @BeforeClass + public static void beforeClass(){ + LettuceConnectionFactory lettuceConnectionFactory = new LettuceConnectionFactory(SENTINEL_CONFIG); + lettuceConnectionFactory.setShareNativeConnection(false); + lettuceConnectionFactory.afterPropertiesSet(); + LettuceSentinelIntegrationTests.lettuceConnectionFactory = lettuceConnectionFactory; + } + + @AfterClass + public static void afterClass(){ + LettuceSentinelIntegrationTests.lettuceConnectionFactory.destroy(); + } + + @Before + public void setUp() { + connectionFactory = lettuceConnectionFactory; + super.setUp(); + } + + @After + public void tearDown() { + super.tearDown(); + } + + /** + * @see DATAREDIS-348 + */ + @Test + public void shouldReadMastersCorrectly() { + + List servers = (List) connectionFactory.getSentinelConnection().masters(); + assertThat(servers.size(), is(1)); + assertThat(servers.get(0).getName(),is(MASTER_NAME)); + } + + /** + * @see DATAREDIS-348 + */ + @Test + public void shouldReadSlavesOfMastersCorrectly() { + + RedisSentinelConnection sentinelConnection = connectionFactory.getSentinelConnection(); + + List servers = (List) sentinelConnection.masters(); + assertThat(servers.size(), is(1)); + + Collection slaves = sentinelConnection.slaves(servers.get(0)); + assertThat(slaves.size(), is(2)); + assertThat(slaves, hasItems(SLAVE_0, SLAVE_1)); + } + +} diff --git a/src/test/java/org/springframework/data/redis/test/util/MinimumRedisVersionRule.java b/src/test/java/org/springframework/data/redis/test/util/MinimumRedisVersionRule.java index e376f41bf..9cc28c785 100644 --- a/src/test/java/org/springframework/data/redis/test/util/MinimumRedisVersionRule.java +++ b/src/test/java/org/springframework/data/redis/test/util/MinimumRedisVersionRule.java @@ -23,6 +23,7 @@ import org.springframework.data.redis.RedisVersionUtils; import org.springframework.data.redis.SettingsUtils; import org.springframework.data.redis.Version; import org.springframework.data.redis.connection.RedisConnection; +import org.springframework.data.redis.connection.jedis.JedisConnectionFactory; import org.springframework.data.redis.connection.lettuce.LettuceConnectionFactory; import org.springframework.test.annotation.IfProfileValue; import org.springframework.util.StringUtils; @@ -45,7 +46,7 @@ public class MinimumRedisVersionRule implements TestRule { private Version readServerVersion() { - LettuceConnectionFactory connectionFactory = new LettuceConnectionFactory(); + JedisConnectionFactory connectionFactory = new JedisConnectionFactory(); connectionFactory.setHostName(SettingsUtils.getHost()); connectionFactory.setPort(SettingsUtils.getPort()); connectionFactory.setTimeout(100); diff --git a/template.mf b/template.mf index e29650657..bcde8e585 100644 --- a/template.mf +++ b/template.mf @@ -29,5 +29,5 @@ Import-Template: org.apache.commons.beanutils.*;resolution:="optional";version=1.8.5, redis.*;resolution:="optional";version="[0.2, 1.0)", com.google.common.*;resolution:="optional";version="[11.0.0, 20.0.0)", - com.lambdaworks.*;resolution:="optional";version="[2.2.0, 3.0.0)", - org.jboss.netty.*;resolution:="optional";version="[3.0.0, 4.0.0)" + com.lambdaworks.redis.*;resolution:="optional";version="[3.0.0, 4.0.0)", + io.netty.channel.*;resolution:="optional";version="[4.0.0, 5.0.0)"