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

DATAREDIS-719 - Polishing.

Browse Source 
Fix generics for accepted Future, Supplier and Converter. Introduce generics for status responses. Add Javadoc, use synchronous API for exists call.

Original pull request: #289.
This commit is contained in:
Mark Paluch
2017-10-20 12:02:31 +02:00
parent 435ace8f55
commit 39541122d1
8 changed files with 233 additions and 183 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -26,7 +26,8 @@ import org.springframework.lang.Nullable;
* @author Jennifer Hickey
* @author Christoph Strobl
* @author Mark Paluch
* @param <T> The data type of the object that holds the future result (usually of type Future)
* @param <T> The data type of the object that holds the future result (usually type of the
* {@link java.util.concurrent.Future} or response wrapper).
*/
public abstract class FutureResult<T> {
@@ -36,7 +37,7 @@ public abstract class FutureResult<T> {
private boolean status = false;
@SuppressWarnings("rawtypes") //
protected @Nullable Converter converter;
protected Converter converter;
/**
* Create new {@link FutureResult} for given object actually holding the result itself.
@@ -70,7 +71,6 @@ public abstract class FutureResult<T> {
* @param defaultConversionResult must not be {@literal null}.
* @since 2.1
*/
@SuppressWarnings("rawtypes")
public FutureResult(T resultHolder, @Nullable Converter converter, Supplier<?> defaultConversionResult) {
this.resultHolder = resultHolder;
@@ -99,7 +99,7 @@ public abstract class FutureResult<T> {
public Object convert(@Nullable Object result) {
if (result == null) {
return computeDefaultResult(result);
return computeDefaultResult(null);
}
return computeDefaultResult(converter.convert(result));
@@ -140,8 +140,8 @@ public abstract class FutureResult<T> {
/**
* Indicate whether or not the actual result needs to be {@link #convert(Object) converted} before handing over.
*
* @return
* @return {@literal true} if result conversion is required.
* @since 2.1
*/
public abstract boolean seeksConversion();
public abstract boolean conversionRequired();
}

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

@@ -16,7 +16,6 @@
package org.springframework.data.redis.connection.convert;
import java.util.ArrayList;
import java.util.LinkedList;
import java.util.List;
import java.util.Queue;
@@ -31,13 +30,13 @@ import org.springframework.data.redis.connection.FutureResult;
*
* @author Jennifer Hickey
* @author Christoph Strobl
* @author Mark Paluch
* @param <T> The type of {@link FutureResult} of the individual tx operations
*/
public class TransactionResultConverter<T> implements Converter<List<Object>, List<Object>> {
private Queue<FutureResult<T>> txResults = new LinkedList<>();
private Converter<Exception, DataAccessException> exceptionConverter;
private final Queue<FutureResult<T>> txResults;
private final Converter<Exception, DataAccessException> exceptionConverter;
public TransactionResultConverter(Queue<FutureResult<T>> txResults,
Converter<Exception, DataAccessException> exceptionConverter) {
@@ -71,7 +70,7 @@ public class TransactionResultConverter<T> implements Converter<List<Object>, Li
: new RedisSystemException("Error reading future result.", source);
}
if (!(futureResult.isStatus())) {
convertedResults.add(futureResult.seeksConversion() ? futureResult.convert(result) : result);
convertedResults.add(futureResult.conversionRequired() ? futureResult.convert(result) : result);
}
}

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

@@ -389,7 +389,7 @@ public class JedisConnection extends AbstractRedisConnection {
Object data = result.get();
if (!result.isStatus()) {
results.add(result.seeksConversion() ? result.convert(data) : data);
results.add(result.conversionRequired() ? result.convert(data) : data);
}
} catch (JedisDataException e) {
DataAccessException dataAccessException = convertJedisAccessException(e);
@@ -554,16 +554,16 @@ public class JedisConnection extends AbstractRedisConnection {
return JedisResultBuilder.forResponse(response).build();
}
<T> JedisResult newJedisResult(Response<T> response, Converter<T, ?> converter) {
<T, R> JedisResult newJedisResult(Response<T> response, Converter<T, R> converter) {
return JedisResultBuilder.forResponse(response).mappedWith(converter)
return JedisResultBuilder.<T, R> forResponse(response).mappedWith(converter)
.convertPipelineAndTxResults(convertPipelineAndTxResults).build();
}
<T> JedisResult newJedisResult(Response<T> response, Converter<T, ?> converter, Supplier<?> defaultValue) {
<T, R> JedisResult newJedisResult(Response<T> response, Converter<T, R> converter, Supplier<R> defaultValue) {
return JedisResultBuilder.forResponse(response).mappedWith(converter)
.convertPipelineAndTxResults(convertPipelineAndTxResults).defaultNullTo(defaultValue).build();
return JedisResultBuilder.<T, R> forResponse(response).mappedWith(converter)
.convertPipelineAndTxResults(convertPipelineAndTxResults).mapNullTo(defaultValue).build();
}
JedisStatusResult newStatusResult(Response<?> response) {

88
src/main/java/org/springframework/data/redis/connection/jedis/JedisResult.java
View File

@@ -30,21 +30,23 @@ import org.springframework.lang.Nullable;
* @author Jennifer Hickey
* @author Christoph Strobl
* @author Mark Paluch
* @param <T> The data type of the object that holds the future result (usually of type Future).
* @param <R> The data type of the result type.
* @since 2.1
*/
class JedisResult<T, S> extends FutureResult<Response<?>> {
class JedisResult<T, R> extends FutureResult<Response<?>> {
private final boolean convertPipelineAndTxResults;
<T> JedisResult(Response<T> resultHolder) {
JedisResult(Response<T> resultHolder) {
this(resultHolder, false, null);
}
<T> JedisResult(Response<T> resultHolder, boolean convertPipelineAndTxResults, @Nullable Converter<T, ?> converter) {
this(resultHolder, null, convertPipelineAndTxResults, converter);
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,
JedisResult(Response<T> resultHolder, Supplier<R> defaultReturnValue, boolean convertPipelineAndTxResults,
@Nullable Converter<T, ?> converter) {
super(resultHolder, converter, defaultReturnValue);
@@ -58,25 +60,26 @@ class JedisResult<T, S> extends FutureResult<Response<?>> {
*/
@Nullable
@Override
@SuppressWarnings("unchecked")
public T get() {
return (T) getResultHolder().get();
}
/*
* (non-Javadoc)
* @see org.springframework.data.redis.connection.FutureResult#seeksConversion()
* @return
* @see org.springframework.data.redis.connection.FutureResult#conversionRequired()
*/
public boolean seeksConversion() {
return convertPipelineAndTxResults && converter != null;
public boolean conversionRequired() {
return convertPipelineAndTxResults;
}
/**
* Jedis specific {@link FutureResult} implementation of a throw away status result.
*/
static class JedisStatusResult extends JedisResult {
static class JedisStatusResult<T, R> extends JedisResult<T, R> {
<T> JedisStatusResult(Response<T> resultHolder, Converter<T, ?> converter) {
@SuppressWarnings("unchecked")
JedisStatusResult(Response<T> resultHolder, Converter<T, R> converter) {
super(resultHolder, false, converter);
setStatus(true);
@@ -87,54 +90,77 @@ class JedisResult<T, S> extends FutureResult<Response<?>> {
* Builder for constructing {@link JedisResult}.
*
* @param <T>
* @param <S>
* @param <R>
* @since 2.1
*/
static class JedisResultBuilder<T, S> {
static class JedisResultBuilder<T, R> {
private final Response<T> response;
private Converter<T, ?> converter;
private Converter<T, R> converter;
private boolean convertPipelineAndTxResults = false;
private Supplier<?> nullValueDefault = () -> null;
private Supplier<R> nullValueDefault = () -> null;
@SuppressWarnings("unchecked")
JedisResultBuilder(Response<T> response) {
this.response = response;
this.converter = (source) -> source;
this.converter = (source) -> (R) source;
}
static <T> JedisResultBuilder<T, ?> forResponse(Response<T> response) {
/**
* Create a new {@link JedisResultBuilder} given {@link Response}.
*
* @param response must not be {@literal null}.
* @param <T> native response type.
* @param <R> resulting response type.
* @return the new {@link JedisResultBuilder}.
*/
static <T, R> JedisResultBuilder<T, R> forResponse(Response<T> response) {
return new JedisResultBuilder<>(response);
}
<S> JedisResultBuilder<T, S> mappedWith(Converter<T, S> converter) {
/**
* Configure a {@link Converter} to convert between {@code T} and {@code R} types.
*
* @param converter must not be {@literal null}.
* @return {@code this} builder.
*/
JedisResultBuilder<T, R> mappedWith(Converter<T, R> converter) {
this.converter = converter;
return (JedisResultBuilder<T, S>) this;
return this;
}
<S> JedisResultBuilder<T, S> defaultNullTo(S value) {
return (defaultNullTo(() -> value));
/**
* Configure a {@link Supplier} to map {@literal null} responses to a different value.
*
* @param supplier must not be {@literal null}.
* @return {@code this} builder.
*/
JedisResultBuilder<T, R> mapNullTo(Supplier<R> supplier) {
this.nullValueDefault = supplier;
return this;
}
<S> JedisResultBuilder<T, S> defaultNullTo(Supplier<S> value) {
this.nullValueDefault = value;
return (JedisResultBuilder<T, S>) this;
}
JedisResultBuilder<T, S> convertPipelineAndTxResults(boolean flag) {
JedisResultBuilder<T, R> convertPipelineAndTxResults(boolean flag) {
convertPipelineAndTxResults = flag;
return this;
}
JedisResult<T, S> build() {
return new JedisResult(response, nullValueDefault, convertPipelineAndTxResults, converter);
/**
* @return a new {@link JedisResult} wrapper with configuration applied from this builder.
*/
JedisResult<T, R> build() {
return new JedisResult<>(response, nullValueDefault, convertPipelineAndTxResults, converter);
}
/**
* @return a new {@link JedisStatusResult} wrapper for status results with configuration applied from this builder.
*/
JedisStatusResult buildStatusResult() {
return new JedisStatusResult(response, converter);
return new JedisStatusResult<>(response, converter);
}
}
}

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

@@ -66,9 +66,6 @@ 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;
@@ -119,35 +116,41 @@ public class LettuceConnection extends AbstractRedisConnection {
return newLettuceResult(resultHolder, (val) -> val);
}
<T> LettuceResult newLettuceResult(Future<T> resultHolder, Converter<T, ?> converter) {
@SuppressWarnings("unchecked")
<T, R> LettuceResult<T, R> newLettuceResult(Future<T> resultHolder, Converter<T, R> converter) {
return LettuceResultBuilder.forResponse(resultHolder).mappedWith(converter)
return LettuceResultBuilder.forResponse(resultHolder).mappedWith((Converter) converter)
.convertPipelineAndTxResults(convertPipelineAndTxResults).build();
}
<T> LettuceResult newLettuceResult(Future<T> resultHolder, Converter<T, ?> converter, Supplier<?> defaultValue) {
@SuppressWarnings("unchecked")
<T, R> LettuceResult<T, R> newLettuceResult(Future<T> resultHolder, Converter<T, R> converter,
Supplier<R> defaultValue) {
return LettuceResultBuilder.forResponse(resultHolder).mappedWith(converter)
return LettuceResultBuilder.forResponse(resultHolder).mappedWith((Converter) converter)
.convertPipelineAndTxResults(convertPipelineAndTxResults).defaultNullTo(defaultValue).build();
}
LettuceStatusResult newLettuceStatusResult(Future<?> resultHolder) {
return new LettuceStatusResult(resultHolder);
<T, R> LettuceResult<T, R> newLettuceStatusResult(Future<T> resultHolder) {
return LettuceResultBuilder.<T, R> forResponse(resultHolder).buildStatusResult();
}
LettuceTxResult newLettuceTxResult(Object resultHolder) {
<T> LettuceTxResult<T, Object> newLettuceTxResult(T resultHolder) {
return newLettuceTxResult(resultHolder, (val) -> val);
}
<T> LettuceTxResult<T> newLettuceTxResult(Object resultHolder, Converter converter) {
@SuppressWarnings("unchecked")
<T> LettuceTxResult<T, Object> newLettuceTxResult(T resultHolder, Converter<T, ?> converter) {
return LettuceResultBuilder.forResponse(resultHolder).mappedWith(converter)
return LettuceResultBuilder.forResponse(resultHolder).mappedWith((Converter) converter)
.convertPipelineAndTxResults(convertPipelineAndTxResults).buildTxResult();
}
<T> LettuceTxResult<T> newLettuceTxResult(Object resultHolder, Converter converter, Supplier<?> defaultValue) {
@SuppressWarnings("unchecked")
<T> LettuceTxResult<T, Object> newLettuceTxResult(T resultHolder, Converter<T, ?> converter,
Supplier<?> defaultValue) {
return LettuceResultBuilder.forResponse(resultHolder).mappedWith(converter)
return LettuceResultBuilder.forResponse(resultHolder).mappedWith((Converter) converter)
.convertPipelineAndTxResults(convertPipelineAndTxResults).defaultNullTo(defaultValue).buildTxResult();
}
@@ -397,6 +400,7 @@ public class LettuceConnection extends AbstractRedisConnection {
Assert.hasText(command, "a valid command needs to be specified");
try {
String name = command.trim().toUpperCase();
CommandType commandType = CommandType.valueOf(name);
@@ -412,17 +416,20 @@ public class LettuceConnection extends AbstractRedisConnection {
CommandOutput expectedOutput = commandOutputTypeHint != null ? commandOutputTypeHint
: typeHints.getTypeHint(commandType);
Command cmd = new Command(commandType, expectedOutput, cmdArg);
if (isPipelined()) {
pipeline(newLettuceResult(connectionImpl.dispatch(cmd.getType(), cmd.getOutput(), cmd.getArgs())));
return null;
} else if (isQueueing()) {
}
if (isQueueing()) {
transaction(newLettuceTxResult(connectionImpl.dispatch(cmd.getType(), cmd.getOutput(), cmd.getArgs())));
return null;
} else {
return await(connectionImpl.dispatch(cmd.getType(), cmd.getOutput(), cmd.getArgs()));
}
return await(connectionImpl.dispatch(cmd.getType(), cmd.getOutput(), cmd.getArgs()));
} catch (RedisException ex) {
throw convertLettuceAccessException(ex);
}
@@ -488,61 +495,62 @@ public class LettuceConnection extends AbstractRedisConnection {
@Override
public List<Object> closePipeline() {
if (isPipelined) {
isPipelined = false;
List<io.lettuce.core.protocol.RedisCommand<?, ?, ?>> futures = new ArrayList<>();
for (LettuceResult<?, ?> result : ppline) {
futures.add(result.getResultHolder());
}
if (!isPipelined) {
try {
boolean done = LettuceFutures.awaitAll(timeout, TimeUnit.MILLISECONDS,
futures.toArray(new RedisFuture[futures.size()]));
return Collections.emptyList();
}
isPipelined = false;
List<io.lettuce.core.protocol.RedisCommand<?, ?, ?>> futures = new ArrayList<>(ppline.size());
for (LettuceResult<?, ?> result : ppline) {
futures.add(result.getResultHolder());
}
List<Object> results = new ArrayList<>(futures.size());
try {
boolean done = LettuceFutures.awaitAll(timeout, TimeUnit.MILLISECONDS,
futures.toArray(new RedisFuture[futures.size()]));
Exception problem = null;
List<Object> results = new ArrayList<>(futures.size());
if (done) {
for (LettuceResult<?, ?> result : ppline) {
Exception problem = null;
if (result.getResultHolder().getOutput().hasError()) {
if (done) {
for (LettuceResult<?, ?> result : ppline) {
Exception err = new InvalidDataAccessApiUsageException(result.getResultHolder().getOutput().getError());
// remember only the first error
if (result.getResultHolder().getOutput().hasError()) {
Exception err = new InvalidDataAccessApiUsageException(result.getResultHolder().getOutput().getError());
// remember only the first error
if (problem == null) {
problem = err;
}
results.add(err);
} else if (!result.isStatus()) {
try {
results.add(result.conversionRequired() ? result.convert(result.get()) : result.get());
} catch (DataAccessException e) {
if (problem == null) {
problem = err;
}
results.add(err);
} else if (!result.isStatus()) {
try {
results.add(result.seeksConversion() ? result.convert(result.get()) : result.get());
} catch (DataAccessException e) {
if (problem == null) {
problem = e;
}
results.add(e);
problem = e;
}
results.add(e);
}
}
}
ppline.clear();
if (problem != null) {
throw new RedisPipelineException(problem, results);
}
if (done) {
return results;
}
throw new RedisPipelineException(new QueryTimeoutException("Redis command timed out"));
} catch (Exception e) {
throw new RedisPipelineException(e);
}
}
ppline.clear();
return Collections.emptyList();
if (problem != null) {
throw new RedisPipelineException(problem, results);
}
if (done) {
return results;
}
throw new RedisPipelineException(new QueryTimeoutException("Redis command timed out"));
} catch (Exception e) {
throw new RedisPipelineException(e);
}
}
/*
@@ -558,7 +566,7 @@ public class LettuceConnection extends AbstractRedisConnection {
return null;
}
if (isQueueing()) {
transaction(new LettuceTxResult(getConnection().echo(message)));
transaction(new LettuceTxResult<>(getConnection().echo(message)));
return null;
}
return getConnection().echo(message);
@@ -575,7 +583,7 @@ public class LettuceConnection extends AbstractRedisConnection {
return null;
}
if (isQueueing()) {
transaction(new LettuceTxResult(getConnection().ping()));
transaction(new LettuceTxResult<>(getConnection().ping()));
return null;
}
return getConnection().ping();
@@ -589,10 +597,10 @@ public class LettuceConnection extends AbstractRedisConnection {
isMulti = false;
try {
if (isPipelined()) {
pipeline(new LettuceStatusResult(((RedisAsyncCommands) getAsyncDedicatedConnection()).discard()));
pipeline(newLettuceStatusResult(getAsyncDedicatedRedisCommands().discard()));
return;
}
((RedisCommands) getDedicatedConnection()).discard();
getDedicatedRedisCommands().discard();
} catch (Exception ex) {
throw convertLettuceAccessException(ex);
} finally {
@@ -608,7 +616,7 @@ public class LettuceConnection extends AbstractRedisConnection {
try {
if (isPipelined()) {
RedisFuture<TransactionResult> exec = ((RedisAsyncCommands) getAsyncDedicatedConnection()).exec();
RedisFuture<TransactionResult> exec = getAsyncDedicatedRedisCommands().exec();
LettuceTransactionResultConverter resultConverter = new LettuceTransactionResultConverter(
new LinkedList<>(txResults), LettuceConverters.exceptionConverter());
@@ -618,7 +626,7 @@ public class LettuceConnection extends AbstractRedisConnection {
return null;
}
TransactionResult transactionResult = ((RedisCommands) getDedicatedConnection()).exec();
TransactionResult transactionResult = (getDedicatedRedisCommands()).exec();
List<Object> results = LettuceConverters.transactionResultUnwrapper().convert(transactionResult);
return convertPipelineAndTxResults
? new LettuceTransactionResultConverter(txResults, LettuceConverters.exceptionConverter()).convert(results)
@@ -641,7 +649,7 @@ public class LettuceConnection extends AbstractRedisConnection {
((RedisAsyncCommands) getAsyncDedicatedConnection()).multi();
return;
}
((RedisCommands) getDedicatedConnection()).multi();
(getDedicatedRedisCommands()).multi();
} catch (Exception ex) {
throw convertLettuceAccessException(ex);
}
@@ -661,7 +669,7 @@ public class LettuceConnection extends AbstractRedisConnection {
this.dbIndex = dbIndex;
if (isQueueing()) {
transaction(new LettuceTxStatusResult(((RedisCommands) getAsyncConnection()).select(dbIndex)));
transaction(newLettuceStatusResult(((RedisCommands)getAsyncConnection()).select(dbIndex)));
return;
}
((RedisCommands) getConnection()).select(dbIndex);
@@ -674,14 +682,14 @@ public class LettuceConnection extends AbstractRedisConnection {
public void unwatch() {
try {
if (isPipelined()) {
pipeline(new LettuceStatusResult(((RedisAsyncCommands) getAsyncDedicatedConnection()).unwatch()));
pipeline(newLettuceStatusResult(((RedisAsyncCommands) getAsyncDedicatedConnection()).unwatch()));
return;
}
if (isQueueing()) {
transaction(new LettuceTxStatusResult(((RedisAsyncCommands) getDedicatedConnection()).unwatch()));
transaction(newLettuceStatusResult(getDedicatedRedisCommands().unwatch()));
return;
}
((RedisCommands) getDedicatedConnection()).unwatch();
getDedicatedRedisCommands().unwatch();
} catch (Exception ex) {
throw convertLettuceAccessException(ex);
}
@@ -694,14 +702,14 @@ public class LettuceConnection extends AbstractRedisConnection {
}
try {
if (isPipelined()) {
pipeline(new LettuceStatusResult(((RedisAsyncCommands) getAsyncDedicatedConnection()).watch((Object[]) keys)));
pipeline(newLettuceStatusResult(getAsyncDedicatedRedisCommands().watch(keys)));
return;
}
if (isQueueing()) {
transaction(new LettuceTxStatusResult(((RedisAsyncCommands) getDedicatedConnection()).watch((Object[]) keys)));
transaction(new LettuceTxStatusResult(getDedicatedRedisCommands().watch(keys)));
return;
}
((RedisCommands) getDedicatedConnection()).watch((Object[]) keys);
getDedicatedRedisCommands().watch(keys);
} catch (Exception ex) {
throw convertLettuceAccessException(ex);
}
@@ -854,6 +862,11 @@ public class LettuceConnection extends AbstractRedisConnection {
return getDedicatedConnection();
}
@SuppressWarnings("unchecked")
private RedisAsyncCommands<byte[], byte[]> getAsyncDedicatedRedisCommands() {
return (RedisAsyncCommands) getAsyncDedicatedConnection();
}
protected RedisClusterAsyncCommands<byte[], byte[]> getAsyncDedicatedConnection() {
if (asyncDedicatedConn == null) {
@@ -876,6 +889,11 @@ public class LettuceConnection extends AbstractRedisConnection {
String.format("%s is not a supported connection type.", asyncDedicatedConn.getClass().getName()));
}
@SuppressWarnings("unchecked")
private RedisCommands<byte[], byte[]> getDedicatedRedisCommands() {
return (RedisCommands) getDedicatedConnection();
}
RedisClusterCommands<byte[], byte[]> getDedicatedConnection() {
if (asyncDedicatedConn == null) {
@@ -898,6 +916,7 @@ public class LettuceConnection extends AbstractRedisConnection {
String.format("%s is not a supported connection type.", asyncDedicatedConn.getClass().getName()));
}
@SuppressWarnings("unchecked")
protected StatefulConnection<byte[], byte[]> doGetAsyncDedicatedConnection() {
return connectionProvider.getConnection(StatefulConnection.class);
}

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

@@ -66,7 +66,7 @@ class LettuceKeyCommands implements RedisKeyCommands {
return null;
}
if (isQueueing()) {
transaction(connection.newLettuceTxResult(getAsyncConnection().exists(new byte[][] { key }),
transaction(connection.newLettuceTxResult(getConnection().exists(new byte[][] { key }),
LettuceConverters.longToBooleanConverter()));
return null;
}
@@ -93,7 +93,7 @@ class LettuceKeyCommands implements RedisKeyCommands {
return null;
}
if (isQueueing()) {
transaction(connection.newLettuceTxResult(getAsyncConnection().exists(keys)));
transaction(connection.newLettuceTxResult(getConnection().exists(keys)));
return null;
}
return getConnection().exists(keys);

120
src/main/java/org/springframework/data/redis/connection/lettuce/LettuceResult.java
View File

@@ -34,19 +34,20 @@ import org.springframework.lang.Nullable;
* @since 2.1
*/
@SuppressWarnings("rawtypes")
class LettuceResult<T, S> extends FutureResult<RedisCommand<?, T, ?>> {
class LettuceResult<T, R> extends FutureResult<RedisCommand<?, T, ?>> {
private final boolean convertPipelineAndTxResults;
<T> LettuceResult(Future<T> resultHolder) {
LettuceResult(Future<T> resultHolder) {
this(resultHolder, false, val -> val);
}
<T> LettuceResult(Future<T> resultHolder, boolean convertPipelineAndTxResults, @Nullable Converter<T, ?> converter) {
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,
@SuppressWarnings("unchecked")
LettuceResult(Future<T> resultHolder, Supplier<R> defaultReturnValue, boolean convertPipelineAndTxResults,
@Nullable Converter<T, ?> converter) {
super((RedisCommand) resultHolder, converter, defaultReturnValue);
@@ -56,59 +57,52 @@ class LettuceResult<T, S> extends FutureResult<RedisCommand<?, T, ?>> {
/*
* (non-Javadoc)
* @see org.springframework.data.redis.connection.FutureResult#get()
* @return
*/
@Nullable
@SuppressWarnings("unchecked")
@Override
@SuppressWarnings("unchecked")
public T get() {
return (T) getResultHolder().getOutput().get();
}
/*
* (non-Javadoc)
* @see org.springframework.data.redis.connection.FutureResult#seeksConversion()
* @return
* @see org.springframework.data.redis.connection.FutureResult#conversionRequired()
*/
@Override
public boolean seeksConversion() {
return convertPipelineAndTxResults && converter != null;
public boolean conversionRequired() {
return convertPipelineAndTxResults;
}
/**
* Lettuce specific {@link FutureResult} implementation of a throw away status result.
*/
static class LettuceStatusResult extends LettuceResult {
static class LettuceStatusResult<T, R> extends LettuceResult<T, R> {
@SuppressWarnings("rawtypes")
LettuceStatusResult(Future resultHolder) {
@SuppressWarnings("unchecked")
LettuceStatusResult(Future<T> 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> {
static class LettuceTxResult<T, R> extends FutureResult<T> {
private final boolean convertPipelineAndTxResults;
LettuceTxResult(T resultHolder) {
this(resultHolder, false, val -> val);
this(resultHolder, false, val -> (R) val);
}
LettuceTxResult(T resultHolder, boolean convertPipelineAndTxResults, Converter<?, ?> converter) {
LettuceTxResult(T resultHolder, boolean convertPipelineAndTxResults, Converter<T, R> converter) {
this(resultHolder, () -> null, convertPipelineAndTxResults, converter);
}
LettuceTxResult(T resultHolder, Supplier<Object> defaultReturnValue, boolean convertPipelineAndTxResults,
Converter<?, ?> converter) {
Converter<T, R> converter) {
super(resultHolder, converter, defaultReturnValue);
this.convertPipelineAndTxResults = convertPipelineAndTxResults;
}
@@ -121,15 +115,14 @@ class LettuceResult<T, S> extends FutureResult<RedisCommand<?, T, ?>> {
@Override
public boolean seeksConversion() {
return convertPipelineAndTxResults && converter != null;
return convertPipelineAndTxResults;
}
}
/**
* Lettuce specific {@link FutureResult} implementation of a throw away status result.
*/
static class LettuceTxStatusResult extends LettuceTxResult {
static class LettuceTxStatusResult extends LettuceTxResult<Object, Object> {
LettuceTxStatusResult(Object resultHolder) {
super(resultHolder);
@@ -141,63 +134,76 @@ class LettuceResult<T, S> extends FutureResult<RedisCommand<?, T, ?>> {
* Builder for constructing {@link LettuceResult}.
*
* @param <T>
* @param <S>
* @param <R>
* @since 2.1
*/
static class LettuceResultBuilder<T, S> {
static class LettuceResultBuilder<T, R> {
private final Object response;
private Converter<T, ?> converter;
private final Future<T> response;
private Converter<T, R> converter;
private boolean convertPipelineAndTxResults = false;
private Supplier<?> nullValueDefault = () -> null;
private Supplier<R> nullValueDefault = () -> null;
LettuceResultBuilder(Object response) {
LettuceResultBuilder(Future<T> response) {
this.response = response;
this.converter = (source) -> source;
this.converter = (source) -> (R) source;
}
static <T> LettuceResultBuilder<T, ?> forResponse(Future<T> response) {
/**
* Create a new {@link LettuceResultBuilder} given {@link Future}.
*
* @param response must not be {@literal null}.
* @param <T> native response type.
* @param <R> resulting response type.
* @return the new {@link LettuceResultBuilder}.
*/
static <T, R> LettuceResultBuilder<T, R> 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) {
/**
* Configure a {@link Converter} to convert between {@code T} and {@code R} types.
*
* @param converter must not be {@literal null}.
* @return {@code this} builder.
*/
LettuceResultBuilder<T, R> mappedWith(Converter<T, R> converter) {
this.converter = converter;
return (LettuceResultBuilder<T, S>) this;
return (LettuceResultBuilder<T, R>) this;
}
<S> LettuceResultBuilder<T, S> defaultNullTo(S value) {
return (defaultNullTo(() -> value));
/**
* Configure a {@link Supplier} to map {@literal null} responses to a different value.
*
* @param supplier must not be {@literal null}.
* @return {@code this} builder.
*/
LettuceResultBuilder<T, R> defaultNullTo(Supplier<R> supplier) {
this.nullValueDefault = supplier;
return this;
}
<S> LettuceResultBuilder<T, S> defaultNullTo(Supplier<S> value) {
this.nullValueDefault = value;
return (LettuceResultBuilder<T, S>) this;
}
LettuceResultBuilder<T, S> convertPipelineAndTxResults(boolean flag) {
LettuceResultBuilder<T, R> convertPipelineAndTxResults(boolean flag) {
convertPipelineAndTxResults = flag;
return this;
}
LettuceResult<T, S> build() {
return new LettuceResult((Future<T>) response, nullValueDefault, convertPipelineAndTxResults, converter);
/**
* @return a new {@link LettuceResult} wrapper with configuration applied from this builder.
*/
LettuceResult<T, R> build() {
return new LettuceResult<>(response, nullValueDefault, convertPipelineAndTxResults, converter);
}
LettuceTxResult<T> buildTxResult() {
return new LettuceTxResult(response, nullValueDefault, convertPipelineAndTxResults, converter);
}
LettuceResult buildStatusResult() {
return new LettuceStatusResult((Future<T>) response);
/**
* @return a new {@link LettuceResult} wrapper for status results with configuration applied from this builder.
*/
LettuceResult<T, R> buildStatusResult() {
return new LettuceStatusResult<>(response);
}
}
}

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

@@ -453,7 +453,7 @@ class LettuceServerCommands implements RedisServerCommands {
throw new UnsupportedOperationException("Cannot be called in pipeline mode.");
}
if (isQueueing()) {
transaction(connection.newLettuceTxResult(getAsyncConnection().clientList(),
transaction(connection.newLettuceTxResult(getConnection().clientList(),
LettuceConverters.stringToRedisClientListConverter()));
return null;
}
@@ -551,11 +551,11 @@ class LettuceServerCommands implements RedisServerCommands {
return connection.isQueueing();
}
private void pipeline(LettuceResult result) {
private void pipeline(LettuceResult<?, ?> result) {
connection.pipeline(result);
}
private void transaction(LettuceTxResult result) {
private void transaction(LettuceTxResult<?, ?> result) {
connection.transaction(result);
}