import { BinaryHeap } from 'cdata' ; import { Task , PromisyClass , tryFinally } from './Task' ; export class TaskQueue < PromiseType extends PromisyClass > { constructor ( private Promise , /** Number of promises allowed to resolve concurrently. */ public concurrency ) { this . Promise = Promise ; this . concurrency = concurrency ; this . nextBound = ( ) => this . next ( 1 ) ; this . nextTickBound = ( ) => { this . timerStamp = 0 ; this . next ( 0 ) ; } ; } param func param delay add ( func , delay = 0 ) { if ( this . busyCount < this . concurrency && ! delay ) { ++ this . busyCount ; return ( tryFinally ( func , this . nextBound , this . Promise ) ) ; } else { const stamp = new Date ( ) . getTime ( ) + delay ; const task = new Task ( func , this . Promise , stamp ) ; this . backlog . insert ( task ) ; return ( task . delay ( ) ) ; } } /** Consider current function idle until promise resolves.
      * Useful for making recursive calls. */ unblock < Result > ( promise ) { this . next ( 1 ) ; const onFinish = ( ) => ++ this . busyCount ; promise . then ( onFinish , onFinish ) ; return ( promise ) ; } /** Wrap a function returning a promise, so that before running
      * it waits until concurrent invocations are below this queue's limit. */ wrap < Result > ( func , thisObject ? ) ; wrap < Result , A > ( func , thisObject ? ) ; wrap < Result , A , B > ( func , thisObject ? ) ; wrap < Result , A , B , C > ( func , thisObject ? ) ; wrap < Result , A , B , C , D > ( func , thisObject ? ) ; wrap < Result > ( func , thisObject ? ) { return ( ( ... args ) => this . add ( ( ) => func . apply ( thisObject , args ) ) ) ; } /** Start the next task from the backlog. */ private next ( ended ) { const stamp = new Date ( ) . getTime ( ) ; let task = null ; this . busyCount -= ended ; if ( this . busyCount < this . concurrency ) task = this . backlog . peekTop ( ) ; if ( ! task ) return ; if ( task . stamp <= stamp ) { ++ this . busyCount ; task = this . backlog . extractTop ( ) ! ; task . resume ( this . nextBound ) ; } else if ( ! this . timerStamp || task . stamp + 1 < this . timerStamp ) { if ( this . timerStamp ) { clearTimeout ( this . timer as any ) ; } this . timer = setTimeout ( this . nextTickBound , ~ ~ ( task . stamp - stamp + 1 ) ) ; this . timerStamp = task . stamp ; } } private nextBound : ( ) => void ; private nextTickBound : ( ) => void ; private busyCount = 0 ; private backlog = new BinaryHeap < Task < any > > ( ( a , b ) => a . stamp - b . stamp ) ; timer : number | NodeJS . Timer ; timerStamp : number ; }