(self, request, state, result, **kwargs)
Called on finishing each part of a Chord header
Called on finishing each part of a Chord header
Called on finishing each part of a Chord header
def on_chord_part_return(self, request, state, result, **kwargs):
        """Called on finishing each part of a Chord header"""
        tid, gid = request.id, request.group
        if not gid or not tid:
            return
        call_callback = False
        with transaction.atomic():
            # We need to know if `count` hits 0.
            # wrap the update in a transaction
            # with a `select_for_update` lock to prevent race conditions.
            # SELECT FOR UPDATE is not supported on all databases
            chord_counter = (
                ChordCounter.objects.select_for_update()
                .filter(group_id=gid).first()
            )
            if chord_counter is None:
                logger.warning("Can't find ChordCounter for Group %s", gid)
                return
            chord_counter.count -= 1
            if chord_counter.count != 0:
                chord_counter.save()
            else:
                # Last task in the chord header has finished
                call_callback = True
                chord_counter.delete()

        if call_callback:
            deps = chord_counter.group_result(app=self.app)
            if deps.ready():
                callback = maybe_signature(request.chord, app=self.app)
                trigger_callback(
                    app=self.app,
                    callback=callback,
                    group_result=deps
                )
def on_chord_part_return ( self , request , state , result , * * kwargs ) : tid , gid = request . id , request . group if not gid or not tid : return call_callback = False with transaction . atomic ( ) : # We need to know if `count` hits 0. # wrap the update in a transaction # with a `select_for_update` lock to prevent race conditions. # SELECT FOR UPDATE is not supported on all databases chord_counter = ( ChordCounter . objects . select_for_update ( ) . filter ( group_id = gid ) . first ( ) ) if chord_counter is None : logger . warning ( "Can't find ChordCounter for Group %s" , gid ) return chord_counter . count -= 1 if chord_counter . count != 0 : chord_counter . save ( ) else : # Last task in the chord header has finished call_callback = True chord_counter . delete ( ) if call_callback : deps = chord_counter . group_result ( app = self . app ) if deps . ready ( ) : callback = maybe_signature ( request . chord , app = self . app ) trigger_callback ( app = self . app , callback = callback , group_result = deps )