baseimage-tutorial-framework/lib/tfw/components/pipe_io_event_handler.py

54 lines
1.5 KiB
Python

from abc import abstractmethod
from json import loads, dumps, JSONDecodeError
from tfw import EventHandlerBase
from tfw.config.logs import logging
from .pipe_io_server import PipeIOServer
LOG = logging.getLogger(__name__)
class PipeIOEventHandlerBase(EventHandlerBase):
def __init__(self, key, in_pipe_path, out_pipe_path, permissions=0o600):
super().__init__(key)
self.pipe_io = CallbackPipeIOServer(
in_pipe_path,
out_pipe_path,
self.handle_pipe_event,
permissions
)
self.pipe_io.start()
@abstractmethod
def handle_pipe_event(self, message):
raise NotImplementedError()
def cleanup(self):
self.pipe_io.stop()
class CallbackPipeIOServer(PipeIOServer):
def __init__(self, in_pipe_path, out_pipe_path, callback, permissions):
super().__init__(in_pipe_path, out_pipe_path, permissions)
self.callback = callback
def handle_message(self, message):
self.callback(message)
class PipeIOEventHandler(PipeIOEventHandlerBase):
def handle_event(self, message):
try:
json_bytes = dumps(message).encode()
self.pipe_io.send_message(json_bytes)
except TypeError:
LOG.error("Message %s not JSON serializable! Ignoring...", message)
def handle_pipe_event(self, message):
try:
json = loads(message)
self.server_connector.send(json)
except JSONDecodeError:
LOG.error("Invalid JSON received on %s! Ignoring...", self.pipe_io.in_pipe)