baseimage-tutorial-framework/tfw/components/pipe_io/pipe_io_handler.py

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)