add features sync callback processing

ml/callback.py

+import asyncio
+
 
 class CallbackManager(object):
     """
@@ -5,6 +7,7 @@ class CallbackManager(object):
     """
     CALLBACK = 'callback'
     ONE_SHOT = 'one_shot'
+    IS_ASYNC_CALLBACK = 'is_async_callback'
     ARGS = 'args'
     KWARGS = 'kwargs'
 
@@ -14,15 +17,31 @@ def __init__(self):
         """
         self._stack = {}
 
-    def add(self, prefix, callback, one_shot, *args, **kwargs):
+    def add(self, prefix, callback, one_shot, is_async, *args, **kwargs):
+        """
+        Appends async callback function to dictionary
+        :param prefix:
+        :param callback:
+        :param one_shot:
+        :param is_async:
+        :param args:
+        :param kwargs:
+        :return:
+        """
         # prepare the stack
         if prefix not in self._stack:
             self._stack[prefix] = []
 
+        if not isinstance(one_shot, bool):
+            raise TypeError
+        if not isinstance(is_async, bool):
+            raise TypeError
+
         # create callback dictionary
         callback_dict = self.create_callback_dict(
             callback,
             one_shot,
+            is_async,
             *args,
             **kwargs
         )
@@ -48,46 +67,39 @@ def clear(self):
         if self._stack:
             self._stack = {}
 
-    def process(self, prefix):
+    def process(self, prefix, loop=asyncio.get_event_loop()):
         if prefix not in self._stack:
             return False
-        for callback_dict in self._stack[prefix]:
-            method = callback_dict[self.CALLBACK]
-            args = callback_dict[self.ARGS]
-            kwargs = callback_dict[self.KWARGS]
-            method(*args, **kwargs)
 
-        for callback_dict in self._stack[prefix]:
-            if callback_dict[self.ONE_SHOT]:
-                self.remove(prefix, callback_dict)
-        return True
-
-    def process_async_task(self, prefix, loop):
-        if prefix not in self._stack:
-            return False
         for callback_dict in self._stack[prefix]:
             method = callback_dict[self.CALLBACK]
             args = callback_dict[self.ARGS]
             kwargs = callback_dict[self.KWARGS]
-            loop.create_task(method(*args, **kwargs))
+            if callback_dict[self.IS_ASYNC_CALLBACK]:
+                loop.create_task(method(*args, **kwargs))
+            else:
+                method(*args, **kwargs)
 
         for callback_dict in self._stack[prefix]:
             if callback_dict[self.ONE_SHOT]:
                 self.remove(prefix, callback_dict)
+
         return True
 
-    def create_callback_dict(self, callback, one_shot, *args, **kwargs):
+    def create_callback_dict(self, callback, one_shot, is_async, *args, **kwargs):
         """
         Create and return callback dictionary
 
         :param method callback:
         :param bool one_shot:
+        :param is_async:
         :return:
         """
 
         return {
             self.CALLBACK: callback,
             self.ONE_SHOT: one_shot,
+            self.IS_ASYNC_CALLBACK: is_async,
             self.ARGS: args,
             self.KWARGS: kwargs
         }