const EventEmitter = require ( 'events' ) ; import * as Types from './kafkaInterface' ; class CascadeConsumer extends EventEmitter { consumer : Types . ConsumerInterface ; topic : string ; groupId : string ; fromBeginning : boolean ; constructor ( kafkaInterface , topic , groupId , fromBeginning = false ) { super ( ) ; this . consumer = kafkaInterface . consumer ( { groupId } ) ; this . topic = topic ; this . groupId = groupId ; this . fromBeginning = fromBeginning ; } connect ( ) { return new Promise ( async ( resolve , reject ) => { try { await this . consumer . connect ( ) ; console . log ( 'Connected to the consumer...' ) ; await this . consumer . subscribe ( { topic : this . topic , fromBeginning : this . fromBeginning } ) ; console . log ( 'Subscribed to the base topic:' , this . topic ) ; let re = new RegExp ( `^${ this . topic }-cascade-retry-.*` ) ; await this . consumer . subscribe ( { topic : re , fromBeginning : this . fromBeginning } ) ; console . log ( 'Connected to the retry topics...' ) ; resolve ( true ) ; } catch ( error ) { this . emit ( 'error' , error ) ; reject ( error ) ; } } ) ; } run ( serviceCB , successCB , rejectCB ) { return this . consumer . run ( { eachMessage : ( msg ) => { try { if ( ! msg . message . headers ) { msg . message . headers = { } ; } if ( ! msg . message . headers . cascadeMetadata ) { msg . message . headers . cascadeMetadata = JSON . stringify ( { status : '' , retries : 0 , topicArr : [ ] , } ) ; } if ( msg . topic === this . topic ) this . emit ( 'receive' , msg ) ; } catch ( error ) { this . emit ( 'error' , error ) ; } try { serviceCB ( msg , successCB , rejectCB ) ; } catch ( error ) { this . emit ( 'serviceError' , error ) ; } } } ) ; } disconnect ( ) { return this . consumer . disconnect ( ) ; } stop ( ) { return this . consumer . stop ( ) ; } pause ( ) { return this . consumer . pause ( ) ; } resume ( ) { return this . consumer . resume ( ) ; } on ( event , callback ) { super . on ( event , callback ) ; } } export default CascadeConsumer ;