diff options
| -rwxr-xr-x | redis/client.py | 47 | ||||
| -rw-r--r-- | tests/test_commands.py | 48 |
2 files changed, 94 insertions, 1 deletions
diff --git a/redis/client.py b/redis/client.py index 995239e..019ff4f 100755 --- a/redis/client.py +++ b/redis/client.py @@ -316,6 +316,12 @@ def parse_xclaim(response, **options): return parse_stream_list(response) +def parse_xautoclaim(response, **options): + if options.get('parse_justid', False): + return response[1] + return parse_stream_list(response[1]) + + def parse_xinfo_stream(response): data = pairs_to_dict(response, decode_keys=True) first = data['first-entry'] @@ -684,6 +690,7 @@ class Redis: 'SSCAN': parse_scan, 'TIME': lambda x: (int(x[0]), int(x[1])), 'XCLAIM': parse_xclaim, + 'XAUTOCLAIM': parse_xautoclaim, 'XGROUP CREATE': bool_ok, 'XGROUP DELCONSUMER': int, 'XGROUP DESTROY': bool, @@ -2601,6 +2608,46 @@ class Redis: pieces.extend(pair) return self.execute_command('XADD', name, *pieces) + def xautoclaim(self, name, groupname, consumername, min_idle_time, + start_id=0, count=None, justid=False): + """ + Transfers ownership of pending stream entries that match the specified + criteria. Conceptually, equivalent to calling XPENDING and then XCLAIM, + but provides a more straightforward way to deal with message delivery + failures via SCAN-like semantics. + name: name of the stream. + groupname: name of the consumer group. + consumername: name of a consumer that claims the message. + min_idle_time: filter messages that were idle less than this amount of + milliseconds. + start_id: filter messages with equal or greater ID. + count: optional integer, upper limit of the number of entries that the + command attempts to claim. Set to 100 by default. + justid: optional boolean, false by default. Return just an array of IDs + of messages successfully claimed, without returning the actual message + """ + try: + if int(min_idle_time) < 0: + raise DataError("XAUTOCLAIM min_idle_time must be a non" + "negative integer") + except TypeError: + pass + + kwargs = {} + pieces = [name, groupname, consumername, min_idle_time, start_id] + + try: + if int(count) < 0: + raise DataError("XPENDING count must be a integer >= 0") + pieces.extend([b'COUNT', count]) + except TypeError: + pass + if justid: + pieces.append(b'JUSTID') + kwargs['parse_justid'] = True + + return self.execute_command('XAUTOCLAIM', *pieces, **kwargs) + def xclaim(self, name, groupname, consumername, min_idle_time, message_ids, idle=None, time=None, retrycount=None, force=False, justid=False): diff --git a/tests/test_commands.py b/tests/test_commands.py index e92be62..60e667d 100644 --- a/tests/test_commands.py +++ b/tests/test_commands.py @@ -2374,13 +2374,59 @@ class TestRedisCommands: r.xadd(stream, {'some': 'other'}, nomkstream=True) assert r.xlen(stream) == 3 + @skip_if_server_version_lt('6.2.0') + def test_xautoclaim(self, r): + stream = 'stream' + group = 'group' + consumer1 = 'consumer1' + consumer2 = 'consumer2' + + message_id1 = r.xadd(stream, {'john': 'wick'}) + message_id2 = r.xadd(stream, {'johny': 'deff'}) + message = get_stream_message(r, stream, message_id1) + r.xgroup_create(stream, group, 0) + + # trying to claim a message that isn't already pending doesn't + # do anything + response = r.xautoclaim(stream, group, consumer2, min_idle_time=0) + assert response == [] + + # read the group as consumer1 to initially claim the messages + r.xreadgroup(group, consumer1, streams={stream: '>'}) + + # claim one message as consumer2 + response = r.xautoclaim(stream, group, consumer2, + min_idle_time=0, count=1) + assert response == [message] + + # reclaim the messages as consumer1, but use the justid argument + # which only returns message ids + assert r.xautoclaim(stream, group, consumer1, min_idle_time=0, + start_id=0, justid=True) == \ + [message_id1, message_id2] + assert r.xautoclaim(stream, group, consumer1, min_idle_time=0, + start_id=message_id2, justid=True) == \ + [message_id2] + + @skip_if_server_version_lt('6.2.0') + def test_xautoclaim_negative(self, r): + stream = 'stream' + group = 'group' + consumer = 'consumer' + with pytest.raises(redis.DataError): + r.xautoclaim(stream, group, consumer, min_idle_time=-1) + with pytest.raises(ValueError): + r.xautoclaim(stream, group, consumer, min_idle_time="wrong") + with pytest.raises(redis.DataError): + r.xautoclaim(stream, group, consumer, min_idle_time=0, + count=-1) + @skip_if_server_version_lt('5.0.0') def test_xclaim(self, r): stream = 'stream' group = 'group' consumer1 = 'consumer1' consumer2 = 'consumer2' - message_id = r.xadd(stream, {'john': 'wick'}) message = get_stream_message(r, stream, message_id) r.xgroup_create(stream, group, 0) |