<commit_before>const highland = require('highland');
const kafka = require('kafka-node');
const config = require('../config/config');

/*
 * This is a KAFKA Consumer base service module, which subscribes to Kafka topic perpetually
 *
 * This module can be used as a base module method, wherever kafka topic subscription is needed
 */

// A client (calee) needs to specify
// - Topic name it wants to subscribe to
// - Consumer group it belongs to
// - Kafka host it wants to connect to
// - A highland Processing Pipeline, how the messages are processed

function run(subscribeTopic, consumerGroup, kafkaHost, processPipeLine) {
  if (!subscribeTopic || subscribeTopic == '') {
    throw new Error('Invalid subscription details for consumer..!');
    return;
  }

  kafkaHost = kafkaHost || config.KAFKA_HOST;
  consumerGroup = consumerGroup || '';

  highland(function(push, next) {
      let client = new kafka.Client(kafkaHost);
      let topics = [{
        topic: subscribeTopic
      }];
      let options = {
        groupId: consumerGroup,
        autoCommit: true //Making it autocommit so that message offset is moved after consuming the message
      }

      let consumer = new kafka.Consumer(client, topics, options);

      consumer.on('message', function(message) {
        // console.log('Message received: ', message);

        //If message is not JSON, parse it as JSON here, before passing it to the rest of the pipeline

        //Push the message from kafka to rest of the pipeline
        push(null, message);

        //Start calling the generator again for listening to next message
        // next(); //Commenting this as processing is currently slower than message producer
      });

      consumer.on('error', function(err) {
        console.log("Error: ", err);

        push(err, null);
        next();
      });
    }).map(function(messageObj) {
      //Temporarily keeping this map method, to intermediary log and verify if messages are coming from Kafka or not
      //Once well tested, this method can be removed
      console.log('[*] Received a message in pipeline: ', messageObj);

      //If not returned, the message will not be propagated to next set of pipeline<commit_msg>Minor bug in the pipeline, fixed<commit_after>const highland = require('highland');
const kafka = require('kafka-node');
const config = require('../config/config');

/*
 * This is a KAFKA Consumer base service module, which subscribes to Kafka topic perpetually
 *
 * This module can be used as a base module method, wherever kafka topic subscription is needed
 */

// A client (calee) needs to specify
// - Topic name it wants to subscribe to
// - Consumer group it belongs to
// - Kafka host it wants to connect to
// - A highland Processing Pipeline, how the messages are processed

function run(subscribeTopic, consumerGroup, kafkaHost, processPipeLine) {
  if (!subscribeTopic || subscribeTopic == '') {
    throw new Error('Invalid subscription details for consumer..!');
    return;
  }

  kafkaHost = kafkaHost || config.KAFKA_HOST;
  consumerGroup = consumerGroup || '';

  highland(function(push, next) {
      let client = new kafka.Client(kafkaHost);
      let topics = [{
        topic: subscribeTopic
      }];
      let options = {
        groupId: consumerGroup,
        autoCommit: true //Making it autocommit so that message offset is moved after consuming the message
      }

      let consumer = new kafka.Consumer(client, topics, options);

      consumer.on('message', function(message) {
        // console.log('Message received: ', message);

        //If message is not JSON, parse it as JSON here, before passing it to the rest of the pipeline

        //Push the message from kafka to rest of the pipeline
        push(null, message);

        //Start calling the generator again for listening to next message
        // next(); //Commenting this as processing is currently slower than message producer
      });

      consumer.on('error', function(err) {
        console.log("Error: ", err);

        push(err, null);
        // next();
      });
    }).map(function(messageObj) {
      //Temporarily keeping this map method, to intermediary log and verify if messages are coming from Kafka or not
      //Once well tested, this method can be removed
      console.log('[*] Received a message in pipeline: ', messageObj);

      //If not returned, the message will not be propagated to next set of pipeline
{'fraction_non_alphanumeric': 0.09685393258426966, 'fraction_numerical': 0.0, 'mean_word_length': 2.9564444444444447, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '5362100', 'n_tokens_mistral': 1171, 'n_tokens_neox': 1120, 'n_words': 486}