add: refactoring finish

Author: cfhamlet

README.md

@@ -6,7 +6,7 @@
 [![PyPI](https://img.shields.io/pypi/v/os-3m-engine.svg)](https://pypi.python.org/pypi/os-3m-engine)
 
 
-Multithread engine for 3(or 2) stage job.
+Multithread engine for 3(or 2) stages job.
 
 
 # Install

requirements.txt

@@ -0,0 +1 @@
+os-config>0.1

src/os_3m_engine/__init__.py

@@ -1,7 +1,6 @@
 import sys
-from .engine import create
 
-__all__ = ['__version__', 'version_info', 'create']
+__all__ = ['__version__', 'version_info']
 
 import pkgutil
 __version__ = pkgutil.get_data(__package__, 'VERSION').decode('ascii').strip()

src/os_3m_engine/backend.py

@@ -1,12 +1,15 @@
-
 import sys
 
+from .utils import load_class
+
 _PY3 = sys.version_info[0] == 3
 
 if _PY3:
     import queue as Queue
+    binary_stdin = sys.stdin.buffer
 else:
     import Queue
+    binary_stdin = sys.stdin
 
 
 class Configurable(object):
@@ -18,5 +21,43 @@ def config(self):
         return self._config
 
 
-class RuntimeContext(object):
+class Workflowable(object):
+    def setup(self):
+        pass
+
+    def cleanup(self):
+        pass
+
+
+class Component(Configurable, Workflowable):
     pass
+
+
+class StandardComponent(Component):
+    def __init__(self, component_config, engine_config, runtime_context, current_thread):
+        super(StandardComponent, self).__init__(component_config)
+        self._engine_config = engine_config
+        self._runtime_context = runtime_context
+        self._current_thread = current_thread
+
+
+class FactoryInterface(object):
+    def create(self, *args, **kwargs):
+        raise NotImplementedError
+
+
+class ConfigurableFactory(Configurable):
+    def create(self, *args, **kwargs):
+        raise NotImplementedError
+
+
+class ComponentFactory(ConfigurableFactory):
+
+    def __init__(self, config, component_cls_string):
+        super(ComponentFactory, self).__init__(config)
+        self._component_cls = load_class(
+            component_cls_string, StandardComponent)
+
+    def create(self, engine_config, runtime_context, current_thread):
+        return self._component_cls(self.config, engine_config,
+                                   runtime_context, current_thread)

src/os_3m_engine/common.py

@@ -0,0 +1,46 @@
+from ..common import Queue, StandardComponent
+from .driver import Driver
+from .othread import Othread
+
+
+class Backend(StandardComponent):
+    def process(self, data):
+        raise NotImplementedError
+
+
+class BackendDriver(Driver):
+
+    def setup(self):
+        if hasattr(self._runtime_context, 'transport_thread'):
+            self._queue = self._runtime_context.backend_thread_queue
+        else:
+            self._queue = self._runtime_context.frontend_thread_queue
+        super(BackendDriver, self).setup()
+
+    @property
+    def queue(self):
+        return self._queue
+
+    def run(self):
+        while True:
+            if self._current_thread.stopping():
+                if self.queue.qsize() <= 0:
+                    if not hasattr(self._runtime_context, 'transport_thread'):
+                        break
+                    elif self._runtime_context.transport_thread.stopped():
+                        break
+
+            try:
+                data = self.queue.get(block=True, timeout=0.1)
+                self._component.process(data)
+            except Queue.Empty:
+                pass
+
+
+class BackendThread(Othread):
+
+    def run(self):
+        super(BackendThread, self).run()
+        self._runtime_context.frontend_thread.stop()
+        if hasattr(self._runtime_context, 'transport_thread'):
+            self._runtime_context.transport_thread.stop()

src/os_3m_engine/core/__init__.py

@@ -0,0 +1,20 @@
+from ..common import Configurable, Workflowable
+
+
+class Driver(Configurable, Workflowable):
+
+    def __init__(self, engine_config, runtime_context, current_thread, component_factory):
+        super(Driver, self).__init__(engine_config)
+        self._runtime_context = runtime_context
+        self._current_thread = current_thread
+        self._component = component_factory.create(
+            engine_config, runtime_context, current_thread)
+
+    def setup(self):
+        self._component.setup()
+
+    def cleanup(self):
+        self._component.cleanup()
+
+    def run(self):
+        raise NotImplementedError

src/os_3m_engine/core/backend.py

@@ -0,0 +1,66 @@
+import threading
+import time
+
+from ..common import Configurable
+from ..utils import getatters
+
+
+class RuntimeContext(object):
+    pass
+
+
+class Engine(Configurable):
+
+    def __init__(self, config, runtime_context):
+        super(Engine, self).__init__(config)
+        self.__start_lock = threading.Lock()
+        self.__started = False
+        self.__stopped = False
+        self._runtime_context = runtime_context
+
+    def start(self):
+
+        self.__acquire_start_lock(False, 'Can not start twice')
+        runtime_context = self._runtime_context
+        m = getatters(runtime_context, [
+                      'backend_thread',
+                      'transport_thread',
+                      'frontend_thread'
+                      ])
+        list(map(lambda x: x.start(), m))
+
+        self.__started = True
+        self.__start_lock.release()
+
+        while not any([x.stopped() for x in m]):
+            time.sleep(0.1)
+
+        m = getatters(runtime_context, [
+            'transport_thread',
+            'backend_thread'
+        ])
+        list(map(lambda x: x.join(), m))
+        self.__stopped = True
+
+    def __acquire_start_lock(self, started, err_msg):
+        self.__start_lock.acquire()
+        if self.__started != started:
+            try:
+                raise RuntimeError(err_msg)
+            finally:
+                self.__start_lock.release()
+
+    def __stop(self):
+        runtime_context = self._runtime_context
+        runtime_context.frontend_thread.stop()
+        if hasattr(runtime_context, 'transport_thread'):
+            runtime_context.transport_thread.stop()
+        elif hasattr(runtime_context, 'backend_thread'):
+            runtime_context.backend_thread.stop()
+        self.__stopped = True
+
+    def stop(self):
+        self.__acquire_start_lock(True, 'Can stop before start')
+        if not self.__stopped:
+            self.__stop()
+        self.__start_lock.release()

src/os_3m_engine/core/driver.py

@@ -0,0 +1,37 @@
+from ..common import Queue, StandardComponent
+from .driver import Driver
+from .othread import Othread
+
+
+class Frontend(StandardComponent):
+
+    def produce(self):
+        raise NotImplementedError
+
+
+class FrontendDriver(Driver):
+
+    @property
+    def queue(self):
+        return self._runtime_context.frontend_thread_queue
+
+    def run(self):
+        for data in self._component.produce():
+            while True:
+                try:
+                    self.queue.put(data, block=False, timeout=1)
+                    break
+                except Queue.Full:
+                    continue
+            if self._current_thread.stopping():
+                break
+
+
+class FrontendThread(Othread):
+
+    def run(self):
+        super(FrontendThread, self).run()
+        if hasattr(self._runtime_context, 'transport_thread'):
+            self._runtime_context.transport_thread.stop()
+        elif hasattr(self._runtime_context, 'backend_thread'):
+            self._runtime_context.backend_thread.stop()