Merge branch 'develop' into feat/package_version

Author: s3rius

poetry.lock

@@ -20,7 +20,7 @@ keywords = ["taskiq", "tasks", "distributed", "async", "aio-pika"]
 
 [tool.poetry.dependencies]
 python = "^3.7"
-taskiq = "^0"
+taskiq = ">=0.6.0,<1"
 aio-pika = "^9.0"
 importlib-metadata = {version = "^4.0.0", python = "<3.8"}
 

pyproject.toml

@@ -5,9 +5,7 @@
 
 from aio_pika import DeliveryMode, ExchangeType, Message, connect_robust
 from aio_pika.abc import AbstractChannel, AbstractQueue, AbstractRobustConnection
-from taskiq.abc.broker import AsyncBroker
-from taskiq.abc.result_backend import AsyncResultBackend
-from taskiq.message import BrokerMessage
+from taskiq import AckableMessage, AsyncBroker, AsyncResultBackend, BrokerMessage
 
 _T = TypeVar("_T")  # noqa: WPS111
 
@@ -267,7 +265,7 @@ async def kick(self, message: BrokerMessage) -> None:
                 routing_key=self._delay_queue_name,
             )
 
-    async def listen(self) -> AsyncGenerator[bytes, None]:
+    async def listen(self) -> AsyncGenerator[AckableMessage, None]:
         """
         Listen to queue.
 
@@ -283,5 +281,8 @@ async def listen(self) -> AsyncGenerator[bytes, None]:
         queue = await self.declare_queues(self.read_channel)
         async with queue.iterator() as iterator:
             async for message in iterator:
-                async with message.process():
-                    yield message.body
+                yield AckableMessage(
+                    data=message.body,
+                    ack=message.ack,
+                    reject=message.reject,
+                )

taskiq_aio_pika/broker.py

@@ -4,12 +4,13 @@
 import pytest
 from aio_pika import Channel, Message
 from aio_pika.exceptions import QueueEmpty
-from taskiq import BrokerMessage
+from taskiq import AckableMessage, BrokerMessage
+from taskiq.utils import maybe_awaitable
 
 from taskiq_aio_pika.broker import AioPikaBroker
 
 
-async def get_first_task(broker: AioPikaBroker) -> bytes:  # type: ignore
+async def get_first_task(broker: AioPikaBroker) -> AckableMessage:  # type: ignore
     """
     Get first message from the queue.
 
@@ -46,7 +47,8 @@ async def test_kick_success(broker: AioPikaBroker) -> None:
 
     message = await asyncio.wait_for(get_first_task(broker), timeout=0.4)
 
-    assert message == sent.message
+    assert message.data == sent.message
+    await maybe_awaitable(message.ack())
 
 
 @pytest.mark.anyio
@@ -111,7 +113,8 @@ async def test_listen(
 
     message = await asyncio.wait_for(get_first_task(broker), timeout=0.4)
 
-    assert message == b"test_message"
+    assert message.data == b"test_message"
+    await maybe_awaitable(message.ack())
 
 
 @pytest.mark.anyio
@@ -133,9 +136,10 @@ async def test_wrong_format(
         routing_key=queue_name,
     )
 
-    msg_bytes = await asyncio.wait_for(get_first_task(broker), 0.4)
+    message = await asyncio.wait_for(get_first_task(broker), 0.4)
 
-    assert msg_bytes == b"wrong"
+    assert message.data == b"wrong"
+    await maybe_awaitable(message.ack())
 
     with pytest.raises(QueueEmpty):
         await queue.get()