import { forEach , map , set } from 'lodash' ; import debugF from 'debug' ; import { Redis } from 'ioredis' ; import { enqueueTasks as enqueueTasksAction } from './enqueue-tasks' ; import { ensureDisconnected , createClientAndLoadLuaScripts , RedisConfig , } from '../utils/redis' ; import { getTaskById } from './get-task-by-id' ; import { getTasksById } from './get-tasks-by-id' ; import { Task } from '../domain/tasks/task' ; import { Event } from '../domain/events/event' ; import { createListener } from './create-listener' ; import { EventType } from '../domain/events/event-type' ; import { TaskStatus } from '../domain/tasks/task-status' ; import { getTaskCounts } from './get-task-counts' ; import { destroyQueue } from './destroy-queue' ; import { removeTaskById } from './remove-task-by-id' ; import { scheduleTasks as scheduleTasksAction } from './schedule-tasks' ; import { getWorkers } from './get-workers' ; import { pauseQueue } from './pause-queue' ; import { resumeQueue } from './resume-queue' ; import { Manager } from '../domain/manager/manager' ; import { TaskResponse } from '../domain/manager/task-response' ; import { OnBeforeEnqueueTask , OnAfterEnqueueTask } from './enqueue-task' ; import { setQueueRateLimit } from './set-queue-rate-limit' ; import { getQueueRateLimitConfig } from './get-queue-rate-limit-config' ; const debug = debugF ( 'conveyor-mq:manager' ) ; export interface ManagerInput { queue : string ; redisConfig : RedisConfig ; redisClient ? : Redis ; hooks ? : { onBeforeEnqueueTask ? : OnBeforeEnqueueTask ; onAfterEnqueueTask ? : OnAfterEnqueueTask ; } ; } param queue param redisConfig param redisClient returns export const createManager = ( { queue , redisConfig , redisClient , hooks , } ) : Manager => { debug ( 'Starting' ) ; debug ( 'Creating client' ) ; const client = redisClient || createClientAndLoadLuaScripts ( redisConfig ) ; const eventSubscriptions = { } ; const setupListener = async ( ) => { const listener = createListener ( { queue , redisConfig , events : [ EventType . TaskComplete ] , } ) ; debug ( 'Created listener' ) ; listener . on ( EventType . TaskComplete , ( { event } ) => { if ( ! event || ! event . task || ! event . task . id ) return ; const { task } = event ; const handlers = eventSubscriptions ?. [ EventType . TaskComplete ] ?. [ task . id ] ; forEach ( handlers || [ ] , ( handler ) => { handler ( { event } ) ; } ) ; if ( handlers ) { delete eventSubscriptions [ EventType . TaskComplete ] [ task . id ] ; } } ) ; debug ( 'Registered listener' ) ; } ; const listenerSetupPromise = setupListener ( ) ; const onTaskComplete = async ( { taskId } ) => { const promise = new Promise ( ( resolve ) => { set ( eventSubscriptions , `${ EventType . TaskComplete }.${ taskId }` , [ ... ( eventSubscriptions ?. [ EventType . TaskComplete ] ?. [ taskId ] || [ ] ) , ( { event } ) => resolve ( event . task ) , ] ) ; } ) as Promise < Task > ; const task = await getTaskById ( { queue , taskId , client } ) ; if ( task && task . status && [ ( TaskStatus . Failed , TaskStatus . Success ) ] . includes ( task . status ) ) { delete eventSubscriptions [ EventType . TaskComplete ] [ taskId ] ; return task ; } return promise ; } ; const enqueueTasks = async ( { tasks , } ) : Promise < TaskResponse [ ] > => { const enqueuedTasks = await enqueueTasksAction ( { queue , tasks , client , onBeforeEnqueueTask : hooks ?. onBeforeEnqueueTask , onAfterEnqueueTask : hooks ?. onAfterEnqueueTask , } ) ; return map ( enqueuedTasks , ( task ) => ( { task , onTaskComplete : ( ) => onTaskComplete ( { taskId : task . id } ) , } ) ) ; } ; const enqueueTask = async ( { task , } ) : Promise < TaskResponse > => { const [ result ] = await enqueueTasks ( { tasks : [ task ] } ) ; return result ; } ; const scheduleTasks = async ( { tasks , } ) : Promise < TaskResponse [ ] > => { const scheduledTasks = await scheduleTasksAction ( { tasks , queue , client } ) ; return map ( scheduledTasks , ( task ) => ( { task , onTaskComplete : ( ) => onTaskComplete ( { taskId : task . id } ) , } ) ) ; } ; const scheduleTask = async ( { task , } ) : Promise < TaskResponse > => { const [ result ] = await scheduleTasks ( { tasks : [ task ] } ) ; return result ; } ; const ready = async ( ) => { await Promise . all ( [ listenerSetupPromise ] ) ; debug ( 'Ready' ) ; } ; const readyPromise = ready ( ) ; return { enqueueTask : async ( task ) => { await readyPromise ; debug ( `enqueueTask ${ task }` ) ; return enqueueTask ( { task } ) ; } , enqueueTasks : async ( tasks ) => { await readyPromise ; debug ( `enqueueTasks ${ tasks }` ) ; return enqueueTasks ( { tasks } ) ; } , scheduleTask : async ( task ) => { await readyPromise ; debug ( `scheduleTask ${ task }` ) ; return scheduleTask ( { task } ) ; } , scheduleTasks : async ( tasks ) => { await readyPromise ; debug ( `scheduleTasks ${ tasks }` ) ; return scheduleTasks ( { tasks } ) ; } , onTaskComplete : async ( taskId ) => { await readyPromise ; debug ( `onTaskComplete ${ taskId }` ) ; return onTaskComplete ( { taskId } ) ; } , getTaskById : async ( taskId ) => { await readyPromise ; debug ( `getTaskById ${ taskId }` ) ; return getTaskById ( { queue , taskId , client } ) ; } , getTasksById : async ( taskIds ) => { await readyPromise ; debug ( `getTasksById ${ taskIds }` ) ; return getTasksById ( { queue , taskIds , client } ) ; } , getTaskCounts : async ( ) => { await readyPromise ; debug ( `getTaskCounts` ) ; return getTaskCounts ( { queue , client } ) ; } , getWorkers : async ( ) => { await readyPromise ; debug ( `getWorkers` ) ; return getWorkers ( { queue , client } ) ; } , removeTaskById : async ( taskId ) => { await readyPromise ; debug ( `removeTaskById ${ taskId }` ) ; return removeTaskById ( { taskId , queue , client } ) ; } , destroyQueue : async ( ) => { await readyPromise ; debug ( `destroyQueue` ) ; return destroyQueue ( { queue , client } ) ; } , pauseQueue : async ( ) => { await readyPromise ; debug ( `pauseQueue` ) ; return pauseQueue ( { queue , client } ) ; } , resumeQueue : async ( ) => { await readyPromise ; debug ( `resumeQueue` ) ; return resumeQueue ( { queue , client } ) ; } , setQueueRateLimit : async ( { points , duration , } ) => { await readyPromise ; await setQueueRateLimit ( { points , duration , queue , client } ) ; } , getQueueRateLimit : async ( ) => { await readyPromise ; return getQueueRateLimitConfig ( { queue , client } ) ; } , onReady : async ( ) => { await readyPromise ; debug ( `onReady` ) ; } , quit : async ( ) => { await readyPromise ; debug ( `quit` ) ; if ( ! redisClient ) { await ensureDisconnected ( { client } ) ; } } , } ; } ;