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

DATAREDIS-525 - Upgrade to lettuce 5.0.0.Beta1.

Browse Source 
- Replace deprecated method usage by using newly introduced API.
- Improve converter names, remove unused code, fix generics.

Original pull request: #229.
This commit is contained in:
Mark Paluch
2016-10-26 10:48:12 +02:00
parent d257ae1074
commit 62b1354e02
30 changed files with 757 additions and 863 deletions
Download Patch File Download Diff File Expand all files Collapse all files

64
src/main/java/org/springframework/data/redis/connection/ReactiveZSetCommands.java
View File

@@ -21,6 +21,7 @@ import java.util.List;
import java.util.Optional;
import java.util.stream.Collectors;
import com.lambdaworks.redis.Range.Boundary;
import org.reactivestreams.Publisher;
import org.springframework.data.domain.Range;
import org.springframework.data.domain.Sort.Direction;
@@ -37,72 +38,11 @@ import reactor.core.publisher.Mono;
/**
* @author Christoph Strobl
* @author Mark Paluch
* @since 2.0
*/
public interface ReactiveZSetCommands {
/**
* @author Christoph Strobl
*/
class AgrumentConverters {
public static Object lowerBoundArgOf(Range<?> range) {
return rangeToLowerBoundArgumentConverter(false).convert(range);
}
public static Object upperBoundArgOf(Range<?> range) {
return rangeToLowerBoundArgumentConverter(true).convert(range);
}
public static Converter<Range<?>, Object> rangeToLowerBoundArgumentConverter(Boolean upper) {
return (source) -> {
// TODO: fix range exclusion pattern when DATACMNS-920 is resolved
DirectFieldAccessFallbackBeanWrapper bw = new DirectFieldAccessFallbackBeanWrapper(source);
Boolean inclusive = upper ? Boolean.valueOf(bw.getPropertyValue("upperInclusive").toString())
: Boolean.valueOf(bw.getPropertyValue("lowerInclusive").toString());
Object value = upper ? source.getUpperBound() : source.getLowerBound();
if (value instanceof Double) {
Object converted = doubleToRangeConverter().convert((Double) value);
if (!(converted instanceof String) && !inclusive) {
return "(" + converted.toString();
}
return converted;
}
if (value instanceof String) {
if (!StringUtils.hasText((String) value)) {
return upper ? "+" : "-";
}
if (ObjectUtils.nullSafeEquals(value, "+") || ObjectUtils.nullSafeEquals(value, "-")) {
return value;
}
return (inclusive ? "[" : "(") + value.toString();
}
return inclusive ? value : "(" + value;
};
}
public static Converter<Double, Object> doubleToRangeConverter() {
return (source) -> {
if (source.equals(Double.NEGATIVE_INFINITY)) {
return "-inf";
}
if (source.equals(Double.POSITIVE_INFINITY)) {
return "+inf";
}
return source;
};
}
}
/**
* @author Christoph Strobl
*/

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

@@ -50,15 +50,6 @@ public class AuthenticatingRedisClient extends RedisClient {
return super.connect(codec);
}
/*
* (non-Javadoc)
* @see com.lambdaworks.redis.RedisClient#connectAsync(com.lambdaworks.redis.codec.RedisCodec)
*/
@Override
public <K, V> RedisAsyncCommands<K, V> connectAsync(RedisCodec<K, V> codec) {
return super.connectAsync(codec);
}
/*
* (non-Javadoc)
* @see com.lambdaworks.redis.RedisClient#connectPubSub(com.lambdaworks.redis.codec.RedisCodec)

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

@@ -18,21 +18,12 @@ package org.springframework.data.redis.connection.lettuce;
import static com.lambdaworks.redis.protocol.CommandType.*;
import java.lang.reflect.Constructor;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collection;
import java.util.Collections;
import java.util.HashMap;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import java.util.*;
import java.util.Map.Entry;
import java.util.Properties;
import java.util.Queue;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;
import org.springframework.beans.BeanUtils;
import org.springframework.core.convert.converter.Converter;
@@ -47,17 +38,7 @@ import org.springframework.data.geo.Point;
import org.springframework.data.redis.ExceptionTranslationStrategy;
import org.springframework.data.redis.FallbackExceptionTranslationStrategy;
import org.springframework.data.redis.RedisConnectionFailureException;
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;
import org.springframework.data.redis.connection.Subscription;
import org.springframework.data.redis.connection.*;
import org.springframework.data.redis.connection.convert.Converters;
import org.springframework.data.redis.connection.convert.ListConverter;
import org.springframework.data.redis.connection.convert.TransactionResultConverter;
@@ -73,23 +54,7 @@ import org.springframework.util.Assert;
import org.springframework.util.ClassUtils;
import org.springframework.util.ObjectUtils;
import com.lambdaworks.redis.AbstractRedisClient;
import com.lambdaworks.redis.GeoArgs;
import com.lambdaworks.redis.GeoCoordinates;
import com.lambdaworks.redis.GeoWithin;
import com.lambdaworks.redis.KeyScanCursor;
import com.lambdaworks.redis.LettuceFutures;
import com.lambdaworks.redis.MapScanCursor;
import com.lambdaworks.redis.RedisClient;
import com.lambdaworks.redis.RedisException;
import com.lambdaworks.redis.RedisFuture;
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.*;
import com.lambdaworks.redis.api.StatefulConnection;
import com.lambdaworks.redis.api.StatefulRedisConnection;
import com.lambdaworks.redis.api.async.RedisAsyncCommands;
@@ -101,20 +66,7 @@ import com.lambdaworks.redis.cluster.api.async.RedisClusterAsyncCommands;
import com.lambdaworks.redis.cluster.api.sync.RedisClusterCommands;
import com.lambdaworks.redis.codec.ByteArrayCodec;
import com.lambdaworks.redis.codec.RedisCodec;
import com.lambdaworks.redis.output.BooleanOutput;
import com.lambdaworks.redis.output.ByteArrayOutput;
import com.lambdaworks.redis.output.CommandOutput;
import com.lambdaworks.redis.output.DateOutput;
import com.lambdaworks.redis.output.DoubleOutput;
import com.lambdaworks.redis.output.IntegerOutput;
import com.lambdaworks.redis.output.KeyListOutput;
import com.lambdaworks.redis.output.KeyValueOutput;
import com.lambdaworks.redis.output.MapOutput;
import com.lambdaworks.redis.output.MultiOutput;
import com.lambdaworks.redis.output.StatusOutput;
import com.lambdaworks.redis.output.ValueListOutput;
import com.lambdaworks.redis.output.ValueOutput;
import com.lambdaworks.redis.output.ValueSetOutput;
import com.lambdaworks.redis.output.*;
import com.lambdaworks.redis.protocol.Command;
import com.lambdaworks.redis.protocol.CommandArgs;
import com.lambdaworks.redis.protocol.CommandType;
@@ -859,12 +811,18 @@ public class LettuceConnection extends AbstractRedisConnection {
isMulti = false;
try {
if (isPipelined()) {
pipeline(new LettuceResult(((RedisAsyncCommands) getAsyncDedicatedConnection()).exec(),
new LettuceTransactionResultConverter(new LinkedList<FutureResult<?>>(txResults),
LettuceConverters.exceptionConverter())));
RedisFuture<TransactionResult> exec = ((RedisAsyncCommands) getAsyncDedicatedConnection()).exec();
LettuceTransactionResultConverter resultConverter = new LettuceTransactionResultConverter(
new LinkedList<FutureResult<?>>(txResults), LettuceConverters.exceptionConverter());
pipeline(new LettuceResult(exec,
source -> resultConverter.convert(LettuceConverters.transactionResultUnwrapper().convert(source))));
return null;
}
List<Object> results = ((RedisCommands) getDedicatedConnection()).exec();
TransactionResult transactionResult = ((RedisCommands) getDedicatedConnection()).exec();
List<Object> results = LettuceConverters.transactionResultUnwrapper().convert(transactionResult);
return convertPipelineAndTxResults
? new LettuceTransactionResultConverter(txResults, LettuceConverters.exceptionConverter()).convert(results)
: results;
@@ -1369,14 +1327,15 @@ public class LettuceConnection extends AbstractRedisConnection {
public List<byte[]> mGet(byte[]... keys) {
try {
if (isPipelined()) {
pipeline(new LettuceResult(getAsyncConnection().mget(keys)));
pipeline(new LettuceResult(getAsyncConnection().mget(keys), LettuceConverters.keyValueListUnwrapper()));
return null;
}
if (isQueueing()) {
transaction(new LettuceTxResult(getConnection().mget(keys)));
transaction(new LettuceTxResult(getConnection().mget(keys), LettuceConverters.keyValueListUnwrapper()));
return null;
}
return getConnection().mget(keys);
return LettuceConverters.<byte[], byte[]> keyValueListUnwrapper().convert(getConnection().mget(keys));
} catch (Exception ex) {
throw convertLettuceAccessException(ex);
}
@@ -2271,19 +2230,16 @@ public class LettuceConnection extends AbstractRedisConnection {
Assert.notNull(range, "Range for ZCOUNT must not be null!");
String min = LettuceConverters.boundaryToStringForZRange(range.getMin(), "-inf");
String max = LettuceConverters.boundaryToStringForZRange(range.getMax(), "+inf");
try {
if (isPipelined()) {
pipeline(new LettuceResult(getAsyncConnection().zcount(key, min, max)));
pipeline(new LettuceResult(getAsyncConnection().zcount(key, LettuceConverters.toRange(range))));
return null;
}
if (isQueueing()) {
transaction(new LettuceTxResult(getConnection().zcount(key, min, max)));
transaction(new LettuceTxResult(getConnection().zcount(key, LettuceConverters.toRange(range))));
return null;
}
return getConnection().zcount(key, min, max);
return getConnection().zcount(key, LettuceConverters.toRange(range));
} catch (Exception ex) {
throw convertLettuceAccessException(ex);
}
@@ -2390,37 +2346,33 @@ public class LettuceConnection extends AbstractRedisConnection {
Assert.notNull(range, "Range for ZRANGEBYSCORE must not be null!");
String min = LettuceConverters.boundaryToStringForZRange(range.getMin(), "-inf");
String max = LettuceConverters.boundaryToStringForZRange(range.getMax(), "+inf");
try {
if (isPipelined()) {
if (limit != null) {
pipeline(
new LettuceResult(getAsyncConnection().zrangebyscore(key, min, max, limit.getOffset(), limit.getCount()),
LettuceConverters.bytesListToBytesSet()));
pipeline(new LettuceResult(getAsyncConnection().zrangebyscore(key, LettuceConverters.toRange(range),
LettuceConverters.toLimit(limit)), LettuceConverters.bytesListToBytesSet()));
} else {
pipeline(new LettuceResult(getAsyncConnection().zrangebyscore(key, min, max),
pipeline(new LettuceResult(getAsyncConnection().zrangebyscore(key, LettuceConverters.toRange(range)),
LettuceConverters.bytesListToBytesSet()));
}
return null;
}
if (isQueueing()) {
if (limit != null) {
transaction(
new LettuceTxResult(getConnection().zrangebyscore(key, min, max, limit.getOffset(), limit.getCount()),
LettuceConverters.bytesListToBytesSet()));
transaction(new LettuceTxResult(
getConnection().zrangebyscore(key, LettuceConverters.toRange(range), LettuceConverters.toLimit(limit)),
LettuceConverters.bytesListToBytesSet()));
} else {
transaction(new LettuceTxResult(getConnection().zrangebyscore(key, min, max),
transaction(new LettuceTxResult(getConnection().zrangebyscore(key, LettuceConverters.toRange(range)),
LettuceConverters.bytesListToBytesSet()));
}
return null;
}
if (limit != null) {
return LettuceConverters
.toBytesSet(getConnection().zrangebyscore(key, min, max, limit.getOffset(), limit.getCount()));
return LettuceConverters.toBytesSet(
getConnection().zrangebyscore(key, LettuceConverters.toRange(range), LettuceConverters.toLimit(limit)));
}
return LettuceConverters.toBytesSet(getConnection().zrangebyscore(key, min, max));
return LettuceConverters.toBytesSet(getConnection().zrangebyscore(key, LettuceConverters.toRange(range)));
} catch (Exception ex) {
throw convertLettuceAccessException(ex);
}
@@ -2441,37 +2393,35 @@ public class LettuceConnection extends AbstractRedisConnection {
Assert.notNull(range, "Range for ZRANGEBYSCOREWITHSCORES must not be null!");
String min = LettuceConverters.boundaryToStringForZRange(range.getMin(), "-inf");
String max = LettuceConverters.boundaryToStringForZRange(range.getMax(), "+inf");
try {
if (isPipelined()) {
if (limit != null) {
pipeline(new LettuceResult(
getAsyncConnection().zrangebyscoreWithScores(key, min, max, limit.getOffset(), limit.getCount()),
LettuceConverters.scoredValuesToTupleSet()));
pipeline(new LettuceResult(getAsyncConnection().zrangebyscoreWithScores(key, LettuceConverters.toRange(range),
LettuceConverters.toLimit(limit)), LettuceConverters.scoredValuesToTupleSet()));
} else {
pipeline(new LettuceResult(getAsyncConnection().zrangebyscoreWithScores(key, min, max),
LettuceConverters.scoredValuesToTupleSet()));
pipeline(
new LettuceResult(getAsyncConnection().zrangebyscoreWithScores(key, LettuceConverters.toRange(range)),
LettuceConverters.scoredValuesToTupleSet()));
}
return null;
}
if (isQueueing()) {
if (limit != null) {
transaction(new LettuceTxResult(
getConnection().zrangebyscoreWithScores(key, min, max, limit.getOffset(), limit.getCount()),
LettuceConverters.scoredValuesToTupleSet()));
transaction(new LettuceTxResult(getConnection().zrangebyscoreWithScores(key, LettuceConverters.toRange(range),
LettuceConverters.toLimit(limit)), LettuceConverters.scoredValuesToTupleSet()));
} else {
transaction(new LettuceTxResult(getConnection().zrangebyscoreWithScores(key, min, max),
LettuceConverters.scoredValuesToTupleSet()));
transaction(
new LettuceTxResult(getConnection().zrangebyscoreWithScores(key, LettuceConverters.toRange(range)),
LettuceConverters.scoredValuesToTupleSet()));
}
return null;
}
if (limit != null) {
return LettuceConverters
.toTupleSet(getConnection().zrangebyscoreWithScores(key, min, max, limit.getOffset(), limit.getCount()));
return LettuceConverters.toTupleSet(getConnection().zrangebyscoreWithScores(key,
LettuceConverters.toRange(range), LettuceConverters.toLimit(limit)));
}
return LettuceConverters.toTupleSet(getConnection().zrangebyscoreWithScores(key, min, max));
return LettuceConverters
.toTupleSet(getConnection().zrangebyscoreWithScores(key, LettuceConverters.toRange(range)));
} catch (Exception ex) {
throw convertLettuceAccessException(ex);
}
@@ -2526,37 +2476,33 @@ public class LettuceConnection extends AbstractRedisConnection {
Assert.notNull(range, "Range for ZREVRANGEBYSCORE must not be null!");
String min = LettuceConverters.boundaryToStringForZRange(range.getMin(), "-inf");
String max = LettuceConverters.boundaryToStringForZRange(range.getMax(), "+inf");
try {
if (isPipelined()) {
if (limit != null) {
pipeline(new LettuceResult(
getAsyncConnection().zrevrangebyscore(key, max, min, limit.getOffset(), limit.getCount()),
LettuceConverters.bytesListToBytesSet()));
pipeline(new LettuceResult(getAsyncConnection().zrevrangebyscore(key, LettuceConverters.toRange(range),
LettuceConverters.toLimit(limit)), LettuceConverters.bytesListToBytesSet()));
} else {
pipeline(new LettuceResult(getAsyncConnection().zrevrangebyscore(key, max, min),
pipeline(new LettuceResult(getAsyncConnection().zrevrangebyscore(key, LettuceConverters.toRange(range)),
LettuceConverters.bytesListToBytesSet()));
}
return null;
}
if (isQueueing()) {
if (limit != null) {
transaction(
new LettuceTxResult(getConnection().zrevrangebyscore(key, max, min, limit.getOffset(), limit.getCount()),
LettuceConverters.bytesListToBytesSet()));
transaction(new LettuceTxResult(
getConnection().zrevrangebyscore(key, LettuceConverters.toRange(range), LettuceConverters.toLimit(limit)),
LettuceConverters.bytesListToBytesSet()));
} else {
transaction(new LettuceTxResult(getConnection().zrevrangebyscore(key, max, min),
transaction(new LettuceTxResult(getConnection().zrevrangebyscore(key, LettuceConverters.toRange(range)),
LettuceConverters.bytesListToBytesSet()));
}
return null;
}
if (limit != null) {
return LettuceConverters
.toBytesSet(getConnection().zrevrangebyscore(key, max, min, limit.getOffset(), limit.getCount()));
return LettuceConverters.toBytesSet(
getConnection().zrevrangebyscore(key, LettuceConverters.toRange(range), LettuceConverters.toLimit(limit)));
}
return LettuceConverters.toBytesSet(getConnection().zrevrangebyscore(key, max, min));
return LettuceConverters.toBytesSet(getConnection().zrevrangebyscore(key, LettuceConverters.toRange(range)));
} catch (Exception ex) {
throw convertLettuceAccessException(ex);
}
@@ -2584,37 +2530,37 @@ public class LettuceConnection extends AbstractRedisConnection {
Assert.notNull(range, "Range for ZREVRANGEBYSCOREWITHSCORES must not be null!");
String min = LettuceConverters.boundaryToStringForZRange(range.getMin(), "-inf");
String max = LettuceConverters.boundaryToStringForZRange(range.getMax(), "+inf");
try {
if (isPipelined()) {
if (limit != null) {
pipeline(new LettuceResult(
getAsyncConnection().zrevrangebyscoreWithScores(key, max, min, limit.getOffset(), limit.getCount()),
LettuceConverters.scoredValuesToTupleSet()));
pipeline(
new LettuceResult(getAsyncConnection().zrevrangebyscoreWithScores(key, LettuceConverters.toRange(range),
LettuceConverters.toLimit(limit)), LettuceConverters.scoredValuesToTupleSet()));
} else {
pipeline(new LettuceResult(getAsyncConnection().zrevrangebyscoreWithScores(key, max, min),
LettuceConverters.scoredValuesToTupleSet()));
pipeline(
new LettuceResult(getAsyncConnection().zrevrangebyscoreWithScores(key, LettuceConverters.toRange(range)),
LettuceConverters.scoredValuesToTupleSet()));
}
return null;
}
if (isQueueing()) {
if (limit != null) {
transaction(new LettuceTxResult(
getConnection().zrevrangebyscoreWithScores(key, max, min, limit.getOffset(), limit.getCount()),
LettuceConverters.scoredValuesToTupleSet()));
transaction(
new LettuceTxResult(getConnection().zrevrangebyscoreWithScores(key, LettuceConverters.toRange(range),
LettuceConverters.toLimit(limit)), LettuceConverters.scoredValuesToTupleSet()));
} else {
transaction(new LettuceTxResult(getConnection().zrevrangebyscoreWithScores(key, max, min),
LettuceConverters.scoredValuesToTupleSet()));
transaction(
new LettuceTxResult(getConnection().zrevrangebyscoreWithScores(key, LettuceConverters.toRange(range)),
LettuceConverters.scoredValuesToTupleSet()));
}
return null;
}
if (limit != null) {
return LettuceConverters
.toTupleSet(getConnection().zrevrangebyscoreWithScores(key, max, min, limit.getOffset(), limit.getCount()));
return LettuceConverters.toTupleSet(getConnection().zrevrangebyscoreWithScores(key,
LettuceConverters.toRange(range), LettuceConverters.toLimit(limit)));
}
return LettuceConverters.toTupleSet(getConnection().zrevrangebyscoreWithScores(key, max, min));
return LettuceConverters
.toTupleSet(getConnection().zrevrangebyscoreWithScores(key, LettuceConverters.toRange(range)));
} catch (Exception ex) {
throw convertLettuceAccessException(ex);
}
@@ -2683,19 +2629,16 @@ public class LettuceConnection extends AbstractRedisConnection {
Assert.notNull(range, "Range for ZREMRANGEBYSCORE must not be null!");
String min = LettuceConverters.boundaryToStringForZRange(range.getMin(), "-inf");
String max = LettuceConverters.boundaryToStringForZRange(range.getMax(), "+inf");
try {
if (isPipelined()) {
pipeline(new LettuceResult(getAsyncConnection().zremrangebyscore(key, min, max)));
pipeline(new LettuceResult(getAsyncConnection().zremrangebyscore(key, LettuceConverters.toRange(range))));
return null;
}
if (isQueueing()) {
transaction(new LettuceTxResult(getConnection().zremrangebyscore(key, min, max)));
transaction(new LettuceTxResult(getConnection().zremrangebyscore(key, LettuceConverters.toRange(range))));
return null;
}
return getConnection().zremrangebyscore(key, min, max);
return getConnection().zremrangebyscore(key, LettuceConverters.toRange(range));
} catch (Exception ex) {
throw convertLettuceAccessException(ex);
}
@@ -2952,14 +2895,14 @@ public class LettuceConnection extends AbstractRedisConnection {
public List<byte[]> hMGet(byte[] key, byte[]... fields) {
try {
if (isPipelined()) {
pipeline(new LettuceResult(getAsyncConnection().hmget(key, fields)));
pipeline(new LettuceResult(getAsyncConnection().hmget(key, fields), LettuceConverters.keyValueListUnwrapper()));
return null;
}
if (isQueueing()) {
transaction(new LettuceTxResult(getConnection().hmget(key, fields)));
transaction(new LettuceTxResult(getConnection().hmget(key, fields), LettuceConverters.keyValueListUnwrapper()));
return null;
}
return getConnection().hmget(key, fields);
return LettuceConverters.<byte[], byte[]> keyValueListUnwrapper().convert(getConnection().hmget(key, fields));
} catch (Exception ex) {
throw convertLettuceAccessException(ex);
}
@@ -3356,7 +3299,8 @@ public class LettuceConnection extends AbstractRedisConnection {
transaction(new LettuceTxResult(getConnection().geohash(key, members)));
return null;
}
return getConnection().geohash(key, members);
return getConnection().geohash(key, members).stream().map(value -> value.getValueOrElse(null))
.collect(Collectors.toList());
} catch (Exception ex) {
throw convertLettuceAccessException(ex);
}
@@ -3586,7 +3530,7 @@ public class LettuceConnection extends AbstractRedisConnection {
}
getConnection().clientKill(client);
} catch (Exception e) {
convertLettuceAccessException(e);
throw convertLettuceAccessException(e);
}
}
@@ -4097,9 +4041,9 @@ public class LettuceConnection extends AbstractRedisConnection {
break;
}
}
long[] lg = new long[weights.length];
double[] lg = new double[weights.length];
for (int i = 0; i < lg.length; i++) {
lg[i] = (long) weights[i];
lg[i] = weights[i];
}
args.weights(lg);
return args;
@@ -4516,39 +4460,37 @@ public class LettuceConnection extends AbstractRedisConnection {
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()));
pipeline(new LettuceResult(
getAsyncConnection().zrangebylex(key, LettuceConverters.toRange(range), LettuceConverters.toLimit(limit)),
LettuceConverters.bytesListToBytesSet()));
} else {
pipeline(new LettuceResult(getAsyncConnection().zrangebylex(key, min, max),
pipeline(new LettuceResult(getAsyncConnection().zrangebylex(key, LettuceConverters.toRange(range)),
LettuceConverters.bytesListToBytesSet()));
}
return null;
}
if (isQueueing()) {
if (limit != null) {
transaction(
new LettuceTxResult(getConnection().zrangebylex(key, min, max, limit.getOffset(), limit.getCount()),
LettuceConverters.bytesListToBytesSet()));
transaction(new LettuceTxResult(
getConnection().zrangebylex(key, LettuceConverters.toRange(range), LettuceConverters.toLimit(limit)),
LettuceConverters.bytesListToBytesSet()));
} else {
transaction(
new LettuceTxResult(getConnection().zrangebylex(key, min, max), LettuceConverters.bytesListToBytesSet()));
transaction(new LettuceTxResult(getConnection().zrangebylex(key, LettuceConverters.toRange(range)),
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, LettuceConverters.toRange(range), LettuceConverters.toLimit(limit)));
}
return LettuceConverters.bytesListToBytesSet().convert(getConnection().zrangebylex(key, min, max));
return LettuceConverters.bytesListToBytesSet()
.convert(getConnection().zrangebylex(key, LettuceConverters.toRange(range)));
} catch (Exception ex) {
throw convertLettuceAccessException(ex);
}

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

@@ -16,18 +16,9 @@
package org.springframework.data.redis.connection.lettuce;
import java.nio.ByteBuffer;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collection;
import java.util.Collections;
import java.util.Date;
import java.util.Iterator;
import java.util.LinkedHashMap;
import java.util.LinkedHashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.*;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;
import org.springframework.core.convert.converter.Converter;
import org.springframework.dao.DataAccessException;
@@ -51,6 +42,7 @@ import org.springframework.data.redis.connection.RedisNode.NodeType;
import org.springframework.data.redis.connection.RedisSentinelConfiguration;
import org.springframework.data.redis.connection.RedisServer;
import org.springframework.data.redis.connection.RedisStringCommands.SetOption;
import org.springframework.data.redis.connection.RedisZSetCommands;
import org.springframework.data.redis.connection.RedisZSetCommands.Range.Boundary;
import org.springframework.data.redis.connection.RedisZSetCommands.Tuple;
import org.springframework.data.redis.connection.ReturnType;
@@ -67,15 +59,7 @@ import org.springframework.util.CollectionUtils;
import org.springframework.util.ObjectUtils;
import org.springframework.util.StringUtils;
import com.lambdaworks.redis.GeoArgs;
import com.lambdaworks.redis.GeoCoordinates;
import com.lambdaworks.redis.GeoWithin;
import com.lambdaworks.redis.KeyValue;
import com.lambdaworks.redis.RedisURI;
import com.lambdaworks.redis.ScoredValue;
import com.lambdaworks.redis.ScriptOutputType;
import com.lambdaworks.redis.SetArgs;
import com.lambdaworks.redis.SortArgs;
import com.lambdaworks.redis.*;
import com.lambdaworks.redis.cluster.models.partitions.Partitions;
import com.lambdaworks.redis.cluster.models.partitions.RedisClusterNode.NodeFlag;
import com.lambdaworks.redis.protocol.LettuceCharsets;
@@ -111,6 +95,9 @@ abstract public class LettuceConverters extends Converters {
private static final Converter<List<byte[]>, Long> BYTES_LIST_TO_TIME_CONVERTER;
private static final Converter<GeoCoordinates, Point> GEO_COORDINATE_TO_POINT_CONVERTER;
private static final ListConverter<GeoCoordinates, Point> GEO_COORDINATE_LIST_TO_POINT_LIST_CONVERTER;
private static final Converter<KeyValue<Object, Object>, Object> KEY_VALUE_UNWRAPPER;
private static final ListConverter<KeyValue<Object, Object>, Object> KEY_VALUE_LIST_UNWRAPPER;
private static final Converter<TransactionResult, List<Object>> TRANSACTION_RESULT_UNWRAPPER;
public static final byte[] PLUS_BYTES;
public static final byte[] MINUS_BYTES;
@@ -167,8 +154,8 @@ abstract public class LettuceConverters extends Converters {
return null;
}
List<byte[]> list = new ArrayList<byte[]>(2);
list.add(source.key);
list.add(source.value);
list.add(source.getKey());
list.add(source.getValue());
return list;
}
};
@@ -218,7 +205,7 @@ abstract public class LettuceConverters extends Converters {
};
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;
return source != null ? new DefaultTuple(source.getValue(), Double.valueOf(source.getScore())) : null;
}
};
BYTES_LIST_TO_TUPLE_LIST_CONVERTER = new Converter<List<byte[]>, List<Tuple>>() {
@@ -329,12 +316,29 @@ abstract public class LettuceConverters extends Converters {
GEO_COORDINATE_TO_POINT_CONVERTER = new Converter<com.lambdaworks.redis.GeoCoordinates, Point>() {
@Override
public Point convert(com.lambdaworks.redis.GeoCoordinates geoCoordinate) {
return geoCoordinate != null ? new Point(geoCoordinate.x.doubleValue(), geoCoordinate.y.doubleValue()) : null;
return geoCoordinate != null ? new Point(geoCoordinate.getX().doubleValue(), geoCoordinate.getY().doubleValue()) : null;
}
};
GEO_COORDINATE_LIST_TO_POINT_LIST_CONVERTER = new ListConverter<GeoCoordinates, Point>(
GEO_COORDINATE_TO_POINT_CONVERTER);
KEY_VALUE_UNWRAPPER = new Converter<KeyValue<Object, Object>, Object>() {
@Override
public Object convert(KeyValue<Object, Object> source) {
return source.getValueOrElse(null);
}
};
KEY_VALUE_LIST_UNWRAPPER = new ListConverter<>(KEY_VALUE_UNWRAPPER);
TRANSACTION_RESULT_UNWRAPPER = new Converter<TransactionResult, List<Object>>() {
@Override
public List<Object> convert(TransactionResult transactionResult) {
return transactionResult.stream().collect(Collectors.toList());
}
};
}
public static List<Tuple> toTuple(List<byte[]> list) {
@@ -541,6 +545,81 @@ abstract public class LettuceConverters extends Converters {
return prefix + value;
}
/**
* Convert a {@link org.springframework.data.redis.connection.RedisZSetCommands.Limit} to a lettuce {@link com.lambdaworks.redis.Limit}.
*
* @param limit
* @return a lettuce {@link com.lambdaworks.redis.Limit}.
* @since 2.0
*/
public static com.lambdaworks.redis.Limit toLimit(RedisZSetCommands.Limit limit){
return Limit.create(limit.getOffset(), limit.getCount());
}
/**
* Convert a {@link org.springframework.data.redis.connection.RedisZSetCommands.Range} to a lettuce {@link Range}.
*
* @param range
* @return
* @since 2.0
*/
public static <T> Range<T> toRange(org.springframework.data.redis.connection.RedisZSetCommands.Range range) {
return Range.from(lowerBoundaryOf(range), upperBoundaryOf(range));
}
/**
* Convert a {@link org.springframework.data.redis.connection.RedisZSetCommands.Range} to a lettuce {@link Range} and
* reverse boundaries.
*
* @param range
* @return
* @since 2.0
*/
public static <T> Range<T> toRevRange(org.springframework.data.redis.connection.RedisZSetCommands.Range range) {
return Range.from(upperBoundaryOf(range), lowerBoundaryOf(range));
}
@SuppressWarnings("unchecked")
private static <T> Range.Boundary<T> lowerBoundaryOf(org.springframework.data.redis.connection.RedisZSetCommands.Range range) {
return (Range.Boundary<T>) rangeToBoundaryArgumentConverter(false).convert(range);
}
@SuppressWarnings("unchecked")
private static <T> Range.Boundary<T> upperBoundaryOf(org.springframework.data.redis.connection.RedisZSetCommands.Range range) {
return (Range.Boundary<T>) rangeToBoundaryArgumentConverter(true).convert(range);
}
private static Converter<org.springframework.data.redis.connection.RedisZSetCommands.Range, Range.Boundary<?>> rangeToBoundaryArgumentConverter(
boolean upper) {
return (source) -> {
Boundary sourceBoundary = upper ? source.getMax() : source.getMin();
if (sourceBoundary == null || sourceBoundary.getValue() == null) {
return Range.Boundary.unbounded();
}
boolean inclusive = sourceBoundary.isIncluding();
Object value = sourceBoundary.getValue();
if (value instanceof Number) {
return inclusive ? Range.Boundary.including((Number) value) : Range.Boundary.excluding((Number) value);
}
if (value instanceof String) {
if (!StringUtils.hasText((String) value) || ObjectUtils.nullSafeEquals(value, "+")
|| ObjectUtils.nullSafeEquals(value, "-")) {
return Range.Boundary.unbounded();
}
return inclusive ? Range.Boundary.including(value.toString().getBytes(LettuceCharsets.UTF8))
: Range.Boundary.excluding(value.toString().getBytes(LettuceCharsets.UTF8));
}
return inclusive ? Range.Boundary.including((byte[]) value) : Range.Boundary.excluding((byte[]) value);
};
}
/**
* @param source List of Maps containing node details from SENTINEL SLAVES or SENTINEL MASTERS. May be empty or
* {@literal null}.
@@ -820,6 +899,19 @@ abstract public class LettuceConverters extends Converters {
return GEO_COORDINATE_LIST_TO_POINT_LIST_CONVERTER;
}
/**
* @return
* @since 2.0
*/
@SuppressWarnings("unchecked")
public static <K, V> ListConverter<KeyValue<K, V>, V> keyValueListUnwrapper() {
return (ListConverter) KEY_VALUE_LIST_UNWRAPPER;
}
public static Converter<TransactionResult, List<Object>> transactionResultUnwrapper() {
return TRANSACTION_RESULT_UNWRAPPER;
}
/**
* @author Christoph Strobl
* @since 1.8
@@ -881,10 +973,10 @@ abstract public class LettuceConverters extends Converters {
@Override
public GeoResult<GeoLocation<byte[]>> convert(GeoWithin<byte[]> source) {
Point point = GEO_COORDINATE_TO_POINT_CONVERTER.convert(source.coordinates);
Point point = GEO_COORDINATE_TO_POINT_CONVERTER.convert(source.getCoordinates());
return new GeoResult<GeoLocation<byte[]>>(new GeoLocation<byte[]>(source.member, point),
new Distance(source.distance != null ? source.distance : 0D, metric));
return new GeoResult<GeoLocation<byte[]>>(new GeoLocation<byte[]>(source.getMember(), point),
new Distance(source.getDistance() != null ? source.getDistance() : 0D, metric));
}
}
}

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

@@ -28,13 +28,14 @@ import org.springframework.data.redis.connection.RedisClusterNode;
import org.springframework.util.Assert;
import com.lambdaworks.redis.RedisException;
import com.lambdaworks.redis.api.reactive.RedisKeyReactiveCommands;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import rx.Observable;
/**
* @author Christoph Strobl.
* @author Christoph Strobl
* @author Mark Paluch
* @since 2.0
*/
public class LettuceReactiveClusterKeyCommands extends LettuceReactiveKeyCommands
@@ -60,19 +61,14 @@ public class LettuceReactiveClusterKeyCommands extends LettuceReactiveKeyCommand
Assert.notNull(pattern, "Pattern must not be null!");
Observable<List<ByteBuffer>> result = cmd.keys(pattern.array()).map(ByteBuffer::wrap).toList();
return Flux.from(LettuceReactiveRedisConnection.<List<ByteBuffer>> monoConverter().convert(result));
return cmd.keys(pattern).collectList();
}).next();
}
@Override
public Mono<ByteBuffer> randomKey(RedisClusterNode node) {
return connection.execute(node, cmd -> {
Observable<ByteBuffer> result = cmd.randomkey().map(ByteBuffer::wrap);
return Flux.from(LettuceReactiveRedisConnection.<ByteBuffer> monoConverter().convert(result));
}).next();
return connection.execute(node, RedisKeyReactiveCommands::randomkey).next();
}
/*
@@ -91,15 +87,13 @@ public class LettuceReactiveClusterKeyCommands extends LettuceReactiveKeyCommand
return super.rename(Mono.just(command));
}
Observable<Boolean> result = cmd.dump(command.getKey().array())
.switchIfEmpty(Observable.error(new RedisSystemException("Cannot rename key that does not exist",
Flux<Boolean> result = cmd.dump(command.getKey())
.otherwiseIfEmpty(Mono.error(new RedisSystemException("Cannot rename key that does not exist",
new RedisException("ERR no such key."))))
.concatMap(value -> cmd.restore(command.getNewName().array(), 0, value)
.concatMap(res -> cmd.del(command.getKey().array())))
.flatMap(value -> cmd.restore(command.getNewName(), 0, value).flatMap(res -> cmd.del(command.getKey())))
.map(LettuceConverters.longToBooleanConverter()::convert);
return LettuceReactiveRedisConnection.<Boolean> monoConverter().convert(result)
.map(val -> new ReactiveRedisConnection.BooleanResponse<>(command, val));
return result.map(val -> new ReactiveRedisConnection.BooleanResponse<>(command, val));
}));
}
@@ -119,25 +113,20 @@ public class LettuceReactiveClusterKeyCommands extends LettuceReactiveKeyCommand
return super.renameNX(Mono.just(command));
}
Observable<Boolean> result =
Flux<Boolean> result = cmd.exists(command.getNewName()).flatMap(exists -> {
cmd.exists(command.getNewName().array()).concatMap(exists -> {
if (exists == 1) {
return Mono.just(Boolean.FALSE);
}
if (exists == 1) {
return Observable.just(Boolean.FALSE);
}
return cmd.dump(command.getKey())
.otherwiseIfEmpty(Mono.error(new RedisSystemException("Cannot rename key that does not exist",
new RedisException("ERR no such key."))))
.flatMap(value -> cmd.restore(command.getNewName(), 0, value).flatMap(res -> cmd.del(command.getKey())))
.map(LettuceConverters.longToBooleanConverter()::convert);
});
return cmd.dump(command.getKey().array())
.switchIfEmpty(Observable.error(new RedisSystemException("Cannot rename key that does not exist",
new RedisException("ERR no such key."))))
.concatMap(value -> cmd.restore(command.getNewName().array(), 0, value)
.concatMap(res -> cmd.del(command.getKey().array())))
.map(LettuceConverters.longToBooleanConverter()::convert);
});
return LettuceReactiveRedisConnection.<Boolean> monoConverter().convert(result)
.map(val -> new ReactiveRedisConnection.BooleanResponse<>(command, val));
return result.map(val -> new ReactiveRedisConnection.BooleanResponse<>(command, val));
}));
}
}

10
src/main/java/org/springframework/data/redis/connection/lettuce/LettuceReactiveClusterListCommands.java
View File

@@ -27,10 +27,10 @@ import org.springframework.util.Assert;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import rx.Observable;
/**
* @author Christoph Strobl
* @author Mark Paluch
* @since 2.0
*/
public class LettuceReactiveClusterListCommands extends LettuceReactiveListCommands
@@ -74,12 +74,10 @@ public class LettuceReactiveClusterListCommands extends LettuceReactiveListComma
return super.rPopLPush(Mono.just(command));
}
Observable<ByteBuffer> result = cmd.rpop(command.getKey().array())
.concatMap(value -> cmd.lpush(command.getDestination().array(), value).map(x -> value)).map(ByteBuffer::wrap);
Flux<ByteBuffer> result = cmd.rpop(command.getKey())
.flatMap(value -> cmd.lpush(command.getDestination(), value).map(x -> value));
return LettuceReactiveRedisConnection.<ByteBuffer> monoConverter().convert(result)
.map(value -> new ReactiveRedisConnection.ByteBufferResponse<>(command, value));
return result.map(value -> new ReactiveRedisConnection.ByteBufferResponse<>(command, value));
}));
}
}

