import { forEach , pickBy } from 'lodash' ; import debugF from 'debug' ; import { getQueueTaskQueuedChannel , getQueueTaskProcessingChannel , getQueueTaskSuccessChannel , getQueueTaskErrorChannel , getQueueTaskStalledChannel , getQueueTaskFailedChannel , getQueueTaskCompleteChannel , getQueueTaskUpdatedChannel , getWorkerStartedChannel , getWorkerPausedChannel , getWorkerShutdownChannel , getQueueTaskScheduledChannel , getQueueTaskProgressUpdatedChannel , getQueueRateLimitUpdatedChannel , } from '../utils/keys' ; import { deSerializeEvent } from '../domain/events/deserialize-event' ; import { Event } from '../domain/events/event' ; import { EventType } from '../domain/events/event-type' ; import { createClientAndLoadLuaScripts , ensureDisconnected , RedisConfig , } from '../utils/redis' ; import { Listener } from '../domain/listener/listener' ; const debug = debugF ( 'conveyor-mq:listener' ) ; param queue param redisConfig param events returns export const createListener = ( { queue , redisConfig , events , } ) : Listener => { debug ( 'Starting' ) ; debug ( 'Creating client' ) ; const client = createClientAndLoadLuaScripts ( redisConfig ) ; const handlers = { } ; const setupListener = async ( ) => { const channelMap = { [ EventType . TaskScheduled ] : getQueueTaskScheduledChannel ( { queue } ) , [ EventType . TaskQueued ] : getQueueTaskQueuedChannel ( { queue } ) , [ EventType . TaskProcessing ] : getQueueTaskProcessingChannel ( { queue } ) , [ EventType . TaskSuccess ] : getQueueTaskSuccessChannel ( { queue } ) , [ EventType . TaskError ] : getQueueTaskErrorChannel ( { queue } ) , [ EventType . TaskStalled ] : getQueueTaskStalledChannel ( { queue } ) , [ EventType . TaskFail ] : getQueueTaskFailedChannel ( { queue } ) , [ EventType . TaskComplete ] : getQueueTaskCompleteChannel ( { queue } ) , [ EventType . TaskUpdated ] : getQueueTaskUpdatedChannel ( { queue } ) , [ EventType . TaskProgressUpdated ] : getQueueTaskProgressUpdatedChannel ( { queue , } ) , [ EventType . WorkerStarted ] : getWorkerStartedChannel ( { queue } ) , [ EventType . WorkerPaused ] : getWorkerPausedChannel ( { queue } ) , [ EventType . WorkerShutdown ] : getWorkerShutdownChannel ( { queue } ) , [ EventType . QueueRateLimitUpdated ] : getQueueRateLimitUpdatedChannel ( { queue , } ) , } ; client . on ( 'message' , ( channel , eventString ) => { const event = deSerializeEvent ( eventString ) ; const handlersToCall = handlers [ event . type ] ; debug ( `Received message ${ eventString } on channel ${ channel }` ) ; forEach ( handlersToCall , ( handler ) => handler ( { event } ) ) ; } ) ; debug ( 'Registered message handler' ) ; const channels = Object . values ( events ? pickBy ( channelMap , ( channel , eventType ) => events . includes ( eventType as EventType ) ) : channelMap ) ; await client . subscribe ( channels ) ; debug ( 'Client subscribed to channels' ) ; } ; const setupPromise = setupListener ( ) ; const ready = async ( ) => { await Promise . all ( [ setupPromise ] ) ; debug ( 'Ready' ) ; } ; const readyPromise = ready ( ) ; return { onReady : async ( ) => { await readyPromise ; } , on : ( eventType , f ) => { handlers [ eventType ] = [ ... ( handlers [ eventType ] || [ ] ) , f ] ; } , quit : async ( ) => { await readyPromise ; await ensureDisconnected ( { client } ) ; } , } ; } ;