import assert from "assert" ; import { EventEmitter } from "events" ; import { IContext , IContextErrorData , IQueuedMessage , IRoutingKey } from "@fluidframework/server-services-core" ; import { DocumentContext } from "./documentContext" ; const LastCheckpointedOffset = { offset : - 1 , partition : - 1 , topic : "" , value : undefined } ; /**
 * The DocumentContextManager manages a set of created DocumentContexts and computes an aggregate checkpoint offset
 * from them.
 */ export class DocumentContextManager extends EventEmitter { private readonly contexts : Set < DocumentContext > = new Set ( ) ; private head = LastCheckpointedOffset ; private tail = LastCheckpointedOffset ; private lastCheckpoint = LastCheckpointedOffset ; private closed = false ; constructor ( private readonly partitionContext ) { super ( ) ; } /**
     * Creates a context that should be used for a single document partition
     * This class is responsible for the lifetime of the context
     */ public createContext ( routingKey , head ) { assert ( head . offset > this . tail . offset && head . offset <= this . head . offset ) ; const context = new DocumentContext ( routingKey , head , this . partitionContext . log , ( ) => this . tail ) ; this . contexts . add ( context ) ; context . addListener ( "checkpoint" , ( ) => this . updateCheckpoint ( ) ) ; context . addListener ( "error" , ( error , errorData ) => this . emit ( "error" , error , errorData ) ) ; return context ; } public removeContext ( context ) { context . close ( ) ; this . contexts . delete ( context ) ; } public getHeadOffset ( ) { return this . head . offset ; } returns public setHead ( head ) { if ( head . offset > this . head . offset ) { this . head = head ; return true ; } return false ; } public setTail ( tail ) { assert ( tail . offset > this . tail . offset && tail . offset <= this . head . offset , `${ tail . offset } > ${ this . tail . offset } && ${ tail . offset } <= ${ this . head . offset }` ) ; this . tail = tail ; this . updateCheckpoint ( ) ; } public close ( ) { this . closed = true ; for ( const context of this . contexts ) { context . close ( ) ; } this . contexts . clear ( ) ; this . removeAllListeners ( ) ; } private updateCheckpoint ( ) { if ( this . closed ) { return ; } let queuedMessage = this . tail ; for ( const context of this . contexts ) { if ( context . hasPendingWork ( ) ) { queuedMessage = queuedMessage . offset > context . tail . offset ? context . tail : queuedMessage ; } } if ( queuedMessage . offset !== this . lastCheckpoint . offset ) { this . partitionContext . checkpoint ( queuedMessage ) ; this . lastCheckpoint = queuedMessage ; } } }