92
src/main/java/org/springframework/data/redis/connection/lettuce/LettuceReactiveClusterSetCommands.java
View File

@@ -29,10 +29,10 @@ import org.springframework.util.Assert;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import rx.Observable;
/**
* @author Christoph Strobl
* @author Mark Paluch
* @since 2.0
*/
public class LettuceReactiveClusterSetCommands extends LettuceReactiveSetCommands
@@ -59,12 +59,10 @@ public class LettuceReactiveClusterSetCommands extends LettuceReactiveSetCommand
return super.sUnion(Mono.just(command));
}
Observable<List<ByteBuffer>> result = Observable
.merge(command.getKeys().stream().map(key -> cmd.smembers(key.array())).collect(Collectors.toList()))
.map(ByteBuffer::wrap).distinct().toList();
Mono<List<ByteBuffer>> result = Flux
.merge(command.getKeys().stream().map(cmd::smembers).collect(Collectors.toList())).distinct().collectList();
return LettuceReactiveRedisConnection.<List<ByteBuffer>> monoConverter().convert(result)
.map(value -> new ReactiveRedisConnection.MultiValueResponse<>(command, value));
return result.map(value -> new ReactiveRedisConnection.MultiValueResponse<>(command, value));
}));
}
@@ -85,10 +83,8 @@ public class LettuceReactiveClusterSetCommands extends LettuceReactiveSetCommand
}
return sUnion(Mono.just(SUnionCommand.keys(command.getKeys()))).next().flatMap(values -> {
Observable<Long> result = cmd.sadd(command.getKey().array(),
values.getOutput().stream().map(ByteBuffer::array).toArray(size -> new byte[size][]));
return LettuceReactiveRedisConnection.<Long> monoConverter().convert(result)
.map(value -> new ReactiveRedisConnection.NumericResponse<>(command, value));
Mono<Long> result = cmd.sadd(command.getKey(), values.getOutput().stream().toArray(ByteBuffer[]::new));
return result.map(value -> new ReactiveRedisConnection.NumericResponse<>(command, value));
});
}));
}
@@ -105,24 +101,21 @@ public class LettuceReactiveClusterSetCommands extends LettuceReactiveSetCommand
return super.sInter(Mono.just(command));
}
Observable<List<ByteBuffer>> sourceSet = cmd.smembers(command.getKeys().get(0).array()).map(ByteBuffer::wrap)
.distinct().toList();
Mono<List<ByteBuffer>> sourceSet = cmd.smembers(command.getKeys().get(0)).distinct().collectList();
List<Observable<List<ByteBuffer>>> intersectingSets = new ArrayList<>();
List<Mono<List<ByteBuffer>>> intersectingSets = new ArrayList<>();
for (int i = 1; i < command.getKeys().size(); i++) {
intersectingSets.add(cmd.smembers(command.getKeys().get(i).array()).map(ByteBuffer::wrap).distinct().toList());
intersectingSets.add(cmd.smembers(command.getKeys().get(i)).distinct().collectList());
}
Observable<List<ByteBuffer>> result = Observable.zip(sourceSet, Observable.merge(intersectingSets),
(source, intersecting) -> {
Flux<List<ByteBuffer>> result = Flux.zip(sourceSet, Flux.merge(intersectingSets), (source, intersecting) -> {
source.retainAll(intersecting);
return source;
});
source.retainAll(intersecting);
return source;
});
return LettuceReactiveRedisConnection.<List<ByteBuffer>> monoConverter().convert(result)
.map(value -> new ReactiveRedisConnection.MultiValueResponse<>(command, value));
return result.map(value -> new ReactiveRedisConnection.MultiValueResponse<>(command, value));
}));
}
@@ -143,10 +136,8 @@ public class LettuceReactiveClusterSetCommands extends LettuceReactiveSetCommand
}
return sInter(Mono.just(SInterCommand.keys(command.getKeys()))).next().flatMap(values -> {
Observable<Long> result = cmd.sadd(command.getKey().array(),
values.getOutput().stream().map(ByteBuffer::array).toArray(size -> new byte[size][]));
return LettuceReactiveRedisConnection.<Long> monoConverter().convert(result)
.map(value -> new ReactiveRedisConnection.NumericResponse<>(command, value));
Mono<Long> result = cmd.sadd(command.getKey(), values.getOutput().stream().toArray(ByteBuffer[]::new));
return result.map(value -> new ReactiveRedisConnection.NumericResponse<>(command, value));
});
}));
}
@@ -163,24 +154,21 @@ public class LettuceReactiveClusterSetCommands extends LettuceReactiveSetCommand
return super.sDiff(Mono.just(command));
}
Observable<List<ByteBuffer>> sourceSet = cmd.smembers(command.getKeys().get(0).array()).map(ByteBuffer::wrap)
.distinct().toList();
Mono<List<ByteBuffer>> sourceSet = cmd.smembers(command.getKeys().get(0)).distinct().collectList();
List<Observable<List<ByteBuffer>>> intersectingSets = new ArrayList<>();
List<Mono<List<ByteBuffer>>> intersectingSets = new ArrayList<>();
for (int i = 1; i < command.getKeys().size(); i++) {
intersectingSets.add(cmd.smembers(command.getKeys().get(i).array()).map(ByteBuffer::wrap).distinct().toList());
intersectingSets.add(cmd.smembers(command.getKeys().get(i)).distinct().collectList());
}
Observable<List<ByteBuffer>> result = Observable.zip(sourceSet, Observable.merge(intersectingSets),
(source, intersecting) -> {
Flux<List<ByteBuffer>> result = Flux.zip(sourceSet, Flux.merge(intersectingSets), (source, intersecting) -> {
source.removeAll(intersecting);
return source;
});
source.removeAll(intersecting);
return source;
});
return LettuceReactiveRedisConnection.<List<ByteBuffer>> monoConverter().convert(result)
.map(value -> new ReactiveRedisConnection.MultiValueResponse<>(command, value));
return result.map(value -> new ReactiveRedisConnection.MultiValueResponse<>(command, value));
}));
}
@@ -202,10 +190,8 @@ public class LettuceReactiveClusterSetCommands extends LettuceReactiveSetCommand
}
return sDiff(Mono.just(SDiffCommand.keys(command.getKeys()))).next().flatMap(values -> {
Observable<Long> result = cmd.sadd(command.getKey().array(),
values.getOutput().stream().map(ByteBuffer::array).toArray(size -> new byte[size][]));
return LettuceReactiveRedisConnection.<Long> monoConverter().convert(result)
.map(value -> new ReactiveRedisConnection.NumericResponse<>(command, value));
Mono<Long> result = cmd.sadd(command.getKey(), values.getOutput().stream().toArray(ByteBuffer[]::new));
return result.map(value -> new ReactiveRedisConnection.NumericResponse<>(command, value));
});
}));
}
@@ -222,28 +208,26 @@ public class LettuceReactiveClusterSetCommands extends LettuceReactiveSetCommand
return super.sMove(Mono.just(command));
}
Observable<Boolean> result = cmd.exists(command.getKey().array()).flatMap(nrKeys -> nrKeys == 0
? Observable.empty() : cmd.sismember(command.getKey().array(), command.getValue().array()))
Flux<Boolean> result = cmd.exists(command.getKey())
.flatMap(nrKeys -> nrKeys == 0 ? Mono.empty() : cmd.sismember(command.getKey(), command.getValue()))
.flatMap(exists -> {
if (!exists) {
return Observable.just(Boolean.FALSE);
return Mono.just(Boolean.FALSE);
}
return cmd.sismember(command.getDestination().array(), command.getValue().array())
.flatMap(existsInTarget -> {
return cmd.sismember(command.getDestination(), command.getValue()).flatMap(existsInTarget -> {
Observable<Boolean> tmp = cmd.srem(command.getKey().array(), command.getValue().array())
.map(nrRemoved -> nrRemoved > 0);
if (!existsInTarget) {
return tmp.flatMap(removed -> cmd.sadd(command.getDestination().array(), command.getValue().array())
.map(LettuceConverters::toBoolean));
}
return tmp;
});
Mono<Boolean> tmp = cmd.srem(command.getKey(), command.getValue()).map(nrRemoved -> nrRemoved > 0);
if (!existsInTarget) {
return tmp.flatMap(removed -> cmd.sadd(command.getDestination(), command.getValue())
.map(LettuceConverters::toBoolean));
}
return tmp;
});
});
return LettuceReactiveRedisConnection.<Boolean> monoConverter().convert(result.defaultIfEmpty(Boolean.FALSE))
return result.defaultIfEmpty(Boolean.FALSE)
.map(value -> new ReactiveRedisConnection.BooleanResponse<>(command, value));
}));
}

