Added coroutine hook system

state_management.py

@@ -2,9 +2,8 @@ from abots.helpers import generator, infinitedict
 from gevent.lock import BoundedSemaphore
 from collections import namedtuple
 
-class AtomicDict(object):#dict):
+class AtomicDict(object):
     def __init__(self, *args, **kwargs):
-        #dict.__init__(self, *args, **kwargs)
         self._dict = dict()
         self._lock = BoundedSemaphore()
 
@@ -71,7 +70,8 @@ def manager(reducer, initial_state=infinitedict()):
         m["reducer"] = reducer
         m["state"] = initial_state
 
-    return meta, dispatch(meta)
+    state = lambda: get_state(meta)
+    return state, dispatch(meta)
 
 Action = namedtuple("Action", ["type", "payload"])
-meta, actions = manager(reducer, {"init": False})
+#meta, actions = manager(reducer)

taijitu.py

@@ -1,67 +1,55 @@
 from state_management import Action, manager
-from abots.helpers import generator, infinitedict
+from abots.helpers import generator, coroutine, infinitedict
 from time import monotonic as time
-from gevent import Greenlet, sleep, joinall
+from gevent import Greenlet, spawn, sleep, joinall
 from gevent.queue import Queue, Empty
+#from queue import Queue, Empty
 
-hooks = dict()
+HOOKS = dict()
 
 def reducer(state, action):
-    name = action.get("name")
+    if action.name == "ADD":
-    data = action.get("data")
+        state["counter"] = state.get("counter", 0) + action.data
-    if name == "ADD":
-        state["counter"] = state.get("counter", 0) + data
 
     return state
 
-@generator
-def dispatcher():
-    state, action = (yield)
-    state = state.copy() if state else infinitedict()
-    reducer(state, action)
-
-dispatch = dispatcher()
-
 def register(func):
-    f = func()
+    #f = generator(func)
-    f.start()
+    #f.start()
-    hooks[func.__name__.lower()] = f
+    HOOKS[func.__name__.lower()] = func#f()
     return func
 
-def loop():
+#@coroutine
-    joinall([func for name, func in hooks.items()])
+def pull(queue):
+    while True:
+        task = queue.get()
+        assert isinstance(task, tuple), f"Expected tuple: {task}"
+        assert len(task) == 2, f"Invalid format: {task}"
+        hook_name, message = task
+        hook = HOOKS.get(hook_name)
+        hook(queue, message)
 
-def transmit(hook_name, message):
+@generator
-    hook = hooks.get(hook_name)
+def push(queue):
-    inbox = getattr(hook, "inbox", None)
+    task = (yield)
-    put = getattr(inbox, "put", None)
+    assert isinstance(task, tuple), f"Expected tuple: {task}"
-    if callable(put):
+    assert len(task) == 2, f"Invalid format: {task}"
-        put(message)
+    queue.put(task)
-
-class Actor(Greenlet):
-    def __init__(self):
-        self.inbox = Queue()
-        Greenlet.__init__(self)
-
-    def send(self, message):
-        NotImplemented
-
-    def _run(self):
-        self.running = True
-        while self.running:
-            message = self.inbox.get()
-            self.send(message)
 
 @register
-class Ping(Actor):
+def ping(queue, message):
-    def send(self, message):
+    print(f"PING: {message}")
-        print(message)
+    task = "pong", message
-        dispatch("pong", "ping")
+    queue.put(task)
-        sleep(0)
 
 @register
-class Pong(Actor):
+def pong(queue, message):
-    def send(self, message):
+    print(f"PONG: {message}")
-        print(message)
+    task = "ping", message
-        dispatch("ping", "pong")
+    queue.put(task)
-        sleep(0)
+
+state, dispatch = manager(reducer)
+queue = Queue()
+yang = push(queue)
+yin = spawn(pull, queue)
+start = lambda: joinall([yin])