import Queue , { RateLimiter } from 'bull' ; import Redis from 'ioredis' ; import { QUEUE_COMPLETED_LIST_SIZE } from 'common/enums' ; import { environment } from 'common/environment' ; import logger from 'common/logger' ; export interface CustomQueueOpts { limiter ? : RateLimiter ; } export const sharedQueueOpts = { createClient ( ) { return new Redis ( { host : environment . queueHost , port : environment . queuePort , } ) ; } , defaultJobOptions : { removeOnComplete : QUEUE_COMPLETED_LIST_SIZE . small , } , } ; export const createQueue = ( queueName , customOpts ? ) => { const queue = new Queue ( queueName , { ... sharedQueueOpts , ... ( customOpts || { } ) , } ) ; queue . on ( 'error' , ( error ) => { logger . error ( `Job errored.` ) ; logger . info ( error ) ; } ) ; queue . on ( 'waiting' , ( jobId ) => { } ) ; queue . on ( 'active' , ( job , jobPromise ) => { } ) ; queue . on ( 'stalled' , ( job ) => { logger . error ( `Job#${ job . id } stalled, processing again.` ) ; logger . error ( { job } ) ; } ) ; queue . on ( 'progress' , ( job , progress ) => { } ) ; queue . on ( 'completed' , ( job , result ) => { } ) ; queue . on ( 'failed' , ( job , err ) => { logger . error ( `Job#${ job . id } failed, with following reason.` ) ; logger . error ( { job , err } ) ; } ) ; queue . on ( 'paused' , ( ) => { } ) ; queue . on ( 'resumed' , ( ) => { } ) ; queue . on ( 'cleaned' , ( jobs , type ) => { } ) ; queue . on ( 'drained' , ( ) => { } ) ; queue . on ( 'removed' , ( job ) => { } ) ; return queue ; } ;