Add support for IDLE option in XPENDING #1537

Original pull request: #1636.

src/main/java/io/lettuce/core/AbstractRedisAsyncCommands.java

@@ -1842,6 +1842,11 @@ public RedisFuture<List<PendingMessage>> xpending(K key, Consumer<K> consumer, R
         return dispatch(commandBuilder.xpending(key, consumer, range, limit));
     }
 
+    @Override
+    public RedisFuture<List<PendingMessage>> xpending(K key, XPendingArgs<K> args) {
+        return dispatch(commandBuilder.xpending(key, args));
+    }
+
     @Override
     public RedisFuture<List<StreamMessage<K, V>>> xrange(K key, Range<String> range) {
         return dispatch(commandBuilder.xrange(key, range, Limit.unlimited()));

src/main/java/io/lettuce/core/AbstractRedisReactiveCommands.java

@@ -1917,6 +1917,11 @@ public Flux<PendingMessage> xpending(K key, Consumer<K> consumer, Range<String>
         return createDissolvingFlux(() -> commandBuilder.xpending(key, consumer, range, limit));
     }
 
+    @Override
+    public Flux<PendingMessage> xpending(K key, XPendingArgs<K> args) {
+        return createDissolvingFlux(() -> commandBuilder.xpending(key, args));
+    }
+
     @Override
     public Flux<StreamMessage<K, V>> xrange(K key, Range<String> range) {
         return createDissolvingFlux(() -> commandBuilder.xrange(key, range, Limit.unlimited()));

src/main/java/io/lettuce/core/RedisCommandBuilder.java

@@ -2619,6 +2619,15 @@ public Command<K, V, List<PendingMessage>> xpending(K key, Consumer<K> consumer,
         return createCommand(XPENDING, new PendingMessageListOutput<>(codec), args);
     }
 
+    public Command<K, V, List<PendingMessage>> xpending(K key, XPendingArgs<K> commandArgs) {
+        notNullKey(key);
+        LettuceAssert.notNull(commandArgs, "XPendingArgs " + MUST_NOT_BE_NULL);
+
+        CommandArgs<K, V> args = new CommandArgs<>(codec).addKey(key);
+        commandArgs.build(args);
+        return createCommand(XPENDING, new PendingMessageListOutput<>(codec), args);
+    }
+
     public Command<K, V, List<StreamMessage<K, V>>> xrange(K key, Range<String> range, Limit limit) {
         notNullKey(key);
         LettuceAssert.notNull(range, "Range " + MUST_NOT_BE_NULL);

src/main/java/io/lettuce/core/XPendingArgs.java

@@ -0,0 +1,112 @@
+/*
+ * Copyright 2021 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
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * 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 io.lettuce.core;
+
+import io.lettuce.core.internal.LettuceAssert;
+import io.lettuce.core.protocol.CommandArgs;
+import io.lettuce.core.protocol.CommandKeyword;
+
+/**
+ * Argument list builder for the Redis <a href="http://redis.io/commands/xpending">XPENDING</a> command.
+ * Static import the methods from {@link XPendingArgs.Builder} and call the methods: {@code block(…)} .
+ * <p>
+ * {@link XPendingArgs} is a mutable object and instances should be used only once to avoid shared mutable state.
+ *
+ * @author dengliming
+ * @since 6.1
+ */
+public class XPendingArgs<K> {
+
+    private Consumer<K> consumer;
+
+    private Range<String> range;
+
+    private Limit limit;
+
+    private Long idle;
+
+    /**
+     * Builder entry points for {@link XPendingArgs}.
+     */
+    public static class Builder {
+
+        /**
+         * Utility constructor.
+         */
+        private Builder() {
+        }
+
+        /**
+         * Create a new {@link XPendingArgs} .
+         *
+         * @param consumer
+         * @param range
+         * @param limit
+         */
+        public static <K> XPendingArgs xpending(Consumer<K> consumer, Range<String> range, Limit limit) {
+            return new XPendingArgs().consumer(consumer).range(range).limit(limit);
+        }
+    }
+
+    public XPendingArgs range(Range<String> range) {
+        LettuceAssert.notNull(range, "Range must not be null");
+
+        this.range = range;
+        return this;
+    }
+
+    public XPendingArgs consumer(Consumer<K> consumer) {
+        LettuceAssert.notNull(consumer, "Consumer must not be null");
+
+        this.consumer = consumer;
+        return this;
+    }
+
+    public XPendingArgs limit(Limit limit) {
+        LettuceAssert.notNull(limit, "Limit must not be null");
+
+        this.limit = limit;
+        return this;
+    }
+
+    public XPendingArgs idle(Long idle) {
+        this.idle = idle;
+        return this;
+    }
+
+    public <V> void build(CommandArgs<K, V> args) {
+        args.addKey(consumer.group);
+
+        if (idle != null) {
+            args.add(CommandKeyword.IDLE).add(idle);
+        }
+
+        if (range.getLower().equals(Range.Boundary.unbounded())) {
+            args.add("-");
+        } else {
+            args.add(range.getLower().getValue());
+        }
+
+        if (range.getUpper().equals(Range.Boundary.unbounded())) {
+            args.add("+");
+        } else {
+            args.add(range.getUpper().getValue());
+        }
+
+        args.add(limit.isLimited() ? limit.getCount() : Long.MAX_VALUE);
+        args.addKey(consumer.name);
+    }
+}

src/main/java/io/lettuce/core/api/async/RedisStreamAsyncCommands.java

@@ -241,6 +241,15 @@ public interface RedisStreamAsyncCommands<K, V> {
      */
     RedisFuture<List<PendingMessage>> xpending(K key, Consumer<K> consumer, Range<String> range, Limit limit);
 
+    /**
+     * Read pending messages from a stream within a specific {@link XPendingArgs}.
+     *
+     * @param key the stream key.
+     * @param args
+     * @return List&lt;Object&gt; array-reply list with members of the resulting stream.
+     */
+    RedisFuture<List<PendingMessage>> xpending(K key, XPendingArgs<K> args);
+
     /**
      * Read messages from a stream within a specific {@link Range}.
      *

src/main/java/io/lettuce/core/api/reactive/RedisStreamReactiveCommands.java

@@ -242,6 +242,15 @@ public interface RedisStreamReactiveCommands<K, V> {
      */
     Flux<PendingMessage> xpending(K key, Consumer<K> consumer, Range<String> range, Limit limit);
 
+    /**
+     * Read pending messages from a stream within a specific {@link XPendingArgs}.
+     *
+     * @param key the stream key.
+     * @param args
+     * @return Object array-reply list with members of the resulting stream.
+     */
+    Flux<PendingMessage> xpending(K key, XPendingArgs<K> args);
+
     /**
      * Read messages from a stream within a specific {@link Range}.
      *

src/main/java/io/lettuce/core/api/sync/RedisStreamCommands.java

@@ -241,6 +241,15 @@ public interface RedisStreamCommands<K, V> {
      */
     List<PendingMessage> xpending(K key, Consumer<K> consumer, Range<String> range, Limit limit);
 
+    /**
+     * Read pending messages from a stream within a specific {@link XPendingArgs}.
+     *
+     * @param key the stream key.
+     * @param args
+     * @return List&lt;Object&gt; array-reply list with members of the resulting stream.
+     */
+    List<PendingMessage> xpending(K key, XPendingArgs<K> args);
+
     /**
      * Read messages from a stream within a specific {@link Range}.
      *

src/main/java/io/lettuce/core/cluster/api/async/NodeSelectionStreamAsyncCommands.java

@@ -241,6 +241,15 @@ public interface NodeSelectionStreamAsyncCommands<K, V> {
      */
     AsyncExecutions<List<PendingMessage>> xpending(K key, Consumer<K> consumer, Range<String> range, Limit limit);
 
+    /**
+     * Read pending messages from a stream within a specific {@link XPendingArgs}.
+     *
+     * @param key the stream key.
+     * @param args
+     * @return List&lt;Object&gt; array-reply list with members of the resulting stream.
+     */
+    AsyncExecutions<List<PendingMessage>> xpending(K key, XPendingArgs<K> args);
+
     /**
      * Read messages from a stream within a specific {@link Range}.
      *

src/main/java/io/lettuce/core/cluster/api/sync/NodeSelectionStreamCommands.java

@@ -241,6 +241,15 @@ public interface NodeSelectionStreamCommands<K, V> {
      */
     Executions<List<PendingMessage>> xpending(K key, Consumer<K> consumer, Range<String> range, Limit limit);
 
+    /**
+     * Read pending messages from a stream within a specific {@link XPendingArgs}.
+     *
+     * @param key the stream key.
+     * @param args
+     * @return List&lt;Object&gt; array-reply list with members of the resulting stream.
+     */
+    Executions<List<PendingMessage>> xpending(K key, XPendingArgs<K> args);
+
     /**
      * Read messages from a stream within a specific {@link Range}.
      *

src/main/kotlin/io/lettuce/core/api/coroutines/RedisStreamCoroutinesCommands.kt

@@ -241,6 +241,15 @@ interface RedisStreamCoroutinesCommands<K : Any, V : Any> {
      */
     fun xpending(key: K, consumer: Consumer<K>, range: Range<String>, limit: Limit): Flow<PendingMessage>
 
+    /**
+     * Read pending messages from a stream within a specific [XPendingArgs].
+     *
+     * @param key the stream key.
+     * @param args
+     * @return List<Any> array-reply list with members of the resulting stream.
+     */
+    fun xpending(key: K, args: XPendingArgs<K>): Flow<PendingMessage>
+
     /**
      * Read messages from a stream within a specific [Range].
      *

src/main/kotlin/io/lettuce/core/api/coroutines/RedisStreamCoroutinesCommandsImpl.kt

@@ -79,6 +79,8 @@ internal class RedisStreamCoroutinesCommandsImpl<K : Any, V : Any>(internal val
 
     override fun xpending(key: K, consumer: Consumer<K>, range: Range<String>, limit: Limit): Flow<PendingMessage> = ops.xpending(key, consumer, range, limit).asFlow()
 
+    override fun xpending(key: K, args: XPendingArgs<K>): Flow<PendingMessage> = ops.xpending(key, args).asFlow()
+
     override fun xrange(key: K, range: Range<String>): Flow<StreamMessage<K, V>> = ops.xrange(key, range).asFlow()
 
     override fun xrange(key: K, range: Range<String>, limit: Limit): Flow<StreamMessage<K, V>> = ops.xrange(key, range, limit).asFlow()

src/main/templates/io/lettuce/core/api/RedisStreamCommands.java

@@ -247,6 +247,15 @@ public interface RedisStreamCommands<K, V> {
      */
     List<PendingMessage> xpending(K key, Consumer<K> consumer, Range<String> range, Limit limit);
 
+    /**
+     * Read pending messages from a stream within a specific {@link XPendingArgs}.
+     *
+     * @param key the stream key.
+     * @param args
+     * @return List&lt;Object&gt; array-reply list with members of the resulting stream.
+     */
+    List<PendingMessage> xpending(K key, XPendingArgs<K> args);
+
     /**
      * Read messages from a stream within a specific {@link Range}.
      *

src/test/java/io/lettuce/core/commands/StreamCommandIntegrationTests.java

@@ -28,6 +28,7 @@
 
 import javax.inject.Inject;
 
+import io.lettuce.core.XPendingArgs;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.api.TestInstance;
@@ -441,6 +442,23 @@ void xpendingRanges() {
                 Limit.from(10))).hasSize(2);
     }
 
+    @Test
+    void xpendingWithArgs() {
+
+        redis.xgroupCreate(StreamOffset.latest(key), "group", XGroupCreateArgs.Builder.mkstream());
+        String id = redis.xadd(key, Collections.singletonMap("key", "value"));
+
+        redis.xreadgroup(Consumer.from("group", "consumer1"), StreamOffset.lastConsumed(key));
+
+        List<PendingMessage> pendingEntries = redis.xpending(key, XPendingArgs.Builder
+                .xpending(Consumer.from("group", "consumer1"), Range.unbounded(), Limit.from(10)).idle(30L));
+
+        PendingMessage message = pendingEntries.get(0);
+        assertThat(message.getId()).isEqualTo(id);
+        assertThat(message.getConsumer()).isEqualTo("consumer1");
+        assertThat(message.getRedeliveryCount()).isEqualTo(1);
+    }
+
     @Test
     void xpendingWithoutMessages() {