pipe-io-server/pipe_io_server/pipe_writer_thread.py

from threading import Thread
from queue import Queue

from .terminate_process_on_failure import terminate_process_on_failure


class PipeWriterThread(Thread):
    def __init__(self, pipe_path, stop_event):
        super().__init__()
        self._pipe_path = pipe_path
        self._stop_event = stop_event
        self._write_queue = Queue()

    def write(self, message):
        self._write_queue.put(message, block=True)

    @terminate_process_on_failure
    def run(self):
        try:
            with open(self._pipe_path, 'w') as pipe:
                while True:
                    message = self._write_queue.get(block=True)
                    if message is None:
                        self._stop_event.set()
                        break
                    pipe.write(f'{message}\n')
                    pipe.flush()
        except BrokenPipeError:
            self._stop_event.set()

    def stop(self):
        self.unblock()
        self.join()

    def unblock(self):
        self._write_queue.put(None)
        open(self._pipe_path, 'r').close()