Ensure PubSub layer closes connection pool before loop shutdown.

Co-authored-by: Carlton Gibson <[email protected]>

Author: bellini666

channels_redis/pubsub.py

@@ -302,7 +302,10 @@ async def flush(self):
                     pass
                 self._receive_task = None
             if self._redis is not None:
-                await self._redis.close()
+                # The pool was created just for this client, so make sure it is closed,
+                # otherwise it will schedule the connection to be closed inside the
+                # __del__ method, which doesn't have a loop running anymore.
+                await self._redis.close(close_connection_pool=True)
                 self._redis = None
                 self._pubsub = None
             self._subscribed_to = set()

tests/test_pubsub.py

@@ -32,6 +32,21 @@ async def other_channel_layer():
     await channel_layer.flush()
 
 
+def test_layer_close():
+    """
+    If the channel layer does not close properly there will be a "Task was destroyed but it is pending!" warning at
+    process exit.
+    """
+
+    async def do_something_with_layer():
+        channel_layer = RedisPubSubChannelLayer(hosts=TEST_HOSTS)
+        await channel_layer.send(
+            "TestChannel", {"type": "test.message", "text": "Ahoy-hoy!"}
+        )
+
+    async_to_sync(do_something_with_layer)()
+
+
 @pytest.mark.asyncio
 async def test_send_receive(channel_layer):
     """