From cb30b810d8cf5277a5077479333f0ed586fa60fc Mon Sep 17 00:00:00 2001
From: NeonKirill <74428618+NeonKirill@users.noreply.github.com>
Date: Wed, 4 Dec 2024 00:39:04 +0100
Subject: [PATCH] [FEAT] Async Consumer-based Klatchat Observer (#109)

* Enabled async consumers for Chatbots Observer and updated neon mq connector requirements
---
 requirements/requirements.txt            | 2 +-
 requirements/services/observer.txt       | 2 +-
 services/klatchat_observer/controller.py | 8 ++++----
 3 files changed, 6 insertions(+), 6 deletions(-)

diff --git a/requirements/requirements.txt b/requirements/requirements.txt
index 749e3073..9905fba2 100644
--- a/requirements/requirements.txt
+++ b/requirements/requirements.txt
@@ -7,7 +7,7 @@ httpx==0.27.0  # required by FastAPI
 Jinja2==3.1.3
 jsbeautifier==1.15.1
 kubernetes==29.0.0
-neon-mq-connector~=0.7
+neon-mq-connector==0.7.2a5
 neon-sftp~=0.1
 neon_utils[sentry]==1.11.1a5
 pre-commit==3.7.0
diff --git a/requirements/services/observer.txt b/requirements/services/observer.txt
index b27bcee3..d75f16a4 100644
--- a/requirements/services/observer.txt
+++ b/requirements/services/observer.txt
@@ -1,3 +1,3 @@
-neon-mq-connector~=0.7
+neon-mq-connector==0.7.2a5
 websocket-client==1.7.0
 websockets==12.0
diff --git a/services/klatchat_observer/controller.py b/services/klatchat_observer/controller.py
index ef959c1c..2a1d1e4c 100644
--- a/services/klatchat_observer/controller.py
+++ b/services/klatchat_observer/controller.py
@@ -58,6 +58,8 @@ class Recipients(Enum):
 class ChatObserver(MQConnector):
     """Observer of conversations states"""
 
+    async_consumers_enabled = True
+
     recipient_prefixes = {
         Recipients.NEON: ["neon"],
         Recipients.UNRESOLVED: ["undefined"],
@@ -509,9 +511,7 @@ def handle_message(self, data: dict):
                 or Recipients.UNRESOLVED
             )
             handler_method = self.recipient_to_handler_method.get(recipient)
-            if not handler_method:
-                LOG.warning(f"Failed to get handler message for recipient={recipient}")
-            else:
+            if handler_method:
                 handler_method(recipient_data=recipient_data, msg_data=data)
         else:
             raise TypeError(f"Malformed data received: {data}")
@@ -569,7 +569,7 @@ def send_translation_response(self, data: dict):
         if request_id and self.__translation_requests.pop(request_id, None):
             self.sio.emit("get_neon_translations", data=data)
         else:
-            LOG.warning(
+            LOG.debug(
                 f"Neon translation response was not sent, "
                 f"as request_id={request_id} was not found among translation requests"
             )