diff --git a/src/main/java/org/springframework/data/redis/connection/DefaultStringRedisConnection.java b/src/main/java/org/springframework/data/redis/connection/DefaultStringRedisConnection.java index 3af27aac6..fc235b271 100644 --- a/src/main/java/org/springframework/data/redis/connection/DefaultStringRedisConnection.java +++ b/src/main/java/org/springframework/data/redis/connection/DefaultStringRedisConnection.java @@ -37,6 +37,8 @@ import org.springframework.data.redis.connection.convert.SetConverter; import org.springframework.data.redis.connection.stream.ByteRecord; import org.springframework.data.redis.connection.stream.Consumer; import org.springframework.data.redis.connection.stream.MapRecord; +import org.springframework.data.redis.connection.stream.PendingMessages; +import org.springframework.data.redis.connection.stream.PendingMessagesSummary; import org.springframework.data.redis.connection.stream.ReadOffset; import org.springframework.data.redis.connection.stream.RecordId; import org.springframework.data.redis.connection.stream.StreamOffset; @@ -3694,6 +3696,44 @@ public class DefaultStringRedisConnection implements StringRedisConnection, Deco return convertAndReturn(delegate.xLen(serialize(key)), identityConverter); } + /* + * (non-Javadoc) + * @see org.springframework.data.redis.connection.StringRedisConnection#xPending(java.lang.String, java.lang.String) + */ + @Override + public PendingMessagesSummary xPending(String key, String groupName) { + return convertAndReturn(delegate.xPending(serialize(key), groupName), identityConverter); + } + + /* + * (non-Javadoc) + * @see org.springframework.data.redis.connection.StringRedisConnection#xPending(java.lang.String, java.lang.String, java.lang.String, org.springframework.data.domain.Range, java.lang.Long) + */ + @Override + public PendingMessages xPending(String key, String groupName, String consumer, + org.springframework.data.domain.Range range, Long count) { + return convertAndReturn(delegate.xPending(serialize(key), groupName, consumer, range, count), identityConverter); + } + + /* + * (non-Javadoc) + * @see org.springframework.data.redis.connection.StringRedisConnection#xPending(java.lang.String, java.lang.String, org.springframework.data.domain.Range, java.lang.Long) + */ + @Override + public PendingMessages xPending(String key, String groupName, org.springframework.data.domain.Range range, + Long count) { + return convertAndReturn(delegate.xPending(serialize(key), groupName, range, count), identityConverter); + } + + /* + * (non-Javadoc) + * @see org.springframework.data.redis.connection.StringRedisConnection#xPending(java.lang.String, org.springframework.data.redis.connection.RedisStreamCommands.XPendingOptions) + */ + @Override + public PendingMessages xPending(String key, String groupName, XPendingOptions options) { + return convertAndReturn(delegate.xPending(serialize(key), groupName, options), identityConverter); + } + /* * (non-Javadoc) * @see org.springframework.data.redis.connection.StringRedisConnection#xRange(java.lang.String, org.springframework.data.domain.Range, org.springframework.data.redis.connection.RedisZSetCommands.Limit) @@ -3719,7 +3759,7 @@ public class DefaultStringRedisConnection implements StringRedisConnection, Deco */ @Override public List xReadGroupAsString(Consumer consumer, StreamReadOptions readOptions, - StreamOffset... streams) { + StreamOffset... streams) { return convertAndReturn(delegate.xReadGroup(consumer, readOptions, serialize(streams)), listByteMapRecordToStringMapRecordConverter); @@ -3808,6 +3848,24 @@ public class DefaultStringRedisConnection implements StringRedisConnection, Deco return delegate.xLen(key); } + /* + * (non-Javadoc) + * @see org.springframework.data.redis.connection.RedisStreamCommands#xPending(byte[], java.lang.String) + */ + @Override + public PendingMessagesSummary xPending(byte[] key, String groupName) { + return delegate.xPending(key, groupName); + } + + /* + * (non-Javadoc) + * @see org.springframework.data.redis.connection.RedisStreamCommands#xPending(byte[], java.lang.String) + */ + @Override + public PendingMessages xPending(byte[] key, String groupName, XPendingOptions options) { + return delegate.xPending(key, groupName, options); + } + /* * (non-Javadoc) * @see org.springframework.data.redis.connection.RedisStreamCommands#xRange(byte[], org.springframework.data.domain.Range, org.springframework.data.redis.connection.RedisZSetCommands.Limit) @@ -3832,7 +3890,7 @@ public class DefaultStringRedisConnection implements StringRedisConnection, Deco */ @Override public List xReadGroup(Consumer consumer, StreamReadOptions readOptions, - StreamOffset... streams) { + StreamOffset... streams) { return delegate.xReadGroup(consumer, readOptions, streams); } diff --git a/src/main/java/org/springframework/data/redis/connection/DefaultedRedisConnection.java b/src/main/java/org/springframework/data/redis/connection/DefaultedRedisConnection.java index 2b7dbe589..5ebef5397 100644 --- a/src/main/java/org/springframework/data/redis/connection/DefaultedRedisConnection.java +++ b/src/main/java/org/springframework/data/redis/connection/DefaultedRedisConnection.java @@ -31,6 +31,8 @@ import org.springframework.data.geo.Point; import org.springframework.data.redis.connection.stream.ByteRecord; import org.springframework.data.redis.connection.stream.Consumer; import org.springframework.data.redis.connection.stream.MapRecord; +import org.springframework.data.redis.connection.stream.PendingMessages; +import org.springframework.data.redis.connection.stream.PendingMessagesSummary; import org.springframework.data.redis.connection.stream.ReadOffset; import org.springframework.data.redis.connection.stream.RecordId; import org.springframework.data.redis.connection.stream.StreamOffset; @@ -486,6 +488,20 @@ public interface DefaultedRedisConnection extends RedisConnection { return streamCommands().xLen(key); } + /** @deprecated in favor of {@link RedisConnection#streamCommands()}}. */ + @Override + @Deprecated + default PendingMessagesSummary xPending(byte[] key, String groupName) { + return streamCommands().xPending(key, groupName); + } + + /** @deprecated in favor of {@link RedisConnection#streamCommands()}}. */ + @Override + @Deprecated + default PendingMessages xPending(byte[] key, String groupName, XPendingOptions options) { + return streamCommands().xPending(key, groupName, options); + } + /** @deprecated in favor of {@link RedisConnection#streamCommands()}}. */ @Override @Deprecated @@ -525,7 +541,7 @@ public interface DefaultedRedisConnection extends RedisConnection { @Override @Deprecated default List xReadGroup(Consumer consumer, StreamReadOptions readOptions, - StreamOffset... streams) { + StreamOffset... streams) { return streamCommands().xReadGroup(consumer, readOptions, streams); } 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 7575ba2fc..b8b1cb531 100644 --- a/src/main/java/org/springframework/data/redis/connection/ReactiveStreamCommands.java +++ b/src/main/java/org/springframework/data/redis/connection/ReactiveStreamCommands.java @@ -30,9 +30,13 @@ import org.springframework.data.domain.Range; import org.springframework.data.redis.connection.ReactiveRedisConnection.CommandResponse; import org.springframework.data.redis.connection.ReactiveRedisConnection.KeyCommand; import org.springframework.data.redis.connection.ReactiveRedisConnection.NumericResponse; +import org.springframework.data.redis.connection.RedisStreamCommands.XPendingOptions; import org.springframework.data.redis.connection.RedisZSetCommands.Limit; import org.springframework.data.redis.connection.stream.ByteBufferRecord; import org.springframework.data.redis.connection.stream.Consumer; +import org.springframework.data.redis.connection.stream.PendingMessage; +import org.springframework.data.redis.connection.stream.PendingMessages; +import org.springframework.data.redis.connection.stream.PendingMessagesSummary; import org.springframework.data.redis.connection.stream.ReadOffset; import org.springframework.data.redis.connection.stream.RecordId; import org.springframework.data.redis.connection.stream.StreamOffset; @@ -40,6 +44,7 @@ import org.springframework.data.redis.connection.stream.StreamReadOptions; import org.springframework.data.redis.connection.stream.StreamRecords; import org.springframework.lang.Nullable; import org.springframework.util.Assert; +import org.springframework.util.StringUtils; /** * Stream-specific Redis commands executed using reactive infrastructure. @@ -410,6 +415,225 @@ public interface ReactiveStreamCommands { */ Flux> xLen(Publisher commands); + /** + * 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 {@link Mono} emitting a summary of pending messages within the given {@literal consumer group}. + * @see Redis Documentation: xpending + * @since 2.3 + */ + default Mono xPending(ByteBuffer key, String groupName) { + + Assert.notNull(key, "Key must not be null!"); + Assert.notNull(groupName, "GroupName must not be null!"); + + return xPendingSummary(Mono.just(new PendingRecordsCommand(key, groupName, null, Range.unbounded(), null))).next() + .map(CommandResponse::getOutput); + } + + /** + * Obtain the {@link PendingMessagesSummary} for a given {@literal consumer group}. + * + * @param commands must not be {@literal null}.. + * @return {@link Flux} emitting a summary of pending messages within the given {@literal consumer group} one by one. + * @see Redis Documentation: xpending + * @since 2.3 + */ + Flux> xPendingSummary( + Publisher commands); + + /** + * 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 {@link Mono} emitting pending messages for the given {@link Consumer}. + * @see Redis Documentation: xpending + * @since 2.3 + */ + @Nullable + default Mono xPending(ByteBuffer key, Consumer consumer) { + return xPending(key, consumer.getGroup(), consumer.getName()); + } + + /** + * Obtained detailed information about all pending messages for a given {@literal consumer}. + * + * @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}. + * @param consumerName the consumer to fetch {@link PendingMessages} for. Must not be {@literal null}. + * @return {@link Mono} emitting pending messages for the given {@link Consumer}. + * @see Redis Documentation: xpending + * @since 2.3 + */ + @Nullable + default Mono xPending(ByteBuffer key, String groupName, String consumerName) { + return xPending(Mono.just(new PendingRecordsCommand(key, groupName, consumerName, Range.unbounded(), null))).next() + .map(CommandResponse::getOutput); + } + + /** + * Obtain detailed information about pending {@link PendingMessage messages} for a given {@link Range} within a + * {@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}. + * @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}. + * @return {@link Mono} emitting pending messages for the given {@literal consumer group}. transaction. + * @see Redis Documentation: xpending + * @since 2.3 + */ + default Mono xPending(ByteBuffer key, String groupName, Range range, Long count) { + return xPending(Mono.just(new PendingRecordsCommand(key, groupName, null, range, count))).next() + .map(CommandResponse::getOutput); + } + + /** + * Obtain detailed information about pending {@link PendingMessage messages} for a given {@link Range} and + * {@link Consumer} within a {@literal consumer group}. + * + * @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}. + * @return {@link Mono} emitting pending messages for the given {@link Consumer}. + * @see Redis Documentation: xpending + * @since 2.3 + */ + default Mono xPending(ByteBuffer key, Consumer consumer, Range range, Long count) { + return xPending(key, consumer.getGroup(), consumer.getName(), range, count); + } + + /** + * Obtain detailed information about pending {@link PendingMessage messages} for a given {@link Range} and + * {@literal consumer} within a {@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}. + * @param consumerName the name of the {@literal 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}. + * @return {@link Mono} emitting pending messages for the given {@literal consumer} in given + * {@literal consumer group}. + * @see Redis Documentation: xpending + * @since 2.3 + */ + default Mono xPending(ByteBuffer key, String groupName, String consumerName, Range range, + Long count) { + return xPending(Mono.just(new PendingRecordsCommand(key, groupName, consumerName, range, count))).next() + .map(CommandResponse::getOutput); + } + + /** + * Obtain detailed information about pending {@link PendingMessage messages} applying given {@link XPendingOptions + * options}. + * + * @param commands must not be {@literal null}. + * @return {@link Flux} emitting pending messages matching given criteria. + * @see Redis Documentation: xpending + * @since 2.3 + */ + Flux> xPending(Publisher commands); + + /** + * Value Object holding parameters for obtaining pending messages. + * + * @author Christoph Strobl + * @since 2.3 + */ + class PendingRecordsCommand extends KeyCommand { + + private final String groupName; + private final @Nullable String consumerName; + private final Range range; + private final @Nullable Long count; + + private PendingRecordsCommand(ByteBuffer key, String groupName, @Nullable String consumerName, Range range, + @Nullable Long count) { + + super(key); + + this.groupName = groupName; + this.consumerName = consumerName; + this.range = range; + this.count = count; + } + + /** + * 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}. + */ + static PendingRecordsCommand pending(ByteBuffer key, String groupName) { + return new PendingRecordsCommand(key, groupName, null, Range.unbounded(), null); + } + + /** + * Create new {@link PendingRecordsCommand} with given {@link Range} and limit. + * + * @return new instance of {@link XPendingOptions}. + */ + public PendingRecordsCommand range(Range range, Long count) { + return new PendingRecordsCommand(getKey(), groupName, consumerName, range, count); + } + + /** + * Append given consumer. + * + * @param consumerName must not be {@literal null}. + * @return new instance of {@link PendingRecordsCommand}. + */ + public PendingRecordsCommand consumer(String consumerName) { + return new PendingRecordsCommand(getKey(), groupName, consumerName, range, count); + } + + public String getGroupName() { + return groupName; + } + + /** + * @return can be {@literal null}. + */ + @Nullable + public String getConsumerName() { + return consumerName; + } + + /** + * @return never {@literal null}. + */ + public Range getRange() { + return range; + } + + /** + * @return can be {@literal null}. + */ + @Nullable + public Long getCount() { + return count; + } + + /** + * @return {@literal true} if a consumer name is present. + */ + public boolean hasConsumer() { + return StringUtils.hasText(consumerName); + } + + /** + * @return {@literal true} count is set. + */ + public boolean isLimited() { + return count != null && count > -1; + } + } + /** * {@code XRANGE}/{@code XREVRANGE} command parameters. * 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 a740a6e50..e775754cf 100644 --- a/src/main/java/org/springframework/data/redis/connection/RedisStreamCommands.java +++ b/src/main/java/org/springframework/data/redis/connection/RedisStreamCommands.java @@ -23,6 +23,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.StringUtils; /** * Stream-specific Redis commands. @@ -162,6 +163,212 @@ public interface RedisStreamCommands { @Nullable Long xLen(byte[] key); + /** + * 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 + * pipeline / transaction. + * @see Redis Documentation: xpending + * @since 2.3 + */ + @Nullable + PendingMessagesSummary xPending(byte[] key, String groupName); + + /** + * 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. + * @see Redis Documentation: xpending + * @since 2.3 + */ + @Nullable + default PendingMessages xPending(byte[] key, Consumer consumer) { + return xPending(key, consumer.getGroup(), consumer.getName()); + } + + /** + * Obtained detailed information about all pending messages for a given {@literal consumer}. + * + * @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}. + * @param consumerName 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. + * @see Redis Documentation: xpending + * @since 2.3 + */ + @Nullable + default PendingMessages xPending(byte[] key, String groupName, String consumerName) { + return xPending(key, groupName, XPendingOptions.unbounded().consumer(consumerName)); + } + + /** + * Obtain detailed information about pending {@link PendingMessage messages} for a given {@link Range} within a + * {@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}. + * @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}. + * @return pending messages for the given {@literal consumer group} or {@literal null} when used in pipeline / + * transaction. + * @see Redis Documentation: xpending + * @since 2.3 + */ + @Nullable + default PendingMessages xPending(byte[] key, String groupName, Range range, Long count) { + return xPending(key, groupName, XPendingOptions.range(range, count)); + } + + /** + * Obtain detailed information about pending {@link PendingMessage messages} for a given {@link Range} and + * {@link Consumer} within a {@literal consumer group}. + * + * @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}. + * @return pending messages for the given {@link Consumer} or {@literal null} when used in pipeline / transaction. + * @see Redis Documentation: xpending + * @since 2.3 + */ + @Nullable + default PendingMessages xPending(byte[] key, Consumer consumer, Range range, Long count) { + return xPending(key, consumer.getGroup(), consumer.getName(), range, count); + } + + /** + * Obtain detailed information about pending {@link PendingMessage messages} for a given {@link Range} and + * {@literal consumer} within a {@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}. + * @param consumerName the name of the {@literal 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}. + * @return pending messages for the given {@literal consumer} in given {@literal consumer group} or {@literal null} + * when used in pipeline / transaction. + * @see Redis Documentation: xpending + * @since 2.3 + */ + @Nullable + default PendingMessages xPending(byte[] key, String groupName, String consumerName, Range range, Long count) { + return xPending(key, groupName, XPendingOptions.range(range, count).consumer(consumerName)); + } + + /** + * Obtain detailed information about pending {@link PendingMessage messages} applying given {@link XPendingOptions + * options}. + * + * @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}. + * @param options the options containing {@literal range}, {@literal consumer} and {@literal count}. Must not be + * {@literal null}. + * @return pending messages matching given criteria or {@literal null} when used in pipeline / transaction. + * @see Redis Documentation: xpending + * @since 2.3 + */ + @Nullable + PendingMessages xPending(byte[] key, String groupName, XPendingOptions options); + + /** + * Value Object holding parameters for obtaining pending messages. + * + * @author Christoph Strobl + * @since 2.3 + */ + class XPendingOptions { + + private final @Nullable String consumerName; + private final Range range; + private final @Nullable Long count; + + private XPendingOptions(@Nullable String consumerName, Range range, @Nullable Long count) { + + this.range = range; + this.count = count; + this.consumerName = consumerName; + } + + /** + * Create new {@link XPendingOptions} with an unbounded {@link Range} ({@literal - +}). + * + * @return new instance of {@link XPendingOptions}. + */ + public static XPendingOptions unbounded() { + return new XPendingOptions(null, Range.unbounded(), null); + } + + /** + * Create new {@link XPendingOptions} with an unbounded {@link Range} ({@literal - +}). + * + * @param count the max number of messages to return. Must not be {@literal null}. + * @return new instance of {@link XPendingOptions}. + */ + public static XPendingOptions unbounded(Long count) { + return new XPendingOptions(null, Range.unbounded(), 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) { + return new XPendingOptions(null, range, count); + } + + /** + * Append given consumer. + * + * @param consumerName must not be {@literal null}. + * @return new instance of {@link XPendingOptions}. + */ + public XPendingOptions consumer(String consumerName) { + return new XPendingOptions(consumerName, range, count); + } + + /** + * @return never {@literal null}. + */ + public Range getRange() { + return range; + } + + /** + * @return can be {@literal null}. + */ + @Nullable + public Long getCount() { + return count; + } + + /** + * @return can be {@literal null}. + */ + @Nullable + public String getConsumerName() { + return consumerName; + } + + /** + * @return {@literal true} if a consumer name is present. + */ + public boolean hasConsumer() { + return StringUtils.hasText(consumerName); + } + + /** + * @return {@literal true} count is set. + */ + public boolean isLimited() { + return count != null && count > -1; + } + } + /** * Retrieve all {@link ByteRecord records} within a specific {@link Range} from the stream stored at {@literal key}. *
diff --git a/src/main/java/org/springframework/data/redis/connection/StringRedisConnection.java b/src/main/java/org/springframework/data/redis/connection/StringRedisConnection.java index 70f5cc2ca..9d6fdcfaf 100644 --- a/src/main/java/org/springframework/data/redis/connection/StringRedisConnection.java +++ b/src/main/java/org/springframework/data/redis/connection/StringRedisConnection.java @@ -23,12 +23,16 @@ import java.util.Map; import java.util.Set; import java.util.concurrent.TimeUnit; +import org.springframework.data.domain.Range; import org.springframework.data.geo.Circle; import org.springframework.data.geo.Distance; import org.springframework.data.geo.GeoResults; import org.springframework.data.geo.Metric; import org.springframework.data.geo.Point; import org.springframework.data.redis.connection.stream.Consumer; +import org.springframework.data.redis.connection.stream.PendingMessage; +import org.springframework.data.redis.connection.stream.PendingMessages; +import org.springframework.data.redis.connection.stream.PendingMessagesSummary; import org.springframework.data.redis.connection.stream.ReadOffset; import org.springframework.data.redis.connection.stream.RecordId; import org.springframework.data.redis.connection.stream.StreamOffset; @@ -2071,6 +2075,68 @@ public interface StringRedisConnection extends RedisConnection { @Nullable Long xLen(String key); + /** + * 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 + * pipeline / transaction. + * @see Redis Documentation: xpending + * @since 2.3 + */ + @Nullable + PendingMessagesSummary xPending(String key, String groupName); + + /** + * Obtain detailed information about pending {@link PendingMessage messages} for a given {@link org.springframework.data.domain.Range} within a + * {@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}. + * @param consumerName the name of the {@literal 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}. + * @return pending messages for the given {@literal consumer group} or {@literal null} when used in pipeline / + * transaction. + * @see Redis Documentation: xpending + * @since 2.3 + */ + @Nullable + PendingMessages xPending(String key, String groupName, String consumerName, org.springframework.data.domain.Range range, Long count); + + /** + * Obtain detailed information about pending {@link PendingMessage messages} for a given {@link org.springframework.data.domain.Range} within a + * {@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}. + * @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}. + * @return pending messages for the given {@literal consumer group} or {@literal null} when used in pipeline / + * transaction. + * @see Redis Documentation: xpending + * @since 2.3 + */ + @Nullable + PendingMessages xPending(String key, String groupName, org.springframework.data.domain.Range range, Long count); + + + /** + * Obtain detailed information about pending {@link PendingMessage messages} applying given {@link XPendingOptions + * options}. + * + * @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}. + * @param options the options containing {@literal range}, {@literal consumer} and {@literal count}. Must not be + * {@literal null}. + * @return pending messages matching given criteria or {@literal null} when used in pipeline / transaction. + * @see Redis Documentation: xpending + * @since 2.3 + */ + @Nullable + PendingMessages xPending(String key, String groupName, XPendingOptions options); + /** * Read records from a stream within a specific {@link Range}. * 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 e2faf5a22..e94129f12 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 @@ -18,11 +18,16 @@ package org.springframework.data.redis.connection.lettuce; import io.lettuce.core.*; import io.lettuce.core.cluster.models.partitions.Partitions; import io.lettuce.core.cluster.models.partitions.RedisClusterNode.NodeFlag; +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.nio.charset.StandardCharsets; +import java.time.Duration; import java.util.*; import java.util.concurrent.TimeUnit; +import java.util.function.BiFunction; import java.util.stream.Collectors; import org.springframework.core.convert.converter.Converter; @@ -64,6 +69,9 @@ import org.springframework.data.redis.connection.convert.Converters; import org.springframework.data.redis.connection.convert.ListConverter; import org.springframework.data.redis.connection.convert.LongToBooleanConverter; import org.springframework.data.redis.connection.convert.StringToRedisClientInfoConverter; +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.core.ScanOptions; import org.springframework.data.redis.core.types.Expiration; import org.springframework.data.redis.core.types.RedisClientInfo; @@ -71,6 +79,7 @@ import org.springframework.data.redis.util.ByteUtils; import org.springframework.lang.Nullable; import org.springframework.util.Assert; import org.springframework.util.CollectionUtils; +import org.springframework.util.NumberUtils; import org.springframework.util.ObjectUtils; import org.springframework.util.StringUtils; @@ -108,6 +117,8 @@ 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; @@ -308,6 +319,70 @@ abstract public class LettuceConverters extends Converters { KEY_VALUE_LIST_UNWRAPPER = new ListConverter<>(KEY_VALUE_UNWRAPPER); 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) { @@ -1042,6 +1117,32 @@ 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 dfeec166b..13213ae05 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 @@ -22,6 +22,7 @@ import io.lettuce.core.cluster.api.reactive.RedisClusterReactiveCommands; import reactor.core.publisher.Flux; import java.nio.ByteBuffer; +import java.util.ArrayList; import java.util.Collection; import java.util.List; import java.util.function.Function; @@ -34,6 +35,8 @@ import org.springframework.data.redis.connection.ReactiveStreamCommands; import org.springframework.data.redis.connection.ReactiveStreamCommands.GroupCommand.GroupCommandAction; import org.springframework.data.redis.connection.stream.ByteBufferRecord; import org.springframework.data.redis.connection.stream.Consumer; +import org.springframework.data.redis.connection.stream.PendingMessages; +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; @@ -172,6 +175,67 @@ class LettuceReactiveStreamCommands implements ReactiveStreamCommands { })); } + /* + * (non-Javadoc) + * @see org.springframework.data.redis.connection.ReactiveStreamCommands#xPendingSummary(org.reactivestreams.Publisher) + */ + @Override + public Flux> xPendingSummary( + Publisher commands) { + + return connection.execute(cmd -> Flux.from(commands).concatMap(command -> { + + Assert.notNull(command.getKey(), "Key must not be null!"); + return cmd.xpending(command.getKey(), ByteUtils.getByteBuffer(command.getGroupName())).collectList().map(it -> { + + // begin + // {* hacking *} + // while (https://github.com/lettuce-io/lettuce-core/issues/1229 != resolved) begin + + ArrayList target = new ArrayList<>(it); + if (target.size() == 2 && target.get(1) instanceof List) { + target.add(1, null); + target.add(1, null); + } + while (target.size() < 4) { + target.add(null); + } + + // end. + // end. + + return LettuceConverters.toPendingMessagesInfo(command.getGroupName(), target); + }).map(value -> new CommandResponse(command, value)); + })); + } + + /* + * (non-Javadoc) + * @see org.springframework.data.redis.connection.ReactiveStreamCommands#xPending(org.reactivestreams.Publisher) + */ + @Override + public Flux> xPending( + Publisher commands) { + return connection.execute(cmd -> Flux.from(commands).concatMap(command -> { + + Assert.notNull(command.getKey(), "Key must not be null!"); + + 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); + + Flux publisher = command.hasConsumer() ? cmd.xpending(command.getKey(), + io.lettuce.core.Consumer.from(groupName, ByteUtils.getByteBuffer(command.getConsumerName())), range, limit) + : cmd.xpending(command.getKey(), groupName, range, limit); + + return publisher.collectList().map(it -> { + + return LettuceConverters.toPendingMessages(command.getGroupName(), command.getRange(), it); + }).map(value -> new CommandResponse(command, value)); + })); + } + /* * (non-Javadoc) * @see org.springframework.data.redis.connection.ReactiveStreamCommands#xRange(org.reactivestreams.Publisher) diff --git a/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceStreamCommands.java b/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceStreamCommands.java index d30f12c68..9f3d13c22 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 @@ -28,11 +28,13 @@ import java.util.function.Function; import org.springframework.dao.DataAccessException; import org.springframework.data.domain.Range; -import org.springframework.data.redis.connection.stream.ByteRecord; -import org.springframework.data.redis.connection.stream.MapRecord; import org.springframework.data.redis.connection.RedisStreamCommands; import org.springframework.data.redis.connection.RedisZSetCommands.Limit; +import org.springframework.data.redis.connection.stream.ByteRecord; import org.springframework.data.redis.connection.stream.Consumer; +import org.springframework.data.redis.connection.stream.MapRecord; +import org.springframework.data.redis.connection.stream.PendingMessages; +import org.springframework.data.redis.connection.stream.PendingMessagesSummary; import org.springframework.data.redis.connection.stream.ReadOffset; import org.springframework.data.redis.connection.stream.RecordId; import org.springframework.data.redis.connection.stream.StreamOffset; @@ -242,6 +244,78 @@ class LettuceStreamCommands implements RedisStreamCommands { } } + /* + * (non-Javadoc) + * @see org.springframework.data.redis.connection.RedisStreamCommands#xPending(byte[], java.lang.String) + */ + @Override + public PendingMessagesSummary xPending(byte[] key, String groupName) { + + byte[] group = LettuceConverters.toBytes(groupName); + try { + if (isPipelined()) { + pipeline(connection.newLettuceResult(getAsyncConnection().xpending(key, group), + it -> LettuceConverters.toPendingMessagesInfo(groupName, it))); + return null; + } + if (isQueueing()) { + transaction(connection.newLettuceResult(getAsyncConnection().xpending(key, group), + it -> LettuceConverters.toPendingMessagesInfo(groupName, it))); + return null; + } + return LettuceConverters.toPendingMessagesInfo(groupName, getConnection().xpending(key, group)); + } catch (Exception ex) { + throw convertLettuceAccessException(ex); + } + } + + /* + * (non-Javadoc) + * @see org.springframework.data.redis.connection.RedisStreamCommands#xPending(byte[], java.lang.String, org.springframework.data.redis.connection.RedisStreamCommands.XPendingOptions) + */ + @Override + public PendingMessages xPending(byte[] key, String groupName, XPendingOptions options) { + + 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); + + 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 (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))); + 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))); + return null; + } + return LettuceConverters.toPendingMessages(groupName, options.getRange(), getConnection().xpending(key, + io.lettuce.core.Consumer.from(group, LettuceConverters.toBytes(options.getConsumerName())), range, limit)); + } + } catch (Exception ex) { + throw convertLettuceAccessException(ex); + } + } + /* * (non-Javadoc) * @see org.springframework.data.redis.connection.RedisStreamCommands#xRange(byte[], org.springframework.data.domain.Range, org.springframework.data.redis.connection.RedisZSetCommands.Limit) diff --git a/src/main/java/org/springframework/data/redis/connection/lettuce/RangeConverter.java b/src/main/java/org/springframework/data/redis/connection/lettuce/RangeConverter.java index 91341da94..bfaa7a10c 100644 --- a/src/main/java/org/springframework/data/redis/connection/lettuce/RangeConverter.java +++ b/src/main/java/org/springframework/data/redis/connection/lettuce/RangeConverter.java @@ -20,9 +20,11 @@ import io.lettuce.core.Range.Boundary; import io.lettuce.core.codec.StringCodec; import java.nio.ByteBuffer; +import java.util.function.BiFunction; import java.util.function.Function; -import org.springframework.core.convert.converter.Converter; +import org.springframework.data.domain.Range.Bound; +import org.springframework.lang.Nullable; import org.springframework.util.ObjectUtils; import org.springframework.util.StringUtils; @@ -36,31 +38,64 @@ class RangeConverter { return toRange(range, StringCodec.UTF8::encodeValue); } + /** + * @param range the source {@link org.springframework.data.domain.Range} to convert. + * @param lowerDefault the lower default to use if {@link org.springframework.data.domain.Range#getLowerBound()} is + * not {@link Bound#isBounded() bounded}. + * @param upperDefault the upper default to use if {@link org.springframework.data.domain.Range#getUpperBound()} is + * not {@link Bound#isBounded() bounded}. + * @param + * @return new instance of {@link Range}. + * @since 2.3 + */ + static Range toRangeWithDefault(org.springframework.data.domain.Range range, @Nullable T lowerDefault, + @Nullable T upperDefault) { + return toRangeWithDefault(range, lowerDefault, upperDefault, StringCodec.UTF8::encodeValue); + } + static Range toRange(org.springframework.data.domain.Range range, Function stringEncoder) { - return Range.from(lowerBoundArgOf(range, stringEncoder), upperBoundArgOf(range, stringEncoder)); + return toRangeWithDefault(range, null, null, stringEncoder); + } + + /** + * @param range the source {@link org.springframework.data.domain.Range} to convert. + * @param lowerDefault the lower default to use if {@link org.springframework.data.domain.Range#getLowerBound()} is + * not {@link Bound#isBounded() bounded}. + * @param upperDefault the upper default to use if {@link org.springframework.data.domain.Range#getUpperBound()} is + * not {@link Bound#isBounded() bounded}. + * @param stringEncoder the encoder to use. + * @param + * @return new instance of {@link Range}. + * @since 2.3 + */ + static Range toRangeWithDefault(org.springframework.data.domain.Range range, @Nullable T lowerDefault, + @Nullable T upperDefault, Function stringEncoder) { + + return Range.from(lowerBoundArgOf(range, lowerDefault, stringEncoder), + upperBoundArgOf(range, upperDefault, stringEncoder)); } @SuppressWarnings("unchecked") private static Boundary lowerBoundArgOf(org.springframework.data.domain.Range range, - Function stringEncoder) { - return (Boundary) rangeToBoundArgumentConverter(false, stringEncoder).convert(range); + @Nullable T lowerDefault, Function stringEncoder) { + return (Boundary) rangeToBoundArgumentConverter(false, stringEncoder).apply(range, lowerDefault); } @SuppressWarnings("unchecked") private static Boundary upperBoundArgOf(org.springframework.data.domain.Range range, - Function stringEncoder) { - return (Boundary) rangeToBoundArgumentConverter(true, stringEncoder).convert(range); + @Nullable T upperDefault, Function stringEncoder) { + return (Boundary) rangeToBoundArgumentConverter(true, stringEncoder).apply(range, upperDefault); } - private static Converter, Boundary> rangeToBoundArgumentConverter( + private static BiFunction> rangeToBoundArgumentConverter( boolean upper, Function stringEncoder) { - return (source) -> { + return (source, defaultValue) -> { Boolean inclusive = upper ? source.getUpperBound().isInclusive() : source.getLowerBound().isInclusive(); - Object value = upper ? source.getUpperBound().getValue().orElse(null) - : source.getLowerBound().getValue().orElse(null); + Object value = upper ? source.getUpperBound().getValue().orElse(defaultValue) + : source.getLowerBound().getValue().orElse(defaultValue); if (value instanceof Number) { return inclusive ? Boundary.including((Number) value) : Boundary.excluding((Number) value); 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 new file mode 100644 index 000000000..ef9258876 --- /dev/null +++ b/src/main/java/org/springframework/data/redis/connection/stream/PendingMessage.java @@ -0,0 +1,117 @@ +/* + * Copyright 2020 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * 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 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.springframework.data.redis.connection.stream; + +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 + */ +public class PendingMessage { + + private final RecordId id; + private final Consumer consumer; + private final Duration elapsedTimeSinceLastDelivery; + private final Long totalDeliveryCount; + + public PendingMessage(RecordId id, Consumer consumer, Duration elapsedTimeSinceLastDelivery, + Long totalDeliveryCount) { + + this.id = id; + this.consumer = consumer; + this.elapsedTimeSinceLastDelivery = elapsedTimeSinceLastDelivery; + this.totalDeliveryCount = totalDeliveryCount; + } + + /** + * @return the message id. + */ + public RecordId getId() { + return id; + } + + /** + * @return the message id as {@link String}. + */ + public String getStringId() { + return id.getValue(); + } + + /** + * The {@link Consumer} to acknowledge the message. + * + * @return never {@literal null}. + */ + public Consumer getConsumer() { + return consumer; + } + + /** + * The {@literal consumer name} to acknowledge the message. + * + * @return never {@literal null}. + */ + public String getConsumerName() { + return consumer.getName(); + } + + /** + * Get the {@literal consumer group}. + * + * @return never {@literal null}. + */ + public String getGroupName() { + return consumer.getGroup(); + } + + /** + * Get the time elapsed since the messages last delivery to the {@link #getConsumer() consumer}. + * + * @return never {@literal null}. + */ + public Duration getElapsedTimeSinceLastDelivery() { + return elapsedTimeSinceLastDelivery; + } + + /** + * Get the milliseconds elapsed since the messages last delivery 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() { + return totalDeliveryCount; + } + + @Override + public String toString() { + return "PendingMessage{" + "id=" + id + ", consumer=" + consumer + ", elapsedTimeSinceLastDeliveryMS=" + + elapsedTimeSinceLastDelivery.toMillis() + ", totalDeliveryCount=" + 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 new file mode 100644 index 000000000..8fe21bc9a --- /dev/null +++ b/src/main/java/org/springframework/data/redis/connection/stream/PendingMessages.java @@ -0,0 +1,115 @@ +/* + * Copyright 2020 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * 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 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.springframework.data.redis.connection.stream; + +import java.util.Iterator; +import java.util.List; + +import org.springframework.data.domain.Range; +import org.springframework.data.util.Streamable; + +/** + * 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. + * + * @author Christoph Strobl + * @since 2.3 + */ +public class PendingMessages implements Streamable { + + private final String groupName; + private final Range range; + private final List pendingMessages; + + public PendingMessages(String groupName, List pendingMessages) { + this(groupName, Range.unbounded(), pendingMessages); + } + + public PendingMessages(String groupName, Range range, List pendingMessages) { + + this.groupName = groupName; + this.range = range; + this.pendingMessages = pendingMessages; + } + + /** + * Adds the range to the current {@link PendingMessages}. + * + * @param range must not be {@literal null}. + * @return new instance of {@link PendingMessages}. + */ + public PendingMessages withinRange(Range range) { + return new PendingMessages(groupName, range, pendingMessages); + } + + /** + * The {@literal consumer group} name. + * + * @return never {@literal null}. + */ + public String getGroupName() { + return groupName; + } + + /** + * The {@link Range} pending messages have been loaded. + * + * @return never {@literal null}. + */ + public Range getRange() { + return range; + } + + /** + * @return {@literal true} if no messages pending within range. + */ + public boolean isEmpty() { + return pendingMessages.isEmpty(); + } + + /** + * @return the number of pending messages in range. + */ + public int size() { + return pendingMessages.size(); + } + + /** + * 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. + */ + public PendingMessage get(int index) { + return pendingMessages.get(index); + } + + /* + * (non-Javadoc) + * @see java.lang.Iterable#iterator() + */ + @Override + public Iterator iterator() { + return pendingMessages.iterator(); + } + + @Override + public String toString() { + return "PendingMessages{" + "groupName='" + groupName + '\'' + ", range=" + range + ", pendingMessages=" + + pendingMessages + '}'; + } +} 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 new file mode 100644 index 000000000..fa3314304 --- /dev/null +++ b/src/main/java/org/springframework/data/redis/connection/stream/PendingMessagesSummary.java @@ -0,0 +1,124 @@ +/* + * Copyright 2020 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * 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 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.springframework.data.redis.connection.stream; + +import java.util.Collections; +import java.util.Map; + +import org.springframework.data.domain.Range; + +/** + * 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 + */ +public class PendingMessagesSummary { + + private final String groupName; + private final Long totalPendingMessages; + private final Range idRange; + private final Map pendingMessagesPerConsumer; + + public PendingMessagesSummary(String groupName, Long totalPendingMessages, Range idRange, + Map pendingMessagesPerConsumer) { + + this.groupName = groupName; + this.totalPendingMessages = totalPendingMessages; + this.idRange = idRange; + this.pendingMessagesPerConsumer = pendingMessagesPerConsumer; + } + + /** + * Get the range between the smallest and greatest ID among the pending messages. + * + * @return never {@literal null}. + */ + public Range getIdRange() { + return idRange; + } + + /** + * Get the smallest ID among the pending messages. + * + * @return never {@literal null}. + */ + public RecordId minRecordId() { + return RecordId.of(minMessageId()); + } + + /** + * Get the greatest ID among the pending messages. + * + * @return never {@literal null}. + */ + public RecordId maxRecordId() { + return RecordId.of(maxMessageId()); + } + + /** + * Get the smallest ID as {@link String} among the pending messages. + * + * @return never {@literal null}. + */ + public String minMessageId() { + return idRange.getLowerBound().getValue().get(); + } + + /** + * Get the greatest ID as {@link String} among the pending messages. + * + * @return never {@literal null}. + */ + public String maxMessageId() { + return idRange.getUpperBound().getValue().get(); + } + + /** + * Get the number of total pending messages within the {@literal consumer group}. + * + * @return + */ + public Long getTotalPendingMessages() { + return totalPendingMessages; + } + + /** + * @return the {@literal consumer group} name. + */ + public String getGroupName() { + return groupName; + } + + /** + * 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() { + return Collections.unmodifiableMap(pendingMessagesPerConsumer); + } + + @Override + public String toString() { + + return "PendingMessagesSummary{" + "groupName='" + groupName + '\'' + ", totalPendingMessages='" + + getTotalPendingMessages() + '\'' + ", minMessageId='" + minMessageId() + '\'' + ", maxMessageId='" + + maxMessageId() + '\'' + ", pendingMessagesPerConsumer=" + pendingMessagesPerConsumer + '}'; + } +} 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 ef843fad1..61699361c 100644 --- a/src/main/java/org/springframework/data/redis/core/DefaultReactiveStreamOperations.java +++ b/src/main/java/org/springframework/data/redis/core/DefaultReactiveStreamOperations.java @@ -15,6 +15,8 @@ */ package org.springframework.data.redis.core; +import org.springframework.data.redis.connection.stream.PendingMessages; +import org.springframework.data.redis.connection.stream.PendingMessagesSummary; import reactor.core.publisher.Flux; import reactor.core.publisher.Mono; @@ -183,6 +185,40 @@ class DefaultReactiveStreamOperations implements ReactiveStreamOperat return createMono(connection -> connection.xGroupDestroy(rawKey(key), group)); } + /* + * (non-Javadoc) + * @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) { + + ByteBuffer rawKey = rawKey(key); + return createMono(connection -> connection.xPending(rawKey, group, range, count)); + } + + /* + * (non-Javadoc) + * @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) { + + ByteBuffer rawKey = rawKey(key); + return createMono(connection -> connection.xPending(rawKey, consumer, range, count)); + } + + /* + * (non-Javadoc) + * @see org.springframework.data.redis.core.StreamOperations#pending(java.lang.Object, java.lang.String) + */ + @Override + public Mono pending(K key, String group) { + + ByteBuffer rawKey = rawKey(key); + return createMono(connection -> connection.xPending(rawKey, group)); + } + + /* * (non-Javadoc) * @see org.springframework.data.redis.core.ReactiveStreamOperations#size(java.lang.Object) 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 919375eda..0aa06372c 100644 --- a/src/main/java/org/springframework/data/redis/core/DefaultStreamOperations.java +++ b/src/main/java/org/springframework/data/redis/core/DefaultStreamOperations.java @@ -29,6 +29,8 @@ import org.springframework.data.redis.connection.RedisZSetCommands.Limit; import org.springframework.data.redis.connection.stream.ByteRecord; import org.springframework.data.redis.connection.stream.Consumer; import org.springframework.data.redis.connection.stream.MapRecord; +import org.springframework.data.redis.connection.stream.PendingMessages; +import org.springframework.data.redis.connection.stream.PendingMessagesSummary; import org.springframework.data.redis.connection.stream.ReadOffset; import org.springframework.data.redis.connection.stream.Record; import org.springframework.data.redis.connection.stream.RecordId; @@ -179,6 +181,39 @@ class DefaultStreamOperations extends AbstractOperations i return execute(connection -> connection.xGroupDestroy(rawKey, group), true); } + /* + * (non-Javadoc) + * @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) { + + byte[] rawKey = rawKey(key); + return execute(connection -> connection.xPending(rawKey, group, range, count), true); + } + + /* + * (non-Javadoc) + * @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) { + + byte[] rawKey = rawKey(key); + return execute(connection -> connection.xPending(rawKey, consumer, range, count), true); + } + + /* + * (non-Javadoc) + * @see org.springframework.data.redis.core.StreamOperations#pending(java.lang.Object, java.lang.String) + */ + @Override + public PendingMessagesSummary pending(K key, String group) { + + byte[] rawKey = rawKey(key); + return execute(connection -> connection.xPending(rawKey, group), true); + } + /* * (non-Javadoc) * @see org.springframework.data.redis.core.StreamOperations#size(java.lang.Object) @@ -279,7 +314,7 @@ class DefaultStreamOperations extends AbstractOperations i return hashValueSerializer() != null; } - @SuppressWarnings({"rawtypes", "unchecked"}) + @SuppressWarnings({ "rawtypes", "unchecked" }) private byte[] serialize(Object value, RedisSerializer serializer) { Object _value = value; @@ -303,7 +338,7 @@ class DefaultStreamOperations extends AbstractOperations i public final List> doInRedis(RedisConnection connection) { List raw = inRedis(connection); - if(raw == null) { + if (raw == null) { return Collections.emptyList(); } 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 ada984b96..21b7564ac 100644 --- a/src/main/java/org/springframework/data/redis/core/ReactiveStreamOperations.java +++ b/src/main/java/org/springframework/data/redis/core/ReactiveStreamOperations.java @@ -24,16 +24,9 @@ import java.util.Map; import org.reactivestreams.Publisher; import org.springframework.data.domain.Range; import org.springframework.data.redis.connection.RedisZSetCommands.Limit; -import org.springframework.data.redis.connection.stream.Consumer; -import org.springframework.data.redis.connection.stream.MapRecord; -import org.springframework.data.redis.connection.stream.ObjectRecord; -import org.springframework.data.redis.connection.stream.ReadOffset; -import org.springframework.data.redis.connection.stream.Record; -import org.springframework.data.redis.connection.stream.RecordId; -import org.springframework.data.redis.connection.stream.StreamOffset; -import org.springframework.data.redis.connection.stream.StreamReadOptions; -import org.springframework.data.redis.connection.stream.StreamRecords; +import org.springframework.data.redis.connection.stream.*; import org.springframework.data.redis.hash.HashMapper; +import org.springframework.lang.Nullable; import org.springframework.util.Assert; /** @@ -201,6 +194,61 @@ public interface ReactiveStreamOperations extends HashMapperProvider< */ Mono destroyGroup(K key, String group); + /** + * 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 group 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 + * pipeline / transaction. + * @see Redis Documentation: xpending + * @since 2.3 + */ + @Nullable + Mono pending(K key, String group); + + /** + * 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. + * @see Redis Documentation: xpending + * @since 2.3 + */ + default Mono pending(K key, Consumer consumer) { + return pending(key, consumer, Range.unbounded(), -1L); + } + + /** + * Obtain detailed information about pending {@link PendingMessage messages} for a given {@link Range} within a + * {@literal consumer group}. + * + * @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}. + * @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); + + /** + * Obtain detailed information about pending {@link PendingMessage messages} for a given {@link Range} and + * {@link Consumer} within a {@literal consumer group}. + * + * @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}. + * @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); + /** * 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 fce6473d1..d3070dd91 100644 --- a/src/main/java/org/springframework/data/redis/core/StreamOperations.java +++ b/src/main/java/org/springframework/data/redis/core/StreamOperations.java @@ -23,15 +23,7 @@ import java.util.Map; import org.springframework.data.domain.Range; import org.springframework.data.redis.connection.RedisZSetCommands.Limit; -import org.springframework.data.redis.connection.stream.Consumer; -import org.springframework.data.redis.connection.stream.MapRecord; -import org.springframework.data.redis.connection.stream.ObjectRecord; -import org.springframework.data.redis.connection.stream.ReadOffset; -import org.springframework.data.redis.connection.stream.Record; -import org.springframework.data.redis.connection.stream.RecordId; -import org.springframework.data.redis.connection.stream.StreamOffset; -import org.springframework.data.redis.connection.stream.StreamReadOptions; -import org.springframework.data.redis.connection.stream.StreamRecords; +import org.springframework.data.redis.connection.stream.*; import org.springframework.data.redis.hash.HashMapper; import org.springframework.lang.Nullable; import org.springframework.util.Assert; @@ -201,6 +193,61 @@ public interface StreamOperations extends HashMapperProvider @Nullable Boolean destroyGroup(K key, String group); + /** + * 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 group 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 + * pipeline / transaction. + * @see Redis Documentation: xpending + * @since 2.3 + */ + @Nullable + PendingMessagesSummary pending(K key, String group); + + /** + * 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. + * @see Redis Documentation: xpending + * @since 2.3 + */ + default PendingMessages pending(K key, Consumer consumer) { + return pending(key, consumer, Range.unbounded(), -1L); + } + + /** + * Obtain detailed information about pending {@link PendingMessage messages} for a given {@link Range} within a + * {@literal consumer group}. + * + * @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}. + * @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); + + /** + * Obtain detailed information about pending {@link PendingMessage messages} for a given {@link Range} and + * {@link Consumer} within a {@literal consumer group}. + * + * @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}. + * @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); + /** * Get the length of a stream. * diff --git a/src/test/java/org/springframework/data/redis/RedisTestProfileValueSource.java b/src/test/java/org/springframework/data/redis/RedisTestProfileValueSource.java index d91c552bc..13010faea 100644 --- a/src/test/java/org/springframework/data/redis/RedisTestProfileValueSource.java +++ b/src/test/java/org/springframework/data/redis/RedisTestProfileValueSource.java @@ -129,4 +129,18 @@ public class RedisTestProfileValueSource implements ProfileValueSource { } return INSTANCE.get(key) != null ? INSTANCE.get(key).equals(value) : value == null; } + + public static boolean atLeast(String key, String value) { + + if (INSTANCE == null) { + INSTANCE = new RedisTestProfileValueSource(); + } + + String current = INSTANCE.get(key); + if(current == null) { + return value == null; + } + + return org.springframework.data.util.Version.parse(current).isGreaterThanOrEqualTo(org.springframework.data.util.Version.parse(value)); + } } 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 e52d4c157..39dd728c7 100644 --- a/src/test/java/org/springframework/data/redis/connection/AbstractConnectionIntegrationTests.java +++ b/src/test/java/org/springframework/data/redis/connection/AbstractConnectionIntegrationTests.java @@ -42,7 +42,6 @@ import org.junit.Before; import org.junit.Ignore; import org.junit.Rule; import org.junit.Test; - import org.springframework.beans.factory.annotation.Autowired; import org.springframework.dao.DataAccessException; import org.springframework.data.domain.Range.Bound; @@ -68,6 +67,8 @@ import org.springframework.data.redis.connection.StringRedisConnection.StringTup import org.springframework.data.redis.connection.ValueEncoding.RedisValueEncoding; import org.springframework.data.redis.connection.stream.Consumer; import org.springframework.data.redis.connection.stream.MapRecord; +import org.springframework.data.redis.connection.stream.PendingMessages; +import org.springframework.data.redis.connection.stream.PendingMessagesSummary; import org.springframework.data.redis.connection.stream.ReadOffset; import org.springframework.data.redis.connection.stream.RecordId; import org.springframework.data.redis.connection.stream.StreamOffset; @@ -3095,6 +3096,131 @@ public abstract class AbstractConnectionIntegrationTests { assertThat(messages.get(1).getValue()).isEqualTo(Collections.singletonMap(KEY_2, VALUE_2)); } + @Test // DATAREDIS-1084 + @IfProfileValue(name = "redisVersion", value = "5.0") + @WithRedisDriver({ RedisDriver.LETTUCE }) + public void xPendingShouldLoadOverviewCorrectly() { + + actual.add(connection.xAdd(KEY_1, Collections.singletonMap(KEY_2, VALUE_2))); + actual.add(connection.xGroupCreate(KEY_1, ReadOffset.from("0"), "my-group")); + actual.add(connection.xReadGroupAsString(Consumer.from("my-group", "my-consumer"), + StreamOffset.create(KEY_1, ReadOffset.lastConsumed()))); + + actual.add(connection.xPending(KEY_1, "my-group")); + + List results = getResults(); + assertThat(results).hasSize(4); + PendingMessagesSummary info = (PendingMessagesSummary) results.get(3); + + assertThat(info.getGroupName()).isEqualTo("my-group"); + assertThat(info.getTotalPendingMessages()).isEqualTo(1L); + assertThat(info.getIdRange()).isNotNull(); + assertThat(info.getPendingMessagesPerConsumer()).hasSize(1).containsEntry("my-consumer", 1L); + } + + @Test // DATAREDIS-1084 + @IfProfileValue(name = "redisVersion", value = "5.0") + @WithRedisDriver({ RedisDriver.LETTUCE }) + public void xPendingShouldLoadEmptyOverviewCorrectly() { + + actual.add(connection.xAdd(KEY_1, Collections.singletonMap(KEY_2, VALUE_2))); + actual.add(connection.xGroupCreate(KEY_1, ReadOffset.from("0"), "my-group")); + actual.add(connection.xPending(KEY_1, "my-group")); + + List results = getResults(); + assertThat(results).hasSize(3); + PendingMessagesSummary info = (PendingMessagesSummary) results.get(2); + + assertThat(info.getGroupName()).isEqualTo("my-group"); + assertThat(info.getTotalPendingMessages()).isEqualTo(0L); + assertThat(info.getIdRange()).isNotNull(); + assertThat(info.getPendingMessagesPerConsumer()).isEmpty(); + } + + @Test // DATAREDIS-1084 + @IfProfileValue(name = "redisVersion", value = "5.0") + @WithRedisDriver({ RedisDriver.LETTUCE }) + public void xPendingShouldLoadPendingMessages() { + + actual.add(connection.xAdd(KEY_1, Collections.singletonMap(KEY_2, VALUE_2))); + actual.add(connection.xGroupCreate(KEY_1, ReadOffset.from("0"), "my-group")); + actual.add(connection.xReadGroupAsString(Consumer.from("my-group", "my-consumer"), + StreamOffset.create(KEY_1, ReadOffset.lastConsumed()))); + + actual.add(connection.xPending(KEY_1, "my-group", org.springframework.data.domain.Range.open("-", "+"), 10L)); + + List results = getResults(); + assertThat(results).hasSize(4); + PendingMessages pending = (PendingMessages) results.get(3); + + assertThat(pending.size()).isOne(); + 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(); + + } + + @Test // DATAREDIS-1084 + @IfProfileValue(name = "redisVersion", value = "5.0") + @WithRedisDriver({ RedisDriver.LETTUCE }) + public void xPendingShouldLoadPendingMessagesForConsumer() { + + actual.add(connection.xAdd(KEY_1, Collections.singletonMap(KEY_2, VALUE_2))); + actual.add(connection.xGroupCreate(KEY_1, ReadOffset.from("0"), "my-group")); + actual.add(connection.xReadGroupAsString(Consumer.from("my-group", "my-consumer"), + StreamOffset.create(KEY_1, ReadOffset.lastConsumed()))); + + actual.add(connection.xPending(KEY_1, "my-group", "my-consumer", org.springframework.data.domain.Range.open("-", "+"), 10L)); + + List results = getResults(); + assertThat(results).hasSize(4); + PendingMessages pending = (PendingMessages) results.get(3); + + assertThat(pending.size()).isOne(); + 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(); + } + + @Test // DATAREDIS-1084 + @IfProfileValue(name = "redisVersion", value = "5.0") + @WithRedisDriver({ RedisDriver.LETTUCE }) + public void xPendingShouldLoadPendingMessagesForNonExistingConsumer() { + + actual.add(connection.xAdd(KEY_1, Collections.singletonMap(KEY_2, VALUE_2))); + actual.add(connection.xGroupCreate(KEY_1, ReadOffset.from("0"), "my-group")); + actual.add(connection.xReadGroupAsString(Consumer.from("my-group", "my-consumer"), + StreamOffset.create(KEY_1, ReadOffset.lastConsumed()))); + + actual.add(connection.xPending(KEY_1, "my-group", "my-consumer-2", org.springframework.data.domain.Range.open("-", "+"), 10L)); + + List results = getResults(); + assertThat(results).hasSize(4); + PendingMessages pending = (PendingMessages) results.get(3); + + assertThat(pending.size()).isZero(); + } + + @Test // DATAREDIS-1084 + @IfProfileValue(name = "redisVersion", value = "5.0") + @WithRedisDriver({ RedisDriver.LETTUCE }) + public void xPendingShouldLoadEmptyPendingMessages() { + + actual.add(connection.xAdd(KEY_1, Collections.singletonMap(KEY_2, VALUE_2))); + actual.add(connection.xGroupCreate(KEY_1, ReadOffset.from("0"), "my-group")); + + actual.add(connection.xPending(KEY_1, "my-group", org.springframework.data.domain.Range.open("-", "+"), 10L)); + + + List results = getResults(); + assertThat(results).hasSize(3); + PendingMessages pending = (PendingMessages) results.get(2); + + assertThat(pending.size()).isZero(); + } + protected void verifyResults(List expected) { assertThat(getResults()).isEqualTo(expected); } 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 ac3d4c9c7..cd02d840b 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 @@ -19,20 +19,20 @@ import static org.assertj.core.api.Assertions.*; import static org.junit.Assume.*; import io.lettuce.core.XReadArgs; -import org.junit.Ignore; -import org.springframework.data.redis.connection.stream.RecordId; import reactor.test.StepVerifier; import java.util.Collections; import org.junit.Before; +import org.junit.Ignore; import org.junit.Test; import org.springframework.data.domain.Range; import org.springframework.data.redis.RedisTestProfileValueSource; +import org.springframework.data.redis.connection.RedisZSetCommands.Limit; import org.springframework.data.redis.connection.stream.Consumer; import org.springframework.data.redis.connection.stream.ReadOffset; +import org.springframework.data.redis.connection.stream.RecordId; import org.springframework.data.redis.connection.stream.StreamOffset; -import org.springframework.data.redis.connection.RedisZSetCommands.Limit; /** * Integration tests for {@link LettuceReactiveStreamCommands}. @@ -45,7 +45,7 @@ public class LettuceReactiveStreamCommandsTests extends LettuceReactiveCommandsT public void before() { // TODO: Upgrade to 5.0 - assumeTrue(RedisTestProfileValueSource.matches("redisVersion", "4.9")); + assumeTrue(RedisTestProfileValueSource.atLeast("redisVersion", "4.9")); } @Test // DATAREDIS-864 @@ -179,9 +179,9 @@ public class LettuceReactiveStreamCommandsTests extends LettuceReactiveCommandsT nativeCommands.xadd(KEY_1, Collections.singletonMap(KEY_2, VALUE_2)); connection.streamCommands().xGroupCreate(KEY_1_BBUFFER, "group-1", ReadOffset.latest()) // - .as(StepVerifier::create) // - .expectNext("OK") // - .verifyComplete(); + .as(StepVerifier::create) // + .expectNext("OK") // + .verifyComplete(); } @Test // DATAREDIS-864 @@ -190,7 +190,8 @@ public class LettuceReactiveStreamCommandsTests extends LettuceReactiveCommandsT String id = nativeCommands.xadd(KEY_1, Collections.singletonMap(KEY_2, VALUE_2)); nativeCommands.xgroupCreate(XReadArgs.StreamOffset.from(KEY_1, id), "group-1"); - nativeCommands.xreadgroup(io.lettuce.core.Consumer.from("group-1", "consumer-1"), XReadArgs.StreamOffset.from(KEY_1, id)); + nativeCommands.xreadgroup(io.lettuce.core.Consumer.from("group-1", "consumer-1"), + XReadArgs.StreamOffset.from(KEY_1, id)); connection.streamCommands().xGroupDelConsumer(KEY_1_BBUFFER, Consumer.from("group-1", "consumer-1")) .as(StepVerifier::create) // @@ -204,12 +205,136 @@ public class LettuceReactiveStreamCommandsTests extends LettuceReactiveCommandsT String id = nativeCommands.xadd(KEY_1, Collections.singletonMap(KEY_2, VALUE_2)); nativeCommands.xgroupCreate(XReadArgs.StreamOffset.from(KEY_1, id), "group-1"); - connection.streamCommands().xGroupDestroy(KEY_1_BBUFFER, "group-1") - .as(StepVerifier::create) // + connection.streamCommands().xGroupDestroy(KEY_1_BBUFFER, "group-1").as(StepVerifier::create) // .expectNext("OK") // .verifyComplete(); } + @Test // DATAREDIS-1084 + public void xPendingShouldLoadOverviewCorrectly() { + String initialMessage = nativeCommands.xadd(KEY_1, KEY_1, VALUE_1); + nativeCommands.xgroupCreate(XReadArgs.StreamOffset.from(KEY_1, initialMessage), "my-group"); + + nativeCommands.xadd(KEY_1, KEY_2, VALUE_2); + + connection.streamCommands() + .xReadGroup(Consumer.from("my-group", "my-consumer"), + StreamOffset.create(KEY_1_BBUFFER, ReadOffset.lastConsumed())) // + .then().as(StepVerifier::create) // + .verifyComplete(); + + nativeCommands.xadd(KEY_1, KEY_2, VALUE_2); + + connection.streamCommands().xPending(KEY_1_BBUFFER, "my-group").as(StepVerifier::create).assertNext(it -> { + + assertThat(it.getGroupName()).isEqualTo("my-group"); + assertThat(it.getTotalPendingMessages()).isEqualTo(1L); + assertThat(it.getIdRange()).isNotNull(); + assertThat(it.getPendingMessagesPerConsumer()).hasSize(1).containsEntry("my-consumer", 1L); + }).verifyComplete(); + } + + @Test // DATAREDIS-1084 + public void xPendingShouldLoadEmptyOverviewCorrectly() { + + String initialMessage = nativeCommands.xadd(KEY_1, KEY_1, VALUE_1); + nativeCommands.xgroupCreate(XReadArgs.StreamOffset.from(KEY_1, initialMessage), "my-group"); + + nativeCommands.xadd(KEY_1, KEY_2, VALUE_2); + + connection.streamCommands().xPending(KEY_1_BBUFFER, "my-group").as(StepVerifier::create).assertNext(it -> { + + assertThat(it.getGroupName()).isEqualTo("my-group"); + assertThat(it.getTotalPendingMessages()).isEqualTo(0L); + assertThat(it.getIdRange()).isNotNull(); + assertThat(it.getPendingMessagesPerConsumer()).isEmpty(); + }).verifyComplete(); + } + + @Test // DATAREDIS-1084 + public void xPendingShouldLoadPendingMessages() { + + String initialMessage = nativeCommands.xadd(KEY_1, KEY_1, VALUE_1); + nativeCommands.xgroupCreate(XReadArgs.StreamOffset.from(KEY_1, initialMessage), "my-group"); + + nativeCommands.xadd(KEY_1, KEY_2, VALUE_2); + + connection.streamCommands() + .xReadGroup(Consumer.from("my-group", "my-consumer"), + StreamOffset.create(KEY_1_BBUFFER, ReadOffset.lastConsumed())) // + .then().as(StepVerifier::create) // + .verifyComplete(); + + connection.streamCommands().xPending(KEY_1_BBUFFER, "my-group", Range.open("-", "+"), 10L).as(StepVerifier::create) + .assertNext(it -> { + + assertThat(it.size()).isOne(); + 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(); + }).verifyComplete(); + } + + @Test // DATAREDIS-1084 + public void xPendingShouldLoadPendingMessagesForConsumer() { + + String initialMessage = nativeCommands.xadd(KEY_1, KEY_1, VALUE_1); + nativeCommands.xgroupCreate(XReadArgs.StreamOffset.from(KEY_1, initialMessage), "my-group"); + + nativeCommands.xadd(KEY_1, KEY_2, VALUE_2); + + connection.streamCommands() + .xReadGroup(Consumer.from("my-group", "my-consumer"), + StreamOffset.create(KEY_1_BBUFFER, ReadOffset.lastConsumed())) // + .then().as(StepVerifier::create) // + .verifyComplete(); + + connection.streamCommands().xPending(KEY_1_BBUFFER, "my-group", "my-consumer", Range.open("-", "+"), 10L).as(StepVerifier::create) + .assertNext(it -> { + + assertThat(it.size()).isOne(); + 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(); + }).verifyComplete(); + } + + @Test // DATAREDIS-1084 + public void xPendingShouldLoadPendingMessagesForNonExistingConsumer() { + + String initialMessage = nativeCommands.xadd(KEY_1, KEY_1, VALUE_1); + nativeCommands.xgroupCreate(XReadArgs.StreamOffset.from(KEY_1, initialMessage), "my-group"); + + nativeCommands.xadd(KEY_1, KEY_2, VALUE_2); + + connection.streamCommands() + .xReadGroup(Consumer.from("my-group", "my-consumer"), + StreamOffset.create(KEY_1_BBUFFER, ReadOffset.lastConsumed())) // + .then().as(StepVerifier::create) // + .verifyComplete(); + + connection.streamCommands().xPending(KEY_1_BBUFFER, "my-group", "my-consumer-2", Range.open("-", "+"), 10L).as(StepVerifier::create) + .assertNext(it -> { + + assertThat(it.size()).isZero(); + }).verifyComplete(); + } + + @Test // DATAREDIS-1084 + public void xPendingShouldLoadEmptyPendingMessages() { + + String initialMessage = nativeCommands.xadd(KEY_1, KEY_1, VALUE_1); + nativeCommands.xgroupCreate(XReadArgs.StreamOffset.from(KEY_1, initialMessage), "my-group"); + + nativeCommands.xadd(KEY_1, KEY_2, VALUE_2); + + connection.streamCommands().xPending(KEY_1_BBUFFER, "my-group", Range.open("-", "+"), 10L).as(StepVerifier::create) + .assertNext(it -> { + assertThat(it.isEmpty()).isTrue(); + }).verifyComplete(); + } } diff --git a/src/test/java/org/springframework/data/redis/core/DefaultReactiveStreamOperationsTests.java b/src/test/java/org/springframework/data/redis/core/DefaultReactiveStreamOperationsTests.java index 3b5cfe8a3..701e1282e 100644 --- a/src/test/java/org/springframework/data/redis/core/DefaultReactiveStreamOperationsTests.java +++ b/src/test/java/org/springframework/data/redis/core/DefaultReactiveStreamOperationsTests.java @@ -40,6 +40,7 @@ import org.springframework.data.redis.connection.RedisConnection; import org.springframework.data.redis.connection.RedisConnectionFactory; import org.springframework.data.redis.connection.RedisZSetCommands.Limit; import org.springframework.data.redis.connection.lettuce.LettuceConnectionFactory; +import org.springframework.data.redis.connection.stream.Consumer; import org.springframework.data.redis.connection.stream.MapRecord; import org.springframework.data.redis.connection.stream.ReadOffset; import org.springframework.data.redis.connection.stream.RecordId; @@ -322,4 +323,53 @@ public class DefaultReactiveStreamOperationsTests { .expectNext(2L) // .verifyComplete(); } + + @Test // DATAREDIS-1084 + public void pendingShouldReadMessageSummary() { + + K key = keyFactory.instance(); + HK hashKey = hashKeyFactory.instance(); + HV value = valueFactory.instance(); + + streamOperations.add(key, Collections.singletonMap(hashKey, value)).then().as(StepVerifier::create) + .verifyComplete(); + + streamOperations.createGroup(key, ReadOffset.from("0-0"), "my-group").then().as(StepVerifier::create) + .verifyComplete(); + + streamOperations.read(Consumer.from("my-group", "my-consumer"), StreamOffset.create(key, ReadOffset.lastConsumed())) + .then().as(StepVerifier::create).verifyComplete(); + + streamOperations.pending(key, "my-group").as(StepVerifier::create).assertNext(pending -> { + + assertThat(pending.getTotalPendingMessages()).isOne(); + assertThat(pending.getGroupName()).isEqualTo("my-group"); + }).verifyComplete(); + } + + @Test // DATAREDIS-1084 + public void pendingShouldReadMessageDetails() { + + K key = keyFactory.instance(); + HK hashKey = hashKeyFactory.instance(); + HV value = valueFactory.instance(); + + streamOperations.add(key, Collections.singletonMap(hashKey, value)).then().as(StepVerifier::create) + .verifyComplete(); + + streamOperations.createGroup(key, ReadOffset.from("0-0"), "my-group").then().as(StepVerifier::create) + .verifyComplete(); + + streamOperations.read(Consumer.from("my-group", "my-consumer"), StreamOffset.create(key, ReadOffset.lastConsumed())) + .then().as(StepVerifier::create).verifyComplete(); + + streamOperations.pending(key, "my-group", Range.unbounded(), 10L).as(StepVerifier::create).assertNext(pending -> { + + assertThat(pending).hasSize(1); + assertThat(pending.get(0).getGroupName()).isEqualTo("my-group"); + assertThat(pending.get(0).getConsumerName()).isEqualTo("my-consumer"); + assertThat(pending.get(0).getTotalDeliveryCount()).isOne(); + }).verifyComplete(); + + } } diff --git a/src/test/java/org/springframework/data/redis/core/DefaultStreamOperationsTests.java b/src/test/java/org/springframework/data/redis/core/DefaultStreamOperationsTests.java index d95832ab0..8fe09913a 100644 --- a/src/test/java/org/springframework/data/redis/core/DefaultStreamOperationsTests.java +++ b/src/test/java/org/springframework/data/redis/core/DefaultStreamOperationsTests.java @@ -36,6 +36,8 @@ import org.springframework.data.redis.ObjectFactory; import org.springframework.data.redis.Person; import org.springframework.data.redis.RedisTestProfileValueSource; import org.springframework.data.redis.connection.stream.Consumer; +import org.springframework.data.redis.connection.stream.PendingMessages; +import org.springframework.data.redis.connection.stream.PendingMessagesSummary; import org.springframework.data.redis.connection.stream.RecordId; import org.springframework.data.redis.connection.stream.MapRecord; import org.springframework.data.redis.connection.stream.ObjectRecord; @@ -72,7 +74,7 @@ public class DefaultStreamOperationsTests { // See https://github.com/xetorthio/jedis/issues/1820 assumeTrue(redisTemplate.getConnectionFactory() instanceof LettuceConnectionFactory); - assumeTrue(RedisTestProfileValueSource.matches("redisVersion", "5.0")); + assumeTrue(RedisTestProfileValueSource.atLeast("redisVersion", "5.0")); this.redisTemplate = redisTemplate; this.keyFactory = keyFactory; @@ -322,4 +324,44 @@ public class DefaultStreamOperationsTests { streamOps.add(key, Collections.singletonMap(hashKey, value)); assertThat(streamOps.size(key)).isEqualTo(2); } + + @Test // DATAREDIS-1084 + public void pendingShouldReadMessageSummary() { + + K key = keyFactory.instance(); + HK hashKey = hashKeyFactory.instance(); + HV value = hashValueFactory.instance(); + + RecordId messageId = streamOps.add(key, Collections.singletonMap(hashKey, value)); + streamOps.createGroup(key, ReadOffset.from("0-0"), "my-group"); + + streamOps.read(Consumer.from("my-group", "my-consumer"), + StreamOffset.create(key, ReadOffset.lastConsumed())); + + PendingMessagesSummary pending = streamOps.pending(key, "my-group"); + + assertThat(pending.getTotalPendingMessages()).isOne(); + assertThat(pending.getGroupName()).isEqualTo("my-group"); + } + + @Test // DATAREDIS-1084 + public void pendingShouldReadMessageDetails() { + + K key = keyFactory.instance(); + HK hashKey = hashKeyFactory.instance(); + HV value = hashValueFactory.instance(); + + RecordId messageId = streamOps.add(key, Collections.singletonMap(hashKey, value)); + streamOps.createGroup(key, ReadOffset.from("0-0"), "my-group"); + + streamOps.read(Consumer.from("my-group", "my-consumer"), + StreamOffset.create(key, ReadOffset.lastConsumed())); + + PendingMessages pending = streamOps.pending(key, "my-group", Range.unbounded(), 10L); + + assertThat(pending).hasSize(1); + assertThat(pending.get(0).getGroupName()).isEqualTo("my-group"); + assertThat(pending.get(0).getConsumerName()).isEqualTo("my-consumer"); + assertThat(pending.get(0).getTotalDeliveryCount()).isOne(); + } }