lib/kafka/Consumer.js
/**
   * Registers a callback to handle messages from a specific kafka topic.
   * If the kafka consumer is not subscribed on the given topic, it does.
   * Note that:
   * - subscriptions are not made immediately, it waits the
   * 'subscription.backoff.min.ms' to do it;
   * - errors happens if the topic does not exist and auto creation is disabled,
   * but in the future if the topic is created, the consumer will subscribe on it
   * according to the 'kafka.topic.metadata.refresh.interval.ms'(configurable by
   * config on constructor).
   *
   * @param {*} topic the target kafka topic, it could be a String or a RegExp
   * @param {*} callback async callback (data): Promise
   * Where data is an object with the following content:
   *   {
   *     value: Buffer.from('hi'), // message contents as a Buffer
   *     size: 2, // size of the message, in bytes
   *     topic: 'librdtesting-01', // topic the message comes from
   *     offset: 1337, // offset the message was read from
   *     partition: 1, // partition the message was on
   *     key: 'someKey', // key of the message if present
   *     timestamp: 1510325354780 // timestamp of message creation
   *   }
   * .
   *
   * @return an identifier (uuidv4 string) that represents the association between the
   * topic and the callback. You need to store it if you intend to remove this association
   * in some moment.
   *
   * @example subscribe(/^notifications\/.*\/critical/), handleAllCriticalNotifications)
   */
Registers a callback to handle messages from a specific kafka topic.
If the kafka consumer is not subscribed on the given topic, it does.
Note that:
- subscriptions are not made immediately, it waits the
'subscription.backoff.min.ms' to do it;
- errors happens if the topic does not exist and auto creation is disabled,
but in the future if the topic is created, the consumer will subscribe on it
according to the 'kafka.topic.metadata.refresh.interval.ms'(configurable by
config on constructor).



@return an identifier (uuidv4 string) that represents the association between the
topic and the callback. You need to store it if you intend to remove this association
in some moment.


Registers a callback to handle messages from a specific kafka topic . If the kafka consumer is not subscribed on the given topic it does . Note that : - subscriptions are not made immediately it waits the ' subscription . backoff . min . ms ' to do it ; - errors happens if the topic does not exist and auto creation is disabled but in the future if the topic is created the consumer will subscribe on it according to the ' kafka . topic . metadata . refresh . interval . ms ' ( configurable by config on constructor ) . @return an identifier ( uuidv4 string ) that represents the association between the topic and the callback . You need to store it if you intend to remove this association in some moment .
registerCallback(topic, callback) {
    logger.debug(`subscribing on topic: ${topic}`);
    let needToRefreshSubscriptions = true;
    const id = uuidv4();
    const registryEntry = {
      callback,
      id,
    };
    if (typeof topic === 'object') {
      registryEntry.regExp = topic;
      this.topicRegExpArray.push(registryEntry);
    } else {
      this.topicMap[topic] = this.topicMap[topic] || [];
      this.topicMap[topic].push(registryEntry);
      if (this.topicMap[topic].length > 1) {
        needToRefreshSubscriptions = false;
      }
    }
    if (needToRefreshSubscriptions) {
      this.refreshSubscriptions();
    }
    return id;
  }
registerCallback ( topic , callback ) { logger . debug ( ` ${ topic } ` ) ; let needToRefreshSubscriptions = true ; const id = uuidv4 ( ) ; const registryEntry = { callback , id , } ; if ( typeof topic === 'object' ) { registryEntry . regExp = topic ; this . topicRegExpArray . push ( registryEntry ) ; } else { this . topicMap [ topic ] = this . topicMap [ topic ] || [ ] ; this . topicMap [ topic ] . push ( registryEntry ) ; if ( this . topicMap [ topic ] . length > 1 ) { needToRefreshSubscriptions = false ; } } if ( needToRefreshSubscriptions ) { this . refreshSubscriptions ( ) ; } return id ; }
Registers a callback to handle messages from a specific kafka topic.
Registers a callback to handle messages from a specific kafka topic .
// init structure // add new entry // update kafka topic subscriptions