add optional redis event handlers

Author: qeternity

README.rst

@@ -51,9 +51,8 @@ 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
+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.
 
@@ -165,6 +164,32 @@ If you're using Django, you may also wish to set this to your site's
         },
     }
 
+``on_disconnect`` / ``on_reconnect``
+~~~~~~~~~~~~
+
+The PubSub layer, which maintains long-running connections to Redis, can drop messages in the event of a network partition.
+To handle such situations the PubSub layer accepts optional arguments which will notify consumers of Redis disconnect/reconnect events.
+A common use-case is for consumers to ensure that they perform a full state re-sync to ensure that no messages have been missed.
+
+.. code-block:: python
+
+    CHANNEL_LAYERS = {
+        "default": {
+            "BACKEND": "channels_redis.pubsub.RedisPubSubChannelLayer",
+            "CONFIG": {
+                "hosts": [...],
+                "on_disconnect": "redis.disconnect",
+            },
+        },
+    }
+
+
+And then in your channels consumer, you can implement the handler:
+
+.. code-block:: python
+
+    async def redis_disconnect(self, *args):
+        # Handle disconnect
 
 Dependencies
 ------------

channels_redis/pubsub.py

@@ -13,7 +13,9 @@ class RedisPubSubChannelLayer:
     Channel Layer that uses Redis's pub/sub functionality.
     """
 
-    def __init__(self, hosts=None, prefix="asgi", **kwargs):
+    def __init__(
+        self, hosts=None, prefix="asgi", on_disconnect=None, on_reconnect=None, **kwargs
+    ):
         if hosts is None:
             hosts = [("localhost", 6379)]
         assert (
@@ -22,6 +24,9 @@ def __init__(self, hosts=None, prefix="asgi", **kwargs):
 
         self.prefix = prefix
 
+        self.on_disconnect = on_disconnect
+        self.on_reconnect = on_reconnect
+
         # Each consumer gets its own *specific* channel, created with the `new_channel()` method.
         # This dict maps `channel_name` to a queue of messages for that channel.
         self.channels = {}
@@ -271,6 +276,7 @@ async def _get_sub_conn(self):
         async with self._lock:
             if self._sub_conn is not None and self._sub_conn.closed:
                 self._sub_conn = None
+                self._notify_consumers(self.channel_layer.on_disconnect)
             if self._sub_conn is None:
                 if self._receive_task is not None:
                     self._receive_task.cancel()
@@ -301,6 +307,7 @@ async def _get_sub_conn(self):
                         self._receiver.channel(name) for name in self._subscribed_to
                     ]
                     await self._sub_conn.subscribe(*resubscribe_to)
+                    self._notify_consumers(self.channel_layer.on_reconnect)
             return self._sub_conn
 
     async def _do_receiving(self):
@@ -317,6 +324,11 @@ async def _do_receiving(self):
                     if channel_name in self.channel_layer.channels:
                         self.channel_layer.channels[channel_name].put_nowait(message)
 
+    def _notify_consumers(self, mtype):
+        if mtype is not None:
+            for channel in self.channel_layer.channels.values():
+                channel.put_nowait(msgpack.packb({"type": mtype}))
+
     async def _do_keepalive(self):
         """
         This task's simple job is just to call `self._get_sub_conn()` periodically.