export class ProcessingQueue < T > { private eventQueue : readonly T [ ] = [ ] ; private pendingQueue : readonly T [ ] = [ ] ; private isProcessing = false ; private interval = undefined ; constructor ( private readonly processingFunction , private readonly processFrequency , private readonly queueName ? ) { } public clear ( ) { this . isProcessing = true ; this . eventQueue = [ ] ; this . pendingQueue = [ ] ; if ( this . interval ) { clearInterval ( this . interval ) ; this . interval = undefined ; } this . isProcessing = false ; } public eventsPendingCount ( ) { return this . eventQueue . length + this . pendingQueue . length ; } public async enqueue ( event ) { if ( this . isProcessing ) { this . pendingQueue = [ ... this . pendingQueue , event ] ; } else { this . eventQueue = [ ... this . eventQueue , event ] ; } this . processQueue ( ) ; } public async enqueueAll ( events ) { if ( this . isProcessing ) { this . pendingQueue = [ ... this . pendingQueue , ... events ] ; } else { this . eventQueue = [ ... this . eventQueue , ... events ] ; } this . processQueue ( ) ; } private async processQueue ( ) { if ( this . isProcessing ) { return ; } this . isProcessing = true ; if ( this . eventQueue . length === 0 ) { if ( this . interval ) { clearInterval ( this . interval ) ; this . interval = undefined ; } this . isProcessing = false ; return ; } if ( ! this . interval ) { this . interval = setInterval ( ( ) => this . processQueue ( ) , this . processFrequency ) ; } let hasEventBeenProcessed = true ; while ( this . eventQueue . length !== 0 && hasEventBeenProcessed ) { const queueAfterProcess = await this . processingFunction ( [ ... this . eventQueue ] ) ; hasEventBeenProcessed = queueAfterProcess . length !== this . eventQueue . length ; this . eventQueue = [ ... queueAfterProcess , ... this . pendingQueue ] ; this . pendingQueue = [ ] ; } this . isProcessing = false ; } }