async beforePerform() {
        const { jobId, buildId } = this.args[0];
        const runningKey = `${runningJobsPrefix}${jobId}`;
        const lastRunningKey = `last_${runningJobsPrefix}${jobId}`;
        const waitingKey = `${waitingJobsPrefix}${jobId}`;
        const deleteKey = `deleted_${jobId}_${buildId}`;
        const enforceBlockedBySelf = String(this.options.blockedBySelf) === 'true'; // because kubernetes value is a string
        const shouldDelete = await this.queueObject.connection.redis.get(deleteKey);
        const runningBuildId = await this.queueObject.connection.redis.get(runningKey);
        const lastRunningBuildId = await this.queueObject.connection.redis.get(lastRunningKey);
        const enableCollapse = String(this.options.collapse) === 'true'; // because kubernetes value is a string
        const buildConfig = await this.queueObject.connection.redis.hget(
            `${queuePrefix}buildConfigs`, buildId);
        const annotations = hoek.reach(JSON.parse(buildConfig), 'annotations', { default: {} });
        const collapse = hoek.reach(annotations, 'screwdriver.cd/collapseBuilds', {
            default: enableCollapse, separator: '>' });
        const timeout = hoek.reach(annotations, 'screwdriver.cd/timeout', { separator: '>' });

        // For retry logic: failed to create pod, so it will retry
        // Current buildId is already set as runningKey. Should proceed
        if (parseInt(runningBuildId, 10) === buildId) {
            return true;
        }

        // Current build is older than last running build for the same job, discard the build
        if (collapse && buildId < parseInt(lastRunningBuildId, 10)) {
            await this.queueObject.connection.redis.lrem(waitingKey, 0, buildId);
            await helper.updateBuildStatus({
                redisInstance: this.queueObject.connection.redis,
                buildId,
                status: 'COLLAPSED',
                statusMessage: `Collapsed to build: ${lastRunningBuildId}`
            }, () => {});
            await this.queueObject.connection.redis.hdel(`${queuePrefix}buildConfigs`, buildId);

            return false;
        }

        // If this build is in the delete list (it was aborted)
        if (shouldDelete !== null) {
            await this.queueObject.connection.redis.del(deleteKey);

            //  Clean up to prevent race condition: stop and beforePerform happen at the same time
            //  stop deletes key runningKey and waitingKey
            //  beforePerform either proceeds or reEnqueue (which adds the key back)
            await this.queueObject.connection.redis.lrem(waitingKey, 0, buildId);

            if (parseInt(runningBuildId, 10) === buildId) {
                await this.queueObject.connection.redis.del(runningKey);
            }

            // Should not proceed since this build was previously aborted
            return false;
        }

        let blockedBy = this.args[0].blockedBy.split(',').map(jid =>
            `${runningJobsPrefix}${jid}`);

        if (!enforceBlockedBySelf) {
            blockedBy = blockedBy.filter(key => key !== `${runningJobsPrefix}${jobId}`); // remove itself from blocking list
        }

        if (blockedBy.length > 0) {
            const blockingBuildIds = [];

            // Get the blocking job
            await Promise.all(blockedBy.map(async (key) => {
                const val = await this.queueObject.connection.redis.get(key);

                if (val !== null) {
                    blockingBuildIds.push(val);
                }
            }));

            // If any blocking job is running, then re-enqueue
            if (blockingBuildIds.length > 0) {
                if (enforceBlockedBySelf && collapse) {
                    await collapseBuilds.call(this, {
                        waitingKey,
                        buildId,
                        blockingBuildIds });
                } else {
                    await this.reEnqueue(waitingKey, buildId, blockingBuildIds);
                }

                return false;
            }
        }

        if (enforceBlockedBySelf) { // only check this if feature is on
            const blocked = await blockedBySelf.call(this, { // pass in this context
                waitingKey,
                buildId,
                runningBuildId,
                collapse
            });

            if (blocked) { return false; } // if blocked then cannot proceed
        } else { // clean up waitingKey
            await this.queueObject.connection.redis.del(waitingKey);
        }

        // Register the curent job as running by setting key
        await this.queueObject.connection.redis.set(runningKey, buildId);
        // Set lastRunningKey
        await this.queueObject.connection.redis.set(lastRunningKey, buildId);

        // Set expire time to take care of the case where
        // afterPerform failed to call and blocked jobs will be stuck forever
        await this.queueObject.connection.redis.expire(runningKey,
            this.blockTimeout(timeout) * 60);
        await this.queueObject.connection.redis.expire(lastRunningKey,
            this.blockTimeout(timeout) * 60);

        // Proceed
        return true;
    }
