DATAREDIS-1084 - Polishing.

Rename PendingMessage.getStringId to PendingMessage.getIdAsString for consistent naming scheme. Remove getElapsedTimeSinceLastDeliveryMS method for now in favor of getElapsedTimeSinceLastDelivery. Use primitive long instead of boxed wrapper to avoid nullability where possible. Move PendingMessage* converters to StreamConverters. Add assertions. Original pull request: spring-projects#512.

Author: mp911de

src/main/java/org/springframework/data/redis/connection/ReactiveStreamCommands.java

@@ -353,7 +353,7 @@ default Flux<ByteBufferRecord> xClaim(ByteBuffer key, String group, String newOw
 
 /**
  * Change the ownership of a pending message to the given new {@literal consumer}.
- * 
+ *
  * @param commands must not be {@literal null}.
  * @return
  * @see <a href="https://redis.io/commands/xclaim">Redis Documentation: XCLAIM</a>
@@ -365,27 +365,28 @@ default Flux<ByteBufferRecord> xClaim(ByteBuffer key, String group, String newOw
  * {@code XCLAIM} command parameters.
  *
  * @see <a href="https://redis.io/commands/xclaim">Redis Documentation: XCLAIM</a>
+ * @since 2.3
  */
 class XClaimCommand extends KeyCommand {
 
 private final String groupName;
-private final String consumerName;
+private final String newOwner;
 private final XClaimOptions options;
 
-private XClaimCommand(@Nullable ByteBuffer key, String groupName, String consumerName, XClaimOptions options) {
+private XClaimCommand(@Nullable ByteBuffer key, String groupName, String newOwner, XClaimOptions options) {
 
 super(key);
 this.groupName = groupName;
-this.consumerName = consumerName;
+this.newOwner = newOwner;
 this.options = options;
 }
 
 public XClaimOptions getOptions() {
 return options;
 }
 
-public String getConsumerName() {
-return consumerName;
+public String getNewOwner() {
+return newOwner;
 }
 
 public String getGroupName() {
@@ -672,7 +673,7 @@ private PendingRecordsCommand(ByteBuffer key, String groupName, @Nullable String
 
 /**
  * Create new unbounded {@link PendingRecordsCommand}.
- * 
+ *
  * @param key the {@literal key} the stream is stored at. Must not be {@literal null}.
  * @param groupName the name of the {@literal consumer group}. Must not be {@literal null}.
  * @return new instance of {@link PendingRecordsCommand}.

src/main/java/org/springframework/data/redis/connection/RedisStreamCommands.java

@@ -27,6 +27,7 @@
 import org.springframework.data.redis.connection.RedisZSetCommands.Limit;
 import org.springframework.data.redis.connection.stream.*;
 import org.springframework.lang.Nullable;
+import org.springframework.util.Assert;
 import org.springframework.util.StringUtils;
 
 /**
@@ -121,7 +122,7 @@ default List<ByteRecord> xClaim(byte[] key, String group, String newOwner, Durat
 
 /**
  * Change the ownership of a pending message to the given new {@literal consumer}.
- * 
+ *
  * @param key the {@literal key} the stream is stored at.
  * @param group the name of the {@literal consumer group}.
  * @param newOwner the name of the new {@literal consumer}.
@@ -145,8 +146,8 @@ class XClaimOptions {
 private final @Nullable Long retryCount;
 private final boolean force;
 
-private XClaimOptions(List<RecordId> ids, Duration minIdleTime, Duration idleTime, Instant unixTime,
-Long retryCount, boolean force) {
+private XClaimOptions(List<RecordId> ids, Duration minIdleTime, @Nullable Duration idleTime,
+@Nullable Instant unixTime, @Nullable Long retryCount, boolean force) {
 
 this.ids = new ArrayList<>(ids);
 this.minIdleTime = minIdleTime;
@@ -159,7 +160,7 @@ private XClaimOptions(List<RecordId> ids, Duration minIdleTime, Duration idleTim
 /**
  * Set the {@literal min-idle-time} to limit the command to messages that have been idle for at at least the given
  * {@link Duration}.
- * 
+ *
  * @param minIdleTime must not be {@literal null}.
  * @return new instance of {@link XClaimOptions}.
  */
@@ -206,7 +207,7 @@ public XClaimOptions time(Instant unixTime) {
  * @param retryCount can be {@literal null}. If {@literal null} no change to the retry counter will be made.
  * @return new instance of {@link XClaimOptions}.
  */
-public XClaimOptions retryCount(@Nullable Long retryCount) {
+public XClaimOptions retryCount(long retryCount) {
 return new XClaimOptions(ids, minIdleTime, idleTime, unixTime, retryCount, force);
 }
 
@@ -249,7 +250,7 @@ public Duration getMinIdleTime() {
 
 /**
  * Get the {@literal IDLE ms} time.
- * 
+ *
  * @return can be {@literal null}.
  */
 @Nullable
@@ -259,7 +260,7 @@ public Duration getIdleTime() {
 
 /**
  * Get the {@literal TIME ms-unix-time}
- * 
+ *
  * @return
  */
 @Nullable
@@ -269,7 +270,7 @@ public Instant getUnixTime() {
 
 /**
  * Get the {@literal RETRYCOUNT count}.
- * 
+ *
  * @return
  */
 @Nullable
@@ -279,7 +280,7 @@ public Long getRetryCount() {
 
 /**
  * Get the {@literal FORCE} flag.
- * 
+ *
  * @return
  */
 public boolean isForce() {
@@ -288,9 +289,12 @@ public boolean isForce() {
 
 public static class XClaimOptionsBuilder {
 
-private Duration minIdleTime;
+private final Duration minIdleTime;
 
 XClaimOptionsBuilder(Duration minIdleTime) {
+
+Assert.notNull(minIdleTime, "Min idle time must not be null!");
+
 this.minIdleTime = minIdleTime;
 }
 
@@ -412,7 +416,7 @@ default Boolean xGroupDelConsumer(byte[] key, String groupName, String consumerN
 
 /**
  * Obtain the {@link PendingMessagesSummary} for a given {@literal consumer group}.
- * 
+ *
  * @param key the {@literal key} the stream is stored at. Must not be {@literal null}.
  * @param groupName the name of the {@literal consumer group}. Must not be {@literal null}.
  * @return a summary of pending messages within the given {@literal consumer group} or {@literal null} when used in
@@ -425,7 +429,7 @@ default Boolean xGroupDelConsumer(byte[] key, String groupName, String consumerN
 
 /**
  * Obtained detailed information about all pending messages for a given {@link Consumer}.
- * 
+ *
  * @param key the {@literal key} the stream is stored at. Must not be {@literal null}.
  * @param consumer the consumer to fetch {@link PendingMessages} for. Must not be {@literal null}.
  * @return pending messages for the given {@link Consumer} or {@literal null} when used in pipeline / transaction.
@@ -542,7 +546,7 @@ private XPendingOptions(@Nullable String consumerName, Range<?> range, @Nullable
 
 /**
  * Create new {@link XPendingOptions} with an unbounded {@link Range} ({@literal - +}).
- * 
+ *
  * @return new instance of {@link XPendingOptions}.
  */
 public static XPendingOptions unbounded() {
@@ -561,7 +565,7 @@ public static XPendingOptions unbounded(Long count) {
 
 /**
  * Create new {@link XPendingOptions} with given {@link Range} and limit.
- * 
+ *
  * @return new instance of {@link XPendingOptions}.
  */
 public static XPendingOptions range(Range<?> range, Long count) {
@@ -570,7 +574,7 @@ public static XPendingOptions range(Range<?> range, Long count) {
 
 /**
  * Append given consumer.
- * 
+ *
  * @param consumerName must not be {@literal null}.
  * @return new instance of {@link XPendingOptions}.
  */

src/main/java/org/springframework/data/redis/connection/lettuce/LettuceConverters.java

@@ -117,8 +117,6 @@ abstract public class LettuceConverters extends Converters {
 private static final Converter<KeyValue<Object, Object>, Object> KEY_VALUE_UNWRAPPER;
 private static final ListConverter<KeyValue<Object, Object>, Object> KEY_VALUE_LIST_UNWRAPPER;
 private static final Converter<TransactionResult, List<Object>> TRANSACTION_RESULT_UNWRAPPER;
-private static final BiFunction<List<Object>, String, org.springframework.data.redis.connection.stream.PendingMessages> PENDING_MESSAGES_CONVERTER;
-private static final BiFunction<List<Object>, String, PendingMessagesSummary> PENDING_MESSAGES_SUMMARY_CONVERTER;
 
 public static final byte[] PLUS_BYTES;
 public static final byte[] MINUS_BYTES;
@@ -320,69 +318,7 @@ private Set<Flag> parseFlags(Set<NodeFlag> source) {
 
 TRANSACTION_RESULT_UNWRAPPER = transactionResult -> transactionResult.stream().collect(Collectors.toList());
 
-PENDING_MESSAGES_CONVERTER = (source, groupName) -> {
 
-List<Object> target = source.stream().map(LettuceConverters::preConvertNativeValues).collect(Collectors.toList());
-List<PendingMessage> pendingMessages = PendingParser.parseRange(target);
-
-List<org.springframework.data.redis.connection.stream.PendingMessage> messages = pendingMessages.stream()
-.map(it -> {
-
-RecordId id = RecordId.of(it.getId());
-Consumer consumer = Consumer.from(groupName, it.getConsumer());
-
-return new org.springframework.data.redis.connection.stream.PendingMessage(id, consumer,
-Duration.ofMillis(it.getMsSinceLastDelivery()), it.getRedeliveryCount());
-
-}).collect(Collectors.toList());
-
-return new org.springframework.data.redis.connection.stream.PendingMessages(groupName, messages);
-
-};
-
-PENDING_MESSAGES_SUMMARY_CONVERTER = (source, groupName) -> {
-
-List<Object> target = source.stream().map(LettuceConverters::preConvertNativeValues).collect(Collectors.toList());
-
-PendingMessages pendingMessages = PendingParser.parse(target);
-org.springframework.data.domain.Range<String> range = org.springframework.data.domain.Range.open(
-pendingMessages.getMessageIds().getLower().getValue(), pendingMessages.getMessageIds().getUpper().getValue());
-
-return new PendingMessagesSummary(groupName, pendingMessages.getCount(), range,
-pendingMessages.getConsumerMessageCount());
-};
-}
-
-/**
- * We need to convert values into the correct target type since lettuce will give us {@link ByteBuffer} or arrays but
- * the parser requires us to have them as {@link String} or numeric values. Oh and {@literal null} values aren't real
- * good citizens as well, so we make them empty strings instead - see it works - somehow ;P
- * 
- * @param value dont't get me started om this.
- * @return preconverted values that Lettuce parsers are able to understand \ö/.
- */
-private static Object preConvertNativeValues(Object value) {
-
-if (value instanceof ByteBuffer || value instanceof byte[]) {
-
-byte[] targetArray = value instanceof ByteBuffer ? ByteUtils.getBytes((ByteBuffer) value) : (byte[]) value;
-String tmp = toString(targetArray);
-
-try {
-return NumberUtils.parseNumber(tmp, Long.class);
-} catch (NumberFormatException e) {
-return tmp;
-}
-}
-if (value instanceof List) {
-List<Object> targetList = new ArrayList<>();
-for (Object it : (List) value) {
-targetList.add(preConvertNativeValues(it));
-}
-return targetList;
-}
-
-return value != null ? value : "";
 }
 
 public static List<Tuple> toTuple(List<byte[]> list) {
@@ -1117,32 +1053,6 @@ static long getUpperBoundIndex(org.springframework.data.domain.Range<Long> range
 return getUpperBound(range).orElse(INDEXED_RANGE_END);
 }
 
-/**
- * Convert the raw lettuce xpending result to {@link PendingMessages}.
- *
- * @param groupName the group name
- * @param range the range of messages requested
- * @param source the raw lettuce response.
- * @return
- * @since 2.3
- */
-static org.springframework.data.redis.connection.stream.PendingMessages toPendingMessages(String groupName,
-org.springframework.data.domain.Range<?> range, List<Object> source) {
-return PENDING_MESSAGES_CONVERTER.apply(source, groupName).withinRange(range);
-}
-
-/**
- * Convert the raw lettuce xpending result to {@link PendingMessagesSummary}.
- * 
- * @param groupName
- * @param source the raw lettuce response.
- * @return
- * @since 2.3
- */
-static PendingMessagesSummary toPendingMessagesInfo(String groupName, List<Object> source) {
-return PENDING_MESSAGES_SUMMARY_CONVERTER.apply(source, groupName);
-}
-
 /**
  * @author Christoph Strobl
  * @since 1.8

src/main/java/org/springframework/data/redis/connection/lettuce/LettuceReactiveStreamCommands.java

@@ -121,7 +121,7 @@ public Flux<CommandResponse<XClaimCommand, Flux<RecordId>>> xClaimJustId(Publish
 
 String[] ids = command.getOptions().getIdsAsStringArray();
 io.lettuce.core.Consumer<ByteBuffer> from = io.lettuce.core.Consumer
-.from(ByteUtils.getByteBuffer(command.getGroupName()), ByteUtils.getByteBuffer(command.getConsumerName()));
+.from(ByteUtils.getByteBuffer(command.getGroupName()), ByteUtils.getByteBuffer(command.getNewOwner()));
 XClaimArgs args = StreamConverters.toXClaimArgs(command.getOptions());
 
 Flux<RecordId> result = cmd.xclaim(command.getKey(), from, args, ids).map(it -> RecordId.of(it.getId()));
@@ -140,7 +140,7 @@ public Flux<CommandResponse<XClaimCommand, Flux<ByteBufferRecord>>> xClaim(Publi
 
 String[] ids = command.getOptions().getIdsAsStringArray();
 io.lettuce.core.Consumer<ByteBuffer> from = io.lettuce.core.Consumer
-.from(ByteUtils.getByteBuffer(command.getGroupName()), ByteUtils.getByteBuffer(command.getConsumerName()));
+.from(ByteUtils.getByteBuffer(command.getGroupName()), ByteUtils.getByteBuffer(command.getNewOwner()));
 XClaimArgs args = StreamConverters.toXClaimArgs(command.getOptions());
 
 Flux<ByteBufferRecord> result = cmd.xclaim(command.getKey(), from, args, ids)
@@ -249,7 +249,7 @@ public Flux<CommandResponse<PendingRecordsCommand, PendingMessagesSummary>> xPen
 // end.
 // end.
 
-return LettuceConverters.toPendingMessagesInfo(command.getGroupName(), target);
+return StreamConverters.toPendingMessagesInfo(command.getGroupName(), target);
 }).map(value -> new CommandResponse<PendingRecordsCommand, PendingMessagesSummary>(command, value));
 }));
 }
@@ -268,15 +268,15 @@ public Flux<CommandResponse<PendingRecordsCommand, PendingMessages>> xPending(
 ByteBuffer groupName = ByteUtils.getByteBuffer(command.getGroupName());
 io.lettuce.core.Range<String> range = RangeConverter.toRangeWithDefault(command.getRange(), "-", "+");
 io.lettuce.core.Limit limit = command.isLimited() ? io.lettuce.core.Limit.from(command.getCount())
-: io.lettuce.core.Limit.from(Long.MAX_VALUE);
+: io.lettuce.core.Limit.unlimited();
 
 Flux<Object> 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);
 
 return publisher.collectList().map(it -> {
 
-return LettuceConverters.toPendingMessages(command.getGroupName(), command.getRange(), it);
+return StreamConverters.toPendingMessages(command.getGroupName(), command.getRange(), it);
 }).map(value -> new CommandResponse<PendingRecordsCommand, PendingMessages>(command, value));
 }));
 }