Merge pull request #606 from dominykas/stop

Expose a public Informer.stop()

Author: k8s-ci-robot

src/cache.ts

@@ -13,6 +13,7 @@ export class ListWatch<T extends KubernetesObject> implements ObjectCache<T>, In
     private readonly indexCache: { [key: string]: T[] } = {};
     private readonly callbackCache: { [key: string]: Array<ObjectCallback<T>> } = {};
     private request: RequestResult | undefined;
+    private stopped: boolean = false;
 
     public constructor(
         private readonly path: string,
@@ -31,14 +32,13 @@ export class ListWatch<T extends KubernetesObject> implements ObjectCache<T>, In
     }
 
     public async start(): Promise<void> {
+        this.stopped = false;
         await this.doneHandler(null);
     }
 
-    public stop(): void {
-        if (this.request) {
-            this.request.abort();
-            this.request = undefined;
-        }
+    public async stop(): Promise<void> {
+        this.stopped = true;
+        this._stop();
     }
 
     public on(verb: string, cb: ObjectCallback<T>): void {
@@ -80,12 +80,23 @@ export class ListWatch<T extends KubernetesObject> implements ObjectCache<T>, In
         return this.resourceVersion;
     }
 
+    private _stop(): void {
+        if (this.request) {
+            this.request.abort();
+            this.request = undefined;
+        }
+    }
+
     private async doneHandler(err: any): Promise<any> {
-        this.stop();
+        this._stop();
         if (err) {
             this.callbackCache[ERROR].forEach((elt: ObjectCallback<T>) => elt(err));
             return;
         }
+        if (this.stopped) {
+            // do not auto-restart
+            return;
+        }
         // TODO: Don't always list here for efficiency
         // try to restart the watch from resourceVersion, but detect 410 GONE and relist in that case.
         // Or if resourceVersion is empty.

src/cache_test.ts

@@ -9,7 +9,7 @@ import { EventEmitter } from 'ws';
 
 import { V1Namespace, V1NamespaceList, V1ObjectMeta, V1Pod, V1ListMeta } from './api';
 import { deleteObject, ListWatch, deleteItems } from './cache';
-import { ADD, UPDATE, DELETE, ListPromise } from './informer';
+import { ADD, UPDATE, DELETE, ERROR, ListPromise } from './informer';
 
 use(chaiAsPromised);
 
@@ -577,6 +577,7 @@ describe('ListWatchCache', () => {
         expect(cache.list('ns2').length).to.equal(0);
         expect(cache.get('name2', 'ns2')).to.equal(undefined);
     });
+
     it('should delete an object correctly', () => {
         const list: V1Pod[] = [
             {
@@ -614,6 +615,7 @@ describe('ListWatchCache', () => {
         } as V1Pod);
         expect(list.length).to.equal(1);
     });
+
     it('should not call handlers which have been unregistered', async () => {
         const fakeWatch = mock.mock(Watch);
         const list: V1Namespace[] = [];
@@ -721,6 +723,7 @@ describe('ListWatchCache', () => {
 
         expect(addedList.length).to.equal(1);
     });
+
     it('should resolve start promise after seeding the cache', async () => {
         const fakeWatch = mock.mock(Watch);
         const list: V1Namespace[] = [
@@ -823,6 +826,127 @@ describe('ListWatchCache', () => {
         expect(addedList1.length).to.equal(2);
         expect(addedList2.length).to.equal(1);
     });
+
+    it('should not auto-restart after explicitly stopping until restarted again', async () => {
+        const fakeWatch = mock.mock(Watch);
+        const list: V1Pod[] = [
+            {
+                metadata: {
+                    name: 'name1',
+                    namespace: 'ns1',
+                } as V1ObjectMeta,
+            } as V1Pod,
+            {
+                metadata: {
+                    name: 'name2',
+                    namespace: 'ns2',
+                } as V1ObjectMeta,
+            } as V1Pod,
+        ];
+        const listObj = {
+            metadata: {
+                resourceVersion: '12345',
+            } as V1ListMeta,
+            items: list,
+        } as V1NamespaceList;
+
+        const listFn: ListPromise<V1Namespace> = function(): Promise<{
+            response: http.IncomingMessage;
+            body: V1NamespaceList;
+        }> {
+            return new Promise<{ response: http.IncomingMessage; body: V1NamespaceList }>(
+                (resolve, reject) => {
+                    resolve({ response: {} as http.IncomingMessage, body: listObj });
+                },
+            );
+        };
+        let promise = new Promise((resolve) => {
+            mock.when(
+                fakeWatch.watch(mock.anything(), mock.anything(), mock.anything(), mock.anything()),
+            ).thenCall(() => {
+                resolve(new FakeRequest());
+            });
+        });
+
+        const cache = new ListWatch('/some/path', mock.instance(fakeWatch), listFn);
+        await promise;
+
+        const [, , , doneHandler] = mock.capture(fakeWatch.watch).last();
+
+        // stop the informer
+        cache.stop();
+
+        await doneHandler(null);
+
+        mock.verify(
+            fakeWatch.watch(mock.anything(), mock.anything(), mock.anything(), mock.anything()),
+        ).once();
+
+        // restart the informer
+        await cache.start();
+
+        mock.verify(
+            fakeWatch.watch(mock.anything(), mock.anything(), mock.anything(), mock.anything()),
+        ).twice();
+    });
+
+    it('does not auto-restart after an error', async () => {
+        const fakeWatch = mock.mock(Watch);
+        const list: V1Pod[] = [
+            {
+                metadata: {
+                    name: 'name1',
+                    namespace: 'ns1',
+                } as V1ObjectMeta,
+            } as V1Pod,
+            {
+                metadata: {
+                    name: 'name2',
+                    namespace: 'ns2',
+                } as V1ObjectMeta,
+            } as V1Pod,
+        ];
+        const listObj = {
+            metadata: {
+                resourceVersion: '12345',
+            } as V1ListMeta,
+            items: list,
+        } as V1NamespaceList;
+
+        const listFn: ListPromise<V1Namespace> = function(): Promise<{
+            response: http.IncomingMessage;
+            body: V1NamespaceList;
+        }> {
+            return new Promise<{ response: http.IncomingMessage; body: V1NamespaceList }>(
+                (resolve, reject) => {
+                    resolve({ response: {} as http.IncomingMessage, body: listObj });
+                },
+            );
+        };
+        let promise = new Promise((resolve) => {
+            mock.when(
+                fakeWatch.watch(mock.anything(), mock.anything(), mock.anything(), mock.anything()),
+            ).thenCall(() => {
+                resolve(new FakeRequest());
+            });
+        });
+
+        const cache = new ListWatch('/some/path', mock.instance(fakeWatch), listFn);
+        await promise;
+
+        let errorEmitted = false;
+        cache.on(ERROR, () => (errorEmitted = true));
+
+        const [, , , doneHandler] = mock.capture(fakeWatch.watch).last();
+
+        const error = new Error('testing');
+        await doneHandler(error);
+
+        mock.verify(
+            fakeWatch.watch(mock.anything(), mock.anything(), mock.anything(), mock.anything()),
+        ).once();
+        expect(errorEmitted).to.equal(true);
+    });
 });
 
 describe('delete items', () => {

src/informer.ts

@@ -21,6 +21,7 @@ export interface Informer<T> {
     on(verb: string, fn: ObjectCallback<T>): void;
     off(verb: string, fn: ObjectCallback<T>): void;
     start(): Promise<void>;
+    stop(): Promise<void>;
 }
 
 export function makeInformer<T>(