Skip to content

E
eterban
Commit 2044d195 authored by Vitaly Lipatov's avatar Vitaly Lipatov
Browse files
Options

gateway: reclaim pending Redis Stream commands

parent 1be387b7
Hide whitespace changes
Inline Side-by-side
Showing with 9 additions and 4 deletions
gateway/usr/share/eterban/eterban_switcher.py
View file @ 2044d195
...@@ -687,10 +687,15 @@ def process_stream_entry(fields): ...@@ -687,10 +687,15 @@ def process_stream_entry(fields):
while True: while True:
try: try:
pending = r.xreadgroup(redis_commands_group, redis_commands_consumer, claimed = r.xautoclaim(redis_commands_stream, redis_commands_group,
{redis_commands_stream: '0'}, count=10) redis_commands_consumer, min_idle_time=60000,
entries = pending or r.xreadgroup(redis_commands_group, redis_commands_consumer, start_id='0-0', count=10)
{redis_commands_stream: '>'}, count=10, block=5000) entries = [(redis_commands_stream, claimed[1])] if claimed[1] else []
if not entries:
pending = r.xreadgroup(redis_commands_group, redis_commands_consumer,
{redis_commands_stream: '0'}, count=10)
entries = pending or r.xreadgroup(redis_commands_group, redis_commands_consumer,
{redis_commands_stream: '>'}, count=10, block=5000)
for stream, messages in entries: for stream, messages in entries:
for message_id, fields in messages: for message_id, fields in messages:
if process_stream_entry(fields): if process_stream_entry(fields):
......
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment