Support terminal resizes

examples/typescript/attach/attach-resize-example.ts

@@ -0,0 +1,29 @@
+import * as k8s from '@kubernetes/client-node';
+
+const kc = new k8s.KubeConfig();
+kc.loadFromDefault();
+
+const terminalSizeQueue = new k8s.TerminalSizeQueue();
+terminalSizeQueue.resize(getSize(process.stdout));
+process.stdout.on('resize', () => {
+    terminalSizeQueue.resize(getSize(process.stdout));
+});
+
+const attach = new k8s.Attach(kc);
+attach.attach(
+    'default',
+    'nginx-4217019353-9gl4s',
+    'nginx',
+    process.stdout,
+    process.stderr,
+    null /* stdin */,
+    false /* tty */,
+    terminalSizeQueue,
+);
+
+function getSize(writeStream: NodeJS.WriteStream): k8s.TerminalSize {
+    return {
+        height: writeStream.rows!,
+        width: writeStream.columns!,
+    };
+}

examples/typescript/exec/exec-resize-example.ts

@@ -0,0 +1,39 @@
+import * as k8s from '@kubernetes/client-node';
+import * as stream from 'stream';
+
+const command = process.argv[2];
+
+const kc = new k8s.KubeConfig();
+kc.loadFromDefault();
+
+const terminalSizeQueue = new k8s.TerminalSizeQueue();
+terminalSizeQueue.resize(getSize(process.stdout));
+process.stdout.on('resize', () => {
+    terminalSizeQueue.resize(getSize(process.stdout));
+});
+
+const exec = new k8s.Exec(kc);
+exec.exec(
+    'tutor',
+    'tutor-environment-operator-deployment-c888b5cd8-qp95f',
+    'default',
+    command,
+    process.stdout as stream.Writable,
+    process.stderr as stream.Writable,
+    process.stdin as stream.Readable,
+    true /* tty */,
+    (status: k8s.V1Status) => {
+        // tslint:disable-next-line:no-console
+        console.log('Exited with status:');
+        // tslint:disable-next-line:no-console
+        console.log(JSON.stringify(status, null, 2));
+    },
+    terminalSizeQueue,
+);
+
+function getSize(writeStream: NodeJS.WriteStream): k8s.TerminalSize {
+    return {
+        height: writeStream.rows!,
+        width: writeStream.columns!,
+    };
+}

src/attach.ts

@@ -3,6 +3,7 @@ import querystring = require('querystring');
 import stream = require('stream');
 
 import { KubeConfig } from './config';
+import { TerminalSizeQueue } from './terminal-size-queue';
 import { WebSocketHandler, WebSocketInterface } from './web-socket-handler';
 
 export class Attach {
@@ -24,6 +25,7 @@ export class Attach {
         stderr: stream.Writable | any,
         stdin: stream.Readable | any,
         tty: boolean,
+        terminalSizeQueue?: TerminalSizeQueue,
     ): Promise<WebSocket> {
         const query = {
             container: containerName,
@@ -43,9 +45,12 @@ export class Attach {
             },
         );
         if (stdin != null) {
-            WebSocketHandler.handleStandardInput(conn, stdin);
+            WebSocketHandler.handleStandardInput(conn, stdin, WebSocketHandler.StdinStream);
         }
 
+        if (terminalSizeQueue != null) {
+            WebSocketHandler.handleStandardInput(conn, terminalSizeQueue, WebSocketHandler.ResizeStream);
+        }
         return conn;
     }
 }

src/attach_test.ts

@@ -1,10 +1,13 @@
 import { expect } from 'chai';
+import { EventEmitter } from 'events';
 import WebSocket = require('isomorphic-ws');
 import { ReadableStreamBuffer, WritableStreamBuffer } from 'stream-buffers';
 import { anyFunction, anything, capture, instance, mock, verify, when } from 'ts-mockito';
+import { CallAwaiter, matchBuffer } from '../test';
 
 import { Attach } from './attach';
 import { KubeConfig } from './config';