74
src/main/java/org/springframework/data/redis/connection/lettuce/LettuceReactiveGeoCommands.java
View File

@@ -18,6 +18,7 @@ package org.springframework.data.redis.connection.lettuce;
import java.nio.ByteBuffer;
import java.util.ArrayList;
import java.util.List;
import java.util.stream.Collectors;
import org.reactivestreams.Publisher;
import org.springframework.core.convert.converter.Converter;
@@ -35,10 +36,12 @@ import org.springframework.data.redis.connection.RedisGeoCommands.GeoLocation;
import org.springframework.util.Assert;
import com.lambdaworks.redis.GeoArgs;
import com.lambdaworks.redis.GeoCoordinates;
import com.lambdaworks.redis.GeoWithin;
import com.lambdaworks.redis.Value;
import reactor.core.publisher.Flux;
import rx.Observable;
import reactor.core.publisher.Mono;
/**
* @author Christoph Strobl
@@ -79,12 +82,10 @@ public class LettuceReactiveGeoCommands implements ReactiveGeoCommands {
values.add(location.getPoint().getX());
values.add(location.getPoint().getY());
values.add(location.getName().array());
values.add(location.getName());
}
return LettuceReactiveRedisConnection.<Long> monoConverter()
.convert(cmd.geoadd(command.getKey().array(), values.toArray()))
.map(value -> new NumericResponse<>(command, value));
return cmd.geoadd(command.getKey(), values.toArray()).map(value -> new NumericResponse<>(command, value));
}));
}
@@ -101,17 +102,16 @@ public class LettuceReactiveGeoCommands implements ReactiveGeoCommands {
Assert.notNull(command.getFrom(), "From member must not be null!");
Assert.notNull(command.getTo(), "To member must not be null!");
Metric metric = command.getMetric().isPresent() ? command.getMetric().get() : RedisGeoCommands.DistanceUnit.METERS;
Metric metric = command.getMetric().isPresent() ? command.getMetric().get()
: RedisGeoCommands.DistanceUnit.METERS;
GeoArgs.Unit geoUnit = LettuceConverters.toGeoArgsUnit(metric);
Converter<Double, Distance> distanceConverter = LettuceConverters.distanceConverterForMetric(metric);
Observable<Distance> result = cmd
.geodist(command.getKey().array(), command.getFrom().array(), command.getTo().array(), geoUnit)
Mono<Distance> result = cmd.geodist(command.getKey(), command.getFrom(), command.getTo(), geoUnit)
.map(distanceConverter::convert);
return LettuceReactiveRedisConnection.<Distance> monoConverter().convert(result)
.map(value -> new CommandResponse<>(command, value));
return result.map(value -> new CommandResponse<>(command, value));
}));
}
@@ -127,10 +127,9 @@ public class LettuceReactiveGeoCommands implements ReactiveGeoCommands {
Assert.notNull(command.getKey(), "Key must not be null!");
Assert.notNull(command.getMembers(), "Members must not be null!");
return LettuceReactiveRedisConnection.<List<String>> monoConverter()
.convert(cmd.geohash(command.getKey().array(),
command.getMembers().stream().map(ByteBuffer::array).toArray(size -> new byte[size][])).toList())
.map(value -> new MultiValueResponse<>(command, value));
return cmd.geohash(command.getKey(), command.getMembers().stream().toArray(ByteBuffer[]::new)).collectList()
.map(value -> new MultiValueResponse<>(command,
value.stream().map(v -> v.getValueOrElse(null)).collect(Collectors.toList())));
}));
}
@@ -146,13 +145,11 @@ public class LettuceReactiveGeoCommands implements ReactiveGeoCommands {
Assert.notNull(command.getKey(), "Key must not be null!");
Assert.notNull(command.getMembers(), "Members must not be null!");
Observable<List<Point>> result = cmd
.geopos(command.getKey().array(),
command.getMembers().stream().map(ByteBuffer::array).toArray(size -> new byte[size][]))
.map(LettuceConverters::geoCoordinatesToPoint).toList();
Mono<List<Value<GeoCoordinates>>> result = cmd
.geopos(command.getKey(), command.getMembers().stream().toArray(ByteBuffer[]::new)).collectList();
return LettuceReactiveRedisConnection.<List<Point>> monoConverter().convert(result)
.map(value -> new MultiValueResponse<>(command, value));
return result.map(value -> new MultiValueResponse<>(command, value.stream().map(v -> v.getValueOrElse(null))
.map(LettuceConverters::geoCoordinatesToPoint).collect(Collectors.toList())));
}));
}
@@ -170,16 +167,15 @@ public class LettuceReactiveGeoCommands implements ReactiveGeoCommands {
Assert.notNull(command.getPoint(), "Point must not be null!");
Assert.notNull(command.getDistance(), "Distance must not be null!");
GeoArgs geoArgs = command.getArgs().isPresent() ? LettuceConverters.toGeoArgs(command.getArgs().get()) : new GeoArgs();
GeoArgs geoArgs = command.getArgs().isPresent() ? LettuceConverters.toGeoArgs(command.getArgs().get())
: new GeoArgs();
Observable<GeoResults<GeoLocation<ByteBuffer>>> result = cmd
.georadius(command.getKey().array(), command.getPoint().getX(), command.getPoint().getY(),
command.getDistance().getValue(), LettuceConverters.toGeoArgsUnit(command.getDistance().getMetric()),
geoArgs)
.map(converter(command.getDistance().getMetric())::convert).toList().map(vals -> new GeoResults<>(vals));
Mono<GeoResults<GeoLocation<ByteBuffer>>> result = cmd.georadius(command.getKey(), command.getPoint().getX(),
command.getPoint().getY(), command.getDistance().getValue(),
LettuceConverters.toGeoArgsUnit(command.getDistance().getMetric()), geoArgs)
.map(converter(command.getDistance().getMetric())::convert).collectList().map(GeoResults::new);
return LettuceReactiveRedisConnection.<GeoResults<GeoLocation<ByteBuffer>>> monoConverter().convert(result)
.map(value -> new CommandResponse<>(command, value));
return result.map(value -> new CommandResponse<>(command, value));
}));
}
@@ -197,28 +193,26 @@ public class LettuceReactiveGeoCommands implements ReactiveGeoCommands {
Assert.notNull(command.getMember(), "Member must not be null!");
Assert.notNull(command.getDistance(), "Distance must not be null!");
GeoArgs geoArgs = command.getArgs().isPresent() ? LettuceConverters.toGeoArgs(command.getArgs().get()) : new GeoArgs();
GeoArgs geoArgs = command.getArgs().isPresent() ? LettuceConverters.toGeoArgs(command.getArgs().get())
: new GeoArgs();
Observable<GeoResults<GeoLocation<ByteBuffer>>> result = cmd
.georadiusbymember(command.getKey().array(), command.getMember().array(), command.getDistance().getValue(),
Mono<GeoResults<GeoLocation<ByteBuffer>>> result = cmd
.georadiusbymember(command.getKey(), command.getMember(), command.getDistance().getValue(),
LettuceConverters.toGeoArgsUnit(command.getDistance().getMetric()), geoArgs)
.map(converter(command.getDistance().getMetric())::convert).toList().map(vals -> new GeoResults<>(vals));
.map(converter(command.getDistance().getMetric())::convert).collectList().map(GeoResults::new);
return LettuceReactiveRedisConnection.<GeoResults<GeoLocation<ByteBuffer>>> monoConverter().convert(result)
.map(value -> new CommandResponse<>(command, value));
return result.map(value -> new CommandResponse<>(command, value));
}));
}
private Converter<GeoWithin<byte[]>, GeoResult<GeoLocation<ByteBuffer>>> converter(Metric metric) {
private Converter<GeoWithin<ByteBuffer>, GeoResult<GeoLocation<ByteBuffer>>> converter(Metric metric) {
return (source) -> {
Point point = LettuceConverters.geoCoordinatesToPoint(source.coordinates);
Point point = LettuceConverters.geoCoordinatesToPoint(source.getCoordinates());
return new GeoResult<>(new GeoLocation<>(ByteBuffer.wrap(source.member), point),
new Distance(source.distance != null ? source.distance : 0D, metric));
return new GeoResult<>(new GeoLocation<>(source.getMember(), point),
new Distance(source.getDistance() != null ? source.getDistance() : 0D, metric));
};
}
}

62
src/main/java/org/springframework/data/redis/connection/lettuce/LettuceReactiveHashCommands.java
View File

@@ -32,11 +32,14 @@ import org.springframework.data.redis.connection.ReactiveRedisConnection.Numeric
import org.springframework.util.Assert;
import org.springframework.util.ObjectUtils;
import com.lambdaworks.redis.KeyValue;
import reactor.core.publisher.Flux;
import rx.Observable;
import reactor.core.publisher.Mono;
/**
* @author Christoph Strobl
* @author Mark Paluch
* @since 2.0
*/
public class LettuceReactiveHashCommands implements ReactiveHashCommands {
@@ -66,25 +69,23 @@ public class LettuceReactiveHashCommands implements ReactiveHashCommands {
Assert.notNull(command.getKey(), "Key must not be null!");
Assert.notNull(command.getFieldValueMap(), "FieldValueMap must not be null!");
Observable<Boolean> result = Observable.empty();
Mono<Boolean> result;
if (command.getFieldValueMap().size() == 1) {
Entry<ByteBuffer, ByteBuffer> entry = command.getFieldValueMap().entrySet().iterator().next();
result = ObjectUtils.nullSafeEquals(command.isUpsert(), Boolean.TRUE)
? cmd.hset(command.getKey().array(), entry.getKey().array(), entry.getValue().array())
: cmd.hsetnx(command.getKey().array(), entry.getKey().array(), entry.getValue().array());
? cmd.hset(command.getKey(), entry.getKey(), entry.getValue())
: cmd.hsetnx(command.getKey(), entry.getKey(), entry.getValue());
} else {
Map<byte[], byte[]> entries = command.getFieldValueMap().entrySet().stream()
.collect(Collectors.toMap(e -> e.getKey().array(), e -> e.getValue().array()));
Map<ByteBuffer, ByteBuffer> entries = command.getFieldValueMap();
result = cmd.hmset(command.getKey().array(), entries).map(LettuceConverters::stringToBoolean);
result = cmd.hmset(command.getKey(), entries).map(LettuceConverters::stringToBoolean);
}
return LettuceReactiveRedisConnection.<Boolean> monoConverter().convert(result)
.map(value -> new BooleanResponse<>(command, value));
return result.map(value -> new BooleanResponse<>(command, value));
}));
}
@@ -100,20 +101,18 @@ public class LettuceReactiveHashCommands implements ReactiveHashCommands {
Assert.notNull(command.getKey(), "Key must not be null!");
Assert.notNull(command.getFields(), "Fields must not be null!");
Observable<List<ByteBuffer>> result = null;
Mono<List<KeyValue<ByteBuffer, ByteBuffer>>> result;
if (command.getFields().size() == 1) {
result = cmd.hget(command.getKey().array(), command.getFields().iterator().next().array()).map(ByteBuffer::wrap)
.map(val -> val != null ? Collections.singletonList(val) : Collections.emptyList());
ByteBuffer key = command.getFields().iterator().next();
result = cmd.hget(command.getKey(), key.duplicate()).map(value -> KeyValue.fromNullable(key, value))
.map(Collections::singletonList).otherwiseReturn(Collections.emptyList());
} else {
result = cmd
.hmget(command.getKey().array(),
command.getFields().stream().map(ByteBuffer::array).toArray(size -> new byte[size][]))
.map(val -> val != null ? ByteBuffer.wrap(val) : null).toList();
result = cmd.hmget(command.getKey(), command.getFields().stream().toArray(ByteBuffer[]::new)).collectList();
}
return LettuceReactiveRedisConnection.<List<ByteBuffer>> monoConverter().convert(result)
.map(value -> new MultiValueResponse<>(command, value));
return result.map(value -> new MultiValueResponse<>(command,
value.stream().map(keyValue -> keyValue.getValueOrElse(null)).collect(Collectors.toList())));
}));
}
@@ -129,9 +128,7 @@ public class LettuceReactiveHashCommands implements ReactiveHashCommands {
Assert.notNull(command.getKey(), "Key must not be null!");
Assert.notNull(command.getName(), "Name must not be null!");
return LettuceReactiveRedisConnection.<Boolean> monoConverter()
.convert(cmd.hexists(command.getKey().array(), command.getField().array()))
.map(value -> new BooleanResponse<>(command, value));
return cmd.hexists(command.getKey(), command.getField()).map(value -> new BooleanResponse<>(command, value));
}));
}
@@ -147,9 +144,7 @@ public class LettuceReactiveHashCommands implements ReactiveHashCommands {
Assert.notNull(command.getKey(), "Key must not be null!");
Assert.notNull(command.getFields(), "Fields must not be null!");
return LettuceReactiveRedisConnection.<Long> monoConverter()
.convert(cmd.hdel(command.getKey().array(),
command.getFields().stream().map(ByteBuffer::array).toArray(size -> new byte[size][])))
return cmd.hdel(command.getKey(), command.getFields().stream().toArray(ByteBuffer[]::new))
.map(value -> new NumericResponse<>(command, value));
}));
}
@@ -165,8 +160,7 @@ public class LettuceReactiveHashCommands implements ReactiveHashCommands {
Assert.notNull(command.getKey(), "Command.getKey() must not be null!");
return LettuceReactiveRedisConnection.<Long> monoConverter().convert(cmd.hlen(command.getKey().array()))
.map(value -> new NumericResponse<>(command, value));
return cmd.hlen(command.getKey()).map(value -> new NumericResponse<>(command, value));
}));
}
@@ -181,10 +175,9 @@ public class LettuceReactiveHashCommands implements ReactiveHashCommands {
Assert.notNull(command.getKey(), "Key must not be null!");
Observable<List<ByteBuffer>> result = cmd.hkeys(command.getKey().array()).map(ByteBuffer::wrap).toList();
Mono<List<ByteBuffer>> result = cmd.hkeys(command.getKey()).collectList();
return LettuceReactiveRedisConnection.<List<ByteBuffer>> monoConverter().convert(result)
.map(value -> new MultiValueResponse<>(command, value));
return result.map(value -> new MultiValueResponse<>(command, value));
}));
}
@@ -199,10 +192,9 @@ public class LettuceReactiveHashCommands implements ReactiveHashCommands {
Assert.notNull(command.getKey(), "Key must not be null!");
Observable<List<ByteBuffer>> result = cmd.hvals(command.getKey().array()).map(ByteBuffer::wrap).toList();
Mono<List<ByteBuffer>> result = cmd.hvals(command.getKey()).collectList();
return LettuceReactiveRedisConnection.<List<ByteBuffer>> monoConverter().convert(result)
.map(value -> new MultiValueResponse<>(command, value));
return result.map(value -> new MultiValueResponse<>(command, value));
}));
}
@@ -217,11 +209,9 @@ public class LettuceReactiveHashCommands implements ReactiveHashCommands {
Assert.notNull(command.getKey(), "Key must not be null!");
Observable<Map<ByteBuffer, ByteBuffer>> result = cmd.hgetall(command.getKey().array()).map(val -> val.entrySet()
.stream().collect(Collectors.toMap(e -> ByteBuffer.wrap(e.getKey()), e -> ByteBuffer.wrap(e.getValue()))));
Mono<Map<ByteBuffer, ByteBuffer>> result = cmd.hgetall(command.getKey());
return LettuceReactiveRedisConnection.<Map<ByteBuffer, ByteBuffer>> monoConverter().convert(result)
.map(value -> new CommandResponse<>(command, value));
return result.map(value -> new CommandResponse<>(command, value));
}));
}
}

