const EventEmitter = require ( 'events' ) ; import * as Types from './kafkaInterface' ; import CascadeProducer from './cascadeProducer' ; import CascadeConsumer from './cascadeConsumer' ; /**
 * CascadeService
 */ class CascadeService extends EventEmitter { kafka : Types . KafkaInterface ; topic : string ; serviceCB : Types . ServiceCallback ; successCB : ( ... args : any [ ] ) => any ; dlqCB : Types . RouteCallback ; producer : CascadeProducer ; consumer : CascadeConsumer ; events = [ 'connect' , 'disconnect' , 'run' , 'stop' , 'pause' , 'resume' , 'receive' , 'success' , 'retry' , 'dlq' , 'error' , 'serviceError' , ] ; /**
     * CascadeService objects should be constructed from [cascade.service] module constructor ( kafka , topic , groupId , serviceCB , successCB , dlqCB ) { super ( ) ; this . kafka = kafka ; this . topic = topic ; this . serviceCB = serviceCB ; this . successCB = successCB ; this . dlqCB = dlqCB ; this . retries = 0 ; this . topicsArr = [ ] ; this . producer = new CascadeProducer ( kafka , topic , dlqCB ) ; this . producer . on ( 'retry' , ( msg ) => this . emit ( 'retry' , msg ) ) ; this . producer . on ( 'dlq' , ( msg ) => this . emit ( 'dlq' , msg ) ) ; this . producer . on ( 'error' , ( error ) => this . emit ( 'error' , 'Error in cascade producer: ' + error ) ) ; this . consumer = new CascadeConsumer ( kafka , topic , groupId , false ) ; this . consumer . on ( 'receive' , ( msg ) => this . emit ( 'receive' , msg ) ) ; this . consumer . on ( 'serviceError' , ( error ) => this . emit ( 'serviceError' , error ) ) ; this . consumer . on ( 'error' , ( error ) => this . emit ( 'error' , 'Error in cascade consumer: ' + error ) ) ; } returns { Promise } connect ( ) { return new Promise ( async ( resolve , reject ) => { try { await this . producer . connect ( ) ; await this . consumer . connect ( ) ; resolve ( true ) ; this . emit ( 'connect' ) ; } catch ( error ) { reject ( error ) ; this . emit ( 'error' , 'Error in cascade.connect(): ' + error ) ; } } ) ; } returns { Promise } disconnect ( ) { return new Promise ( ( resolve , reject ) => { this . producer . stop ( ) . then ( ( ) => { this . producer . disconnect ( ) . then ( ( ) => { this . consumer . disconnect ( ) . then ( ( ) => { resolve ( true ) ; this . emit ( 'disconnect' ) ; } ) . catch ( error => { reject ( error ) ; this . emit ( 'error' , 'Error in cascade.disconnect(): [CONSUMER]' + error ) ; } ) ; } ) . catch ( error => { reject ( error ) ; this . emit ( 'error' , 'Error in cascade.disconnect(): [PRODUCER:DISCONNECT]' + error ) ; } ) ; } ) . catch ( error => { reject ( error ) ; this . emit ( 'error' , 'Error in cascade.disconnect(): [PRODUCER:STOP]' + error ) ; } ) ; } ) ; } param { number } levels param { object } options returns { promise } setDefaultRoute ( levels , options ? ) { return new Promise ( ( resolve , reject ) => { this . producer . setDefaultRoute ( levels , options ) . then ( res => resolve ( res ) ) . catch ( error => { reject ( error ) ; this . emit ( 'error' , error ) ; } ) ; } ) ; } param { string } status param { number } levels param { object } options returns { Promise } setRoute ( status , levels , options ? ) { return new Promise ( ( resolve , reject ) => { this . producer . setRoute ( status , levels , options ) . then ( res => resolve ( res ) ) . catch ( error => { reject ( error ) ; this . emit ( 'error' , error ) ; } ) ; } ) ; } returns { string [ ] } getKafkaTopics ( ) { let topics = [ ] ; this . producer . routes . forEach ( route => topics = topics . concat ( route . topics ) ) ; return topics ; } returns { Promise } run ( ) { return new Promise ( async ( resolve , reject ) => { try { const status = await this . consumer . run ( this . serviceCB , ( ... args ) => { this . emit ( 'success' , ... args ) ; this . successCB ( ... args ) ; } , async ( msg , status = '' ) => { try { await this . producer . send ( msg , status ) ; } catch ( error ) { this . emit ( 'error' , 'Error in cascade producer.send(): ' + error ) ; } } ) ; resolve ( status ) ; this . emit ( 'run' ) ; } catch ( error ) { reject ( error ) ; this . emit ( 'error' , 'Error in cascade.run(): ' + error ) ; } } ) ; } returns { Promise } stop ( ) { return new Promise ( async ( resolve , reject ) => { try { await this . consumer . stop ( ) ; await this . producer . stop ( ) ; resolve ( true ) ; this . emit ( 'stop' ) ; } catch ( error ) { reject ( error ) ; this . emit ( 'error' , 'Error in cascade.stop(): ' + error ) ; } } ) ; } returns { Promise } async pause ( ) { if ( ! this . producer . paused ) { return new Promise ( async ( resolve , reject ) => { try { await this . consumer . pause ( ) ; this . producer . pause ( ) ; resolve ( true ) ; this . emit ( 'pause' ) ; } catch ( error ) { reject ( error ) ; this . emit ( 'error' , 'Error in cascade.pause(): ' + error ) ; } } ) ; } else { console . log ( 'cascade.pause() called while service is already paused!' ) ; } } returns { boolean } paused ( ) { return this . producer . paused ; } returns { Promise } async resume ( ) { if ( this . producer . paused ) { return new Promise ( async ( resolve , reject ) => { try { await this . consumer . resume ( ) ; await this . producer . resume ( ) ; resolve ( true ) ; this . emit ( 'resume' ) ; } catch ( error ) { reject ( error ) ; this . emit ( 'error' , 'Error in cascade.resume(): ' + error ) ; } } ) ; } else { console . log ( 'cascade.resume() called while service is already running!' ) ; } } on ( event , callback ) { if ( ! this . events . includes ( event ) ) throw new Error ( 'Unknown event: ' + event ) ; super . on ( event , callback ) ; } } export default CascadeService ;