import { ObjectSerializer } from './api' ; import { ADD , CHANGE , CONNECT , DELETE , ERROR , ErrorCallback , Informer , ListPromise , ObjectCallback , UPDATE , } from './informer' ; import { KubernetesObject } from './types' ; import { RequestResult , Watch } from './watch' ; export interface ObjectCache < T > { get ( name , namespace ? ) : T | undefined ; list ( namespace ? ) : ReadonlyArray < T > ; } export class ListWatch < T extends KubernetesObject > implements ObjectCache < T > , Informer < T > { private objects : T [ ] = [ ] ; private resourceVersion : string ; private readonly indexCache : { [ key ] : T [ ] ; } = { } ; private readonly callbackCache : { [ key ] : Array < ObjectCallback < T > | ErrorCallback > ; } = { } ; private request : RequestResult | undefined ; private stopped : boolean = false ; public constructor ( private readonly path , private readonly watch , private readonly listFn , autoStart = true , private readonly labelSelector ? ) { this . callbackCache [ ADD ] = [ ] ; this . callbackCache [ UPDATE ] = [ ] ; this . callbackCache [ DELETE ] = [ ] ; this . callbackCache [ ERROR ] = [ ] ; this . callbackCache [ CONNECT ] = [ ] ; this . resourceVersion = '' ; if ( autoStart ) { this . doneHandler ( null ) ; } } public async start ( ) { this . stopped = false ; await this . doneHandler ( null ) ; } public async stop ( ) { this . stopped = true ; this . _stop ( ) ; } public on ( verb , cb ) ; public on ( verb , cb ) ; public on ( verb , cb ) { if ( verb === CHANGE ) { this . on ( 'add' , cb ) ; this . on ( 'update' , cb ) ; this . on ( 'delete' , cb ) ; return ; } if ( this . callbackCache [ verb ] === undefined ) { throw new Error ( `Unknown verb: ${ verb }` ) ; } this . callbackCache [ verb ] . push ( cb ) ; } public off ( verb , cb ) ; public off ( verb , cb ) ; public off ( verb , cb ) { if ( verb === CHANGE ) { this . off ( 'add' , cb ) ; this . off ( 'update' , cb ) ; this . off ( 'delete' , cb ) ; return ; } if ( this . callbackCache [ verb ] === undefined ) { throw new Error ( `Unknown verb: ${ verb }` ) ; } const indexToRemove = this . callbackCache [ verb ] . findIndex ( ( cachedCb ) => cachedCb === cb ) ; if ( indexToRemove === - 1 ) { return ; } this . callbackCache [ verb ] . splice ( indexToRemove , 1 ) ; } public get ( name , namespace ? ) { return this . objects . find ( ( obj ) : boolean => { return obj . metadata ! . name === name && ( ! namespace || obj . metadata ! . namespace === namespace ) ; } ) ; } public list ( namespace ? ) { if ( ! namespace ) { return this . objects ; } return this . indexCache [ namespace ] as ReadonlyArray < T > ; } public latestResourceVersion ( ) { return this . resourceVersion ; } private _stop ( ) { if ( this . request ) { this . request . abort ( ) ; this . request = undefined ; } } private async doneHandler ( err ) { this . _stop ( ) ; if ( err && err . statusCode === 410 ) { this . resourceVersion = '' ; } else if ( err ) { this . callbackCache [ ERROR ] . forEach ( ( elt ) => elt ( err ) ) ; return ; } if ( this . stopped ) { return ; } this . callbackCache [ CONNECT ] . forEach ( ( elt ) => elt ( undefined ) ) ; if ( ! this . resourceVersion ) { const promise = this . listFn ( ) ; const result = await promise ; const list = result . body ; this . objects = deleteItems ( this . objects , list . items , this . callbackCache [ DELETE ] . slice ( ) ) ; Object . keys ( this . indexCache ) . forEach ( ( key ) => { const updateObjects = deleteItems ( this . indexCache [ key ] , list . items ) ; if ( updateObjects . length !== 0 ) { this . indexCache [ key ] = updateObjects ; } else { delete this . indexCache [ key ] ; } } ) ; this . addOrUpdateItems ( list . items ) ; this . resourceVersion = list . metadata ! . resourceVersion ! ; } const queryParams = { resourceVersion : this . resourceVersion , } as { resourceVersion : string | undefined ; labelSelector : string | undefined ; } ; if ( this . labelSelector !== undefined ) { queryParams . labelSelector = ObjectSerializer . serialize ( this . labelSelector , 'string' ) ; } this . request = await this . watch . watch ( this . path , queryParams , this . watchHandler . bind ( this ) , this . doneHandler . bind ( this ) ) ; } private addOrUpdateItems ( items ) { items . forEach ( ( obj ) => { addOrUpdateObject ( this . objects , obj , this . callbackCache [ ADD ] . slice ( ) , this . callbackCache [ UPDATE ] . slice ( ) ) ; if ( obj . metadata ! . namespace ) { this . indexObj ( obj ) ; } } ) ; } private indexObj ( obj ) { let namespaceList = this . indexCache [ obj . metadata ! . namespace ! ] as T [ ] ; if ( ! namespaceList ) { namespaceList = [ ] ; this . indexCache [ obj . metadata ! . namespace ! ] = namespaceList ; } addOrUpdateObject ( namespaceList , obj ) ; } private watchHandler ( phase , obj , watchObj ? ) { switch ( phase ) { case 'ADDED' : case 'MODIFIED' : addOrUpdateObject ( this . objects , obj , this . callbackCache [ ADD ] . slice ( ) , this . callbackCache [ UPDATE ] . slice ( ) ) ; if ( obj . metadata ! . namespace ) { this . indexObj ( obj ) ; } break ; case 'DELETED' : deleteObject ( this . objects , obj , this . callbackCache [ DELETE ] . slice ( ) ) ; if ( obj . metadata ! . namespace ) { const namespaceList = this . indexCache [ obj . metadata ! . namespace ! ] as T [ ] ; if ( namespaceList ) { deleteObject ( namespaceList , obj ) ; } } break ; case 'BOOKMARK' : break ; } if ( watchObj && watchObj . metadata ) { this . resourceVersion = watchObj . metadata . resourceVersion ; } } } export function deleteItems < T extends KubernetesObject > ( oldObjects , newObjects , deleteCallback ? ) { return oldObjects . filter ( ( obj ) => { if ( findKubernetesObject ( newObjects , obj ) === - 1 ) { if ( deleteCallback ) { deleteCallback . forEach ( ( fn ) => fn ( obj ) ) ; } return false ; } return true ; } ) ; } export function addOrUpdateObject < T extends KubernetesObject > ( objects , obj , addCallback ? , updateCallback ? ) { const ix = findKubernetesObject ( objects , obj ) ; if ( ix === - 1 ) { objects . push ( obj ) ; if ( addCallback ) { addCallback . forEach ( ( elt ) => elt ( obj ) ) ; } } else { if ( ! isSameVersion ( objects [ ix ] , obj ) ) { objects [ ix ] = obj ; if ( updateCallback ) { updateCallback . forEach ( ( elt ) => elt ( obj ) ) ; } } } } function isSameObject < T extends KubernetesObject > ( o1 , o2 ) { return o1 . metadata ! . name === o2 . metadata ! . name && o1 . metadata ! . namespace === o2 . metadata ! . namespace ; } function isSameVersion < T extends KubernetesObject > ( o1 , o2 ) { return ( o1 . metadata ! . resourceVersion !== undefined && o1 . metadata ! . resourceVersion !== null && o1 . metadata ! . resourceVersion === o2 . metadata ! . resourceVersion ) ; } function findKubernetesObject < T extends KubernetesObject > ( objects , obj ) { return objects . findIndex ( ( elt ) => { return isSameObject ( elt , obj ) ; } ) ; } export function deleteObject < T extends KubernetesObject > ( objects , obj , deleteCallback ? ) { const ix = findKubernetesObject ( objects , obj ) ; if ( ix !== - 1 ) { objects . splice ( ix , 1 ) ; if ( deleteCallback ) { deleteCallback . forEach ( ( elt ) => elt ( obj ) ) ; } } }