Merge pull request #187 from Maplemx/dev

v3.4.1.1

Author: Maplemx

Agently/Agent/Agent.py

@@ -172,6 +172,8 @@ def export(self):
         return self.register_agent_component(name, AttachedWorkflow)
 
     async def start_async(self, request_type: str=None, *, return_generator:bool=False):
+        if return_generator:
+            return self.get_delta_generator()
         try:
             is_debug = self.settings.get_trace_back("is_debug")
             # Auto Save Agent runtime_ctx
@@ -265,12 +267,8 @@ async def handle_response(response):
 
             await handle_response({ "event": "response:finally", "data": self.request.response_cache })
 
-            if return_generator:
-                self.request_runtime_ctx.empty()
-                return self.response_generator.start()
-            else:
-                self.request_runtime_ctx.empty()
-                return self.request.response_cache["reply"]
+            self.request_runtime_ctx.empty()
+            return self.request.response_cache["reply"]
         except Exception as e:
             retry_time = self.settings.get_trace_back("request.retry_times")
             if not isinstance(retry_time, int):
@@ -288,6 +286,8 @@ async def handle_response(response):
                 return await self.start_async(request_type, return_generator=return_generator)
 
     def start(self, request_type: str=None, *, return_generator:bool=False):
+        if return_generator:
+            return self.get_delta_generator()
         reply_queue = queue.Queue()
         is_debug = self.settings.get_trace_back("is_debug")
         def start_in_theard():
@@ -307,15 +307,12 @@ def start_in_theard():
                     loop.close()
         theard = threading.Thread(target=start_in_theard)
         theard.start()
-        if return_generator:
-            return self.response_generator.start()
-        else:
-            theard.join()
-            try:        
-                reply = reply_queue.get_nowait()
-            except:
-                reply = None
-            return reply
+        theard.join()
+        try:        
+            reply = reply_queue.get_nowait()
+        except:
+            reply = None
+        return reply
 
     def start_websocket_server(self, port:int=15365):
         is_debug = self.settings.get_trace_back("is_debug")

Agently/Stage/EventEmitter.py

@@ -0,0 +1,58 @@
+from .Stage import Stage
+
+class EventEmitter:
+    def __init__(self, private_max_workers=None, max_concurrent_tasks=None, on_error=None):
+        self._listeners = {}
+        self._once = {}
+        self._stage = Stage(
+            private_max_workers=private_max_workers,
+            max_concurrent_tasks=max_concurrent_tasks,
+            on_error=on_error,
+            is_daemon=True,
+        )
+    
+    def add_listener(self, event, listener):
+        if event not in self._listeners:
+            self._listeners.update({ event: [] })
+        if listener not in self._listeners[event]:
+            self._listeners[event].append(listener)
+        return listener
+    
+    def remove_listener(self, event, listener):
+        if event in self._listeners and listener in self._listeners[event]:
+            self._listeners[event].remove(listener)
+
+    def remove_all_listeners(self, event_list):
+        if isinstance(event_list, str):
+            event_list = [event_list]
+        for event in event_list:
+            self._listeners.update({ event: [] })
+
+    def on(self, event, listener):
+        return self.add_listener(event, listener)
+    
+    def off(self, event, listener):
+        return self.remove_listener(event, listener)
+
+    def once(self, event, listener):
+        if event not in self._once:
+            self._once.update({ event: [] })
+        if listener not in self._listeners[event] and listener not in self._once[event]:
+            self._once[event].append(listener)
+        return listener
+    
+    def listener_count(self, event):
+        return len(self._listeners[event]) + len(self._once[event])
+        
+    def emit(self, event, *args, **kwargs):
+        listeners_to_execute = []
+        if event in self._listeners:
+            for listener in self._listeners[event]:
+                listeners_to_execute.append((listener, args, kwargs))
+        if event in self._once:
+            for listener in self._once[event]:
+                listeners_to_execute.append((listener, args, kwargs))
+            self._once.update({ event: [] })
+        for listener, args, kwargs in listeners_to_execute:
+            self._stage.go(listener, *args, **kwargs)
+        return len(listeners_to_execute)

Agently/Stage/MessageCenter.py