/**
     * Checks if there are any blocking jobs running.
     * If yes, re-enqueue. If no, check if there is the same job waiting.
     * If buildId is not the same, re-enqueue. Otherwise, proceeds and set the current job as running
     * @method beforePerform
     * @return {Promise}
     */
Checks if there are any blocking jobs running.
If yes, re-enqueue. If no, check if there is the same job waiting.
If buildId is not the same, re-enqueue. Otherwise, proceeds and set the current job as running
@method beforePerform
@return {Promise}
Checks if there are any blocking jobs running . If yes re - enqueue . If no check if there is the same job waiting . If buildId is not the same re - enqueue . Otherwise proceeds and set the current job as running @method beforePerform @return { Promise }
async beforePerform() {
        const { jobId, buildId } = this.args[0];
        const runningKey = `${runningJobsPrefix}${jobId}`;
        const lastRunningKey = `last_${runningJobsPrefix}${jobId}`;
        const waitingKey = `${waitingJobsPrefix}${jobId}`;
        const deleteKey = `deleted_${jobId}_${buildId}`;
        const enforceBlockedBySelf = String(this.options.blockedBySelf) === 'true'; 
        const shouldDelete = await this.queueObject.connection.redis.get(deleteKey);
        const runningBuildId = await this.queueObject.connection.redis.get(runningKey);
        const lastRunningBuildId = await this.queueObject.connection.redis.get(lastRunningKey);
        const enableCollapse = String(this.options.collapse) === 'true'; 
        const buildConfig = await this.queueObject.connection.redis.hget(
            `${queuePrefix}buildConfigs`, buildId);
        const annotations = hoek.reach(JSON.parse(buildConfig), 'annotations', { default: {} });
        const collapse = hoek.reach(annotations, 'screwdriver.cd/collapseBuilds', {
            default: enableCollapse, separator: '>' });
        const timeout = hoek.reach(annotations, 'screwdriver.cd/timeout', { separator: '>' });
        if (parseInt(runningBuildId, 10) === buildId) {
            return true;
        }
        if (collapse && buildId < parseInt(lastRunningBuildId, 10)) {
            await this.queueObject.connection.redis.lrem(waitingKey, 0, buildId);
            await helper.updateBuildStatus({
                redisInstance: this.queueObject.connection.redis,
                buildId,
                status: 'COLLAPSED',
                statusMessage: `Collapsed to build: ${lastRunningBuildId}`
            }, () => {});
            await this.queueObject.connection.redis.hdel(`${queuePrefix}buildConfigs`, buildId);
            return false;
        }
        if (shouldDelete !== null) {
            await this.queueObject.connection.redis.del(deleteKey);
            await this.queueObject.connection.redis.lrem(waitingKey, 0, buildId);
            if (parseInt(runningBuildId, 10) === buildId) {
                await this.queueObject.connection.redis.del(runningKey);
            }
            return false;
        }
        let blockedBy = this.args[0].blockedBy.split(',').map(jid =>
            `${runningJobsPrefix}${jid}`);
        if (!enforceBlockedBySelf) {
            blockedBy = blockedBy.filter(key => key !== `${runningJobsPrefix}${jobId}`); 
        }
        if (blockedBy.length > 0) {
            const blockingBuildIds = [];
            await Promise.all(blockedBy.map(async (key) => {
                const val = await this.queueObject.connection.redis.get(key);
                if (val !== null) {
                    blockingBuildIds.push(val);
                }
            }));
            if (blockingBuildIds.length > 0) {
                if (enforceBlockedBySelf && collapse) {
                    await collapseBuilds.call(this, {
                        waitingKey,
                        buildId,
                        blockingBuildIds });
                } else {
                    await this.reEnqueue(waitingKey, buildId, blockingBuildIds);
                }
                return false;
            }
        }
        if (enforceBlockedBySelf) { 
            const blocked = await blockedBySelf.call(this, { 
                waitingKey,
                buildId,
                runningBuildId,
                collapse
            });
            if (blocked) { return false; } 
        } else { 
            await this.queueObject.connection.redis.del(waitingKey);
        }
        await this.queueObject.connection.redis.set(runningKey, buildId);
        await this.queueObject.connection.redis.set(lastRunningKey, buildId);
        await this.queueObject.connection.redis.expire(runningKey,
            this.blockTimeout(timeout) * 60);
        await this.queueObject.connection.redis.expire(lastRunningKey,
            this.blockTimeout(timeout) * 60);
        return true;
    }