15
src/main/java/org/springframework/data/redis/connection/lettuce/LettuceReactiveHyperLogLogCommands.java
View File

@@ -55,9 +55,7 @@ public class LettuceReactiveHyperLogLogCommands implements ReactiveHyperLogLogCo
Assert.notNull(command.getKey(), "key must not be null!");
return LettuceReactiveRedisConnection.<Long> monoConverter()
.convert(cmd.pfadd(command.getKey().array(),
command.getValues().stream().map(ByteBuffer::array).toArray(size -> new byte[size][])))
return cmd.pfadd(command.getKey(), command.getValues().stream().toArray(ByteBuffer[]::new))
.map(value -> new NumericResponse<>(command, value));
}));
@@ -74,8 +72,7 @@ public class LettuceReactiveHyperLogLogCommands implements ReactiveHyperLogLogCo
Assert.notEmpty(command.getKeys(), "Keys must not be empty for PFCOUNT.");
return LettuceReactiveRedisConnection.<Long> monoConverter()
.convert(cmd.pfcount(command.getKeys().stream().map(ByteBuffer::array).toArray(size -> new byte[size][])))
return cmd.pfcount(command.getKeys().stream().toArray(ByteBuffer[]::new))
.map(value -> new NumericResponse<>(command, value));
}));
}
@@ -92,12 +89,8 @@ public class LettuceReactiveHyperLogLogCommands implements ReactiveHyperLogLogCo
Assert.notNull(command.getKey(), "Destination key must not be null for PFMERGE.");
Assert.notEmpty(command.getSourceKeys(), "Source keys must not be null for PFMERGE.");
return LettuceReactiveRedisConnection.<Boolean> monoConverter()
.convert(cmd
.pfmerge(command.getKey().array(),
command.getSourceKeys().stream().map(ByteBuffer::array).toArray(size -> new byte[size][]))
.map(LettuceConverters::stringToBoolean))
.map(value -> new BooleanResponse<>(command, value));
return cmd.pfmerge(command.getKey(), command.getSourceKeys().stream().toArray(ByteBuffer[]::new))
.map(LettuceConverters::stringToBoolean).map(value -> new BooleanResponse<>(command, value));
}));
}

32
src/main/java/org/springframework/data/redis/connection/lettuce/LettuceReactiveKeyCommands.java
View File

