import { ChildProcess , spawn , execSync } from 'child_process' ; import * as net from 'net' ; import { Socket } from 'net' ; import path from 'path' ; import fs from 'fs' ; import { URL } from 'url' ; import Timer = NodeJS . Timer ; let SERVER = ( process . env . TRAVIS ) ? 'nats-server/nats-server' : 'nats-server' ; let PID_DIR = ( process . env . TRAVIS ) ? process . env . TRAVIS_BUILD_DIR : process . env . TMPDIR ; let SERVER_VERSION ; export interface SC { server : Server ; servers : Server [ ] ; } export interface Ports { nats : string [ ] ; monitoring : string [ ] | undefined ; cluster : string [ ] | undefined ; profile : string [ ] | undefined ; } export interface Server extends ChildProcess { args : string [ ] ; nats : string ; ports : Ports ; port : number ; clusterPort : number ; } export function natsURL ( s ) { return s . nats ; } export function wsURL ( s ) { return s . nats ; } export function getPort ( urlString ) { let u = new URL ( urlString ) ; return parseInt ( u . port , 10 ) ; } export function addClusterMember ( s , opt_flags ? ) { return new Promise ( ( resolve , reject ) => { opt_flags = opt_flags || [ ] ; if ( opt_flags . indexOf ( '--routes' ) !== - 1 ) { reject ( new Error ( 'addClusterMember doesn\'t take a --routes flag as an option' ) ) ; return ; } opt_flags = opt_flags . concat ( [ '--routes' , `nats://127.0.0.1:${ s . clusterPort }` ] ) ; startServer ( opt_flags ) . then ( ( v ) => { resolve ( v ) ; } ) . catch ( ( err ) => { reject ( err ) ; } ) ; } ) ; } export function startServer ( opt_flags ? ) { return new Promise ( ( resolve , reject ) => { opt_flags = opt_flags || [ ] ; let flags = [ ] ; if ( opt_flags . indexOf ( '-a' ) === - 1 ) { flags = flags . concat ( [ '-a' , '127.0.0.1' ] ) ; } if ( opt_flags . indexOf ( '-p' ) === - 1 && opt_flags . indexOf ( '--port' ) === - 1 ) { flags = flags . concat ( [ '-p' , '-1' ] ) ; } if ( opt_flags . indexOf ( '--cluster' ) === - 1 ) { flags = flags . concat ( [ '--cluster' , 'nats://127.0.0.1:-1' ] ) ; } if ( opt_flags . indexOf ( '--http_port' ) === - 1 && opt_flags . indexOf ( '-m' ) === - 1 ) { flags = flags . concat ( [ '--m' , '-1' ] ) ; } flags = flags . concat ( [ '--ports_file_dir' , PID_DIR ] as string [ ] ) ; let port = - 1 ; if ( opt_flags ) { flags = flags . concat ( opt_flags ) ; } if ( process . env . PRINT_LAUNCH_CMD ) { console . log ( flags . join ( ' ' ) ) ; } let server = spawn ( SERVER , flags ) as Server ; server . args = flags ; let start = Date . now ( ) ; let wait = 0 ; let maxWait = 5 * 1000 ; let delta = 250 ; let socket ; let timer ; function resetSocket ( ) { if ( socket ) { socket . removeAllListeners ( ) ; socket . destroy ( ) ; socket = null ; } } function finish ( err ? ) { resetSocket ( ) ; if ( timer ) { clearInterval ( timer ) ; timer = null ; } if ( err === undefined ) { resolve ( server ) ; return ; } reject ( err ) ; } let count = 50 ; new Promise < any > ( ( rslv , rjct ) => { let t = setInterval ( ( ) => { -- count ; if ( count === 0 ) { clearInterval ( t ) ; rjct ( 'Unable to find the pid' ) ; } let portsFile = path . join ( PID_DIR , `nats-server_${ server . pid }.ports` ) ; if ( fs . existsSync ( portsFile ) ) { let data = fs . readFileSync ( portsFile ) . toString ( ) ; let s = ( server as Server ) ; let ports = JSON . parse ( data ) ; s . nats = ports . nats [ 0 ] ; s . port = port = getPort ( server . nats ) ; s . clusterPort = getPort ( ports . cluster [ 0 ] ) ; s . ports = ports ; clearInterval ( t ) ; rslv ( ) ; } } , 150 ) ; } ) . then ( ( ) => { timer = < any > setInterval ( function ( ) { resetSocket ( ) ; wait = Date . now ( ) - start ; if ( wait > maxWait ) { finish ( new Error ( 'Can\'t connect to server on port: ' + port ) ) ; } socket = net . createConnection ( port ) ; socket . on ( 'connect' , function ( ) { if ( server . pid === null ) { finish ( new Error ( 'Server already running on port: ' + port ) ) ; } else { finish ( ) ; } } ) ; socket . on ( 'error' , function ( error ) { finish ( new Error ( 'Problem connecting to server on port: ' + port + ' (' + error + ')' ) ) ; } ) ; } , delta ) ; } ) . catch ( ( err ) => { reject ( err ) ; } ) ; server . on ( 'exit' , function ( code , signal ) { if ( code === 1 ) { finish ( new Error ( 'Server exited with bad code, already running? (' + code + ' / ' + signal + ')' ) ) ; } } ) ; server . stderr . on ( 'data' , function ( data ) { if ( /^execvp\(\)/ . test ( data . toString ( ) ) ) { if ( timer ) { clearInterval ( timer ) ; } finish ( new Error ( 'Can\'t find the ' + SERVER ) ) ; } } ) ; } ) ; } function wait ( server , done ? ) { if ( server . killed ) { if ( done ) { done ( ) ; } } else { setTimeout ( function ( ) { wait ( server , done ) ; } , 0 ) ; } } export function stopServer ( server , done ? ) { if ( server && ! server . killed ) { server . kill ( ) ; wait ( server , done ) ; } else if ( done ) { done ( ) ; } } export function serverVersion ( ) { if ( SERVER_VERSION === undefined ) { SERVER_VERSION = initServerVersion ( ) ; } return SERVER_VERSION ; } function initServerVersion ( ) { let v = execSync ( SERVER + ' -v' , { timeout : 1000 } ) . toString ( ) ; return normalizeVersion ( v ) ; } function normalizeVersion ( s ) { s = s . replace ( 'version ' , 'v' ) ; s = s . replace ( ':' , '' ) ; s = s . replace ( 'nats-server ' , '' ) ; s = s . replace ( 'nats-server' , '' ) ; s = s . replace ( 'v' , '' ) ; let i = s . indexOf ( '-' ) ; if ( i !== - 1 ) { s = s . substring ( 0 , i ) ; } let a = s . split ( '.' ) ; a . forEach ( ( v , i ) => { let vv = parseInt ( v , 10 ) ; if ( ! isNaN ( vv ) ) { a [ i ] = vv ; } } ) ; return a ; }