+import { TerminalSize, TerminalSizeQueue } from './terminal-size-queue';
 import { WebSocketHandler, WebSocketInterface } from './web-socket-handler';
 
 describe('Attach', () => {
@@ -46,11 +49,14 @@ describe('Attach', () => {
 
         it('should correctly attach to streams', async () => {
             const kc = new KubeConfig();
-            const fakeWebSocket: WebSocketInterface = mock(WebSocketHandler);
-            const attach = new Attach(kc, instance(fakeWebSocket));
+            const fakeWebSocketInterface: WebSocketInterface = mock(WebSocketHandler);
+            const fakeWebSocket: WebSocket = mock(WebSocket);
+            const callAwaiter: CallAwaiter = new CallAwaiter();
+            const attach = new Attach(kc, instance(fakeWebSocketInterface));
             const osStream = new WritableStreamBuffer();
             const errStream = new WritableStreamBuffer();
             const isStream = new ReadableStreamBuffer();
+            const terminalSizeQueue = new TerminalSizeQueue();
 
             const namespace = 'somenamespace';
             const pod = 'somepod';
@@ -59,11 +65,24 @@ describe('Attach', () => {
             const path = `/api/v1/namespaces/${namespace}/pods/${pod}/attach`;
             const args = `container=${container}&stderr=true&stdin=true&stdout=true&tty=false`;
 
-            const fakeConn: WebSocket = mock(WebSocket);
-            when(fakeWebSocket.connect(`${path}?${args}`, null, anyFunction())).thenResolve(fakeConn);
-
-            await attach.attach(namespace, pod, container, osStream, errStream, isStream, false);
-            const [, , outputFn] = capture(fakeWebSocket.connect).last();
+            const fakeConn: WebSocket = instance(fakeWebSocket);
+            when(fakeWebSocketInterface.connect(`${path}?${args}`, null, anyFunction())).thenResolve(
+                fakeConn,
+            );
+            when(fakeWebSocket.send(anything())).thenCall(callAwaiter.resolveCall('send'));
+            when(fakeWebSocket.close()).thenCall(callAwaiter.resolveCall('close'));
+
+            await attach.attach(
+                namespace,
+                pod,
+                container,
+                osStream,
+                errStream,
+                isStream,
+                false,
+                terminalSizeQueue,
+            );
+            const [, , outputFn] = capture(fakeWebSocketInterface.connect).last();
 
             /* tslint:disable:no-unused-expression */
             expect(outputFn).to.not.be.null;
@@ -91,11 +110,23 @@ describe('Attach', () => {
             }
 
             const msg = 'This is test data';
+            const inputPromise = callAwaiter.awaitCall('send');
             isStream.put(msg);
-            verify(fakeConn.send(msg));
-
+            await inputPromise;
+            verify(fakeWebSocket.send(matchBuffer(WebSocketHandler.StdinStream, msg))).called();
+
+            const terminalSize: TerminalSize = { height: 80, width: 120 };
+            const resizePromise = callAwaiter.awaitCall('send');
+            terminalSizeQueue.resize(terminalSize);
+            await resizePromise;
+            verify(
+                fakeWebSocket.send(matchBuffer(WebSocketHandler.ResizeStream, JSON.stringify(terminalSize))),
+            ).called();
+
+            const closePromise = callAwaiter.awaitCall('close');
             isStream.stop();
-            verify(fakeConn.close());
+            await closePromise;
+            verify(fakeWebSocket.close()).called();
         });
     });
 });

src/exec.ts

@@ -4,6 +4,7 @@ import stream = require('stream');
 
 import { V1Status } from './api';
 import { KubeConfig } from './config';
+import { TerminalSizeQueue } from './terminal-size-queue';
 import { WebSocketHandler, WebSocketInterface } from './web-socket-handler';
 
 export class Exec {
@@ -40,6 +41,7 @@ export class Exec {
         stdin: stream.Readable | null,
         tty: boolean,
         statusCallback?: (status: V1Status) => void,
+        terminalSizeQueue?: TerminalSizeQueue,
     ): Promise<WebSocket> {
         const query = {
             stdout: stdout != null,
@@ -66,7 +68,10 @@ export class Exec {
             },
         );
         if (stdin != null) {
-            WebSocketHandler.handleStandardInput(conn, stdin);
+            WebSocketHandler.handleStandardInput(conn, stdin, WebSocketHandler.StdinStream);
+        }
+        if (terminalSizeQueue != null) {
+            WebSocketHandler.handleStandardInput(conn, terminalSizeQueue, WebSocketHandler.ResizeStream);
         }
         return conn;
     }

src/exec_test.ts

@@ -1,11 +1,13 @@
 import { expect } from 'chai';
 import WebSocket = require('isomorphic-ws');
 import { ReadableStreamBuffer, WritableStreamBuffer } from 'stream-buffers';
-import { anyFunction, capture, instance, mock, verify, when } from 'ts-mockito';
+import { anyFunction, anything, capture, instance, mock, verify, when } from 'ts-mockito';
 
+import { CallAwaiter, matchBuffer } from '../test';
 import { V1Status } from './api';
 import { KubeConfig } from './config';
 import { Exec } from './exec';
+import { TerminalSize, TerminalSizeQueue } from './terminal-size-queue';
 import { WebSocketHandler, WebSocketInterface } from './web-socket-handler';
 
 describe('Exec', () => {
@@ -55,11 +57,14 @@ describe('Exec', () => {
 
         it('should correctly attach to streams', async () => {
             const kc = new KubeConfig();
-            const fakeWebSocket: WebSocketInterface = mock(WebSocketHandler);
-            const exec = new Exec(kc, instance(fakeWebSocket));
+            const fakeWebSocketInterface: WebSocketInterface = mock(WebSocketHandler);
+            const fakeWebSocket: WebSocket = mock(WebSocket);
+            const callAwaiter: CallAwaiter = new CallAwaiter();
+            const exec = new Exec(kc, instance(fakeWebSocketInterface));
             const osStream = new WritableStreamBuffer();
             const errStream = new WritableStreamBuffer();
             const isStream = new ReadableStreamBuffer();
+            const terminalSizeQueue = new TerminalSizeQueue();
 
             const namespace = 'somenamespace';
             const pod = 'somepod';
@@ -71,8 +76,12 @@ describe('Exec', () => {
 
             let statusOut = {} as V1Status;
 
-            const fakeConn: WebSocket = mock(WebSocket);
-            when(fakeWebSocket.connect(`${path}?${args}`, null, anyFunction())).thenResolve(fakeConn);
+            const fakeConn: WebSocket = instance(fakeWebSocket);
+            when(fakeWebSocketInterface.connect(`${path}?${args}`, null, anyFunction())).thenResolve(
+                fakeConn,
+            );
+            when(fakeWebSocket.send(anything())).thenCall(callAwaiter.resolveCall('send'));
+            when(fakeWebSocket.close()).thenCall(callAwaiter.resolveCall('close'));
 
             await exec.exec(
                 namespace,
@@ -86,9 +95,10 @@ describe('Exec', () => {
                 (status: V1Status) => {
                     statusOut = status;
                 },
+                terminalSizeQueue,
             );
 
-            const [, , outputFn] = capture(fakeWebSocket.connect).last();
+            const [, , outputFn] = capture(fakeWebSocketInterface.connect).last();
 
             /* tslint:disable:no-unused-expression */
             expect(outputFn).to.not.be.null;
@@ -116,19 +126,30 @@ describe('Exec', () => {
             }
 
             const msg = 'This is test data';
+            const inputPromise = callAwaiter.awaitCall('send');
             isStream.put(msg);
-            verify(fakeConn.send(msg));
+            await inputPromise;
+            verify(fakeWebSocket.send(matchBuffer(WebSocketHandler.StdinStream, msg))).called();
+
+            const terminalSize: TerminalSize = { height: 80, width: 120 };
+            const resizePromise = callAwaiter.awaitCall('send');
+            terminalSizeQueue.resize(terminalSize);
+            await resizePromise;
+            verify(
+                fakeWebSocket.send(matchBuffer(WebSocketHandler.ResizeStream, JSON.stringify(terminalSize))),
+            ).called();
 
             const statusIn = {
                 code: 100,
                 message: 'this is a test',
             } as V1Status;
-
             outputFn(WebSocketHandler.StatusStream, Buffer.from(JSON.stringify(statusIn)));
             expect(statusOut).to.deep.equal(statusIn);
 
+            const closePromise = callAwaiter.awaitCall('close');
             isStream.stop();
-            verify(fakeConn.close());
+            await closePromise;
+            verify(fakeWebSocket.close()).called();
         });
     });
 });

src/index.ts

@@ -8,3 +8,4 @@ export * from './portforward';
 export * from './types';
 export * from './yaml';
 export * from './log';
+export * from './terminal-size-queue';

src/terminal-size-queue.ts

@@ -0,0 +1,20 @@
+import { Readable, ReadableOptions } from 'stream';
+
+export interface TerminalSize {
+    height: number;
+    width: number;
+}
+
+export class TerminalSizeQueue extends Readable {
+    constructor(opts: ReadableOptions = {}) {
+        super({
+            ...opts,
+            // tslint:disable-next-line:no-empty
+            read() {},
+        });
+    }
+
+    public resize(size: TerminalSize) {
+        this.push(JSON.stringify(size));
+    }
+}

src/web-socket-handler.ts

@@ -19,6 +19,7 @@ export class WebSocketHandler implements WebSocketInterface {
     public static readonly StdoutStream = 1;
     public static readonly StderrStream = 2;
     public static readonly StatusStream = 3;
+    public static readonly ResizeStream = 4;
 
     public static handleStandardStreams(
         streamNum: number,

test/call-awaiter.ts

@@ -0,0 +1,13 @@
+import { EventEmitter } from 'events';
+
+export class CallAwaiter extends EventEmitter {
+  public awaitCall(event: string) {
+    return new Promise<any[]>((resolve) => {
+      this.once(event, resolve);
+    });
+  }
+
+  public resolveCall(event: string) {
+    return (...args: any[]) => this.emit(event, ...args);
+  }
+}

test/index.ts

@@ -0,0 +1,2 @@
+export * from './call-awaiter';
+export * from './match-buffer';

test/match-buffer.ts

@@ -0,0 +1,26 @@
+import { Matcher } from 'ts-mockito/lib/matcher/type/Matcher';
+
+export function matchBuffer(channel: number, contents: string): Matcher {
+  return new StringBufferMatcher(channel, contents);
+}
+
+class StringBufferMatcher extends Matcher {
+  constructor(private channel: number, private contents: string) {
+      super();
+  }
+
+  public match(value: any): boolean {
+      if (value instanceof Buffer) {
+          const buffer = value as Buffer;
+          const channel: number = buffer.readInt8(0);
+          const contents: string = buffer.toString('utf8', 1);
+          return this.channel === channel && this.contents === contents;
+      }
+
+      return false;
+  }
+
+  public toString(): string {
+      return `buffer did not contain "${this.contents}"`;
+  }
+}