import * as events from 'events' ; import * as tls from 'tls' ; import { ConnectionOptions } from 'tls' ; import { ErrorCode , INVALID_ENCODING_MSG_PREFIX , NatsError } from './error' ; import { createInbox , extend } from './util' ; import { ProtocolHandler } from './protocolhandler' ; import { DEFAULT_JITTER , DEFAULT_JITTER_TLS , DEFAULT_MAX_PING_OUT , DEFAULT_MAX_RECONNECT_ATTEMPTS , DEFAULT_PING_INTERVAL , DEFAULT_PRE , DEFAULT_RECONNECT_TIME_WAIT , DEFAULT_URI } from './const' ; import { next } from 'nuid' ; export { ErrorCode , NatsError } ; /** Version of the ts-nats library */ export const VERSION = require ( '../package.json' ) . version ; internal export interface Base { subject : string ; callback : MsgCallback ; received : number ; timeout ? : NodeJS . Timer ; max ? : number | undefined ; draining ? : boolean ; } internal export function defaultSub ( ) { return { sid : 0 , subject : '' , received : 0 } as Sub ; } /** ServerInfo received from the server */ export interface ServerInfo { tls_required ? : boolean ; tls_verify ? : boolean ; connect_urls ? : string [ ] ; max_payload : number ; client_id : number ; proto : number ; server_id : string ; version : string ; echo ? : boolean ; nonce ? : string ; nkey ? : string ; } /** Argument provided to `subscribe` and `unsubscribe` event handlers. */ export interface SubEvent { /** subscription subject */ subject : string ; /** subscription id */ sid : number ; /** subscription queue name if a queue subscription */ queue ? : string ; } /** Argument provided to `serversChanged` event handlers. */ export interface ServersChangedEvent { /** Server URLs learned via cluster gossip */ added : string [ ] ; /** Removed server URLs (only added servers are removed). */ deleted : string [ ] ; } internal export interface Sub extends Base { sid : number ; queue ? : string | null ; } internal export interface Req extends Base { token : string ; } /**
 * Message object provided to subscription and requests.
 */ export interface Msg { /** subject used to publish the message */ subject : string ; /** optional reply subject where replies may be sent. */ reply ? : string ; /** optional payload for the message. Type is controlled by [[NatsConnectionOptions.payload]]. */ data ? : any ; /** Internal subscription id */ sid : number ; /** Number of bytes in the payload */ size : number ; } /**
 * Payload specifies the type of [[Msg.data]] that will be sent and received by the client.
 * The payload affects all client subscribers and publishers. If using mixed types, either
 * create multiple connections, or select [[Payload.BINARY]] and perform your own decoding.
 */ export enum Payload { /** Specifies a string payload. This is default [[NatsConnectionOptions.payload]] setting */ STRING = 'string' , /** Specifies payloads are JSON. */ JSON = 'json' , /** Specifies payloads are binary (Buffer) */ BINARY = 'binary' } /** Optional callback interface for 'connect' and 'reconnect' events */ export interface ConnectReconnectCallback { ( connection , serverURL , info ) : void ; } /** Optional callback interface for 'disconnect' and 'reconnecting' events */ export interface ReconnectingDisconnectCallback { ( serverURL ) : void ; } /** Optional callback interface for 'permissionError' events */ export interface PermissionsErrorCallback { ( err ) : void ; } /** Optional callback for 'serversChanged' events */ export interface ServersChangedCallback { ( e ) : void ; } /** Optional callback for 'subscribe' and 'unsubscribe' events */ export interface SubscribeUnsubscribeCallback { ( e ) : void ; } /** Optional callback for 'yield'events */ export interface YieldCallback { ( ) : void ; } /** Optional callback argument for [[Client.flush]] */ export interface FlushCallback { ( err ? ) : void ; } /** [[Client.subscribe]] callbacks. First argument will be an error if an error occurred (such as a timeout) or null. Message argument is the received message (which should be treated as debug information when an error is provided). */ export interface MsgCallback { ( err , msg ) : void ; } /** Signs a challenge from the server with an NKEY, a function matching this interface must be provided when manually signing nonces via the `nonceSigner` connect option. */ export interface NonceSigner { ( nonce ) : Buffer ; } /** Returns an user JWT - can be specified in `userJWT` connect option as a way of dynamically providing a JWT when required. */ export interface JWTProvider { ( ) : string ; } /** Additional options that can be provided to [[Client.subscribe]]. */ export interface SubscriptionOptions { /** Name of the queue group for the subscription */ queue ? : string ; /** Maximum number of messages expected. When this number of messages is received the subscription auto [[Subscription.unsubscribe]]. */ max ? : number ; } hidden export interface RequestOptions { max ? : number ; timeout ? : number ; } export interface NatsConnectionOptions { /** Requires server support 1.2.0+. When set to `true`, the server will not forward messages published by the client
     * to the client's subscriptions. By default value is ignored unless it is set to `true` explicitly */ noEcho ? : boolean ; /** Sets the encoding type used when dealing with [[Payload.STRING]] messages. Only node-supported encoding allowed. */ encoding ? : BufferEncoding ; /** Maximum number of client PINGs that can be outstanding before the connection is considered stale. */ maxPingOut ? : number ; /** Maximum number of consecutive reconnect attempts before the client closes the connection. Specify `-1` to retry forever. */ maxReconnectAttempts ? : number ; /** A name for the client. Useful for identifying a client on the server monitoring and logs. */ name ? : string ; /** When `true` does not randomize the order of servers provided to connect. */ noRandomize ? : boolean ; /** User password. */ pass ? : string ; /** Specifies the payload type of the message. See [[Payload]]. Payload determines [[Msg.data]] types and types published. */ payload ? : Payload ; hidden pedantic ? : boolean ; /** Interval in milliseconds that the client will send PINGs to the server. See [[NatsConnectionOptions.maxPingOut]] */ pingInterval ? : number ; /** Specifies the port on the localhost to make a connection. */ port ? : number ; /** Specifies whether the client should attempt reconnects. */ reconnect ? : boolean ; /** Specifies the interval in milliseconds between reconnect attempts. */ reconnectTimeWait ? : number ; /** Random number between 0 and specified value to add to reconnectTimeWait */ reconnectJitter ? : number ; /** Random number between 0 and specified value to add to reconnectTimeWait when TLS options are specified. */ reconnectJitterTLS ? : number ; /** Specifies a function that determines how long to wait before the next connection attempt if the server has not been connected to. When specified,
     * this disables `reconnectTimeWait`, `reconnectJitter` and `reconnectJitterTLS`. */ reconnectDelayHandler ? : ( ) => number ; /** A list of server URLs where the client should attempt a connection. */ servers ? : Array < string > ; /** If true, or set as a tls.TlsOption object, requires the connection to be secure. Fine grain tls settings, such as certificates can be specified by using a tls.TlsOptions object. */ tls ? : boolean | tls . TlsOptions ; /** Token to use for authentication. */ token ? : string ; /** Server URL where the client should attempt a connection */ url ? : string ; /** NATS username. */ user ? : string ; hidden verbose ? : boolean ; /** If true, the client will perform reconnect logic if it fails to connect when first started. Normal behaviour is for the client to try the supplied list of servers and close if none succeed. */ waitOnFirstConnect ? : boolean ; /** Specifies the max amount of time the client is allowed to process inbound messages before yielding for other IO tasks. When exceeded the client will yield. */ yieldTime ? : number ; /** Nonce signer - a function that signs the nonce challenge sent by the server. */ nonceSigner ? : NonceSigner ; /** Public NKey Identifying the user. */ nkey ? : string ; /** A JWT identifying the user. Can be a static JWT string, or a function that returns a JWT when called. */ userJWT ? : string | JWTProvider ; /** Credentials file path - will automatically setup an `nkey` and `nonceSigner` that references the specified credentials file.*/ userCreds ? : string ; /** nkey file path - will automatically setup an `nkey` and `nonceSigner` that references the specified nkey seed file.*/ nkeyCreds ? : string ; /** number of milliseconds when making a connection to wait for the connection to succeed. Must be greater than zero. */ timeout ? : number ; } internal function defaultReq ( ) { return { token : '' , subject : '' , received : 0 , max : 1 } as Req ; } /**
 * NATS server Client object.
 */ export class Client extends events . EventEmitter { /** Returns an unique and properly formatted inbox subject that can be used for replies */ createInbox = createInbox ; private protocolHandler ! : ProtocolHandler ; internal constructor ( ) { super ( ) ; events . EventEmitter . call ( this ) ; } internal static connect ( opts ? ) { return new Promise ( ( resolve , reject ) => { let options = Client . parseOptions ( opts ) ; let client = new Client ( ) ; ProtocolHandler . connect ( client , options ) . then ( ( ph ) => { client . protocolHandler = ph ; resolve ( client ) ; } ) . catch ( ( ex ) => { reject ( ex ) ; } ) ; } ) ; } internal private static defaultOptions ( ) { return { encoding : 'utf8' , maxPingOut : DEFAULT_MAX_PING_OUT , maxReconnectAttempts : DEFAULT_MAX_RECONNECT_ATTEMPTS , noRandomize : false , pedantic : false , pingInterval : DEFAULT_PING_INTERVAL , reconnect : true , reconnectJitter : DEFAULT_JITTER , reconnectJitterTLS : DEFAULT_JITTER_TLS , reconnectTimeWait : DEFAULT_RECONNECT_TIME_WAIT , tls : undefined , verbose : false , waitOnFirstConnect : false } as ConnectionOptions ; } internal private static parseOptions ( args ? ) { if ( args === undefined || args === null ) { args = { url : DEFAULT_URI } as NatsConnectionOptions ; } if ( typeof args === 'number' ) { args = { url : DEFAULT_PRE + args } as NatsConnectionOptions ; } else if ( typeof args === 'string' ) { args = { url : args . toString ( ) } as NatsConnectionOptions ; } else if ( typeof args === 'object' ) { if ( args . port !== undefined ) { args . url = DEFAULT_PRE + args . port ; } } let options = extend ( Client . defaultOptions ( ) , args ) ; if ( options . user && options . token ) { throw NatsError . errorForCode ( ErrorCode . BAD_AUTHENTICATION ) ; } if ( options . nonceSigner && typeof options . nonceSigner !== 'function' ) { throw NatsError . errorForCode ( ErrorCode . NONCE_SIGNER_NOTFUNC ) ; } let bufEncoding = options . encoding as BufferEncoding ; if ( ! Buffer . isEncoding ( bufEncoding ) ) { throw new NatsError ( INVALID_ENCODING_MSG_PREFIX + options . encoding , ErrorCode . INVALID_ENCODING ) ; } if ( options . reconnectDelayHandler && typeof options . reconnectDelayHandler !== 'function' ) { throw NatsError . errorForCode ( ErrorCode . RECONNECT_DELAY_NOTFUNC ) ; } else { options . reconnectDelayHandler = ( ) => { let extra = options . tls ? options . reconnectJitterTLS : options . reconnectJitter ; if ( extra ) { extra ++ ; extra = Math . floor ( Math . random ( ) * extra ) ; } return options . reconnectTimeWait + extra ; } ; } return options ; } param cb return flush ( cb ? ) { if ( cb === undefined ) { return new Promise ( ( resolve , reject ) => { this . protocolHandler . flush ( ( err ) => { if ( ! err ) { resolve ( ) ; } else { reject ( err ) ; } } ) ; } ) ; } else { this . protocolHandler . flush ( cb ) ; } } param subject param data param reply publish ( subject , data = undefined , reply = '' ) { if ( ! subject ) { throw NatsError . errorForCode ( ErrorCode . BAD_SUBJECT ) ; } this . protocolHandler . publish ( subject , data , reply ) ; } param subject param cb param opts return subscribe ( subject , cb , opts = { } ) { return new Promise < Subscription > ( ( resolve , reject ) => { if ( ! subject ) { reject ( NatsError . errorForCode ( ErrorCode . BAD_SUBJECT ) ) ; } if ( ! cb ) { reject ( new NatsError ( 'subscribe requires a callback' , ErrorCode . API_ERROR ) ) ; } let s = defaultSub ( ) ; extend ( s , opts ) ; s . subject = subject ; s . callback = cb ; resolve ( this . protocolHandler . subscribe ( s ) ) ; } ) ; } see drain ( ) { return this . protocolHandler . drain ( ) ; } param subject param timeout param data return request ( subject , timeout = 1000 , data = undefined ) { return new Promise < Msg > ( ( resolve , reject ) => { if ( this . isClosed ( ) ) { reject ( NatsError . errorForCode ( ErrorCode . CONN_CLOSED ) ) ; } if ( ! subject ) { reject ( NatsError . errorForCode ( ErrorCode . BAD_SUBJECT ) ) ; } let r = defaultReq ( ) ; let opts = { max : 1 } as RequestOptions ; extend ( r , opts ) ; r . token = next ( ) ; let request = this . protocolHandler . request ( r ) ; r . timeout = setTimeout ( ( ) => { request . cancel ( ) ; reject ( NatsError . errorForCode ( ErrorCode . REQ_TIMEOUT ) ) ; } , timeout ) ; r . callback = ( error , msg ? ) => { if ( error ) { reject ( error ) ; } else { resolve ( msg ) ; } } ; try { this . publish ( subject , data , `${ this . protocolHandler . muxSubscriptions . baseInbox }${ r . token }` ) ; } catch ( err ) { reject ( err ) ; request . cancel ( ) ; } } ) ; } ; /**
     * Closes the connection to the NATS server. A closed client cannot be reconnected.
     */ close ( ) { this . protocolHandler . close ( ) ; } return isClosed ( ) { return this . protocolHandler . isClosed ( ) ; } return { Number } numSubscriptions ( ) { return this . protocolHandler . numSubscriptions ( ) ; } } param opts return export function connect ( opts ? ) { return Client . connect ( opts ) ; } /**
 * Type returned when a subscribe call resolved. Provides methods to manage the subscription.
 */ export class Subscription { sid : number ; private protocol : ProtocolHandler ; hidden constructor ( sub , protocol ) { this . sid = sub . sid ; this . protocol = protocol ; } hidden static cancelTimeout ( s ) { if ( s && s . timeout ) { clearTimeout ( s . timeout ) ; delete s . timeout ; } } param max see unsubscribe ( max ? ) { this . protocol . unsubscribe ( this . sid , max ) ; } return see drain ( ) { return this . protocol . drainSubscription ( this . sid ) ; } /**
     * Returns true if the subscription has an associated timeout.
     */ hasTimeout ( ) { let sub = this . protocol . subscriptions . get ( this . sid ) ; return sub !== null && sub . hasOwnProperty ( 'timeout' ) ; } /**
     * Cancels a timeout associated with the subscription
     */ cancelTimeout ( ) { let sub = this . protocol . subscriptions . get ( this . sid ) ; Subscription . cancelTimeout ( sub ) ; } param millis setTimeout ( millis ) { let sub = this . protocol . subscriptions . get ( this . sid ) ; Subscription . cancelTimeout ( sub ) ; if ( sub ) { sub . timeout = setTimeout ( ( ) => { if ( sub && sub . callback ) { sub . callback ( NatsError . errorForCode ( ErrorCode . SUB_TIMEOUT ) , { } as Msg ) ; } this . unsubscribe ( ) ; } , millis ) ; return true ; } return false ; } /**
     * Returns the number of messages received by the subscription.
     */ getReceived ( ) { let sub = this . protocol . subscriptions . get ( this . sid ) ; if ( sub ) { return sub . received ; } return 0 ; } /**
     * Returns the number of messages expected by the subscription.
     * If `0`, the subscription was not found or was auto-cancelled.
     * If `-1`, the subscription didn't specify a count for expected messages.
     */ getMax ( ) { let sub = this . protocol . subscriptions . get ( this . sid ) ; if ( ! sub ) { return 0 ; } if ( sub && sub . max ) { return sub . max ; } return - 1 ; } return isCancelled ( ) { return this . protocol . subscriptions . get ( this . sid ) === null ; } return see isDraining ( ) { let sub = this . protocol . subscriptions . get ( this . sid ) ; if ( sub ) { return sub . draining === true ; } return false ; } }