Remove the dramatiq_group_callback_barrier_ttl environment variable

Author: mikeroll

docs/source/changelog.rst

@@ -19,12 +19,17 @@ Breaking Changes
 * The ``backend`` argument to the |Results| middleware is now required.
   Previously, not supplying this argument would result in a non-functional |Results| middleware.
   (`#728`_, `@LincolnPuzey`_)
+* The `dramatiq_group_callback_barrier_ttl` environment variable has been removed.
+  Instead, use the `barrier_ttl` parameter of the `GroupCallbacks` middleware.
+  (`#775`_, `@mikeroll`_)
 
 .. _#95: https://github.com/Bogdanp/dramatiq/issues/95
 .. _#345: https://github.com/Bogdanp/dramatiq/issues/345
 .. _#688: https://github.com/Bogdanp/dramatiq/pull/688
 .. _@azmeuk: https://github.com/azmeuk
 .. _#728: https://github.com/Bogdanp/dramatiq/pull/728
+.. _#775: https://github.com/Bogdanp/dramatiq/pull/775
+.. _@mikeroll: https://github.com/mikeroll 
 
 Fixed
 ^^^^^

docs/source/reference.rst

@@ -35,6 +35,7 @@ Message Composition
    :members:
 .. autoclass:: pipeline
    :members:
+.. autoclass:: dramatiq.middleware.GroupCallbacks
 
 Message Encoders
 ^^^^^^^^^^^^^^^^
@@ -105,6 +106,7 @@ The following middleware classes are available, but not enabled by default.
 .. autoclass:: dramatiq.middleware.CurrentMessage
    :members: get_current_message
    :member-order: bysource
+.. autoclass:: dramatiq.middleware.GroupCallbacks
 .. autoclass:: dramatiq.middleware.prometheus.Prometheus
 
 .. py:class:: dramatiq.results.Results
@@ -259,9 +261,6 @@ These are the environment variables that dramatiq reads
    * - ``dramatiq_dead_message_ttl``
      - 604800000 (One week)
      - The maximum amount of time a message can be in the dead letter queue for the RabbitMQ Broker (in milliseconds).
-   * - ``dramatiq_group_callback_barrier_ttl``
-     - 86400000 (One day)
-     -
    * - ``dramatiq_prom_db``
      - tempfile.gettempdir()/dramatiq-prometheus
      - The path to store the prometheus database files. See :ref:`gotchas-with-prometheus`.

dramatiq/composition.py

@@ -264,15 +264,13 @@ def run(self, *, delay=None):
           group: This same group.
         """
         if self.completion_callbacks:
-            from .middleware.group_callbacks import (
-                GROUP_CALLBACK_BARRIER_TTL,
-                GroupCallbacks,
-            )
+            from .middleware.group_callbacks import GroupCallbacks
 
             rate_limiter_backend = None
             for middleware in self.broker.middleware:
                 if isinstance(middleware, GroupCallbacks):
                     rate_limiter_backend = middleware.rate_limiter_backend
+                    barrier_ttl = middleware.barrier_ttl
                     break
             else:
                 raise RuntimeError(
@@ -285,7 +283,7 @@ def run(self, *, delay=None):
             # group is re-run, the barriers are all separate.
             # Re-using a barrier's name is an unsafe operation.
             completion_uuid = str(uuid4())
-            completion_barrier = Barrier(rate_limiter_backend, completion_uuid, ttl=GROUP_CALLBACK_BARRIER_TTL)
+            completion_barrier = Barrier(rate_limiter_backend, completion_uuid, ttl=barrier_ttl)
             completion_barrier.create(len(self.children))
 
             children = []

dramatiq/middleware/group_callbacks.py

@@ -17,17 +17,19 @@
 
 from __future__ import annotations
 
-import os
-
 from ..rate_limits import Barrier, RateLimiterBackend
 from .middleware import Middleware
 
-GROUP_CALLBACK_BARRIER_TTL = int(os.getenv("dramatiq_group_callback_barrier_ttl", "86400000"))
-
 
 class GroupCallbacks(Middleware):
-    def __init__(self, rate_limiter_backend: RateLimiterBackend) -> None:
+    def __init__(
+        self,
+        rate_limiter_backend: RateLimiterBackend,
+        *,
+        barrier_ttl: int = 86400 * 1000,
+    ) -> None:
         self.rate_limiter_backend = rate_limiter_backend
+        self.barrier_ttl = barrier_ttl
 
     def after_process_message(self, broker, message, *, result=None, exception=None):
         from ..message import Message
@@ -39,7 +41,7 @@ def after_process_message(self, broker, message, *, result=None, exception=None)
                 barrier = Barrier(
                     self.rate_limiter_backend,
                     group_completion_uuid,
-                    ttl=GROUP_CALLBACK_BARRIER_TTL,
+                    ttl=self.barrier_ttl,
                 )
                 if barrier.wait(block=False):
                     for message in group_completion_callbacks: