terminals

sandbox/print_terminal.py

@@ -0,0 +1,26 @@
+import time
+import sys
+
+sys.path.append('../')
+
+from synergine2.terminals import Terminal, TerminalPackage, TerminalManager
+
+
+class PrintTerminal(Terminal):
+    def receive(self, package: TerminalPackage):
+        print(package.value)
+        sys.stdout.flush()
+
+    def daemon(self):
+        while self.read():
+            print('Hello world')
+            sys.stdout.flush()
+            time.sleep(1)
+
+
+terminals_manager = TerminalManager(terminals=[PrintTerminal()])
+for i in range(3):
+    time.sleep(2)
+    terminals_manager.send(TerminalPackage('Just print me'))
+
+terminals_manager.stop()

synergine2/terminals.py

@@ -0,0 +1,98 @@
+from multiprocessing import Queue
+
+from multiprocessing import Process
+from queue import Empty
+
+import time
+
+STOP_SIGNAL = '__STOP_SIGNAL__'
+
+
+class TerminalPackage(object):
+    def __init__(self, value):
+        self._value = value
+
+    @property
+    def value(self):
+        return self._value
+
+
+class Terminal(object):
+    """Default behaviour is to do nothing.
+    DEFAULT_SLEEP is sleep time between each queue read"""
+    DEFAULT_SLEEP = 1
+
+    def __init__(self):
+        self._input_queue = None
+        self._output_queue = None
+        self._stop_required = False
+
+    def start(self, input_queue: Queue, output_queue: Queue) -> None:
+        self._input_queue = input_queue
+        self._output_queue = output_queue
+        self.run()
+
+    def run(self):
+        """
+        Override this method to create your daemon terminal
+        """
+        while self.read():
+            time.sleep(self.DEFAULT_SLEEP)
+
+    def read(self):
+        while True:
+            try:
+                package = self._input_queue.get(block=False, timeout=None)
+                if package == STOP_SIGNAL:
+                    self._stop_required = True
+                    return False
+
+                self.receive(package)
+            except Empty:
+                return True  # Finished to read Queue
+
+    def receive(self, package: TerminalPackage):
+        raise NotImplementedError()
+
+    def send(self, package: TerminalPackage):
+        self._output_queue.put(package)
+
+
+class TerminalManager(object):
+    def __init__(self, terminals: [Terminal]):
+        self._terminals = terminals
+        self._outputs_queues = []
+        self._inputs_queues = []
+
+    def start(self) -> None:
+        for terminal in self._terminals:
+            output_queue = Queue()
+            self._outputs_queues.append(output_queue)
+
+            input_queue = Queue()
+            self._inputs_queues.append(input_queue)
+
+            process = Process(target=terminal.start, kwargs=dict(
+                input_queue=output_queue,
+                output_queue=input_queue,
+            ))
+            process.start()
+
+    def stop(self):
+        for output_queue in self._outputs_queues:
+            output_queue.put(STOP_SIGNAL)
+
+    def send(self, package: TerminalPackage):
+        for output_queue in self._outputs_queues:
+            output_queue.put(package)
+
+    def receive(self) -> []:
+        packages = []
+        for input_queue in self._inputs_queues:
+            try:
+                while True:
+                    packages.append(input_queue.get(block=False, timeout=None))
+            except Empty:
+                pass  # Queue is empty
+
+        return packages

tests/test_terminals.py

@@ -0,0 +1,34 @@
+import time
+
+from synergine2.terminals import Terminal
+from synergine2.terminals import TerminalPackage
+from synergine2.terminals import TerminalManager
+from tests import BaseTest
+
+
+class FakeTerminal(Terminal):
+    def receive(self, package: TerminalPackage):
+        self.send(TerminalPackage(package.value * 2))
+        self.send(TerminalPackage(package.value * 4))
+
+
+class TestTerminals(BaseTest):
+    def test_terminals_communications(self):
+        terminals_manager = TerminalManager(
+            terminals=[
+                FakeTerminal(),
+            ]
+        )
+        terminals_manager.start()
+        terminals_manager.send(TerminalPackage(42))
+
+        time.sleep(2)  # TODO: Replace by lock
+        packages = terminals_manager.receive()
+
+        assert 2 == len(packages)
+        values = [p.value for p in packages]
+        assert 84 in values
+        assert 168 in values
+
+        terminals_manager.stop()  # pytest must execute this if have fail
+        # TODO: Tester avec plusieurs terminaux