Add an informer implementation.

Author: brendandburns

examples/cache-example.js

@@ -10,7 +10,8 @@ const watch = new k8s.Watch(kc);
 const listFn = (fn) => {
     k8sApi.listNamespacedPod('default')
         .then((res) => {
-            fn(res.body.items);
+            const podList = res.body;
+            fn(podList.items, podList.metadata.resourceVersin);
         })
         .catch((err) => {
             console.log(err);

src/cache.ts

@@ -1,3 +1,4 @@
+import { Informer, ObjectCallback } from './informer';
 import { KubernetesObject } from './types';
 import { Watch } from './watch';
 
@@ -6,11 +7,12 @@ export interface ObjectCache<T> {
     list(namespace?: string): ReadonlyArray<T>;
 }
 
-export type ListCallback<T extends KubernetesObject> = (list: T[]) => void;
+export type ListCallback<T extends KubernetesObject> = (list: T[], ResourceVersion: string) => void;
 
-export class ListWatch<T extends KubernetesObject> implements ObjectCache<T> {
+export class ListWatch<T extends KubernetesObject> implements ObjectCache<T>, Informer<T> {
     private objects: T[] = [];
     private readonly indexCache: { [key: string]: T[] } = {};
+    private readonly callbackCache: { [key: string]: ObjectCallback<T> } = {};
 
     public constructor(
         private readonly path: string,
@@ -22,6 +24,10 @@ export class ListWatch<T extends KubernetesObject> implements ObjectCache<T> {
         this.doneHandler(null);
     }
 
+    public on(verb: string, cb: ObjectCallback<T>) {
+        this.callbackCache[verb] = cb;
+    }
+
     public get(name: string, namespace?: string): T | undefined {
         return this.objects.find(
             (obj: T): boolean => {
@@ -38,12 +44,12 @@ export class ListWatch<T extends KubernetesObject> implements ObjectCache<T> {
     }
 
     private doneHandler(err: any) {
-        this.listFn((result: T[]) => {
+        this.listFn((result: T[], resourceVersion: string) => {
             this.objects = result;
             for (const elt of this.objects) {
                 this.indexObj(elt);
             }
-            this.watch.watch(this.path, {}, this.watchHandler.bind(this), this.doneHandler.bind(this));
+            this.watch.watch(this.path, { resourceVersion }, this.watchHandler.bind(this), this.doneHandler.bind(this));
         });
     }
 
@@ -53,24 +59,24 @@ export class ListWatch<T extends KubernetesObject> implements ObjectCache<T> {
             namespaceList = [];
             this.indexCache[obj.metadata!.namespace!] = namespaceList;
         }
-        addOrUpdateObject(namespaceList, obj);
+        addOrUpdateObject(namespaceList, obj, this.callbackCache['add'], this.callbackCache['update']);
     }
 
     private watchHandler(phase: string, obj: T) {
         switch (phase) {
             case 'ADDED':
             case 'MODIFIED':
-                addOrUpdateObject(this.objects, obj);
+                addOrUpdateObject(this.objects, obj,  this.callbackCache['add'], this.callbackCache['update']);
                 if (obj.metadata!.namespace) {
                     this.indexObj(obj);
                 }
                 break;
             case 'DELETED':
-                deleteObject(this.objects, obj);
+                deleteObject(this.objects, obj, this.callbackCache['delete']);
                 if (obj.metadata!.namespace) {
                     const namespaceList = this.indexCache[obj.metadata!.namespace!] as T[];
                     if (namespaceList) {
-                        deleteObject(namespaceList, obj);
+                        deleteObject(namespaceList, obj, this.callbackCache['delete']);
                     }
                 }
                 break;
@@ -79,12 +85,16 @@ export class ListWatch<T extends KubernetesObject> implements ObjectCache<T> {
 }
 
 // Only public for testing.
-export function addOrUpdateObject<T extends KubernetesObject>(objects: T[], obj: T) {
+export function addOrUpdateObject<T extends KubernetesObject>(
+    objects: T[], obj: T,
+    addCallback: ObjectCallback<T>, updateCallback: ObjectCallback<T>) {
     const ix = findKubernetesObject(objects, obj);
     if (ix === -1) {
         objects.push(obj);
+        addCallback(obj);
     } else {
         objects[ix] = obj;
+        updateCallback(obj);
     }
 }
 
@@ -99,9 +109,11 @@ function findKubernetesObject<T extends KubernetesObject>(objects: T[], obj: T):
 }
 
 // Public for testing.
-export function deleteObject<T extends KubernetesObject>(objects: T[], obj: T) {
+export function deleteObject<T extends KubernetesObject>(
+    objects: T[], obj: T, deleteCallback: ObjectCallback<T>) {
     const ix = findKubernetesObject(objects, obj);
     if (ix !== -1) {
         objects.splice(ix, 1);
+        deleteCallback(obj);
     }
 }

src/informer.ts

@@ -0,0 +1,7 @@
+import { KubernetesObject } from './types';
+
+export type ObjectCallback<T extends KubernetesObject> = (obj: T) => void;
+
+export interface Informer<T> {
+    on(verb: string, fn: ObjectCallback<T>);
+}