import { IDeltaQueue , IDeltaQueueEvents } from "@fluidframework/container-definitions" ; import { assert , performance , Deferred , TypedEventEmitter } from "@fluidframework/common-utils" ; import Deque from "double-ended-queue" ; export interface IDeltaQueueWriter < T > { push ( task ) : void ; clear ( ) : void ; } export class DeltaQueue < T > extends TypedEventEmitter < IDeltaQueueEvents < T > > implements IDeltaQueue < T > , IDeltaQueueWriter < T > { private isDisposed : boolean = false ; private readonly q = new Deque < T > ( ) ; /**
     * Tracks the number of pause requests for the queue
     * The DeltaQueue is create initially paused.
     */ private pauseCount = 1 ; private error : any | undefined ; /**
     * When processing is ongoing, holds a deferred that will resolve once processing stops.
     * Undefined when not processing.
     */ private processingDeferred : Deferred < void > | undefined ; public get disposed ( ) : boolean { return this . isDisposed ; } returns public get paused ( ) : boolean { return this . pauseCount !== 0 ; } public get length ( ) : number { return this . q . length ; } public get idle ( ) : boolean { return this . processingDeferred === undefined && this . q . length === 0 ; } public async waitTillProcessingDone ( ) { if ( this . processingDeferred !== undefined ) { return this . processingDeferred . promise ; } } param worker param logger constructor ( private readonly worker ) { super ( ) ; } public dispose ( ) { throw new Error ( "Not implemented." ) ; this . isDisposed = true ; } public clear ( ) { this . q . clear ( ) ; } public peek ( ) { return this . q . peekFront ( ) ; } public toArray ( ) { return this . q . toArray ( ) ; } public push ( task ) { this . q . push ( task ) ; this . emit ( "push" , task ) ; this . ensureProcessing ( ) ; } public async pause ( ) { this . pauseCount ++ ; return this . waitTillProcessingDone ( ) ; } public resume ( ) { assert ( this . pauseCount > 0 , 0x0f4 ) ; this . pauseCount -- ; this . ensureProcessing ( ) ; } /**
     * There are several actions that may need to kick off delta processing, so we want to guard against
     * accidental reentrancy. ensureProcessing can be called safely to start the processing loop if it is
     * not already started.
     */ private ensureProcessing ( ) { if ( ! this . paused && this . processingDeferred === undefined ) { this . processingDeferred = new Deferred < void > ( ) ; Promise . resolve ( ) . then ( ( ) => { this . processDeltas ( ) ; if ( this . processingDeferred !== undefined ) { this . processingDeferred . resolve ( ) ; this . processingDeferred = undefined ; } } ) ; } } /**
     * Executes the delta processing loop until a stop condition is reached.
     */ private processDeltas ( ) { const start = performance . now ( ) ; let count = 0 ; while ( ! ( this . q . length === 0 || this . paused || this . error !== undefined ) ) { const next = this . q . shift ( ) ; count ++ ; try { this . worker ( next ! ) ; this . emit ( "op" , next ) ; } catch ( error ) { this . error = error ; this . emit ( "error" , error ) ; } } if ( this . q . length === 0 ) { this . emit ( "idle" , count , performance . now ( ) - start ) ; } } }