import async from 'async' ; import { Queue , Worker , QueueScheduler } from 'bullmq' ; import { QueueInterface } from './queue-interface' ; import { Server } from '../server' ; const Redis = require ( 'ioredis' ) ; interface QueueWithWorker { queue : Queue ; worker : Worker ; scheduler : QueueScheduler ; } export class RedisQueueDriver implements QueueInterface { /**
     * The queues with workers list.
     */ protected queueWithWorker : Map < string , QueueWithWorker > = new Map ( ) ; /**
     * Initialize the Redis Queue Driver.
     */ constructor ( protected server ) { } /**
     * Add a new event with data to queue.
     */ addToQueue ( queueName , data = { } ) { return new Promise ( resolve => { let queueWithWorker = this . queueWithWorker . get ( queueName ) ; if ( ! queueWithWorker ) { return resolve ( ) ; } queueWithWorker . queue . add ( 'webhook' , data ) . then ( ( ) => resolve ( ) ) ; } ) ; } /**
     * Register the code to run when handing the queue.
     */ processQueue ( queueName , callback ) { return new Promise ( resolve => { if ( ! this . queueWithWorker . has ( queueName ) ) { const connection = new Redis ( { maxRetriesPerRequest : null , enableReadyCheck : false , ... this . server . options . database . redis , keyPrefix : undefined , } ) ; const prefix = this . server . options . database . redis . keyPrefix . replace ( /:$/ , '' ) ; const sharedOptions = { prefix , connection } ; this . queueWithWorker . set ( queueName , { queue : new Queue ( queueName , { ... sharedOptions , defaultJobOptions : { attempts : 6 , backoff : { type : 'exponential' , delay : 1000 , } , removeOnComplete : true , removeOnFail : true , } , } ) , worker : new Worker ( queueName , callback as any , { ... sharedOptions , concurrency : this . server . options . queue . redis . concurrency , } ) , scheduler : new QueueScheduler ( queueName , sharedOptions ) , } ) ; } resolve ( ) ; } ) ; } /**
     * Clear the queues for a graceful shutdown.
     */ clear ( ) { return async . each ( [ ... this . queueWithWorker ] , ( [ queueName , { queue , worker , scheduler } ] , callback ) => { scheduler . close ( ) . then ( ( ) => { worker . close ( ) . then ( ( ) => callback ( ) ) ; } ) ; } ) ; } }