@@ -19,6 +19,7 @@ import java.nio.ByteBuffer;
import java.util.List;
import java.util.stream.Collectors;
import com.lambdaworks.redis.api.reactive.RedisKeyReactiveCommands;
import org.reactivestreams.Publisher;
import org.springframework.data.redis.connection.DataType;
import org.springframework.data.redis.connection.ReactiveKeyCommands;
@@ -62,9 +63,8 @@ public class LettuceReactiveKeyCommands implements ReactiveKeyCommands {
Assert.notNull(command.getKey(), "Key must not be null!");
return LettuceReactiveRedisConnection.<BooleanResponse<KeyCommand>> monoConverter()
.convert(cmd.exists(command.getKey().array()).map(LettuceConverters.longToBooleanConverter()::convert)
.map((value) -> new BooleanResponse<>(command, value)));
return cmd.exists(command.getKey()).map(LettuceConverters.longToBooleanConverter()::convert)
.map((value) -> new BooleanResponse<>(command, value));
}));
}
@@ -79,8 +79,7 @@ public class LettuceReactiveKeyCommands implements ReactiveKeyCommands {
Assert.notNull(command.getKey(), "Key must not be null!");
return LettuceReactiveRedisConnection.<DataType> monoConverter()
.convert(cmd.type(command.getKey().array()).map(LettuceConverters::toDataType))
return cmd.type(command.getKey()).map(LettuceConverters::toDataType)
.map(respValue -> new CommandResponse<>(command, respValue));
}));
}
@@ -96,8 +95,7 @@ public class LettuceReactiveKeyCommands implements ReactiveKeyCommands {
Assert.notNull(command.getKey(), "Key must not be null!");
return LettuceReactiveRedisConnection.<NumericResponse<KeyCommand, Long>> monoConverter()
.convert(cmd.del(command.getKey().array()).map((value) -> new NumericResponse<>(command, value)));
return cmd.del(command.getKey()).map((value) -> new NumericResponse<>(command, value));
}));
}
@@ -112,10 +110,8 @@ public class LettuceReactiveKeyCommands implements ReactiveKeyCommands {
Assert.notEmpty(keys, "Keys must not be null!");
return LettuceReactiveRedisConnection.<NumericResponse<List<ByteBuffer>, Long>> monoConverter()
.convert(cmd
.del(keys.stream().map(ByteBuffer::array).collect(Collectors.toList()).toArray(new byte[keys.size()][]))
.map((value) -> new NumericResponse<>(keys, value)));
return cmd.del(keys.stream().collect(Collectors.toList()).toArray(new ByteBuffer[keys.size()]))
.map((value) -> new NumericResponse<>(keys, value));
}));
}
@@ -129,9 +125,7 @@ public class LettuceReactiveKeyCommands implements ReactiveKeyCommands {
Assert.notNull(pattern, "Pattern must not be null!");
return LettuceReactiveRedisConnection.<List<ByteBuffer>> monoConverter()
.convert(cmd.keys(pattern.array()).map(ByteBuffer::wrap).toList())
.map(value -> new MultiValueResponse<>(pattern, value));
return cmd.keys(pattern).collectList().map(value -> new MultiValueResponse<>(pattern, value));
}));
}
@@ -141,8 +135,7 @@ public class LettuceReactiveKeyCommands implements ReactiveKeyCommands {
*/
@Override
public Mono<ByteBuffer> randomKey() {
return connection.execute(cmd -> LettuceReactiveRedisConnection.<ByteBuffer> monoConverter()
.convert(cmd.randomkey().map(ByteBuffer::wrap))).next();
return connection.execute(RedisKeyReactiveCommands::randomkey).next();
}
/*
@@ -157,8 +150,7 @@ public class LettuceReactiveKeyCommands implements ReactiveKeyCommands {
Assert.notNull(command.getKey(), "Key must not be null!");
Assert.notNull(command.getNewName(), "New name must not be null!");
return LettuceReactiveRedisConnection.<Boolean> monoConverter().convert(
cmd.rename(command.getKey().array(), command.getNewName().array()).map(LettuceConverters::stringToBoolean))
return cmd.rename(command.getKey(), command.getNewName()).map(LettuceConverters::stringToBoolean)
.map(value -> new BooleanResponse<>(command, value));
}));
}
@@ -175,9 +167,7 @@ public class LettuceReactiveKeyCommands implements ReactiveKeyCommands {
Assert.notNull(command.getKey(), "Key must not be null!");
Assert.notNull(command.getNewName(), "New name must not be null!");
return LettuceReactiveRedisConnection.<Boolean> monoConverter()
.convert(cmd.renamenx(command.getKey().array(), command.getNewName().array()))
.map(value -> new BooleanResponse<>(command, value));
return cmd.renamenx(command.getKey(), command.getNewName()).map(value -> new BooleanResponse<>(command, value));
}));
}
}

84
src/main/java/org/springframework/data/redis/connection/lettuce/LettuceReactiveListCommands.java
View File

@@ -18,7 +18,6 @@ package org.springframework.data.redis.connection.lettuce;
import java.nio.ByteBuffer;
import java.time.temporal.ChronoUnit;
import java.util.Arrays;
import java.util.List;
import org.reactivestreams.Publisher;
import org.springframework.dao.InvalidDataAccessApiUsageException;
@@ -34,10 +33,11 @@ import org.springframework.util.Assert;
import org.springframework.util.ObjectUtils;
import reactor.core.publisher.Flux;
import rx.Observable;
import reactor.core.publisher.Mono;
/**
* @author Christoph Strobl
* @author Mark Paluch
* @since 2.0
*/
public class LettuceReactiveListCommands implements ReactiveListCommands {
@@ -72,20 +72,19 @@ public class LettuceReactiveListCommands implements ReactiveListCommands {
String.format("%s PUSHX only allows one value!", command.getDirection()));
}
byte[][] values = command.getValues().stream().map(ByteBuffer::array).toArray(size -> new byte[size][]);
Observable<Long> pushResult = null;
Mono<Long> pushResult = null;
if (ObjectUtils.nullSafeEquals(Direction.RIGHT, command.getDirection())) {
pushResult = command.getUpsert() ? cmd.rpush(command.getKey().array(), values)
: cmd.rpushx(command.getKey().array(), values[0]);
pushResult = command.getUpsert()
? cmd.rpush(command.getKey(), command.getValues().stream().toArray(ByteBuffer[]::new))
: cmd.rpushx(command.getKey(), command.getValues().get(0));
} else {
pushResult = command.getUpsert() ? cmd.lpush(command.getKey().array(), values)
: cmd.lpushx(command.getKey().array(), values[0]);
pushResult = command.getUpsert()
? cmd.lpush(command.getKey(), command.getValues().stream().toArray(ByteBuffer[]::new))
: cmd.lpushx(command.getKey(), command.getValues().get(0));
}
return LettuceReactiveRedisConnection.<Long> monoConverter().convert(pushResult)
.map(value -> new NumericResponse<>(command, value));
return pushResult.map(value -> new NumericResponse<>(command, value));
}));
}
@@ -100,8 +99,7 @@ public class LettuceReactiveListCommands implements ReactiveListCommands {
Assert.notNull(command.getKey(), "Key must not be null!");
return LettuceReactiveRedisConnection.<Long> monoConverter().convert(cmd.llen(command.getKey().array()))
.map(value -> new NumericResponse<>(command, value));
return cmd.llen(command.getKey()).map(value -> new NumericResponse<>(command, value));
}));
}
@@ -117,11 +115,8 @@ public class LettuceReactiveListCommands implements ReactiveListCommands {
Assert.notNull(command.getKey(), "Key must not be null!");
Assert.notNull(command.getRange(), "Range must not be null!");
return LettuceReactiveRedisConnection.<List<ByteBuffer>> monoConverter()
.convert(cmd
.lrange(command.getKey().array(), command.getRange().getLowerBound(), command.getRange().getUpperBound())
.map(ByteBuffer::wrap).toList())
.map(value -> new MultiValueResponse<>(command, value));
return cmd.lrange(command.getKey(), command.getRange().getLowerBound(), command.getRange().getUpperBound())
.collectList().map(value -> new MultiValueResponse<>(command, value));
}));
}
@@ -137,11 +132,8 @@ public class LettuceReactiveListCommands implements ReactiveListCommands {
Assert.notNull(command.getKey(), "Key must not be null!");
Assert.notNull(command.getRange(), "Range must not be null!");
return LettuceReactiveRedisConnection.<Boolean> monoConverter()
.convert(cmd
.ltrim(command.getKey().array(), command.getRange().getLowerBound(), command.getRange().getUpperBound())
.map(LettuceConverters::stringToBoolean))
.map(value -> new BooleanResponse<>(command, value));
return cmd.ltrim(command.getKey(), command.getRange().getLowerBound(), command.getRange().getUpperBound())
.map(LettuceConverters::stringToBoolean).map(value -> new BooleanResponse<>(command, value));
}));
}
@@ -157,9 +149,7 @@ public class LettuceReactiveListCommands implements ReactiveListCommands {
Assert.notNull(command.getKey(), "Key must not be null!");
Assert.notNull(command.getIndex(), "Index value must not be null!");
return LettuceReactiveRedisConnection.<ByteBuffer> monoConverter()
.convert(cmd.lindex(command.getKey().array(), command.getIndex()).map(ByteBuffer::wrap))
.map(value -> new ByteBufferResponse<>(command, value));
return cmd.lindex(command.getKey(), command.getIndex()).map(value -> new ByteBufferResponse<>(command, value));
}));
}
@@ -177,10 +167,8 @@ public class LettuceReactiveListCommands implements ReactiveListCommands {
Assert.notNull(command.getPivot(), "Pivot must not be null!");
Assert.notNull(command.getPosition(), "Position must not be null!");
return LettuceReactiveRedisConnection.<Long> monoConverter()
.convert(cmd.linsert(command.getKey().array(), Position.BEFORE.equals(command.getPosition()),
command.getPivot().array(), command.getValue().array()))
.map(value -> new NumericResponse<>(command, value));
return cmd.linsert(command.getKey(), Position.BEFORE.equals(command.getPosition()), command.getPivot(),
command.getValue()).map(value -> new NumericResponse<>(command, value));
}));
}
@@ -199,10 +187,8 @@ public class LettuceReactiveListCommands implements ReactiveListCommands {
Assert.notNull(command.getValue(), "value must not be null!");
Assert.notNull(command.getIndex(), "Index must not be null!");
return LettuceReactiveRedisConnection.<Boolean> monoConverter()
.convert(cmd.lset(command.getKey().array(), command.getIndex(), command.getValue().array())
.map(LettuceConverters::stringToBoolean))
.map(value -> new BooleanResponse<>(command, value));
return cmd.lset(command.getKey(), command.getIndex(), command.getValue())
.map(LettuceConverters::stringToBoolean).map(value -> new BooleanResponse<>(command, value));
});
});
}
@@ -220,8 +206,7 @@ public class LettuceReactiveListCommands implements ReactiveListCommands {
Assert.notNull(command.getValue(), "Value must not be null!");
Assert.notNull(command.getCount(), "Count must not be null!");
return LettuceReactiveRedisConnection.<Long> monoConverter()
.convert(cmd.lrem(command.getKey().array(), command.getCount(), command.getValue().array()))
return cmd.lrem(command.getKey(), command.getCount(), command.getValue())
.map(value -> new NumericResponse<>(command, value));
}));
}
@@ -238,11 +223,10 @@ public class LettuceReactiveListCommands implements ReactiveListCommands {
Assert.notNull(command.getKey(), "Key must not be null!");
Assert.notNull(command.getDirection(), "Direction must not be null!");
Observable<byte[]> popResult = ObjectUtils.nullSafeEquals(Direction.RIGHT, command.getDirection())
? cmd.rpop(command.getKey().array()) : cmd.lpop(command.getKey().array());
Mono<ByteBuffer> popResult = ObjectUtils.nullSafeEquals(Direction.RIGHT, command.getDirection())
? cmd.rpop(command.getKey()) : cmd.lpop(command.getKey());
return LettuceReactiveRedisConnection.<ByteBuffer> monoConverter().convert(popResult.map(ByteBuffer::wrap))
.map(value -> new ByteBufferResponse<>(command, value));
return popResult.map(value -> new ByteBufferResponse<>(command, value));
}));
}
@@ -258,16 +242,14 @@ public class LettuceReactiveListCommands implements ReactiveListCommands {
Assert.notNull(command.getKeys(), "Keys must not be null!");
Assert.notNull(command.getDirection(), "Direction must not be null!");
byte[][] keys = command.getKeys().stream().map(ByteBuffer::array).toArray(size -> new byte[size][]);
long timeout = command.getTimeout().get(ChronoUnit.SECONDS);
Observable<PopResult> mappedObservable = (ObjectUtils.nullSafeEquals(Direction.RIGHT, command.getDirection())
? cmd.brpop(timeout, keys) : cmd.blpop(timeout, keys))
.map(kv -> Arrays.asList(ByteBuffer.wrap(kv.key), ByteBuffer.wrap(kv.value)))
.map(val -> new PopResult(val));
Mono<PopResult> mappedMono = (ObjectUtils.nullSafeEquals(Direction.RIGHT, command.getDirection())
? cmd.brpop(timeout, command.getKeys().stream().toArray(ByteBuffer[]::new))
: cmd.blpop(timeout, command.getKeys().stream().toArray(ByteBuffer[]::new)))
.map(kv -> Arrays.asList(kv.getKey(), kv.getValue())).map(PopResult::new);
return LettuceReactiveRedisConnection.<PopResult> monoConverter().convert(mappedObservable)
.map(value -> new PopResponse(command, value));
return mappedMono.map(value -> new PopResponse(command, value));
}));
}
@@ -283,8 +265,7 @@ public class LettuceReactiveListCommands implements ReactiveListCommands {
Assert.notNull(command.getKey(), "Key must not be null!");
Assert.notNull(command.getDestination(), "Destination key must not be null!");
return LettuceReactiveRedisConnection.<ByteBuffer> monoConverter()
.convert(cmd.rpoplpush(command.getKey().array(), command.getDestination().array()).map(ByteBuffer::wrap))
return cmd.rpoplpush(command.getKey(), command.getDestination())
.map(value -> new ByteBufferResponse<>(command, value));
}));
}
@@ -302,9 +283,8 @@ public class LettuceReactiveListCommands implements ReactiveListCommands {
Assert.notNull(command.getDestination(), "Destination key must not be null!");
Assert.notNull(command.getTimeout(), "Timeout must not be null!");
return LettuceReactiveRedisConnection.<ByteBuffer> monoConverter()
.convert(cmd.brpoplpush(command.getTimeout().get(ChronoUnit.SECONDS), command.getKey().array(),
command.getDestination().array()).map(ByteBuffer::wrap))
return cmd.brpoplpush(command.getTimeout().get(ChronoUnit.SECONDS), command.getKey(), command.getDestination())
.map(value -> new ByteBufferResponse<>(command, value));
}));
}

42
src/main/java/org/springframework/data/redis/connection/lettuce/LettuceReactiveNumberCommands.java
View File

@@ -23,10 +23,11 @@ import org.springframework.util.Assert;
import org.springframework.util.NumberUtils;
import reactor.core.publisher.Flux;
import rx.Observable;
import reactor.core.publisher.Mono;
/**
* @author Christoph Strobl
* @author Mark Paluch
* @since 2.0
*/
public class LettuceReactiveNumberCommands implements ReactiveNumberCommands {
@@ -55,8 +56,7 @@ public class LettuceReactiveNumberCommands implements ReactiveNumberCommands {
Assert.notNull(command.getKey(), "Key must not be null!");
return LettuceReactiveRedisConnection.<Long> monoConverter().convert(cmd.incr(command.getKey().array()))
.map(value -> new NumericResponse<>(command, value));
return cmd.incr(command.getKey()).map(value -> new NumericResponse<>(command, value));
}));
}
@@ -74,16 +74,15 @@ public class LettuceReactiveNumberCommands implements ReactiveNumberCommands {
T incrBy = command.getValue();
Observable<? extends Number> result = null;
Mono<? extends Number> result = null;
if (incrBy instanceof Double || incrBy instanceof Float) {
result = cmd.incrbyfloat(command.getKey().array(), incrBy.doubleValue());
result = cmd.incrbyfloat(command.getKey(), incrBy.doubleValue());
} else {
result = cmd.incrby(command.getKey().array(), incrBy.longValue());
result = cmd.incrby(command.getKey(), incrBy.longValue());
}
return LettuceReactiveRedisConnection.<T> monoConverter()
.convert(result.map(val -> NumberUtils.convertNumberToTargetClass(val, incrBy.getClass())))
.map(res -> new NumericResponse<>(command, res));
return result.map(val -> NumberUtils.convertNumberToTargetClass(val, incrBy.getClass()))
.map(res -> new NumericResponse<>(command, (T) res));
}));
}
@@ -98,8 +97,7 @@ public class LettuceReactiveNumberCommands implements ReactiveNumberCommands {
Assert.notNull(command.getKey(), "Key must not be null!");
return LettuceReactiveRedisConnection.<Long> monoConverter().convert(cmd.decr(command.getKey().array()))
.map(value -> new NumericResponse<>(command, value));
return cmd.decr(command.getKey()).map(value -> new NumericResponse<>(command, value));
}));
}
@@ -117,16 +115,15 @@ public class LettuceReactiveNumberCommands implements ReactiveNumberCommands {
T decrBy = command.getValue();
Observable<? extends Number> result = null;
Mono<? extends Number> result = null;
if (decrBy instanceof Double || decrBy instanceof Float) {
result = cmd.incrbyfloat(command.getKey().array(), decrBy.doubleValue() * (-1.0D));
result = cmd.incrbyfloat(command.getKey(), decrBy.doubleValue() * (-1.0D));
} else {
result = cmd.decrby(command.getKey().array(), decrBy.longValue());
result = cmd.decrby(command.getKey(), decrBy.longValue());
}
return LettuceReactiveRedisConnection.<T> monoConverter()
.convert(result.map(val -> NumberUtils.convertNumberToTargetClass(val, decrBy.getClass())))
.map(res -> new NumericResponse<>(command, res));
return result.map(val -> NumberUtils.convertNumberToTargetClass(val, decrBy.getClass()))
.map(res -> new NumericResponse<>(command, (T) res));
}));
}
@@ -144,17 +141,16 @@ public class LettuceReactiveNumberCommands implements ReactiveNumberCommands {
T incrBy = command.getValue();
Observable<? extends Number> result = null;
Mono<? extends Number> result = null;
if (incrBy instanceof Double || incrBy instanceof Float) {
result = cmd.hincrbyfloat(command.getKey().array(), command.getField().array(), incrBy.doubleValue());
result = cmd.hincrbyfloat(command.getKey(), command.getField(), incrBy.doubleValue());
} else {
result = cmd.hincrby(command.getKey().array(), command.getField().array(), incrBy.longValue());
result = cmd.hincrby(command.getKey(), command.getField(), incrBy.longValue());
}
return LettuceReactiveRedisConnection.<T> monoConverter()
.convert(result.map(val -> NumberUtils.convertNumberToTargetClass(val, incrBy.getClass())))
.map(value -> new NumericResponse<>(command, value));
return result.map(val -> NumberUtils.convertNumberToTargetClass(val, incrBy.getClass()))
.map(value -> new NumericResponse<>(command, (T) value));
}));
}

13
src/main/java/org/springframework/data/redis/connection/lettuce/LettuceReactiveRedisClusterConnection.java
View File