async beforePerform ( ) { const { jobId , buildId } = this . args [ 0 ] ; const runningKey = ` ${ runningJobsPrefix } ${ jobId } ` ; const lastRunningKey = ` ${ runningJobsPrefix } ${ jobId } ` ; const waitingKey = ` ${ waitingJobsPrefix } ${ jobId } ` ; const deleteKey = ` ${ jobId } ${ buildId } ` ; const enforceBlockedBySelf = String ( this . options . blockedBySelf ) === 'true' ; const shouldDelete = await this . queueObject . connection . redis . get ( deleteKey ) ; const runningBuildId = await this . queueObject . connection . redis . get ( runningKey ) ; const lastRunningBuildId = await this . queueObject . connection . redis . get ( lastRunningKey ) ; const enableCollapse = String ( this . options . collapse ) === 'true' ; const buildConfig = await this . queueObject . connection . redis . hget ( ` ${ queuePrefix } ` , buildId ) ; const annotations = hoek . reach ( JSON . parse ( buildConfig ) , 'annotations' , { default : { } } ) ; const collapse = hoek . reach ( annotations , 'screwdriver.cd/collapseBuilds' , { default : enableCollapse , separator : '>' } ) ; const timeout = hoek . reach ( annotations , 'screwdriver.cd/timeout' , { separator : '>' } ) ; if ( parseInt ( runningBuildId , 10 ) === buildId ) { return true ; } if ( collapse && buildId < parseInt ( lastRunningBuildId , 10 ) ) { await this . queueObject . connection . redis . lrem ( waitingKey , 0 , buildId ) ; await helper . updateBuildStatus ( { redisInstance : this . queueObject . connection . redis , buildId , status : 'COLLAPSED' , statusMessage : ` ${ lastRunningBuildId } ` } , ( ) => { } ) ; await this . queueObject . connection . redis . hdel ( ` ${ queuePrefix } ` , buildId ) ; return false ; } if ( shouldDelete !== null ) { await this . queueObject . connection . redis . del ( deleteKey ) ; await this . queueObject . connection . redis . lrem ( waitingKey , 0 , buildId ) ; if ( parseInt ( runningBuildId , 10 ) === buildId ) { await this . queueObject . connection . redis . del ( runningKey ) ; } return false ; } let blockedBy = this . args [ 0 ] . blockedBy . split ( ',' ) . map ( jid => ` ${ runningJobsPrefix } ${ jid } ` ) ; if ( ! enforceBlockedBySelf ) { blockedBy = blockedBy . filter ( key => key !== ` ${ runningJobsPrefix } ${ jobId } ` ) ; } if ( blockedBy . length > 0 ) { const blockingBuildIds = [ ] ; await Promise . all ( blockedBy . map ( async ( key ) => { const val = await this . queueObject . connection . redis . get ( key ) ; if ( val !== null ) { blockingBuildIds . push ( val ) ; } } ) ) ; if ( blockingBuildIds . length > 0 ) { if ( enforceBlockedBySelf && collapse ) { await collapseBuilds . call ( this , { waitingKey , buildId , blockingBuildIds } ) ; } else { await this . reEnqueue ( waitingKey , buildId , blockingBuildIds ) ; } return false ; } } if ( enforceBlockedBySelf ) { const blocked = await blockedBySelf . call ( this , { waitingKey , buildId , runningBuildId , collapse } ) ; if ( blocked ) { return false ; } } else { await this . queueObject . connection . redis . del ( waitingKey ) ; } await this . queueObject . connection . redis . set ( runningKey , buildId ) ; await this . queueObject . connection . redis . set ( lastRunningKey , buildId ) ; await this . queueObject . connection . redis . expire ( runningKey , this . blockTimeout ( timeout ) * 60 ) ; await this . queueObject . connection . redis . expire ( lastRunningKey , this . blockTimeout ( timeout ) * 60 ) ; return true ; }
Checks if there are any blocking jobs running.
Checks if there are any blocking jobs running .
// because kubernetes value is a string // because kubernetes value is a string // For retry logic: failed to create pod, so it will retry // Current buildId is already set as runningKey. Should proceed // Current build is older than last running build for the same job, discard the build // If this build is in the delete list (it was aborted) //  Clean up to prevent race condition: stop and beforePerform happen at the same time //  stop deletes key runningKey and waitingKey //  beforePerform either proceeds or reEnqueue (which adds the key back) // Should not proceed since this build was previously aborted // remove itself from blocking list // Get the blocking job // If any blocking job is running, then re-enqueue // only check this if feature is on // pass in this context // if blocked then cannot proceed // clean up waitingKey // Register the curent job as running by setting key // Set lastRunningKey // Set expire time to take care of the case where // afterPerform failed to call and blocked jobs will be stuck forever // Proceed