lib/harvest.js
/**
   * @method connect
   * @description connect to redis and kafka, subscribe to harvest
   * update kafka topic
   */
@method connect
@description connect to redis and kafka, subscribe to harvest
update kafka topic
@method connect @description connect to redis and kafka subscribe to harvest update kafka topic
async connect() {
    logger.info('waiting from redis');
    await waitUntil(config.redis.host, config.redis.port);
    logger.info('redis online, connecting');
    await redis.connect();
    logger.info('waiting for kafka');
    await waitUntil(config.kafka.host, config.kafka.port);
    logger.info('kafka online, connecting');
    this.consumer = new Consumer({
      'group.id': config.kafka.groups.harvest,
      'metadata.broker.list': KAFKA_HOST_PORT,
    },{
      'auto.offset.reset' : 'earliest'
    });
    this.producer = new Producer({
      'metadata.broker.list': KAFKA_HOST_PORT,
    });
    logger.info('Connecting top kafka topic: '+config.kafka.topics.harvestRunning);
    await this.producer.connect();
    await this.consumer.connect();
    await this.ensureTopic(config.kafka.topics.harvestQueue);
    await this.ensureTopic(config.kafka.topics.harvestRunning);
    logger.info('Subscribing to topic: '+config.kafka.topics.harvestRunning);
    await this.consumer.subscribe([config.kafka.topics.harvestRunning]);
    await this.consumer.consume(msg => this.handleHarvestUpdate(msg));
  }
async connect ( ) { logger . info ( 'waiting from redis' ) ; await waitUntil ( config . redis . host , config . redis . port ) ; logger . info ( 'redis online, connecting' ) ; await redis . connect ( ) ; logger . info ( 'waiting for kafka' ) ; await waitUntil ( config . kafka . host , config . kafka . port ) ; logger . info ( 'kafka online, connecting' ) ; this . consumer = new Consumer ( { 'group.id' : config . kafka . groups . harvest , 'metadata.broker.list' : KAFKA_HOST_PORT , } , { 'auto.offset.reset' : 'earliest' } ) ; this . producer = new Producer ( { 'metadata.broker.list' : KAFKA_HOST_PORT , } ) ; logger . info ( 'Connecting top kafka topic: ' + config . kafka . topics . harvestRunning ) ; await this . producer . connect ( ) ; await this . consumer . connect ( ) ; await this . ensureTopic ( config . kafka . topics . harvestQueue ) ; await this . ensureTopic ( config . kafka . topics . harvestRunning ) ; logger . info ( 'Subscribing to topic: ' + config . kafka . topics . harvestRunning ) ; await this . consumer . subscribe ( [ config . kafka . topics . harvestRunning ] ) ; await this . consumer . consume ( msg => this . handleHarvestUpdate ( msg ) ) ; }
@method connect
 @description connect to redis and kafka, subscribe to harvest
 update kafka topic
@method connect @description connect to redis and kafka subscribe to harvest update kafka topic
// make sure we read any unread messages