From 87d3bb5660c8ab1ebec34ea17afdc64ad15d007c Mon Sep 17 00:00:00 2001 From: Mark Paluch Date: Tue, 10 Mar 2020 09:39:50 +0100 Subject: [PATCH] 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: #512. --- .../connection/ReactiveStreamCommands.java | 15 +-- .../redis/connection/RedisStreamCommands.java | 34 +++--- .../connection/lettuce/LettuceConverters.java | 90 --------------- .../LettuceReactiveStreamCommands.java | 10 +- .../lettuce/LettuceStreamCommands.java | 45 ++++---- .../connection/lettuce/StreamConverters.java | 108 +++++++++++++++++- .../connection/stream/PendingMessage.java | 28 ++--- .../connection/stream/PendingMessages.java | 20 ++-- .../stream/PendingMessagesSummary.java | 24 ++-- .../core/DefaultReactiveStreamOperations.java | 4 +- .../redis/core/DefaultStreamOperations.java | 4 +- .../redis/core/ReactiveStreamOperations.java | 8 +- .../data/redis/core/StreamOperations.java | 10 +- .../AbstractConnectionIntegrationTests.java | 8 +- .../LettuceReactiveStreamCommandsTests.java | 4 +- 15 files changed, 216 insertions(+), 196 deletions(-) diff --git a/src/main/java/org/springframework/data/redis/connection/ReactiveStreamCommands.java b/src/main/java/org/springframework/data/redis/connection/ReactiveStreamCommands.java index 4f76a0735..fdebeedd1 100644 --- a/src/main/java/org/springframework/data/redis/connection/ReactiveStreamCommands.java +++ b/src/main/java/org/springframework/data/redis/connection/ReactiveStreamCommands.java @@ -353,7 +353,7 @@ public interface ReactiveStreamCommands { /** * Change the ownership of a pending message to the given new {@literal consumer}. - * + * * @param commands must not be {@literal null}. * @return * @see Redis Documentation: XCLAIM @@ -365,18 +365,19 @@ public interface ReactiveStreamCommands { * {@code XCLAIM} command parameters. * * @see Redis Documentation: XCLAIM + * @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; } @@ -384,8 +385,8 @@ public interface ReactiveStreamCommands { return options; } - public String getConsumerName() { - return consumerName; + public String getNewOwner() { + return newOwner; } public String getGroupName() { @@ -672,7 +673,7 @@ public interface ReactiveStreamCommands { /** * 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}. diff --git a/src/main/java/org/springframework/data/redis/connection/RedisStreamCommands.java b/src/main/java/org/springframework/data/redis/connection/RedisStreamCommands.java index a2984b6e9..50dfa4b73 100644 --- a/src/main/java/org/springframework/data/redis/connection/RedisStreamCommands.java +++ b/src/main/java/org/springframework/data/redis/connection/RedisStreamCommands.java @@ -27,6 +27,7 @@ import org.springframework.data.domain.Range; 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 @@ public interface RedisStreamCommands { /** * 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 @@ public interface RedisStreamCommands { private final @Nullable Long retryCount; private final boolean force; - private XClaimOptions(List ids, Duration minIdleTime, Duration idleTime, Instant unixTime, - Long retryCount, boolean force) { + private XClaimOptions(List 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 @@ public interface RedisStreamCommands { /** * 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 interface RedisStreamCommands { * @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 interface RedisStreamCommands { /** * Get the {@literal IDLE ms} time. - * + * * @return can be {@literal null}. */ @Nullable @@ -259,7 +260,7 @@ public interface RedisStreamCommands { /** * Get the {@literal TIME ms-unix-time} - * + * * @return */ @Nullable @@ -269,7 +270,7 @@ public interface RedisStreamCommands { /** * Get the {@literal RETRYCOUNT count}. - * + * * @return */ @Nullable @@ -279,7 +280,7 @@ public interface RedisStreamCommands { /** * Get the {@literal FORCE} flag. - * + * * @return */ public boolean isForce() { @@ -288,9 +289,12 @@ public interface RedisStreamCommands { 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 @@ public interface RedisStreamCommands { /** * 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 @@ public interface RedisStreamCommands { /** * 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 @@ public interface RedisStreamCommands { /** * 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 interface RedisStreamCommands { /** * 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 interface RedisStreamCommands { /** * Append given consumer. - * + * * @param consumerName must not be {@literal null}. * @return new instance of {@link XPendingOptions}. */ diff --git a/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceConverters.java b/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceConverters.java index e94129f12..44be5a725 100644 --- a/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceConverters.java +++ b/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, Object> KEY_VALUE_UNWRAPPER; private static final ListConverter, Object> KEY_VALUE_LIST_UNWRAPPER; private static final Converter> TRANSACTION_RESULT_UNWRAPPER; - private static final BiFunction, String, org.springframework.data.redis.connection.stream.PendingMessages> PENDING_MESSAGES_CONVERTER; - private static final BiFunction, String, PendingMessagesSummary> PENDING_MESSAGES_SUMMARY_CONVERTER; public static final byte[] PLUS_BYTES; public static final byte[] MINUS_BYTES; @@ -320,69 +318,7 @@ abstract public class LettuceConverters extends Converters { TRANSACTION_RESULT_UNWRAPPER = transactionResult -> transactionResult.stream().collect(Collectors.toList()); - PENDING_MESSAGES_CONVERTER = (source, groupName) -> { - List target = source.stream().map(LettuceConverters::preConvertNativeValues).collect(Collectors.toList()); - List pendingMessages = PendingParser.parseRange(target); - - List 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 target = source.stream().map(LettuceConverters::preConvertNativeValues).collect(Collectors.toList()); - - 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()); - }; - } - - /** - * 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 targetList = new ArrayList<>(); - for (Object it : (List) value) { - targetList.add(preConvertNativeValues(it)); - } - return targetList; - } - - return value != null ? value : ""; } public static List toTuple(List list) { @@ -1117,32 +1053,6 @@ abstract public class LettuceConverters extends Converters { 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 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 source) { - return PENDING_MESSAGES_SUMMARY_CONVERTER.apply(source, groupName); - } - /** * @author Christoph Strobl * @since 1.8 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 67c5a8392..1ed1fdda1 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 @@ -121,7 +121,7 @@ class LettuceReactiveStreamCommands implements ReactiveStreamCommands { String[] ids = command.getOptions().getIdsAsStringArray(); io.lettuce.core.Consumer 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 result = cmd.xclaim(command.getKey(), from, args, ids).map(it -> RecordId.of(it.getId())); @@ -140,7 +140,7 @@ class LettuceReactiveStreamCommands implements ReactiveStreamCommands { String[] ids = command.getOptions().getIdsAsStringArray(); io.lettuce.core.Consumer 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 result = cmd.xclaim(command.getKey(), from, args, ids) @@ -249,7 +249,7 @@ class LettuceReactiveStreamCommands implements ReactiveStreamCommands { // end. // end. - return LettuceConverters.toPendingMessagesInfo(command.getGroupName(), target); + return StreamConverters.toPendingMessagesInfo(command.getGroupName(), target); }).map(value -> new CommandResponse(command, value)); })); } @@ -268,7 +268,7 @@ class LettuceReactiveStreamCommands implements ReactiveStreamCommands { ByteBuffer groupName = ByteUtils.getByteBuffer(command.getGroupName()); io.lettuce.core.Range 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 publisher = command.hasConsumer() ? cmd.xpending(command.getKey(), io.lettuce.core.Consumer.from(groupName, ByteUtils.getByteBuffer(command.getConsumerName())), range, limit) @@ -276,7 +276,7 @@ class LettuceReactiveStreamCommands implements ReactiveStreamCommands { 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(command, value)); })); } 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 fd26ed339..769105e83 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 @@ -322,15 +322,15 @@ class LettuceStreamCommands implements RedisStreamCommands { try { if (isPipelined()) { pipeline(connection.newLettuceResult(getAsyncConnection().xpending(key, group), - it -> LettuceConverters.toPendingMessagesInfo(groupName, it))); + it -> StreamConverters.toPendingMessagesInfo(groupName, it))); return null; } if (isQueueing()) { transaction(connection.newLettuceResult(getAsyncConnection().xpending(key, group), - it -> LettuceConverters.toPendingMessagesInfo(groupName, it))); + it -> StreamConverters.toPendingMessagesInfo(groupName, it))); return null; } - return LettuceConverters.toPendingMessagesInfo(groupName, getConnection().xpending(key, group)); + return StreamConverters.toPendingMessagesInfo(groupName, getConnection().xpending(key, group)); } catch (Exception ex) { throw convertLettuceAccessException(ex); } @@ -346,37 +346,40 @@ class LettuceStreamCommands implements RedisStreamCommands { byte[] group = LettuceConverters.toBytes(groupName); io.lettuce.core.Range range = RangeConverter.toRangeWithDefault(options.getRange(), "-", "+"); io.lettuce.core.Limit limit = options.isLimited() ? io.lettuce.core.Limit.from(options.getCount()) - : io.lettuce.core.Limit.from(Long.MAX_VALUE); + : io.lettuce.core.Limit.unlimited(); try { - if (!options.hasConsumer()) { - if (isPipelined()) { - pipeline(connection.newLettuceResult(getAsyncConnection().xpending(key, group, range, limit), - it -> LettuceConverters.toPendingMessages(groupName, options.getRange(), (List) it))); - return null; - } - if (isQueueing()) { - transaction(connection.newLettuceResult(getAsyncConnection().xpending(key, group, range, limit), - it -> LettuceConverters.toPendingMessages(groupName, options.getRange(), (List) it))); - return null; - } - return LettuceConverters.toPendingMessages(groupName, options.getRange(), - getConnection().xpending(key, group, range, limit)); - } else { + if (options.hasConsumer()) { + if (isPipelined()) { pipeline(connection.newLettuceResult(getAsyncConnection().xpending(key, io.lettuce.core.Consumer.from(group, LettuceConverters.toBytes(options.getConsumerName())), range, limit), - it -> LettuceConverters.toPendingMessages(groupName, options.getRange(), (List) it))); + it -> StreamConverters.toPendingMessages(groupName, options.getRange(), it))); return null; } if (isQueueing()) { transaction(connection.newLettuceResult(getAsyncConnection().xpending(key, io.lettuce.core.Consumer.from(group, LettuceConverters.toBytes(options.getConsumerName())), range, limit), - it -> LettuceConverters.toPendingMessages(groupName, options.getRange(), (List) it))); + it -> StreamConverters.toPendingMessages(groupName, options.getRange(), it))); return null; } - return LettuceConverters.toPendingMessages(groupName, options.getRange(), getConnection().xpending(key, + return StreamConverters.toPendingMessages(groupName, options.getRange(), getConnection().xpending(key, io.lettuce.core.Consumer.from(group, LettuceConverters.toBytes(options.getConsumerName())), range, limit)); + + } else { + + if (isPipelined()) { + pipeline(connection.newLettuceResult(getAsyncConnection().xpending(key, group, range, limit), + it -> StreamConverters.toPendingMessages(groupName, options.getRange(), it))); + return null; + } + if (isQueueing()) { + transaction(connection.newLettuceResult(getAsyncConnection().xpending(key, group, range, limit), + it -> StreamConverters.toPendingMessages(groupName, options.getRange(), it))); + return null; + } + return StreamConverters.toPendingMessages(groupName, options.getRange(), + getConnection().xpending(key, group, range, limit)); } } 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 77bbc8298..75e126905 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 @@ -18,17 +18,28 @@ package org.springframework.data.redis.connection.lettuce; import io.lettuce.core.StreamMessage; 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; +import java.util.ArrayList; import java.util.List; +import java.util.function.BiFunction; +import java.util.stream.Collectors; import org.springframework.core.convert.converter.Converter; import org.springframework.data.redis.connection.RedisStreamCommands.XClaimOptions; import org.springframework.data.redis.connection.convert.ListConverter; import org.springframework.data.redis.connection.stream.ByteRecord; +import org.springframework.data.redis.connection.stream.Consumer; +import org.springframework.data.redis.connection.stream.PendingMessagesSummary; import org.springframework.data.redis.connection.stream.RecordId; import org.springframework.data.redis.connection.stream.StreamReadOptions; import org.springframework.data.redis.connection.stream.StreamRecords; -import org.springframework.lang.Nullable; +import org.springframework.data.redis.util.ByteUtils; +import org.springframework.util.NumberUtils; /** * Converters for Redis Stream-specific types. @@ -46,6 +57,40 @@ 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 = ( + source, groupName) -> { + + List target = source.stream().map(StreamConverters::preConvertNativeValues).collect(Collectors.toList()); + List pendingMessages = PendingParser.parseRange(target); + + List 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); + + }; + + private static final BiFunction, String, PendingMessagesSummary> PENDING_MESSAGES_SUMMARY_CONVERTER = ( + source, groupName) -> { + + List target = source.stream().map(StreamConverters::preConvertNativeValues).collect(Collectors.toList()); + + 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()); + }; + /** * Convert {@link StreamReadOptions} to Lettuce's {@link XReadArgs}. * @@ -83,6 +128,64 @@ class StreamConverters { return MESSAGEs_TO_IDs; } + /** + * 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 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 source) { + return PENDING_MESSAGES_SUMMARY_CONVERTER.apply(source, groupName); + } + + /** + * 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 = LettuceConverters.toString(targetArray); + + try { + return NumberUtils.parseNumber(tmp, Long.class); + } catch (NumberFormatException e) { + return tmp; + } + } + if (value instanceof List) { + List targetList = new ArrayList<>(); + for (Object it : (List) value) { + targetList.add(preConvertNativeValues(it)); + } + return targetList; + } + + return value != null ? value : ""; + } + /** * {@link Converter} to convert {@link StreamReadOptions} to Lettuce's {@link XReadArgs}. */ @@ -116,13 +219,12 @@ class StreamConverters { /** * {@link Converter} to convert {@link XClaimOptions} to Lettuce's {@link XClaimArgs}. - * + * * @since 2.3 */ enum XClaimOptionsToXClaimArgsConverter implements Converter { INSTANCE; - @Nullable @Override public XClaimArgs convert(XClaimOptions source) { diff --git a/src/main/java/org/springframework/data/redis/connection/stream/PendingMessage.java b/src/main/java/org/springframework/data/redis/connection/stream/PendingMessage.java index ef9258876..db8abb067 100644 --- a/src/main/java/org/springframework/data/redis/connection/stream/PendingMessage.java +++ b/src/main/java/org/springframework/data/redis/connection/stream/PendingMessage.java @@ -5,7 +5,7 @@ * you may not use this file except in compliance with the License. * You may obtain a copy of the License at * - * http://www.apache.org/licenses/LICENSE-2.0 + * https://www.apache.org/licenses/LICENSE-2.0 * * Unless required by applicable law or agreed to in writing, software * distributed under the License is distributed on an "AS IS" BASIS, @@ -21,7 +21,7 @@ import java.time.Duration; * Value object representing a single pending message containing its {@literal ID}, the {@literal consumer} that fetched * the message and has still to acknowledge it, the time elapsed since the messages last delivery and the the total * number of times delivered. - * + * * @author Christoph Strobl * @since 2.3 */ @@ -33,7 +33,7 @@ public class PendingMessage { private final Long totalDeliveryCount; public PendingMessage(RecordId id, Consumer consumer, Duration elapsedTimeSinceLastDelivery, - Long totalDeliveryCount) { + long totalDeliveryCount) { this.id = id; this.consumer = consumer; @@ -51,13 +51,13 @@ public class PendingMessage { /** * @return the message id as {@link String}. */ - public String getStringId() { + public String getIdAsString() { return id.getValue(); } /** * The {@link Consumer} to acknowledge the message. - * + * * @return never {@literal null}. */ public Consumer getConsumer() { @@ -83,8 +83,9 @@ public class PendingMessage { } /** - * Get the time elapsed since the messages last delivery to the {@link #getConsumer() consumer}. - * + * Get the elapsed time (with milliseconds precision) since the messages last delivery to the {@link #getConsumer() + * consumer}. + * * @return never {@literal null}. */ public Duration getElapsedTimeSinceLastDelivery() { @@ -92,20 +93,11 @@ public class PendingMessage { } /** - * Get the milliseconds elapsed since the messages last delivery to the {@link #getConsumer() consumer}. + * Get the total number of times the messages has been delivered to the {@link #getConsumer() consumer}. * * @return never {@literal null}. */ - public Long getElapsedTimeSinceLastDeliveryMS() { - return elapsedTimeSinceLastDelivery.toMillis(); - } - - /** - * Get the total number of times the messages has been delivered to the {@link #getConsumer() consumer}. - * - * @return never {@literal null}. - */ - public Long getTotalDeliveryCount() { + public long getTotalDeliveryCount() { return totalDeliveryCount; } diff --git a/src/main/java/org/springframework/data/redis/connection/stream/PendingMessages.java b/src/main/java/org/springframework/data/redis/connection/stream/PendingMessages.java index 8fe21bc9a..29db327f6 100644 --- a/src/main/java/org/springframework/data/redis/connection/stream/PendingMessages.java +++ b/src/main/java/org/springframework/data/redis/connection/stream/PendingMessages.java @@ -5,7 +5,7 @@ * you may not use this file except in compliance with the License. * You may obtain a copy of the License at * - * http://www.apache.org/licenses/LICENSE-2.0 + * https://www.apache.org/licenses/LICENSE-2.0 * * Unless required by applicable law or agreed to in writing, software * distributed under the License is distributed on an "AS IS" BASIS, @@ -20,11 +20,12 @@ import java.util.List; import org.springframework.data.domain.Range; import org.springframework.data.util.Streamable; +import org.springframework.util.Assert; /** * Value object holding detailed information about pending messages in {@literal consumer group} for a given - * {@link org.springframework.data.redis.connection.SortParameters.Range} and offset. - * + * {@link org.springframework.data.domain.Range} and offset. + * * @author Christoph Strobl * @since 2.3 */ @@ -40,6 +41,9 @@ public class PendingMessages implements Streamable { public PendingMessages(String groupName, Range range, List pendingMessages) { + Assert.notNull(range, "Range must not be null"); + Assert.notNull(pendingMessages, "Pending Messages must not be null"); + this.groupName = groupName; this.range = range; this.pendingMessages = pendingMessages; @@ -47,7 +51,7 @@ public class PendingMessages implements Streamable { /** * Adds the range to the current {@link PendingMessages}. - * + * * @param range must not be {@literal null}. * @return new instance of {@link PendingMessages}. */ @@ -57,7 +61,7 @@ public class PendingMessages implements Streamable { /** * The {@literal consumer group} name. - * + * * @return never {@literal null}. */ public String getGroupName() { @@ -66,7 +70,7 @@ public class PendingMessages implements Streamable { /** * The {@link Range} pending messages have been loaded. - * + * * @return never {@literal null}. */ public Range getRange() { @@ -89,7 +93,7 @@ public class PendingMessages implements Streamable { /** * Get the {@link PendingMessage} at the given position. - * + * * @param index * @return the {@link PendingMessage} a the given index. * @throws IndexOutOfBoundsException if the index is out of range. @@ -99,7 +103,7 @@ public class PendingMessages implements Streamable { } /* - * (non-Javadoc) + * (non-Javadoc) * @see java.lang.Iterable#iterator() */ @Override diff --git a/src/main/java/org/springframework/data/redis/connection/stream/PendingMessagesSummary.java b/src/main/java/org/springframework/data/redis/connection/stream/PendingMessagesSummary.java index fa3314304..a814b3362 100644 --- a/src/main/java/org/springframework/data/redis/connection/stream/PendingMessagesSummary.java +++ b/src/main/java/org/springframework/data/redis/connection/stream/PendingMessagesSummary.java @@ -5,7 +5,7 @@ * you may not use this file except in compliance with the License. * You may obtain a copy of the License at * - * http://www.apache.org/licenses/LICENSE-2.0 + * https://www.apache.org/licenses/LICENSE-2.0 * * Unless required by applicable law or agreed to in writing, software * distributed under the License is distributed on an "AS IS" BASIS, @@ -19,11 +19,12 @@ import java.util.Collections; import java.util.Map; import org.springframework.data.domain.Range; +import org.springframework.util.Assert; /** * Value Object summarizing pending messages in a {@literal consumer group}. It contains the total number and ID range * of pending messages for this consumer group, as well as a collection of total pending messages per consumer. - * + * * @since 2.3 * @author Christoph Strobl */ @@ -34,9 +35,12 @@ public class PendingMessagesSummary { private final Range idRange; private final Map pendingMessagesPerConsumer; - public PendingMessagesSummary(String groupName, Long totalPendingMessages, Range idRange, + public PendingMessagesSummary(String groupName, long totalPendingMessages, Range idRange, Map pendingMessagesPerConsumer) { + Assert.notNull(idRange, "ID Range must not be null"); + Assert.notNull(pendingMessagesPerConsumer, "Pending Messages must not be null"); + this.groupName = groupName; this.totalPendingMessages = totalPendingMessages; this.idRange = idRange; @@ -54,7 +58,7 @@ public class PendingMessagesSummary { /** * Get the smallest ID among the pending messages. - * + * * @return never {@literal null}. */ public RecordId minRecordId() { @@ -63,7 +67,7 @@ public class PendingMessagesSummary { /** * Get the greatest ID among the pending messages. - * + * * @return never {@literal null}. */ public RecordId maxRecordId() { @@ -72,7 +76,7 @@ public class PendingMessagesSummary { /** * Get the smallest ID as {@link String} among the pending messages. - * + * * @return never {@literal null}. */ public String minMessageId() { @@ -81,7 +85,7 @@ public class PendingMessagesSummary { /** * Get the greatest ID as {@link String} among the pending messages. - * + * * @return never {@literal null}. */ public String maxMessageId() { @@ -91,9 +95,9 @@ public class PendingMessagesSummary { /** * Get the number of total pending messages within the {@literal consumer group}. * - * @return + * @return never {@literal null}. */ - public Long getTotalPendingMessages() { + public long getTotalPendingMessages() { return totalPendingMessages; } @@ -107,7 +111,7 @@ public class PendingMessagesSummary { /** * Obtain a map of every {@literal consumer} in the {@literal consumer group} with at least one pending message, and * the number of pending messages. - * + * * @return never {@literal null}. */ public Map getPendingMessagesPerConsumer() { diff --git a/src/main/java/org/springframework/data/redis/core/DefaultReactiveStreamOperations.java b/src/main/java/org/springframework/data/redis/core/DefaultReactiveStreamOperations.java index 61699361c..8360ed59e 100644 --- a/src/main/java/org/springframework/data/redis/core/DefaultReactiveStreamOperations.java +++ b/src/main/java/org/springframework/data/redis/core/DefaultReactiveStreamOperations.java @@ -190,7 +190,7 @@ class DefaultReactiveStreamOperations implements ReactiveStreamOperat * @see org.springframework.data.redis.core.StreamOperations#pending(java.lang.Object, java.lang.String, org.springframework.data.domain.Range, java.lang.Long) */ @Override - public Mono pending(K key, String group, Range range, Long count) { + public Mono pending(K key, String group, Range range, long count) { ByteBuffer rawKey = rawKey(key); return createMono(connection -> connection.xPending(rawKey, group, range, count)); @@ -201,7 +201,7 @@ class DefaultReactiveStreamOperations implements ReactiveStreamOperat * @see org.springframework.data.redis.core.StreamOperations#pending(java.lang.Object, org.springframework.data.redis.connection.stream.Consumer, org.springframework.data.domain.Range, java.lang.Long) */ @Override - public Mono pending(K key, Consumer consumer, Range range, Long count) { + public Mono pending(K key, Consumer consumer, Range range, long count) { ByteBuffer rawKey = rawKey(key); return createMono(connection -> connection.xPending(rawKey, consumer, range, count)); diff --git a/src/main/java/org/springframework/data/redis/core/DefaultStreamOperations.java b/src/main/java/org/springframework/data/redis/core/DefaultStreamOperations.java index 0aa06372c..6b15159c4 100644 --- a/src/main/java/org/springframework/data/redis/core/DefaultStreamOperations.java +++ b/src/main/java/org/springframework/data/redis/core/DefaultStreamOperations.java @@ -186,7 +186,7 @@ class DefaultStreamOperations extends AbstractOperations i * @see org.springframework.data.redis.core.StreamOperations#pending(java.lang.Object, java.lang.String, org.springframework.data.domain.Range, java.lang.Long) */ @Override - public PendingMessages pending(K key, String group, Range range, Long count) { + public PendingMessages pending(K key, String group, Range range, long count) { byte[] rawKey = rawKey(key); return execute(connection -> connection.xPending(rawKey, group, range, count), true); @@ -197,7 +197,7 @@ class DefaultStreamOperations extends AbstractOperations i * @see org.springframework.data.redis.core.StreamOperations#pending(java.lang.Object, org.springframework.data.redis.connection.stream.Consumer, org.springframework.data.domain.Range, java.lang.Long) */ @Override - public PendingMessages pending(K key, Consumer consumer, Range range, Long count) { + public PendingMessages pending(K key, Consumer consumer, Range range, long count) { byte[] rawKey = rawKey(key); return execute(connection -> connection.xPending(rawKey, consumer, range, count), true); diff --git a/src/main/java/org/springframework/data/redis/core/ReactiveStreamOperations.java b/src/main/java/org/springframework/data/redis/core/ReactiveStreamOperations.java index 21b7564ac..1c4e9305f 100644 --- a/src/main/java/org/springframework/data/redis/core/ReactiveStreamOperations.java +++ b/src/main/java/org/springframework/data/redis/core/ReactiveStreamOperations.java @@ -227,13 +227,13 @@ public interface ReactiveStreamOperations extends HashMapperProvider< * @param key the {@literal key} the stream is stored at. Must not be {@literal null}. * @param group the name of the {@literal consumer group}. Must not be {@literal null}. * @param range the range of messages ids to search within. Must not be {@literal null}. - * @param count limit the number of results. Must not be {@literal null}. + * @param count limit the number of results. * @return pending messages for the given {@literal consumer group} or {@literal null} when used in pipeline / * transaction. * @see Redis Documentation: xpending * @since 2.3 */ - Mono pending(K key, String group, Range range, Long count); + Mono pending(K key, String group, Range range, long count); /** * Obtain detailed information about pending {@link PendingMessage messages} for a given {@link Range} and @@ -242,12 +242,12 @@ public interface ReactiveStreamOperations extends HashMapperProvider< * @param key the {@literal key} the stream is stored at. Must not be {@literal null}. * @param consumer the name of the {@link Consumer}. Must not be {@literal null}. * @param range the range of messages ids to search within. Must not be {@literal null}. - * @param count limit the number of results. Must not be {@literal null}. + * @param count limit the number of results. * @return pending messages for the given {@link Consumer} or {@literal null} when used in pipeline / transaction. * @see Redis Documentation: xpending * @since 2.3 */ - Mono pending(K key, Consumer consumer, Range range, Long count); + Mono pending(K key, Consumer consumer, Range range, long count); /** * Get the length of a stream. diff --git a/src/main/java/org/springframework/data/redis/core/StreamOperations.java b/src/main/java/org/springframework/data/redis/core/StreamOperations.java index d3070dd91..a1137b0c3 100644 --- a/src/main/java/org/springframework/data/redis/core/StreamOperations.java +++ b/src/main/java/org/springframework/data/redis/core/StreamOperations.java @@ -226,13 +226,13 @@ public interface StreamOperations extends HashMapperProvider * @param key the {@literal key} the stream is stored at. Must not be {@literal null}. * @param group the name of the {@literal consumer group}. Must not be {@literal null}. * @param range the range of messages ids to search within. Must not be {@literal null}. - * @param count limit the number of results. Must not be {@literal null}. + * @param count limit the number of results. * @return pending messages for the given {@literal consumer group} or {@literal null} when used in pipeline / * transaction. * @see Redis Documentation: xpending * @since 2.3 */ - PendingMessages pending(K key, String group, Range range, Long count); + PendingMessages pending(K key, String group, Range range, long count); /** * Obtain detailed information about pending {@link PendingMessage messages} for a given {@link Range} and @@ -241,12 +241,12 @@ public interface StreamOperations extends HashMapperProvider * @param key the {@literal key} the stream is stored at. Must not be {@literal null}. * @param consumer the name of the {@link Consumer}. Must not be {@literal null}. * @param range the range of messages ids to search within. Must not be {@literal null}. - * @param count limit the number of results. Must not be {@literal null}. + * @param count limit the number of results. * @return pending messages for the given {@link Consumer} or {@literal null} when used in pipeline / transaction. * @see Redis Documentation: xpending * @since 2.3 */ - PendingMessages pending(K key, Consumer consumer, Range range, Long count); + PendingMessages pending(K key, Consumer consumer, Range range, long count); /** * Get the length of a stream. @@ -327,7 +327,7 @@ public interface StreamOperations extends HashMapperProvider /** * Read records from one or more {@link StreamOffset}s as {@link ObjectRecord}. - * + * * @param targetType the target type of the payload. * @param streams the streams to read from. * @return list with members of the resulting stream. {@literal null} when used in pipeline / transaction. diff --git a/src/test/java/org/springframework/data/redis/connection/AbstractConnectionIntegrationTests.java b/src/test/java/org/springframework/data/redis/connection/AbstractConnectionIntegrationTests.java index df348758c..32d9dad6d 100644 --- a/src/test/java/org/springframework/data/redis/connection/AbstractConnectionIntegrationTests.java +++ b/src/test/java/org/springframework/data/redis/connection/AbstractConnectionIntegrationTests.java @@ -3158,8 +3158,7 @@ public abstract class AbstractConnectionIntegrationTests { assertThat(pending.get(0).getConsumerName()).isEqualTo("my-consumer"); assertThat(pending.get(0).getGroupName()).isEqualTo("my-group"); assertThat(pending.get(0).getTotalDeliveryCount()).isOne(); - assertThat(pending.get(0).getStringId()).isNotNull(); - + assertThat(pending.get(0).getIdAsString()).isNotNull(); } @Test // DATAREDIS-1084 @@ -3183,7 +3182,7 @@ public abstract class AbstractConnectionIntegrationTests { assertThat(pending.get(0).getConsumerName()).isEqualTo("my-consumer"); assertThat(pending.get(0).getGroupName()).isEqualTo("my-group"); assertThat(pending.get(0).getTotalDeliveryCount()).isOne(); - assertThat(pending.get(0).getStringId()).isNotNull(); + assertThat(pending.get(0).getIdAsString()).isNotNull(); } @Test // DATAREDIS-1084 @@ -3234,7 +3233,8 @@ public abstract class AbstractConnectionIntegrationTests { StreamOffset.create(KEY_1, ReadOffset.lastConsumed()))); List> messages = (List>) getResults().get(2); - TimeUnit.MILLISECONDS.sleep(5); + + TimeUnit.MILLISECONDS.sleep(15); actual.add(connection.xClaim(KEY_1, "my-group", "my-consumer", XClaimOptions.minIdle(Duration.ofMillis(1)).ids(messages.get(0).getId()))); 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 82946f18d..327223d45 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 @@ -275,7 +275,7 @@ public class LettuceReactiveStreamCommandsTests extends LettuceReactiveCommandsT assertThat(it.get(0).getConsumerName()).isEqualTo("my-consumer"); assertThat(it.get(0).getGroupName()).isEqualTo("my-group"); assertThat(it.get(0).getTotalDeliveryCount()).isOne(); - assertThat(it.get(0).getStringId()).isNotNull(); + assertThat(it.get(0).getIdAsString()).isNotNull(); }).verifyComplete(); } @@ -300,7 +300,7 @@ public class LettuceReactiveStreamCommandsTests extends LettuceReactiveCommandsT assertThat(it.get(0).getConsumerName()).isEqualTo("my-consumer"); assertThat(it.get(0).getGroupName()).isEqualTo("my-group"); assertThat(it.get(0).getTotalDeliveryCount()).isOne(); - assertThat(it.get(0).getStringId()).isNotNull(); + assertThat(it.get(0).getIdAsString()).isNotNull(); }).verifyComplete(); }