import * as Bull from 'bull' ; import * as config from '../config' ; import { IBullJobData , IBullWebhookResponse } from './Interfaces' ; import * as ActiveExecutions from './ActiveExecutions' ; import * as WebhookHelpers from './WebhookHelpers' ; export class Queue { private activeExecutions : ActiveExecutions . ActiveExecutions ; private jobQueue : Bull . Queue ; constructor ( ) { this . activeExecutions = ActiveExecutions . getInstance ( ) ; const prefix = config . get ( 'queue.bull.prefix' ) as string ; const redisOptions = config . get ( 'queue.bull.redis' ) as object ; this . jobQueue = new Bull ( 'jobs' , { prefix , redis : redisOptions , enableReadyCheck : false } ) ; this . jobQueue . on ( 'global:progress' , ( jobId , progress ) => { this . activeExecutions . resolveResponsePromise ( progress . executionId , WebhookHelpers . decodeWebhookResponse ( progress . response ) ) ; } ) ; } async add ( jobData , jobOptions ) { return this . jobQueue . add ( jobData , jobOptions ) ; } async getJob ( jobId ) { return this . jobQueue . getJob ( jobId ) ; } async getJobs ( jobTypes ) { return this . jobQueue . getJobs ( jobTypes ) ; } getBullObjectInstance ( ) { return this . jobQueue ; } param job returns async stopJob ( job ) { if ( await job . isActive ( ) ) { await job . progress ( - 1 ) ; return true ; } try { await job . remove ( ) ; return true ; } catch ( e ) { await job . progress ( - 1 ) ; } return false ; } } let activeQueueInstance ; export function getInstance ( ) { if ( activeQueueInstance === undefined ) { activeQueueInstance = new Queue ( ) ; } return activeQueueInstance ; }