feat(client-node): add support for watch streaming (#486)

backends/kubernetes-client-node.js

@@ -1,6 +1,7 @@
 'use strict'
 
 const pascalcase = require('pascalcase')
+const { Readable } = require('stream')
 
 //
 // https://github.com/kubernetes-client/javascript
@@ -28,7 +29,26 @@ class ClientNodeBackend {
   }
 
   async getWatchObjectStream (options) {
-    throw new Error('getWatchObjectStream is not implemented')
+    const watch = new this.client.Watch(this.kubeconfig)
+
+    const stream = new Readable({
+      objectMode: true,
+      read: () => { /* .watch callback pushes to stream below */ },
+      destroy: (err, cb) => {
+        req.destroy(err)
+        req.abort()
+        cb(err)
+      }
+    })
+
+    const req = watch.watch(
+      options.pathname,
+      Object.assign({}, options.qs, options.parameters),
+      (type, object) => stream.push({ type, object }),
+      err => stream.destroy(err)
+    )
+
+    return stream
   }
 
   http (options) {

test-integration/stream.js

@@ -1,6 +1,7 @@
 /* eslint-env mocha */
 'use strict'
 
+const expect = require('chai').expect
 const fs = require('fs')
 const yaml = require('js-yaml')
 
@@ -44,14 +45,18 @@ describe('test-integration/stream', () => {
 
     const stream = await client.api.v1.watch.namespaces(namespace).pods.getObjectStream()
 
+    let gotData = false
     await new Promise((resolve, reject) => {
       stream.on('data', data => {
+        gotData = true
+        expect(['ADDED', 'DELETED', 'MODIFIED']).to.include(data.type)
+        expect(data.object).to.be.an('object')
         stream.destroy()
         resolve()
       })
       stream.on('error', err => {
-        stream.abort()
-        reject(err)
+        stream.destroy()
+        if (!gotData) reject(err)
       })
     })
   })