Merge pull request #2 from cfhamlet/develop

First Release

Author: cfhamlet

.travis.yml

@@ -21,6 +21,7 @@ deploy:
   provider: pypi
   user: cfhamlet
   password:
+    secure: eL8ptaVhcvilnSvCFlETjDlBFUL0ecbLjN8BWGwRBkjDcYqHPOze/E9sff0sdDBjiyiQbKu7MD554xAfSn8FsqlQLuQfVqT8RbgSKs2vdmwjOHwCENlINY4fmk+HCtvjWnK8r2hxWB87pxQOZ6Uqb4tYizRpQtOFxNEO9QlfjRRz/M4M6fjzO7BWybEUlxnPQn9IHXP+eGoSdHVg63d4EBmh7CcNCR4/lG7OSo89fJ0IjgWgDHv2P5vHOGeiZW7gc1jiHPeppc3vtCUfwLZ2D6bAps65zv/8jpgnzIiKz+XPtzzkcxMG9ZydN1vLMRFKHp+w2uPWXFn3BHxQ0ImQ5uW5hbyT9I7LehtWkdkEZhiyKTTfCUltns8lsSVG+w74SAHPUHyVj09msWvYioHx5gVKqSKzteEek/dZGxTBi6S1ifaMs4mWyqs2wCvcMkPGlLVCdGuO0soYqxXj4HmMQ0Fq9l4Bo4tRXZNjcshNO8Dje3MSTP6eO2XCnOf547D5/X3cZ6MqBPjZnqv6bg/qZFrxathJEXfqcHseWPUM4eTBEGu/69QRSEE72cx0Gmzo9HW18RQE1O/7+/Gn/rzKkGj7S/JrR3fYZVUj4AQnxroq2qgBYsW7R348jJHDq0EVRwzvRrVpUDrzIOHpdlJRBiZ6mtEObBRdrYlWRhFbuBM=
   true:
     tags: true
     condition: ${TOXENV} == py27

README.md

