Add support for min_idle_time in Redis StreamSub and XAUTOCLAIM

Signed-off-by: Victor Maleca <powersemmi@gmail.com>

Author: powersemmi

faststream/redis/schemas/stream_sub.py

@@ -33,6 +33,12 @@ class StreamSub(NameRequired):
             https://redis.io/docs/latest/commands/xreadgroup/#differences-between-xread-and-xreadgroup
         polling_interval:
             Polling interval in seconds.
+        min_idle_time:
+            Minimum idle time in milliseconds for a message to be eligible for claiming via XAUTOCLAIM.
+            Messages that have been pending (unacknowledged) for at least this duration can be
+            reclaimed by this consumer. Only applicable when using consumer groups.
+
+            https://redis.io/docs/latest/commands/xautoclaim/
     """
 
     __slots__ = (
@@ -42,6 +48,7 @@ class StreamSub(NameRequired):
         "last_id",
         "max_records",
         "maxlen",
+        "min_idle_time",
         "name",
         "no_ack",
         "polling_interval",
@@ -58,11 +65,15 @@ def __init__(
         last_id: str | None = None,
         maxlen: int | None = None,
         max_records: int | None = None,
+        min_idle_time: int | None = None,
     ) -> None:
         if (group and not consumer) or (not group and consumer):
             msg = "You should specify `group` and `consumer` both"
             raise SetupError(msg)
 
+        if last_id is None:
+            last_id = ">" if group and consumer else "$"
+
         if group and consumer:
             if last_id != ">":
                 if polling_interval:
@@ -86,9 +97,6 @@ def __init__(
                     stacklevel=1,
                 )
 
-        if last_id is None:
-            last_id = ">" if group and consumer else "$"
-
         super().__init__(stream)
 
         self.group = group
@@ -99,6 +107,7 @@ def __init__(
         self.last_id = last_id
         self.maxlen = maxlen
         self.max_records = max_records
+        self.min_idle_time = min_idle_time
 
     def add_prefix(self, prefix: str) -> "StreamSub":
         new_stream = deepcopy(self)

faststream/redis/subscriber/usecases/stream_subscriber.py

@@ -1,5 +1,5 @@
 import math
-from collections.abc import AsyncIterator, Awaitable, Callable
+from collections.abc import AsyncIterator, Awaitable, Callable, Coroutine
 from typing import TYPE_CHECKING, Any, Optional, TypeAlias
 
 from redis.exceptions import ResponseError
@@ -47,6 +47,8 @@ def __init__(
         assert config.stream_sub
         self._stream_sub = config.stream_sub
         self.last_id = config.stream_sub.last_id
+        self.min_idle_time = config.stream_sub.min_idle_time
+        self.autoclaim_start_id = "0-0"
 
     @property
     def stream_sub(self) -> "StreamSub":
@@ -139,7 +141,7 @@ def read(
                     noack=stream.no_ack,
                 )
 
-        else:
+        elif self.stream_sub.min_idle_time is None:
 
             def read(
                 last_id: str,
@@ -164,6 +166,50 @@ def read(
                     count=stream.max_records,
                 )
 
+        else:
+
+            def read(
+                _: str,
+            ) -> Coroutine[
+                Any,
+                Any,
+                tuple[
+                    tuple[
+                        TopicName,
+                        tuple[
+                            tuple[
+                                Offset,
+                                dict[bytes, bytes],
+                            ],
+                            ...,
+                        ],
+                    ],
+                    ...,
+                ],
+            ]:
+                async def xautoclaim() -> tuple[
+                    tuple[TopicName, tuple[tuple[Offset, dict[bytes, bytes]], ...]], ...
+                ]:
+                    stream_message = await client.xautoclaim(
+                        name=self.stream_sub.name,
+                        groupname=self.stream_sub.group,
+                        consumername=self.stream_sub.consumer,
+                        min_idle_time=self.min_idle_time,
+                        start_id=self.autoclaim_start_id,
+                        count=1,
+                    )
+                    stream_name = self.stream_sub.name.encode()
+                    (next_id, messages, _) = stream_message
+                    # Update start_id for next call
+                    self.autoclaim_start_id = (
+                        next_id.decode() if isinstance(next_id, bytes) else next_id
+                    )
+                    if not messages:
+                        return ()
+                    return ((stream_name, messages),)
+
+                return xautoclaim()
+
         await super().start(read)
 
     @override
@@ -175,17 +221,43 @@ async def get_one(
         assert not self.calls, (
             "You can't use `get_one` method if subscriber has registered handlers."
         )
+        if self.min_idle_time is None:
+            if self.stream_sub.group and self.stream_sub.consumer:
+                stream_message = await self._client.xreadgroup(
+                    groupname=self.stream_sub.group,
+                    consumername=self.stream_sub.consumer,
+                    streams={self.stream_sub.name: self.last_id},
+                    block=math.ceil(timeout * 1000),
+                    count=1,
+                )
+            else:
+                stream_message = await self._client.xread(
+                    {self.stream_sub.name: self.last_id},
+                    block=math.ceil(timeout * 1000),
+                    count=1,
+                )
+            if not stream_message:
+                return None
 
-        stream_message = await self._client.xread(
-            {self.stream_sub.name: self.last_id},
-            block=math.ceil(timeout * 1000),
-            count=1,
-        )
-
-        if not stream_message:
-            return None
-
-        ((stream_name, ((message_id, raw_message),)),) = stream_message
+            ((stream_name, ((message_id, raw_message),)),) = stream_message
+        else:
+            stream_message = await self._client.xautoclaim(
+                name=self.stream_sub.name,
+                groupname=self.stream_sub.group,
+                consumername=self.stream_sub.consumer,
+                min_idle_time=self.min_idle_time,
+                start_id=self.autoclaim_start_id,
+                count=1,
+            )
+            (next_id, messages, _) = stream_message
+            # Update start_id for next call
+            self.autoclaim_start_id = (
+                next_id.decode() if isinstance(next_id, bytes) else next_id
+            )
+            if not messages:
+                return None
+            stream_name = self.stream_sub.name.encode()
+            ((message_id, raw_message),) = messages
 
         self.last_id = message_id.decode()
 
@@ -221,16 +293,43 @@ async def __aiter__(self) -> AsyncIterator["RedisStreamMessage"]:  # type: ignor
         async_parser, async_decoder = self._get_parser_and_decoder()
 
         while True:
-            stream_message = await self._client.xread(
-                {self.stream_sub.name: self.last_id},
-                block=math.ceil(timeout * 1000),
-                count=1,
-            )
-
-            if not stream_message:
-                continue
-
-            ((stream_name, ((message_id, raw_message),)),) = stream_message
+            if self.min_idle_time is None:
+                if self.stream_sub.group and self.stream_sub.consumer:
+                    stream_message = await self._client.xreadgroup(
+                        groupname=self.stream_sub.group,
+                        consumername=self.stream_sub.consumer,
+                        streams={self.stream_sub.name: self.last_id},
+                        block=math.ceil(timeout * 1000),
+                        count=1,
+                    )
+                else:
+                    stream_message = await self._client.xread(
+                        {self.stream_sub.name: self.last_id},
+                        block=math.ceil(timeout * 1000),
+                        count=1,
+                    )
+                if not stream_message:
+                    continue
+
+                ((stream_name, ((message_id, raw_message),)),) = stream_message
+            else:
+                stream_message = await self._client.xautoclaim(
+                    name=self.stream_sub.name,
+                    groupname=self.stream_sub.group,
+                    consumername=self.stream_sub.consumer,
+                    min_idle_time=self.min_idle_time,
+                    start_id=self.autoclaim_start_id,
+                    count=1,
+                )
+                (next_id, messages, _) = stream_message
+                # Update start_id for next call
+                self.autoclaim_start_id = (
+                    next_id.decode() if isinstance(next_id, bytes) else next_id
+                )
+                if not messages:
+                    continue
+                stream_name = self.stream_sub.name.encode()
+                ((message_id, raw_message),) = messages
 
             self.last_id = message_id.decode()
 

tests/brokers/redis/test_config.py

@@ -48,35 +48,31 @@ def test_stream_with_group() -> None:
 
 
 @pytest.mark.redis()
-def test_custom_ack() -> None:
+def test_stream_with_group_and_min_idle_time() -> None:
     config = RedisSubscriberConfig(
         _outer_config=MagicMock(),
         stream_sub=StreamSub(
             "test_stream",
             group="test_group",
             consumer="test_consumer",
+            min_idle_time=1000,
         ),
-        _ack_policy=AckPolicy.ACK,
     )
-    assert config.ack_policy is AckPolicy.ACK
+    assert config.ack_policy is AckPolicy.REJECT_ON_ERROR
 
 
 @pytest.mark.redis()
-def test_stream_sub_with_no_ack_group() -> None:
-    with pytest.warns(
-        RuntimeWarning,
-        match="`no_ack` is not supported by consumer group with last_id other than `>`",
-    ):
-        config = RedisSubscriberConfig(
-            _outer_config=MagicMock(),
-            stream_sub=StreamSub(
-                "test_stream",
-                group="test_group",
-                consumer="test_consumer",
-                no_ack=True,
-            ),
-        )
-    assert config.ack_policy is AckPolicy.MANUAL
+def test_custom_ack() -> None:
+    config = RedisSubscriberConfig(
+        _outer_config=MagicMock(),
+        stream_sub=StreamSub(
+            "test_stream",
+            group="test_group",
+            consumer="test_consumer",
+        ),
+        _ack_policy=AckPolicy.ACK,
+    )
+    assert config.ack_policy is AckPolicy.ACK
 
 
 @pytest.mark.redis()