Added run safe async method to run external callbacks on internal eve…

…ntloop

ably/executer/decorator.py

@@ -38,7 +38,6 @@ def wrapper(*args, **kwargs):
 def optional_sync(fn):
     '''
     Executes async function as a sync function if sync_enabled property on the given instance is true.
-    Blocks execution of caller till result is returned.
     This decorator should only be used on async methods/coroutines.
     '''
     import asyncio
@@ -48,7 +47,7 @@ def wrapper(self, *args, **kwargs):
         if not hasattr(self, 'sync_enabled'):
             raise Exception("sync_enabled property should exist on instance to enable this feature")
 
-        # Return awaitable like a normal async method if sync is not enabled
+        # Return awaitable as is!
         if not self.sync_enabled:
             return fn(self, *args, **kwargs)
 
@@ -74,6 +73,42 @@ def wrapper(self, *args, **kwargs):
     return wrapper
 
 
+def run_safe_async(fn):
+    '''
+    USAGE :
+    Executes given async function/coroutine on the internal eventloop.
+    This is to make sure external thread/eventloop doesn't block/disrupt internal workflow.
+    This will be mainly needed on realtime-client public methods.
+    More information - https://github.com/ably/ably-python/issues/534
+    This decorator should only be used on async methods/coroutines.
+    '''
+    import asyncio
+
+    @functools.wraps(fn)
+    def wrapper(*args, **kwargs):
+        caller_eventloop = None
+        try:
+            caller_eventloop: events = asyncio.get_running_loop()
+        except Exception:
+            pass
+        app_loop: events = AppEventLoop.current().loop
+
+        res = fn(*args, **kwargs)
+        if asyncio.iscoroutine(res):
+            # Handle calls from app eventloop on the same loop, return awaitable
+            if caller_eventloop is not None and caller_eventloop == app_loop:
+                return res
+
+            # Handle calls from external eventloop, post them on app eventloop
+            # Return awaitable back to external_eventloop
+            future = asyncio.run_coroutine_threadsafe(res, app_loop)
+            return asyncio.wrap_future(future)
+
+        return res
+
+    return wrapper
+
+
 def close_app_eventloop(fn):
     import asyncio