import { EventEmitter } from "events" ; import ILogger from "../ILogger" ; import IOperationQueue from "./IOperationQueue" ; import uuid = require ( "uuid" ) ; interface IOperation { id : string ; op : ( ) => Promise < any > ; } export default class OperationQueue implements IOperationQueue { private operations : IOperation [ ] ; private emitter : EventEmitter ; private runningConcurrency : number ; public constructor ( private maxConcurrency , private readonly logger ) { this . operations = [ ] ; this . runningConcurrency = 0 ; this . emitter = new EventEmitter ( ) ; } template T param { Promise < T > } op param { string } contextId returns { Promise < T > } memberof public async operate < T > ( op , contextId ? ) { const id = uuid ( ) ; this . operations . push ( { id , op } ) ; this . logger . debug ( `OperationQueue.operate() Schedule incoming job ${ id }` , contextId ) ; this . execute ( contextId ) ; return new Promise < T > ( ( resolve , reject ) => { this . emitter . once ( id , res => { this . logger . debug ( `OperationQueue.operate() Job ${ id } completes callback, resolve.` , contextId ) ; this . emitter . removeAllListeners ( "error_" + id ) ; resolve ( res ) ; process . nextTick ( ( ) => { this . execute ( contextId ) ; } ) ; } ) . once ( "error_" + id , err => { this . logger . debug ( `OperationQueue.operate() Job ${ id } error, reject.` , contextId ) ; this . emitter . removeAllListeners ( id ) ; reject ( err ) ; process . nextTick ( ( ) => { this . execute ( contextId ) ; } ) ; } ) ; } ) ; } private param { string } contextId returns memberof private async execute < T > ( contextId ? ) { this . logger . debug ( `OperationQueue:execute() Current runningConcurrency:${ this . runningConcurrency } maxConcurrency:${ this . maxConcurrency } operations.length:${ this . operations . length }` , contextId ) ; if ( this . runningConcurrency < this . maxConcurrency ) { if ( this . operations . length === 0 ) { this . logger . debug ( `OperationQueue:execute() return. Operation.length === 0` , contextId ) ; return ; } this . runningConcurrency ++ ; const head = this . operations . shift ( ) ; let res ; try { res = await head ! . op ( ) ; } catch ( err ) { this . runningConcurrency -- ; this . emitter . emit ( `error_${ head ! . id }` , err ) ; return ; } this . runningConcurrency -- ; this . emitter . emit ( head ! . id , res ) ; } } }