Distributed automation workers

dtable_events/app/app.py

@@ -43,10 +43,13 @@ def __init__(self, config, seafile_config, task_mode):
 
         self._stats_sender = StatsSender(config)
 
-        self._playwright_manager = get_playwright_manager()
+        # automations pipeline, to put test tasks to redis for foreground mode and to put real tasks to redis for background mode
+        # but not necessary to start in foreground mode
+        self._automations_pipeline = AutomationsPipeline(config)
 
         if self._enable_foreground_tasks:
             self._dtable_io_server = DTableIOServer(self, config)
+            self._playwright_manager = get_playwright_manager()
 
         if self._enable_background_tasks:
             # redis client subscriber
@@ -80,14 +83,12 @@ def __init__(self, config, seafile_config, task_mode):
             self._virus_scanner = VirusScanner(config, seafile_config)
             # ai stats, listen redis and cron
             self.ai_stats_worker = AIStatsWorker(config)
-            # automations pipeline
-            self._automations_pipeline = AutomationsPipeline(config)
 
     def serve_forever(self):
 
-        self._playwright_manager.start()
 
         if self._enable_foreground_tasks:
+            self._playwright_manager.start()                 # always True
             self._dtable_io_server.start()                   # always True
 
         if self._enable_background_tasks:

dtable_events/app/event_redis.py

@@ -81,6 +81,15 @@ def delete(self, key):
     def publish(self, channel_name, message):
         return self.connection.publish(channel_name, message)
 
+    def lpush(self, key, value):
+        return self.connection.lpush(key, value)
+
+    def rpop(self, key):
+        return self.connection.rpop(key)
+
+    def llen(self, key):
+        return self.connection.llen(key)
+
 
 class RedisCache(object):
     def __init__(self):