<commit_before>const async = require('async');
const schedule = require('node-schedule');

const werelogs = require('werelogs');

const config = require('../conf/Config');
const zkConfig = config.zookeeper;
const kafkaConfig = config.kafka;
const extConfigs = config.extensions;
const qpConfig = config.queuePopulator;
const httpsConfig = config.internalHttps;
const mConfig = config.metrics;
const rConfig = config.redis;
const QueuePopulator = require('../lib/queuePopulator/QueuePopulator');
const { ProbeServer, DEFAULT_LIVE_ROUTE } = require('arsenal').network.probe.ProbeServer;

const log = new werelogs.Logger('Backbeat:QueuePopulator');

werelogs.configure({ level: config.log.logLevel,
    dump: config.log.dumpLevel });

/* eslint-disable no-param-reassign */
function queueBatch(queuePopulator, taskState) {
    if (taskState.batchInProgress) {
        log.debug('skipping replication batch: previous one still in progress');
        return undefined;
    }
    const onTimeout = () => {
        // reset the flag to allow a new batch to start in case the
        // previous batch timed out
        taskState.batchInProgress = false;
    };
    log.debug('start queueing replication batch');
    taskState.batchInProgress = true;
    const maxRead = qpConfig.batchMaxRead;
    queuePopulator.processAllLogEntries({ maxRead, onTimeout }, err => {
        taskState.batchInProgress = false;
        if (err) {
            log.error('an error occurred during replication', {
                method: 'QueuePopulator::task.queueBatch',
                error: err,
            });
        }
    });
    return undefined;
}
/* eslint-enable no-param-reassign */

const queuePopulator = new QueuePopulator(
    zkConfig, kafkaConfig, qpConfig, httpsConfig, mConfig, rConfig, extConfigs);

const probeServer = new ProbeServer(qpConfig.probeServer);

async.waterfall([
    done => {
        probeServer.addHandler(
            DEFAULT_LIVE_ROUTE,
            (res, log) => queuePopulator.handleLiveness(res, log)
        );
        probeServer._cbOnListening = done;
        probeServer.start();
    },
    done => queuePopulator.open(done),
    done => {
        const taskState = {
            batchInProgress: false,
        };<commit_msg>bugfix: S3C-4447 Probe server off by default<commit_after>const async = require('async');
const schedule = require('node-schedule');

const werelogs = require('werelogs');

const config = require('../conf/Config');
const zkConfig = config.zookeeper;
const kafkaConfig = config.kafka;
const extConfigs = config.extensions;
const qpConfig = config.queuePopulator;
const httpsConfig = config.internalHttps;
const mConfig = config.metrics;
const rConfig = config.redis;
const QueuePopulator = require('../lib/queuePopulator/QueuePopulator');
const { ProbeServer, DEFAULT_LIVE_ROUTE } = require('arsenal').network.probe.ProbeServer;

const log = new werelogs.Logger('Backbeat:QueuePopulator');

werelogs.configure({ level: config.log.logLevel,
    dump: config.log.dumpLevel });

/* eslint-disable no-param-reassign */
function queueBatch(queuePopulator, taskState) {
    if (taskState.batchInProgress) {
        log.debug('skipping replication batch: previous one still in progress');
        return undefined;
    }
    const onTimeout = () => {
        // reset the flag to allow a new batch to start in case the
        // previous batch timed out
        taskState.batchInProgress = false;
    };
    log.debug('start queueing replication batch');
    taskState.batchInProgress = true;
    const maxRead = qpConfig.batchMaxRead;
    queuePopulator.processAllLogEntries({ maxRead, onTimeout }, err => {
        taskState.batchInProgress = false;
        if (err) {
            log.error('an error occurred during replication', {
                method: 'QueuePopulator::task.queueBatch',
                error: err,
            });
        }
    });
    return undefined;
}
/* eslint-enable no-param-reassign */

const queuePopulator = new QueuePopulator(
    zkConfig, kafkaConfig, qpConfig, httpsConfig, mConfig, rConfig, extConfigs);

let probeServer;
if (process.env.BACKBEAT_ENABLE_PROBE_SERVER === 'true' &&
    qpConfig.probeServer !== undefined) {
    probeServer = new ProbeServer(qpConfig.probeServer);
}

async.waterfall([
    done => {
        if (probeServer === undefined) {
            return done();
        }
        probeServer.addHandler(
            DEFAULT_LIVE_ROUTE,
            (res, log) => queuePopulator.handleLiveness(res, log)
        );
        probeServer._cbOnListening = done;
        probeServer.start();
        return undefined;
    },
    done => queuePopulator.open(done),
    done => {
        const taskState = {
            batchInProgress: false,
        };
{'fraction_non_alphanumeric': 0.11321948134092347, 'fraction_numerical': 0.0010541851149061775, 'mean_word_length': 3.1038062283737022, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 15, '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': '3585368', 'n_tokens_mistral': 1367, 'n_tokens_neox': 1349, 'n_words': 319}