import Base from "./base" ; import { Message } from "../kombu/message" ; export default class Worker extends Base { handlers : object = { } ; activeTasks : Set < Promise < any > > = new Set ( ) ; method param { String } name param { Function } handler example public register ( name , handler ) { if ( ! handler ) { throw new Error ( "Undefined handler" ) ; } if ( this . handlers [ name ] ) { throw new Error ( "Handler is already set" ) ; } this . handlers [ name ] = function registHandler ( ... args ) : Promise < any > { try { return Promise . resolve ( handler ( ... args ) ) ; } catch ( err ) { return Promise . reject ( err ) ; } } ; } method example public start ( ) { console . info ( "celery.node worker starting..." ) ; console . info ( `registered task: ${ Object . keys ( this . handlers ) }` ) ; return this . run ( ) . catch ( err => console . error ( err ) ) ; } method private returns { Promise } private run ( ) { return this . isReady ( ) . then ( ( ) => this . processTasks ( ) ) ; } method private returns private processTasks ( ) { const consumer = this . getConsumer ( this . conf . CELERY_QUEUE ) ; return consumer ( ) ; } method private param { String } queue private getConsumer ( queue ) { const onMessage = this . createTaskHandler ( ) ; return ( ) : any => this . broker . subscribe ( queue , onMessage ) ; } public createTaskHandler ( ) { const onTaskReceived = ( message ) : any => { if ( ! message ) { return Promise . resolve ( ) ; } let payload = null ; let taskName = message . headers [ "task" ] ; if ( ! taskName ) { payload = message . decode ( ) ; taskName = payload [ "task" ] ; } let body ; let headers ; if ( payload == null && ! ( "args" in message . decode ( ) ) ) { body = message . decode ( ) ; headers = message . headers ; } else { const args = payload [ "args" ] || [ ] ; const kwargs = payload [ "kwargs" ] || { } ; const embed = { callbacks : payload [ "callbacks" ] , errbacks : payload [ "errbacks" ] , chord : payload [ "chord" ] , chain : null } ; body = [ args , kwargs , embed ] ; headers = { lang : payload [ "lang" ] , task : payload [ "task" ] , id : payload [ "id" ] , rootId : payload [ "root_id" ] , parantId : payload [ "parentId" ] , group : payload [ "group" ] , meth : payload [ "meth" ] , shadow : payload [ "shadow" ] , eta : payload [ "eta" ] , expires : payload [ "expires" ] , retries : payload [ "retries" ] || 0 , timelimit : payload [ "timelimit" ] || [ null , null ] , kwargsrepr : payload [ "kwargsrepr" ] , origin : payload [ "origin" ] } ; } const [ args , kwargs ] = body ; const taskId = headers [ "id" ] ; const handler = this . handlers [ taskName ] ; if ( ! handler ) { throw new Error ( `Missing process handler for task ${ taskName }` ) ; } console . info ( `celery.node Received task: ${ taskName }[${ taskId }], args: ${ args }, kwargs: ${ JSON . stringify ( kwargs ) }` ) ; const timeStart = process . hrtime ( ) ; const taskPromise = handler ( ... args , kwargs ) . then ( result => { const diff = process . hrtime ( timeStart ) ; console . info ( `celery.node Task ${ taskName }[${ taskId }] succeeded in ${ diff [ 0 ] + diff [ 1 ] / 1e9 }s: ${ result }` ) ; this . backend . storeResult ( taskId , result , "SUCCESS" ) ; this . activeTasks . delete ( taskPromise ) ; } ) . catch ( err => { console . info ( `celery.node Task ${ taskName }[${ taskId }] failed: [${ err }]` ) ; this . backend . storeResult ( taskId , err , "FAILURE" ) ; this . activeTasks . delete ( taskPromise ) ; } ) ; this . activeTasks . add ( taskPromise ) ; return taskPromise ; } ; return onTaskReceived ; } method returns public async whenCurrentJobsFinished ( ) { return Promise . all ( Array . from ( this . activeTasks ) ) ; } method todo public stop ( ) { throw new Error ( "not implemented yet" ) ; } }