#15 - Implement Backpressure UPDATE & IDLE, setup AMQ handler in its own thread

amqp/service/amq_message_handler.py

@@ -63,6 +63,7 @@ class DataMessageHandler:
         rabbit_mq_client: RabbitMQClient,
         amq_configuration: AMQConfiguration,
         loop,
+        backpressure_handler: BackpressureHandler,
         file_handler: FileHandler = StreamingFileHandler(),
     ):
         self.service_adapter: CleverThisServiceAdapter = service_adapter
@@ -82,9 +83,7 @@ class DataMessageHandler:
             amq_configuration.backpressure.time_window / 1000
         )  # convert to seconds
         self.file_handler: FileHandler = file_handler
-        self.backpressure_handler = BackpressureHandler(
-            channel=rabbit_mq_client.channel, loop=loop
-        )
+        self.backpressure_handler: BackpressureHandler = backpressure_handler
         _backpressure_treshold = amq_configuration.backpressure.threshold
         """
         Determine the number of parallel workers based on the backpressure threshold and PARALLEL_WORKERS env var.
@@ -143,19 +142,15 @@ class DataMessageHandler:
         :return: DataResponse
         """
         # Increment the counter for parallel executions
-        BackpressureHandler.current_parallel_executions += 1
+        self.backpressure_handler.increase_parallel_executions()
         logging_info(
-            f"PARALLEL: Current Capacity [{BackpressureHandler.current_parallel_executions}] of [{BackpressureHandler.parallel_workers}]"
+            f"PARALLEL: Current Capacity [{self.backpressure_handler.current_parallel_executions}] of [{self.backpressure_handler.parallel_workers}]"
         )
         if message.reply_to:
             self.reply_to = message.reply_to
-        if (
-            BackpressureHandler.current_parallel_executions
-            > BackpressureHandler.parallel_workers - 2
-        ):
-            await self.backpressure_handler.handle_backpressure_overload_event()
 
         try:
+            await self.backpressure_handler.check_overload_condition()
             _a_message = AMQMessage(amq_message, self.rabbit_mq_client.connection)
             _response: DataResponse = await self.service_adapter.on_message(_a_message)
             logging_debug(f"Result in thread: {_response}")
@@ -175,11 +170,10 @@ class DataMessageHandler:
                         os.remove(filename)
                     except OSError as e:
                         logging_error(f"Deleting file {filename}: {e}")
-            BackpressureHandler.current_parallel_executions -= 1
-            return
         except Exception as e:
             logging_error(f"Request handler: {e}")
-            return
+
+        self.backpressure_handler.decrease_parallel_executions()
 
     async def reconstitute_data_message(
         self, message: AbstractIncomingMessage

amqp/service/amq_service.py

@@ -3,9 +3,12 @@ import logging.config
 import os
 import asyncio
 from asyncio import Future
+from threading import Thread
+from typing import Optional
 
 from aio_pika.abc import AbstractRobustExchange
 
+from amqp.adapter.backpressure_handler import BackpressureHandler
 from amqp.adapter.data_message_factory import DataMessageFactory
 from amqp.adapter.amq_route_factory import AMQRouteFactory
 from amqp.adapter.logging_utils import logging_info, logging_warning
@@ -33,13 +36,14 @@ class AMQService:
     CLEVER_AMQP_CLIENT_VERSION = os.getenv("CLEVER_AMQP_CLIENT_VERSION")
     MAX_WAIT = 300  # defaults to 5 minutes (300 seconds)
 
-    def __init__(self, amq_configuration: AMQConfiguration, service_adapter):
+    def __init__(self, amq_configuration: AMQConfiguration, service_adapter=None):
         logging.info("***********************************************")
         logging.info("**********  AMQ   Service   *******************")
         logging.info("***********************************************")
-        self.loop = asyncio.get_event_loop()
+        self.loop = asyncio.new_event_loop()
 
         self.amq_configuration = amq_configuration
+        self.service_adapter = service_adapter
         self.data_message_factory = DataMessageFactory.get_instance(
             self.amq_configuration.amq_adapter.generator_id
         )
@@ -49,8 +53,12 @@ class AMQService:
         self.service_message_factory = ServiceMessageFactory(
             self.amq_configuration.amq_adapter.service_name
         )
+        self.backpressure_thread: Optional[Thread] = None
+
+    def run(self):
         # self.init(loop) will initialize RabbitMQ connection
-        self.loop.run_until_complete(self.init(self.loop, service_adapter))
+        asyncio.set_event_loop(self.loop)
+        self.loop.run_until_complete(self.init(self.loop, self.service_adapter))
 
         logging.info("***********************************************")
         logging.info("******  AMQ   Service  Init  Complete *********")
@@ -61,6 +69,9 @@ class AMQService:
         _reply_to_exchange: AbstractRobustExchange = await self.rabbit_mq_client.init(
             loop
         )
+        backpressure_handler = BackpressureHandler(
+            channel=self.rabbit_mq_client.channel, loop=loop
+        )
         self.amq_data_message_handler = DataMessageHandler(
             service_adapter,
             self.tracer,
@@ -70,9 +81,11 @@ class AMQService:
             self.rabbit_mq_client,
             self.amq_configuration,
             loop=loop,
+            backpressure_handler=backpressure_handler,
         )
         await self.register_routes()
         await self.rabbit_mq_client.request_routes()
+        self.backpressure_thread = backpressure_handler.start_backpressure_monitor()
 
     def cleanup(self):
         logging_info("AMQService.cleanup() rmqc=%s", self.rabbit_mq_client)