From 825f0f38b27a4343564142996be3a954d879383b Mon Sep 17 00:00:00 2001 From: Antoine Rabany Date: Wed, 16 Jun 2021 09:44:15 +0200 Subject: [PATCH] feat(pipeline): forget result of group pipelines --- remoulade/composition.py | 1 + remoulade/middleware/pipelines.py | 8 ++++++-- 2 files changed, 7 insertions(+), 2 deletions(-) diff --git a/remoulade/composition.py b/remoulade/composition.py index 6c70597c8..cc28b0d7b 100644 --- a/remoulade/composition.py +++ b/remoulade/composition.py @@ -41,6 +41,7 @@ class GroupInfo(namedtuple("GroupInfo", ("group_id", "children_count", "cancel_o group_id(str): The id of the group children_count(int) cancel_on_error(bool) + forget_on_enqueue """ def __new__(cls, *, group_id: str, children_count: int, cancel_on_error: bool): diff --git a/remoulade/middleware/pipelines.py b/remoulade/middleware/pipelines.py index c4ca935f1..dd4e0708f 100644 --- a/remoulade/middleware/pipelines.py +++ b/remoulade/middleware/pipelines.py @@ -94,8 +94,12 @@ def _send_next_message(self, pipe_target, broker, result, group_info): broker.enqueue(next_message) + if not pipe_ignore and group_info: + # We can forget the data now that it has been enqueued + self._group_results(group_info, broker, forget=True) + @staticmethod - def _group_results(group_info, broker): + def _group_results(group_info, broker, forget=False): """ Get the result of a group (fetch the group members message_ids then all the results) """ from ..collection_results import CollectionResults @@ -106,7 +110,7 @@ def _group_results(group_info, broker): message_ids = result_backend.get_group_message_ids(group_id=group_info.group_id) results = CollectionResults.from_message_ids(message_ids) - return list(results.get()) + return list(results.get(forget=forget)) def _group_completed(self, message, group_info, broker): """ Returns true if a group is completed, and increment the completion count of the group