import { EventEmitter } from "events" ; import { ResultCallback } from "./core/callback" ; import { Table } from "./core/table" ; import { PersistenceError } from "./persistenceError" ; hidden interface Task { operation : number ; wait : number ; arg : any ; callback : ResultCallback < any > ; finished : boolean ; next ? : Task ; } hidden export class TaskQueue extends EventEmitter { private _execute : ( operation : number , arg : any , callback : ResultCallback < any > ) => void ; private _activeCounts : Table < number > = { } ; private _active : number = 0 ; private _head : Task ; private _tail : Task ; private _closed : boolean ; private _invalid : boolean ; constructor ( execute ) { super ( ) ; this . _execute = execute ; } add ( operation , wait , arg , callback ? ) { var err ; if ( this . _closed ) { err = new PersistenceError ( "Session is closed." ) ; } if ( this . _invalid ) { err = new PersistenceError ( "Session is invalid. An error occurred during a previous action." ) ; } if ( err ) { callback ? callback ( err ) : this . _unhandledError ( err ) ; return ; } var task = { operation : operation , wait : wait , arg : arg , callback : callback , finished : false } ; if ( this . _activeCounts [ operation ] === undefined ) { this . _activeCounts [ operation ] = 0 ; } if ( this . _head ) { this . _tail = this . _tail . next = task ; } else { this . _head = this . _tail = task ; process . nextTick ( ( ) => this . _process ( ) ) ; } } /**
     * Returns true if the TaskQueue is invalid because an error occurred; otherwise, return false.
     */ get invalid ( ) : boolean { return this . _invalid ; } /**
     * Clears the current queue and prevents any new actions from being added to the queue. Note that any currently
     * executing tasks may still return and call their callback.
     */ close ( ) { this . _closed = true ; this . _clear ( ) ; } private _process ( ) { var task = this . _head ; while ( task && ! ( task . wait & this . _active ) ) { this . _head = this . _head . next ; if ( ! this . _activeCounts [ task . operation ] ++ ) { this . _active |= task . operation ; } this . _execute ( task . operation , task . arg , this . _finished ( task ) ) ; task = this . _head ; } } private _finished ( task ) { return ( err , result ) => { if ( task . finished ) { throw new PersistenceError ( "Callback for task can only be called once." ) ; } task . finished = true ; if ( ! ( -- this . _activeCounts [ task . operation ] ) ) { this . _active &= ~ task . operation ; } if ( err ) { this . _invalid = true ; } if ( task . callback ) { task . callback ( err , result ) ; } if ( err ) { if ( ! task . callback ) { this . _unhandledError ( err ) ; } this . _clear ( ) ; return ; } if ( this . _head ) { this . _process ( ) ; } } ; } private _unhandledError ( err ) { var task = this . _head ; while ( task && ! task . callback ) { task = task . next ; } if ( task && task . callback ) { task . callback ( err ) ; } else { this . emit ( 'error' , err ) ; } } private _clear ( ) { this . _head = this . _tail = null ; } }