diff --git a/faststream/redis/subscriber/usecases/stream_subscriber.py b/faststream/redis/subscriber/usecases/stream_subscriber.py index 44b9c6e39d..ae9439a3ae 100644 --- a/faststream/redis/subscriber/usecases/stream_subscriber.py +++ b/faststream/redis/subscriber/usecases/stream_subscriber.py @@ -246,6 +246,7 @@ async def get_one( stream_message = await self._xreadgroup( count=1, block=math.ceil(timeout * 1000), + noack=self.stream_sub.no_ack, ) if not stream_message: return None @@ -323,6 +324,7 @@ async def __aiter__(self) -> AsyncIterator["RedisStreamMessage"]: stream_message = await self._xreadgroup( count=1, block=math.ceil(timeout * 1000), + noack=self.stream_sub.no_ack, ) if not stream_message: continue diff --git a/tests/brokers/redis/test_consume.py b/tests/brokers/redis/test_consume.py index 456d132308..984ff36e0a 100644 --- a/tests/brokers/redis/test_consume.py +++ b/tests/brokers/redis/test_consume.py @@ -2,6 +2,7 @@ from typing import Any from unittest.mock import ANY, AsyncMock, MagicMock, call, patch +import anyio import pytest from dirty_equals import IsPartialDict from pydantic import BaseModel @@ -979,6 +980,30 @@ async def test_get_one_timeout( mock(await subscriber.get_one(timeout=1e-24)) mock.assert_called_once_with(None) + @pytest.mark.parametrize("no_ack", (False, True)) + async def test_get_one_respects_no_ack( + self, + queue: str, + no_ack: bool, + ) -> None: + broker = self.get_broker() + sub = broker.subscriber( + stream=StreamSub(queue, group=queue, consumer="reader", no_ack=no_ack), + ) + + async with self.patch_broker(broker) as br: + await br.start() + await br.publish("hello", stream=queue) + + message = await sub.get_one(timeout=3) + + assert message is not None + assert await message.decode() == "hello" + + client = await br.connect() + pending = await client.xpending(queue, queue) # type: ignore[no-untyped-call] + assert pending["pending"] == (0 if no_ack else 1) + @pytest.mark.flaky(reruns=3, reruns_delay=1) async def test_concurrent_consume_stream( self, @@ -1054,6 +1079,30 @@ async def consume() -> None: calls = [call(msg) for msg in expected_messages] mock.assert_has_calls(calls=calls) + @pytest.mark.parametrize("no_ack", (False, True)) + async def test_iterator_respects_no_ack( + self, + queue: str, + no_ack: bool, + ) -> None: + broker = self.get_broker() + sub = broker.subscriber( + stream=StreamSub(queue, group=queue, consumer="reader", no_ack=no_ack), + ) + + async with self.patch_broker(broker) as br: + await br.start() + await br.publish("hello", stream=queue) + + with anyio.fail_after(3): + message = await anext(aiter(sub)) + + assert await message.decode() == "hello" + + client = await br.connect() + pending = await client.xpending(queue, queue) # type: ignore[no-untyped-call] + assert pending["pending"] == (0 if no_ack else 1) + async def test_get_one_group_cursor_advances( self, queue: str,