@@ -5,15 +5,162 @@
 [![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)
 
+Multi-thread engine for 3(or 2) stages job.
+
+
+
+Although multi-thread of python is buggy and slow, it is still necessary to write multi-thread program. Typically, producer-consumer is the most common model(so called 2-stage job), further more a transporter is needed between them(3-stage job). This library is used to simplify creating 3(or 2) stages program. 
+
+
+
+The 3 stages are:
+
+* Frontend: think as producer, usually used to create or receive data.
+* Transport: receive data from frontend stage, transform and send the transformed data to backend.
+* Backend: think as consumer, process the data received from transport.
+
+
+
+Something else need to know:
+
+* Each of the stage can be multi-thread.
+* Frontend is required, transport or backend can be omitted.
+
+
 
-Multithread engine for 3(or 2) stages job.
- 
 
 # Install
 
 `pip install os-3m-engine`
 
-# Usage
+# API
+
+* Create default engine, a typical 3-stage job:
+
+    - frontend: ``os_3m_engine.ootb.StdinFrontend``, read from stdin, send to transport stage
+    - transport: ``os_3m_engine.ootb.LogTransport``, log data received from fronted, send to backend stage
+    - backend: ``os_3m_engine.ootb.LogBackend``, log data received from transport
+
+    ```
+    from os_3m_engine.launcher import create
+    
+    engine = create()
+    ```
+
+* Create engine with custom defined stage:
+
+    ```
+    from os_3m_engine.launcher import create
+    
+    engine = create(transport_cls='transport_class_path_or_class')
+    ```
+
+* Create engine with custom engine config:
+
+    ```
+    from os_3m_engine.launcher import create
+    
+    config = WhateverOjbectYouWant
+    config.thread_num = 10
+    engine = create(transport_cls='your_transport_cls', engine_transport_config=config) 
+    ```
+
+* Start the engine:
+
+    ``start`` will block the current thread until all of the stage threads stopped
+    
+    ```
+    engine.start()
+    ```
+
+* The regular practice of stopping the engine
+
+    ```
+    from signal
+    from os_3m_engine.launcher import create
+    
+    engine = create()
+    
+    def stop(signum, frame):
+        engine.stop()
+    
+    signal.signal(signal.SIGINT, stop)
+    
+    engine.start()
+    ```
+    
+* Custom frontend class:
+
+    - inherit from ``os_3m_engine.core.frontend.Frontend``
+    - define ``produce`` method as generator
+
+    ```
+    from os_3m_engine.core.frontend import Frontend
+    
+    class CustomFrontend(Frontend):
+        def produce(self):
+            yield 'Hello world!'
+    ```
+    
+* Custom transport class:
+
+    - inherit from ``os_3m_engine.core.transport.Transport``
+    - define ``transport`` method, the only parameter is the data received, the return value will be sent to backend
+
+    ```
+    from os_3m_engine.core.transport import Transport
+    
+    class CustomTransport(Transport):
+        def transport(self, data):
+            return data
+    ```
+
+* Custom backend class:
+
+    - inherit from ``os_3m_engine.core.backend.Backend``
+    - define ``process`` method, the only parameter is the data received, no need return
+
+    ```
+    from os_3m_engine.core.backend import Backend
+    
+    class CustomBackend(Backend):
+        def process(self, data):
+            print(data)
+    ```
+
+* Passing parameters
+
+    - create engine with custom config object
+    - use ``self.config`` to get the config in stage class
+
+    ```
+    from os_3m_engine.launcher import create
+    
+    config = WhateverOjbectYouWant
+    engine = create(app_config=config)
+    ```
+
+
+* Custom ``setup``, ``cleanup`` behavior
+  
+    - each stage class can define ``setup``, ``cleanup`` methods
+    - these will be called at each thread start/stop
+    
+    
+    ```
+    from os_3m_engine.core.backend import Backend
+    
+    class CustomBackend(Backend):
+    
+        def setup(self):
+            print('Setup')
+            
+        def cleanup(self):
+            print('Cleanup')
+        
+        def process(self, data):
+            print(data)
+    ```
 
 
 

requirements.txt

@@ -1 +1,2 @@
+six
 os-config>0.1

src/os_3m_engine/utils.py

@@ -1,20 +1,26 @@
 import inspect
 from importlib import import_module
 
+import six
+
 
 def getatters(obj, attrs):
     return [getattr(obj, attr) for attr in attrs if hasattr(obj, attr)]
 
 
 def load_class(class_path, base_class=None, include_base_class=False):
-    module_path, class_name = class_path.rsplit('.', 1)
-    _mod = import_module(module_path)
-    if not hasattr(_mod, class_name):
-        return None
-    _cls = getattr(_mod, class_name)
-    if not inspect.isclass(_cls):
-        return None
-    elif base_class is None:
+    if isinstance(class_path,  six.string_types):
+        module_path, class_name = class_path.rsplit('.', 1)
+        _mod = import_module(module_path)
+        if not hasattr(_mod, class_name):
+            return None
+        _cls = getattr(_mod, class_name)
+        if not inspect.isclass(_cls):
+            return None
+    elif inspect.isclass(class_path):
+        _cls = class_path
+
+    if base_class is None:
         return _cls
     elif issubclass(_cls, base_class) and (include_base_class or _cls != base_class):
         return _cls

tests/test_engine.py

@@ -37,3 +37,14 @@ def test_engine_002(config):
     engine.stop()
     t1.join()
     assert engine.stopped()
+
+
+def test_engine_003(config):
+    from misc import OneRecordFrontend
+
+    engine = create(
+        frontend_cls=OneRecordFrontend,
+        transport_cls='misc.ConfirmTransport',
+        backend_cls='misc.ConfirmBackend',
+    )
+    engine.start()