Merge pull request kubernetes-client#857 from Nicklason/master

Move watch callback out of try-catch

Author: k8s-ci-robot

src/watch.ts

@@ -121,12 +121,16 @@ export class Watch {
         stream.on('error', doneCallOnce);
         stream.on('close', () => doneCallOnce(null));
         stream.on('data', (line) => {
+            let data;
+
             try {
-                const data = JSON.parse(line);
-                callback(data.type, data.object, data);
+                data = JSON.parse(line);
             } catch (ignore) {
                 // ignore parse errors
+                return;
             }
+
+            callback(data.type, data.object, data);
         });
 
         req.pipe(stream);

src/watch_test.ts

@@ -364,6 +364,67 @@ describe('Watch', () => {
         expect(receivedObjects).to.deep.equal([obj.object]);
     });
 
+    it('should not ignore handler errors', async () => {
+        const kc = new KubeConfig();
+        Object.assign(kc, fakeConfig);
+        const fakeRequestor = mock(DefaultRequest);
+        const watch = new Watch(kc, instance(fakeRequestor));
+
+        const obj = {
+            type: 'MODIFIED',
+            object: {
+                baz: 'blah',
+            },
+        };
+
+        const fakeRequest = new FakeRequest();
+        fakeRequest.pipe = function(stream) {
+            stream.write(JSON.stringify(obj) + '\n');
+            stream.emit('close');
+        };
+
+        when(fakeRequestor.webRequest(anything())).thenReturn(fakeRequest);
+
+        const path = '/some/path/to/object';
+
+        const receivedTypes: string[] = [];
+        const receivedObjects: string[] = [];
+        let doneCalled = false;
+        let doneErr: any;
+        let catchCalled = false;
+        let catchErr: any;
+
+        const handlerError = new Error('handler error');
+
+        await watch.watch(
+            path,
+            {},
+            (recievedType: string, recievedObject: string) => {
+                receivedTypes.push(recievedType);
+                receivedObjects.push(recievedObject);
+                throw handlerError;
+            },
+            (err: any) => {
+                doneCalled = true;
+                doneErr = err;
+            },
+        ).catch((err) => {
+            catchCalled = true;
+            catchErr = err;
+        });
+
+        verify(fakeRequestor.webRequest(anything()));
+
+        expect(receivedTypes).to.deep.equal([obj.type]);
+        expect(receivedObjects).to.deep.equal([obj.object]);
+
+        expect(doneCalled).to.equal(false);
+        expect(doneErr).to.be.undefined;
+
+        expect(catchCalled).to.equal(true);
+        expect(catchErr).to.equal(handlerError);
+    });
+
     it('should throw on empty config', () => {
         const kc = new KubeConfig();
         const watch = new Watch(kc);