import { setIntervalAsync , clearIntervalAsync , } from 'set-interval-async/dynamic' ; import debugF from 'debug' ; import { map } from 'lodash' ; import { Redis } from 'ioredis' ; import { processStalledTasks } from './process-stalled-tasks' ; import { enqueueScheduledTasks } from './enqueue-scheduled-tasks' ; import { quit as redisQuit , createClientAndLoadLuaScripts , RedisConfig , } from '../utils/redis' ; import { acknowledgeOrphanedProcessingTasks } from './acknowledge-orphaned-processing-tasks' ; import { Orchestrator } from '../domain/orchestrator/orchestrator' ; const debug = debugF ( 'conveyor-mq:orchestrator' ) ; param queue param redisConfig param redisClient param stalledCheckInterval param scheduledTasksCheckInterval param defaultStallTimeout returns export const createOrchestrator = ( { queue , redisConfig , redisClient , stalledCheckInterval = 1000 , scheduledTasksCheckInterval = 1000 , defaultStallTimeout = 1000 , } ) : Orchestrator => { debug ( 'Starting' ) ; if ( ! redisClient && ! redisConfig ) { throw new Error ( 'redisClient or redisConfig must be provided' ) ; } debug ( 'Creating client' ) ; const client = redisClient || createClientAndLoadLuaScripts ( redisConfig ! ) ; const processStalledTasksTick = async ( ) => { try { debug ( 'acknowledgeOrphanedProcessingTasks' ) ; await acknowledgeOrphanedProcessingTasks ( { queue , defaultStallTimeout , client , } ) ; debug ( 'processStalledTasks' ) ; await processStalledTasks ( { queue , client } ) ; } catch ( e ) { console . error ( e . toString ( ) ) ; } } ; const stalledTimer = setIntervalAsync ( processStalledTasksTick , stalledCheckInterval ) ; const enqueueScheduledTasksTick = async ( ) => { debug ( 'enqueueScheduledTasks' ) ; try { await enqueueScheduledTasks ( { queue , client } ) ; } catch ( e ) { console . error ( e . toString ( ) ) ; } } ; const enqueueDelayedTasksTimer = setIntervalAsync ( enqueueScheduledTasksTick , scheduledTasksCheckInterval ) ; const quit = async ( ) => { debug ( 'quit' ) ; await Promise . all ( map ( [ stalledTimer , enqueueDelayedTasksTimer ] , ( timer ) => clearIntervalAsync ( timer ) ) ) ; if ( ! redisClient ) { await redisQuit ( { client } ) ; } } ; const ready = async ( ) => { debug ( 'Ready' ) ; } ; const readyPromise = ready ( ) ; return { onReady : async ( ) => { debug ( 'onReady' ) ; await readyPromise ; } , quit , } ; } ;