import assert from "assert" ; import { Deferred } from "@fluidframework/common-utils" ; import { IConsumer , IQueuedMessage } from "@fluidframework/server-services-core" ; export class CheckpointManager { private checkpointing = false ; private closed = false ; private commitedCheckpoint : IQueuedMessage | undefined ; private lastCheckpoint : IQueuedMessage | undefined ; private pendingCheckpoint : Deferred < void > | undefined ; private error : any ; constructor ( private readonly id , private readonly consumer ) { } /**
     * Requests a checkpoint at the given offset
     */ public async checkpoint ( queuedMessage ) { assert ( this . lastCheckpoint === undefined || queuedMessage . offset >= this . lastCheckpoint . offset ) ; if ( this . closed ) { return ; } if ( this . error ) { return Promise . reject ( this . error ) ; } if ( this . commitedCheckpoint === queuedMessage ) { return ; } this . lastCheckpoint = queuedMessage ; if ( this . checkpointing ) { if ( ! this . pendingCheckpoint ) { this . pendingCheckpoint = new Deferred < void > ( ) ; } return this . pendingCheckpoint . promise ; } this . checkpointing = true ; const commitP = this . consumer . commitCheckpoint ( this . id , queuedMessage ) ; return commitP . then ( ( ) => { this . commitedCheckpoint = queuedMessage ; this . checkpointing = false ; if ( this . lastCheckpoint && this . lastCheckpoint !== this . commitedCheckpoint ) { assert ( this . pendingCheckpoint , "Differing offsets will always result in pendingCheckpoint" ) ; const nextCheckpointP = this . checkpoint ( this . lastCheckpoint ) ; this . pendingCheckpoint . resolve ( nextCheckpointP ) ; this . pendingCheckpoint = undefined ; } else if ( this . pendingCheckpoint ) { this . pendingCheckpoint . resolve ( ) ; this . pendingCheckpoint = undefined ; } } , ( error ) => { this . error = error ; if ( this . pendingCheckpoint ) { this . pendingCheckpoint . reject ( this . error ) ; } return Promise . reject ( error ) ; } ) ; } /**
     * Checkpoints at the last received offset.
     */ public async flush ( ) { if ( this . lastCheckpoint ) { return this . checkpoint ( this . lastCheckpoint ) ; } } /**
     * Closes the checkpoint manager - this will stop it from performing any future checkpoints
     */ public close ( ) { this . closed = true ; } }