@@ -16,20 +16,23 @@
package org.springframework.data.redis.connection.lettuce;
import java.nio.ByteBuffer;
import org.springframework.data.redis.connection.ReactiveRedisClusterConnection;
import org.springframework.data.redis.connection.RedisNode;
import org.springframework.util.Assert;
import org.springframework.util.StringUtils;
import com.lambdaworks.redis.api.rx.RedisReactiveCommands;
import com.lambdaworks.redis.api.reactive.RedisReactiveCommands;
import com.lambdaworks.redis.cluster.RedisClusterClient;
import com.lambdaworks.redis.cluster.api.StatefulRedisClusterConnection;
import com.lambdaworks.redis.cluster.api.rx.RedisClusterReactiveCommands;
import com.lambdaworks.redis.cluster.api.reactive.RedisClusterReactiveCommands;
import reactor.core.publisher.Flux;
/**
* @author Christoph Strobl
* @author Mark Paluch
* @since 2.0
*/
public class LettuceReactiveRedisClusterConnection extends LettuceReactiveRedisConnection
@@ -101,7 +104,7 @@ public class LettuceReactiveRedisClusterConnection extends LettuceReactiveRedisC
}
@Override
protected StatefulRedisClusterConnection<byte[], byte[]> getConnection() {
protected StatefulRedisClusterConnection<ByteBuffer, ByteBuffer> getConnection() {
Assert.isInstanceOf(StatefulRedisClusterConnection.class, super.getConnection(),
"Connection needs to be instance of StatefulRedisClusterConnection");
@@ -109,11 +112,11 @@ public class LettuceReactiveRedisClusterConnection extends LettuceReactiveRedisC
return (StatefulRedisClusterConnection) super.getConnection();
}
protected RedisClusterReactiveCommands<byte[], byte[]> getCommands() {
protected RedisClusterReactiveCommands<ByteBuffer, ByteBuffer> getCommands() {
return getConnection().reactive();
}
protected RedisReactiveCommands<byte[], byte[]> getCommands(RedisNode node) {
protected RedisReactiveCommands<ByteBuffer, ByteBuffer> getCommands(RedisNode node) {
if (!(getConnection() instanceof StatefulRedisClusterConnection)) {
throw new IllegalArgumentException("o.O connection needs to be cluster compatible " + getConnection());

45
src/main/java/org/springframework/data/redis/connection/lettuce/LettuceReactiveRedisConnection.java
View File

@@ -15,6 +15,7 @@
*/
package org.springframework.data.redis.connection.lettuce;
import java.nio.ByteBuffer;
import java.util.function.Function;
import org.reactivestreams.Publisher;
@@ -30,8 +31,7 @@ import com.lambdaworks.redis.api.StatefulConnection;
import com.lambdaworks.redis.api.StatefulRedisConnection;
import com.lambdaworks.redis.cluster.RedisClusterClient;
import com.lambdaworks.redis.cluster.api.StatefulRedisClusterConnection;
import com.lambdaworks.redis.cluster.api.rx.RedisClusterReactiveCommands;
import com.lambdaworks.redis.codec.ByteArrayCodec;
import com.lambdaworks.redis.cluster.api.reactive.RedisClusterReactiveCommands;
import com.lambdaworks.redis.codec.RedisCodec;
import reactor.core.publisher.Flux;
@@ -46,7 +46,7 @@ public class LettuceReactiveRedisConnection implements ReactiveRedisConnection {
private StatefulConnection<ByteBuffer, ByteBuffer> connection;
private static final RedisCodec<byte[], byte[]> CODEC = new ByteArrayCodec();
private static final RedisCodec<ByteBuffer, ByteBuffer> CODEC = ByteBufferCodec.INSTANCE;
public LettuceReactiveRedisConnection(AbstractRedisClient client) {
@@ -124,16 +124,16 @@ public class LettuceReactiveRedisConnection implements ReactiveRedisConnection {
connection.close();
}
protected StatefulConnection<byte[], byte[]> getConnection() {
protected StatefulConnection<ByteBuffer, ByteBuffer> getConnection() {
return connection;
}
protected RedisClusterReactiveCommands<byte[], byte[]> getCommands() {
protected RedisClusterReactiveCommands<ByteBuffer, ByteBuffer> getCommands() {
if (connection instanceof StatefulRedisConnection) {
return ((StatefulRedisConnection<byte[], byte[]>) connection).reactive();
return ((StatefulRedisConnection<ByteBuffer, ByteBuffer>) connection).reactive();
} else if (connection instanceof StatefulRedisClusterConnection) {
return ((StatefulRedisClusterConnection<byte[], byte[]>) connection).reactive();
return ((StatefulRedisClusterConnection<ByteBuffer, ByteBuffer>) connection).reactive();
}
throw new RuntimeException("o.O unknown connection type " + connection);
@@ -157,6 +157,35 @@ public class LettuceReactiveRedisConnection implements ReactiveRedisConnection {
}
interface LettuceReactiveCallback<T> {
Publisher<T> doWithCommands(RedisClusterReactiveCommands<byte[], byte[]> cmd);
Publisher<T> doWithCommands(RedisClusterReactiveCommands<ByteBuffer, ByteBuffer> cmd);
}
static enum ByteBufferCodec implements RedisCodec<ByteBuffer, ByteBuffer> {
INSTANCE;
@Override
public ByteBuffer decodeKey(ByteBuffer bytes) {
ByteBuffer buffer = ByteBuffer.allocate(bytes.remaining());
buffer.put(bytes);
buffer.flip();
return buffer;
}
@Override
public ByteBuffer decodeValue(ByteBuffer bytes) {
return decodeKey(bytes);
}
@Override
public ByteBuffer encodeKey(ByteBuffer key) {
return key.duplicate();
}
@Override
public ByteBuffer encodeValue(ByteBuffer value) {
return value.duplicate();
}
}
}

62
src/main/java/org/springframework/data/redis/connection/lettuce/LettuceReactiveSetCommands.java
View File

@@ -27,13 +27,13 @@ import org.springframework.data.redis.connection.ReactiveRedisConnection.MultiVa
import org.springframework.data.redis.connection.ReactiveRedisConnection.NumericResponse;
import org.springframework.data.redis.connection.ReactiveSetCommands;
import org.springframework.util.Assert;
import org.springframework.util.ObjectUtils;
import reactor.core.publisher.Flux;
import rx.Observable;
import reactor.core.publisher.Mono;
/**
* @author Christoph Strobl
* @author Mark Paluch
* @since 2.0
*/
public class LettuceReactiveSetCommands implements ReactiveSetCommands {
@@ -63,9 +63,7 @@ public class LettuceReactiveSetCommands implements ReactiveSetCommands {
Assert.notNull(command.getKey(), "Key must not be null!");
Assert.notNull(command.getValues(), "Values must not be null!");
return LettuceReactiveRedisConnection.<Long> monoConverter()
.convert(cmd.sadd(command.getKey().array(),
command.getValues().stream().map(ByteBuffer::array).toArray(size -> new byte[size][])))
return cmd.sadd(command.getKey(), command.getValues().stream().toArray(ByteBuffer[]::new))
.map(value -> new NumericResponse<>(command, value));
}));
}
@@ -82,9 +80,7 @@ public class LettuceReactiveSetCommands implements ReactiveSetCommands {
Assert.notNull(command.getKey(), "Key must not be null!");
Assert.notNull(command.getValues(), "Values must not be null!");
return LettuceReactiveRedisConnection.<Long> monoConverter()
.convert(cmd.srem(command.getKey().array(),
command.getValues().stream().map(ByteBuffer::array).toArray(size -> new byte[size][])))
return cmd.srem(command.getKey(), command.getValues().stream().toArray(ByteBuffer[]::new))
.map(value -> new NumericResponse<>(command, value));
}));
}
@@ -100,9 +96,7 @@ public class LettuceReactiveSetCommands implements ReactiveSetCommands {
Assert.notNull(command.getKey(), "Key must not be null!");
return LettuceReactiveRedisConnection.<ByteBuffer> monoConverter()
.convert(cmd.spop(command.getKey().array()).map(ByteBuffer::wrap))
.map(value -> new ByteBufferResponse<>(command, value));
return cmd.spop(command.getKey()).map(value -> new ByteBufferResponse<>(command, value));
}));
}
@@ -119,8 +113,7 @@ public class LettuceReactiveSetCommands implements ReactiveSetCommands {
Assert.notNull(command.getDestination(), "Destination key must not be null!");
Assert.notNull(command.getValue(), "Value must not be null!");
return LettuceReactiveRedisConnection.<Boolean> monoConverter()
.convert(cmd.smove(command.getKey().array(), command.getDestination().array(), command.getValue().array()))
return cmd.smove(command.getKey(), command.getDestination(), command.getValue())
.map(value -> new BooleanResponse<>(command, value));
}));
}
@@ -136,8 +129,7 @@ public class LettuceReactiveSetCommands implements ReactiveSetCommands {
Assert.notNull(command.getKey(), "Key must not be null!");
return LettuceReactiveRedisConnection.<Long> monoConverter().convert(cmd.scard(command.getKey().array()))
.map(value -> new NumericResponse<>(command, value));
return cmd.scard(command.getKey()).map(value -> new NumericResponse<>(command, value));
}));
}
@@ -153,9 +145,7 @@ public class LettuceReactiveSetCommands implements ReactiveSetCommands {
Assert.notNull(command.getKey(), "Key must not be null!");
Assert.notNull(command.getValue(), "Value must not be null!");
return LettuceReactiveRedisConnection.<Boolean> monoConverter()
.convert(cmd.sismember(command.getKey().array(), command.getValue().array()))
.map(value -> new BooleanResponse<>(command, value));
return cmd.sismember(command.getKey(), command.getValue()).map(value -> new BooleanResponse<>(command, value));
}));
}
@@ -170,9 +160,7 @@ public class LettuceReactiveSetCommands implements ReactiveSetCommands {
Assert.notNull(command.getKeys(), "Keys must not be null!");
return LettuceReactiveRedisConnection.<List<ByteBuffer>> monoConverter()
.convert(cmd.sinter(command.getKeys().stream().map(ByteBuffer::array).toArray(size -> new byte[size][]))
.map(ByteBuffer::wrap).toList())
return cmd.sinter(command.getKeys().stream().toArray(ByteBuffer[]::new)).collectList()
.map(value -> new MultiValueResponse<>(command, value));
}));
}
@@ -189,9 +177,7 @@ public class LettuceReactiveSetCommands implements ReactiveSetCommands {
Assert.notNull(command.getKeys(), "Keys must not be null!");
Assert.notNull(command.getKey(), "Destination key must not be null!");
return LettuceReactiveRedisConnection.<Long> monoConverter()
.convert(cmd.sinterstore(command.getKey().array(),
command.getKeys().stream().map(ByteBuffer::array).toArray(size -> new byte[size][])))
return cmd.sinterstore(command.getKey(), command.getKeys().stream().toArray(ByteBuffer[]::new))
.map(value -> new NumericResponse<>(command, value));
}));
}
@@ -207,9 +193,7 @@ public class LettuceReactiveSetCommands implements ReactiveSetCommands {
Assert.notNull(command.getKeys(), "Keys must not be null!");
return LettuceReactiveRedisConnection.<List<ByteBuffer>> monoConverter()
.convert(cmd.sunion(command.getKeys().stream().map(ByteBuffer::array).toArray(size -> new byte[size][]))
.map(ByteBuffer::wrap).toList())
return cmd.sunion(command.getKeys().stream().toArray(ByteBuffer[]::new)).collectList()
.map(value -> new MultiValueResponse<>(command, value));
}));
}
@@ -226,9 +210,7 @@ public class LettuceReactiveSetCommands implements ReactiveSetCommands {
Assert.notNull(command.getKeys(), "Keys must not be null!");
Assert.notNull(command.getKey(), "Destination key must not be null!");
return LettuceReactiveRedisConnection.<Long> monoConverter()
.convert(cmd.sunionstore(command.getKey().array(),
command.getKeys().stream().map(ByteBuffer::array).toArray(size -> new byte[size][])))
return cmd.sunionstore(command.getKey(), command.getKeys().stream().toArray(ByteBuffer[]::new))
.map(value -> new NumericResponse<>(command, value));
}));
}
@@ -244,9 +226,7 @@ public class LettuceReactiveSetCommands implements ReactiveSetCommands {
Assert.notNull(command.getKeys(), "Keys must not be null!");
return LettuceReactiveRedisConnection.<List<ByteBuffer>> monoConverter()
.convert(cmd.sdiff(command.getKeys().stream().map(ByteBuffer::array).toArray(size -> new byte[size][]))
.map(ByteBuffer::wrap).toList())
return cmd.sdiff(command.getKeys().stream().toArray(ByteBuffer[]::new)).collectList()
.map(value -> new MultiValueResponse<>(command, value));
}));
}
@@ -263,9 +243,7 @@ public class LettuceReactiveSetCommands implements ReactiveSetCommands {
Assert.notNull(command.getKeys(), "Keys must not be null!");
Assert.notNull(command.getKey(), "Destination key must not be null!");
return LettuceReactiveRedisConnection.<Long> monoConverter()
.convert(cmd.sdiffstore(command.getKey().array(),
command.getKeys().stream().map(ByteBuffer::array).toArray(size -> new byte[size][])))
return cmd.sdiffstore(command.getKey(), command.getKeys().stream().toArray(ByteBuffer[]::new))
.map(value -> new NumericResponse<>(command, value));
}));
}
@@ -281,9 +259,7 @@ public class LettuceReactiveSetCommands implements ReactiveSetCommands {
Assert.notNull(command.getKey(), "Key must not be null!");
return LettuceReactiveRedisConnection.<List<ByteBuffer>> monoConverter()
.convert(cmd.smembers(command.getKey().array()).map(ByteBuffer::wrap).toList())
.map(value -> new MultiValueResponse<>(command, value));
return cmd.smembers(command.getKey()).collectList().map(value -> new MultiValueResponse<>(command, value));
}));
}
@@ -301,12 +277,10 @@ public class LettuceReactiveSetCommands implements ReactiveSetCommands {
boolean singleElement = !command.getCount().isPresent() || command.getCount().get().equals(1L);
Observable<List<ByteBuffer>> result = singleElement
? cmd.srandmember(command.getKey().array()).map(ByteBuffer::wrap).map(Collections::singletonList)
: cmd.srandmember(command.getKey().array(), command.getCount().get()).map(ByteBuffer::wrap).toList();
Mono<List<ByteBuffer>> result = singleElement ? cmd.srandmember(command.getKey()).map(Collections::singletonList)
: cmd.srandmember(command.getKey(), command.getCount().get()).collectList();
return LettuceReactiveRedisConnection.<List<ByteBuffer>> monoConverter().convert(result)
.map(value -> new MultiValueResponse<>(command, value));
return result.map(value -> new MultiValueResponse<>(command, value));
}));
}

98
src/main/java/org/springframework/data/redis/connection/lettuce/LettuceReactiveStringCommands.java
View File

