import { IQueuedMessage } from "./queue" ; import { IContext , IContextErrorData } from "./lambdas" ; /**
 * Allows checkpointing the minimum offset for multiple lambdas
 * This is useful to use for a CombinedLambda
 */ export class CombinedContext { private currentCheckpoint : IQueuedMessage | undefined ; private readonly checkpoints : ( IQueuedMessage | undefined ) [ ] ; constructor ( private readonly context ) { this . checkpoints = [ ] ; } public createContext ( ) { const id = this . checkpoints . push ( undefined ) - 1 ; return { log : this . context . log , checkpoint : ( message ) => this . checkpoint ( id , message ) , error : ( error , errorData ) => this . error ( id , error , errorData ) } ; } private checkpoint ( id , queuedMessage ) { this . checkpoints [ id ] = queuedMessage ; const lowestMessage = this . getLowestMessage ( ) ; if ( lowestMessage !== undefined && ( this . currentCheckpoint === undefined || this . currentCheckpoint . offset < lowestMessage . offset ) ) { this . currentCheckpoint = lowestMessage ; this . context . checkpoint ( lowestMessage ) ; } } private error ( _id , error , errorData ) { this . context . error ( error , errorData ) ; } /**
     * Returns the lowest checkpoint for all the lambdas
     * This will return undefined if a lambda has not checkpointed
     */ private getLowestMessage ( ) { let lowestMessage ; for ( const message of this . checkpoints ) { if ( message === undefined ) { return undefined ; } if ( lowestMessage === undefined || lowestMessage . offset > message . offset ) { lowestMessage = message ; } } return lowestMessage ; } }