Fix sending protocols for web sockets.

Author: brendandburns

examples/typescript/exec/exec-example.ts

@@ -6,5 +6,10 @@ const kc = new k8s.KubeConfig();
 kc.loadFromDefault();
 
 const exec = new k8s.Exec(kc);
-exec.exec('default', 'nginx-4217019353-9gl4s', 'nginx', command,
-    process.stdout, process.stderr, process.stdin, true /* tty */);
+exec.exec('default', 'demo', 'demo', command,
+    process.stdout, process.stderr, process.stdin,
+    (status: k8s.V1Status) => {
+        console.log('Exited with status:');
+        console.log(JSON.stringify(status, null, 2));
+    },
+    true /* tty */);

src/exec.ts

@@ -2,6 +2,7 @@ import WebSocket = require('isomorphic-ws');
 import querystring = require('querystring');
 import stream = require('stream');
 
+import { V1Status } from './api';
 import { KubeConfig } from './config';
 import { WebSocketHandler, WebSocketInterface } from './web-socket-handler';
 
@@ -18,7 +19,8 @@ export class Exec {
 
     // TODO: make command an array and support multiple args
     public async exec(namespace: string, podName: string, containerName: string, command: string,
-                      stdout: stream.Writable | any, stderr: stream.Writable | any, stdin: stream.Readable | any,
+                      stdout: stream.Writable | null, stderr: stream.Writable | null, stdin: stream.Readable | null,
+                      statusCallback: (status: V1Status) => void,
                       tty: boolean): Promise<WebSocket> {
         const query = {
             stdout: stdout != null,
@@ -31,7 +33,13 @@ export class Exec {
         const queryStr = querystring.stringify(query);
         const path = `/api/v1/namespaces/${namespace}/pods/${podName}/exec?${queryStr}`;
         const conn = await this.handler.connect(path, null, (streamNum: number, buff: Buffer): boolean => {
-            WebSocketHandler.handleStandardStreams(streamNum, buff, stdout, stderr);
+            const status = WebSocketHandler.handleStandardStreams(streamNum, buff, stdout, stderr);
+            if (status != null) {
+                if (statusCallback) {
+                    statusCallback(status);
+                }
+                return false;
+            }
             return true;
         });
         if (stdin != null) {

src/exec_test.ts

@@ -8,6 +8,9 @@ import { Exec } from './exec';
 import { WebSocketHandler, WebSocketInterface } from './web-socket-handler';
 
 describe('Exec', () => {
+    // tslint:disable-next-line:no-empty
+    const nop = () => {};
+
     describe('basic', () => {
         it('should correctly exec to a url', async () => {
             const kc = new KubeConfig();
@@ -24,27 +27,27 @@ describe('Exec', () => {
             const path = `/api/v1/namespaces/${namespace}/pods/${pod}/exec`;
 
             await exec.exec(
-                namespace, pod, container, cmd, osStream, errStream, isStream, false);
+                namespace, pod, container, cmd, osStream, errStream, isStream, nop, false);
             let args = `stdout=true&stderr=true&stdin=true&tty=false&command=${cmd}&container=${container}`;
             verify(fakeWebSocket.connect(`${path}?${args}`, null,  anyFunction())).called();
 
             await exec.exec(
-                namespace, pod, container, cmd, null, errStream, isStream, false);
+                namespace, pod, container, cmd, null, errStream, isStream, nop, false);
             args = `stdout=false&stderr=true&stdin=true&tty=false&command=${cmd}&container=${container}`;
             verify(fakeWebSocket.connect(`${path}?${args}`, null,  anyFunction())).called();
 
             await exec.exec(
-                namespace, pod, container, cmd, null, null, isStream, false);
+                namespace, pod, container, cmd, null, null, isStream, nop, false);
             args = `stdout=false&stderr=false&stdin=true&tty=false&command=${cmd}&container=${container}`;
             verify(fakeWebSocket.connect(`${path}?${args}`, null,  anyFunction())).called();
 
             await exec.exec(
-                namespace, pod, container, cmd, null, null, null, false);
+                namespace, pod, container, cmd, null, null, null, nop, false);
             args = `stdout=false&stderr=false&stdin=false&tty=false&command=${cmd}&container=${container}`;
             verify(fakeWebSocket.connect(`${path}?${args}`, null,  anyFunction())).called();
 
             await exec.exec(
-                namespace, pod, container, cmd, null, errStream, isStream, true);
+                namespace, pod, container, cmd, null, errStream, isStream, nop, true);
             args = `stdout=false&stderr=true&stdin=true&tty=true&command=${cmd}&container=${container}`;
             verify(fakeWebSocket.connect(`${path}?${args}`, null,  anyFunction())).called();
         });
@@ -69,7 +72,7 @@ describe('Exec', () => {
             when(fakeWebSocket.connect(`${path}?${args}`, null, anyFunction())).thenResolve(fakeConn);
 
             await exec.exec(
-                namespace, pod, container, cmd, osStream, errStream, isStream, false);
+                namespace, pod, container, cmd, osStream, errStream, isStream, nop, false);
 
             const [, , outputFn] = capture(fakeWebSocket.connect).last();
 

src/web-socket-handler.ts

@@ -27,21 +27,21 @@ export class WebSocketHandler implements WebSocketInterface {
 
     public static handleStandardStreams(
         streamNum: number, buff: Buffer,
-        stdout: stream.Writable, stderr: stream.Writable,
+        stdout: stream.Writable | null, stderr: stream.Writable | null,
     ): V1Status | null {
         if (buff.length < 1) {
             return null;
         }
-        if (streamNum === WebSocketHandler.StdoutStream) {
+        if (stdout && streamNum === WebSocketHandler.StdoutStream) {
             stdout.write(buff);
-        } else if (streamNum === WebSocketHandler.StderrStream) {
+        } else if (stderr && streamNum === WebSocketHandler.StderrStream) {
             stderr.write(buff);
         } else if (streamNum === WebSocketHandler.StatusStream) {
             // stream closing.
-            if (stdout) {
+            if (stdout && stdout !== process.stdout) {
                 stdout.end();
             }
-            if (stderr) {
+            if (stderr && stderr !== process.stderr) {
                 stderr.end();
             }
             return JSON.parse(buff.toString('utf8')) as V1Status;
@@ -109,7 +109,7 @@ export class WebSocketHandler implements WebSocketInterface {
         this.config.applytoHTTPSOptions(opts);
 
         return new Promise((resolve, reject) => {
-            const client = (this.socketFactory ? this.socketFactory(uri, opts) : new WebSocket(uri, opts));
+            const client = (this.socketFactory ? this.socketFactory(uri, opts) : new WebSocket(uri, protocols, opts));
             let resolved = false;
 
             client.onopen = () => {