@@ -0,0 +1,27 @@
+import uuid
+from .Stage import Stage
+
+class MessageCenter:
+    def __init__(self, private_max_workers=None, max_concurrent_tasks=None, on_error=None):
+        self._stage = Stage(
+            private_max_workers=private_max_workers,
+            max_concurrent_tasks=max_concurrent_tasks,
+            on_error=on_error,
+            is_daemon=True
+        )
+        self._consumers = {}
+    
+    def register_consumer(self, handler):
+        consumer_id = uuid.uuid4()
+        self._consumers.update({ consumer_id: handler })
+        return consumer_id
+    
+    def remove_consumer(self, consumer_id):
+        del self._consumers[consumer_id]
+    
+    def put(self, data):
+        for _, consumer_handler in self._consumers.items():
+            self._stage.go(consumer_handler, data)
+    
+    def close(self):
+        self._stage.close()

Agently/utils/Stage/Stage.py renamed to Agently/Stage/Stage.py

@@ -1,3 +1,4 @@
+import atexit
 import inspect
 import threading
 import asyncio
@@ -7,42 +8,92 @@
 from .StageFunction import StageFunctionMixin
 
 class BaseStage:
-    def __init__(self, max_workers=5, max_concurrent_tasks=None, on_error=None, close_when_exception=False):
-        self._max_workers = max_workers
+    _global_executor = None
+    _global_max_workers = 5
+    _executor_lock = threading.RLock()
+
+    @staticmethod
+    def _get_global_executor():
+        if BaseStage._global_executor is None:
+            with BaseStage._executor_lock:
+                BaseStage._global_executor = ThreadPoolExecutor(max_workers=BaseStage._global_max_workers)
+        return BaseStage._global_executor
+        
+    @staticmethod
+    def set_global_max_workers(global_max_workers):
+        BaseStage._global_max_workers = global_max_workers
+        with BaseStage._executor_lock:
+            if BaseStage._global_executor:
+                BaseStage._global_executor.shutdown(wait=True)
+                BaseStage._global_executor = ThreadPoolExecutor(max_workers=BaseStage._global_max_workers)
+                atexit.register(BaseStage._global_executor.shutdown)
+    
+    def __init__(self, private_max_workers=5, max_concurrent_tasks=None, on_error=None, is_daemon=False):
+        self._private_max_workers = private_max_workers
         self._max_concurrent_tasks = max_concurrent_tasks
         self._on_error = on_error
+        self._is_daemon = is_daemon
         self._semaphore = None
         self._loop_thread = None
         self._loop = None
-        self._executor = None
-        self._responses = []
-        #self._initialize()
-    
-    def _initialize(self):
+        self._current_executor = None
         self._loop_ready = threading.Event()
-        self._loop_thread = threading.Thread(target=self._start_loop)
-        self._loop_thread.start()
-        self._executor = ThreadPoolExecutor(max_workers=self._max_workers)
-        self._loop_ready.wait()
-        del self._loop_ready
+        self._responses = set()
+        self._closed = False
+        if self._is_daemon:
+            atexit.register(self.close)
+        self._initialize()
+    
+    def __enter__(self):
+        self._initialize()
+        return self
+    
+    def __exit__(self, type, value, traceback):
+        self.close()
+        if type is not None and self._on_error is not None:
+            self._on_error(value)
+        return False
 
-    def _loop_exception_handler(self, loop, context):
-        if self._on_error is not None:
-            loop.call_soon_threadsafe(self._on_error, context["exception"])
+    @property
+    def _executor(self):
+        if self._current_executor is not None:
+            return self._current_executor
+        if self._private_max_workers:
+            self._current_executor = ThreadPoolExecutor(max_workers=self._private_max_workers)
+            return self._current_executor
         else:
-            raise context["exception"]
+            self._current_executor = BaseStage._get_global_executor()
+            return self._current_executor
 
+    def _initialize(self):
+        self._closed = False
+        if (
+            not self._loop_thread
+            or not self._loop_thread.is_alive()
+            or not self._loop
+            or not self._loop.is_running()
+        ):
+            self._loop_thread = threading.Thread(target=self._start_loop, daemon=self._is_daemon)
+            self._loop_thread.start()
+            self._loop_ready.wait()
+
     def _start_loop(self):
         self._loop = asyncio.new_event_loop()
         self._loop.set_exception_handler(self._loop_exception_handler)
