From 23f6df7b07c3c2bdc765ea7cb2a3f5033da576e3 Mon Sep 17 00:00:00 2001
From: DESPRES Damien <ddespres@neogeo.fr>
Date: Thu, 14 Oct 2021 17:37:34 +0200
Subject: [PATCH] check next queue size
---
workers/doc_enricher.py | 2 ++
workers/doc_processor.py | 2 ++
2 files changed, 4 insertions(+)
diff --git a/workers/doc_enricher.py b/workers/doc_enricher.py
index ab2000d..5f1b18b 100644
--- a/workers/doc_enricher.py
+++ b/workers/doc_enricher.py
@@ -183,7 +183,9 @@ def enrich_docs( channel, method, properties, body ):
while q_len > 50:
logging.info('Waiting for next queue to process elements...')
time.sleep(5)
+ dest_queue = channel.queue_declare(queue=doc_pages_to_process_qn, durable=True, arguments={'x-message-ttl' : cfg['rabbitmq']['ttl']}, passive=True)
q_len = dest_queue.method.message_count
+ logging.info('Queue Size is %i for %s...' % ( q_len, doc_pages_to_process_qn))
channel.basic_publish( exchange=exchange,
routing_key=doc_pages_to_process_rk,
diff --git a/workers/doc_processor.py b/workers/doc_processor.py
index dec3f76..df73a8d 100644
--- a/workers/doc_processor.py
+++ b/workers/doc_processor.py
@@ -172,7 +172,9 @@ def process_docs( channel, method, properties, body ):
while q_len > 50:
logging.info('Waiting for next queue to process elements...')
time.sleep(5)
+ dest_queue = channel.queue_declare(queue=docs_to_index_qn, durable=True, arguments={'x-message-ttl' : cfg['rabbitmq']['ttl']}, passive=True)
q_len = dest_queue.method.message_count
+ logging.info('Queue Size is %i for %s...' % ( q_len, docs_to_index_qn))
channel.basic_publish( exchange=exchange,
routing_key=docs_to_index_rk,
--
GitLab