diff --git a/workers/doc_enricher.py b/workers/doc_enricher.py index 1d36785247ff8ed2d0ed1cc56a3213b22c99d1d8..ab2000dd1c0af63acb39c56e2095cc1a070a1813 100644 --- a/workers/doc_enricher.py +++ b/workers/doc_enricher.py @@ -138,7 +138,7 @@ def enrich_docs( channel, method, properties, body ): channel.exchange_declare(exchange=exchange, exchange_type='direct') - channel.queue_declare(queue=doc_pages_to_process_qn, durable=True, arguments={'x-message-ttl' : cfg['rabbitmq']['ttl']}) + dest_queue = channel.queue_declare(queue=doc_pages_to_process_qn, durable=True, arguments={'x-message-ttl' : cfg['rabbitmq']['ttl']}) channel.queue_bind(exchange=exchange, queue=doc_pages_to_process_qn, routing_key=doc_pages_to_process_rk) @@ -178,7 +178,13 @@ def enrich_docs( channel, method, properties, body ): msg['body'] = doc_page the_body = msgpack.packb(msg, use_bin_type=True, default=encode_datetime) - + q_len = dest_queue.method.message_count + logging.info('Queue Size is %i for %s...' % ( q_len, doc_pages_to_process_qn)) + while q_len > 50: + logging.info('Waiting for next queue to process elements...') + time.sleep(5) + q_len = dest_queue.method.message_count + channel.basic_publish( exchange=exchange, routing_key=doc_pages_to_process_rk, body=the_body, diff --git a/workers/doc_processor.py b/workers/doc_processor.py index ffd51d42d1f41f8501cc7b1b46add9b7e1f2f586..dec3f762f36f2ec4019143d7621d9b7af892baeb 100644 --- a/workers/doc_processor.py +++ b/workers/doc_processor.py @@ -169,6 +169,11 @@ def process_docs( channel, method, properties, body ): q_len = dest_queue.method.message_count logging.info('Queue Size is %i for %s...' % ( q_len, docs_to_index_qn)) + while q_len > 50: + logging.info('Waiting for next queue to process elements...') + time.sleep(5) + q_len = dest_queue.method.message_count + channel.basic_publish( exchange=exchange, routing_key=docs_to_index_rk, body=the_body,