diff --git a/pom.xml b/pom.xml index 1403fde28..96e5280b8 100644 --- a/pom.xml +++ b/pom.xml @@ -22,7 +22,7 @@ 1.9.2 1.4.12 2.7.0 - 5.3.3.RELEASE + 6.0.0.RC1 3.3.0 1.01 4.1.51.Final diff --git a/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceReactiveScriptingCommands.java b/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceReactiveScriptingCommands.java index fca79aec8..ef3d0e680 100644 --- a/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceReactiveScriptingCommands.java +++ b/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceReactiveScriptingCommands.java @@ -26,6 +26,7 @@ import java.util.List; import org.springframework.data.redis.connection.ReactiveScriptingCommands; import org.springframework.data.redis.connection.ReturnType; +import org.springframework.data.redis.util.ByteUtils; import org.springframework.util.Assert; /** @@ -80,7 +81,7 @@ class LettuceReactiveScriptingCommands implements ReactiveScriptingCommands { Assert.notNull(script, "Script must not be null!"); - return connection.execute(cmd -> cmd.scriptLoad(script)).next(); + return connection.execute(cmd -> cmd.scriptLoad(ByteUtils.getBytes(script))).next(); } /* diff --git a/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceReactiveStreamCommands.java b/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceReactiveStreamCommands.java index 2cb940c0a..cf1baeb95 100644 --- a/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceReactiveStreamCommands.java +++ b/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceReactiveStreamCommands.java @@ -21,15 +21,16 @@ import io.lettuce.core.XGroupCreateArgs; import io.lettuce.core.XReadArgs; import io.lettuce.core.XReadArgs.StreamOffset; import io.lettuce.core.cluster.api.reactive.RedisClusterReactiveCommands; +import io.lettuce.core.models.stream.PendingMessage; import reactor.core.publisher.Flux; import java.nio.ByteBuffer; -import java.util.ArrayList; import java.util.Collection; import java.util.List; import java.util.function.Function; import org.reactivestreams.Publisher; + import org.springframework.data.redis.connection.ReactiveRedisConnection.CommandResponse; import org.springframework.data.redis.connection.ReactiveRedisConnection.KeyCommand; import org.springframework.data.redis.connection.ReactiveRedisConnection.NumericResponse; @@ -204,7 +205,7 @@ class LettuceReactiveStreamCommands implements ReactiveStreamCommands { .xgroupDelconsumer(command.getKey(), io.lettuce.core.Consumer.from(ByteUtils.getByteBuffer(command.getGroupName()), ByteUtils.getByteBuffer(command.getConsumerName()))) - .map(it -> new CommandResponse<>(command, Boolean.TRUE.equals(it) ? "OK" : "Error")); + .map(it -> new CommandResponse<>(command, "OK")); } if (command.getAction().equals(GroupCommandAction.DESTROY)) { @@ -243,25 +244,8 @@ class LettuceReactiveStreamCommands implements ReactiveStreamCommands { return connection.execute(cmd -> Flux.from(commands).concatMap(command -> { Assert.notNull(command.getKey(), "Key must not be null!"); - return cmd.xpending(command.getKey(), ByteUtils.getByteBuffer(command.getGroupName())).collectList().map(it -> { - - // begin - // {* hacking *} - // while (https://github.com/lettuce-io/lettuce-core/issues/1229 != resolved) begin - - ArrayList target = new ArrayList<>(it); - if (target.size() == 2 && target.get(1) instanceof List) { - target.add(1, null); - target.add(1, null); - } - while (target.size() < 4) { - target.add(null); - } - - // end. - // end. - - return StreamConverters.toPendingMessagesInfo(command.getGroupName(), target); + return cmd.xpending(command.getKey(), ByteUtils.getByteBuffer(command.getGroupName())).map(it -> { + return StreamConverters.toPendingMessagesInfo(command.getGroupName(), it); }).map(value -> new CommandResponse<>(command, value)); })); } @@ -282,7 +266,7 @@ class LettuceReactiveStreamCommands implements ReactiveStreamCommands { io.lettuce.core.Limit limit = command.isLimited() ? io.lettuce.core.Limit.from(command.getCount()) : io.lettuce.core.Limit.unlimited(); - Flux publisher = command.hasConsumer() ? cmd.xpending(command.getKey(), + Flux publisher = command.hasConsumer() ? cmd.xpending(command.getKey(), io.lettuce.core.Consumer.from(groupName, ByteUtils.getByteBuffer(command.getConsumerName())), range, limit) : cmd.xpending(command.getKey(), groupName, range, limit); @@ -353,7 +337,7 @@ class LettuceReactiveStreamCommands implements ReactiveStreamCommands { .map(it -> StreamRecords.newRecord().in(it.getStream()).withId(it.getId()).ofBuffer(it.getBody())); } - /* + /* * (non-Javadoc) * @see org.springframework.data.redis.connection.ReactiveStreamCommands#xInfo(org.reactivestreams.Publisher) */ @@ -370,7 +354,7 @@ class LettuceReactiveStreamCommands implements ReactiveStreamCommands { } - /* + /* * (non-Javadoc) * @see org.springframework.data.redis.connection.ReactiveStreamCommands#xInfoGroups(org.reactivestreams.Publisher) */ @@ -386,7 +370,7 @@ class LettuceReactiveStreamCommands implements ReactiveStreamCommands { })); } - /* + /* * (non-Javadoc) * @see org.springframework.data.redis.connection.ReactiveStreamCommands#xInfoConsumers(org.reactivestreams.Publisher) */ diff --git a/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceStreamCommands.java b/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceStreamCommands.java index af2f4cc38..0afc6b6ef 100644 --- a/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceStreamCommands.java +++ b/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceStreamCommands.java @@ -24,6 +24,7 @@ import io.lettuce.core.cluster.api.sync.RedisClusterCommands; import java.util.Arrays; import java.util.List; +import java.util.Objects; import java.util.function.Function; import org.springframework.dao.DataAccessException; @@ -268,14 +269,16 @@ class LettuceStreamCommands implements RedisStreamCommands { io.lettuce.core.Consumer lettuceConsumer = toConsumer(consumer); if (isPipelined()) { - pipeline(connection.newLettuceResult(getAsyncConnection().xgroupDelconsumer(key, lettuceConsumer))); + pipeline(connection.newLettuceResult(getAsyncConnection().xgroupDelconsumer(key, lettuceConsumer), + Objects::nonNull)); return null; } if (isQueueing()) { - transaction(connection.newLettuceResult(getAsyncConnection().xgroupDelconsumer(key, lettuceConsumer))); + transaction(connection.newLettuceResult(getAsyncConnection().xgroupDelconsumer(key, lettuceConsumer), + Objects::nonNull)); return null; } - return getConnection().xgroupDelconsumer(key, lettuceConsumer); + return Objects.nonNull(getConnection().xgroupDelconsumer(key, lettuceConsumer)); } catch (Exception ex) { throw convertLettuceAccessException(ex); } diff --git a/src/main/java/org/springframework/data/redis/connection/lettuce/StreamConverters.java b/src/main/java/org/springframework/data/redis/connection/lettuce/StreamConverters.java index 75e126905..733f10795 100644 --- a/src/main/java/org/springframework/data/redis/connection/lettuce/StreamConverters.java +++ b/src/main/java/org/springframework/data/redis/connection/lettuce/StreamConverters.java @@ -20,7 +20,6 @@ import io.lettuce.core.XClaimArgs; import io.lettuce.core.XReadArgs; import io.lettuce.core.models.stream.PendingMessage; import io.lettuce.core.models.stream.PendingMessages; -import io.lettuce.core.models.stream.PendingParser; import java.nio.ByteBuffer; import java.time.Duration; @@ -51,19 +50,17 @@ import org.springframework.util.NumberUtils; * @author Christoph Strobl * @since 2.2 */ -@SuppressWarnings({ "unchecked", "rawtypes" }) +@SuppressWarnings({ "rawtypes" }) class StreamConverters { private static final Converter>, List> MESSAGEs_TO_IDs = new ListConverter<>( messageToIdConverter()); - private static final BiFunction, String, org.springframework.data.redis.connection.stream.PendingMessages> PENDING_MESSAGES_CONVERTER = ( + private static final BiFunction, String, org.springframework.data.redis.connection.stream.PendingMessages> PENDING_MESSAGES_CONVERTER = ( source, groupName) -> { - List target = source.stream().map(StreamConverters::preConvertNativeValues).collect(Collectors.toList()); - List pendingMessages = PendingParser.parseRange(target); - List messages = pendingMessages.stream() + List messages = source.stream() .map(it -> { RecordId id = RecordId.of(it.getId()); @@ -78,17 +75,15 @@ class StreamConverters { }; - private static final BiFunction, String, PendingMessagesSummary> PENDING_MESSAGES_SUMMARY_CONVERTER = ( + private static final BiFunction PENDING_MESSAGES_SUMMARY_CONVERTER = ( source, groupName) -> { - List target = source.stream().map(StreamConverters::preConvertNativeValues).collect(Collectors.toList()); + org.springframework.data.domain.Range range = source.getMessageIds().isUnbounded() + ? org.springframework.data.domain.Range.unbounded() + : org.springframework.data.domain.Range.open(source.getMessageIds().getLower().getValue(), + source.getMessageIds().getUpper().getValue()); - PendingMessages pendingMessages = PendingParser.parse(target); - org.springframework.data.domain.Range range = org.springframework.data.domain.Range.open( - pendingMessages.getMessageIds().getLower().getValue(), pendingMessages.getMessageIds().getUpper().getValue()); - - return new PendingMessagesSummary(groupName, pendingMessages.getCount(), range, - pendingMessages.getConsumerMessageCount()); + return new PendingMessagesSummary(groupName, source.getCount(), range, source.getConsumerMessageCount()); }; /** @@ -138,7 +133,7 @@ class StreamConverters { * @since 2.3 */ static org.springframework.data.redis.connection.stream.PendingMessages toPendingMessages(String groupName, - org.springframework.data.domain.Range range, List source) { + org.springframework.data.domain.Range range, List source) { return PENDING_MESSAGES_CONVERTER.apply(source, groupName).withinRange(range); } @@ -150,7 +145,7 @@ class StreamConverters { * @return * @since 2.3 */ - static PendingMessagesSummary toPendingMessagesInfo(String groupName, List source) { + static PendingMessagesSummary toPendingMessagesInfo(String groupName, PendingMessages source) { return PENDING_MESSAGES_SUMMARY_CONVERTER.apply(source, groupName); } diff --git a/src/test/java/org/springframework/data/redis/connection/lettuce/LettuceConnectionFactoryTests.java b/src/test/java/org/springframework/data/redis/connection/lettuce/LettuceConnectionFactoryTests.java index 096399808..0c9974477 100644 --- a/src/test/java/org/springframework/data/redis/connection/lettuce/LettuceConnectionFactoryTests.java +++ b/src/test/java/org/springframework/data/redis/connection/lettuce/LettuceConnectionFactoryTests.java @@ -22,12 +22,14 @@ import io.lettuce.core.EpollProvider; import io.lettuce.core.KqueueProvider; import io.lettuce.core.ReadFrom; import io.lettuce.core.RedisException; +import io.lettuce.core.RedisFuture; import io.lettuce.core.api.async.RedisAsyncCommands; import io.lettuce.core.api.reactive.BaseRedisReactiveCommands; import reactor.test.StepVerifier; import java.io.File; import java.time.Duration; +import java.util.concurrent.ExecutionException; import java.util.function.Consumer; import org.apache.commons.pool2.impl.GenericObjectPoolConfig; @@ -230,11 +232,12 @@ public class LettuceConnectionFactoryTests { assertThat(conn2.isClosed()).isTrue(); // Give some time for native connection to asynchronously close Thread.sleep(100); + RedisFuture future = ((RedisAsyncCommands) conn2.getNativeConnection()).ping(); try { - ((RedisAsyncCommands) conn2.getNativeConnection()).ping(); + future.get(); fail("The native connection should be closed"); - } catch (RedisException e) { - // expected + } catch (ExecutionException e) { + // expected, Lettuce async failures are signalled on the Future } } diff --git a/src/test/java/org/springframework/data/redis/connection/lettuce/LettuceConnectionIntegrationTests.java b/src/test/java/org/springframework/data/redis/connection/lettuce/LettuceConnectionIntegrationTests.java index 2c842dd6c..78f9822bd 100644 --- a/src/test/java/org/springframework/data/redis/connection/lettuce/LettuceConnectionIntegrationTests.java +++ b/src/test/java/org/springframework/data/redis/connection/lettuce/LettuceConnectionIntegrationTests.java @@ -17,7 +17,6 @@ package org.springframework.data.redis.connection.lettuce; import static org.assertj.core.api.Assertions.*; import static org.junit.Assume.*; -import static org.springframework.data.redis.SpinBarrier.*; import io.lettuce.core.api.async.RedisAsyncCommands; diff --git a/src/test/java/org/springframework/data/redis/connection/lettuce/LettuceReactiveStreamCommandsTests.java b/src/test/java/org/springframework/data/redis/connection/lettuce/LettuceReactiveStreamCommandsTests.java index 57b6ec78c..61773cb71 100644 --- a/src/test/java/org/springframework/data/redis/connection/lettuce/LettuceReactiveStreamCommandsTests.java +++ b/src/test/java/org/springframework/data/redis/connection/lettuce/LettuceReactiveStreamCommandsTests.java @@ -50,9 +50,7 @@ public class LettuceReactiveStreamCommandsTests extends LettuceReactiveCommandsT @Before public void before() { - - // TODO: Upgrade to 5.0 - assumeTrue(RedisTestProfileValueSource.atLeast("redisVersion", "4.9")); + assumeTrue(RedisTestProfileValueSource.atLeast("redisVersion", "5.0")); } @Test // DATAREDIS-864 diff --git a/src/test/java/org/springframework/data/redis/test/util/LettuceRedisClientProvider.java b/src/test/java/org/springframework/data/redis/test/util/LettuceRedisClientProvider.java index b68287821..eb4ec8bab 100644 --- a/src/test/java/org/springframework/data/redis/test/util/LettuceRedisClientProvider.java +++ b/src/test/java/org/springframework/data/redis/test/util/LettuceRedisClientProvider.java @@ -15,10 +15,13 @@ */ package org.springframework.data.redis.test.util; +import io.lettuce.core.ClientOptions; import io.lettuce.core.RedisClient; import io.lettuce.core.RedisURI; +import io.lettuce.core.protocol.ProtocolVersion; import org.junit.rules.ExternalResource; + import org.springframework.data.redis.SettingsUtils; import org.springframework.data.redis.connection.lettuce.LettuceTestClientResources; @@ -44,6 +47,8 @@ public class LettuceRedisClientProvider extends ExternalResource { client = RedisClient.create(LettuceTestClientResources.getSharedClientResources(), RedisURI.builder().withHost(host).withPort(port).build()); + client.setOptions( + ClientOptions.builder().protocolVersion(ProtocolVersion.RESP2).pingBeforeActivateConnection(false).build()); } @Override diff --git a/src/test/java/org/springframework/data/redis/test/util/LettuceRedisClusterClientProvider.java b/src/test/java/org/springframework/data/redis/test/util/LettuceRedisClusterClientProvider.java index 59c64a030..7f5eabbd0 100644 --- a/src/test/java/org/springframework/data/redis/test/util/LettuceRedisClusterClientProvider.java +++ b/src/test/java/org/springframework/data/redis/test/util/LettuceRedisClusterClientProvider.java @@ -15,12 +15,15 @@ */ package org.springframework.data.redis.test.util; -import org.junit.rules.ExternalResource; -import org.springframework.data.redis.connection.lettuce.LettuceTestClientResources; - import io.lettuce.core.RedisURI; +import io.lettuce.core.cluster.ClusterClientOptions; import io.lettuce.core.cluster.RedisClusterClient; import io.lettuce.core.cluster.api.StatefulRedisClusterConnection; +import io.lettuce.core.protocol.ProtocolVersion; + +import org.junit.rules.ExternalResource; + +import org.springframework.data.redis.connection.lettuce.LettuceTestClientResources; /** * @author Christoph Strobl @@ -44,6 +47,8 @@ public class LettuceRedisClusterClientProvider extends ExternalResource { client = RedisClusterClient.create(LettuceTestClientResources.getSharedClientResources(), RedisURI.builder().withHost(host).withPort(port).build()); + client.setOptions(ClusterClientOptions.builder().protocolVersion(ProtocolVersion.RESP2) + .pingBeforeActivateConnection(false).build()); } @Override