join that extends and revokes

join.py

def join(resultset, task_options=(), queue_results=False, propagate=True,
         interval=0.5, cleanup=None):
    pending_results = deque(resultset.results)
    resultset = ResultSet(list(resultset.results))
    completed_results = []
    def handle_abort():
        logger.warning(
            "task_control.join is aborting ! Revoking taskset %s and calling cleanup function %r.",
            resultset,
            cleanup
        )
        revoke(resultset)
        if cleanup:
            cleanup()
    try:
        while pending_results:
            result = pending_results.popleft()
            try:
                resultvalue = result.wait(
                    timeout = interval,
                    propagate = propagate,
                    interval = interval
                )
                completed_results.append(resultvalue)
                if queue_results and isinstance(resultvalue, list):
                    for maybe_signature in resultvalue:
                        if isinstance(maybe_signature, Signature):
                            newresult = maybe_signature.apply_async(
                                **task_options
                            )
                            pending_results.append(newresult)
                            resultset.results.append(newresult)
            except TimeoutError:
                pending_results.append(result)
            check_termination(handle_abort)
    except Exception:
        handle_abort()
        raise
    return completed_results