pipe-io-server/pipe_io_server.py

109 lines
3.1 KiB
Python
Raw Normal View History

2018-12-13 21:06:13 +00:00
from threading import Thread
from queue import Queue
from os import mkfifo, remove, getpid, kill
2018-12-13 21:06:13 +00:00
from os.path import exists, join
from signal import SIGTERM
2018-12-13 21:06:13 +00:00
from secrets import token_urlsafe
2018-12-13 21:44:46 +00:00
from collections import namedtuple
from abc import ABC, abstractmethod
from traceback import print_exc
2018-12-13 21:06:13 +00:00
class PipeWriterThread(Thread):
2018-12-13 21:44:46 +00:00
def __init__(self, pipe_path):
2018-12-13 21:06:13 +00:00
super().__init__()
self._pipe_path = pipe_path
2018-12-13 21:44:46 +00:00
self._write_queue = Queue()
def write(self, message):
self._write_queue.put(message, block=True)
2018-12-13 21:06:13 +00:00
def run(self):
while True:
2018-12-13 21:44:46 +00:00
message = self._write_queue.get(block=True)
2018-12-13 21:06:13 +00:00
if message is None:
break
with open(self._pipe_path, 'wb') as pipe:
pipe.write(message)
def stop(self):
2018-12-13 21:44:46 +00:00
self._write_queue.put(None)
2018-12-13 21:06:13 +00:00
self.join()
class PipeReaderThread(Thread):
_stop_sequence = b'stop_reading'
2018-12-13 21:44:46 +00:00
def __init__(self, pipe_path, message_handler):
2018-12-13 21:06:13 +00:00
super().__init__()
2018-12-13 21:44:46 +00:00
self._message_handler = message_handler
2018-12-13 21:06:13 +00:00
self._pipe_path = pipe_path
def run(self):
while True:
with open(self._pipe_path, 'rb') as pipe:
message = pipe.read()
if message == self._stop_sequence:
break
try:
self._message_handler(message)
2018-12-14 16:25:54 +00:00
except: # pylint: disable=bare-except
print_exc()
kill(getpid(), SIGTERM)
2018-12-13 21:06:13 +00:00
def stop(self):
with open(self._pipe_path, 'wb') as pipe:
pipe.write(self._stop_sequence)
self.join()
class PipeHandler:
def __init__(self, *pipe_paths):
self._pipe_paths = pipe_paths
2018-12-13 21:06:13 +00:00
def recreate(self):
self.remove()
for pipe_path in self._pipe_paths:
mkfifo(pipe_path)
2018-12-13 21:06:13 +00:00
def remove(self):
for pipe_path in self._pipe_paths:
if exists(pipe_path):
remove(pipe_path)
2018-12-13 21:06:13 +00:00
class PipeIOServer(ABC):
2018-12-13 21:06:13 +00:00
def __init__(self, in_pipe=None, out_pipe=None):
self.in_pipe, self.out_pipe = in_pipe, out_pipe
self._create_pipes()
2018-12-13 21:44:46 +00:00
io_threads_dict = {
'reader': PipeReaderThread(self.in_pipe, self._handle_message),
'writer': PipeWriterThread(self.out_pipe)
2018-12-13 21:06:13 +00:00
}
2018-12-13 21:44:46 +00:00
IOThreadsTuple = namedtuple('IOThreadsTuple', sorted(io_threads_dict.keys()))
self._io_threads = IOThreadsTuple(**io_threads_dict)
2018-12-13 21:06:13 +00:00
def _create_pipes(self):
if not self.in_pipe or not self.out_pipe:
pipe_id = token_urlsafe(6)
self.in_pipe = join('/tmp', f'in_pipe_{pipe_id}')
self.out_pipe = join('/tmp', f'out_pipe_{pipe_id}')
PipeHandler(self.in_pipe, self.out_pipe).recreate()
2018-12-13 21:06:13 +00:00
@abstractmethod
2018-12-13 21:44:46 +00:00
def _handle_message(self, message):
raise NotImplementedError()
def send(self, message):
2018-12-13 21:44:46 +00:00
self._io_threads.writer.write(message)
2018-12-13 21:06:13 +00:00
def run(self):
2018-12-13 21:44:46 +00:00
for thread in self._io_threads:
2018-12-13 21:06:13 +00:00
thread.start()
def stop(self):
2018-12-13 21:44:46 +00:00
for thread in self._io_threads:
2018-12-13 21:06:13 +00:00
thread.stop()
PipeHandler(self.in_pipe, self.out_pipe).remove()