diff --git a/workers/doc_enricher.py b/workers/doc_enricher.py index ab2000dd1c0af63acb39c56e2095cc1a070a1813..5f1b18be506f6d70dc00e2009402db5a6aaf2486 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 dec3f762f36f2ec4019143d7621d9b7af892baeb..df73a8d1ecdbdfde11b29d7f19e925fa07fd40e8 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,