import logging from abc import abstractmethod from json import loads, dumps from pipe_io_server import PipeIOServer LOG = logging.getLogger(__name__) DEFAULT_PERMISSIONS = 0o600 class PipeIOHandlerBase: keys = [''] def __init__(self, in_pipe_path, out_pipe_path, permissions=DEFAULT_PERMISSIONS): self.connector = None self.in_pipe = in_pipe_path self.out_pipe = out_pipe_path self.pipe_io = PipeIOServer( in_pipe_path, out_pipe_path, permissions ) self.pipe_io.handle_message = self._server_handle_message self.pipe_io.start() def _server_handle_message(self, message): try: self.handle_pipe_event(message) except: # pylint: disable=bare-except LOG.exception('Failed to handle message %s from pipe %s!', message, self.in_pipe) @abstractmethod def handle_pipe_event(self, message_bytes): raise NotImplementedError() def cleanup(self): self.pipe_io.stop() class PipeIOHandler(PipeIOHandlerBase): def handle_event(self, message, _): json_bytes = dumps(message).encode() self.pipe_io.send_message(json_bytes) def handle_pipe_event(self, message_bytes): json = loads(message_bytes) self.connector.send_message(json)