grpc

daemons/base.py

@@ -1,3 +1,18 @@
+import os
+import grpc
+import tasks_pb2_grpc
+
+
+stage = os.getenv("STAGE", 'local')
+if stage == 'local':
+    QUEUES_URL = 'localhost:50051'
+else:
+    QUEUES_URL = 'queues-grpc:50051'
+
+
 class Daemon:
+    def __init__(self):
+        self.channel = grpc.insecure_channel(QUEUES_URL)
+        self.stub = tasks_pb2_grpc.TasksStub(channel=self.channel)     
     def execute(self):
         raise NotImplemented

daemons/mailbox.py

@@ -23,12 +23,6 @@ class Daemon(base.Daemon, queues.TasksHandlerMixin):
     def queue_name(self):
         return 'botalka_mailbox'
     
-    def before_execute(self, task: dict):
-        locks.acquire(task['id'])
-
-    def after_execute(self, task: dict):
-        locks.release(task['id'])
-    
     def process(self, payload: dict):
         message = Message.model_validate(payload)
         bot = platform.platform_client.get_config('bots')[message.project][message.name]

daemons/poll.py

@@ -47,7 +47,7 @@ class Daemon(base.Daemon):
     def start_polling(self, bot: telebot.TeleBot, queue: str) -> threading.Thread:
         @bot.message_handler(content_types=['audio', 'photo', 'voice', 'video', 'document', 'animation', 'text', 'location', 'contact', 'sticker', 'video_note'])
         def do_action(message: telebot.types.Message):
-            queues.set_task(queue, message.json, 1)
+            queues.set_task(self.stub, queue, message.json, 1)
         thread = threading.Thread(target=bot.polling)
         thread.start()
         return thread