<commit_before>var {EventEmitter} = require('fbemitter');
var helper = require('./helper.js');

class channelManager {
  constructor() {
    this.emitter = new EventEmitter();
    this.channels = {};
    this.streamRef = {};
    this.receive = this.receive.bind(this);
  }
  setConfig(config) {
    this.config = config;
  }
  
  // Receive: This method will be executed whenever dependency value changes
  // It receives which dependency changes and which channeldId should be affected.
  receive(depend, channelId) {
    let self = this;
    let channelObj = this.channels[channelId];
    let queryObj = this.queryBuild(channelObj.depends, channelObj.previousSelectedSensor, channelObj.size);
    let validQuery = true;
    try {
      validQuery = !queryObj.body.aggs && queryObj.body.query.bool.should.length === 0 ? false : true;
    } catch(e) { }
    
    if(validQuery) {
      // apply search query and emit historic queryResult
      helper.appbaseRef.search(queryObj).on('data', function(data) {
        let obj = {
          method: 'historic',
          data: data
        };
        self.emitter.emit(channelId, obj);
      }).on('error', function(error) {
        console.log(error);
      });
      // apply searchStream query and emit streaming data
      if(this.streamRef[channelId]) {
        this.streamRef[channelId].stop();
      } 
      this.streamRef[channelId] = helper.appbaseRef.searchStream(queryObj).on('data', function(data) {
        let obj = {
          method: 'stream',
          data: data
        };
        self.emitter.emit(channelId, obj);
      }).on('error', function(error) {
        console.log(error);
      });
    } else {
      let obj = {
        method: 'historic',
        data: {
          _shards: {},
          hits: {
            hits: []
          }
        }
      };
      self.emitter.emit(channelId, obj);
    }
  }

  // queryBuild
  // Builds the query by using depends object and values of sensor
  queryBuild(depends, previousSelectedSensor, size) {
    let aggs = null;
    let mustArray = [];
    let shouldArray = [];
    for(let depend in depends) {
      if(depend !== 'aggs') {
        let queryObj = null;
        if(depends[depend].defaultQuery) {
          queryObj = depends[depend].defaultQuery(previousSelectedSensor[depend]);
        } else {
          queryObj = singleQuery(depend);
        }
        if(queryObj) {
          if(depends[depend].operation === 'must') {
            mustArray.push(queryObj);
          }
          else if(depends[depend].operation === 'should') {
            shouldArray.push(queryObj);
          }
        }
      } else {
        aggs = aggsQuery(depend);
      }
    }  
    
    function singleQuery(depend) {
      let sensorInfo = helper.selectedSensor.get(depend, 'sensorInfo');
      let s_query = null
      if(previousSelectedSensor[depend]) {
        s_query = {}
        s_query[sensorInfo.queryType] = {};
        s_query[sensorInfo.queryType][sensorInfo.inputData] = previousSelectedSensor[depend];
      }
      return s_query;
    }
    
    function aggsQuery(depend) {
      let aggsObj = depends[depend];
      let order, type;
      if(aggsObj.sort=="count"){
        order = "desc";
        type = "_count";
      }
      else if(aggsObj.sort=="asc"){
        order = "asc";
        type = "_term";
      }
      else{
        order = "desc";
        type = "_term";
      }
      let orderQuery = `{ 
        "${type}" : "${order}" 
      }`;<commit_msg>Allow sensor to pass defaultQuery<commit_after>var {EventEmitter} = require('fbemitter');
var helper = require('./helper.js');

class channelManager {
  constructor() {
    this.emitter = new EventEmitter();
    this.channels = {};
    this.streamRef = {};
    this.receive = this.receive.bind(this);
  }
  setConfig(config) {
    this.config = config;
  }
  
  // Receive: This method will be executed whenever dependency value changes
  // It receives which dependency changes and which channeldId should be affected.
  receive(depend, channelId) {
    let self = this;
    let channelObj = this.channels[channelId];
    let queryObj = this.queryBuild(channelObj.depends, channelObj.previousSelectedSensor, channelObj.size);
    let validQuery = true;
    try {
      validQuery = !queryObj.body.aggs && queryObj.body.query.bool.should.length === 0 ? false : true;
    } catch(e) { }
    
    if(validQuery) {
      // apply search query and emit historic queryResult
      helper.appbaseRef.search(queryObj).on('data', function(data) {
        let obj = {
          method: 'historic',
          data: data
        };
        self.emitter.emit(channelId, obj);
      }).on('error', function(error) {
        console.log(error);
      });
      // apply searchStream query and emit streaming data
      if(this.streamRef[channelId]) {
        this.streamRef[channelId].stop();
      } 
      this.streamRef[channelId] = helper.appbaseRef.searchStream(queryObj).on('data', function(data) {
        let obj = {
          method: 'stream',
          data: data
        };
        self.emitter.emit(channelId, obj);
      }).on('error', function(error) {
        console.log(error);
      });
    } else {
      let obj = {
        method: 'historic',
        data: {
          _shards: {},
          hits: {
            hits: []
          }
        }
      };
      self.emitter.emit(channelId, obj);
    }
  }

  // queryBuild
  // Builds the query by using depends object and values of sensor
  queryBuild(depends, previousSelectedSensor, size) {
    let aggs = null;
    let mustArray = [];
    let shouldArray = [];
    for(let depend in depends) {
      if(depend !== 'aggs') {
        let queryObj = null;
        if(depends[depend].defaultQuery) {
          queryObj = depends[depend].defaultQuery(previousSelectedSensor[depend]);
        } else {
          queryObj = singleQuery(depend);
        }
        if(queryObj) {
          if(depends[depend].operation === 'must') {
            mustArray.push(queryObj);
          }
          else if(depends[depend].operation === 'should') {
            shouldArray.push(queryObj);
          }
        }
      } else {
        aggs = aggsQuery(depend);
      }
    }  
    
    // build single query or if default query present in sensor itself use that
    function singleQuery(depend) {
      let sensorInfo = helper.selectedSensor.get(depend, 'sensorInfo');
      let s_query = null
      if(sensorInfo.defaultQuery) {
        s_query = sensorInfo.defaultQuery(previousSelectedSensor[depend]);
      }
      else if(previousSelectedSensor[depend]) {
        s_query = {}
        s_query[sensorInfo.queryType] = {};
        s_query[sensorInfo.queryType][sensorInfo.inputData] = previousSelectedSensor[depend];
      }
      return s_query;
    }
    
    function aggsQuery(depend) {
      let aggsObj = depends[depend];
      let order, type;
      if(aggsObj.sort=="count"){
        order = "desc";
        type = "_count";
      }
      else if(aggsObj.sort=="asc"){
        order = "asc";
        type = "_term";
      }
      else{
        order = "desc";
        type = "_term";
      }
      let orderQuery = `{ 
        "${type}" : "${order}" 
      }`;
{'fraction_non_alphanumeric': 0.12940849763954457, 'fraction_numerical': 0.00027770063871146905, 'mean_word_length': 2.1371951219512195, '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': '3081060', 'n_tokens_mistral': 2111, 'n_tokens_neox': 2100, 'n_words': 443}