try async generators

Author: 1st1

asyncio/base_events.py

@@ -294,9 +294,6 @@ def __init__(self):
         # Set to True when `loop.shutdown_asyncgens` is called.
         self._asyncgens_shutdown_called = False
 
-        # Future that isn't resolved while the loop is running.
-        self._forever_fut = None
-
     def __repr__(self):
         return ('<%s running=%s closed=%s debug=%s>'
                 % (self.__class__.__name__, self.is_running(),
@@ -433,12 +430,8 @@ def shutdown_asyncgens(self):
                     'asyncgen': agen
                 })
 
-    def get_forever_future(self):
-        return self._forever_fut
-
     def run_forever(self):
         """Run until stop() is called."""
-        self._forever_fut = self.create_future()
         self._check_closed()
         if self.is_running():
             raise RuntimeError('This event loop is already running')
@@ -457,14 +450,7 @@ def run_forever(self):
                 self._run_once()
                 if self._stopping:
                     break
-        except BaseException as ex:
-            self._forever_fut.set_exception(ex)
-            self._forever_fut._log_traceback = False
-            raise ex
-        else:
-            self._forever_fut.set_result(None)
         finally:
-            self._forever_fut = None
             self._stopping = False
             self._thread_id = None
             events._set_running_loop(None)

asyncio/events.py

@@ -512,9 +512,6 @@ def get_debug(self):
     def set_debug(self, enabled):
         raise NotImplementedError
 
-    def get_forever_future(self):
-        raise NotImplementedError
-
 
 class AbstractEventLoopPolicy:
     """Abstract policy for accessing the event loop."""

asyncio/run.py

@@ -2,12 +2,19 @@
 
 __all__ = ['run', 'forever']
 
+import inspect
 import threading
 
 from . import coroutines
 from . import events
 
 
+def _isasyncgen(obj):
+    if hasattr(inspect, 'isasyncgen'):
+        return inspect.isasyncgen(obj)
+    return False
+
+
 @coroutines.coroutine
 def forever():
     """Wait until the current event loop stops running.
@@ -67,8 +74,10 @@ async def main():
     if not isinstance(threading.current_thread(), threading._MainThread):
         raise RuntimeError(
             "asyncio.run() must be called from the main thread")
-    if not coroutines.iscoroutine(coro):
-        raise ValueError("a coroutine was expected, got {!r}".format(coro))
+    if not coroutines.iscoroutine(coro) and not _isasyncgen(coro):
+        raise ValueError(
+            "a coroutine or an asynchronous generator was expected, "
+            "got {!r}".format(coro))
 
     loop = events.new_event_loop()
     try:
@@ -77,15 +86,26 @@ async def main():
         if debug:
             loop.set_debug(True)
 
-        task = loop.create_task(coro)
-        task.add_done_callback(lambda task: loop.stop())
+        if _isasyncgen(coro):
+            result = None
+            loop.run_until_complete(coro.asend(None))
+            try:
+                loop.run_forever()
+            except BaseException as ex:
+                try:
+                    loop.run_until_complete(coro.athrow(ex))
+                except StopAsyncIteration as ex:
+                    if ex.args:
+                        result = ex.args[0]
+            else:
+                try:
+                    loop.run_until_complete(coro.asend(None))
+                except StopAsyncIteration as ex:
+                    if ex.args:
+                        result = ex.args[0]
 
-        try:
-            loop.run_forever()
-        except BaseException as ex:
-            result = loop.run_until_complete(task)
         else:
-            result = task.result()
+            result = loop.run_until_complete(coro)
 
         try:
             # `shutdown_asyncgens` was added in Python 3.6; not all