lib/offset.js
function fetchOffsets (offset, topics, cb, when) {
  if (!offset.ready) {
    if (when === -1) {
      offset.once('ready', () => offset.fetchLatestOffsets(topics, cb));
    } else if (when === -2) {
      offset.once('ready', () => offset.fetchEarliestOffsets(topics, cb));
    }
    return;
  }
  async.waterfall(
    [
      callback => {
        offset.client.loadMetadataForTopics(topics, callback);
      },
      (topicsMetaData, callback) => {
        var payloads = [];
        var metaDatas = topicsMetaData[1].metadata;
        Object.keys(metaDatas).forEach(function (topicName) {
          var topic = metaDatas[topicName];
          Object.keys(topic).forEach(function (partition) {
            payloads.push({
              topic: topicName,
              partition: partition,
              time: when
            });
          });
        });

        if (payloads.length === 0) {
          return callback(new Error('Topic(s) does not exist'));
        }

        offset.fetch(payloads, callback);
      },
      function (results, callback) {
        Object.keys(results).forEach(function (topicName) {
          var topic = results[topicName];

          Object.keys(topic).forEach(function (partitionName) {
            topic[partitionName] = topic[partitionName][0];
          });
        });
        callback(null, results);
      }
    ],
    cb
  );
}
function fetchOffsets (offset, topics, cb, when) {
  if (!offset.ready) {
    if (when === -1) {
      offset.once('ready', () => offset.fetchLatestOffsets(topics, cb));
    } else if (when === -2) {
      offset.once('ready', () => offset.fetchEarliestOffsets(topics, cb));
    }
    return;
  }
  async.waterfall(
    [
      callback => {
        offset.client.loadMetadataForTopics(topics, callback);
      },
      (topicsMetaData, callback) => {
        var payloads = [];
        var metaDatas = topicsMetaData[1].metadata;
        Object.keys(metaDatas).forEach(function (topicName) {
          var topic = metaDatas[topicName];
          Object.keys(topic).forEach(function (partition) {
            payloads.push({
              topic: topicName,
              partition: partition,
              time: when
            });
          });
        });

        if (payloads.length === 0) {
          return callback(new Error('Topic(s) does not exist'));
        }

        offset.fetch(payloads, callback);
      },
      function (results, callback) {
        Object.keys(results).forEach(function (topicName) {
          var topic = results[topicName];

          Object.keys(topic).forEach(function (partitionName) {
            topic[partitionName] = topic[partitionName][0];
          });
        });
        callback(null, results);
      }
    ],
    cb
  );
}
function fetchOffsets ( offset , topics , cb , when ) { if ( ! offset . ready ) { if ( when === - 1 ) { offset . once ( 'ready' , ( ) => offset . fetchLatestOffsets ( topics , cb ) ) ; } else if ( when === - 2 ) { offset . once ( 'ready' , ( ) => offset . fetchEarliestOffsets ( topics , cb ) ) ; } return ; } async . waterfall ( [ callback => { offset . client . loadMetadataForTopics ( topics , callback ) ; } , ( topicsMetaData , callback ) => { var payloads = [ ] ; var metaDatas = topicsMetaData [ 1 ] . metadata ; Object . keys ( metaDatas ) . forEach ( function ( topicName ) { var topic = metaDatas [ topicName ] ; Object . keys ( topic ) . forEach ( function ( partition ) { payloads . push ( { topic : topicName , partition : partition , time : when } ) ; } ) ; } ) ; if ( payloads . length === 0 ) { return callback ( new Error ( 'Topic(s) does not exist' ) ) ; } offset . fetch ( payloads , callback ) ; } , function ( results , callback ) { Object . keys ( results ) . forEach ( function ( topicName ) { var topic = results [ topicName ] ; Object . keys ( topic ) . forEach ( function ( partitionName ) { topic [ partitionName ] = topic [ partitionName ] [ 0 ] ; } ) ; } ) ; callback ( null , results ) ; } ] , cb ) ; }
private helper
private helper