mrtoth
/
pipe-io-server
1
0
Fork 0
Code Issues Pull Requests Releases Wiki Activity

Make PipeIOServer usable as a base class

This commit is contained in:
Kristóf Tóth 2018-12-13 23:00:54 +01:00
parent 43578cd3a0
commit 88ef32ede3
1 changed files with 12 additions and 2 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

14
pipe_io_server.py
View File

@ -5,6 +5,7 @@ from os.path import exists, join
from signal import signal, SIGTERM, SIGINT
from secrets import token_urlsafe
from collections import namedtuple
from abc import ABC, abstractmethod
class PipeWriterThread(Thread):
@ -66,7 +67,7 @@ class PipeHandler:
remove(pipe_path)
class PipeIOServer:
class PipeIOServer(ABC):
def __init__(self, in_pipe=None, out_pipe=None):
self.in_pipe, self.out_pipe = in_pipe, out_pipe
self._create_pipes()
@ -85,7 +86,11 @@ class PipeIOServer:
self.out_pipe = join('/tmp', f'out_pipe_{pipe_id}')
PipeHandler(self.in_pipe, self.out_pipe).recreate()
@abstractmethod
def _handle_message(self, message):
raise NotImplementedError()
def send(self, message):
self._io_threads.writer.write(message)
def run(self):
@ -98,8 +103,13 @@ class PipeIOServer:
PipeHandler(self.in_pipe, self.out_pipe).remove()
class EchoPipeIOServer(PipeIOServer):
def _handle_message(self, message):
self.send(message)
if __name__ == "__main__":
pipe_io = PipeIOServer()
pipe_io = EchoPipeIOServer()
signal(SIGTERM, lambda a, b: pipe_io.stop())
signal(SIGINT, lambda a, b: pipe_io.stop())
pipe_io.run()