Provide an alternative channel layer implementation that uses Redis Pub/Sub

.github/workflows/tests.yml

@@ -3,7 +3,7 @@ name: Tests
 on:
   push:
    branches:
-   - master
+   - main
   pull_request:
 
 jobs:
@@ -18,7 +18,6 @@ jobs:
         - 3.7
         - 3.8
         - 3.9
-
     services:
       redis:
         image: redis
@@ -29,6 +28,19 @@ jobs:
           --health-interval 10s
           --health-timeout 5s
           --health-retries 5
+      sentinel:
+        image: bitnami/redis-sentinel
+        ports:
+          - 26379:26379
+        options: >-
+          --health-cmd "redis-cli -p 26379 ping"
+          --health-interval 10s
+          --health-timeout 5s
+          --health-retries 5
+        env:
+          REDIS_MASTER_HOST: redis
+          REDIS_MASTER_SET: sentinel
+          REDIS_SENTINEL_QUORUM: "1"
 
     steps:
       - uses: actions/checkout@v2

README.rst

@@ -31,6 +31,17 @@ Set up the channel layer in your Django settings file like so::
         },
     }
 
+Or, you can use the alternate implementation which uses Redis Pub/Sub::
+
+    CHANNEL_LAYERS = {
+        "default": {
+            "BACKEND": "channels_redis.pubsub.RedisPubSubChannelLayer",
+            "CONFIG": {
+                "hosts": [("localhost", 6379)],
+            },
+        },
+    }
+
 Possible options for ``CONFIG`` are listed below.
 
 ``hosts``
@@ -40,6 +51,12 @@ The server(s) to connect to, as either URIs, ``(host, port)`` tuples, or dicts c
 Defaults to ``['localhost', 6379]``. Pass multiple hosts to enable sharding,
 but note that changing the host list will lose some sharded data.
 
+Sentinel connections require dicts conforming to `create_sentinel
+<https://aioredis.readthedocs.io/en/v1.3.0/sentinel.html#aioredis.sentinel.
+create_sentinel>` with an additional `master_name` key specifying the Sentinel
+master set. Plain Redis and Sentinel connections can be mixed and matched if
+sharding.
+
 If your server is listening on a UNIX domain socket, you can also use that to connect: ``["unix:///path/to/redis.sock"]``.
 This should be slightly faster than a loopback TCP connection.
 

channels_redis/core.py

@@ -43,14 +43,15 @@ def _wrapper(self, *args, **kwargs):
 class ConnectionPool:
     """
     Connection pool manager for the channel layer.
-
     It manages a set of connections for the given host specification and
     taking into account asyncio event loops.
     """
 
     def __init__(self, host):
-        self.host = host
+        self.host = host.copy()
+        self.master_name = self.host.pop("master_name", None)
         self.conn_map = {}
+        self.sentinel_map = {}
         self.in_use = {}
 
     def _ensure_loop(self, loop):
@@ -68,16 +69,27 @@ def _ensure_loop(self, loop):
 
         return self.conn_map[loop], loop
 
+    async def create_conn(self, loop):
+        # One connection per pool since we are emulating a single connection
+        kwargs = {"minsize": 1, "maxsize": 1, **self.host}
+        if not (sys.version_info >= (3, 8, 0) and AIOREDIS_VERSION >= (1, 3, 1)):
+            kwargs["loop"] = loop
+        if self.master_name is None:
+            return await aioredis.create_redis_pool(**kwargs)
+        else:
+            kwargs = {"timeout": 2, **kwargs}  # aioredis default is way too low
+            sentinel = await aioredis.sentinel.create_sentinel(**kwargs)
+            conn = sentinel.master_for(self.master_name)
+            self.sentinel_map[conn] = sentinel
+            return conn
+
     async def pop(self, loop=None):
         """
         Get a connection for the given identifier and loop.
         """
         conns, loop = self._ensure_loop(loop)
         if not conns:
-            if sys.version_info >= (3, 8, 0) and AIOREDIS_VERSION >= (1, 3, 1):
-                conn = await aioredis.create_redis(**self.host)
-            else:
-                conn = await aioredis.create_redis(**self.host, loop=loop)
+            conn = await self.create_conn(loop)
             conns.append(conn)
         conn = conns.pop()
         if conn.closed:
@@ -96,48 +108,58 @@ def push(self, conn):
             conns, _ = self._ensure_loop(loop)
             conns.append(conn)
 
-    def conn_error(self, conn):
+    async def conn_error(self, conn):
         """
         Handle a connection that produced an error.
         """
-        conn.close()
+        await self._close_conn(conn)
         del self.in_use[conn]
 
     def reset(self):
         """
         Clear all connections from the pool.
         """
         self.conn_map = {}
+        self.sentinel_map = {}
         self.in_use = {}
 
+    async def _close_conn(self, conn, sentinel_map=None):
+        if sentinel_map is None:
+            sentinel_map = self.sentinel_map
+        if conn in sentinel_map:
+            sentinel_map[conn].close()
+            await sentinel_map[conn].wait_closed()
+            del sentinel_map[conn]
+        conn.close()
+        await conn.wait_closed()
+
     async def close_loop(self, loop):
         """
         Close all connections owned by the pool on the given loop.
         """
         if loop in self.conn_map:
             for conn in self.conn_map[loop]:
-                conn.close()
-                await conn.wait_closed()
+                await self._close_conn(conn)
             del self.conn_map[loop]
 
         for k, v in self.in_use.items():
             if v is loop:
+                await self._close_conn(k)
                 self.in_use[k] = None
 
     async def close(self):
         """
         Close all connections owned by the pool.
         """
         conn_map = self.conn_map
+        sentinel_map = self.sentinel_map
         in_use = self.in_use
         self.reset()
         for conns in conn_map.values():
             for conn in conns:
-                conn.close()
-                await conn.wait_closed()
+                await self._close_conn(conn, sentinel_map)
         for conn in in_use:
-            conn.close()
-            await conn.wait_closed()
+            await self._close_conn(conn, sentinel_map)
 
 
 class ChannelLock:
@@ -262,6 +284,7 @@ def decode_hosts(self, hosts):
             raise ValueError(
                 "You must pass a list of Redis hosts, even if there is only one."
             )
+
         # Decode each hosts entry into a kwargs dict
         result = []
         for entry in hosts:
@@ -888,7 +911,7 @@ async def __aenter__(self):
 
         async def __aexit__(self, exc_type, exc, tb):
             if exc:
-                self.pool.conn_error(self.conn)
+                await self.pool.conn_error(self.conn)
             else:
                 self.pool.push(self.conn)
             self.conn = None