avoid busy wait

run-llama · Jan 7, 2025 · 90832b9 · 90832b9
1 parent 34d46e3
commit 90832b9
Show file tree

Hide file tree

Showing 2 changed files with 3 additions and 2 deletions.
diff --git a/e2e_tests/message_queues/redis/test_message_queue.py b/e2e_tests/message_queues/redis/test_message_queue.py
@@ -20,6 +20,7 @@ def message_handler(message: QueueMessage) -> None:
     )
     start_consuming_callable = await mq.register_consumer(test_consumer, topic="test")
     t = asyncio.create_task(start_consuming_callable())
+    await asyncio.sleep(1.0)
 
     # produce a message
     test_message = QueueMessage(type="test_message", data={"message": "this is a test"})

diff --git a/llama_deploy/message_queues/redis.py b/llama_deploy/message_queues/redis.py
@@ -112,7 +112,7 @@ async def start_consuming_callable() -> None:
                         decoded_message = json.loads(message["data"])
                         queue_message = QueueMessage.model_validate(decoded_message)
                         await consumer.process_message(queue_message)
-                    await asyncio.sleep(0)
+                    await asyncio.sleep(0.01)
             finally:
                 return
 
@@ -165,7 +165,7 @@ async def cleanup_local(
         for consumer_metadata in self._consumers.values():
             if consumer_metadata.pubsub:
                 await consumer_metadata.pubsub.unsubscribe()
-                await consumer_metadata.pubsub.close()
+                await consumer_metadata.pubsub.aclose()
 
         # Clear consumers
         self._consumers = {}