summaryrefslogtreecommitdiff
diff options
context:
space:
mode:
authorAvital Fine <79420960+AvitalFineRedis@users.noreply.github.com>2021-08-05 12:24:40 +0300
committerGitHub <noreply@github.com>2021-08-05 12:24:40 +0300
commit9c60670deea5c593e20204bbd5f172ccfcd1d9db (patch)
treef7d88c36d3af71187535d655e724287d3c97a9a3
parent238f69e36e0ff5ac9b892706e3a5478b138069cd (diff)
downloadredis-py-9c60670deea5c593e20204bbd5f172ccfcd1d9db.tar.gz
add idle to xpending (#1523)
-rwxr-xr-xredis/client.py42
-rw-r--r--tests/test_commands.py46
2 files changed, 74 insertions, 14 deletions
diff --git a/redis/client.py b/redis/client.py
index e5cd647..995239e 100755
--- a/redis/client.py
+++ b/redis/client.py
@@ -2747,7 +2747,7 @@ class Redis:
return self.execute_command('XPENDING', name, groupname)
def xpending_range(self, name, groupname, min, max, count,
- consumername=None):
+ consumername=None, idle=None):
"""
Returns information about pending messages, in a range.
name: name of the stream.
@@ -2756,21 +2756,35 @@ class Redis:
max: maximum stream ID.
count: number of messages to return
consumername: name of a consumer to filter by (optional).
+ idle: available from version 6.2. filter entries by their
+ idle-time, given in milliseconds (optional).
"""
+ if {min, max, count} == {None}:
+ if idle is not None or consumername is not None:
+ raise DataError("if XPENDING is provided with idle time"
+ " or consumername, it must be provided"
+ " with min, max and count parameters")
+ return self.xpending(name, groupname)
+
pieces = [name, groupname]
- if min is not None or max is not None or count is not None:
- if min is None or max is None or count is None:
- raise DataError("XPENDING must be provided with min, max "
- "and count parameters, or none of them. ")
- if not isinstance(count, int) or count < -1:
- raise DataError("XPENDING count must be a integer >= -1")
- pieces.extend((min, max, str(count)))
- if consumername is not None:
- if min is None or max is None or count is None:
- raise DataError("if XPENDING is provided with consumername,"
- " it must be provided with min, max and"
- " count parameters")
- pieces.append(consumername)
+ if min is None or max is None or count is None:
+ raise DataError("XPENDING must be provided with min, max "
+ "and count parameters, or none of them.")
+ # idle
+ try:
+ if int(idle) < 0:
+ raise DataError("XPENDING idle must be a integer >= 0")
+ pieces.extend(['IDLE', idle])
+ except TypeError:
+ pass
+ # count
+ try:
+ if int(count) < 0:
+ raise DataError("XPENDING count must be a integer >= 0")
+ pieces.extend([min, max, count])
+ except TypeError:
+ pass
+
return self.execute_command('XPENDING', *pieces, parse_detail=True)
def xrange(self, name, min='-', max='+', count=None):
diff --git a/tests/test_commands.py b/tests/test_commands.py
index dbe0093..e92be62 100644
--- a/tests/test_commands.py
+++ b/tests/test_commands.py
@@ -2636,6 +2636,52 @@ class TestRedisCommands:
assert response[1]['message_id'] == m2
assert response[1]['consumer'] == consumer2.encode()
+ @skip_if_server_version_lt('6.2.0')
+ def test_xpending_range_idle(self, r):
+ stream = 'stream'
+ group = 'group'
+ consumer1 = 'consumer1'
+ consumer2 = 'consumer2'
+ r.xadd(stream, {'foo': 'bar'})
+ r.xadd(stream, {'foo': 'bar'})
+ r.xgroup_create(stream, group, 0)
+
+ # read 1 message from the group with each consumer
+ r.xreadgroup(group, consumer1, streams={stream: '>'}, count=1)
+ r.xreadgroup(group, consumer2, streams={stream: '>'}, count=1)
+
+ response = r.xpending_range(stream, group,
+ min='-', max='+', count=5)
+ assert len(response) == 2
+ response = r.xpending_range(stream, group,
+ min='-', max='+', count=5, idle=1000)
+ assert len(response) == 0
+
+ def test_xpending_range_negative(self, r):
+ stream = 'stream'
+ group = 'group'
+ with pytest.raises(redis.DataError):
+ r.xpending_range(stream, group, min='-', max='+', count=None)
+ with pytest.raises(ValueError):
+ r.xpending_range(stream, group, min='-', max='+', count="one")
+ with pytest.raises(redis.DataError):
+ r.xpending_range(stream, group, min='-', max='+', count=-1)
+ with pytest.raises(ValueError):
+ r.xpending_range(stream, group, min='-', max='+', count=5,
+ idle="one")
+ with pytest.raises(redis.exceptions.ResponseError):
+ r.xpending_range(stream, group, min='-', max='+', count=5,
+ idle=1.5)
+ with pytest.raises(redis.DataError):
+ r.xpending_range(stream, group, min='-', max='+', count=5,
+ idle=-1)
+ with pytest.raises(redis.DataError):
+ r.xpending_range(stream, group, min=None, max=None, count=None,
+ idle=0)
+ with pytest.raises(redis.DataError):
+ r.xpending_range(stream, group, min=None, max=None, count=None,
+ consumername=0)
+
@skip_if_server_version_lt('5.0.0')
def test_xrange(self, r):
stream = 'stream'