Rework buffer for providers

src/transport/msg_providers/common.py

@@ -1,6 +1,5 @@
 import abc
 import logging
-from collections import deque
 from typing import Any, List, Optional
 
 from schema import Schema, SchemaError
@@ -15,30 +14,15 @@ class BaseMessageProvider(abc.ABC):
 
     def __init__(self, message_schema: Schema):
         self.message_schema = message_schema
-        self._queue: deque[Any] = deque([])
 
     def get_messages(self) -> List[dict]:
-        messages = []
+        """
+        Fetches new messages, processes them, and filters out only the valid ones.
 
-        for _ in range(self.MAX_MESSAGES_RECEIVE):
-            msg = self._fetch_message()
-
-            if msg is None:
-                break
-
-            value = self._process_msg(msg)
-
-            if value and self._is_valid(value):
-                messages.append(value)
-
-        return messages
-
-    def _fetch_message(self) -> Optional[Any]:
-        if not self._queue:
-            messages = self._fetch_messages()
-            if messages:
-                self._queue.extend(messages)
-        return None if not self._queue else self._queue.popleft()
+        Returns:
+            List[Dict]: A list of processed and valid messages.
+        """
+        return [msg for msg in (self._process_msg(m) for m in self._fetch_messages()) if msg and self._is_valid(msg)]
 
     @abc.abstractmethod
     def _fetch_messages(self) -> List[Any]:

src/transport/msg_providers/kafka.py

@@ -51,7 +51,7 @@ def __init__(self, message_schema: Schema, client: str):
     def __del__(self):
         self.kafka.close()
 
-    def _receive_message(self) -> Optional[dict]:
+    def _receive_message(self) -> Optional[str]:
         msg = self.kafka.poll(timeout=1)
         if msg is None:
             return None
@@ -63,8 +63,15 @@ def _receive_message(self) -> Optional[dict]:
         return msg.value()
 
     def _fetch_messages(self) -> List[Any]:
-        msg = self._receive_message()
-        return [] if msg is None else [msg]
+        messages = []
+
+        for _ in range(self.MAX_MESSAGES_RECEIVE):
+            msg = self._receive_message()
+            if msg is None:
+                break
+            messages.append(msg)
+
+        return messages
 
     def _process_msg(self, msg: str) -> Optional[dict]:
         try:

src/transport/msg_providers/rabbit.py

@@ -74,6 +74,17 @@ def _recreate_client(self):
     def __del__(self):
         self.client.disconnect()
 
+    def _fetch_messages(self) -> List[Any]:
+        messages = []
+
+        for _ in range(self.MAX_MESSAGES_RECEIVE):
+            msg = self._receive_message()
+            if msg is None:
+                break
+            messages.append(msg)
+
+        return messages
+
     def _receive_message(self) -> Optional[dict]:
         if not self.connection:
             raise ConnectionError('Connection RabbitMQ was lost.')
@@ -83,10 +94,6 @@ def _receive_message(self) -> Optional[dict]:
         except IndexError:
             return None
 
-    def _fetch_messages(self) -> List[Any]:
-        msg = self._receive_message()
-        return [] if msg is None else [msg]
-
     def _receive_message_from_queue(self, body):
         self._queue.append(body)