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

DATAREDIS-719 - Rework FutureResult and implementations for Jedis/Lettuce.

Browse Source 
We now encapsulate deferred results for pipelining and transactions entirely within FutureResult and its subtypes. FutureResult accepts a Supplier<T> for default values, if operations return null and reports whether its result requires conversion. JedisResult and LettuceResult are now top-level classes and no longer inner classes of their connection factories.

Original pull request: #289.
This commit is contained in:
Christoph Strobl
2017-10-13 10:41:38 +02:00
committed by Mark Paluch
parent 6ea78c77dd
commit 435ace8f55
31 changed files with 791 additions and 415 deletions
Download Patch File Download Diff File Expand all files Collapse all files

31
src/main/java/org/springframework/data/redis/connection/DefaultStringRedisConnection.java
View File

@@ -15,8 +15,17 @@
*/
package org.springframework.data.redis.connection;
import java.util.*;
import java.util.ArrayList;
import java.util.Collection;
import java.util.HashMap;
import java.util.LinkedHashMap;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
import java.util.Properties;
import java.util.Queue;
import java.util.Set;
import java.util.concurrent.TimeUnit;
import org.apache.commons.logging.Log;
@@ -725,7 +734,7 @@ public class DefaultStringRedisConnection implements StringRedisConnection, Deco
*/
@Override
public Boolean mSet(Map<byte[], byte[]> tuple) {
return delegate.mSet(tuple);
return convertAndReturn(delegate.mSet(tuple), identityConverter);
}
/*
@@ -923,7 +932,7 @@ public class DefaultStringRedisConnection implements StringRedisConnection, Deco
*/
@Override
public Boolean set(byte[] key, byte[] value) {
return delegate.set(key, value);
return convertAndReturn(delegate.set(key, value), identityConverter);
}
/*
@@ -932,7 +941,7 @@ public class DefaultStringRedisConnection implements StringRedisConnection, Deco
*/
@Override
public Boolean set(byte[] key, byte[] value, Expiration expiration, SetOption option) {
return delegate.set(key, value, expiration, option);
return convertAndReturn(delegate.set(key, value, expiration, option), identityConverter);
}
/*
@@ -959,7 +968,7 @@ public class DefaultStringRedisConnection implements StringRedisConnection, Deco
*/
@Override
public Boolean setEx(byte[] key, long seconds, byte[] value) {
return delegate.setEx(key, seconds, value);
return convertAndReturn(delegate.setEx(key, seconds, value), identityConverter);
}
/*
@@ -968,7 +977,7 @@ public class DefaultStringRedisConnection implements StringRedisConnection, Deco
*/
@Override
public Boolean pSetEx(byte[] key, long milliseconds, byte[] value) {
return delegate.pSetEx(key, milliseconds, value);
return convertAndReturn(delegate.pSetEx(key, milliseconds, value), identityConverter);
}
/*
@@ -2106,7 +2115,7 @@ public class DefaultStringRedisConnection implements StringRedisConnection, Deco
*/
@Override
public Boolean mSetString(Map<String, String> tuple) {
return delegate.mSet(serialize(tuple));
return mSet(serialize(tuple));
}
/*
@@ -2241,7 +2250,7 @@ public class DefaultStringRedisConnection implements StringRedisConnection, Deco
*/
@Override
public Boolean set(String key, String value) {
return delegate.set(serialize(key), serialize(value));
return set(serialize(key), serialize(value));
}
/*
@@ -2268,7 +2277,7 @@ public class DefaultStringRedisConnection implements StringRedisConnection, Deco
*/
@Override
public Boolean setEx(String key, long seconds, String value) {
return delegate.setEx(serialize(key), seconds, serialize(value));
return setEx(serialize(key), seconds, serialize(value));
}
/*
@@ -3441,7 +3450,7 @@ public class DefaultStringRedisConnection implements StringRedisConnection, Deco
*/
@Override
public Set<String> zRangeByLex(String key) {
return this.zRangeByLex(key, Range.unbounded());
return zRangeByLex(key, Range.unbounded());
}
/*
@@ -3450,7 +3459,7 @@ public class DefaultStringRedisConnection implements StringRedisConnection, Deco
*/
@Override
public Set<String> zRangeByLex(String key, Range range) {
return this.zRangeByLex(key, range, null);
return zRangeByLex(key, range, null);
}
/*

69
src/main/java/org/springframework/data/redis/connection/FutureResult.java
View File

@@ -15,6 +15,8 @@
*/
package org.springframework.data.redis.connection;
import java.util.function.Supplier;
import org.springframework.core.convert.converter.Converter;
import org.springframework.lang.Nullable;
@@ -28,23 +30,60 @@ import org.springframework.lang.Nullable;
*/
public abstract class FutureResult<T> {
protected T resultHolder;
private T resultHolder;
private final Supplier<?> defaultConversionResult;
protected boolean status = false;
private boolean status = false;
@SuppressWarnings("rawtypes") //
protected @Nullable Converter converter;
/**
* Create new {@link FutureResult} for given object actually holding the result itself.
*
* @param resultHolder must not be {@literal null}.
*/
public FutureResult(T resultHolder) {
this.resultHolder = resultHolder;
this(resultHolder, val -> val);
}
/**
* Create new {@link FutureResult} for given object actually holding the result itself and a converter capable of
* transforming the result via {@link #convert(Object)}.
*
* @param resultHolder must not be {@literal null}.
* @param converter can be {@literal null} and will be defaulted to an identity converter {@code value -> value} to
* preserve the original value.
*/
@SuppressWarnings("rawtypes")
public FutureResult(T resultHolder, Converter converter) {
this.resultHolder = resultHolder;
this.converter = converter;
public FutureResult(T resultHolder, @Nullable Converter converter) {
this(resultHolder, converter, () -> null);
}
/**
* Create new {@link FutureResult} for given object actually holding the result itself and a converter capable of
* transforming the result via {@link #convert(Object)}.
*
* @param resultHolder must not be {@literal null}.
* @param converter can be {@literal null} and will be defaulted to an identity converter {@code value -> value} to
* preserve the original value.
* @param defaultConversionResult must not be {@literal null}.
* @since 2.1
*/
@SuppressWarnings("rawtypes")
public FutureResult(T resultHolder, @Nullable Converter converter, Supplier<?> defaultConversionResult) {
this.resultHolder = resultHolder;
this.converter = converter != null ? converter : val -> val;
this.defaultConversionResult = defaultConversionResult;
}
/**
* Get the object holding the actual result.
*
* @return never {@literal null}.
* @since 1.1
*/
public T getResultHolder() {
return resultHolder;
}
@@ -60,14 +99,18 @@ public abstract class FutureResult<T> {
public Object convert(@Nullable Object result) {
if (result == null) {
return null;
return computeDefaultResult(result);
}
return (converter != null) ? converter.convert(result) : result;
return computeDefaultResult(converter.convert(result));
}
@Nullable
private Object computeDefaultResult(@Nullable Object source) {
return source != null ? source : defaultConversionResult.get();
}
@SuppressWarnings("rawtypes")
@Nullable
public Converter getConverter() {
return converter;
}
@@ -93,4 +136,12 @@ public abstract class FutureResult<T> {
*/
@Nullable
public abstract Object get();
/**
* Indicate whether or not the actual result needs to be {@link #convert(Object) converted} before handing over.
*
* @return
* @since 2.1
*/
public abstract boolean seeksConversion();
}

2
src/main/java/org/springframework/data/redis/connection/convert/TransactionResultConverter.java
View File

@@ -71,7 +71,7 @@ public class TransactionResultConverter<T> implements Converter<List<Object>, Li
: new RedisSystemException("Error reading future result.", source);
}
if (!(futureResult.isStatus())) {
convertedResults.add(futureResult.convert(result));
convertedResults.add(futureResult.seeksConversion() ? futureResult.convert(result) : result);
}
}

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

@@ -29,6 +29,7 @@ import java.util.Collections;
import java.util.LinkedList;
import java.util.List;
import java.util.Queue;
import java.util.function.Supplier;
import org.springframework.core.convert.converter.Converter;
import org.springframework.dao.DataAccessException;
@@ -39,6 +40,8 @@ import org.springframework.data.redis.RedisConnectionFailureException;
import org.springframework.data.redis.RedisSystemException;
import org.springframework.data.redis.connection.*;
import org.springframework.data.redis.connection.convert.TransactionResultConverter;
import org.springframework.data.redis.connection.jedis.JedisResult.JedisResultBuilder;
import org.springframework.data.redis.connection.jedis.JedisResult.JedisStatusResult;
import org.springframework.lang.Nullable;
import org.springframework.util.Assert;
import org.springframework.util.CollectionUtils;
@@ -75,43 +78,9 @@ public class JedisConnection extends AbstractRedisConnection {
private final int dbIndex;
private final String clientName;
private boolean convertPipelineAndTxResults = true;
private List<FutureResult<Response<?>>> pipelinedResults = new ArrayList<>();
private List<JedisResult> pipelinedResults = new ArrayList<>();
private Queue<FutureResult<Response<?>>> txResults = new LinkedList<>();
class JedisResult extends FutureResult<Response<?>> {
public <T> JedisResult(Response<T> resultHolder, Converter<T, ?> converter) {
super(resultHolder, converter);
}
public <T> JedisResult(Response<T> resultHolder) {
super(resultHolder);
}
@SuppressWarnings("unchecked")
public Object get() {
Object raw = resultHolder.get();
if (!convertPipelineAndTxResults || raw == null) {
return raw;
}
return converter != null ? converter.convert(raw) : raw;
}
}
private class JedisStatusResult extends JedisResult {
public JedisStatusResult(Response<?> resultHolder) {
super(resultHolder);
setStatus(true);
}
public <T> JedisStatusResult(Response<T> resultHolder, Converter<T, ?> converter) {
super(resultHolder, converter);
setStatus(true);
}
}
/**
* Constructs a new <code>JedisConnection</code> instance.
*
@@ -285,9 +254,9 @@ public class JedisConnection extends AbstractRedisConnection {
Response<Object> result = JedisClientUtils
.getResponse(isPipelined() ? getRequiredPipeline() : getRequiredTransaction());
if (isPipelined()) {
pipeline(new JedisResult(result));
pipeline(newJedisResult(result));
} else {
transaction(new JedisResult(result));
transaction(newJedisResult(result));
}
return null;
}
@@ -414,11 +383,13 @@ public class JedisConnection extends AbstractRedisConnection {
List<Object> results = new ArrayList<>();
getRequiredPipeline().sync();
Exception cause = null;
for (FutureResult<Response<?>> result : pipelinedResults) {
for (JedisResult result : pipelinedResults) {
try {
Object data = result.get();
if (!convertPipelineAndTxResults || !(result.isStatus())) {
results.add(data);
if (!result.isStatus()) {
results.add(result.seeksConversion() ? result.convert(data) : data);
}
} catch (JedisDataException e) {
DataAccessException dataAccessException = convertJedisAccessException(e);
@@ -439,7 +410,7 @@ public class JedisConnection extends AbstractRedisConnection {
return results;
}
void pipeline(FutureResult<Response<?>> result) {
void pipeline(JedisResult result) {
if (isQueueing()) {
transaction(result);
} else {
@@ -459,11 +430,11 @@ public class JedisConnection extends AbstractRedisConnection {
public byte[] echo(byte[] message) {
try {
if (isPipelined()) {
pipeline(new JedisResult(getRequiredPipeline().echo(message)));
pipeline(newJedisResult(getRequiredPipeline().echo(message)));
return null;
}
if (isQueueing()) {
transaction(new JedisResult(getRequiredTransaction().echo(message)));
transaction(newJedisResult(getRequiredTransaction().echo(message)));
return null;
}
return jedis.echo(message);
@@ -480,11 +451,11 @@ public class JedisConnection extends AbstractRedisConnection {
public String ping() {
try {
if (isPipelined()) {
pipeline(new JedisResult(getRequiredPipeline().ping()));
pipeline(newJedisResult(getRequiredPipeline().ping()));
return null;
}
if (isQueueing()) {
transaction(new JedisResult(getRequiredTransaction().ping()));
transaction(newJedisResult(getRequiredTransaction().ping()));
return null;
}
return jedis.ping();
@@ -501,7 +472,7 @@ public class JedisConnection extends AbstractRedisConnection {
public void discard() {
try {
if (isPipelined()) {
pipeline(new JedisStatusResult(getRequiredPipeline().discard()));
pipeline(newStatusResult(getRequiredPipeline().discard()));
return;
}
getRequiredTransaction().discard();
@@ -521,7 +492,7 @@ public class JedisConnection extends AbstractRedisConnection {
public List<Object> exec() {
try {
if (isPipelined()) {
pipeline(new JedisResult(getRequiredPipeline().exec(),
pipeline(newJedisResult(getRequiredPipeline().exec(),
new TransactionResultConverter<>(new LinkedList<>(txResults), JedisConverters.exceptionConverter())));
return null;
}
@@ -529,8 +500,10 @@ public class JedisConnection extends AbstractRedisConnection {
if (transaction == null) {
throw new InvalidDataAccessApiUsageException("No ongoing transaction. Did you forget to call multi?");
}
List<Object> results = transaction.exec();
return convertPipelineAndTxResults && !CollectionUtils.isEmpty(results)
return !CollectionUtils.isEmpty(results)
? new TransactionResultConverter<>(txResults, JedisConverters.exceptionConverter()).convert(results)
: results;
} catch (Exception ex) {
@@ -578,19 +551,23 @@ public class JedisConnection extends AbstractRedisConnection {
}
JedisResult newJedisResult(Response<?> response) {
return new JedisResult(response);
return JedisResultBuilder.forResponse(response).build();
}
<T> JedisResult newJedisResult(Response<T> response, Converter<T, ?> converter) {
return new JedisResult(response, converter);
return JedisResultBuilder.forResponse(response).mappedWith(converter)
.convertPipelineAndTxResults(convertPipelineAndTxResults).build();
}
<T> JedisResult newJedisResult(Response<T> response, Converter<T, ?> converter, Supplier<?> defaultValue) {
return JedisResultBuilder.forResponse(response).mappedWith(converter)
.convertPipelineAndTxResults(convertPipelineAndTxResults).defaultNullTo(defaultValue).build();
}
JedisStatusResult newStatusResult(Response<?> response) {
return new JedisStatusResult(response);
}
<T> JedisStatusResult newStatusResult(Response<T> response, Converter<T, ?> converter) {
return new JedisStatusResult(response, converter);
return JedisResultBuilder.forResponse(response).buildStatusResult();
}
/*
@@ -621,11 +598,11 @@ public class JedisConnection extends AbstractRedisConnection {
public void select(int dbIndex) {
try {
if (isPipelined()) {
pipeline(new JedisStatusResult(getRequiredPipeline().select(dbIndex)));
pipeline(newStatusResult(getRequiredPipeline().select(dbIndex)));
return;
}
if (isQueueing()) {
transaction(new JedisStatusResult(getRequiredTransaction().select(dbIndex)));
transaction(newStatusResult(getRequiredTransaction().select(dbIndex)));
return;
}
jedis.select(dbIndex);
@@ -659,7 +636,7 @@ public class JedisConnection extends AbstractRedisConnection {
try {
for (byte[] key : keys) {
if (isPipelined()) {
pipeline(new JedisStatusResult(getRequiredPipeline().watch(key)));
pipeline(newStatusResult(getRequiredPipeline().watch(key)));
} else {
jedis.watch(key);
}
@@ -681,11 +658,11 @@ public class JedisConnection extends AbstractRedisConnection {
public Long publish(byte[] channel, byte[] message) {
try {
if (isPipelined()) {
pipeline(new JedisResult(getRequiredPipeline().publish(channel, message)));
pipeline(newJedisResult(getRequiredPipeline().publish(channel, message)));
return null;
}
if (isQueueing()) {
transaction(new JedisResult(getRequiredTransaction().publish(channel, message)));
transaction(newJedisResult(getRequiredTransaction().publish(channel, message)));
return null;
}
return jedis.publish(channel, message);

23
src/main/java/org/springframework/data/redis/connection/jedis/JedisGeoCommands.java
View File

@@ -32,7 +32,6 @@ import org.springframework.data.geo.Metric;
import org.springframework.data.geo.Point;
import org.springframework.data.redis.connection.RedisGeoCommands;
import org.springframework.data.redis.connection.convert.ListConverter;
import org.springframework.data.redis.connection.jedis.JedisConnection.JedisResult;
import org.springframework.util.Assert;
/**
@@ -63,9 +62,8 @@ class JedisGeoCommands implements RedisGeoCommands {
return null;
}
if (isQueueing()) {
transaction(
connection
.newJedisResult(connection.getRequiredTransaction().geoadd(key, point.getX(), point.getY(), member)));
transaction(connection
.newJedisResult(connection.getRequiredTransaction().geoadd(key, point.getX(), point.getY(), member)));
return null;
}
@@ -159,9 +157,8 @@ class JedisGeoCommands implements RedisGeoCommands {
return null;
}
if (isQueueing()) {
transaction(
connection.newJedisResult(connection.getRequiredTransaction().geodist(key, member1, member2),
distanceConverter));
transaction(connection.newJedisResult(connection.getRequiredTransaction().geodist(key, member1, member2),
distanceConverter));
return null;
}
@@ -194,8 +191,7 @@ class JedisGeoCommands implements RedisGeoCommands {
}
if (isQueueing()) {
transaction(connection.newJedisResult(
connection.getRequiredTransaction().geodist(key, member1, member2, geoUnit),
distanceConverter));
connection.getRequiredTransaction().geodist(key, member1, member2, geoUnit), distanceConverter));
return null;
}
@@ -323,8 +319,7 @@ class JedisGeoCommands implements RedisGeoCommands {
}
if (isQueueing()) {
transaction(connection.newJedisResult(connection.getRequiredTransaction().georadius(key,
within.getCenter().getX(),
within.getCenter().getY(), within.getRadius().getValue(),
within.getCenter().getX(), within.getCenter().getY(), within.getRadius().getValue(),
JedisConverters.toGeoUnit(within.getRadius().getMetric()), geoRadiusParam), converter));
return null;
}
@@ -396,10 +391,8 @@ class JedisGeoCommands implements RedisGeoCommands {
return null;
}
if (isQueueing()) {
transaction(connection.newJedisResult(
connection.getRequiredTransaction().georadiusByMember(key, member, radius.getValue(), geoUnit,
geoRadiusParam),
converter));
transaction(connection.newJedisResult(connection.getRequiredTransaction().georadiusByMember(key, member,
radius.getValue(), geoUnit, geoRadiusParam), converter));
return null;
}
return converter

1
src/main/java/org/springframework/data/redis/connection/jedis/JedisHashCommands.java
View File

@@ -26,7 +26,6 @@ import java.util.Map.Entry;
import java.util.Set;
import org.springframework.data.redis.connection.RedisHashCommands;
import org.springframework.data.redis.connection.jedis.JedisConnection.JedisResult;
import org.springframework.data.redis.core.Cursor;
import org.springframework.data.redis.core.KeyBoundCursor;
import org.springframework.data.redis.core.ScanIteration;

1
src/main/java/org/springframework/data/redis/connection/jedis/JedisHyperLogLogCommands.java
View File

@@ -19,7 +19,6 @@ import lombok.NonNull;
import lombok.RequiredArgsConstructor;
import org.springframework.data.redis.connection.RedisHyperLogLogCommands;
import org.springframework.data.redis.connection.jedis.JedisConnection.JedisResult;
import org.springframework.util.Assert;
/**

7
src/main/java/org/springframework/data/redis/connection/jedis/JedisKeyCommands.java
View File

@@ -28,7 +28,6 @@ import org.springframework.data.redis.connection.DataType;
import org.springframework.data.redis.connection.RedisKeyCommands;
import org.springframework.data.redis.connection.SortParameters;
import org.springframework.data.redis.connection.convert.Converters;
import org.springframework.data.redis.connection.jedis.JedisConnection.JedisResult;
import org.springframework.data.redis.core.Cursor;
import org.springframework.data.redis.core.ScanCursor;
import org.springframework.data.redis.core.ScanIteration;
@@ -83,13 +82,11 @@ class JedisKeyCommands implements RedisKeyCommands {
try {
if (isPipelined()) {
pipeline(
connection.newJedisResult(connection.getRequiredPipeline().exists(keys)));
pipeline(connection.newJedisResult(connection.getRequiredPipeline().exists(keys)));
return null;
}
if (isQueueing()) {
transaction(connection
.newJedisResult(connection.getRequiredTransaction().exists(keys)));
transaction(connection.newJedisResult(connection.getRequiredTransaction().exists(keys)));
return null;
}
return connection.getJedis().exists(keys);

1
src/main/java/org/springframework/data/redis/connection/jedis/JedisListCommands.java
View File

@@ -23,7 +23,6 @@ import java.util.ArrayList;
import java.util.List;
import org.springframework.data.redis.connection.RedisListCommands;
import org.springframework.data.redis.connection.jedis.JedisConnection.JedisResult;
import org.springframework.util.Assert;
/**

140
src/main/java/org/springframework/data/redis/connection/jedis/JedisResult.java Normal file
View File

@@ -0,0 +1,140 @@
/*
* Copyright 2017 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.data.redis.connection.jedis;
import redis.clients.jedis.Response;
import java.util.function.Supplier;
import org.springframework.core.convert.converter.Converter;
import org.springframework.data.redis.connection.FutureResult;
import org.springframework.lang.Nullable;
/**
* Jedis specific {@link FutureResult} implementation. <br />
*
* @author Costin Leau
* @author Jennifer Hickey
* @author Christoph Strobl
* @author Mark Paluch
* @since 2.1
*/
class JedisResult<T, S> extends FutureResult<Response<?>> {
private final boolean convertPipelineAndTxResults;
<T> JedisResult(Response<T> resultHolder) {
this(resultHolder, false, null);
}
<T> JedisResult(Response<T> resultHolder, boolean convertPipelineAndTxResults, @Nullable Converter<T, ?> converter) {
this(resultHolder, null, convertPipelineAndTxResults, converter);
}
<T> JedisResult(Response<T> resultHolder, Supplier<S> defaultReturnValue, boolean convertPipelineAndTxResults,
@Nullable Converter<T, ?> converter) {
super(resultHolder, converter, defaultReturnValue);
this.convertPipelineAndTxResults = convertPipelineAndTxResults;
}
/*
* (non-Javadoc)
* @see org.springframework.data.redis.connection.FutureResult#get()
* @return
*/
@Nullable
@Override
public T get() {
return (T) getResultHolder().get();
}
/*
* (non-Javadoc)
* @see org.springframework.data.redis.connection.FutureResult#seeksConversion()
* @return
*/
public boolean seeksConversion() {
return convertPipelineAndTxResults && converter != null;
}
/**
* Jedis specific {@link FutureResult} implementation of a throw away status result.
*/
static class JedisStatusResult extends JedisResult {
<T> JedisStatusResult(Response<T> resultHolder, Converter<T, ?> converter) {
super(resultHolder, false, converter);
setStatus(true);
}
}
/**
* Builder for constructing {@link JedisResult}.
*
* @param <T>
* @param <S>
* @since 2.1
*/
static class JedisResultBuilder<T, S> {
private final Response<T> response;
private Converter<T, ?> converter;
private boolean convertPipelineAndTxResults = false;
private Supplier<?> nullValueDefault = () -> null;
JedisResultBuilder(Response<T> response) {
this.response = response;
this.converter = (source) -> source;
}
static <T> JedisResultBuilder<T, ?> forResponse(Response<T> response) {
return new JedisResultBuilder<>(response);
}
<S> JedisResultBuilder<T, S> mappedWith(Converter<T, S> converter) {
this.converter = converter;
return (JedisResultBuilder<T, S>) this;
}
<S> JedisResultBuilder<T, S> defaultNullTo(S value) {
return (defaultNullTo(() -> value));
}
<S> JedisResultBuilder<T, S> defaultNullTo(Supplier<S> value) {
this.nullValueDefault = value;
return (JedisResultBuilder<T, S>) this;
}
JedisResultBuilder<T, S> convertPipelineAndTxResults(boolean flag) {
convertPipelineAndTxResults = flag;
return this;
}
JedisResult<T, S> build() {
return new JedisResult(response, nullValueDefault, convertPipelineAndTxResults, converter);
}
JedisStatusResult buildStatusResult() {
return new JedisStatusResult(response, converter);
}
}
}

1
src/main/java/org/springframework/data/redis/connection/jedis/JedisServerCommands.java
View File

@@ -25,7 +25,6 @@ import org.springframework.data.redis.connection.RedisNode;
import org.springframework.data.redis.connection.RedisServerCommands;
import org.springframework.data.redis.connection.ReturnType;
import org.springframework.data.redis.connection.convert.Converters;
import org.springframework.data.redis.connection.jedis.JedisConnection.JedisResult;
import org.springframework.data.redis.core.types.RedisClientInfo;
import org.springframework.lang.Nullable;
import org.springframework.util.Assert;

1
src/main/java/org/springframework/data/redis/connection/jedis/JedisSetCommands.java
View File

@@ -24,7 +24,6 @@ import java.util.List;
import java.util.Set;
import org.springframework.data.redis.connection.RedisSetCommands;
import org.springframework.data.redis.connection.jedis.JedisConnection.JedisResult;
import org.springframework.data.redis.core.Cursor;
import org.springframework.data.redis.core.KeyBoundCursor;
import org.springframework.data.redis.core.ScanIteration;

41
src/main/java/org/springframework/data/redis/connection/jedis/JedisStringCommands.java
View File

@@ -24,7 +24,6 @@ import java.util.concurrent.TimeUnit;
import org.springframework.data.redis.connection.RedisStringCommands;
import org.springframework.data.redis.connection.convert.Converters;
import org.springframework.data.redis.connection.jedis.JedisConnection.JedisResult;
import org.springframework.data.redis.core.types.Expiration;
import org.springframework.util.Assert;
import org.springframework.util.ObjectUtils;
@@ -126,12 +125,12 @@ class JedisStringCommands implements RedisStringCommands {
try {
if (isPipelined()) {
pipeline(connection.newStatusResult(connection.getRequiredPipeline().set(key, value),
pipeline(connection.newJedisResult(connection.getRequiredPipeline().set(key, value),
Converters.stringToBooleanConverter()));
return null;
}
if (isQueueing()) {
transaction(connection.newStatusResult(connection.getRequiredTransaction().set(key, value),
transaction(connection.newJedisResult(connection.getRequiredTransaction().set(key, value),
Converters.stringToBooleanConverter()));
return null;
}
@@ -165,14 +164,14 @@ class JedisStringCommands implements RedisStringCommands {
if (isPipelined()) {
pipeline(connection.newStatusResult(connection.getRequiredPipeline().set(key, value, nxxx),
Converters.stringToBooleanConverter()));
pipeline(connection.newJedisResult(connection.getRequiredPipeline().set(key, value, nxxx),
Converters.stringToBooleanConverter(), () -> false));
return null;
}
if (isQueueing()) {
transaction(connection.newStatusResult(connection.getRequiredTransaction().set(key, value, nxxx),
Converters.stringToBooleanConverter()));
transaction(connection.newJedisResult(connection.getRequiredTransaction().set(key, value, nxxx),
Converters.stringToBooleanConverter(), () -> false));
return null;
}
@@ -205,9 +204,9 @@ class JedisStringCommands implements RedisStringCommands {
"Expiration.expirationTime must be less than Integer.MAX_VALUE for pipeline in Jedis.");
}
pipeline(connection.newStatusResult(
pipeline(connection.newJedisResult(
connection.getRequiredPipeline().set(key, value, nxxx, expx, (int) expiration.getExpirationTime()),
Converters.stringToBooleanConverter()));
Converters.stringToBooleanConverter(), () -> false));
return null;
}
if (isQueueing()) {
@@ -217,9 +216,9 @@ class JedisStringCommands implements RedisStringCommands {
"Expiration.expirationTime must be less than Integer.MAX_VALUE for transactions in Jedis.");
}
transaction(connection.newStatusResult(
transaction(connection.newJedisResult(
connection.getRequiredTransaction().set(key, value, nxxx, expx, (int) expiration.getExpirationTime()),
Converters.stringToBooleanConverter()));
Converters.stringToBooleanConverter(), () -> false));
return null;
}
@@ -277,13 +276,13 @@ class JedisStringCommands implements RedisStringCommands {
try {
if (isPipelined()) {
pipeline(connection.newStatusResult(connection.getRequiredPipeline().setex(key, (int) seconds, value),
Converters.stringToBooleanConverter()));
pipeline(connection.newJedisResult(connection.getRequiredPipeline().setex(key, (int) seconds, value),
Converters.stringToBooleanConverter(), () -> false));
return null;
}
if (isQueueing()) {
transaction(connection.newStatusResult(connection.getRequiredTransaction().setex(key, (int) seconds, value),
Converters.stringToBooleanConverter()));
transaction(connection.newJedisResult(connection.getRequiredTransaction().setex(key, (int) seconds, value),
Converters.stringToBooleanConverter(), () -> false));
return null;
}
return Converters.stringToBoolean(connection.getJedis().setex(key, (int) seconds, value));
@@ -304,13 +303,13 @@ class JedisStringCommands implements RedisStringCommands {
try {
if (isPipelined()) {
pipeline(connection.newStatusResult(connection.getRequiredPipeline().psetex(key, milliseconds, value),
Converters.stringToBooleanConverter()));
pipeline(connection.newJedisResult(connection.getRequiredPipeline().psetex(key, milliseconds, value),
Converters.stringToBooleanConverter(), () -> false));
return null;
}
if (isQueueing()) {
transaction(connection.newStatusResult(connection.getRequiredTransaction().psetex(key, milliseconds, value),
Converters.stringToBooleanConverter()));
transaction(connection.newJedisResult(connection.getRequiredTransaction().psetex(key, milliseconds, value),
Converters.stringToBooleanConverter(), () -> false));
return null;
}
return Converters.stringToBoolean(connection.getJedis().psetex(key, milliseconds, value));
@@ -330,13 +329,13 @@ class JedisStringCommands implements RedisStringCommands {
try {
if (isPipelined()) {
pipeline(connection.newStatusResult(connection.getRequiredPipeline().mset(JedisConverters.toByteArrays(tuples)),
pipeline(connection.newJedisResult(connection.getRequiredPipeline().mset(JedisConverters.toByteArrays(tuples)),
Converters.stringToBooleanConverter()));
return null;
}
if (isQueueing()) {
transaction(
connection.newStatusResult(connection.getRequiredTransaction().mset(JedisConverters.toByteArrays(tuples)),
connection.newJedisResult(connection.getRequiredTransaction().mset(JedisConverters.toByteArrays(tuples)),
Converters.stringToBooleanConverter()));
return null;
}

1
src/main/java/org/springframework/data/redis/connection/jedis/JedisZSetCommands.java
View File

@@ -25,7 +25,6 @@ import java.nio.charset.StandardCharsets;
import java.util.Set;
import org.springframework.data.redis.connection.RedisZSetCommands;
import org.springframework.data.redis.connection.jedis.JedisConnection.JedisResult;
import org.springframework.data.redis.core.Cursor;
import org.springframework.data.redis.core.KeyBoundCursor;
import org.springframework.data.redis.core.ScanIteration;

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

@@ -53,6 +53,7 @@ import java.util.Queue;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
import java.util.function.Supplier;
import org.springframework.beans.BeanUtils;
import org.springframework.core.convert.converter.Converter;
@@ -64,6 +65,10 @@ import org.springframework.data.redis.FallbackExceptionTranslationStrategy;
import org.springframework.data.redis.connection.*;
import org.springframework.data.redis.connection.convert.TransactionResultConverter;
import org.springframework.data.redis.connection.lettuce.LettuceConnectionProvider.TargetAware;
import org.springframework.data.redis.connection.lettuce.LettuceResult.LettuceResultBuilder;
import org.springframework.data.redis.connection.lettuce.LettuceResult.LettuceStatusResult;
import org.springframework.data.redis.connection.lettuce.LettuceResult.LettuceTxResult;
import org.springframework.data.redis.connection.lettuce.LettuceResult.LettuceTxStatusResult;
import org.springframework.data.redis.core.RedisCommand;
import org.springframework.data.redis.core.ScanOptions;
import org.springframework.lang.Nullable;
@@ -110,106 +115,46 @@ public class LettuceConnection extends AbstractRedisConnection {
/** flag indicating whether the connection needs to be dropped or not */
private boolean convertPipelineAndTxResults = true;
@SuppressWarnings("rawtypes")
class LettuceResult extends FutureResult<io.lettuce.core.protocol.RedisCommand<?, ?, ?>> {
public <T> LettuceResult(Future<T> resultHolder, Converter<T, ?> converter) {
super((io.lettuce.core.protocol.RedisCommand) resultHolder, converter);
}
public LettuceResult(Future resultHolder) {
super((io.lettuce.core.protocol.RedisCommand) resultHolder);
}
@SuppressWarnings("unchecked")
@Override
public Object get() {
try {
if (convertPipelineAndTxResults && converter != null) {
return converter.convert(resultHolder.getOutput().get());
}
return resultHolder.getOutput().get();
} catch (Exception e) {
throw LettuceConnection.this.convertLettuceAccessException(e);
}
}
}
LettuceResult newLettuceResult(Future<?> resultHolder) {
return new LettuceResult(resultHolder);
return newLettuceResult(resultHolder, (val) -> val);
}
<T> LettuceResult newLettuceResult(Future<T> resultHolder, Converter<T, ?> converter) {
return new LettuceResult(resultHolder, converter);
return LettuceResultBuilder.forResponse(resultHolder).mappedWith(converter)
.convertPipelineAndTxResults(convertPipelineAndTxResults).build();
}
private class LettuceStatusResult extends LettuceResult {
@SuppressWarnings("rawtypes")
LettuceStatusResult(Future resultHolder) {
super(resultHolder);
setStatus(true);
}
<T> LettuceResult newLettuceResult(Future<T> resultHolder, Converter<T, ?> converter, Supplier<?> defaultValue) {
<T> LettuceStatusResult(Future<T> resultHolder, Converter<T, ?> converter) {
super(resultHolder, converter);
setStatus(true);
}
return LettuceResultBuilder.forResponse(resultHolder).mappedWith(converter)
.convertPipelineAndTxResults(convertPipelineAndTxResults).defaultNullTo(defaultValue).build();
}
LettuceStatusResult newLettuceStatusResult(Future<?> resultHolder) {
return new LettuceStatusResult(resultHolder);
}
<T> LettuceStatusResult newLettuceStatusResult(Future<T> resultHolder, Converter<T, ?> converter) {
return new LettuceStatusResult(resultHolder, converter);
}
class LettuceTxResult extends FutureResult<Object> {
public LettuceTxResult(Object resultHolder, Converter<?, ?> converter) {
super(resultHolder, converter);
}
public LettuceTxResult(Object resultHolder) {
super(resultHolder);
}
@SuppressWarnings("unchecked")
@Override
public Object get() {
if (convertPipelineAndTxResults && converter != null) {
return converter.convert(resultHolder);
}
return resultHolder;
}
}
LettuceTxResult newLettuceTxResult(Object resultHolder) {
return new LettuceTxResult(resultHolder);
return newLettuceTxResult(resultHolder, (val) -> val);
}
LettuceTxResult newLettuceTxResult(Object resultHolder, Converter<?, ?> converter) {
return new LettuceTxResult(resultHolder, converter);
<T> LettuceTxResult<T> newLettuceTxResult(Object resultHolder, Converter converter) {
return LettuceResultBuilder.forResponse(resultHolder).mappedWith(converter)
.convertPipelineAndTxResults(convertPipelineAndTxResults).buildTxResult();
}
private class LettuceTxStatusResult extends LettuceTxResult {
LettuceTxStatusResult(Object resultHolder) {
super(resultHolder);
setStatus(true);
}
<T> LettuceTxResult<T> newLettuceTxResult(Object resultHolder, Converter converter, Supplier<?> defaultValue) {
LettuceTxStatusResult(Object resultHolder, Converter converter) {
super(resultHolder, converter);
setStatus(true);
}
return LettuceResultBuilder.forResponse(resultHolder).mappedWith(converter)
.convertPipelineAndTxResults(convertPipelineAndTxResults).defaultNullTo(defaultValue).buildTxResult();
}
LettuceTxStatusResult newLettuceTxStatusResult(Object resultHolder) {
return new LettuceTxStatusResult(resultHolder);
}
LettuceTxStatusResult newLettuceTxStatusResult(Object resultHolder, Converter<?, ?> converter) {
return new LettuceTxStatusResult(resultHolder, converter);
}
private class LettuceTransactionResultConverter<T> extends TransactionResultConverter<T> {
public LettuceTransactionResultConverter(Queue<FutureResult<T>> txResults,
Converter<Exception, DataAccessException> exceptionConverter) {
@@ -440,7 +385,7 @@ public class LettuceConnection extends AbstractRedisConnection {
/**
* 'Native' or 'raw' execution of the given command along-side the given arguments.
*
* @see RedisCommands#execute(String, byte[]...)
* @see RedisConnection#execute(String, byte[]...)
* @param command Command to execute
* @param commandOutputTypeHint Type of Output to use, may be (may be {@literal null}).
* @param args Possible command arguments (may be {@literal null})
@@ -469,11 +414,11 @@ public class LettuceConnection extends AbstractRedisConnection {
Command cmd = new Command(commandType, expectedOutput, cmdArg);
if (isPipelined()) {
pipeline(new LettuceResult(connectionImpl.dispatch(cmd.getType(), cmd.getOutput(), cmd.getArgs())));
pipeline(newLettuceResult(connectionImpl.dispatch(cmd.getType(), cmd.getOutput(), cmd.getArgs())));
return null;
} else if (isQueueing()) {
transaction(new LettuceTxResult(connectionImpl.dispatch(cmd.getType(), cmd.getOutput(), cmd.getArgs())));
transaction(newLettuceTxResult(connectionImpl.dispatch(cmd.getType(), cmd.getOutput(), cmd.getArgs())));
return null;
} else {
return await(connectionImpl.dispatch(cmd.getType(), cmd.getOutput(), cmd.getArgs()));
@@ -546,7 +491,7 @@ public class LettuceConnection extends AbstractRedisConnection {
if (isPipelined) {
isPipelined = false;
List<io.lettuce.core.protocol.RedisCommand<?, ?, ?>> futures = new ArrayList<>();
for (LettuceResult result : ppline) {
for (LettuceResult<?, ?> result : ppline) {
futures.add(result.getResultHolder());
}
@@ -559,7 +504,7 @@ public class LettuceConnection extends AbstractRedisConnection {
Exception problem = null;
if (done) {
for (LettuceResult result : ppline) {
for (LettuceResult<?, ?> result : ppline) {
if (result.getResultHolder().getOutput().hasError()) {
@@ -569,10 +514,10 @@ public class LettuceConnection extends AbstractRedisConnection {
problem = err;
}
results.add(err);
} else if (!convertPipelineAndTxResults || !(result.isStatus())) {
} else if (!result.isStatus()) {
try {
results.add(result.get());
results.add(result.seeksConversion() ? result.convert(result.get()) : result.get());
} catch (DataAccessException e) {
if (problem == null) {
problem = e;
@@ -609,7 +554,7 @@ public class LettuceConnection extends AbstractRedisConnection {
public byte[] echo(byte[] message) {
try {
if (isPipelined()) {
pipeline(new LettuceResult(getAsyncConnection().echo(message)));
pipeline(newLettuceResult(getAsyncConnection().echo(message)));
return null;
}
if (isQueueing()) {
@@ -626,7 +571,7 @@ public class LettuceConnection extends AbstractRedisConnection {
public String ping() {
try {
if (isPipelined()) {
pipeline(new LettuceResult(getAsyncConnection().ping()));
pipeline(newLettuceResult(getAsyncConnection().ping()));
return null;
}
if (isQueueing()) {
@@ -658,7 +603,9 @@ public class LettuceConnection extends AbstractRedisConnection {
@Override
@SuppressWarnings({ "unchecked", "rawtypes" })
public List<Object> exec() {
isMulti = false;
try {
if (isPipelined()) {
RedisFuture<TransactionResult> exec = ((RedisAsyncCommands) getAsyncDedicatedConnection()).exec();
@@ -666,8 +613,8 @@ public class LettuceConnection extends AbstractRedisConnection {
LettuceTransactionResultConverter resultConverter = new LettuceTransactionResultConverter(
new LinkedList<>(txResults), LettuceConverters.exceptionConverter());
pipeline(new LettuceResult(exec,
source -> resultConverter.convert(LettuceConverters.transactionResultUnwrapper().convert(source))));
pipeline(newLettuceResult(exec, source -> resultConverter
.convert(LettuceConverters.transactionResultUnwrapper().convert((TransactionResult) source))));
return null;
}
@@ -768,11 +715,11 @@ public class LettuceConnection extends AbstractRedisConnection {
public Long publish(byte[] channel, byte[] message) {
try {
if (isPipelined()) {
pipeline(new LettuceResult(getAsyncConnection().publish(channel, message)));
pipeline(newLettuceResult(getAsyncConnection().publish(channel, message)));
return null;
}
if (isQueueing()) {
transaction(new LettuceTxResult(getConnection().publish(channel, message)));
transaction(newLettuceTxResult(getConnection().publish(channel, message)));
return null;
}
return getConnection().publish(channel, message);

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

@@ -40,8 +40,7 @@ import org.springframework.data.geo.Metric;
import org.springframework.data.geo.Point;
import org.springframework.data.redis.connection.RedisGeoCommands;
import org.springframework.data.redis.connection.convert.ListConverter;
import org.springframework.data.redis.connection.lettuce.LettuceConnection.LettuceResult;
import org.springframework.data.redis.connection.lettuce.LettuceConnection.LettuceTxResult;
import org.springframework.data.redis.connection.lettuce.LettuceResult.LettuceTxResult;
import org.springframework.lang.Nullable;
import org.springframework.util.Assert;

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

@@ -29,8 +29,7 @@ import java.util.Set;
import org.springframework.dao.DataAccessException;
import org.springframework.data.redis.connection.RedisHashCommands;
import org.springframework.data.redis.connection.lettuce.LettuceConnection.LettuceResult;
import org.springframework.data.redis.connection.lettuce.LettuceConnection.LettuceTxResult;
import org.springframework.data.redis.connection.lettuce.LettuceResult.LettuceTxResult;
import org.springframework.data.redis.core.Cursor;
import org.springframework.data.redis.core.KeyBoundCursor;
import org.springframework.data.redis.core.ScanIteration;

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

@@ -24,8 +24,7 @@ import lombok.RequiredArgsConstructor;
import org.springframework.dao.DataAccessException;
import org.springframework.data.redis.connection.RedisHyperLogLogCommands;
import org.springframework.data.redis.connection.lettuce.LettuceConnection.LettuceResult;
import org.springframework.data.redis.connection.lettuce.LettuceConnection.LettuceTxResult;
import org.springframework.data.redis.connection.lettuce.LettuceResult.LettuceTxResult;
import org.springframework.util.Assert;
/**

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

@@ -32,8 +32,7 @@ import org.springframework.data.redis.connection.DataType;
import org.springframework.data.redis.connection.RedisKeyCommands;
import org.springframework.data.redis.connection.SortParameters;
import org.springframework.data.redis.connection.convert.Converters;
import org.springframework.data.redis.connection.lettuce.LettuceConnection.LettuceResult;
import org.springframework.data.redis.connection.lettuce.LettuceConnection.LettuceTxResult;
import org.springframework.data.redis.connection.lettuce.LettuceResult.LettuceTxResult;
import org.springframework.data.redis.core.Cursor;
import org.springframework.data.redis.core.ScanCursor;
import org.springframework.data.redis.core.ScanIteration;

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

@@ -24,8 +24,7 @@ import java.util.List;
import org.springframework.dao.DataAccessException;
import org.springframework.data.redis.connection.RedisListCommands;
import org.springframework.data.redis.connection.lettuce.LettuceConnection.LettuceResult;
import org.springframework.data.redis.connection.lettuce.LettuceConnection.LettuceTxResult;
import org.springframework.data.redis.connection.lettuce.LettuceResult.LettuceTxResult;
import org.springframework.util.Assert;
/**

203
src/main/java/org/springframework/data/redis/connection/lettuce/LettuceResult.java Normal file
View File

@@ -0,0 +1,203 @@
/*
* Copyright 2017 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.data.redis.connection.lettuce;
import io.lettuce.core.protocol.RedisCommand;
import java.util.concurrent.Future;
import java.util.function.Supplier;
import org.springframework.core.convert.converter.Converter;
import org.springframework.data.redis.connection.FutureResult;
import org.springframework.lang.Nullable;
/**
* Lettuce specific {@link FutureResult} implementation. <br />
*
* @author Costin Leau
* @author Jennifer Hickey
* @author Christoph Strobl
* @author Mark Paluch
* @since 2.1
*/
@SuppressWarnings("rawtypes")
class LettuceResult<T, S> extends FutureResult<RedisCommand<?, T, ?>> {
private final boolean convertPipelineAndTxResults;
<T> LettuceResult(Future<T> resultHolder) {
this(resultHolder, false, val -> val);
}
<T> LettuceResult(Future<T> resultHolder, boolean convertPipelineAndTxResults, @Nullable Converter<T, ?> converter) {
this(resultHolder, () -> null, convertPipelineAndTxResults, converter);
}
<T> LettuceResult(Future<T> resultHolder, Supplier<S> defaultReturnValue, boolean convertPipelineAndTxResults,
@Nullable Converter<T, ?> converter) {
super((RedisCommand) resultHolder, converter, defaultReturnValue);
this.convertPipelineAndTxResults = convertPipelineAndTxResults;
}
/*
* (non-Javadoc)
* @see org.springframework.data.redis.connection.FutureResult#get()
* @return
*/
@Nullable
@SuppressWarnings("unchecked")
@Override
public T get() {
return (T) getResultHolder().getOutput().get();
}
/*
* (non-Javadoc)
* @see org.springframework.data.redis.connection.FutureResult#seeksConversion()
* @return
*/
@Override
public boolean seeksConversion() {
return convertPipelineAndTxResults && converter != null;
}
/**
* Lettuce specific {@link FutureResult} implementation of a throw away status result.
*/
static class LettuceStatusResult extends LettuceResult {
@SuppressWarnings("rawtypes")
LettuceStatusResult(Future resultHolder) {
super(resultHolder);
setStatus(true);
}
<T> LettuceStatusResult(Future<T> resultHolder, boolean convertPipelineAndTxResults, Converter<T, ?> converter) {
super(resultHolder, convertPipelineAndTxResults, converter);
setStatus(true);
}
}
/**
* Lettuce specific {@link FutureResult} implementation of a transaction result.
*/
static class LettuceTxResult<T> extends FutureResult<Object> {
private final boolean convertPipelineAndTxResults;
LettuceTxResult(T resultHolder) {
this(resultHolder, false, val -> val);
}
LettuceTxResult(T resultHolder, boolean convertPipelineAndTxResults, Converter<?, ?> converter) {
this(resultHolder, () -> null, convertPipelineAndTxResults, converter);
}
LettuceTxResult(T resultHolder, Supplier<Object> defaultReturnValue, boolean convertPipelineAndTxResults,
Converter<?, ?> converter) {
super(resultHolder, converter, defaultReturnValue);
this.convertPipelineAndTxResults = convertPipelineAndTxResults;
}
@SuppressWarnings("unchecked")
@Override
public Object get() {
return getResultHolder();
}
@Override
public boolean seeksConversion() {
return convertPipelineAndTxResults && converter != null;
}
}
/**
* Lettuce specific {@link FutureResult} implementation of a throw away status result.
*/
static class LettuceTxStatusResult extends LettuceTxResult {
LettuceTxStatusResult(Object resultHolder) {
super(resultHolder);
setStatus(true);
}
}
/**
* Builder for constructing {@link LettuceResult}.
*
* @param <T>
* @param <S>
* @since 2.1
*/
static class LettuceResultBuilder<T, S> {
private final Object response;
private Converter<T, ?> converter;
private boolean convertPipelineAndTxResults = false;
private Supplier<?> nullValueDefault = () -> null;
LettuceResultBuilder(Object response) {
this.response = response;
this.converter = (source) -> source;
}
static <T> LettuceResultBuilder<T, ?> forResponse(Future<T> response) {
return new LettuceResultBuilder<>(response);
}
static <T> LettuceResultBuilder<T, ?> forResponse(T response) {
return new LettuceResultBuilder<>(response);
}
<S> LettuceResultBuilder<T, S> mappedWith(Converter<T, S> converter) {
this.converter = converter;
return (LettuceResultBuilder<T, S>) this;
}
<S> LettuceResultBuilder<T, S> defaultNullTo(S value) {
return (defaultNullTo(() -> value));
}
<S> LettuceResultBuilder<T, S> defaultNullTo(Supplier<S> value) {
this.nullValueDefault = value;
return (LettuceResultBuilder<T, S>) this;
}
LettuceResultBuilder<T, S> convertPipelineAndTxResults(boolean flag) {
convertPipelineAndTxResults = flag;
return this;
}
LettuceResult<T, S> build() {
return new LettuceResult((Future<T>) response, nullValueDefault, convertPipelineAndTxResults, converter);
}
LettuceTxResult<T> buildTxResult() {
return new LettuceTxResult(response, nullValueDefault, convertPipelineAndTxResults, converter);
}
LettuceResult buildStatusResult() {
return new LettuceStatusResult((Future<T>) response);
}
}
}

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

@@ -25,8 +25,7 @@ import org.springframework.core.convert.converter.Converter;
import org.springframework.dao.DataAccessException;
import org.springframework.data.redis.connection.RedisScriptingCommands;
import org.springframework.data.redis.connection.ReturnType;
import org.springframework.data.redis.connection.lettuce.LettuceConnection.LettuceResult;
import org.springframework.data.redis.connection.lettuce.LettuceConnection.LettuceTxResult;
import org.springframework.data.redis.connection.lettuce.LettuceResult.LettuceTxResult;
import org.springframework.util.Assert;
/**

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

@@ -27,8 +27,7 @@ import org.springframework.dao.DataAccessException;
import org.springframework.data.redis.connection.RedisNode;
import org.springframework.data.redis.connection.RedisServerCommands;
import org.springframework.data.redis.connection.convert.Converters;
import org.springframework.data.redis.connection.lettuce.LettuceConnection.LettuceResult;
import org.springframework.data.redis.connection.lettuce.LettuceConnection.LettuceTxResult;
import org.springframework.data.redis.connection.lettuce.LettuceResult.LettuceTxResult;
import org.springframework.data.redis.core.types.RedisClientInfo;
import org.springframework.lang.Nullable;
import org.springframework.util.Assert;

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

@@ -29,8 +29,7 @@ import java.util.Set;
import org.springframework.core.convert.converter.Converter;
import org.springframework.dao.DataAccessException;
import org.springframework.data.redis.connection.RedisSetCommands;
import org.springframework.data.redis.connection.lettuce.LettuceConnection.LettuceResult;
import org.springframework.data.redis.connection.lettuce.LettuceConnection.LettuceTxResult;
import org.springframework.data.redis.connection.lettuce.LettuceResult.LettuceTxResult;
import org.springframework.data.redis.core.Cursor;
import org.springframework.data.redis.core.KeyBoundCursor;
import org.springframework.data.redis.core.ScanIteration;

40
src/main/java/org/springframework/data/redis/connection/lettuce/LettuceStringCommands.java
View File

@@ -27,8 +27,7 @@ import java.util.concurrent.Future;
import org.springframework.dao.DataAccessException;
import org.springframework.data.redis.connection.RedisStringCommands;
import org.springframework.data.redis.connection.convert.Converters;
import org.springframework.data.redis.connection.lettuce.LettuceConnection.LettuceResult;
import org.springframework.data.redis.connection.lettuce.LettuceConnection.LettuceTxResult;
import org.springframework.data.redis.connection.lettuce.LettuceResult.LettuceTxResult;
import org.springframework.data.redis.core.types.Expiration;
import org.springframework.util.Assert;
@@ -131,11 +130,13 @@ class LettuceStringCommands implements RedisStringCommands {
try {
if (isPipelined()) {
pipeline(connection.newLettuceStatusResult(getAsyncConnection().set(key, value), Converters.stringToBooleanConverter()));
pipeline(
connection.newLettuceResult(getAsyncConnection().set(key, value), Converters.stringToBooleanConverter()));
return null;
}
if (isQueueing()) {
transaction(connection.newLettuceTxStatusResult(getConnection().set(key, value), Converters.stringToBooleanConverter()));
transaction(
connection.newLettuceTxResult(getConnection().set(key, value), Converters.stringToBooleanConverter()));
return null;
}
return Converters.stringToBoolean(getConnection().set(key, value));
@@ -158,16 +159,19 @@ class LettuceStringCommands implements RedisStringCommands {
try {
if (isPipelined()) {
pipeline(connection.newLettuceStatusResult(
getAsyncConnection().set(key, value, LettuceConverters.toSetArgs(expiration, option)), Converters.stringToBooleanConverter()));
pipeline(connection.newLettuceResult(
getAsyncConnection().set(key, value, LettuceConverters.toSetArgs(expiration, option)),
Converters.stringToBooleanConverter(), () -> false));
return null;
}
if (isQueueing()) {
transaction(connection.newLettuceTxStatusResult(
getConnection().set(key, value, LettuceConverters.toSetArgs(expiration, option)), Converters.stringToBooleanConverter()));
transaction(connection.newLettuceTxResult(
getConnection().set(key, value, LettuceConverters.toSetArgs(expiration, option)),
Converters.stringToBooleanConverter(), () -> false));
return null;
}
return Converters.stringToBoolean(getConnection().set(key, value, LettuceConverters.toSetArgs(expiration, option)));
return Converters
.stringToBoolean(getConnection().set(key, value, LettuceConverters.toSetArgs(expiration, option)));
} catch (Exception ex) {
throw convertLettuceAccessException(ex);
}
@@ -210,11 +214,13 @@ class LettuceStringCommands implements RedisStringCommands {
try {
if (isPipelined()) {
pipeline(connection.newLettuceStatusResult(getAsyncConnection().setex(key, seconds, value), Converters.stringToBooleanConverter()));
pipeline(connection.newLettuceResult(getAsyncConnection().setex(key, seconds, value),
Converters.stringToBooleanConverter()));
return null;
}
if (isQueueing()) {
transaction(connection.newLettuceTxStatusResult(getConnection().setex(key, seconds, value), Converters.stringToBooleanConverter()));
transaction(connection.newLettuceTxResult(getConnection().setex(key, seconds, value),
Converters.stringToBooleanConverter()));
return null;
}
return Converters.stringToBoolean(getConnection().setex(key, seconds, value));
@@ -235,11 +241,13 @@ class LettuceStringCommands implements RedisStringCommands {
try {
if (isPipelined()) {
pipeline(connection.newLettuceStatusResult(getAsyncConnection().psetex(key, milliseconds, value), Converters.stringToBooleanConverter()));
pipeline(connection.newLettuceResult(getAsyncConnection().psetex(key, milliseconds, value),
Converters.stringToBooleanConverter()));
return null;
}
if (isQueueing()) {
transaction(connection.newLettuceTxStatusResult(getConnection().psetex(key, milliseconds, value), Converters.stringToBooleanConverter()));
transaction(connection.newLettuceTxResult(getConnection().psetex(key, milliseconds, value),
Converters.stringToBooleanConverter()));
return null;
}
return Converters.stringToBoolean(getConnection().psetex(key, milliseconds, value));
@@ -259,13 +267,11 @@ class LettuceStringCommands implements RedisStringCommands {
try {
if (isPipelined()) {
pipeline(connection.newLettuceStatusResult(getAsyncConnection().mset(tuples),
Converters.stringToBooleanConverter()));
pipeline(connection.newLettuceResult(getAsyncConnection().mset(tuples), Converters.stringToBooleanConverter()));
return null;
}
if (isQueueing()) {
transaction(
connection.newLettuceTxStatusResult(getConnection().mset(tuples), Converters.stringToBooleanConverter()));
transaction(connection.newLettuceTxResult(getConnection().mset(tuples), Converters.stringToBooleanConverter()));
return null;
}
return Converters.stringToBoolean(getConnection().mset(tuples));

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

@@ -29,8 +29,7 @@ import java.util.Set;
import org.springframework.dao.DataAccessException;
import org.springframework.data.redis.connection.RedisZSetCommands;
import org.springframework.data.redis.connection.lettuce.LettuceConnection.LettuceResult;
import org.springframework.data.redis.connection.lettuce.LettuceConnection.LettuceTxResult;
import org.springframework.data.redis.connection.lettuce.LettuceResult.LettuceTxResult;
import org.springframework.data.redis.core.Cursor;
import org.springframework.data.redis.core.KeyBoundCursor;
import org.springframework.data.redis.core.ScanIteration;