|
@@ -6,10 +6,11 @@ from celery.task.sets import TaskSet, subtask
|
|
|
|
|
|
|
|
|
|
@current_app.task(name="celery.chord_unlock", max_retries=None)
|
|
@current_app.task(name="celery.chord_unlock", max_retries=None)
|
|
-def _unlock_chord(setid, callback, interval=1, max_retries=None):
|
|
|
|
|
|
+def _unlock_chord(setid, callback, interval=1, propagate=False,
|
|
|
|
+ max_retries=None):
|
|
result = TaskSetResult.restore(setid)
|
|
result = TaskSetResult.restore(setid)
|
|
if result.ready():
|
|
if result.ready():
|
|
- subtask(callback).delay(result.join())
|
|
|
|
|
|
+ subtask(callback).delay(result.join(propagate=propagate))
|
|
result.delete()
|
|
result.delete()
|
|
else:
|
|
else:
|
|
_unlock_chord.retry(countdown=interval, max_retries=max_retries)
|
|
_unlock_chord.retry(countdown=interval, max_retries=max_retries)
|
|
@@ -19,7 +20,8 @@ class Chord(current_app.Task):
|
|
accept_magic_kwargs = False
|
|
accept_magic_kwargs = False
|
|
name = "celery.chord"
|
|
name = "celery.chord"
|
|
|
|
|
|
- def run(self, set, body, interval=1, max_retries=None, **kwargs):
|
|
|
|
|
|
+ def run(self, set, body, interval=1, max_retries=None,
|
|
|
|
+ propagate=False, **kwargs):
|
|
if not isinstance(set, TaskSet):
|
|
if not isinstance(set, TaskSet):
|
|
set = TaskSet(set)
|
|
set = TaskSet(set)
|
|
r = []
|
|
r = []
|
|
@@ -30,7 +32,8 @@ class Chord(current_app.Task):
|
|
r.append(current_app.AsyncResult(uuid))
|
|
r.append(current_app.AsyncResult(uuid))
|
|
current_app.TaskSetResult(setid, r).save()
|
|
current_app.TaskSetResult(setid, r).save()
|
|
self.backend.on_chord_apply(setid, body, interval,
|
|
self.backend.on_chord_apply(setid, body, interval,
|
|
- max_retries=max_retries)
|
|
|
|
|
|
+ max_retries=max_retries,
|
|
|
|
+ propagate=propagate)
|
|
return set.apply_async(taskset_id=setid)
|
|
return set.apply_async(taskset_id=setid)
|
|
|
|
|
|
|
|
|