import WebSocket = require ( 'isomorphic-ws' ) ; import stream = require ( 'stream' ) ; import { V1Status } from './api' ; import { KubeConfig } from './config' ; const protocols = [ 'v4.channel.k8s.io' , 'v3.channel.k8s.io' , 'v2.channel.k8s.io' , 'channel.k8s.io' ] ; export interface WebSocketInterface { connect ( path , textHandler , binaryHandler ) : Promise < WebSocket > ; } export class WebSocketHandler implements WebSocketInterface { public static readonly StdinStream : number = 0 ; public static readonly StdoutStream : number = 1 ; public static readonly StderrStream : number = 2 ; public static readonly StatusStream : number = 3 ; public static readonly ResizeStream : number = 4 ; public static handleStandardStreams ( streamNum , buff , stdout , stderr ) { if ( buff . length < 1 ) { return null ; } if ( stdout && streamNum === WebSocketHandler . StdoutStream ) { stdout . write ( buff ) ; } else if ( stderr && streamNum === WebSocketHandler . StderrStream ) { stderr . write ( buff ) ; } else if ( streamNum === WebSocketHandler . StatusStream ) { if ( stdout && stdout !== process . stdout ) { stdout . end ( ) ; } if ( stderr && stderr !== process . stderr ) { stderr . end ( ) ; } return JSON . parse ( buff . toString ( 'utf8' ) ) as V1Status ; } else { throw new Error ( 'Unknown stream: ' + streamNum ) ; } return null ; } public static handleStandardInput ( ws , stdin , streamNum = 0 ) { stdin . on ( 'data' , ( data ) => { const buff = Buffer . alloc ( data . length + 1 ) ; buff . writeInt8 ( streamNum , 0 ) ; if ( data instanceof Buffer ) { data . copy ( buff , 1 ) ; } else { buff . write ( data , 1 ) ; } ws . send ( buff ) ; } ) ; stdin . on ( 'end' , ( ) => { ws . close ( ) ; } ) ; return true ; } public static async processData ( data , ws , createWS , streamNum = 0 , retryCount = 3 ) { const buff = Buffer . alloc ( data . length + 1 ) ; buff . writeInt8 ( streamNum , 0 ) ; if ( data instanceof Buffer ) { data . copy ( buff , 1 ) ; } else { buff . write ( data , 1 ) ; } let i = 0 ; for ( ; i < retryCount ; ++ i ) { if ( ws !== null && ws . readyState === WebSocket . OPEN ) { ws . send ( buff ) ; break ; } else { ws = await createWS ( ) ; } } if ( i >= retryCount ) { throw new Error ( "can't send data to ws" ) ; } return ws ; } public static restartableHandleStandardInput ( createWS , stdin , streamNum = 0 , retryCount = 3 ) { if ( retryCount < 0 ) { throw new Error ( "retryCount can't be lower than 0." ) ; } let queue = Promise . resolve ( ) ; let ws = null ; stdin . on ( 'data' , ( data ) => { queue = queue . then ( async ( ) => { ws = await WebSocketHandler . processData ( data , ws , createWS , streamNum , retryCount ) ; } ) ; } ) ; stdin . on ( 'end' , ( ) => { if ( ws ) { ws . close ( ) ; } } ) ; return ( ) => ws ; } public constructor ( readonly config , readonly socketFactory ? ) { } param path param textHandler param binaryHandler public async connect ( path , textHandler , binaryHandler ) { const cluster = this . config . getCurrentCluster ( ) ; if ( ! cluster ) { throw new Error ( 'No cluster is defined.' ) ; } const server = cluster . server ; const ssl = server . startsWith ( 'https://' ) ; const target = ssl ? server . substr ( 8 ) : server . substr ( 7 ) ; const proto = ssl ? 'wss' : 'ws' ; const uri = `${ proto }://${ target }${ path }` ; const opts = { } ; await this . config . applytoHTTPSOptions ( opts ) ; return await new Promise < WebSocket > ( ( resolve , reject ) => { const client = this . socketFactory ? this . socketFactory ( uri , opts ) : new WebSocket ( uri , protocols , opts ) ; let resolved = false ; client . onopen = ( ) => { resolved = true ; resolve ( client ) ; } ; client . onerror = ( err ) => { if ( ! resolved ) { reject ( err ) ; } } ; client . onmessage = ( { data } ) => { if ( typeof data === 'string' ) { if ( textHandler && ! textHandler ( data ) ) { client . close ( ) ; } } else if ( data instanceof Buffer ) { const streamNum = data . readInt8 ( 0 ) ; if ( binaryHandler && ! binaryHandler ( streamNum , data . slice ( 1 ) ) ) { client . close ( ) ; } } } ; } ) ; } }