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(); }