78 lines
2.2 KiB
Python
78 lines
2.2 KiB
Python
from typing import Dict
|
|
|
|
from .abstract_plugin import AbstractCommand, AbstractPlugin, AbstractCommandCompiler
|
|
import logging
|
|
import time
|
|
from config import Config
|
|
|
|
import redis
|
|
from pyprocsync import ProcSync
|
|
|
|
|
|
class SyncCommand(AbstractCommand):
|
|
|
|
def __init__(self, logger: logging.Logger, procsync_instance: ProcSync, name: str, nodes: int):
|
|
|
|
if type(nodes) not in [int]:
|
|
raise ValueError("Nodes must be int")
|
|
|
|
if nodes <= 1:
|
|
raise ValueError("Nodes must be greater than 1")
|
|
|
|
self._logger = logger
|
|
self._procsync_instance = procsync_instance
|
|
self._name = name
|
|
self._nodes = nodes
|
|
|
|
def execute(self):
|
|
self._logger.debug(f"Synchronizing on {self._name} with {self._nodes} nodes...")
|
|
self._procsync_instance.sync(self._name, self._nodes, Config.SYNC_TIMEOUT)
|
|
self._logger.debug(f"Event {self._name} synchronized!")
|
|
|
|
def abort(self):
|
|
# TODO
|
|
pass
|
|
|
|
def describe(self) -> dict:
|
|
return {
|
|
"command": "sync",
|
|
"params": {
|
|
"name": self._name,
|
|
"nodes": self._nodes,
|
|
},
|
|
"config": {
|
|
"SYNC_TIMEOUT": Config.SYNC_TIMEOUT,
|
|
"RUN_ID": Config.RUN_ID
|
|
}
|
|
}
|
|
|
|
|
|
class SyncCompiler(AbstractCommandCompiler):
|
|
|
|
def __init__(self, logger: logging.Logger, procsync_instance: ProcSync):
|
|
self._logger = logger
|
|
self._procsync_instance = procsync_instance
|
|
|
|
def compile(self, name: str, nodes: int) -> AbstractCommand:
|
|
return SyncCommand(self._logger, self._procsync_instance, name, nodes)
|
|
|
|
|
|
class SyncPlugin(AbstractPlugin):
|
|
plugin_name = "sync"
|
|
|
|
def __init__(self):
|
|
self._logger = logging.getLogger("plugin").getChild("sync")
|
|
self._procsync_instance = ProcSync(
|
|
redis_client=redis.from_url(Config.REDIS_URL),
|
|
run_id=Config.RUN_ID,
|
|
delay=Config.SYNC_DELAY
|
|
)
|
|
|
|
def load_compilers(self) -> Dict[str, AbstractCommandCompiler]:
|
|
return {
|
|
"sync": SyncCompiler(self._logger, self._procsync_instance)
|
|
}
|
|
|
|
def close(self):
|
|
self._procsync_instance.close()
|