import * as url from 'url' ; import { DEFAULT_PORT , DEFAULT_URI } from './const' ; import { shuffle } from './util' ; import { ServerInfo , ServersChangedEvent } from './nats' ; hidden export class Server { url : url . Url ; didConnect : boolean ; reconnects : number ; lastConnect : number ; implicit : boolean ; constructor ( u , implicit = false ) { if ( ! /^.*:\/\/.*/ . test ( u ) ) { u = `nats://${ u }` ; } this . url = url . parse ( u ) ; if ( ! this . url . port ) { this . url . port = `${ DEFAULT_PORT }` ; } this . didConnect = false ; this . reconnects = 0 ; this . lastConnect = 0 ; this . implicit = implicit ; } toString ( ) { return this . url . href || '' ; } getCredentials ( ) { if ( 'auth' in this . url && ! ! this . url . auth ) { return this . url . auth . split ( ':' ) ; } return undefined ; } } hidden export class Servers { private firstSelect : boolean = true ; private readonly servers : Server [ ] ; private currentServer : Server ; constructor ( randomize , urls , firstServer ? ) { this . servers = [ ] as Server [ ] ; if ( urls ) { urls . forEach ( element => { this . servers . push ( new Server ( element ) ) ; } ) ; if ( randomize ) { this . servers = shuffle ( this . servers ) ; } } if ( firstServer ) { let index = urls . indexOf ( firstServer ) ; if ( index === - 1 ) { this . addServer ( firstServer , false ) ; } else { let fs = this . servers [ index ] ; this . servers . splice ( index , 1 ) ; this . servers . unshift ( fs ) ; } } else { if ( this . servers . length === 0 ) { this . addServer ( DEFAULT_URI , false ) ; } } this . currentServer = this . servers [ 0 ] ; } getCurrentServer ( ) { return this . currentServer ; } addServer ( u , implicit = false ) { this . servers . push ( new Server ( u , implicit ) ) ; } selectServer ( ) { if ( this . firstSelect ) { this . firstSelect = false ; return this . currentServer ; } let t = this . servers . shift ( ) ; if ( t ) { this . servers . push ( t ) ; this . currentServer = t ; } return t ; } removeCurrentServer ( ) { this . removeServer ( this . currentServer ) ; } removeServer ( server ) { if ( server ) { let index = this . servers . indexOf ( server ) ; this . servers . splice ( index , 1 ) ; } } length ( ) { return this . servers . length ; } next ( ) { return this . servers . length ? this . servers [ 0 ] : undefined ; } getServers ( ) { return this . servers ; } processServerUpdate ( info ) { let added = [ ] ; let deleted = [ ] ; if ( info . connect_urls && info . connect_urls . length > 0 ) { let discovered = { } ; info . connect_urls . forEach ( server => { let protocol = this . currentServer . url . protocol ; let u = `${ protocol }//${ server }` ; discovered [ u ] = new Server ( u , true ) ; } ) ; let toDelete = [ ] ; this . servers . forEach ( ( s , index ) => { let u = s . toString ( ) ; if ( s . implicit && this . currentServer . url . href !== u && discovered [ u ] === undefined ) { toDelete . push ( index ) ; } delete discovered [ u ] ; } ) ; toDelete . reverse ( ) ; toDelete . forEach ( index => { let removed = this . servers . splice ( index , 1 ) ; deleted = deleted . concat ( removed [ 0 ] . url . toString ( ) ) ; } ) ; for ( let k in discovered ) { if ( discovered . hasOwnProperty ( k ) ) { this . servers . push ( discovered [ k ] ) ; added . push ( k ) ; } } } return { added : added , deleted : deleted } as ServersChangedEvent ; } }