-        asyncio.set_event_loop(self._loop)
         if self._max_concurrent_tasks:
             self._semaphore = asyncio.Semaphore(self._max_concurrent_tasks)
-        self._loop_ready.set()
+        asyncio.set_event_loop(self._loop)
+        self._loop.call_soon(lambda: self._loop_ready.set())
         self._loop.run_forever()
 
+    def _loop_exception_handler(self, loop, context):
+        if self._on_error is not None:
+            loop.call_soon_threadsafe(self._on_error, context["exception"])
+        else:
+            raise context["exception"]
+    
     def go(self, task, *args, on_success=None, on_error=None, lazy=False, async_gen_interval=0.1, **kwargs):
-        if not self._executor or not self._loop or not self._loop.is_running():
+        if not self._loop or self._loop.is_running():
             self._initialize()
         response_kwargs = {
             "on_success": on_success,
@@ -93,7 +144,7 @@ async def async_generator():
         elif inspect.isfunction(task) or inspect.ismethod(task):
             return StageResponse(self, self._loop.run_in_executor(self._executor, lambda: task(*args, **kwargs)), **response_kwargs)
         else:
-            return task
+            raise TypeError(f"Task seems like a value or an executed function not an executable task: { task }")
 
     def go_all(self, *task_list):
         response_list = []
@@ -134,7 +185,11 @@ def on_error(self, handler):
         self._on_error = handler
 
     def close(self):
-        for response in self._responses:
+        if self._closed:
+            return
+        self._closed = True
+        
+        for response in self._responses.copy():
             response._result_ready.wait()
 
         if self._loop and self._loop.is_running():
@@ -143,15 +198,15 @@ def close(self):
             pending = asyncio.all_tasks(self._loop)
             if pending:
                 self._loop.run_until_complete(asyncio.gather(*pending, return_exceptions=True))
+        if self._private_max_workers and self._current_executor is not None:
+            self._current_executor.shutdown(wait=True)
+            self._current_executor = None
         if self._loop_thread and self._loop_thread.is_alive():
             self._loop_thread.join()
             self._loop_thread = None
-        if self._loop and not self._loop.is_closed:
+        if self._loop and not self._loop.is_closed():
             self._loop.close()
         self._loop = None
-        if self._executor:
-            self._executor.shutdown(wait=True)
-            self._executor = None
 
 class Stage(BaseStage, StageFunctionMixin):
     pass

Agently/utils/Stage/StageFunction.py renamed to Agently/Stage/StageFunction.py

@@ -5,7 +5,7 @@
 class StageHybridGenerator:
     def __init__(self, stage, task, on_success=None, on_error=None, lazy=None, async_gen_interval=0.1):
         self._stage = stage
-        self._stage._responses.append(self)
+        self._stage._responses.add(self)
         self._loop = stage._loop
         self._on_success = on_success
         self._on_error = on_error
@@ -29,11 +29,12 @@ def _on_consume_async_gen_done(self, future):
         future.result()
         if self._error is not None:
             def raise_error():
-                raise Exception(self._error)
+                raise self._error
             self._loop.call_soon_threadsafe(raise_error)
         if self._on_success:
             self._final_result = self._on_success(self._result)
         self._result_ready.set()
+        self._stage._responses.discard(self)
 
     async def _consume_async_gen(self, task):
         try:

Agently/utils/Stage/StageHybridGenerator.py renamed to Agently/Stage/StageHybridGenerator.py

@@ -4,7 +4,7 @@
 class StageResponse:
     def __init__(self, stage, task, on_success=None, on_error=None):
         self._stage = stage
-        self._stage._responses.append(self)
+        self._stage._responses.add(self)
         self._loop = self._stage._loop
         self._on_success = on_success
         self._on_error = on_error
@@ -26,14 +26,17 @@ def _on_task_done(self, future):
             if self._on_success:
                 self._final_result = self._on_success(self._result)
             self._result_ready.set()
+            self._stage._responses.discard(self)
         except Exception as e:
             self._status = False
             self._error = e
             if self._on_error:
                 self._final_result = self._on_error(self._error)
                 self._result_ready.set()
+                self._stage._responses.discard(self)
             else:
                 self._result_ready.set()
+                self._stage._responses.discard(self)
                 raise self._error
 
     def get(self):