/**
 * Copyright (c) 2011 Bruno Jouhier <bruno.jouhier@sage.com>
 * MIT License
 */ import { _ } from 'streamline-runtime' ; import { parse as parseUrl } from 'url' ; import * as os from 'os' ; import * as generic from './devices/generic' ; import { Reader } from './reader' ; import { Writer } from './writer' ; import * as http from 'http' ; import * as https from 'https' ; import * as net from 'net' ; export interface Emitter extends NodeJS . EventEmitter { end ? : ( data ? : any , encoding ? : string ) => void ; close ? : ( ) => void ; destroySoon ? : ( ) => void ; } function nop ( ) { } export class Wrapper < EmitterT extends Emitter > { _emitter : EmitterT ; _closed : boolean ; _onClose : ( err ? : Error ) => void ; _autoClosed : ( ( ) => void ) [ ] ; _doesNotEmitClose : boolean ; constructor ( emitter ) { this . _emitter = emitter ; this . _closed = false ; emitter . on ( 'close' , ( ) => { this . _onClose && this . _onClose ( ) ; } ) ; this . _autoClosed = [ ] ; this . _onClose = this . _trackClose ; } _trackClose ( ) { this . _closed = true ; this . _autoClosed . forEach ( fn => { fn . call ( this ) ; } ) ; } close ( _ ) { _ . cast ( ( callback ) => { if ( typeof callback !== 'function' ) throw new TypeError ( "bad callback parameter: " + typeof callback ) ; if ( this . _closed ) return callback ( ) ; const close = this . _emitter . end || this . _emitter . close || this . _emitter . destroySoon ; if ( typeof close !== "function" ) return callback ( ) ; this . _onClose = err => { this . _closed = true ; this . _onClose = nop ; callback ( err ) ; callback = nop ; } ; if ( this . _doesNotEmitClose ) { this . _emitter . emit ( "close" ) ; } close . call ( this . _emitter ) ; } ) ( _ ) ; } get closed ( ) { return this . _closed ; } unwrap ( ) { this . _emitter . removeAllListeners ( ) ; this . _closed = true ; return this . _emitter ; } get emitter ( ) { return this . _emitter ; } } export interface ReadableOptions { lowMark ? : number ; highMark ? : number ; } export type Data = string | Buffer ; export class ReadableStream < EmitterT extends NodeJS . ReadableStream > extends Wrapper < EmitterT > { _low : number ; _high : number ; _paused : boolean ; _current : number ; _chunks : Data [ ] ; _error : Error ; _done : boolean ; _encoding : string | null ; _onData : ( err ? : Error , chunk ? : Data ) => void ; reader : Reader < any > ; constructor ( emitter , options ? ) { super ( emitter ) ; options = options || { } ; this . _low = Math . max ( options . lowMark || 0 , 0 ) ; this . _high = Math . max ( options . highMark || 0 , this . _low ) ; this . _paused = false ; this . _current = 0 ; this . _chunks = [ ] ; this . _done = false ; this . _onData = this . _trackData ; emitter . on ( 'error' , ( err ) => { this . _onData ( err ) ; } ) ; emitter . on ( 'data' , ( chunk ) => { this . _onData ( undefined , chunk ) ; } ) ; emitter . on ( 'end' , ( ) => { this . _onData ( ) ; } ) ; this . _autoClosed . push ( ( ) => { if ( ! this . _done ) this . _onData ( new Error ( "stream was closed unexpectedly" ) ) ; } ) ; this . reader = generic . reader ( this . _readChunk . bind ( this ) , this . stop . bind ( this ) ) ; } _trackData ( err , chunk ? ) { if ( err ) this . _error = err ; else if ( chunk ) { this . _chunks . push ( chunk ) ; this . _current += chunk . length ; if ( this . _current > this . _high && ! this . _paused && ! this . _done && ! this . _error && ! this . _closed ) { this . _emitter . pause ( ) ; this . _paused = true ; } } else this . _done = true ; } _readChunk ( callback ) { if ( this . _chunks . length > 0 ) { const chunk = this . _chunks . splice ( 0 , 1 ) [ 0 ] ; this . _current -= chunk . length ; if ( this . _current <= this . _low && this . _paused && ! this . _done && ! this . _error && ! this . _closed ) { this . _emitter . resume ( ) ; this . _paused = false ; } return callback ( undefined , chunk ) ; } else if ( this . _done ) { if ( this . _paused ) { try { if ( ! this . _closed ) this . _emitter . resume ( ) ; this . _paused = false ; } catch ( e ) { } } return callback ( undefined ) ; } else if ( this . _error ) { return callback ( this . _error ) ; } else { var replied = false ; this . _onData = ( err , chunk ) => { if ( err ) this . _error = err ; else if ( ! chunk ) this . _done = true ; this . _onData = this . _trackData ; if ( ! replied ) callback ( err , chunk != null ? chunk : undefined ) ; replied = true ; } ; } } _concat ( chunks , total ) { if ( this . _encoding ) return chunks . join ( '' ) ; if ( chunks . length == 1 ) return chunks [ 0 ] ; const result = Buffer . alloc ( total ) ; chunks . reduce ( ( val , chunk ) => { if ( ! Buffer . isBuffer ( chunk ) ) throw new Error ( 'chunk is not a buffer: ' + typeof chunk ) ; chunk . copy ( result , val ) ; return val + chunk . length ; } , 0 ) ; return result ; } setEncoding ( enc ) { this . _encoding = enc ; if ( enc ) this . _emitter . setEncoding ( enc ) ; return this ; } read ( _ , len ? ) { if ( this . _closed && ! this . _chunks . length ) return undefined ; if ( len == null ) return this . reader . read ( _ ) ; if ( len < 0 ) len = Infinity ; if ( len == 0 ) return this . _encoding ? "" : Buffer . alloc ( 0 ) ; const chunks = [ ] ; var total = 0 ; while ( total < len ) { var chunk = this . reader . read ( _ ) ; if ( ! chunk ) return chunks . length == 0 ? undefined : this . _concat ( chunks , total ) ; if ( total + chunk . length <= len ) { chunks . push ( chunk ) ; total += chunk . length ; } else { chunks . push ( chunk . slice ( 0 , len - total ) ) ; this . unread ( chunk . slice ( len - total ) ) ; total = len ; } } return this . _concat ( chunks , total ) ; } readAll ( _ ) { const result = this . read ( _ , - 1 ) ; return result === undefined ? null : result ; } unread ( chunk ) { if ( chunk ) { this . _chunks . splice ( 0 , 0 , chunk ) ; this . _current += chunk . length ; } return this ; } available ( ) { return this . _chunks . reduce ( ( count , chunk ) => { return count + chunk . length ; } , 0 ) ; } stop ( _ , arg ? ) { if ( arg && arg !== true ) this . _error = this . _error || arg ; this . unwrap ( ) ; } get events ( ) { return [ "error" , "data" , "end" , "close" ] ; } } export interface WritableOptions { encoding ? : string ; } export class WritableStream < EmitterT extends NodeJS . WritableStream > extends Wrapper < EmitterT > { _error : Error ; _onDrain : ( err ? : Error ) => void ; _encoding ? : string ; writer : Writer < Data > ; constructor ( emitter , options ? ) { super ( emitter ) ; options = options || { } ; this . _encoding = options . encoding ; emitter . on ( 'error' , ( err ) => { if ( this . _onDrain ) this . _onDrain ( err ) ; else this . _error = err ; } ) ; emitter . on ( 'drain' , ( ) => { if ( this . _onDrain ) this . _onDrain ( ) ; } ) ; this . _autoClosed . push ( ( ) => { const err = new Error ( "stream was closed unexpectedly" ) ; if ( this . _onDrain ) this . _onDrain ( err ) ; else this . _error = err ; } ) ; this . writer = generic . writer ( ( _ , data ? ) => { if ( this . _error ) throw new Error ( this . _error . message ) ; if ( data != null ) { if ( ! data . length ) return this . writer ; if ( typeof data === "string" ) data = Buffer . from ( data , this . _encoding || "utf8" ) ; if ( ! this . _emitter . write ( data ) ) this . _drain ( _ ) ; } else { _ . cast ( this . _emitter . end ) . call ( this . _emitter , _ ) ; } return this . writer ; } ) ; } _drain ( _ ) { _ . cast ( ( callback ) => { this . _onDrain = ( err ) => { this . _onDrain = nop ; callback ( err ) ; callback = nop ; } ; } ) ( _ ) ; } ; write ( _ , data ? , enc ? ) { if ( typeof data === "string" ) data = Buffer . from ( data , enc || this . _encoding || "utf8" ) ; else if ( data === null ) data = undefined ; this . writer . write ( _ , data ) ; return this ; } end ( data ? , enc ? ) { if ( this . writer . ended ) { if ( data != null ) throw new Error ( "invalid attempt to write after end" ) ; return this ; } if ( typeof data === "string" ) data = Buffer . from ( data , enc || this . _encoding || "utf8" ) ; else if ( data === null ) data = undefined ; if ( data !== undefined ) { _ . run ( _ => this . writer . write ( _ , data ) , err => { if ( err ) throw err ; this . end ( ) ; } ) ; } else { _ . run ( _ => this . writer . write ( _ ) , err => { if ( err ) throw err ; } ) ; } return this ; } get events ( ) { return [ "drain" , "close" ] ; } } export type Headers = { [ key ] : string ; } ; function _getSupportedEncoding ( enc ) { switch ( enc . trim ( ) . toLowerCase ( ) ) { case 'utf8' : case 'utf-8' : return 'utf8' ; case 'utf16le' : case 'utf-16le' : return 'utf16le' ; case 'us-ascii' : return 'ascii' ; case 'iso-8859-1' : case 'win-1252' : return 'binary' ; } return null ; } function _getEncodingDefault ( headers ) { const comps = ( headers [ 'content-type' ] || 'text/plain' ) . split ( ';' ) ; const ctype = comps [ 0 ] ; for ( var i = 1 ; i < comps . length ; i ++ ) { const pair = comps [ i ] . split ( '=' ) ; if ( pair . length == 2 && pair [ 0 ] . trim ( ) == 'charset' ) { return _getSupportedEncoding ( pair [ 1 ] ) ; } } if ( ctype . indexOf ( 'text' ) >= 0 || ctype . indexOf ( 'json' ) >= 0 ) return "utf8" ; return null ; } function _getEncodingStrict ( headers ) { if ( ! headers [ 'content-type' ] || headers [ 'content-encoding' ] ) return null ; const comps = headers [ 'content-type' ] . split ( ';' ) ; const ctype = comps [ 0 ] ; for ( var i = 1 ; i < comps . length ; i ++ ) { const pair = comps [ i ] . split ( '=' ) ; if ( pair . length === 2 && pair [ 0 ] . trim ( ) === 'charset' ) { return _getSupportedEncoding ( pair [ 1 ] ) ; } } return null ; } export interface EncodingOptions { detectEncoding ? : 'strict' | 'disable' | ( ( Headers : Headers ) => string ) ; } function _getEncoding ( headers , options ? ) { if ( headers [ 'content-encoding' ] ) return null ; if ( ! options ) return _getEncodingDefault ( headers ) ; if ( typeof options . detectEncoding === "function" ) return options . detectEncoding ( headers ) ; switch ( options . detectEncoding ) { case 'strict' : return _getEncodingStrict ( headers ) ; case 'disable' : return null ; default : return _getEncodingDefault ( headers ) ; } } export interface HttpServerOptions { createServer ? : ( listener : ( request : http . ServerRequest , response : http . ServerResponse ) => void ) => http . Server | https . Server ; secure ? : boolean ; } export class HttpServerRequest extends ReadableStream < http . ServerRequest > { constructor ( req , options ? ) { super ( req , options ) ; this . setEncoding ( _getEncoding ( req . headers , options ) ) ; if ( ( req as any ) . session ) ( this as any ) . session = ( req as any ) . session ; this . _doesNotEmitClose = true ; } get method ( ) { return this . _emitter . method ! ; } set method ( val ) { this . _emitter . method = val ; } get url ( ) { return this . _emitter . url ! ; } set url ( val ) { this . _emitter . url = val ; } get headers ( ) { return this . _emitter . headers ; } set headers ( val ) { this . _emitter . headers = val ; } get trailers ( ) { return this . _emitter . trailers ; } set trailers ( val ) { this . _emitter . trailers = val ; } get rawHeaders ( ) { return this . _emitter . rawHeaders ; } get rawTrailers ( ) { return this . _emitter . rawTrailers ; } get httpVersion ( ) { return this . _emitter . httpVersion ; } get connection ( ) { return this . _emitter . connection ; } get socket ( ) { return this . _emitter . socket ; } get statusCode ( ) { return this . _emitter . statusCode ; } get statusMessage ( ) { return this . _emitter . statusMessage ; } get client ( ) { return ( this . _emitter as any ) . client ; } } Object . defineProperty ( HttpServerRequest . prototype , '_request' , { get ( this ) { return this . _emitter ; } } ) ; export class HttpServerResponse extends WritableStream < http . ServerResponse > { constructor ( resp , options ? ) { super ( resp , options ) ; this . _doesNotEmitClose = true ; } writeContinue ( ) { this . _emitter . writeContinue ( ) ; return this ; } writeHead ( statusCode , headers ? ) ; writeHead ( statusCode , reasonPhrase ? , headers ? ) { this . _emitter . writeHead ( statusCode , reasonPhrase , headers ) ; return this ; } setHeader ( name , value ) { this . _emitter . setHeader ( name , value ) ; return this ; } getHeader ( name ) { return this . _emitter . getHeader ( name ) ; } removeHeader ( name ) { this . _emitter . removeHeader ( name ) ; return this ; } addTrailers ( trailers ) { this . _emitter . addTrailers ( trailers ) ; return this ; } get statusCode ( ) { return this . _emitter . statusCode ; } set statusCode ( val ) { this . _emitter . statusCode = val ; } get statusMessage ( ) { return this . _emitter . statusMessage ; } set statusMessage ( val ) { this . _emitter . statusMessage = val ; } } function _fixHttpServerOptions ( options ? ) { var opts = options || { } ; opts . createServer = function ( listener ) : http . Server | https . Server { if ( typeof listener !== 'function' ) throw new TypeError ( "bad listener parameter: " + typeof listener ) ; return opts . secure ? https . createServer ( opts , listener ) : http . createServer ( listener ) ; } ; return opts ; } export interface ServerEmitter extends Emitter { listen ( ... args ) : void ; } export class Server < EmitterT extends ServerEmitter > extends Wrapper < EmitterT > { constructor ( emitter ) { super ( emitter ) ; } listen ( _ , ... args ) { return _ . cast ( ( callback ) => { if ( this . _closed ) throw new Error ( "cannot listen: server is closed" ) ; var replied = false ; var reply = ( err , result ? ) => { if ( ! replied ) callback ( err , result ) ; replied = true ; } ; args . push ( ( ) => { reply ( undefined , this ) ; } ) ; this . _autoClosed . push ( ( ) => { reply ( new Error ( "server was closed unexpectedly" ) ) ; } ) ; this . _emitter . on ( 'error' , reply ) ; this . _emitter . listen . apply ( this . _emitter , args ) ; } ) ( _ ) ; } } export type HttpListener = ( request : HttpServerRequest , response : HttpServerResponse , _ : _ ) => void ; export function httpListener ( listener , options ) { options = options || { } ; return ( request , response ) => { return _ . withContext ( ( ) => { return _ . run ( _ => listener ( new HttpServerRequest ( request , options ) , new HttpServerResponse ( response , options ) , _ ) , err => { if ( ! /^(post|put)$/i . test ( request . method || 'get' ) ) request . removeAllListeners ( ) ; if ( err ) throw err ; } ) ; } ) ( ) ; } ; } ; export function createHttpServer ( requestListener , options ) { return new HttpServer ( requestListener , options ) ; } ; export class HttpServer extends Server < http . Server | https . Server > { constructor ( requestListener , options ) { var opts = _fixHttpServerOptions ( options ) ; super ( opts . createServer ! ( httpListener ( requestListener , options ) ) ) ; } setTimeout ( msecs , callback ) { if ( ( this . _emitter as any ) . setTimeout ) ( this . _emitter as http . Server ) . setTimeout ( msecs , callback ) ; return this ; } } export interface HttpClientResponseOptions extends ReadableOptions { } export class HttpClientResponse extends ReadableStream < http . ClientResponse > { constructor ( resp , options ? ) { super ( resp , options ) ; this . setEncoding ( _getEncoding ( resp . headers , options ) ) ; } get statusCode ( ) { return this . _emitter . statusCode ; } get statusMessage ( ) { return this . _emitter . statusMessage ; } get httpVersion ( ) { return this . _emitter . httpVersion ; } get headers ( ) { return this . _emitter . headers ; } get trailers ( ) { return this . _emitter . trailers ; } get rawHeaders ( ) { return this . _emitter . rawHeaders ; } get rawTrailers ( ) { return this . _emitter . rawTrailers ; } checkStatus ( statuses ) { if ( typeof statuses === 'number' ) statuses = [ statuses ] ; if ( this . statusCode == null || statuses . indexOf ( this . statusCode ) < 0 ) throw new Error ( "invalid status: " + this . statusCode ) ; return this ; } ; } export interface HttpClientOptions { url ? : string ; protocol ? : string ; host ? : string ; port ? : string ; path ? : string ; method ? : string ; headers ? : Headers ; module ? : any ; user ? : string ; password ? : string ; proxy ? : any ; proxyAuthenticate ? : any ; isHttps ? : boolean ; socket ? : net . Socket ; agent ? : boolean ; } function _fixHttpClientOptions ( options ) { if ( ! options ) throw new Error ( "request error: no options" ) ; var opts = options ; if ( typeof opts === "string" ) opts = { url : opts } ; if ( opts . url ) { const parsed = parseUrl ( opts . url ) ; opts . protocol = parsed . protocol ; opts . host = parsed . hostname ; opts . port = parsed . port ; opts . path = parsed . pathname + ( parsed . query ? "?" + parsed . query : "" ) ; } opts . protocol = opts . protocol || "http:" ; opts . port = opts . port || ( opts . protocol === "https:" ? '443' : '80' ) ; opts . path = opts . path || "/" ; if ( ! opts . host ) throw new Error ( "request error: no host" ) ; opts . method = opts . method || "GET" ; opts . headers = Object . keys ( opts . headers || { } ) . reduce ( ( headers , key ) => { if ( opts . headers ! [ key ] != null ) headers [ key ] = opts . headers ! [ key ] ; return headers ; } , { } as Headers ) ; opts . module = require ( opts . protocol . substring ( 0 , opts . protocol . length - 1 ) ) ; if ( opts . user != null ) { var token = opts . user + ":" + ( opts . password || "" ) ; token = Buffer . from ( token , "utf8" ) . toString ( "base64" ) ; opts . headers [ 'Authorization' ] = "Basic " + token ; } if ( opts . proxy ) { if ( opts . host !== os . hostname ( ) ) { if ( typeof opts . proxy === "string" ) { opts . proxy = parseUrl ( opts . proxy ) ; opts . proxy . host = opts . proxy . hostname ; } if ( ! opts . proxy . force && opts . proxy . excludes && opts . proxy . excludes . indexOf ( opts . host . toLowerCase ( ) ) !== - 1 ) { } else { opts . proxy . port = opts . proxy . port || opts . port ; if ( ! opts . proxy . host ) throw new Error ( "proxy configuration error: no host" ) ; if ( ! opts . proxy . port ) throw new Error ( "proxy configuration error: no port" ) ; opts . proxy . protocol = opts . proxy . protocol || "http:" ; opts . isHttps = opts . protocol . substr ( 0 , 5 ) === "https" ; if ( opts . isHttps ) { opts . proxy . module = require ( opts . proxy . protocol . substring ( 0 , opts . proxy . protocol . length - 1 ) ) ; opts . proxy . headers = opts . proxy . headers || { } ; opts . proxy . headers . host = opts . host ; } else { opts . path = opts . protocol + "//" + opts . host + ":" + opts . port + opts . path ; opts . host = opts . proxy . host ; opts . port = opts . proxy . port ; if ( opts . host ) opts . headers [ 'host' ] = opts . host ; } if ( opts . proxy . auth ) { if ( opts . proxy . auth . toLowerCase ( ) === "basic" ) { if ( ! opts . proxy . user ) throw new Error ( "request error: no proxy user" ) ; var proxyToken = opts . proxy . user + ":" + ( opts . proxy . password || "" ) ; proxyToken = Buffer . from ( proxyToken , "utf8" ) . toString ( "base64" ) ; opts . headers [ "Proxy-Authorization" ] = "Basic " + proxyToken ; } else if ( opts . proxy . auth . toLowerCase ( ) === "ntlm" ) { const proxyAuthenticator = opts . proxy . proxyAuthenticator ; if ( ! proxyAuthenticator ) throw new Error ( "Proxy Authenticator module required" ) ; if ( ! proxyAuthenticator . authenticate ) throw new Error ( "NTLM Engine module MUST provide 'authenticate' function" ) ; opts . proxyAuthenticate = proxyAuthenticator . authenticate ; } else if ( opts . proxy . auth . toLowerCase ( ) === "digest" ) { throw new Error ( "Proxy Digest authentication not yet implemented" ) ; } } } } } return opts ; } export class HttpClientRequest extends WritableStream < http . ClientRequest > { _response : http . ClientResponse ; _done : boolean ; _onResponse : ( err : Error | undefined , response ? : http . ClientResponse ) => void ; _options : HttpClientOptions ; constructor ( options ) { var request = options . module . request ( options , ( response ) => { this . _onResponse ( undefined , response ) ; } ) ; super ( request , options ) ; this . _options = options ; this . _done = false ; this . _emitter . on ( 'error' , ( err ) => { if ( ! this . _done ) this . _onResponse ( err ) ; } ) ; this . _autoClosed . push ( ( ) => { if ( ! this . _done ) this . _onResponse ( new Error ( "stream was closed unexpectedly" ) ) ; } ) ; this . _onResponse = this . _trackResponse ; } _trackResponse ( err , resp ? ) { this . _done = true ; if ( err ) this . _error = err ; if ( resp ) this . _response = resp ; } _responseCb ( callback ) { var replied = false ; if ( typeof callback !== 'function' ) throw new TypeError ( "bad callback parameter: " + typeof callback ) ; if ( this . _done ) return callback ( this . _error , this . _response ) ; else this . _onResponse = ( err , resp ) => { this . _done = true ; if ( ! replied ) callback ( err , resp ) ; replied = true ; } ; } response ( _ ) { var response = this . _response || _ . cast ( this . _responseCb ) . call ( this , _ ) ; return new HttpClientResponse ( response , this . _options ) ; } setTimeout ( ms ) { this . _emitter . setTimeout ( ms , ( ) => { this . _emitter . emit ( 'error' , 'timeout' ) ; } ) ; return this ; } proxyConnect ( _ ) { return this ; } } export class HttpProxyClientRequest { _options : HttpClientOptions ; constructor ( options ) { this . _options = _fixHttpClientOptions ( options ) ; } proxyConnect ( _ ) { const options = this . _options ; if ( options . isHttps ) { return _ . cast ( ( callback ) => { const proxyOpt = { host : options . proxy . host , port : options . proxy . port , method : 'CONNECT' , path : options . host + ":" + options . port , headers : options . proxy . headers } ; var replied = false ; options . proxy . module . request ( proxyOpt ) . on ( 'connect' , ( res , socket , head ) => { options . socket = socket ; options . agent = false ; if ( ! replied ) callback ( undefined , new HttpClientRequest ( options ) ) ; replied = true ; } ) . on ( 'error' , ( err ) => { if ( ! replied ) callback ( err ) ; replied = true ; } ) . end ( ) ; return this ; } ) ( _ ) ; } else { if ( options . proxyAuthenticate ) { options . proxyAuthenticate ( _ , options ) ; } return new HttpClientRequest ( options ) ; } } response ( _ ) { throw new Error ( "proxyConnect(_) call missing" ) ; } } export function httpRequest ( options ) { options = _fixHttpClientOptions ( options ) ; if ( options . isHttps || options . proxyAuthenticate ) return new HttpProxyClientRequest ( options ) ; else return new HttpClientRequest ( options ) ; } export interface SocketOptions extends ReadableOptions , WritableOptions { read ? : ReadableOptions ; write ? : WritableOptions ; } export class SocketStream extends ReadableStream < net . Socket & NodeJS . ReadableStream > { _writableStream : WritableStream < net . Socket > ; constructor ( emitter , options ? ) { super ( emitter as any , ( options && options . read ) || options ) ; this . _writableStream = new WritableStream ( emitter , ( options && options . write ) || options ) ; } write ( _ , data ? , enc ? ) { this . _writableStream . write ( _ , data , enc ) ; return this ; } end ( data ? , enc ? ) { this . _writableStream . end ( data , enc ) ; return this ; } get writer ( ) { return this . _writableStream . writer ; } setTimeout ( ms , callback ? ) { this . _emitter . setTimeout ( ms , callback ) ; return this ; } setNoDelay ( noDelay ? ) { this . _emitter . setNoDelay ( noDelay ) ; return this ; } setKeepAlive ( enable ? ) { this . _emitter . setKeepAlive ( enable ) ; return this ; } ref ( ) { this . _emitter . ref ( ) ; return this ; } unref ( ) { this . _emitter . unref ( ) ; return this ; } destroy ( ) { this . _emitter . destroy ( ) ; return this ; } address ( ) { return this . _emitter . address ( ) ; } get localAddress ( ) { return this . _emitter . localAddress ; } get localPort ( ) { return this . _emitter . localPort ; } get remoteAddress ( ) { return this . _emitter . remoteAddress ; } get remotePort ( ) { return this . _emitter . remotePort ; } } export function tcpClient ( port , host ? , options ? ) { host = host || "localhost" ; options = options || { } ; return new SocketClient ( options , port , host ) ; } ; export function socketClient ( path , options ? ) { options = options || { } ; return new SocketClient ( options , path ) ; } ; export class SocketClient { _options ? : SocketOptions ; _connection : net . Socket ; _error : Error ; _done : boolean ; _onConnect : ( err ? : Error ) => void ; constructor ( options ? , ... args ) { this . _options = options ; this . _connection = net . createConnection . apply ( net , args ) ; this . _connection . on ( 'error' , ( err ) => { if ( ! this . _done ) this . _onConnect ( err ) ; this . _onConnect = nop ; } ) ; this . _connection . on ( 'connect' , ( ) => { this . _onConnect ( ) ; this . _onConnect = nop ; } ) ; this . _onConnect = this . _trackConnect ; } _trackConnect ( err ? ) { this . _done = true ; if ( err ) this . _error = err ; } connect ( callback ) { if ( typeof callback !== 'function' ) throw new TypeError ( "bad callback parameter: " + typeof callback ) ; if ( this . _done ) return callback ( this . _error , new SocketStream ( this . _connection , this . _options ) ) ; else { this . _onConnect = ( err ) => { this . _done = true ; callback ( err , new SocketStream ( this . _connection , this . _options ) ) ; callback = nop ; } ; } } } export interface SocketServerOptions { } export type SocketServerListener = ( stream : SocketStream , _ : _ ) => void ; export function createNetServer ( serverOptions , connectionListener , streamOptions ) { return new SocketServer ( serverOptions , connectionListener , streamOptions ) ; } ; export class SocketServer extends Server < net . Server > { constructor ( serverOptions , connectionListener , streamOptions ) { if ( typeof ( serverOptions ) === 'function' ) { streamOptions = connectionListener ; connectionListener = serverOptions ; serverOptions = { } ; } var emitter = net . createServer ( serverOptions , ( connection ) => { _ . withContext ( ( ) => { _ . run ( _ => connectionListener ( new SocketStream ( connection , streamOptions || { } ) , _ ) , ( err ? ) => { if ( err ) throw err ; } ) ; } ) ( ) ; } ) ; super ( emitter ) ; } } exports . using = function ( this , _ , constructor , emitter , options ? , fn ? ) { if ( ! fn && typeof options === 'function' ) fn = options , options = null ; if ( ! fn ) throw new Error ( "using body missing" ) ; const stream = new constructor ( emitter , options ) ; try { return fn . call ( this , _ , stream ) ; } finally { stream . close ( _ ) ; } } ; exports . usingReadable = function ( this , _ , emitter , options ? , fn ? ) { return exports . using . call ( this , _ , ReadableStream , emitter , options , fn ) ; } ; exports . usingWritable = function ( this , _ , emitter , options ? , fn ? ) { return exports . using . call ( this , _ , WritableStream , emitter , options , fn ) ; } ; exports . pump = function ( _ , inStream , outStream ) { var data ; while ( data = inStream . read ( _ ) ) outStream . write ( _ , data ) ; } ;