@@ -16,10 +16,7 @@
package org.springframework.data.redis.connection.lettuce;
import java.nio.ByteBuffer;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import java.util.stream.Collectors;
import org.reactivestreams.Publisher;
import org.springframework.data.domain.Range;
@@ -35,14 +32,16 @@ import org.springframework.util.Assert;
import com.lambdaworks.redis.SetArgs;
import reactor.core.publisher.Flux;
import rx.Observable;
import reactor.core.publisher.Mono;
/**
* @author Christoph Strobl
* @author Mark Paluch
* @since 2.0
*/
public class LettuceReactiveStringCommands implements ReactiveStringCommands {
private final static ByteBuffer EMPTY_BYTE_BUFFER = ByteBuffer.wrap(new byte[0]);
private final LettuceReactiveRedisConnection connection;
/**
@@ -67,11 +66,8 @@ public class LettuceReactiveStringCommands implements ReactiveStringCommands {
Assert.notNull(keys, "Keys must not be null!");
return LettuceReactiveRedisConnection.<MultiValueResponse<List<ByteBuffer>, ByteBuffer>> monoConverter()
.convert(cmd
.mget(keys.stream().map(ByteBuffer::array).collect(Collectors.toList()).toArray(new byte[keys.size()][]))
.map((value) -> value != null ? ByteBuffer.wrap(value) : ByteBuffer.allocate(0)).toList()
.map((values) -> new MultiValueResponse<>(keys, values)));
return cmd.mget(keys.stream().toArray(ByteBuffer[]::new)).map((value) -> value.getValueOrElse(EMPTY_BYTE_BUFFER))
.collectList().map((values) -> new MultiValueResponse<>(keys, values));
}));
}
@@ -94,11 +90,9 @@ public class LettuceReactiveStringCommands implements ReactiveStringCommands {
command.getOption().isPresent() ? command.getOption().get() : null);
}
return LettuceReactiveRedisConnection.<Boolean> monoConverter().convert(
args != null ? cmd.set(command.getKey().array(), command.getValue().array(), args)
: cmd.set(command.getKey().array(), command.getValue().array()).map(LettuceConverters::stringToBoolean))
.map((value) -> new BooleanResponse<>(command, value));
Mono<String> mono = args != null ? cmd.set(command.getKey(), command.getValue(), args)
: cmd.set(command.getKey(), command.getValue());
return mono.map(LettuceConverters::stringToBoolean).map((value) -> new BooleanResponse<>(command, value));
}));
}
@@ -118,10 +112,8 @@ public class LettuceReactiveStringCommands implements ReactiveStringCommands {
throw new IllegalArgumentException("Command must not define exipiration nor option for GETSET.");
}
return LettuceReactiveRedisConnection.<ByteBufferResponse<SetCommand>> monoConverter()
.convert(cmd.getset(command.getKey().array(), command.getValue().array())
.map((value) -> new ByteBufferResponse<>(command, ByteBuffer.wrap(value)))
.defaultIfEmpty(new ByteBufferResponse<>(command, ByteBuffer.allocate(0))));
return cmd.getset(command.getKey(), command.getValue()).map((value) -> new ByteBufferResponse<>(command, value))
.defaultIfEmpty(new ByteBufferResponse<>(command, EMPTY_BYTE_BUFFER));
}));
}
@@ -136,10 +128,8 @@ public class LettuceReactiveStringCommands implements ReactiveStringCommands {
Assert.notNull(command.getKey(), "Key must not be null!");
return LettuceReactiveRedisConnection.<ByteBuffer> monoConverter()
.convert(cmd.get(command.getKey().array()).map(ByteBuffer::wrap))
.map((value) -> new ByteBufferResponse<>(command, value))
.defaultIfEmpty(new ByteBufferResponse<>(command, ByteBuffer.allocate(0)));
return cmd.get(command.getKey()).map((value) -> new ByteBufferResponse<>(command, value))
.defaultIfEmpty(new ByteBufferResponse<>(command, EMPTY_BYTE_BUFFER));
}));
}
@@ -155,9 +145,7 @@ public class LettuceReactiveStringCommands implements ReactiveStringCommands {
Assert.notNull(command.getKey(), "Key must not be null!");
Assert.notNull(command.getValue(), "Value must not be null!");
return LettuceReactiveRedisConnection.<Boolean> monoConverter()
.convert(cmd.setnx(command.getKey().array(), command.getValue().array()))
.map((value) -> new BooleanResponse<>(command, value));
return cmd.setnx(command.getKey(), command.getValue()).map((value) -> new BooleanResponse<>(command, value));
}));
}
@@ -173,9 +161,7 @@ public class LettuceReactiveStringCommands implements ReactiveStringCommands {
Assert.notNull(command.getValue(), "Value must not be null!");
Assert.isTrue(command.getExpiration().isPresent(), "Expiration time must not be null!");
return LettuceReactiveRedisConnection.<String> monoConverter()
.convert(cmd.setex(command.getKey().array(), command.getExpiration().get().getExpirationTimeInSeconds(),
command.getValue().array()))
return cmd.setex(command.getKey(), command.getExpiration().get().getExpirationTimeInSeconds(), command.getValue())
.map(LettuceConverters::stringToBoolean).map((value) -> new BooleanResponse<>(command, value));
}));
}
@@ -193,9 +179,8 @@ public class LettuceReactiveStringCommands implements ReactiveStringCommands {
Assert.notNull(command.getValue(), "Value must not be null!");
Assert.isTrue(command.getExpiration().isPresent(), "Expiration time must not be null!");
return LettuceReactiveRedisConnection.<String> monoConverter()
.convert(cmd.psetex(command.getKey().array(), command.getExpiration().get().getExpirationTimeInMilliseconds(),
command.getValue().array()))
return cmd
.psetex(command.getKey(), command.getExpiration().get().getExpirationTimeInMilliseconds(), command.getValue())
.map(LettuceConverters::stringToBoolean).map((value) -> new BooleanResponse<>(command, value));
}));
}
@@ -211,11 +196,8 @@ public class LettuceReactiveStringCommands implements ReactiveStringCommands {
Assert.notEmpty(command.getKeyValuePairs(), "Pairs must not be null or empty!");
Map<byte[], byte[]> map = new LinkedHashMap<>();
command.getKeyValuePairs().entrySet().forEach(entry -> map.put(entry.getKey().array(), entry.getValue().array()));
return LettuceReactiveRedisConnection.<String> monoConverter().convert(cmd.mset(map))
.map(LettuceConverters::stringToBoolean).map((value) -> new BooleanResponse<>(command, value));
return cmd.mset(command.getKeyValuePairs()).map(LettuceConverters::stringToBoolean)
.map((value) -> new BooleanResponse<>(command, value));
}));
}
@@ -230,11 +212,7 @@ public class LettuceReactiveStringCommands implements ReactiveStringCommands {
Assert.notEmpty(command.getKeyValuePairs(), "Pairs must not be null or empty!");
Map<byte[], byte[]> map = new LinkedHashMap<>();
command.getKeyValuePairs().entrySet().forEach(entry -> map.put(entry.getKey().array(), entry.getValue().array()));
return LettuceReactiveRedisConnection.<Boolean> monoConverter().convert(cmd.msetnx(map))
.map((value) -> new BooleanResponse<>(command, value));
return cmd.msetnx(command.getKeyValuePairs()).map((value) -> new BooleanResponse<>(command, value));
}));
}
@@ -250,9 +228,7 @@ public class LettuceReactiveStringCommands implements ReactiveStringCommands {
Assert.notNull(command.getKey(), "Key must not be null!");
Assert.notNull(command.getValue(), "Value must not be null!");
return LettuceReactiveRedisConnection.<Long> monoConverter()
.convert(cmd.append(command.getKey().array(), command.getValue().array()))
.map((value) -> new NumericResponse<>(command, value));
return cmd.append(command.getKey(), command.getValue()).map((value) -> new NumericResponse<>(command, value));
}));
}
@@ -270,9 +246,7 @@ public class LettuceReactiveStringCommands implements ReactiveStringCommands {
Range<Long> range = command.getRange();
return LettuceReactiveRedisConnection
.<ByteBuffer> monoConverter().convert(cmd
.getrange(command.getKey().array(), range.getLowerBound(), range.getUpperBound()).map(ByteBuffer::wrap))
return cmd.getrange(command.getKey(), range.getLowerBound(), range.getUpperBound())
.map((value) -> new ByteBufferResponse<>(command, value));
}));
}
@@ -290,8 +264,7 @@ public class LettuceReactiveStringCommands implements ReactiveStringCommands {
Assert.notNull(command.getValue(), "Value must not be null!");
Assert.notNull(command.getOffset(), "Offset must not be null!");
return LettuceReactiveRedisConnection.<Long> monoConverter()
.convert(cmd.setrange(command.getKey().array(), command.getOffset(), command.getValue().array()))
return cmd.setrange(command.getKey(), command.getOffset(), command.getValue())
.map((value) -> new NumericResponse<>(command, value));
}));
}
@@ -308,8 +281,7 @@ public class LettuceReactiveStringCommands implements ReactiveStringCommands {
Assert.notNull(command.getKey(), "Key must not be null!");
Assert.notNull(command.getOffset(), "Offset must not be null!");
return LettuceReactiveRedisConnection.<Boolean> monoConverter()
.convert(cmd.getbit(command.getKey().array(), command.getOffset()).map(LettuceConverters::toBoolean))
return cmd.getbit(command.getKey(), command.getOffset()).map(LettuceConverters::toBoolean)
.map(value -> new BooleanResponse<>(command, value));
}));
}
@@ -327,10 +299,8 @@ public class LettuceReactiveStringCommands implements ReactiveStringCommands {
Assert.notNull(command.getValue(), "Value must not be null!");
Assert.notNull(command.getOffset(), "Offset must not be null!");
return LettuceReactiveRedisConnection.<Boolean> monoConverter()
.convert(cmd.setbit(command.getKey().array(), command.getOffset(), command.getValue() ? 1 : 0)
.map(LettuceConverters::toBoolean))
.map(respValue -> new BooleanResponse<>(command, respValue));
return cmd.setbit(command.getKey(), command.getOffset(), command.getValue() ? 1 : 0)
.map(LettuceConverters::toBoolean).map(respValue -> new BooleanResponse<>(command, respValue));
}));
}
@@ -346,10 +316,8 @@ public class LettuceReactiveStringCommands implements ReactiveStringCommands {
Assert.notNull(command.getKey(), "Key must not be null!");
Range<Long> range = command.getRange();
return LettuceReactiveRedisConnection.<Long> monoConverter()
.convert(range != null ? cmd.bitcount(command.getKey().array(), range.getLowerBound(), range.getUpperBound())
: cmd.bitcount(command.getKey().array()))
.map(responseValue -> new NumericResponse<>(command, responseValue));
return (range != null ? cmd.bitcount(command.getKey(), range.getLowerBound(), range.getUpperBound())
: cmd.bitcount(command.getKey())).map(responseValue -> new NumericResponse<>(command, responseValue));
}));
}
@@ -365,9 +333,9 @@ public class LettuceReactiveStringCommands implements ReactiveStringCommands {
Assert.notNull(command.getDestinationKey(), "DestinationKey must not be null!");
Assert.notEmpty(command.getKeys(), "Keys must not be null or empty");
Observable<Long> result = null;
byte[] destinationKey = command.getDestinationKey().array();
byte[][] sourceKeys = command.getKeys().stream().map(ByteBuffer::array).toArray(size -> new byte[size][]);
Mono<Long> result = null;
ByteBuffer destinationKey = command.getDestinationKey();
ByteBuffer[] sourceKeys = command.getKeys().stream().toArray(ByteBuffer[]::new);
switch (command.getBitOp()) {
case AND:
@@ -389,8 +357,7 @@ public class LettuceReactiveStringCommands implements ReactiveStringCommands {
throw new IllegalArgumentException(String.format("Unknown BITOP '%s'.", command.getBitOp()));
}
return LettuceReactiveRedisConnection.<Long> monoConverter().convert(result)
.map(value -> new NumericResponse<>(command, value));
return result.map(value -> new NumericResponse<>(command, value));
}));
}
@@ -403,8 +370,7 @@ public class LettuceReactiveStringCommands implements ReactiveStringCommands {
return connection.execute(cmd -> {
return Flux.from(commands).flatMap(command -> {
return LettuceReactiveRedisConnection.<Long> monoConverter().convert(cmd.strlen(command.getKey().array()))
.map(respValue -> new NumericResponse<>(command, respValue));
return cmd.strlen(command.getKey()).map(respValue -> new NumericResponse<>(command, respValue));
});
});
}

319
src/main/java/org/springframework/data/redis/connection/lettuce/LettuceReactiveZSetCommands.java
View File

@@ -19,6 +19,7 @@ import java.nio.ByteBuffer;
import java.util.List;
import org.reactivestreams.Publisher;
import org.springframework.core.convert.converter.Converter;
import org.springframework.data.domain.Sort.Direction;
import org.springframework.data.redis.connection.DefaultTuple;
import org.springframework.data.redis.connection.ReactiveRedisConnection.KeyCommand;
@@ -27,18 +28,25 @@ import org.springframework.data.redis.connection.ReactiveRedisConnection.Numeric
import org.springframework.data.redis.connection.ReactiveZSetCommands;
import org.springframework.data.redis.connection.RedisZSetCommands.Aggregate;
import org.springframework.data.redis.connection.RedisZSetCommands.Tuple;
import org.springframework.data.util.DirectFieldAccessFallbackBeanWrapper;
import org.springframework.util.Assert;
import org.springframework.util.ObjectUtils;
import org.springframework.util.StringUtils;
import com.lambdaworks.redis.Range;
import com.lambdaworks.redis.Range.Boundary;
import com.lambdaworks.redis.ScoredValue;
import com.lambdaworks.redis.ZAddArgs;
import com.lambdaworks.redis.ZStoreArgs;
import com.lambdaworks.redis.codec.StringCodec;
import com.lambdaworks.redis.protocol.LettuceCharsets;
import reactor.core.publisher.Flux;
import rx.Observable;
import reactor.core.publisher.Mono;
/**
* @author Christoph Strobl
* @author Mark Paluch
* @since 2.0
*/
public class LettuceReactiveZSetCommands implements ReactiveZSetCommands {
@@ -71,7 +79,8 @@ public class LettuceReactiveZSetCommands implements ReactiveZSetCommands {
ZAddArgs args = null;
if (command.getIncr().isPresent() || command.getUpsert().isPresent() || command.getReturnTotalChanged().isPresent()) {
if (command.getIncr().isPresent() || command.getUpsert().isPresent()
|| command.getReturnTotalChanged().isPresent()) {
if (command.getIncr().isPresent() && ObjectUtils.nullSafeEquals(command.getIncr().get(), Boolean.TRUE)) {
@@ -81,12 +90,12 @@ public class LettuceReactiveZSetCommands implements ReactiveZSetCommands {
Tuple tuple = command.getTuples().iterator().next();
return LettuceReactiveRedisConnection.<Double> monoConverter()
.convert(cmd.zaddincr(command.getKey().array(), tuple.getScore(), tuple.getValue()))
return cmd.zaddincr(command.getKey(), tuple.getScore(), ByteBuffer.wrap(tuple.getValue()))
.map(value -> new NumericResponse<>(command, value));
}
if (command.getReturnTotalChanged().isPresent() && ObjectUtils.nullSafeEquals(command.getReturnTotalChanged().get(), Boolean.TRUE)) {
if (command.getReturnTotalChanged().isPresent()
&& ObjectUtils.nullSafeEquals(command.getReturnTotalChanged().get(), Boolean.TRUE)) {
args = ZAddArgs.Builder.ch();
}
@@ -100,15 +109,13 @@ public class LettuceReactiveZSetCommands implements ReactiveZSetCommands {
}
}
ScoredValue<byte[]>[] values = (ScoredValue<byte[]>[]) command.getTuples().stream()
.map(tuple -> new ScoredValue<byte[]>(tuple.getScore(), tuple.getValue()))
.toArray(size -> new ScoredValue[size]);
ScoredValue<ByteBuffer>[] values = (ScoredValue<ByteBuffer>[]) command.getTuples().stream()
.map(tuple -> ScoredValue.fromNullable(tuple.getScore(), ByteBuffer.wrap(tuple.getValue())))
.toArray(ScoredValue[]::new);
Observable<Long> result = args == null ? cmd.zadd(command.getKey().array(), values)
: cmd.zadd(command.getKey().array(), args, values);
Mono<Long> result = args == null ? cmd.zadd(command.getKey(), values) : cmd.zadd(command.getKey(), args, values);
return LettuceReactiveRedisConnection.<Long> monoConverter().convert(result)
.map(value -> new NumericResponse<>(command, value));
return result.map(value -> new NumericResponse<>(command, value));
}));
}
@@ -124,9 +131,7 @@ public class LettuceReactiveZSetCommands implements ReactiveZSetCommands {
Assert.notNull(command.getKey(), "Key must not be null!");
Assert.notEmpty(command.getValues(), "Values must not be null or empty!");
return LettuceReactiveRedisConnection.<Long> monoConverter()
.convert(cmd.zrem(command.getKey().array(),
command.getValues().stream().map(ByteBuffer::array).toArray(size -> new byte[size][])))
return cmd.zrem(command.getKey(), command.getValues().stream().toArray(ByteBuffer[]::new))
.map(value -> new NumericResponse<>(command, value));
}));
}
@@ -144,9 +149,7 @@ public class LettuceReactiveZSetCommands implements ReactiveZSetCommands {
Assert.notNull(command.getValue(), "Member must not be null!");
Assert.notNull(command.getIncrement(), "Increment value must not be null!");
return LettuceReactiveRedisConnection.<Double> monoConverter()
.convert(
cmd.zincrby(command.getKey().array(), command.getIncrement().doubleValue(), command.getValue().array()))
return cmd.zincrby(command.getKey(), command.getIncrement().doubleValue(), command.getValue())
.map(value -> new NumericResponse<>(command, value));
}));
}
@@ -163,12 +166,10 @@ public class LettuceReactiveZSetCommands implements ReactiveZSetCommands {
Assert.notNull(command.getKey(), "Key must not be null!");
Assert.notNull(command.getValue(), "Value must not be null!");
Observable<Long> result = ObjectUtils.nullSafeEquals(command.getDirection(), Direction.ASC)
? cmd.zrank(command.getKey().array(), command.getValue().array())
: cmd.zrevrank(command.getKey().array(), command.getValue().array());
Mono<Long> result = ObjectUtils.nullSafeEquals(command.getDirection(), Direction.ASC)
? cmd.zrank(command.getKey(), command.getValue()) : cmd.zrevrank(command.getKey(), command.getValue());
return LettuceReactiveRedisConnection.<Long> monoConverter().convert(result)
.map(value -> new NumericResponse<>(command, value));
return result.map(value -> new NumericResponse<>(command, value));
}));
}
@@ -184,34 +185,38 @@ public class LettuceReactiveZSetCommands implements ReactiveZSetCommands {
Assert.notNull(command.getKey(), "Key must not be null!");
Assert.notNull(command.getRange(), "Range must not be null!");
Observable<List<Tuple>> result = Observable.empty();
Mono<List<Tuple>> result;
if (ObjectUtils.nullSafeEquals(command.getDirection(), Direction.ASC)) {
if (command.getWithScores().isPresent() && ObjectUtils.nullSafeEquals(command.getWithScores().get(), Boolean.TRUE)) {
result = cmd.zrangeWithScores(command.getKey().array(), command.getRange().getLowerBound(),
command.getRange().getUpperBound()).map(sc -> (Tuple) new DefaultTuple(sc.value, sc.score)).toList();
} else {
if (command.getWithScores().isPresent()
&& ObjectUtils.nullSafeEquals(command.getWithScores().get(), Boolean.TRUE)) {
result = cmd
.zrange(command.getKey().array(), command.getRange().getLowerBound(), command.getRange().getUpperBound())
.map(value -> (Tuple) new DefaultTuple(value, Double.NaN)).toList();
.zrangeWithScores(command.getKey(), command.getRange().getLowerBound(),
command.getRange().getUpperBound())
.map(sc -> (Tuple) new DefaultTuple(getBytes(sc), sc.getScore())).collectList();
} else {
result = cmd.zrange(command.getKey(), command.getRange().getLowerBound(), command.getRange().getUpperBound())
.map(value -> (Tuple) new DefaultTuple(getBytes(value), Double.NaN)).collectList();
}
} else {
if (command.getWithScores().isPresent() && ObjectUtils.nullSafeEquals(command.getWithScores().get(), Boolean.TRUE)) {
result = cmd.zrevrangeWithScores(command.getKey().array(), command.getRange().getLowerBound(),
command.getRange().getUpperBound()).map(sc -> (Tuple) new DefaultTuple(sc.value, sc.score)).toList();
if (command.getWithScores().isPresent()
&& ObjectUtils.nullSafeEquals(command.getWithScores().get(), Boolean.TRUE)) {
result = cmd
.zrevrangeWithScores(command.getKey(), command.getRange().getLowerBound(),
command.getRange().getUpperBound())
.map(sc -> (Tuple) new DefaultTuple(getBytes(sc), sc.getScore())).collectList();
} else {
result = cmd
.zrevrange(command.getKey().array(), command.getRange().getLowerBound(),
command.getRange().getUpperBound())
.map(value -> (Tuple) new DefaultTuple(value, Double.NaN)).toList();
.zrevrange(command.getKey(), command.getRange().getLowerBound(), command.getRange().getUpperBound())
.map(value -> (Tuple) new DefaultTuple(getBytes(value), Double.NaN)).collectList();
}
}
return LettuceReactiveRedisConnection.<List<Tuple>> monoConverter().convert(result)
.map(value -> new MultiValueResponse<>(command, value));
return result.map(value -> new MultiValueResponse<>(command, value));
}));
}
@@ -227,86 +232,70 @@ public class LettuceReactiveZSetCommands implements ReactiveZSetCommands {
Assert.notNull(command.getKey(), "Key must not be null!");
Assert.notNull(command.getRange(), "Range must not be null!");
Object lowerBound = AgrumentConverters.lowerBoundArgOf(command.getRange());
Object upperBound = AgrumentConverters.upperBoundArgOf(command.getRange());
boolean requiresStringConversion = lowerBound instanceof String || upperBound instanceof String;
boolean isLimited = command.getLimit().isPresent();
Observable<List<Tuple>> result = Observable.empty();
Mono<List<Tuple>> result;
if (ObjectUtils.nullSafeEquals(command.getDirection(), Direction.ASC)) {
if (command.getWithScores().isPresent() && ObjectUtils.nullSafeEquals(command.getWithScores().get(), Boolean.TRUE)) {
Range<Number> range = ArgumentConverters.toRange(command.getRange());
if (command.getWithScores().isPresent()
&& ObjectUtils.nullSafeEquals(command.getWithScores().get(), Boolean.TRUE)) {
if (!isLimited) {
result = (requiresStringConversion
? cmd.zrangebyscoreWithScores(command.getKey().array(), lowerBound.toString(), upperBound.toString())
: cmd.zrangebyscoreWithScores(command.getKey().array(), (Double) lowerBound, (Double) upperBound))
.map(sc -> (Tuple) new DefaultTuple(sc.value, sc.score)).toList();
result = cmd.zrangebyscoreWithScores(command.getKey(), range)
.map(sc -> (Tuple) new DefaultTuple(getBytes(sc.getValue()), sc.getScore())).collectList();
} else {
result = (requiresStringConversion
? cmd.zrangebyscoreWithScores(command.getKey().array(), lowerBound.toString(), upperBound.toString(),
command.getLimit().get().getOffset(), command.getLimit().get().getCount())
: cmd.zrangebyscoreWithScores(command.getKey().array(), (Double) lowerBound, (Double) upperBound,
command.getLimit().get().getOffset(), command.getLimit().get().getCount()))
.map(sc -> (Tuple) new DefaultTuple(sc.value, sc.score)).toList();
result = cmd
.zrangebyscoreWithScores(command.getKey(), range,
LettuceConverters.toLimit(command.getLimit().get()))
.map(sc -> (Tuple) new DefaultTuple(getBytes(sc.getValue()), sc.getScore())).collectList();
}
} else {
if (!isLimited) {
result = (requiresStringConversion
? cmd.zrangebyscore(command.getKey().array(), lowerBound.toString(), upperBound.toString())
: cmd.zrangebyscore(command.getKey().array(), (Double) lowerBound, (Double) upperBound))
.map(value -> (Tuple) new DefaultTuple(value, Double.NaN)).toList();
result = cmd.zrangebyscore(command.getKey(), range)
.map(value -> (Tuple) new DefaultTuple(getBytes(value), Double.NaN)).collectList();
} else {
result = (requiresStringConversion
? cmd.zrangebyscore(command.getKey().array(), lowerBound.toString(), upperBound.toString(),
command.getLimit().get().getOffset(), command.getLimit().get().getCount())
: cmd.zrangebyscore(command.getKey().array(), (Double) lowerBound, (Double) upperBound,
command.getLimit().get().getOffset(), command.getLimit().get().getCount()))
.map(value -> (Tuple) new DefaultTuple(value, Double.NaN)).toList();
result = cmd
.zrangebyscore(command.getKey(), range, LettuceConverters.toLimit(command.getLimit().get()))
.map(value -> (Tuple) new DefaultTuple(getBytes(value), Double.NaN)).collectList();
}
}
} else {
if (command.getWithScores().isPresent() && ObjectUtils.nullSafeEquals(command.getWithScores().get(), Boolean.TRUE)) {
Range<Number> range = ArgumentConverters.toRevRange(command.getRange());
if (command.getWithScores().isPresent()
&& ObjectUtils.nullSafeEquals(command.getWithScores().get(), Boolean.TRUE)) {
if (!isLimited) {
result = (requiresStringConversion
? cmd.zrevrangebyscoreWithScores(command.getKey().array(), lowerBound.toString(), upperBound.toString())
: cmd.zrevrangebyscoreWithScores(command.getKey().array(), (Double) lowerBound, (Double) upperBound))
.map(sc -> (Tuple) new DefaultTuple(sc.value, sc.score)).toList();
result = cmd.zrevrangebyscoreWithScores(command.getKey(), range)
.map(sc -> (Tuple) new DefaultTuple(getBytes(sc.getValue()), sc.getScore())).collectList();
} else {
result = (requiresStringConversion
? cmd.zrevrangebyscoreWithScores(command.getKey().array(), lowerBound.toString(), upperBound.toString(),
command.getLimit().get().getOffset(), command.getLimit().get().getCount())
: cmd.zrevrangebyscoreWithScores(command.getKey().array(), (Double) lowerBound, (Double) upperBound,
command.getLimit().get().getOffset(), command.getLimit().get().getCount()))
.map(sc -> (Tuple) new DefaultTuple(sc.value, sc.score)).toList();
result = cmd
.zrevrangebyscoreWithScores(command.getKey(), range,
LettuceConverters.toLimit(command.getLimit().get()))
.map(sc -> (Tuple) new DefaultTuple(getBytes(sc.getValue()), sc.getScore())).collectList();
}
} else {
if (!isLimited) {
result = (requiresStringConversion
? cmd.zrevrangebyscore(command.getKey().array(), lowerBound.toString(), upperBound.toString())
: cmd.zrevrangebyscore(command.getKey().array(), (Double) lowerBound, (Double) upperBound))
.map(value -> (Tuple) new DefaultTuple(value, Double.NaN)).toList();
result = cmd.zrevrangebyscore(command.getKey(), range)
.map(value -> (Tuple) new DefaultTuple(getBytes(value), Double.NaN)).collectList();
} else {
result = (requiresStringConversion
? cmd.zrevrangebyscore(command.getKey().array(), lowerBound.toString(), upperBound.toString(),
command.getLimit().get().getOffset(), command.getLimit().get().getCount())
: cmd.zrevrangebyscore(command.getKey().array(), (Double) lowerBound, (Double) upperBound,
command.getLimit().get().getOffset(), command.getLimit().get().getCount()))
.map(value -> (Tuple) new DefaultTuple(value, Double.NaN)).toList();
result = cmd
.zrevrangebyscore(command.getKey(), range, LettuceConverters.toLimit(command.getLimit().get()))
.map(value -> (Tuple) new DefaultTuple(getBytes(value), Double.NaN)).collectList();
}
}
}
return LettuceReactiveRedisConnection.<List<Tuple>> monoConverter().convert(result)
.map(value -> new MultiValueResponse<>(command, value));
return result.map(value -> new MultiValueResponse<>(command, value));
}));
}
@@ -322,19 +311,10 @@ public class LettuceReactiveZSetCommands implements ReactiveZSetCommands {
Assert.notNull(command.getKey(), "Key must not be null!");
Assert.notNull(command.getRange(), "Range must not be null!");
Object lowerBound = AgrumentConverters.lowerBoundArgOf(command.getRange());
Object upperBound = AgrumentConverters.upperBoundArgOf(command.getRange());
Range<Number> range = ArgumentConverters.toRange(command.getRange());
Mono<Long> result = cmd.zcount(command.getKey(), range);
Observable<Long> result = Observable.empty();
if (lowerBound instanceof String || upperBound instanceof String) {
result = cmd.zcount(command.getKey().array(), lowerBound.toString(), upperBound.toString());
} else {
result = cmd.zcount(command.getKey().array(), (Double) lowerBound, (Double) upperBound);
}
return LettuceReactiveRedisConnection.<Long> monoConverter().convert(result)
.map(value -> new NumericResponse<>(command, value));
return result.map(value -> new NumericResponse<>(command, value));
}));
}
@@ -349,8 +329,7 @@ public class LettuceReactiveZSetCommands implements ReactiveZSetCommands {
Assert.notNull(command.getKey(), "Key must not be null!");
return LettuceReactiveRedisConnection.<Long> monoConverter().convert(cmd.zcard(command.getKey().array()))
.map(value -> new NumericResponse<>(command, value));
return cmd.zcard(command.getKey()).map(value -> new NumericResponse<>(command, value));
}));
}
@@ -366,9 +345,7 @@ public class LettuceReactiveZSetCommands implements ReactiveZSetCommands {
Assert.notNull(command.getKey(), "Key must not be null!");
Assert.notNull(command.getValue(), "Value must not be null!");
return LettuceReactiveRedisConnection.<Double> monoConverter()
.convert(cmd.zscore(command.getKey().array(), command.getValue().array()))
.map(value -> new NumericResponse<>(command, value));
return cmd.zscore(command.getKey(), command.getValue()).map(value -> new NumericResponse<>(command, value));
}));
}
@@ -385,9 +362,8 @@ public class LettuceReactiveZSetCommands implements ReactiveZSetCommands {
Assert.notNull(command.getKey(), "Key must not be null!");
Assert.notNull(command.getRange(), "Range must not be null!");
return LettuceReactiveRedisConnection
.<Long> monoConverter().convert(cmd.zremrangebyrank(command.getKey().array(),
command.getRange().getLowerBound(), command.getRange().getUpperBound()))
return cmd
.zremrangebyrank(command.getKey(), command.getRange().getLowerBound(), command.getRange().getUpperBound())
.map(value -> new NumericResponse<>(command, value));
}));
}
@@ -405,15 +381,10 @@ public class LettuceReactiveZSetCommands implements ReactiveZSetCommands {
Assert.notNull(command.getKey(), "Key must not be null!");
Assert.notNull(command.getRange(), "Range must not be null!");
Object lowerBound = AgrumentConverters.lowerBoundArgOf(command.getRange());
Object upperBound = AgrumentConverters.upperBoundArgOf(command.getRange());
Range<Number> range = ArgumentConverters.toRange(command.getRange());
Mono<Long> result = cmd.zremrangebyscore(command.getKey(), range);
Observable<Long> result = (lowerBound instanceof String || upperBound instanceof String)
? cmd.zremrangebyscore(command.getKey().array(), lowerBound.toString(), upperBound.toString())
: cmd.zremrangebyscore(command.getKey().array(), (Double) lowerBound, (Double) upperBound);
return LettuceReactiveRedisConnection.<Long> monoConverter().convert(result)
.map(value -> new NumericResponse<>(command, value));
return result.map(value -> new NumericResponse<>(command, value));
}));
}
@@ -431,14 +402,14 @@ public class LettuceReactiveZSetCommands implements ReactiveZSetCommands {
ZStoreArgs args = null;
if (command.getAggregateFunction().isPresent() || !command.getWeights().isEmpty()) {
args = zStoreArgs(command.getAggregateFunction().isPresent() ? command.getAggregateFunction().get() : null, command.getWeights());
args = zStoreArgs(command.getAggregateFunction().isPresent() ? command.getAggregateFunction().get() : null,
command.getWeights());
}
byte[][] sourceKeys = command.getSourceKeys().stream().map(ByteBuffer::array).toArray(size -> new byte[size][]);
Observable<Long> result = args != null ? cmd.zunionstore(command.getKey().array(), args, sourceKeys)
: cmd.zunionstore(command.getKey().array(), sourceKeys);
return LettuceReactiveRedisConnection.<Long> monoConverter().convert(result)
.map(value -> new NumericResponse<>(command, value));
ByteBuffer[] sourceKeys = command.getSourceKeys().stream().toArray(ByteBuffer[]::new);
Mono<Long> result = args != null ? cmd.zunionstore(command.getKey(), args, sourceKeys)
: cmd.zunionstore(command.getKey(), sourceKeys);
return result.map(value -> new NumericResponse<>(command, value));
}));
}
@@ -456,14 +427,14 @@ public class LettuceReactiveZSetCommands implements ReactiveZSetCommands {
ZStoreArgs args = null;
if (command.getAggregateFunction().isPresent() || !command.getWeights().isEmpty()) {
args = zStoreArgs(command.getAggregateFunction().isPresent() ? command.getAggregateFunction().get() : null, command.getWeights());
args = zStoreArgs(command.getAggregateFunction().isPresent() ? command.getAggregateFunction().get() : null,
command.getWeights());
}
byte[][] sourceKeys = command.getSourceKeys().stream().map(ByteBuffer::array).toArray(size -> new byte[size][]);
Observable<Long> result = args != null ? cmd.zinterstore(command.getKey().array(), args, sourceKeys)
: cmd.zinterstore(command.getKey().array(), sourceKeys);
return LettuceReactiveRedisConnection.<Long> monoConverter().convert(result)
.map(value -> new NumericResponse<>(command, value));
ByteBuffer[] sourceKeys = command.getSourceKeys().stream().toArray(ByteBuffer[]::new);
Mono<Long> result = args != null ? cmd.zinterstore(command.getKey(), args, sourceKeys)
: cmd.zinterstore(command.getKey(), sourceKeys);
return result.map(value -> new NumericResponse<>(command, value));
}));
}
@@ -478,33 +449,25 @@ public class LettuceReactiveZSetCommands implements ReactiveZSetCommands {
Assert.notNull(command.getKey(), "Destination key must not be null!");
Observable<byte[]> result = Observable.empty();
Flux<ByteBuffer> result;
String lowerBound = AgrumentConverters.lowerBoundArgOf(command.getRange()).toString();
String upperBound = AgrumentConverters.upperBoundArgOf(command.getRange()).toString();
if (command.getLimit() != null) {
if (ObjectUtils.nullSafeEquals(command.getDirection(), Direction.ASC)) {
result = cmd.zrangebylex(command.getKey().array(), lowerBound, upperBound, command.getLimit().getOffset(),
command.getLimit().getCount());
result = cmd.zrangebylex(command.getKey(), ArgumentConverters.toRange(command.getRange()), LettuceConverters.toLimit(command.getLimit()));
} else {
// TODO: fix when https://github.com/mp911de/lettuce/issues/369 resolved
throw new UnsupportedOperationException("Lettuce does not support ZREVRANGEBYLEX.");
result = cmd.zrevrangebylex(command.getKey(), ArgumentConverters.toRevRange(command.getRange()), LettuceConverters.toLimit(command.getLimit()));
}
} else {
if (ObjectUtils.nullSafeEquals(command.getDirection(), Direction.ASC)) {
result = cmd.zrangebylex(command.getKey().array(), lowerBound, upperBound);
result = cmd.zrangebylex(command.getKey(), ArgumentConverters.toRange(command.getRange()));
} else {
// TODO: fix when https://github.com/mp911de/lettuce/issues/369 resolved
throw new UnsupportedOperationException("Lettuce does not support ZREVRANGEBYLEX.");
result = cmd.zrevrangebylex(command.getKey(), ArgumentConverters.toRevRange(command.getRange()));
}
}
return LettuceReactiveRedisConnection.<List<ByteBuffer>> monoConverter()
.convert(result.map(ByteBuffer::wrap).toList()).map(value -> new MultiValueResponse<>(command, value));
return result.collectList().map(value -> new MultiValueResponse<>(command, value));
}));
}
@@ -525,9 +488,8 @@ public class LettuceReactiveZSetCommands implements ReactiveZSetCommands {
}
}
// TODO: fix when https://github.com/mp911de/lettuce/issues/368 resolved
if (weights != null) {
long[] lg = new long[weights.size()];
double[] lg = new double[weights.size()];
for (int i = 0; i < lg.length; i++) {
lg[i] = weights.get(i).longValue();
}
@@ -536,7 +498,74 @@ public class LettuceReactiveZSetCommands implements ReactiveZSetCommands {
return args;
}
private static byte[] getBytes(ScoredValue<ByteBuffer> scoredValue) {
return scoredValue.optional().map(LettuceReactiveZSetCommands::getBytes).orElse(new byte[0]);
}
private static byte[] getBytes(ByteBuffer byteBuffer) {
byte[] bytes = new byte[byteBuffer.remaining()];
byteBuffer.get(bytes);
return bytes;
}
protected LettuceReactiveRedisConnection getConnection() {
return connection;
}
/**
* @author Christoph Strobl
* @author Mark Paluch
*/
static class ArgumentConverters {
public static <T> Range<T> toRange(org.springframework.data.domain.Range<?> range) {
return Range.from(lowerBoundArgOf(range), upperBoundArgOf(range));
}
public static <T> Range<T> toRevRange(org.springframework.data.domain.Range<?> range) {
return Range.from(upperBoundArgOf(range), lowerBoundArgOf(range));
}
@SuppressWarnings("unchecked")
public static <T> Boundary<T> lowerBoundArgOf(org.springframework.data.domain.Range<?> range) {
return (Boundary<T>) rangeToBoundArgumentConverter(false).convert(range);
}
@SuppressWarnings("unchecked")
public static <T> Boundary<T> upperBoundArgOf(org.springframework.data.domain.Range<?> range) {
return (Boundary<T>) rangeToBoundArgumentConverter(true).convert(range);
}
private static Converter<org.springframework.data.domain.Range<?>, Boundary<?>> rangeToBoundArgumentConverter(
Boolean upper) {
return (source) -> {
// TODO: fix range exclusion pattern when DATACMNS-920 is resolved
DirectFieldAccessFallbackBeanWrapper bw = new DirectFieldAccessFallbackBeanWrapper(source);
Boolean inclusive = upper ? Boolean.valueOf(bw.getPropertyValue("upperInclusive").toString())
: Boolean.valueOf(bw.getPropertyValue("lowerInclusive").toString());
Object value = upper ? source.getUpperBound() : source.getLowerBound();
if (value instanceof Number) {
return inclusive ? Boundary.including((Number) value) : Boundary.excluding((Number) value);
}
if (value instanceof String) {
StringCodec stringCodec = new StringCodec(LettuceCharsets.UTF8);
if (!StringUtils.hasText((String) value) || ObjectUtils.nullSafeEquals(value, "+")
|| ObjectUtils.nullSafeEquals(value, "-")) {
return Boundary.unbounded();
}
return inclusive ? Boundary.including(stringCodec.encodeValue((String) value))
: Boundary.excluding(stringCodec.encodeValue((String) value));
}
return inclusive ? Boundary.including((ByteBuffer) value) : Boundary.excluding((ByteBuffer) value);
};
}
}
}