Disable group_expiry, regardless of the spec

See django/channels#1371 for rationale.

Also, document "normal channels" (i.e., job queues) and what users
should do if they want job queues.

[closes #18]

Author: adamhooper

CHANGELOG.rst

@@ -1,3 +1,10 @@
+v1.2.0 - 2019-12-30
+~~~~~~~~~~~~~~~~~~~
+
+* Support Python 3.8 (upgrade to aioamqp=0.14.0) [issue #20] [PR #21]
+* Deprecate ``group_expiry`` [issue #18] [PR #19]. Rationale:
+  https://github.com/django/channels/issues/1371
+
 v1.1.5 - 2019-08-22
 ~~~~~~~~~~~~~~~~~~~
 

README.rst

@@ -44,15 +44,6 @@ Defaults to ``60``. You generally shouldn't need to change this, but you may
 want to turn it down if you have peaky traffic you wish to drop, or up if you
 have peaky traffic you want to backlog until you get to it.
 
-``group_expiry``
-~~~~~~~~~~~~~~~~
-
-Group expiry in seconds. Defaults to ``86400``. Channels will be removed from
-the group after this amount of time. It's recommended that you increase this
-parameter to ``86400000`` (1 year) and rely on explicit ``group_discard()`` to
-cancel subscriptions. (If your process halts, the group membership will
-disappear from RabbitMQ immediately: you needn't worry about leaks.)
-
 ``local_capacity``
 ~~~~~~~~~~~~~~~~~~
 
@@ -149,6 +140,52 @@ messages from RabbitMQ and routes them to receiver queues; each ``receive()``
 queries receiver queues. Empty queues are deleted. TODO delete queues that
 only contain expired messages, so we don't leak when sending to dead channels.
 
+Deviations from the Channel Layer Specification
+~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
+
+The `Channel Layer Specification
+<https://channels.readthedocs.io/en/latest/channel_layer_spec.html>`_ bends to
+Redis-related restrictions. RabbitMQ cannot emulate Redis. Here are the
+differences:
+
+* **No ``flush`` extension**: To flush all state, simply disconnect all clients.
+  (RabbitMQ won't allow one client to delete another client's data structures.)
+* **No ``group_expiry`` option**: the `group_expiry option
+  <https://channels.readthedocs.io/en/latest/channel_layer_spec.html#persistence>`_
+  recovers when a ``group_add()`` has no matching ``group_discard()``. But the
+  "group membership expiry" logic has a fatal flaw: it disconnects legitimate
+  members. ``channels_rabbitmq`` addresses the root problems directly:
+    * Web-server crash: RabbitMQ cleans all traces of a web server when it
+      disconnects. There's no problem here for ``group_expiry`` to solve.
+    * Programming errors: You may err and call ``group_add()`` without
+      eventually calling ``group_discard()``. Redis can't detect this
+      programming error (because it can't detect web-server crashes.) RabbitMQ
+      can. The ``local_expiry`` option keeps your site running when you
+      erroneously miss a ``group_discard()``. The channel layer warns when
+      discarding expired messages. Monitor your server logs to detect your
+      errors.
+* **No "normal channels"**: `normal channels
+  <https://channels.readthedocs.io/en/latest/channel_layer_spec.html#channels>`_
+  are job queues. In most projects, "normal channel" readers are worker
+  processes, ideally divorced from Websockets and Django.
+
+  You are welcome to submit a ``channels_rabbitmq`` pull request to support this
+  nigh-undocumented aspect of the Channel Layer Specification. But why reinvent
+  the wheel? There are thousands of job-queue implementations out there already.
+  Django Channels is a bad fit, because it is tuned for Websockets.
+
+  If you want an async, RabbitMQ-based job queue, investigate `aiormq
+  <https://github.com/mosquito/aiormq>`_ and `aioamqp
+  <https://github.com/polyconseil/aioamqp>`_. You can even send your jobs
+  to a separate RabbitMQ server or virtual host.
+
+  Currently, this project's strategy is to wait for `Celery 5.0.0
+  <https://github.com/celery/celery/milestone/7>`_, evaluate it, and then
+  recommend an alternative to "normal channels." (With Celery 4, it's
+  inefficient for workers to send messages to the Django Channels layer, because
+  they need to launch a new event loop and RabbitMQ connection per task.
+  Celery 5 may fix this.)
+
 Dependencies
 ------------
 

channels_rabbitmq/connection.py

@@ -135,14 +135,13 @@ async def get(self):
 
             return item
 
-    def __init__(self, loop, capacity, local_expiry, group_expiry):
+    def __init__(self, loop, capacity, local_expiry):
         self.loop = loop
         self.capacity = capacity
         self.local_expiry = local_expiry
-        self.group_expiry = group_expiry
         self.n = 0
 
-        self.local_groups = defaultdict(dict)  # group => {channel => expire}
+        self.local_groups = {}  # group => {channel, ...}
         self._out = defaultdict(lambda: MultiQueue.OutQueue(self))
         self._closed = asyncio.Event(loop=loop)
         self._putter_wakeup = asyncio.Event(loop=loop)
@@ -293,8 +292,8 @@ def group_add(self, group, asgi_channel):
         if self._closed.is_set():
             return None
 
-        channels = self.local_groups[group]  # may create set
-        channels[asgi_channel] = time.time() + self.group_expiry
+        channels = self.local_groups.setdefault(group, set())
+        channels.add(asgi_channel)
         return len(channels)
 
     def group_discard(self, group, asgi_channel):
@@ -311,20 +310,10 @@ def group_discard(self, group, asgi_channel):
 
         channels = self.local_groups[group]
         try:
-            del channels[asgi_channel]
+            channels.remove(asgi_channel)
         except KeyError:
             return None  # it was already removed
 
-        # Discard stale group memberships. These will happen if a
-        # group_add() has no matching group_discard(). We only discard
-        # within this one group because after we've discarded memberships,
-        # the caller needs to check whether it should unbind from RabbitMQ.
-        other_keys = list(channels.keys())
-        now = time.time()
-        for other_key in other_keys:
-            if channels[other_key] < now:
-                del channels[other_key]
-
         ret = len(channels)
 
         if ret == 0:
@@ -445,7 +434,6 @@ def __init__(
         prefetch_count=10,
         expiry=60,
         local_expiry=None,
-        group_expiry=86400,
         ssl_context=None,
         groups_exchange="groups",
     ):
@@ -458,17 +446,14 @@ def __init__(
         self.prefetch_count = prefetch_count
         self.expiry = expiry
         self.local_expiry = local_expiry
-        self.group_expiry = group_expiry
         self.queue_name = queue_name
         self.ssl_context = ssl_context
         self.groups_exchange = groups_exchange
 
         # incoming_messages: await `get()` on any channel-name queue to receive
         # the next message. If the `get()` is canceled, that's probably because
         # the caller is going away: we'll delete the queue in that case.
-        self._incoming_messages = MultiQueue(
-            loop, local_capacity, local_expiry, group_expiry
-        )
+        self._incoming_messages = MultiQueue(loop, local_capacity, local_expiry)
 
         # pending_puts: a "purgatory" for messages as we put them into
         # incoming_messages.

channels_rabbitmq/core.py

@@ -3,6 +3,7 @@
 import string
 import threading
 import types
+import warnings
 
 from channels.layers import BaseChannelLayer
 
@@ -50,7 +51,7 @@ def __init__(
         prefetch_count=10,
         expiry=60,
         local_expiry=None,
-        group_expiry=86400,
+        group_expiry=None,
         ssl_context=None,
         groups_exchange="groups",
     ):
@@ -60,10 +61,19 @@ def __init__(
         self.prefetch_count = prefetch_count
         self.expiry = expiry
         self.local_expiry = local_expiry
-        self.group_expiry = 86400
         self.ssl_context = ssl_context
         self.groups_exchange = groups_exchange
 
+        if group_expiry is not None:
+            warnings.warn(
+                (
+                    "channels_rabbitmq does not support group_expiry. Please do not configure it. "
+                    "For rationale, see "
+                    "https://github.com/CJWorkbench/channels_rabbitmq/issues/18#issuecomment-547052373"
+                ),
+                category=DeprecationWarning,
+            )
+
         # In inefficient client code (e.g., async_to_sync()), there may be
         # several send() or receive() calls within different event loops --
         # meaning callers might be coming from different threads at the same
@@ -94,7 +104,6 @@ def _create_connection(self, loop):
             prefetch_count=self.prefetch_count,
             expiry=self.expiry,
             local_expiry=self.local_expiry,
-            group_expiry=self.group_expiry,
             ssl_context=self.ssl_context,
             groups_exchange=self.groups_exchange,
         )

tests/test_core.py

@@ -153,3 +153,14 @@ def run():
     thread = threading.Thread(target=run, daemon=True)
     thread.start()
     thread.join()
+
+
+def test_warn_when_group_expiry_set():
+    with pytest.warns(DeprecationWarning) as record:
+        RabbitmqChannelLayer(host=HOST, group_expiry=86400)
+    assert str(record[0].message) == (
+        "channels_rabbitmq does not support group_expiry. Please do not configure it. "
+        "For rationale, see "
+        "https://github.com/CJWorkbench/channels_rabbitmq/issues/18#issuecomment-547052373"
+    )
+    assert len(record) == 1