[Redis 6.2] Add idle option to XPENDING

Fix: #1125

Author: Aaron Israel

CHANGELOG.md

@@ -5,6 +5,7 @@
 - Use `MD5` for hashing server nodes in `Redis::Distributed`. This should improve keys distribution among servers. See #1089.
 - Changed `sadd` and `srem` to now always return an Integer.
 - Added `sadd?` and `srem?` which always return a Boolean.
+- Added support for `IDLE` paramter in `xpending`.
 - Cluster support has been moved to a `redis_cluster` companion gem.
 - `select` no longer record the current database. If the client has to reconnect after `select` was used, it will reconnect to the original database.
 - Better support Float timeout in blocking commands. See #977.

lib/redis/commands/streams.rb

@@ -331,6 +331,8 @@ def xautoclaim(key, group, consumer, min_idle_time, start, count: nil, justid: f
       #   redis.xpending('mystream', 'mygroup')
       # @example With range options
       #   redis.xpending('mystream', 'mygroup', '-', '+', 10)
+      # @example With range and idle time options
+      #   redis.xpending('mystream', 'mygroup', '-', '+', 10, idle: 9000)
       # @example With range and consumer options
       #   redis.xpending('mystream', 'mygroup', '-', '+', 10, 'consumer1')
       #
@@ -341,10 +343,13 @@ def xautoclaim(key, group, consumer, min_idle_time, start, count: nil, justid: f
       # @param count    [Integer] count the number of entries as limit
       # @param consumer [String]  the consumer name
       #
+      # @option opts [Integer] :idle       pending message minimum idle time in milliseconds
+      #
       # @return [Hash]        the summary of pending entries
       # @return [Array<Hash>] the pending entries details if options were specified
-      def xpending(key, group, *args)
+      def xpending(key, group, *args, idle: nil)
         command_args = [:xpending, key, group]
+        command_args << 'IDLE' << Integer(idle) if idle
         case args.size
         when 0, 3, 4
           command_args.concat(args)

test/lint/streams.rb

@@ -753,6 +753,51 @@ def test_xpending_with_range_options
       assert_equal 1, actual[2]['count']
     end
 
+    def test_xpending_with_range_and_idle_options
+      target_version "6.2" do
+        redis.xadd('s1', { f: 'v1' }, id: '0-1')
+        redis.xgroup(:create, 's1', 'g1', '$')
+        redis.xadd('s1', { f: 'v2' }, id: '0-2')
+        redis.xadd('s1', { f: 'v3' }, id: '0-3')
+        redis.xreadgroup('g1', 'c1', 's1', '>')
+
+        actual = redis.xpending('s1', 'g1', '-', '+', 10)
+        assert_equal 2, actual.size
+        actual = redis.xpending('s1', 'g1', '-', '+', 10, idle: 10)
+        assert_equal 0, actual.size
+        sleep 0.1
+        actual = redis.xpending('s1', 'g1', '-', '+', 10, idle: 10)
+        assert_equal 2, actual.size
+
+        redis.xadd('s1', { f: 'v4' }, id: '0-4')
+        redis.xreadgroup('g1', 'c2', 's1', '>')
+
+        actual = redis.xpending('s1', 'g1', '-', '+', 10, idle: 1000)
+        assert_equal 0, actual.size
+
+        actual = redis.xpending('s1', 'g1', '-', '+', 10)
+        assert_equal 3, actual.size
+        actual = redis.xpending('s1', 'g1', '-', '+', 10, idle: 10)
+        assert_equal 2, actual.size
+        sleep 0.01
+        actual = redis.xpending('s1', 'g1', '-', '+', 10, idle: 10)
+        assert_equal 3, actual.size
+
+        assert_equal '0-2', actual[0]['entry_id']
+        assert_equal 'c1', actual[0]['consumer']
+        assert_equal true, actual[0]['elapsed'] >= 0
+        assert_equal 1, actual[0]['count']
+        assert_equal '0-3', actual[1]['entry_id']
+        assert_equal 'c1', actual[1]['consumer']
+        assert_equal true, actual[1]['elapsed'] >= 0
+        assert_equal 1, actual[1]['count']
+        assert_equal '0-4', actual[2]['entry_id']
+        assert_equal 'c2', actual[2]['consumer']
+        assert_equal true, actual[2]['elapsed'] >= 0
+        assert_equal 1, actual[2]['count']
+      end
+    end
+
     def test_xpending_with_range_and_consumer_options
       redis.xadd('s1', { f: 'v1' }, id: '0-1')
       redis.xgroup(:create, 's1', 'g1', '$')