Remove risk of getting "coroutine 'Queue.get' was never awaited" warning. (#687)

This could happen when the coroutine object returned from
asyncio.Queue.get() was handed over to a newly created task which was then
immediately cancelled so that the coroutine object was never awaited.

Instead, let the newly created task itself make the call to
asyncio.Queue.get() before immediately awaiting it.

Resolves #646.

Signed-off-by: Knut Aksel Røysland <[email protected]>

Author: knutroy

nats/aio/subscription.py

@@ -172,6 +172,10 @@ async def next_msg(self, timeout: Optional[float] = 1.0) -> Msg:
             msg = await sub.next_msg(timeout=1)
 
         """
+
+        async def timed_get() -> Msg:
+            return await asyncio.wait_for(self._pending_queue.get(), timeout)
+
         if self._conn.is_closed:
             raise errors.ConnectionClosedError
 
@@ -182,9 +186,7 @@ async def next_msg(self, timeout: Optional[float] = 1.0) -> Msg:
 
         task_name = str(uuid4())
         try:
-            future = asyncio.create_task(
-                asyncio.wait_for(self._pending_queue.get(), timeout)
-            )
+            future = asyncio.create_task(timed_get())
             self._pending_next_msgs_calls[task_name] = future
             msg = await future
         except asyncio.TimeoutError:

tests/test_js.py

@@ -829,6 +829,28 @@ async def wait_for_mock(future, _):
 
         await nc.close()
 
+    @async_test
+    async def test_fetch_being_cancelled_early(self):
+        nc = NATS()
+        await nc.connect()
+
+        js = nc.jetstream()
+        await js.add_stream(name="test", subjects=["test.a"])
+        sub = await js.pull_subscribe("test.a", "test", stream="test")
+
+        task = asyncio.create_task(sub.fetch())
+
+        # Allow "task" to reach the point where it has just finished calling
+        # asyncio.create_task() inside nats.client.Subscription.next_msg().
+        await asyncio.sleep(0)
+
+        # Cancel the call to sub.fetch() to provoke issue #646.
+        task.cancel()
+        with self.assertRaises(asyncio.CancelledError):
+            await task
+
+        await nc.close()
+
     @async_test
     async def test_ephemeral_pull_subscribe(self):
         nc = NATS()