Handle async proxy calls after event loop closes

bellows/thread.py

@@ -88,6 +88,23 @@ def __getattr__(self, name):
                 )
             )
 
+        if asyncio.iscoroutinefunction(func):
+
+            async def async_func_wrapper(*args, **kwargs):
+                loop = self._obj_loop
+                curr_loop = asyncio.get_running_loop()
+                call = functools.partial(func, *args, **kwargs)
+                if loop == curr_loop:
+                    return await call()
+                if loop.is_closed():
+                    # Disconnected
+                    LOGGER.warning("Attempted to use a closed event loop")
+                    return None
+                future = asyncio.run_coroutine_threadsafe(call(), loop)
+                return await asyncio.wrap_future(future, loop=curr_loop)
+
+            return async_func_wrapper
+
         def func_wrapper(*args, **kwargs):
             loop = self._obj_loop
             curr_loop = asyncio.get_running_loop()
@@ -98,21 +115,17 @@ def func_wrapper(*args, **kwargs):
                 # Disconnected
                 LOGGER.warning("Attempted to use a closed event loop")
                 return
-            if asyncio.iscoroutinefunction(func):
-                future = asyncio.run_coroutine_threadsafe(call(), loop)
-                return asyncio.wrap_future(future, loop=curr_loop)
-            else:
-
-                def check_result_wrapper():
-                    result = call()
-                    if result is not None:
-                        raise TypeError(
-                            (
-                                "ThreadsafeProxy can only wrap functions with no return"
-                                "value \nUse an async method to return values: {}.{}"
-                            ).format(self._obj.__class__.__name__, name)
-                        )
-
-                loop.call_soon_threadsafe(check_result_wrapper)
+
+            def check_result_wrapper():
+                result = call()
+                if result is not None:
+                    raise TypeError(
+                        (
+                            "ThreadsafeProxy can only wrap functions with no return"
+                            "value \nUse an async method to return values: {}.{}"
+                        ).format(self._obj.__class__.__name__, name)
+                    )
+
+            loop.call_soon_threadsafe(check_result_wrapper)
 
         return func_wrapper

tests/test_thread.py

@@ -161,6 +161,20 @@ async def test_proxy_loop_closed():
     assert obj.test.call_count == 0
 
 
+async def test_proxy_async_loop_closed():
+    loop = asyncio.new_event_loop()
+    obj = mock.MagicMock()
+
+    async def test():
+        return mock.sentinel.result
+
+    obj.test = test
+    proxy = ThreadsafeProxy(obj, loop)
+    loop.close()
+
+    assert await proxy.test() is None
+
+
 async def test_thread_task_cancellation_after_stop(thread):
     loop = asyncio.get_event_loop()
     obj = mock.MagicMock()