lib/kafka/Consumer.js
/**
   * Unregister a callback previously registered by the 'registerCallback' method.
   * @param {*} registerId the subscription id that belongs to the callback register
   * that you desire to remove
   */
Unregister a callback previously registered by the 'registerCallback' method.
@param {*} registerId the subscription id that belongs to the callback register
that you desire to remove
Unregister a callback previously registered by the ' registerCallback ' method . @param { * } registerId the subscription id that belongs to the callback register that you desire to remove
unregisterCallback(registerId) {
    logger.debug(`unsubscribing id: ${registerId}`);
    let needToRefreshSubscriptions = false;
    const topicsToBeRemoved = [];
    const topicsRegExpBefore = this.topicRegExpArray.length;
    this.topicRegExpArray = this.topicRegExpArray.filter(
      (entry) => entry.id !== registerId,
    );
    if (topicsRegExpBefore !== this.topicRegExpArray.length) {
      needToRefreshSubscriptions = true;
    } else {
      Object.keys(this.topicMap).forEach((topic) => {
        this.topicMap[topic] = this.topicMap[topic].filter(
          (entry) => entry.id !== registerId,
        );
        if (this.topicMap[topic].length === 0) {
          needToRefreshSubscriptions = true;
          topicsToBeRemoved.push(topic);
        }
      });
      topicsToBeRemoved.forEach((topic) => {
        delete this.topicMap[topic];
      });
    }
    if (needToRefreshSubscriptions) {
      this.refreshSubscriptions();
    }
  }
unregisterCallback ( registerId ) { logger . debug ( ` ${ registerId } ` ) ; let needToRefreshSubscriptions = false ; const topicsToBeRemoved = [ ] ; const topicsRegExpBefore = this . topicRegExpArray . length ; this . topicRegExpArray = this . topicRegExpArray . filter ( ( entry ) => entry . id !== registerId , ) ; if ( topicsRegExpBefore !== this . topicRegExpArray . length ) { needToRefreshSubscriptions = true ; } else { Object . keys ( this . topicMap ) . forEach ( ( topic ) => { this . topicMap [ topic ] = this . topicMap [ topic ] . filter ( ( entry ) => entry . id !== registerId , ) ; if ( this . topicMap [ topic ] . length === 0 ) { needToRefreshSubscriptions = true ; topicsToBeRemoved . push ( topic ) ; } } ) ; topicsToBeRemoved . forEach ( ( topic ) => { delete this . topicMap [ topic ] ; } ) ; } if ( needToRefreshSubscriptions ) { this . refreshSubscriptions ( ) ; } }
Unregister a callback previously registered by the 'registerCallback' method.
Unregister a callback previously registered by the ' registerCallback ' method .
// check if the id belongs to a regular expression registry