Handle JSON parse errors in watch implementation

package.json

@@ -53,9 +53,9 @@
     "@types/request": "^2.47.1",
     "@types/underscore": "^1.8.9",
     "@types/ws": "^6.0.1",
-    "byline": "^5.0.0",
     "isomorphic-ws": "^4.0.1",
     "js-yaml": "^3.12.0",
+    "json-stream": "^1.0.0",
     "jsonpath-plus": "^0.19.0",
     "request": "^2.88.0",
     "shelljs": "^0.8.2",

src/watch.ts

@@ -1,4 +1,4 @@
-import { LineStream } from 'byline';
+import JSONStream from 'json-stream';
 import request = require('request');
 import { KubeConfig } from './config';
 
@@ -55,21 +55,8 @@ export class Watch {
         };
         this.config.applyToRequest(requestOptions);
 
-        const stream = new LineStream();
-        stream.on('data', (data) => {
-            let obj: WatchUpdate;
-            if (data instanceof Buffer) {
-                obj = JSON.parse(data.toString()) as WatchUpdate;
-            } else {
-                obj = JSON.parse(data) as WatchUpdate;
-            }
-            if (typeof obj === 'object' && obj.object) {
-                callback(obj.type, obj.object);
-            } else {
-                throw new Error(`unexpected ${typeof obj}: ${JSON.stringify(obj)}`);
-            }
-        });
-
+        const stream = new JSONStream();
+        stream.on('data', (data) => callback(data.type, data.object));
         const req = this.requestImpl.webRequest(requestOptions, (error, response, body) => {
             if (error) {
                 done(error);

src/watch_test.ts

@@ -7,6 +7,32 @@ import { KubeConfig } from './config';
 import { Cluster, Context, User } from './config_types';
 import { DefaultRequest, Watch } from './watch';
 
+const server = 'foo.company.com';
+
+const fakeConfig: {
+    clusters: Cluster[];
+    contexts: Context[];
+    users: User[];
+} = {
+    clusters: [
+        {
+            name: 'cluster',
+            server,
+        } as Cluster,
+    ],
+    contexts: [
+        {
+            cluster: 'cluster',
+            user: 'user',
+        } as Context,
+    ],
+    users: [
+        {
+            name: 'user',
+        } as User,
+    ],
+};
+
 describe('Watch', () => {
     it('should construct correctly', () => {
         const kc = new KubeConfig();
@@ -15,24 +41,7 @@ describe('Watch', () => {
 
     it('should watch correctly', () => {
         const kc = new KubeConfig();
-        const server = 'foo.company.com';
-        kc.clusters = [
-            {
-                name: 'cluster',
-                server,
-            } as Cluster,
-        ] as Cluster[];
-        kc.contexts = [
-            {
-                cluster: 'cluster',
-                user: 'user',
-            } as Context,
-        ] as Context[];
-        kc.users = [
-            {
-                name: 'user',
-            } as User,
-        ];
+        Object.assign(kc, fakeConfig);
         const fakeRequestor = mock(DefaultRequest);
         const watch = new Watch(kc, instance(fakeRequestor));
 
@@ -102,4 +111,52 @@ describe('Watch', () => {
         doneCallback(errIn, null, null);
         expect(doneErr).to.deep.equal(errIn);
     });
+
+    it('should ignore JSON parse errors', () => {
+        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 = {
+            pipe: (stream) => {
+                stream.write(JSON.stringify(obj) + '\n');
+                stream.write('{"truncated json\n');
+            },
+        };
+
+        when(fakeRequestor.webRequest(anything(), anyFunction())).thenReturn(fakeRequest);
+
+        const path = '/some/path/to/object';
+
+        const receivedTypes: string[] = [];
+        const receivedObjects: string[] = [];
+
+        watch.watch(
+            path,
+            {},
+            (recievedType: string, recievedObject: string) => {
+                receivedTypes.push(recievedType);
+                receivedObjects.push(recievedObject);
+            },
+            () => {
+                /* ignore */
+            },
+        );
+
+        verify(fakeRequestor.webRequest(anything(), anyFunction()));
+
+        const [opts, doneCallback] = capture(fakeRequestor.webRequest).last();
+        const reqOpts: request.OptionsWithUri = opts as request.OptionsWithUri;
+
+        expect(receivedTypes).to.deep.equal([obj.type]);
+        expect(receivedObjects).to.deep.equal([obj.object]);
+    });
 });