import { extractBoxcar , IContext , IQueuedMessage , IPartitionConfig , IPartitionLambda , IPartitionLambdaFactory , LambdaCloseType , IContextErrorData , IDocumentLambdaServerConfiguration , } from "@fluidframework/server-services-core" ; import { DocumentContextManager } from "./contextManager" ; import { DocumentPartition } from "./documentPartition" ; export class DocumentLambda implements IPartitionLambda { private readonly documents = new Map < string , DocumentPartition > ( ) ; private readonly contextManager : DocumentContextManager ; private activityCheckTimer : NodeJS . Timeout | undefined ; constructor ( private readonly factory , private readonly config , private readonly context , private readonly documentLambdaServerConfiguration ) { this . contextManager = new DocumentContextManager ( context ) ; this . contextManager . on ( "error" , ( error , errorData ) => { context . error ( error , errorData ) ; } ) ; this . activityCheckTimer = setInterval ( this . inactivityCheck . bind ( this ) , documentLambdaServerConfiguration . partitionActivityCheckInterval ) ; } public handler ( message ) { if ( ! this . contextManager . setHead ( message ) ) { this . context . log ?. warn ( "Unexpected head offset. " + `head offset: ${ this . contextManager . getHeadOffset ( ) }, message offset: ${ message . offset }` ) ; return undefined ; } this . handlerCore ( message ) ; this . contextManager . setTail ( message ) ; return undefined ; } public close ( closeType ) { if ( this . activityCheckTimer !== undefined ) { clearInterval ( this . activityCheckTimer ) ; this . activityCheckTimer = undefined ; } this . contextManager . close ( ) ; for ( const [ , partition ] of this . documents ) { partition . close ( closeType ) ; } this . documents . clear ( ) ; } private handlerCore ( message ) { const boxcar = extractBoxcar ( message ) ; if ( ! boxcar . documentId || ! boxcar . tenantId ) { return ; } message . value = boxcar ; const routingKey = `${ boxcar . tenantId }/${ boxcar . documentId }` ; let document = this . documents . get ( routingKey ) ; if ( ! document ) { const documentContext = this . contextManager . createContext ( boxcar , message ) ; document = new DocumentPartition ( this . factory , this . config , boxcar . tenantId , boxcar . documentId , documentContext , this . documentLambdaServerConfiguration . partitionActivityTimeout ) ; this . documents . set ( routingKey , document ) ; } else { document . context . setHead ( message ) ; } document . process ( message ) ; } /**
     * Closes inactive documents
     */ private inactivityCheck ( ) { const now = Date . now ( ) ; const documentPartitions = Array . from ( this . documents ) ; for ( const [ routingKey , documentPartition ] of documentPartitions ) { if ( documentPartition . isInactive ( now ) ) { this . contextManager . removeContext ( documentPartition . context ) ; documentPartition . close ( LambdaCloseType . ActivityTimeout ) ; this . documents . delete ( routingKey ) ; } } } }