lib/kafka/Consumer.js
/**
   * Initializes the consumer.
   * @returns a Promise that is fullfil when the consumer becomes ready.
   */
Initializes the consumer.
@returns a Promise that is fullfil when the consumer becomes ready.
Initializes the consumer . @returns a Promise that is fullfil when the consumer becomes ready .
async init() {
    return new Promise((resolve, reject) => {
      this.consumer.on('event.error', (event) => {
        logger.warn(`Kafka event.error: ${event}`);
      });
      this.consumer.on('data', this.onData.bind(this));
      this.consumer.on('ready', () => {
        logger.info('Consumer is ready');
        this.isReady = true;
        this.commitManager.init();
        this.refreshSubscriptions();
        this.consumer.consume();
        this.eventEmitter.emit('ready');
        return resolve();
      });
      this.consumer.on('disconnected', (arg) => {
        this.isReady = false;
        this.isWaitingForRefreshSubscriptions = false;
        this.topicMap = {};
        this.topicRegExpArray = [];
        this.currQueueBytes = 0;
        this.isPaused = false;
        this.msgQueue = null;
        this.unackMsgBytes = 0;
        this.epoch = 0;
        this.eventEmitter.emit('disconnected');
        logger.info(`Consumer disconnected. ${util.inspect(arg)}`);
      });
      this.consumer.connect(undefined, (error) => {
        if (error) {
          logger.error(`Error on connect: ${error}`);
          this.eventEmitter.emit('error.connecting');
          reject(error);
        }
      });
    });
  }
async init ( ) { return new Promise ( ( resolve , reject ) => { this . consumer . on ( 'event.error' , ( event ) => { logger . warn ( ` ${ event } ` ) ; } ) ; this . consumer . on ( 'data' , this . onData . bind ( this ) ) ; this . consumer . on ( 'ready' , ( ) => { logger . info ( 'Consumer is ready' ) ; this . isReady = true ; this . commitManager . init ( ) ; this . refreshSubscriptions ( ) ; this . consumer . consume ( ) ; this . eventEmitter . emit ( 'ready' ) ; return resolve ( ) ; } ) ; this . consumer . on ( 'disconnected' , ( arg ) => { this . isReady = false ; this . isWaitingForRefreshSubscriptions = false ; this . topicMap = { } ; this . topicRegExpArray = [ ] ; this . currQueueBytes = 0 ; this . isPaused = false ; this . msgQueue = null ; this . unackMsgBytes = 0 ; this . epoch = 0 ; this . eventEmitter . emit ( 'disconnected' ) ; logger . info ( ` ${ util . inspect ( arg ) } ` ) ; } ) ; this . consumer . connect ( undefined , ( error ) => { if ( error ) { logger . error ( ` ${ error } ` ) ; this . eventEmitter . emit ( 'error.connecting' ) ; reject ( error ) ; } } ) ; } ) ; }
Initializes the consumer.
Initializes the consumer .
// register handler for kafka events // error // data // ready // note: the ready function is called just once /**
       * this event is reached only when consumer.disconnect() is called
       * otherwise the consumer will keep trying to reconnect
       */ // connect to kafka