import { Client , defaultSub , FlushCallback , NatsConnectionOptions , Payload , Req , ServerInfo , Sub , SubEvent , Subscription , VERSION } from './nats' ; import { MuxSubscriptions } from './muxsubscriptions' ; import { Callback , Transport , TransportHandlers } from './transport' ; import { CONN_ERR_PREFIX , ErrorCode , NatsError } from './error' ; import { EventEmitter } from 'events' ; import { CR_LF , DEFAULT_PING_INTERVAL , EMPTY } from './const' ; import { Server , Servers } from './servers' ; import { TCPTransport } from './tcptransport' ; import { Subscriptions } from './subscriptions' ; import { DataBuffer } from './databuffer' ; import { MsgBuffer } from './messagebuffer' ; import { delay , settle } from './util' ; import * as fs from 'fs' ; import * as url from 'url' ; import * as nkeys from 'ts-nkeys' ; import Timeout = NodeJS . Timeout ; const MSG = /^MSG\s+([^\s\r\n]+)\s+([^\s\r\n]+)\s+(([^\s\r\n]+)[^\S\r\n]+)?(\d+)\r\n/i , OK = /^\+OK\s*\r\n/i , ERR = /^-ERR\s+('.+')?\r\n/i , PING = /^PING\r\n/i , PONG = /^PONG\r\n/i , INFO = /^INFO\s+([^\r\n]+)\r\n/i , SUBRE = /^SUB\s+([^\r\n]+)\r\n/i , CREDS = /\s*(?:(?:[-]{3,}[^\n]*[-]{3,}\n)(.+)(?:\n\s*[-]{3,}[^\n]*[-]{3,}\n))/i , SUB = 'SUB' , UNSUB = 'UNSUB' , CONNECT = 'CONNECT' , FLUSH_THRESHOLD = 65536 ; const CRLF_BUF = Buffer . from ( '\r\n' ) ; enum ParserState { CLOSED = - 1 , AWAITING_CONTROL = 0 , AWAITING_MSG_PAYLOAD = 1 } enum TlsRequirement { OFF = - 1 , ANY = 0 , ON = 1 } hidden export class ProtocolHandler extends EventEmitter { options : NatsConnectionOptions ; subscriptions : Subscriptions ; muxSubscriptions = new MuxSubscriptions ( ) ; private client : Client ; private closed : boolean = false ; private connected : boolean = false ; private currentServer ! : Server ; private encoding : BufferEncoding ; private inbound = new DataBuffer ( ) ; private info : ServerInfo = { } as ServerInfo ; private infoReceived : boolean = false ; private msgBuffer ? : MsgBuffer | null ; private outbound = new DataBuffer ( ) ; private payload : Payload ; private pingTimer ? : Timeout ; private pongs : any [ ] = [ ] ; private pout : number = 0 ; private reconnecting : boolean = false ; private servers : Servers ; private state : ParserState = ParserState . AWAITING_CONTROL ; private transport : Transport ; private url ! : url . UrlObject ; private wasConnected : boolean = false ; private draining : boolean = false ; private noMorePublishing : boolean = false ; private connectionTimer ? : NodeJS . Timeout | undefined ; constructor ( client , options ) { super ( ) ; EventEmitter . call ( this ) ; this . client = client ; this . options = options ; this . encoding = options . encoding || 'utf8' ; this . payload = options . payload || Payload . STRING ; this . subscriptions = new Subscriptions ( ) ; this . subscriptions . on ( 'subscribe' , ( sub ) => { this . client . emit ( 'subscribe' , sub ) ; } ) ; this . subscriptions . on ( 'unsubscribe' , ( unsub ) => { this . client . emit ( 'unsubscribe' , unsub ) ; } ) ; this . servers = new Servers ( ! this . options . noRandomize , this . options . servers || [ ] , this . options . url ) ; this . transport = new TCPTransport ( this . getTransportHandlers ( ) ) ; } static connect ( client , opts ) { let lastError ; return new Promise < ProtocolHandler > ( async ( resolve , reject ) => { let ph = new ProtocolHandler ( client , opts ) ; while ( true ) { let wait = ph . options . reconnectDelayHandler ( ) ; let maxWait = wait ; for ( let i = 0 ; i < ph . servers . length ( ) ; i ++ ) { const srv = ph . selectServer ( ) ; if ( srv ) { const now = Date . now ( ) ; if ( srv . lastConnect === 0 || srv . lastConnect + wait <= now ) { try { await ph . connect ( ) ; resolve ( ph ) ; ph . startHandshakeTimeout ( ) ; return ; } catch ( err ) { lastError = err ; if ( ! ph . options . waitOnFirstConnect ) { ph . servers . removeCurrentServer ( ) ; } } } else { maxWait = Math . min ( maxWait , srv . lastConnect + wait - now ) ; } } } if ( ph . servers . length ( ) === 0 ) { const err = lastError || NatsError . errorForCode ( ErrorCode . UNABLE_TO_CONNECT ) ; reject ( err ) ; return ; } await delay ( maxWait ) ; } } ) ; } startHandshakeTimeout ( ) { if ( this . options . timeout ) { this . connectionTimer = setTimeout ( ( ) => { this . processErr ( 'conn_timeout' ) ; } , this . options . timeout - this . transport . dialDuration ( ) ) ; } } flush ( cb ) { if ( this . closed ) { if ( typeof cb === 'function' ) { cb ( NatsError . errorForCode ( ErrorCode . CONN_CLOSED ) ) ; return ; } else { throw NatsError . errorForCode ( ErrorCode . CONN_CLOSED ) ; } } this . pongs . push ( cb ) ; this . sendCommand ( ProtocolHandler . buildProtocolMessage ( 'PING' ) ) ; } closeAndEmit ( ) { this . close ( ) ; this . client . emit ( 'close' ) ; } close ( ) { this . cancelHeartbeat ( ) ; this . closed = true ; this . removeAllListeners ( ) ; this . closeStream ( ) ; this . subscriptions . close ( ) ; this . muxSubscriptions . close ( ) ; this . state = ParserState . CLOSED ; this . pongs = [ ] ; this . outbound . reset ( ) ; } ; drain ( ) { if ( this . closed ) { return Promise . reject ( NatsError . errorForCode ( ErrorCode . CONN_CLOSED ) ) ; } if ( this . draining ) { return Promise . reject ( NatsError . errorForCode ( ErrorCode . CONN_DRAINING ) ) ; } this . draining = true ; let subs = this . subscriptions . all ( ) ; let promises = [ ] ; subs . forEach ( ( sub ) => { let p = this . drainSubscription ( sub . sid ) ; promises . push ( p ) ; } ) ; return new Promise ( ( resolve ) => { settle ( promises ) . then ( ( a ) => { this . noMorePublishing = true ; process . nextTick ( ( ) => { this . flush ( ( ) => { this . close ( ) ; resolve ( a ) ; } ) ; } ) ; } ) . catch ( ( ) => { } ) ; } ) ; } publish ( subject , data , reply = '' ) { if ( this . closed ) { throw NatsError . errorForCode ( ErrorCode . CONN_CLOSED ) ; } if ( this . noMorePublishing ) { throw NatsError . errorForCode ( ErrorCode . CONN_DRAINING ) ; } data = this . toBuffer ( data ) ; let len = data . length ; let proto ; if ( reply ) { proto = `PUB ${ subject } ${ reply } ${ len }` ; } else { proto = `PUB ${ subject } ${ len }` ; } this . sendCommand ( ProtocolHandler . buildProtocolMessage ( proto , data ) ) ; } subscribe ( s ) { if ( this . isClosed ( ) ) { throw NatsError . errorForCode ( ErrorCode . CONN_CLOSED ) ; } if ( this . draining ) { throw NatsError . errorForCode ( ErrorCode . CONN_DRAINING ) ; } let sub = this . subscriptions . add ( s ) as Sub ; if ( sub . queue ) { this . sendCommand ( ProtocolHandler . buildProtocolMessage ( `SUB ${ sub . subject } ${ sub . queue } ${ sub . sid }` ) ) ; } else { this . sendCommand ( ProtocolHandler . buildProtocolMessage ( `SUB ${ sub . subject } ${ sub . sid }` ) ) ; } if ( s . max ) { this . unsubscribe ( sub . sid , s . max ) ; } return new Subscription ( sub , this ) ; } drainSubscription ( sid ) { if ( this . closed ) { return Promise . reject ( NatsError . errorForCode ( ErrorCode . CONN_CLOSED ) ) ; } if ( ! sid ) { return Promise . reject ( NatsError . errorForCode ( ErrorCode . SUB_CLOSED ) ) ; } let s = this . subscriptions . get ( sid ) ; if ( ! s ) { return Promise . reject ( NatsError . errorForCode ( ErrorCode . SUB_CLOSED ) ) ; } if ( s . draining ) { return Promise . reject ( NatsError . errorForCode ( ErrorCode . SUB_DRAINING ) ) ; } let sub = s ; return new Promise ( ( resolve ) => { sub . draining = true ; this . sendCommand ( ProtocolHandler . buildProtocolMessage ( `UNSUB ${ sid }` ) ) ; this . flush ( ( ) => { this . subscriptions . cancel ( sub ) ; resolve ( { sid : sub . sid , subject : sub . subject , queue : sub . queue } as SubEvent ) ; } ) ; } ) ; } unsubscribe ( sid , max ? ) { if ( ! sid || this . closed ) { return ; } let s = this . subscriptions . get ( sid ) ; if ( s ) { if ( max ) { this . sendCommand ( ProtocolHandler . buildProtocolMessage ( `${ UNSUB } ${ sid } ${ max }` ) ) ; } else { this . sendCommand ( ProtocolHandler . buildProtocolMessage ( `${ UNSUB } ${ sid }` ) ) ; } s . max = max ; if ( s . max === undefined || s . received >= s . max ) { this . subscriptions . cancel ( s ) ; } } } request ( r ) { if ( this . closed ) { throw NatsError . errorForCode ( ErrorCode . CONN_CLOSED ) ; } if ( this . draining ) { throw NatsError . errorForCode ( ErrorCode . CONN_DRAINING ) ; } this . initMux ( ) ; this . muxSubscriptions . add ( r ) ; return new Request ( r , this ) ; } numSubscriptions ( ) { return this . subscriptions . length ; } isClosed ( ) { return this . closed ; } cancelRequest ( token , max ? ) { if ( ! token || this . isClosed ( ) ) { return ; } let r = this . muxSubscriptions . get ( token ) ; if ( r ) { r . max = max ; if ( r . max === undefined || r . received >= r . max ) { this . muxSubscriptions . cancel ( r ) ; } } } private connect ( ) { this . currentServer . lastConnect = Date . now ( ) ; this . prepareConnection ( ) ; if ( this . reconnecting ) { this . currentServer . reconnects += 1 ; this . client . emit ( 'reconnecting' , this . url . href ) ; } return this . transport . connect ( this . url , this . options . timeout ) ; } private flushPending ( ) { if ( ! this . infoReceived ) { return ; } if ( this . outbound . size ( ) ) { let d = this . outbound . drain ( ) ; this . transport . write ( d ) ; } } private static buildProtocolMessage ( protocol , payload ? ) { let protoLen = Buffer . byteLength ( protocol ) ; let cmd = protoLen + 2 ; let len = cmd ; if ( payload ) { len += payload . byteLength + 2 ; } let buf = Buffer . allocUnsafe ( len ) ; buf . write ( protocol ) ; CRLF_BUF . copy ( buf , protoLen ) ; if ( payload ) { payload . copy ( buf , cmd ) ; CRLF_BUF . copy ( buf , buf . byteLength - 2 ) ; } return buf ; } private sendCommand ( cmd ) { if ( this . closed ) { return ; } let buf ; if ( typeof cmd === 'string' ) { let len = Buffer . byteLength ( cmd ) ; buf = Buffer . allocUnsafe ( len ) ; buf . write ( cmd , 0 , len , 'utf8' ) ; } else { buf = cmd as Buffer ; } if ( buf . byteLength === 0 ) { return ; } this . outbound . fill ( buf ) ; if ( this . outbound . length ( ) === 1 ) { setImmediate ( ( ) => { this . flushPending ( ) ; } ) ; } else if ( this . outbound . size ( ) > FLUSH_THRESHOLD ) { this . flushPending ( ) ; } } private getTransportHandlers ( ) { let handlers = { } as TransportHandlers ; handlers . connect = ( ) => { this . connected = true ; } ; handlers . close = ( ) => { this . cancelHeartbeat ( ) ; let wasConnected = this . connected ; this . closeStream ( ) ; if ( wasConnected ) { this . client . emit ( 'disconnect' , this . currentServer . url . href ) ; } if ( this . closed ) { this . closeAndEmit ( ) ; } else { this . scheduleReconnect ( ) ; } } ; handlers . error = ( exception ) => { if ( this . wasConnected && this . currentServer . didConnect ) { return ; } if ( ! this . wasConnected && ! this . currentServer . didConnect ) { if ( this . options . waitOnFirstConnect ) { this . currentServer . didConnect = true ; } else { this . servers . removeCurrentServer ( ) ; } } if ( ! this . wasConnected && this . servers . length ( ) === 0 ) { this . client . emit ( 'error' , new NatsError ( CONN_ERR_PREFIX + exception , ErrorCode . CONN_ERR , exception ) ) ; } this . closeStream ( ) ; } ; handlers . data = ( data ) => { this . inbound . fill ( data ) ; this . processInbound ( ) ; } ; return handlers ; } private prepareConnection ( ) { let buffers = this . outbound . reset ( ) ; let pongs = [ ] as Callback [ ] ; if ( buffers . length ) { let pongIndex = 0 ; buffers . forEach ( ( buf ) => { let cmd = buf . toString ( 'binary' ) ; if ( PING . test ( cmd ) && this . pongs !== null && pongIndex < this . pongs . length ) { let f = this . pongs [ pongIndex ++ ] ; if ( f ) { this . outbound . fill ( buf ) ; pongs . push ( f ) ; } } else if ( cmd . length > 3 && cmd [ 0 ] === 'P' && cmd [ 1 ] === 'U' && cmd [ 2 ] === 'B' ) { this . outbound . fill ( buf ) ; } } ) ; } this . pongs = pongs ; this . state = ParserState . AWAITING_CONTROL ; this . info = { } as ServerInfo ; this . infoReceived = false ; } ; getInfo ( ) { if ( this . infoReceived ) { return this . info ; } return null ; } api private stripPendingSubs ( ) { if ( this . outbound . size ( ) === 0 ) { return ; } let buffers = this . outbound . reset ( ) ; for ( let i = 0 ; i < buffers . length ; i ++ ) { let s = buffers [ i ] . toString ( 'binary' ) ; if ( ! SUBRE . test ( s ) ) { this . sendCommand ( buffers [ i ] ) ; } } } api private sendSubscriptions ( ) { if ( this . subscriptions . length === 0 || ! this . transport . isConnected ( ) ) { return ; } let cmds = [ ] ; this . subscriptions . all ( ) . forEach ( ( s ) => { if ( s . queue ) { cmds . push ( `${ SUB } ${ s . subject } ${ s . queue } ${ s . sid }${ CR_LF }` ) ; } else { cmds . push ( `${ SUB } ${ s . subject } ${ s . sid }${ CR_LF }` ) ; } if ( s . max ) { const max = s . max - s . received ; if ( max > 0 ) { cmds . push ( `${ UNSUB } ${ s . sid } ${ max }${ CR_LF }` ) ; } else { cmds . push ( `${ UNSUB } ${ s . sid }${ CR_LF }` ) ; } } } ) ; if ( cmds . length ) { this . transport . write ( cmds . join ( '' ) ) ; } } api private processInbound ( ) { let m ; let start ; if ( ! this . transport ) { return ; } this . transport . resume ( ) ; if ( this . options . yieldTime !== undefined ) { start = Date . now ( ) ; } while ( ! this . closed && this . inbound . size ( ) ) { switch ( this . state ) { case ParserState . AWAITING_CONTROL : let len = this . inbound . protoLen ( ) ; if ( len === - 1 ) { return ; } let bb = this . inbound . drain ( len ) ; if ( bb . byteLength === 0 ) { return ; } let buf = bb . toString ( ) ; if ( ( m = MSG . exec ( buf ) ) !== null ) { this . msgBuffer = new MsgBuffer ( m , this . payload , this . encoding ) ; this . state = ParserState . AWAITING_MSG_PAYLOAD ; } else if ( ( m = OK . exec ( buf ) ) !== null ) { } else if ( ( m = ERR . exec ( buf ) ) !== null ) { if ( this . processErr ( m [ 1 ] ) ) { return ; } } else if ( ( m = PONG . exec ( buf ) ) !== null ) { this . pout = 0 ; let cb = this . pongs && this . pongs . shift ( ) ; if ( cb ) { try { cb ( ) ; } catch ( err ) { console . error ( 'error while processing pong' , err ) ; } } } else if ( ( m = PING . exec ( buf ) ) !== null ) { this . sendCommand ( ProtocolHandler . buildProtocolMessage ( 'PONG' ) ) ; } else if ( ( m = INFO . exec ( buf ) ) !== null ) { this . info = JSON . parse ( m [ 1 ] ) ; if ( this . checkTLSMismatch ( ) ) { return ; } if ( this . checkNoEchoMismatch ( ) ) { return ; } if ( this . checkNonceSigner ( ) ) { return ; } let change = this . servers . processServerUpdate ( this . info ) ; if ( change . deleted . length > 0 || change . added . length > 0 ) { this . client . emit ( 'serversChanged' , change ) ; } if ( ! this . infoReceived ) { let encrypted = this . transport . isEncrypted ( ) ; if ( this . info . tls_required === true && ! encrypted ) { this . transport . upgrade ( this . options . tls , ( ) => { this . flushPending ( ) ; } ) ; } let cs = JSON . stringify ( new Connect ( this . currentServer , this . options , this . info ) ) ; this . transport . write ( `${ CONNECT } ${ cs }${ CR_LF }` ) ; this . pongs . unshift ( ( ) => { if ( this . connectionTimer ) { clearTimeout ( this . connectionTimer ) ; this . connectionTimer = undefined ; } this . sendSubscriptions ( ) ; this . stripPendingSubs ( ) ; this . scheduleHeartbeat ( ) ; this . connectCB ( ) ; } ) ; this . transport . write ( ProtocolHandler . buildProtocolMessage ( 'PING' ) ) ; this . flushPending ( ) ; this . infoReceived = true ; } } else { return ; } break ; case ParserState . AWAITING_MSG_PAYLOAD : if ( ! this . msgBuffer ) { break ; } if ( this . inbound . size ( ) < this . msgBuffer . length ) { return ; } let dd = this . inbound . drain ( this . msgBuffer . length ) ; this . msgBuffer . fill ( dd ) ; this . processMsg ( ) ; this . state = ParserState . AWAITING_CONTROL ; this . msgBuffer = null ; if ( start !== undefined && this . options && this . options . yieldTime ) { if ( ( Date . now ( ) - start ) > this . options . yieldTime ) { this . transport . pause ( ) ; this . client . emit ( 'yield' ) ; setImmediate ( ( ) => { this . processInbound ( ) ; } ) ; return ; } } break ; } } } private clientTLSRequirement ( ) { if ( this . options . tls === undefined ) { return TlsRequirement . ANY ; } if ( this . options . tls === false ) { return TlsRequirement . OFF ; } return TlsRequirement . ON ; } api private checkTLSMismatch ( ) { switch ( this . clientTLSRequirement ( ) ) { case TlsRequirement . OFF : if ( this . info . tls_required ) { this . client . emit ( 'error' , NatsError . errorForCode ( ErrorCode . SECURE_CONN_REQ ) ) ; this . closeStream ( ) ; return true ; } break ; case TlsRequirement . ON : if ( ! this . info . tls_required ) { this . client . emit ( 'error' , NatsError . errorForCode ( ErrorCode . NON_SECURE_CONN_REQ ) ) ; this . closeStream ( ) ; return true ; } break ; case TlsRequirement . ANY : break ; } let cert = false ; if ( this . options . tls && typeof this . options . tls === 'object' ) { cert = this . options . tls . cert != null ; } if ( this . info . tls_verify && ! cert ) { this . client . emit ( 'error' , NatsError . errorForCode ( ErrorCode . CLIENT_CERT_REQ ) ) ; this . closeStream ( ) ; return true ; } return false ; } api private checkNoEchoMismatch ( ) { if ( ( this . info . proto === undefined || this . info . proto < 1 ) && this . options . noEcho ) { this . client . emit ( 'error' , NatsError . errorForCode ( ErrorCode . NO_ECHO_NOT_SUPPORTED ) ) ; this . closeStream ( ) ; return true ; } return false ; } private checkNonceSigner ( ) { if ( this . info . nonce === undefined ) { return false ; } if ( this . options . nkeyCreds ) { try { let seed = nkeys . fromSeed ( this . getNkeyCreds ( ) ) ; this . options . nkey = seed . getPublicKey ( ) . toString ( ) ; this . options . nonceSigner = ( nonce ) : any => { return this . nkeyNonceSigner ( Buffer . from ( nonce ) ) ; } ; } catch ( err ) { this . client . emit ( 'error' , err ) ; this . closeStream ( ) ; return true ; } } if ( this . options . userCreds ) { try { this . getUserCreds ( true ) ; this . options . nonceSigner = ( nonce ) : any => { return this . credsNonceSigner ( Buffer . from ( nonce ) ) ; } ; this . options . userJWT = ( ) => { return this . loadJwt ( ) ; } ; } catch ( err ) { this . client . emit ( 'error' , err ) ; this . closeStream ( ) ; return true ; } } if ( this . options . nonceSigner === undefined ) { this . client . emit ( 'error' , NatsError . errorForCode ( ErrorCode . SIGNATURE_REQUIRED ) ) ; this . closeStream ( ) ; return true ; } if ( this . options . nkey === undefined && this . options . userJWT === undefined ) { this . client . emit ( 'error' , NatsError . errorForCode ( ErrorCode . NKEY_OR_JWT_REQ ) ) ; this . closeStream ( ) ; return true ; } return false ; } private getNkeyCreds ( ) { if ( this . options . nkeyCreds ) { return fs . readFileSync ( this . options . nkeyCreds ) ; } throw NatsError . errorForCode ( ErrorCode . BAD_NKEY_SEED ) ; } private getUserCreds ( jwt = false ) { if ( this . options . userCreds ) { let buf = fs . readFileSync ( this . options . userCreds ) ; if ( buf ) { let re = jwt ? CREDS : new RegExp ( CREDS , 'g' ) ; let contents = buf . toString ( ) ; let m = re . exec ( contents ) ; if ( m === null ) { throw NatsError . errorForCode ( ErrorCode . BAD_CREDS ) ; } if ( jwt ) { return m ; } m = re . exec ( contents ) ; if ( m === null ) { throw NatsError . errorForCode ( ErrorCode . BAD_CREDS ) ; } return m ; } } throw NatsError . errorForCode ( ErrorCode . BAD_CREDS ) ; } private nkeyNonceSigner ( nonce ) { try { let m = this . getNkeyCreds ( ) ; let sk = nkeys . fromSeed ( m ) ; return sk . sign ( nonce ) ; } catch ( ex ) { this . closeStream ( ) ; this . client . emit ( 'error' , ex ) ; } } private credsNonceSigner ( nonce ) { try { let m = this . getUserCreds ( ) ; let sk = nkeys . fromSeed ( Buffer . from ( m [ 1 ] ) ) ; return sk . sign ( nonce ) ; } catch ( ex ) { this . closeStream ( ) ; this . client . emit ( 'error' , ex ) ; } } private loadJwt ( ) { try { let m = this . getUserCreds ( true ) ; return m [ 1 ] ; } catch ( ex ) { this . closeStream ( ) ; this . client . emit ( 'error' , ex ) ; } } api private processMsg ( ) { if ( this . subscriptions . length === 0 || ! this . msgBuffer ) { return ; } let sub = this . subscriptions . get ( this . msgBuffer . msg . sid ) ; if ( ! sub ) { return ; } sub . received += 1 ; if ( sub . timeout && ( sub . max === undefined || sub . received >= sub . max ) ) { Subscription . cancelTimeout ( sub ) ; } if ( sub . max !== undefined && sub . received >= sub . max ) { this . unsubscribe ( sub . sid ) ; } if ( sub . callback ) { try { if ( this . msgBuffer . error ) { sub . callback ( this . msgBuffer . error , this . msgBuffer . msg ) ; } else { sub . callback ( null , this . msgBuffer . msg ) ; } } catch ( error ) { this . client . emit ( 'error' , error ) ; } } } ; static toError ( s ) { let t = s ? s . toLowerCase ( ) : '' ; if ( t . indexOf ( 'permissions violation' ) !== - 1 ) { return new NatsError ( s , ErrorCode . PERMISSIONS_VIOLATION ) ; } else if ( t . indexOf ( 'authorization violation' ) !== - 1 ) { return new NatsError ( s , ErrorCode . AUTHORIZATION_VIOLATION ) ; } else if ( t . indexOf ( 'conn_timeout' ) !== - 1 ) { return NatsError . errorForCode ( ErrorCode . CONN_TIMEOUT ) ; } else { return new NatsError ( s , ErrorCode . NATS_PROTOCOL_ERR ) ; } } api private processErr ( s ) { let err = ProtocolHandler . toError ( s ) ; switch ( err . code ) { case ErrorCode . AUTHORIZATION_VIOLATION : this . client . emit ( 'error' , err ) ; this . closeStream ( ) ; return true ; case ErrorCode . PERMISSIONS_VIOLATION : this . client . emit ( 'permissionError' , err ) ; return false ; case ErrorCode . CONN_TIMEOUT : this . client . emit ( 'error' , NatsError . errorForCode ( ErrorCode . CONN_TIMEOUT ) ) ; this . closeStream ( ) ; return true ; default : this . client . emit ( 'error' , err ) ; this . closeStream ( ) ; return true ; } } ; api private closeStream ( ) { this . transport . destroy ( ) ; if ( this . connected || this . closed ) { this . pongs = [ ] ; this . pout = 0 ; this . connected = false ; this . inbound . reset ( ) ; } } ; api private scheduleReconnect ( ) { if ( this . closed ) { return ; } if ( this . servers . length ( ) === 0 || this . options . reconnect === false ) { this . closeAndEmit ( ) ; return ; } if ( this . wasConnected ) { this . reconnecting = true ; } let wait = this . options . reconnectDelayHandler ( ) ; let maxWait = wait ; const now = Date . now ( ) ; for ( let i = 0 ; i < this . servers . length ( ) ; i ++ ) { const srv = this . selectServer ( ) ; if ( srv ) { const mra = this . options . maxReconnectAttempts || 0 ; if ( mra !== - 1 && srv . reconnects >= mra ) { this . servers . removeCurrentServer ( ) ; continue ; } if ( srv . lastConnect === 0 || srv . lastConnect + wait <= now ) { this . reconnect ( ) ; return ; } maxWait = Math . min ( maxWait , srv . lastConnect + wait - now ) ; } } if ( this . servers . length ( ) === 0 ) { this . closeAndEmit ( ) ; return ; } setTimeout ( ( ) => { this . scheduleReconnect ( ) ; } , maxWait ) ; } private scheduleHeartbeat ( ) { this . pingTimer = setTimeout ( ( ) => { this . client . emit ( 'pingtimer' ) ; if ( this . closed ) { return ; } if ( this . transport . isConnected ( ) ) { this . client . emit ( 'pingcount' , this . pout ) ; this . pout ++ ; if ( this . pout > this . options . maxPingOut ) { this . processErr ( ErrorCode . STALE_CONNECTION_ERR ) ; return ; } else { this . sendCommand ( ProtocolHandler . buildProtocolMessage ( 'PING' ) ) ; if ( this . pongs ) { this . pongs . push ( undefined ) ; } } } this . scheduleHeartbeat ( ) ; } , this . options . pingInterval || DEFAULT_PING_INTERVAL , this ) ; } private cancelHeartbeat ( ) { if ( this . pingTimer ) { clearTimeout ( this . pingTimer ) ; delete this . pingTimer ; } } api private reconnect ( ) { if ( this . closed ) { return ; } const ph = this ; this . connect ( ) . then ( ( ) => { ph . startHandshakeTimeout ( ) ; } ) . catch ( ( ) => { } ) ; } api private selectServer ( ) { let server = this . servers . selectServer ( ) ; if ( server === undefined ) { return undefined ; } this . currentServer = server ; this . url = server . url ; return this . currentServer ; } private toBuffer ( data = undefined ) { if ( this . options . payload === Payload . JSON ) { data = data === undefined ? null : data ; try { data = JSON . stringify ( data ) ; } catch ( e ) { throw NatsError . errorForCode ( ErrorCode . BAD_JSON ) ; } } else { data = data || EMPTY ; } if ( ! Buffer . isBuffer ( data ) ) { data = Buffer . from ( data ) ; } return data ; } private initMux ( ) { let mux = this . subscriptions . getMux ( ) ; if ( ! mux ) { let inbox = this . muxSubscriptions . init ( ) ; let sub = defaultSub ( ) ; sub . subject = `${ inbox }*` ; sub . callback = this . muxSubscriptions . dispatcher ( ) ; this . subscriptions . setMux ( sub ) ; this . subscribe ( sub ) ; } } api private connectCB ( ) { let event = this . reconnecting ? 'reconnect' : 'connect' ; this . reconnecting = false ; this . wasConnected = true ; if ( this . currentServer ) { this . currentServer . didConnect = true ; this . currentServer . reconnects = 0 ; } let info = { } as ServerInfo ; try { info = JSON . parse ( JSON . stringify ( this . info ) ) ; } catch ( err ) { } this . client . emit ( event , this . client , this . currentServer . url . href , info ) ; this . flushPending ( ) ; } } export class Request { token : string ; private protocol : ProtocolHandler ; constructor ( req , protocol ) { this . token = req . token ; this . protocol = protocol ; } cancel ( ) { this . protocol . cancelRequest ( this . token , 0 ) ; } } export class Connect { lang : string = 'typescript' ; version : string = VERSION ; verbose : boolean = false ; pedantic : boolean = false ; protocol : number = 1 ; user ? : string ; pass ? : string ; auth_token ? : string ; name ? : string ; echo ? : boolean ; sig ? : string ; jwt ? : string ; nkey ? : string ; constructor ( server , opts , info ) { opts = opts || { } as NatsConnectionOptions ; if ( opts . user ) { this . user = opts . user ; this . pass = opts . pass ; } if ( opts . token ) { this . auth_token = opts . token ; } let auth = server . getCredentials ( ) ; if ( auth ) { if ( auth . length !== 1 ) { if ( this . user === undefined ) { this . user = auth [ 0 ] ; } if ( this . pass === undefined ) { this . pass = auth [ 1 ] ; } } else if ( this . auth_token === undefined ) { this . auth_token = auth [ 0 ] ; } } if ( opts . name ) { this . name = opts . name ; } if ( opts . verbose !== undefined ) { this . verbose = opts . verbose ; } if ( opts . pedantic !== undefined ) { this . pedantic = opts . pedantic ; } if ( opts . noEcho ) { this . echo = false ; } if ( info . nonce && opts . nonceSigner ) { let sig = opts . nonceSigner ( info . nonce ) ; this . sig = sig . toString ( 'base64' ) ; } if ( opts . userJWT ) { if ( typeof opts . userJWT === 'function' ) { this . jwt = opts . userJWT ( ) ; } else { this . jwt = opts . userJWT ; } } if ( opts . nkey ) { this . nkey = opts . nkey ; } } }