DATAREDIS-1084 - Add support for XPENDING.

Original pull request: spring-projects#512.

Author: christophstrobl

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

@@ -37,6 +37,8 @@
 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 Long xLen(String key) {
 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<String> 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<String> 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 List<StringRecord> xReadAsString(StreamReadOptions readOptions, StreamOff
  */
 @Override
 public List<StringRecord> xReadGroupAsString(Consumer consumer, StreamReadOptions readOptions,
- StreamOffset<String>... streams) {
+StreamOffset<String>... streams) {
 
 return convertAndReturn(delegate.xReadGroup(consumer, readOptions, serialize(streams)),
 listByteMapRecordToStringMapRecordConverter);
@@ -3808,6 +3848,24 @@ public Long xLen(byte[] key) {
 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 List<ByteRecord> xRead(StreamReadOptions readOptions, StreamOffset<byte[]
  */
 @Override
 public List<ByteRecord> xReadGroup(Consumer consumer, StreamReadOptions readOptions,
- StreamOffset<byte[]>... streams) {
+StreamOffset<byte[]>... streams) {
 return delegate.xReadGroup(consumer, readOptions, streams);
 }
 

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

@@ -31,6 +31,8 @@
 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 @@ default Long xLen(byte[] key) {
 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 @@ default List<ByteRecord> xReadGroup(Consumer consumer, StreamOffset<byte[]>... s
 @Override
 @Deprecated
 default List<ByteRecord> xReadGroup(Consumer consumer, StreamReadOptions readOptions,
-StreamOffset<byte[]>... streams) {
+StreamOffset<byte[]>... streams) {
 return streamCommands().xReadGroup(consumer, readOptions, streams);
 }
 

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

@@ -30,16 +30,21 @@
 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;
 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 @@ default Mono<Long> xLen(ByteBuffer key) {
  */
 Flux<NumericResponse<KeyCommand, Long>> xLen(Publisher<KeyCommand> 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 <a href="https://redis.io/commands/xpending">Redis Documentation: xpending</a>
+ * @since 2.3
+ */
+default Mono<PendingMessagesSummary> 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 <a href="https://redis.io/commands/xpending">Redis Documentation: xpending</a>
+ * @since 2.3
+ */
+Flux<CommandResponse<PendingRecordsCommand, PendingMessagesSummary>> xPendingSummary(
+Publisher<PendingRecordsCommand> 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 <a href="https://redis.io/commands/xpending">Redis Documentation: xpending</a>
+ * @since 2.3
+ */
+@Nullable
+default Mono<PendingMessages> 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 <a href="https://redis.io/commands/xpending">Redis Documentation: xpending</a>
+ * @since 2.3
+ */
+@Nullable
+default Mono<PendingMessages> 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 <a href="https://redis.io/commands/xpending">Redis Documentation: xpending</a>
+ * @since 2.3
+ */
+default Mono<PendingMessages> 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 <a href="https://redis.io/commands/xpending">Redis Documentation: xpending</a>
+ * @since 2.3
+ */
+default Mono<PendingMessages> 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 <a href="https://redis.io/commands/xpending">Redis Documentation: xpending</a>
+ * @since 2.3
+ */
+default Mono<PendingMessages> 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 <a href="https://redis.io/commands/xpending">Redis Documentation: xpending</a>
+ * @since 2.3
+ */
+Flux<CommandResponse<PendingRecordsCommand, PendingMessages>> xPending(Publisher<PendingRecordsCommand> 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<String> 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.
  *