Handle JSON parse errors in watch implementation

This change "handles" JSON parse parse errors in the watch implementation by
ignoring them: the JSONStream implementation ignores parsing errors.

The request might result in invalid JSON if, for example, the HTTP connection
is abruptly terminated (and a partial line is piped to the transform stream).

The code this commit replaces would raise an uncaught exception exception on
parse errors. An alternative approach to this commit would invoke the `done`
callback with parse errors, but that could result in multiple invocations of
the `done` callback.

Author: Silas Boyd-Wickizer

package-lock.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",

package.json

@@ -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.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]);
+    });
 });