import * as net from 'net' ; import * as tls from 'tls' ; import { ConnectionOptions , TLSSocket } from 'tls' ; import { Transport , TransportHandlers } from './transport' ; import { UrlObject } from 'url' ; import { Messages , NatsError , ErrorCode } from "./error" ; hidden export class TCPTransport implements Transport { connectedOnce : boolean = false ; stream : net . Socket | TLSSocket | null = null ; handlers : TransportHandlers ; closed : boolean = false ; dialTime : number = 0 ; constructor ( handlers ) { this . handlers = handlers ; } connect ( url , timeout ? ) { let dialStart = 0 ; if ( timeout ) { dialStart = Date . now ( ) ; } return new Promise ( ( resolve , reject ) => { if ( this . stream ) { this . destroy ( ) ; } let connected = false ; let to ; if ( timeout ) { to = setTimeout ( ( ) => { if ( ! this . connectedOnce ) { reject ( NatsError . errorForCode ( ErrorCode . CONN_TIMEOUT ) ) ; this . destroy ( ) ; } else { this . handlers . error ( NatsError . errorForCode ( ErrorCode . CONN_TIMEOUT ) ) ; } } , timeout ) ; } this . stream = net . createConnection ( parseInt ( url . port , 10 ) , url . hostname , ( ) => { if ( to ) { this . dialTime = Date . now ( ) - dialStart ; clearTimeout ( to ) ; to = undefined ; } resolve ( ) ; connected = true ; this . connectedOnce = true ; this . handlers . connect ( ) ; } ) ; this . stream . setNoDelay ( true ) ; this . stream . on ( 'error' , ( error ) => { if ( ! this . connectedOnce ) { reject ( error ) ; this . destroy ( ) ; } else { this . handlers . error ( error ) ; } } ) ; this . stream . on ( 'close' , ( ) => { if ( this . connectedOnce ) { this . handlers . close ( ) ; } } ) ; this . stream . on ( 'data' , ( data ) => { this . handlers . data ( data ) ; } ) ; } ) ; } isClosed ( ) { return this . closed ; } isConnected ( ) { return this . stream != null && ! this . stream . connecting ; } isEncrypted ( ) { return this . stream instanceof TLSSocket && this . stream . encrypted ; } isAuthorized ( ) { return this . stream instanceof TLSSocket && this . stream . authorized ; } upgrade ( tlsOptions , done ) { if ( this . stream ) { let opts ; if ( 'object' === typeof tlsOptions ) { opts = tlsOptions as ConnectionOptions ; } else { opts = { } as ConnectionOptions ; } opts . socket = this . stream ; this . stream . removeAllListeners ( ) ; try { this . stream = tls . connect ( opts , ( ) => { done ( ) ; } ) ; this . stream . on ( 'error' , ( error ) => { this . handlers . error ( error ) ; } ) ; this . stream . on ( 'close' , ( ) => { this . handlers . close ( ) ; } ) ; this . stream . on ( 'data' , ( data ) => { this . handlers . data ( data ) ; } ) ; } catch ( err ) { this . handlers . error ( new NatsError ( Messages . getMessage ( ErrorCode . SSL_ERR ) , ErrorCode . SSL_ERR , err ) ) ; } } } write ( data ) { if ( this . stream ) { this . stream . write ( data ) ; } } destroy ( ) { if ( ! this . stream ) { return ; } if ( this . closed ) { this . stream . removeAllListeners ( ) ; } this . stream . destroy ( ) ; this . stream = null ; } close ( ) { this . closed = true ; this . destroy ( ) ; } pause ( ) { if ( this . stream ) { this . stream . pause ( ) ; } } resume ( ) { if ( this . stream && this . stream . isPaused ( ) ) { this . stream . resume ( ) ; } } dialDuration ( ) { return this . dialTime ; } }