add: refactoring code struct

Author: cfhamlet

MANIFEST.in

@@ -1,6 +1,6 @@
 include LICENSE
 include README.md
 include MANIFEST.in
-include src/os_m3_engine/VERSION
+include src/os_3m_engine/VERSION
 recursive-include tests *.py
 

README.md

@@ -1,17 +1,17 @@
-# os-m3-engine
+# os-3m-engine
 
-[![Build Status](https://www.travis-ci.org/cfhamlet/os-m3-engine.svg?branch=master)](https://www.travis-ci.org/cfhamlet/os-m3-engine)
-[![codecov](https://codecov.io/gh/cfhamlet/os-m3-engine/branch/master/graph/badge.svg)](https://codecov.io/gh/cfhamlet/os-m3-engine)
-[![PyPI - Python Version](https://img.shields.io/pypi/pyversions/os-m3-engine.svg)](https://pypi.python.org/pypi/os-m3-engine)
-[![PyPI](https://img.shields.io/pypi/v/os-m3-engine.svg)](https://pypi.python.org/pypi/os-m3-engine)
+[![Build Status](https://www.travis-ci.org/cfhamlet/os-3m-engine.svg?branch=master)](https://www.travis-ci.org/cfhamlet/os-3m-engine)
+[![codecov](https://codecov.io/gh/cfhamlet/os-3m-engine/branch/master/graph/badge.svg)](https://codecov.io/gh/cfhamlet/os-3m-engine)
+[![PyPI - Python Version](https://img.shields.io/pypi/pyversions/os-3m-engine.svg)](https://pypi.python.org/pypi/os-3m-engine)
+[![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.
 
 
 # Install
 
-`pip install os-m3-engine`
+`pip install os-3m-engine`
 
 # Usage
 

setup.py

@@ -14,8 +14,8 @@ def read(*filenames, **kwargs):
 
 
 setup(
-    name='os-m3-engine',
-    version=read('src/os_m3_engine/VERSION'),
+    name='os-3m-engine',
+    version=read('src/os_3m_engine/VERSION'),
     packages=find_packages(where='src'),
     package_dir={'': 'src'},
     include_package_data=True,
@@ -24,7 +24,7 @@ def read(*filenames, **kwargs):
     long_description=open('README.md').read(),
     author='Ozzy',
     author_email='cfhamlet@gmail.com',
-    url='https://github.com/cfhamlet/os-m3-engine',
+    url='https://github.com/cfhamlet/os-3m-engine',
     zip_safe=False,
     classifiers=[
         'Development Status :: 2 - Pre-Alpha',

src/os_m3_engine/VERSION renamed to src/os_3m_engine/VERSION

@@ -0,0 +1,9 @@
+import sys
+
+__all__ = ['__version__', 'version_info']
+
+import pkgutil
+__version__ = pkgutil.get_data(__package__, 'VERSION').decode('ascii').strip()
+version_info = tuple(int(v) if v.isdigit() else v
+                     for v in __version__.split('.'))
+del pkgutil

src/os_3m_engine/__init__.py

@@ -0,0 +1,45 @@
+from .common import Queue
+from .driver import Driver
+from .othread import Othread
+
+
+class BackendDriver(Driver):
+    def __init__(self, config, current_thread):
+        super(BackendDriver, self).__init__(config, current_thread)
+        self._transport_exists = False
+        if hasattr(self.runtime_context, 'transport_thread'):
+            self._transport_exists = True
+            self._queue = self.runtime_context.backend_thread_queue
+        else:
+            self._queue = self.runtime_context.frontend_thread_queue
+
+        self._processor = config.engine.backend.processor_cls(config)
+
+    @property
+    def queue(self):
+        return self._queue
+
+    def run(self):
+        while True:
+            if self.current_thread.stopping():
+                if self.queue.qsize() <= 0:
+                    if not self._transport_exists:
+                        break
+                    elif self.runtime_context.transport_thread.stopped():
+                        break
+
+            try:
+                data = self.queue.get(block=True, timeout=0.1)
+                self._processor.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/backend.py

@@ -0,0 +1,21 @@
+
+import sys
+_PY3 = sys.version_info[0] == 3
+
+if _PY3:
+    import queue as Queue
+else:
+    import Queue
+
+
+class Configurable(object):
+    def __init__(self, config):
+        self._config = config
+
+    @property
+    def config(self):
+        return self._config
+
+
+class RuntimeContext(object):
+    pass

src/os_3m_engine/common.py

@@ -0,0 +1,19 @@
+from .common import Configurable
+
+
+class Driver(Configurable):
+
+    def __init__(self, config, current_thread):
+        super(Driver, self).__init__(config)
+        self._current_thead = current_thread
+
+    @property
+    def current_thread(self):
+        return self._current_thead
+
+    @property
+    def runtime_context(self):
+        return self.config.runtime_context
+
+    def run(self):
+        raise NotImplementedError

src/os_3m_engine/driver.py

@@ -0,0 +1,108 @@
+import logging
+import signal
+import sys
+import threading
+import time
+from .common import Configurable, RuntimeContext, Queue
+from .frontend import FrontendThread
+from .transport import TransportThread
+from .backend import BackendThread
+from .othread import MultithreadManager
+from .utils import getatters
+
+
+class MultithreadEngine(Configurable):
+
+    def __init__(self, config):
+        super(MultithreadEngine, self).__init__(config)
+        self.__ensure()
+        self.__start_lock = threading.Lock()
+        self.__started = False
+        self.__stopped = False
+        self._runtime_context = RuntimeContext()
+
+    def __ensure(self):
+        engine_config = self.config.engine
+        assert any([hasattr(engine_config, x)
+                    for x in ('transport', 'backend')])
+
+    def __setup(self):
+        runtime_context = self._runtime_context
+        engine_config = self.config.engine
+
+        runtime_context.frontend_thread_queue = Queue.Queue(
+            engine_config.frontend.queue_size)
+
+        runtime_context.frontend_thread = MultithreadManager(
+            self.config, runtime_context,
+            FrontendThread, engine_config.frontend.thread_num,
+            engine_config.frontend.driver_cls)
+        runtime_context.frontend_thread.setDaemon(True)
+
+        if hasattr(engine_config, 'transport'):
+
+            if hasattr(engine_config, 'backend'):
+                runtime_context.backend_thread_queue = Queue.Queue(
+                    engine_config.backend.queue_size)
+
+            runtime_context.transport_thread = MultithreadManager(
+                self.config, runtime_context,
+                TransportThread, engine_config.transport.thread_num,
+                engine_config.transport.driver_cls)
+
+        if hasattr(engine_config, 'backend'):
+            runtime_context.backend_thread = MultithreadManager(
+                self.config, runtime_context,
+                BackendThread, engine_config.backend.thread_num,
+                engine_config.backend.driver_cls)
+
+    def __start(self):
+
+        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 start(self):
+        self.__acquire_start_lock(False, 'Can not start twice')
+        self.__setup()
+        self.__start()
+
+    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/engine.py

@@ -0,0 +1,40 @@
+from .driver import Driver
+from .common import Queue
+from .othread import Othread
+
+
+class FrontendDriver(Driver):
+    def __init__(self, config, current_thread):
+        super(FrontendDriver, self).__init__(config, current_thread)
+        self._processor = self.config.engine.frontend.processor_cls(config)
+
+    @property
+    def queue(self):
+        return self.runtime_context.frontend_thread_queue
+
+    def run(self):
+        for data in self._processor.process():
+            data = data.strip()
+            while True:
+                if not data:
+                    if self.current_thread.stopping():
+                        return
+                    else:
+                        break
+                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()