avoid calls to deprecated asyncio.get_event_loop

examples/asyncio/coroutines.py

@@ -45,7 +45,7 @@ async def sender() -> None:
         await asyncio.sleep(1)
 
 
-asyncio.get_event_loop().run_until_complete(
+asyncio.run(
     asyncio.wait(
         [
             ping(),

examples/asyncio/helloworld_pubsub_dealerrouter.py

@@ -61,7 +61,7 @@ def __init__(self, url: str = '127.0.0.1', port: int = 5555):
     def main(self) -> None:
 
         # activate publishers / subscribers
-        asyncio.get_event_loop().run_until_complete(
+        asyncio.run(
             asyncio.wait(
                 [
                     self.hello_world_pub(),

zmq/asyncio.py

@@ -100,6 +100,16 @@ class _AsyncIO:
     _READ = selectors.EVENT_READ
 
     def _default_loop(self):
+        if sys.version_info >= (3, 7):
+            try:
+                return asyncio.get_running_loop()
+            except RuntimeError:
+                warnings.warn(
+                    "No running event loop. zmq.asyncio should be used from within an asyncio loop.",
+                    RuntimeWarning,
+                    stacklevel=4,
+                )
+        # get_event_loop deprecated in 3.10:
         return asyncio.get_event_loop()
 
 

zmq/tests/test_asyncio.py

@@ -35,14 +35,16 @@ def run(self):
         async def never_ending_task(socket):
             await socket.recv()  # never ever receive anything
 
-        loop = asyncio.get_event_loop()
+        loop = asyncio.new_event_loop()
         coro = asyncio.wait_for(never_ending_task(socket), timeout=1)
         try:
             loop.run_until_complete(coro)
         except asyncio.TimeoutError:
             pass  # expected timeout
         else:
             assert False, "never_ending_task was completed unexpectedly"
+        finally:
+            loop.close()
 
 
 class TestAsyncIOSocket(BaseZMQTestCase):
@@ -389,7 +391,7 @@ async def test():
             r.close()
             w.close()
 
-        loop = asyncio.get_event_loop()
+        loop = asyncio.new_event_loop()
         loop.run_until_complete(test())
 
     def test_multiple_loops(self):