From 73d1771205295083c6b7ea0ca8d8322402b0c19e Mon Sep 17 00:00:00 2001 From: Will Date: Thu, 1 Feb 2024 14:07:47 -0800 Subject: [PATCH] Restricted two input queues to 1 doc at a time -Restricted input queue for NLP container to 1 doc at a time -Restricted input queue for file ingest in enrichment to 1 doc at a time --- cmd/enrichment/enrichment/containers.py | 2 +- cmd/nlp/nlp/app.py | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) diff --git a/cmd/enrichment/enrichment/containers.py b/cmd/enrichment/enrichment/containers.py index e66b4d1..a166234 100644 --- a/cmd/enrichment/enrichment/containers.py +++ b/cmd/enrichment/enrichment/containers.py @@ -109,7 +109,7 @@ class Container(containers.DeclarativeContainer): # Format: inputq__ # inputq_alert_slackwebhookalert = providers.Resource(create_consumer, config.rabbitmq_connection_uri, constants.Q_ALERT, pb.Alert, "slackwebhookalert") - inputq_filedata_fileprocessor = providers.Resource(create_consumer, config.rabbitmq_connection_uri, constants.Q_FILE_DATA, pb.FileDataIngestionMessage, "fileprocessor") + inputq_filedata_fileprocessor = providers.Resource(create_consumer, config.rabbitmq_connection_uri, constants.Q_FILE_DATA, pb.FileDataIngestionMessage, "fileprocessor", 1) # limit to 1 file at a time inputq_filedataenriched_fileprocessor = providers.Resource(create_consumer, config.rabbitmq_connection_uri, constants.Q_FILE_DATA_ENRICHED, pb.FileDataEnrichedMessage, "fileprocessor") inputq_process_processcategorizer = providers.Resource(create_consumer, config.rabbitmq_connection_uri, constants.Q_PROCESS, pb.ProcessIngestionMessage, "processcategorizer") inputq_service_servicecategorizer = providers.Resource(create_consumer, config.rabbitmq_connection_uri, constants.Q_SERVICE, pb.ServiceIngestionMessage, "servicecategorizer") diff --git a/cmd/nlp/nlp/app.py b/cmd/nlp/nlp/app.py index c8cd925..6ccb56c 100644 --- a/cmd/nlp/nlp/app.py +++ b/cmd/nlp/nlp/app.py @@ -91,7 +91,7 @@ class App: async def start_indexing_service(self) -> None: async with ( await NemesisRabbitMQConsumer.create( - self.cfg.rabbitmq_connection_uri, constants.Q_FILE_DATA_PLAINTEXT, pb.FileDataPlaintextMessage, "indexingservice" + self.cfg.rabbitmq_connection_uri, constants.Q_FILE_DATA_PLAINTEXT, pb.FileDataPlaintextMessage, "indexingservice", 1 ) as textQ, ): service = IndexingService(textQ, self.cfg, self.storage)