import { EventEmitter } from "events" ; import { IConsumer , IQueuedMessage , IPartitionConfig , IPartitionLambda , IPartitionLambdaFactory , ILogger , LambdaCloseType , IContextErrorData , } from "@fluidframework/server-services-core" ; import { QueueObject , queue } from "async" ; import * as _ from "lodash" ; import { Lumberjack } from "@fluidframework/server-services-telemetry" ; import { CheckpointManager } from "./checkpointManager" ; import { Context } from "./context" ; /**
 * Partition of a message stream. Manages routing messages to individual handlers. And then maintaining the
 * overall partition offset.
 */ export class Partition extends EventEmitter { private readonly q : QueueObject < IQueuedMessage > ; private lambdaP : Promise < IPartitionLambda > | undefined ; private lambda : IPartitionLambda | undefined ; private readonly checkpointManager : CheckpointManager ; private readonly context : Context ; private closed = false ; constructor ( private readonly id , leaderEpoch , factory , consumer , private readonly logger ? ) { super ( ) ; const partitionConfig = { leaderEpoch } ; this . checkpointManager = new CheckpointManager ( id , consumer ) ; this . context = new Context ( this . checkpointManager , this . logger ) ; this . context . on ( "error" , ( error , errorData ) => { this . emit ( "error" , error , errorData ) ; } ) ; this . q = queue ( ( message , callback ) => { try { const optionalPromise = this . lambda ! . handler ( message ) ; if ( optionalPromise ) { optionalPromise . then ( callback as any ) . catch ( callback ) ; return ; } callback ( ) ; } catch ( error ) { callback ( error ) ; } } , 1 ) ; this . q . pause ( ) ; this . lambdaP = factory . create ( partitionConfig , this . context ) ; this . lambdaP . then ( ( lambda ) => { this . lambda = lambda ; this . lambdaP = undefined ; this . q . resume ( ) ; } , ( error ) => { if ( this . closed ) { return ; } const errorData = { restart : true } ; this . emit ( "error" , error , errorData ) ; this . q . kill ( ) ; } ) ; this . q . error ( ( error ) => { const errorData = { restart : true } ; this . emit ( "error" , error , errorData ) ; } ) ; } public process ( rawMessage ) { if ( this . closed ) { return ; } void this . q . push ( rawMessage ) ; } public close ( closeType ) { this . closed = true ; this . q . kill ( ) ; this . checkpointManager . close ( ) ; this . context . close ( ) ; if ( this . lambda ) { this . lambda . close ( closeType ) ; this . lambda = undefined ; } else if ( this . lambdaP ) { this . lambdaP . then ( ( lambda ) => { lambda . close ( closeType ) ; } , ( error ) => { } ) . finally ( ( ) => { this . lambda = undefined ; this . lambdaP = undefined ; } ) ; } this . removeAllListeners ( ) ; } /**
     * Stops processing on the partition
     */ public async drain ( ) { const drainedP = new Promise < void > ( ( resolve , reject ) => { if ( this . q . length ( ) === 0 ) { this . logger ?. info ( `No pending work for partition ${ this . id }. Exiting early` ) ; Lumberjack . info ( `No pending work for partition ${ this . id }. Exiting early` ) ; return resolve ( ) ; } this . logger ?. info ( `Waiting for queue to drain for partition ${ this . id }` ) ; Lumberjack . info ( `Waiting for queue to drain for partition ${ this . id }` ) ; this . q . drain ( ( ) => { this . logger ?. info ( `Drained partition ${ this . id }` ) ; Lumberjack . info ( `Drained partition ${ this . id }` ) ; resolve ( ) ; } ) ; } ) ; await drainedP ; await this . checkpointManager . flush ( ) ; } }