import byline = require ( 'byline' ) ; import request = require ( 'request' ) ; import { Duplex } from 'stream' ; import { KubeConfig } from './config' ; export interface WatchUpdate { type : string ; object : object ; } export interface RequestResult { pipe ( stream ) : void ; on ( ev , cb ) : void ; abort ( ) : void ; } export interface Response { statusCode : number ; statusMessage : string ; } export interface RequestInterface { webRequest ( opts ) : RequestResult ; } export class DefaultRequest implements RequestInterface { private requestImpl : ( opts : request . OptionsWithUri ) => request . Request ; constructor ( requestImpl ? ) { this . requestImpl = requestImpl ? requestImpl : request ; } public webRequest ( opts ) { const req = this . requestImpl ( opts ) ; req . pause ( ) ; req . on ( 'response' , ( resp ) => { if ( resp . statusCode === 200 ) { req . resume ( ) ; } else { const error = new Error ( resp . statusMessage ) as Error & { statusCode : number | undefined ; } ; error . statusCode = resp . statusCode ; req . emit ( 'error' , error ) ; } } ) ; return req ; } } export class Watch { public static SERVER_SIDE_CLOSE : object = { error : 'Connection closed on server' } ; public config : KubeConfig ; private readonly requestImpl : RequestInterface ; public constructor ( config , requestImpl ? ) { this . config = config ; if ( requestImpl ) { this . requestImpl = requestImpl ; } else { this . requestImpl = new DefaultRequest ( ) ; } } public async watch ( path , queryParams , callback , done ) { const cluster = this . config . getCurrentCluster ( ) ; if ( ! cluster ) { throw new Error ( 'No currently active cluster' ) ; } const url = cluster . server + path ; queryParams . watch = true ; const headerParams = { } ; const requestOptions = { method : 'GET' , qs : queryParams , headers : headerParams , uri : url , useQuerystring : true , json : true , pool : false , } ; await this . config . applyToRequest ( requestOptions ) ; let req ; let doneCalled = false ; const doneCallOnce = ( err ) => { if ( ! doneCalled ) { req . abort ( ) ; doneCalled = true ; done ( err ) ; } } ; req = this . requestImpl . webRequest ( requestOptions ) ; const stream = byline . createStream ( ) ; req . on ( 'error' , doneCallOnce ) ; req . on ( 'socket' , ( socket ) => { socket . setTimeout ( 30000 ) ; socket . setKeepAlive ( true , 30000 ) ; } ) ; stream . on ( 'error' , doneCallOnce ) ; stream . on ( 'close' , ( ) => doneCallOnce ( null ) ) ; stream . on ( 'data' , ( line ) => { try { const data = JSON . parse ( line ) ; callback ( data . type , data . object , data ) ; } catch ( ignore ) { } } ) ; req . pipe ( stream ) ; return req ; } }