- Fixed issue with empty queues.

issue #283

Author: albamig

packages/plugins/minos-broker-rabbitmq/minos/plugins/rabbitmq/subscriber.py

@@ -13,6 +13,7 @@
 from aio_pika import (
     connect,
 )
+from aio_pika.exceptions import QueueEmpty
 
 from minos.common import (
     MinosConfig, Config,
@@ -55,7 +56,7 @@ def _from_config(cls, config: MinosConfig, **kwargs) -> RabbitMQBrokerSubscriber
 
     async def _setup(self) -> None:
         await super()._setup()
-        self.connection = await connect(f"amqp://guest:guest@{self.broker_host}/")
+        self.connection = await connect(f"amqp://guest:guest@{self.broker_host}:{self.broker_port}/")
 
     async def _destroy(self) -> None:
         await self.connection.close()
@@ -65,8 +66,11 @@ async def _receive(self) -> BrokerMessage:
             async with self.connection:
                 channel = await self.connection.channel()
                 queue = await channel.declare_queue(topic)
-                message = await queue.get(fail=False)
-                return BrokerMessage.from_avro_bytes(message.body if message.body else None)
+                try:
+                    message = await queue.get()
+                    return BrokerMessage.from_avro_bytes(message.body)
+                except QueueEmpty:
+                    pass
 
 
 class RabbitMQBrokerSubscriberBuilder(BrokerSubscriberBuilder):

packages/plugins/minos-broker-rabbitmq/tests/test_rabbitmq/test_integration.py

@@ -17,7 +17,7 @@
 
 
 class IntegrationTests(IsolatedAsyncioTestCase):
-    async def test_integration(self):
+    async def test_one_topic(self):
         message = BrokerMessageV1("foo", BrokerMessageV1Payload("bar"))
 
         async with RabbitMQBrokerPublisher.from_config(CONFIG_FILE_PATH) as publisher:
@@ -27,3 +27,8 @@ async def test_integration(self):
             observed = await subscriber.receive()
 
         self.assertEqual(message.content, observed.content)
+
+    async def test_empty_topic(self):
+        async with RabbitMQBrokerSubscriber.from_config(CONFIG_FILE_PATH, topics={"empty_topic"}) as subscriber:
+            observed = await subscriber.receive()
+