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

Refactor PipeIOServer.__init__()

Browse Source
This commit is contained in:
Kristóf Tóth 2018-12-16 19:04:51 +01:00
parent 5091380133
commit 09abb0331d
1 changed files with 10 additions and 7 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

17
pipe_io_server/pipe_io_server.py
View File

@ -26,14 +26,9 @@ class PipeHandler:
class PipeIOServer(ABC):
def __init__(self, in_pipe=None, out_pipe=None):
self.in_pipe, self.out_pipe = in_pipe, out_pipe
self._io_threads = None
self._create_pipes()
io_threads_dict = {
'reader': PipeReaderThread(self.in_pipe, self._handle_message),
'writer': PipeWriterThread(self.out_pipe)
}
IOThreadsTuple = namedtuple('IOThreadsTuple', sorted(io_threads_dict.keys()))
self._io_threads = IOThreadsTuple(**io_threads_dict)
self._init_io_threads()
def _create_pipes(self):
if not self.in_pipe or not self.out_pipe:
@ -42,6 +37,14 @@ class PipeIOServer(ABC):
self.out_pipe = join('/tmp', f'out_pipe_{pipe_id}')
PipeHandler(self.in_pipe, self.out_pipe).recreate()
def _init_io_threads(self):
io_threads_dict = {
'reader': PipeReaderThread(self.in_pipe, self._handle_message),
'writer': PipeWriterThread(self.out_pipe)
}
IOThreadsTuple = namedtuple('IOThreadsTuple', sorted(io_threads_dict.keys()))
self._io_threads = IOThreadsTuple(**io_threads_dict)
@abstractmethod
def _handle_message(self, message):
raise NotImplementedError()