Properly declare queues

alephdata · Jul 30, 2024 · 128e0a7 · 128e0a7
1 parent 008806a
commit 128e0a7
Showing 1 changed file with 3 additions and 1 deletion.
diff --git a/servicelayer/taskqueue.py b/servicelayer/taskqueue.py
@@ -705,12 +705,14 @@ def consume_queue(queue):
             for method, properties, body in channel.consume(
                 queue,
                 inactivity_timeout=10,
-                prefetch_count=self.prefetch_count_mapping[queue],
             ):
                 if method:
                     self.on_message(channel, method, properties, body)
 
         for queue in self.queues:
+            declare_rabbitmq_queue(
+                channel, queue, prefetch_count=self.prefetch_count_mapping[queue]
+            )
             thread = threading.Thread(target=consume_queue, args=(queue,))
             thread.daemon = True
             thread.start()