@@ -53,6 +53,7 @@ import java.util.concurrent.ConcurrentHashMap;
|
||||
import java.util.concurrent.Future;
|
||||
import java.util.concurrent.TimeUnit;
|
||||
import java.util.concurrent.atomic.AtomicLong;
|
||||
import java.util.function.Function;
|
||||
import java.util.function.Supplier;
|
||||
|
||||
import org.springframework.beans.BeanUtils;
|
||||
@@ -936,6 +937,79 @@ public class LettuceConnection extends AbstractRedisConnection {
|
||||
}
|
||||
}
|
||||
|
||||
<S, T> void executeInPipeline(Function<RedisClusterAsyncCommands, RedisFuture<S>> command,
|
||||
Converter<S, T> converter) {
|
||||
|
||||
try {
|
||||
pipeline(newLettuceResult(command.apply(getAsyncConnection()), converter));
|
||||
} catch (Exception ex) {
|
||||
throw convertLettuceAccessException(ex);
|
||||
}
|
||||
}
|
||||
|
||||
<S, T> void executeInTx(Function<RedisClusterAsyncCommands, RedisFuture<S>> command, Converter<S, T> converter) {
|
||||
|
||||
try {
|
||||
transaction(newLettuceResult(command.apply(getAsyncConnection()), converter));
|
||||
} catch (Exception ex) {
|
||||
throw convertLettuceAccessException(ex);
|
||||
}
|
||||
}
|
||||
|
||||
<T> NullableResult<T> execute(Function<RedisClusterCommands, T> command) {
|
||||
|
||||
try {
|
||||
|
||||
T result = command.apply(getConnection());
|
||||
return NullableResult.of(result);
|
||||
} catch (Exception ex) {
|
||||
throw convertLettuceAccessException(ex);
|
||||
}
|
||||
}
|
||||
|
||||
<T> RedisFuture<T> executeAsync(Function<RedisClusterAsyncCommands, RedisFuture<T>> command) {
|
||||
|
||||
try {
|
||||
return command.apply(getAsyncConnection());
|
||||
} catch (Exception ex) {
|
||||
throw convertLettuceAccessException(ex);
|
||||
}
|
||||
}
|
||||
|
||||
<S, T> T invoke(RedisFuture<S> future, Converter<S, T> converter) {
|
||||
|
||||
if (isPipelined()) {
|
||||
pipeline(newLettuceResult(future, converter));
|
||||
return null;
|
||||
}
|
||||
if (isQueueing()) {
|
||||
transaction(newLettuceResult(future, converter));
|
||||
return null;
|
||||
}
|
||||
return NullableResult.of((S) await(future)).convert(converter).get();
|
||||
}
|
||||
|
||||
<T> T execute(Function<RedisClusterCommands, T> sync, Function<RedisClusterAsyncCommands, RedisFuture<T>> async) {
|
||||
return execute(sync, async, val -> val);
|
||||
}
|
||||
|
||||
// use a future here and only async.
|
||||
<S, T> T execute(Function<RedisClusterCommands, S> sync, Function<RedisClusterAsyncCommands, RedisFuture<S>> async,
|
||||
Converter<S, T> converter) {
|
||||
|
||||
if (isPipelined()) {
|
||||
executeInPipeline(async, converter);
|
||||
return null;
|
||||
}
|
||||
if (isQueueing()) {
|
||||
executeInTx(async, converter);
|
||||
return null;
|
||||
}
|
||||
|
||||
// return execute(sync).convert(converter).get();
|
||||
return NullableResult.of((S) await(executeAsync(async))).convert(converter).get();
|
||||
}
|
||||
|
||||
void transaction(FutureResult<?> result) {
|
||||
txResults.add(result);
|
||||
}
|
||||
|
||||
@@ -36,6 +36,7 @@ import org.springframework.data.geo.Distance;
|
||||
import org.springframework.data.geo.GeoResults;
|
||||
import org.springframework.data.geo.Metric;
|
||||
import org.springframework.data.geo.Point;
|
||||
import org.springframework.data.redis.connection.NullableResult;
|
||||
import org.springframework.data.redis.connection.RedisGeoCommands;
|
||||
import org.springframework.data.redis.connection.convert.ListConverter;
|
||||
import org.springframework.lang.Nullable;
|
||||
@@ -165,23 +166,12 @@ class LettuceGeoCommands implements RedisGeoCommands {
|
||||
GeoArgs.Unit geoUnit = LettuceConverters.toGeoArgsUnit(metric);
|
||||
Converter<Double, Distance> distanceConverter = LettuceConverters.distanceConverterForMetric(metric);
|
||||
|
||||
try {
|
||||
if (isPipelined()) {
|
||||
pipeline(connection.newLettuceResult(getAsyncConnection().geodist(key, member1, member2, geoUnit),
|
||||
distanceConverter));
|
||||
return null;
|
||||
}
|
||||
if (isQueueing()) {
|
||||
transaction(connection.newLettuceResult(getAsyncConnection().geodist(key, member1, member2, geoUnit),
|
||||
distanceConverter));
|
||||
return null;
|
||||
}
|
||||
// return connection.execute(
|
||||
// sync -> sync.geodist(key, member1, member2, geoUnit),
|
||||
// async -> async.geodist(key, member1, member2, geoUnit),
|
||||
// distanceConverter);
|
||||
|
||||
Double distance = getConnection().geodist(key, member1, member2, geoUnit);
|
||||
return distance != null ? distanceConverter.convert(distance) : null;
|
||||
} catch (Exception ex) {
|
||||
throw convertLettuceAccessException(ex);
|
||||
}
|
||||
return connection.invoke(connection.getAsyncConnection().geodist(key, member1, member2, geoUnit), distanceConverter);
|
||||
}
|
||||
|
||||
/*
|
||||
|
||||
Reference in New Issue
Block a user