add green threads syntax

Author: czheo

setup.py

@@ -11,6 +11,7 @@
     packages=find_packages(),
     install_requires=[
         'multiprocess',
+        'eventlet',
     ],
     setup_requires=[
         'pytest-runner',

syntax_sugar/pipe.py

@@ -1,6 +1,7 @@
 from functools import partial
 from .composable import compose, composable
 from multiprocess.pool import ThreadPool, Pool
+from eventlet import GreenPool
 
 __all__ = [
     'END',
@@ -10,8 +11,10 @@
     'puts',
     'process_syntax',
     'thread_syntax',
+    'green_thread_syntax',
     'p',
     't',
+    'g',
 ]
 
 def puts(data, end="\n"):
@@ -49,6 +52,12 @@ class ProcessSyntax(MultiTaskSyntax):
 class ThreadSyntax(MultiTaskSyntax):
     pass
 
+class GreenThreadSyntax(MultiTaskSyntax):
+    pass
+
+class EventSyntax(MultiTaskSyntax):
+    pass
+
 def multitask(fn, poolsize, data, pool_constructor):
     with pool_constructor(poolsize) as p:
         if not hasattr(data, '__iter__'):
@@ -63,8 +72,14 @@ def multiprocess(fn, poolsize, data):
 def multithread(fn, poolsize, data):
     return multitask(fn, poolsize, data, ThreadPool)
 
+def multigreenthread(fn, poolsize, data):
+    p = GreenPool(poolsize)
+    return list(p.imap(fn, data))
+
+
 process_syntax = p = ProcessSyntax()
 thread_syntax = t = ThreadSyntax()
+green_thread_syntax = g = GreenThreadSyntax()
 
 class pipe:
     def __init__(self, data = None):
@@ -110,6 +125,8 @@ def __or__(self, rhs):
             self.action = compose(partial(multiprocess, rhs.func, rhs.poolsize), self.action)
         elif isinstance(rhs, ThreadSyntax):
             self.action = compose(partial(multithread, rhs.func, rhs.poolsize), self.action)
+        elif isinstance(rhs, GreenThreadSyntax):
+            self.action = compose(partial(multigreenthread, rhs.func, rhs.poolsize), self.action)
         elif isinstance(rhs, tuple):
             self.action = compose(partial(*rhs), self.action)
         elif isinstance(rhs, pipe):

tests/test_pipe.py

@@ -30,11 +30,14 @@ def test_pipe_connect():
 # not all processes in the pool are necessarily used 
 # 
 def test_pipe_multiprocess():
-    assert pipe(range(10)) | p[lambda x: x**2] * 3 | END == [x ** 2 for x in range(10)]
+    assert pipe(100) | range | p[lambda x: x**2] * 3 | sorted | END == [x ** 2 for x in range(100)]
 
 def test_pipe_multithread():
-    assert pipe(range(10)) | [lambda x: x**2] * 3 | END == [x ** 2 for x in range(10)]
-    assert pipe(range(10)) | t[lambda x: x**2] * 3 | END == [x ** 2 for x in range(10)]
+    assert pipe(100) | range | [lambda x: x**2] * 3 | sorted | END == [x ** 2 for x in range(100)]
+    assert pipe(100) | range | t[lambda x: x**2] * 3 | sorted | END == [x ** 2 for x in range(100)]
+
+def test_pipe_multigreenthread():
+    assert pipe(100) | range | g[lambda x: x**2] * 3 | sorted | END == [x ** 2 for x in range(100)]
 
 def test_redirect():
     p = pipe(range(10)) | each(str) | ''.join