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

Accept CompletableFuture subtypes for Lettuce pipelining.

Browse Source 
We now no longer require RedisCommand but resort to CompletableFuture as the general asynchronous result type for Lettuce pipelining to allow subtypes such as PipelinedRedisFuture.

Closes: #2888
Original Pull Request: #2889
This commit is contained in:
Mark Paluch
2024-04-04 09:15:26 +02:00
committed by Christoph Strobl
parent eac13566d4
commit 7c04f6baf2
4 changed files with 60 additions and 17 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -49,7 +49,9 @@ import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import java.util.Queue;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicLong;
@@ -581,7 +583,7 @@ public class LettuceConnection extends AbstractRedisConnection {
pipeliningFlushState = null;
isPipelined = false;
List<io.lettuce.core.protocol.RedisCommand<?, ?, ?>> futures = new ArrayList<>(ppline.size());
List<CompletableFuture<?>> futures = new ArrayList<>(ppline.size());
for (LettuceResult<?, ?> result : ppline) {
futures.add(result.getResultHolder());
@@ -598,10 +600,24 @@ public class LettuceConnection extends AbstractRedisConnection {
if (done) {
for (LettuceResult<?, ?> result : ppline) {
if (result.getResultHolder().getOutput().hasError()) {
CompletableFuture<?> resultHolder = result.getResultHolder();
if (resultHolder.isCompletedExceptionally()) {
Exception exception = new InvalidDataAccessApiUsageException(result.getResultHolder()
.getOutput().getError());
String message;
if (resultHolder instanceof io.lettuce.core.protocol.RedisCommand<?, ?, ?> rc) {
message = rc.getOutput().getError();
} else {
try {
resultHolder.get();
message = "";
} catch (InterruptedException ignore) {
message = "";
} catch (ExecutionException e) {
message = e.getCause().getMessage();
}
}
Exception exception = new InvalidDataAccessApiUsageException(message);
// remember only the first error
if (problem == null) {

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

@@ -15,8 +15,7 @@
*/
package org.springframework.data.redis.connection.lettuce;
import io.lettuce.core.protocol.RedisCommand;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.Future;
import java.util.function.Supplier;
@@ -34,7 +33,7 @@ import org.springframework.lang.Nullable;
* @since 2.1
*/
@SuppressWarnings("rawtypes")
class LettuceResult<T, R> extends FutureResult<RedisCommand<?, T, ?>> {
class LettuceResult<T, R> extends FutureResult<CompletableFuture<T>> {
private final boolean convertPipelineAndTxResults;
@@ -51,7 +50,7 @@ class LettuceResult<T, R> extends FutureResult<RedisCommand<?, T, ?>> {
LettuceResult(Future<T> resultHolder, Supplier<R> defaultReturnValue, boolean convertPipelineAndTxResults,
@Nullable Converter<T, R> converter) {
super((RedisCommand) resultHolder, converter, defaultReturnValue);
super((CompletableFuture<T>) resultHolder, converter, defaultReturnValue);
this.convertPipelineAndTxResults = convertPipelineAndTxResults;
}
@@ -59,7 +58,7 @@ class LettuceResult<T, R> extends FutureResult<RedisCommand<?, T, ?>> {
@Override
@SuppressWarnings("unchecked")
public T get() {
return (T) getResultHolder().getOutput().get();
return (T) getResultHolder().join();
}
@Override

1
src/test/java/org/springframework/data/redis/connection/ClusterTestVariables.java
View File

@@ -25,6 +25,7 @@ public abstract class ClusterTestVariables {
public static final String KEY_1 = "key1";
public static final String KEY_2 = "key2";
public static final String KEY_3 = "key3";
public static final String KEY_4 = "key4";
public static final String VALUE_1 = "value1";
public static final String VALUE_2 = "value2";

43
src/test/java/org/springframework/data/redis/connection/lettuce/LettuceClusterConnectionTests.java
View File

@@ -83,6 +83,7 @@ public class LettuceClusterConnectionTests implements ClusterConnectionTests {
private static final byte[] KEY_1_BYTES = LettuceConverters.toBytes(KEY_1);
private static final byte[] KEY_2_BYTES = LettuceConverters.toBytes(KEY_2);
private static final byte[] KEY_3_BYTES = LettuceConverters.toBytes(KEY_3);
private static final byte[] KEY_4_BYTES = LettuceConverters.toBytes(KEY_4);
private static final byte[] SAME_SLOT_KEY_1_BYTES = LettuceConverters.toBytes(SAME_SLOT_KEY_1);
private static final byte[] SAME_SLOT_KEY_2_BYTES = LettuceConverters.toBytes(SAME_SLOT_KEY_2);
@@ -91,6 +92,7 @@ public class LettuceClusterConnectionTests implements ClusterConnectionTests {
private static final byte[] VALUE_1_BYTES = LettuceConverters.toBytes(VALUE_1);
private static final byte[] VALUE_2_BYTES = LettuceConverters.toBytes(VALUE_2);
private static final byte[] VALUE_3_BYTES = LettuceConverters.toBytes(VALUE_3);
private static final byte[] VALUE_4_BYTES = LettuceConverters.toBytes(VALUE_4);
private static final GeoLocation<String> ARIGENTO = new GeoLocation<>("arigento", POINT_ARIGENTO);
private static final GeoLocation<String> CATANIA = new GeoLocation<>("catania", POINT_CATANIA);
@@ -179,7 +181,32 @@ public class LettuceClusterConnectionTests implements ClusterConnectionTests {
} finally {
factory.destroy();
}
}
@Test // GH-2888
void shouldPipelineAdvancedClusterApi() {
LettuceConnectionFactory factory = createConnectionFactory();
ConnectionVerifier.create(factory) //
.execute(connection -> {
connection.set(KEY_1_BYTES, VALUE_1_BYTES);
connection.set(KEY_2_BYTES, VALUE_2_BYTES);
connection.set(KEY_4_BYTES, VALUE_4_BYTES);
connection.openPipeline();
connection.keyCommands().randomKey();
connection.stringCommands().mGet(KEY_1_BYTES, KEY_2_BYTES);
List<Object> objects = connection.closePipeline();
assertThat(objects).hasSize(1);
assertThat(objects).element(0).isInstanceOf(List.class);
List<Object> mget = (List<Object>) objects.get(0);
assertThat(mget).containsExactly(VALUE_1_BYTES, VALUE_2_BYTES);
}).verifyAndClose();
}
@Test // DATAREDIS-315
@@ -2821,13 +2848,13 @@ public class LettuceClusterConnectionTests implements ClusterConnectionTests {
assertThat(clusterConnection.stringCommands().bitField(LettuceConverters.toBytes(KEY_1),
create().incr(unsigned(2)).valueAt(BitFieldSubCommands.Offset.offset(102L)).overflow(FAIL).by(1L)))
.containsExactly(1L);
.containsExactly(1L);
assertThat(clusterConnection.stringCommands().bitField(LettuceConverters.toBytes(KEY_1),
create().incr(unsigned(2)).valueAt(BitFieldSubCommands.Offset.offset(102L)).overflow(FAIL).by(1L)))
.containsExactly(2L);
.containsExactly(2L);
assertThat(clusterConnection.stringCommands().bitField(LettuceConverters.toBytes(KEY_1),
create().incr(unsigned(2)).valueAt(BitFieldSubCommands.Offset.offset(102L)).overflow(FAIL).by(1L)))
.containsExactly(3L);
.containsExactly(3L);
assertThat(clusterConnection.stringCommands().bitField(LettuceConverters.toBytes(KEY_1),
create().incr(unsigned(2)).valueAt(BitFieldSubCommands.Offset.offset(102L)).overflow(FAIL).by(1L))).isNotNull();
}
@@ -2837,7 +2864,7 @@ public class LettuceClusterConnectionTests implements ClusterConnectionTests {
assertThat(clusterConnection.stringCommands().bitField(LettuceConverters.toBytes(KEY_1),
create().incr(signed(5)).valueAt(BitFieldSubCommands.Offset.offset(100L)).by(1L).get(unsigned(4)).valueAt(0L)))
.containsExactly(1L, 0L);
.containsExactly(1L, 0L);
}
@Test // DATAREDIS-562
@@ -2847,13 +2874,13 @@ public class LettuceClusterConnectionTests implements ClusterConnectionTests {
clusterConnection.stringCommands().bitField(LettuceConverters.toBytes(KEY_1),
create().set(INT_8).valueAt(BitFieldSubCommands.Offset.offset(0L).multipliedByTypeLength()).to(100L)
.set(INT_8).valueAt(BitFieldSubCommands.Offset.offset(1L).multipliedByTypeLength()).to(200L)))
.containsExactly(0L, 0L);
.containsExactly(0L, 0L);
assertThat(
clusterConnection.stringCommands()
.bitField(LettuceConverters.toBytes(KEY_1),
create().get(INT_8).valueAt(BitFieldSubCommands.Offset.offset(0L).multipliedByTypeLength()).get(INT_8)
.valueAt(BitFieldSubCommands.Offset.offset(1L).multipliedByTypeLength()))).containsExactly(100L,
-56L);
.valueAt(BitFieldSubCommands.Offset.offset(1L).multipliedByTypeLength())))
.containsExactly(100L, -56L);
}
@Test // DATAREDIS-1103
@@ -2864,7 +2891,7 @@ public class LettuceClusterConnectionTests implements ClusterConnectionTests {
assertThat(
clusterConnection.stringCommands().set(KEY_1_BYTES, VALUE_2_BYTES, Expiration.keepTtl(), SetOption.upsert()))
.isTrue();
.isTrue();
assertThat(nativeConnection.ttl(KEY_1)).isCloseTo(expireSeconds, Offset.offset(5L));
assertThat(nativeConnection.get(KEY_1)).isEqualTo(VALUE_2);