Merge pull request #369 from brendandburns/watch

Add some missing error handling.

Author: k8s-ci-robot

src/watch.ts

@@ -18,6 +18,7 @@ export class DefaultRequest implements RequestInterface {
 }
 
 export class Watch {
+    public static SERVER_SIDE_CLOSE = { error: 'Connection closed on server' };
     public config: KubeConfig;
     private readonly requestImpl: RequestInterface;
 
@@ -64,6 +65,10 @@ export class Watch {
                 // ignore parse errors
             }
         });
+        stream.on('error', (err) => {
+            done(err);
+        });
+        stream.on('close', () => done(null));
         const req = this.requestImpl.webRequest(requestOptions, (error, response, body) => {
             if (error) {
                 done(error);

src/watch_test.ts

@@ -112,6 +112,123 @@ describe('Watch', () => {
         expect(doneErr).to.deep.equal(errIn);
     });
 
+    it('should handle errors correctly', () => {
+        const kc = new KubeConfig();
+        Object.assign(kc, fakeConfig);
+        const fakeRequestor = mock(DefaultRequest);
+        const watch = new Watch(kc, instance(fakeRequestor));
+
+        const obj1 = {
+            type: 'ADDED',
+            object: {
+                foo: 'bar',
+            },
+        };
+
+        const errIn = { error: 'err' };
+        const fakeRequest = {
+            pipe: (stream) => {
+                stream.write(JSON.stringify(obj1) + '\n');
+                stream.emit('error', errIn);
+            },
+        };
+
+        when(fakeRequestor.webRequest(anything(), anyFunction())).thenReturn(fakeRequest);
+
+        const path = '/some/path/to/object';
+
+        const receivedTypes: string[] = [];
+        const receivedObjects: string[] = [];
+        let doneCalled = false;
+        let doneErr: any;
+
+        watch.watch(
+            path,
+            {},
+            (phase: string, obj: string) => {
+                receivedTypes.push(phase);
+                receivedObjects.push(obj);
+            },
+            (err: any) => {
+                doneCalled = true;
+                doneErr = err;
+            },
+        );
+
+        verify(fakeRequestor.webRequest(anything(), anyFunction()));
+
+        const [opts, doneCallback] = capture(fakeRequestor.webRequest).last();
+        const reqOpts: request.OptionsWithUri = opts as request.OptionsWithUri;
+
+        expect(reqOpts.uri).to.equal(`${server}${path}`);
+        expect(reqOpts.method).to.equal('GET');
+        expect(reqOpts.json).to.equal(true);
+
+        expect(receivedTypes).to.deep.equal([obj1.type]);
+        expect(receivedObjects).to.deep.equal([obj1.object]);
+
+        expect(doneCalled).to.equal(true);
+        expect(doneErr).to.deep.equal(errIn);
+    });
+
+    it('should handle server side close correctly', () => {
+        const kc = new KubeConfig();
+        Object.assign(kc, fakeConfig);
+        const fakeRequestor = mock(DefaultRequest);
+        const watch = new Watch(kc, instance(fakeRequestor));
+
+        const obj1 = {
+            type: 'ADDED',
+            object: {
+                foo: 'bar',
+            },
+        };
+
+        const fakeRequest = {
+            pipe: (stream) => {
+                stream.write(JSON.stringify(obj1) + '\n');
+                stream.emit('close');
+            },
+        };
+
+        when(fakeRequestor.webRequest(anything(), anyFunction())).thenReturn(fakeRequest);
+
+        const path = '/some/path/to/object';
+
+        const receivedTypes: string[] = [];
+        const receivedObjects: string[] = [];
+        let doneCalled = false;
+        let doneErr: any;
+
+        watch.watch(
+            path,
+            {},
+            (phase: string, obj: string) => {
+                receivedTypes.push(phase);
+                receivedObjects.push(obj);
+            },
+            (err: any) => {
+                doneCalled = true;
+                doneErr = err;
+            },
+        );
+
+        verify(fakeRequestor.webRequest(anything(), anyFunction()));
+
+        const [opts, doneCallback] = capture(fakeRequestor.webRequest).last();
+        const reqOpts: request.OptionsWithUri = opts as request.OptionsWithUri;
+
+        expect(reqOpts.uri).to.equal(`${server}${path}`);
+        expect(reqOpts.method).to.equal('GET');
+        expect(reqOpts.json).to.equal(true);
+
+        expect(receivedTypes).to.deep.equal([obj1.type]);
+        expect(receivedObjects).to.deep.equal([obj1.object]);
+
+        expect(doneCalled).to.equal(true);
+        expect(doneErr).to.be.null;
+    });
+
     it('should ignore JSON parse errors', () => {
         const kc = new KubeConfig();
         Object.assign(kc, fakeConfig);