<commit_before>'use strict';

var util = require('util');
var EventEmitter = require('events').EventEmitter;
var debug = require('./util/debug')('batch');
var queue = require('queue-async');
var HostScheduler = require('./scheduler/host-scheduler');

var EMPTY_QUEUE = true;

function Batch(name, jobSubscriber, jobQueue, jobRunner, jobService, jobPublisher, redisPool, logger) {
    EventEmitter.call(this);
    this.name = name || 'batch';
    this.jobSubscriber = jobSubscriber;
    this.jobQueue = jobQueue;
    this.jobRunner = jobRunner;
    this.jobService = jobService;
    this.jobPublisher = jobPublisher;
    this.logger = logger;
    this.hostScheduler = new HostScheduler(this.name, { run: this.processJob.bind(this) }, redisPool);

    // map: user => jobId. Will be used for draining jobs.
    this.workInProgressJobs = {};
}
util.inherits(Batch, EventEmitter);

module.exports = Batch;

Batch.prototype.start = function () {
    var self = this;

    this.jobSubscriber.subscribe(
        function onJobHandler(user, host) {
            debug('[%s] onJobHandler(%s, %s)', self.name, user, host);
            self.hostScheduler.add(host, user, function(err) {
                if (err) {
                    return debug(
                        'Could not schedule host=%s user=%s from %s. Reason: %s',
                        host, self.name, user, err.message
                    );
                }
            });
        },
        function onJobSubscriberReady(err) {
            if (err) {
                return self.emit('error', err);
            }

            self.emit('ready');
        }
    );
};

Batch.prototype.processJob = function (user, callback) {
    var self = this;
    self.jobQueue.dequeue(user, function (err, jobId) {
        if (err) {
            return callback(new Error('Could not get job from "' + user + '". Reason: ' + err.message), !EMPTY_QUEUE);
        }

        if (!jobId) {
            debug('Queue empty user=%s', user);
            return callback(null, EMPTY_QUEUE);
        }

        self.setWorkInProgressJob(user, jobId, function (err) {
            if (err) {
                return callback(new Error('Could not add job to work-in-progress list. Reason: ' + err.message));
            }

            self.jobRunner.run(jobId, function (err, job) {
                self.clearWorkInProgressJob(user, jobId, function (clearError) {
                    if (clearError) {
                        return callback(
                            new Error('Could not clear job from work-in-progress list. Reason: ' + err.message)
                        );
                    }

                    if (err) {
                        debug(err);
                        if (err.name === 'JobNotRunnable') {
                            return callback(null, !EMPTY_QUEUE);
                        }
                        return callback(err, !EMPTY_QUEUE);
                    }

                    debug(
                        '[%s] Job=%s status=%s user=%s (failed_reason=%s)',
                        self.name, jobId, job.data.status, user, job.failed_reason
                    );

                    self.logger.log(job);

                    return callback(null, !EMPTY_QUEUE);
                });
            });
        });
    });
};

Batch.prototype.drain = function (callback) {
    var self = this;
    var workingUsers = this.getWorkInProgressUsers();
    var batchQueues = queue(workingUsers.length);

    workingUsers.forEach(function (user) {
        batchQueues.defer(self._drainJob.bind(self), user);
    });

    batchQueues.awaitAll(function (err) {
        if (err) {
            debug('Something went wrong draining', err);
        } else {
            debug('Drain complete');
        }

        callback();
    });
};

Batch.prototype._drainJob = function (user, callback) {
    var self = this;<commit_msg>Hide add/clear job from work in progress list<commit_after>'use strict';

var util = require('util');
var EventEmitter = require('events').EventEmitter;
var debug = require('./util/debug')('batch');
var queue = require('queue-async');
var HostScheduler = require('./scheduler/host-scheduler');

var EMPTY_QUEUE = true;

function Batch(name, jobSubscriber, jobQueue, jobRunner, jobService, jobPublisher, redisPool, logger) {
    EventEmitter.call(this);
    this.name = name || 'batch';
    this.jobSubscriber = jobSubscriber;
    this.jobQueue = jobQueue;
    this.jobRunner = jobRunner;
    this.jobService = jobService;
    this.jobPublisher = jobPublisher;
    this.logger = logger;
    this.hostScheduler = new HostScheduler(this.name, { run: this.processJob.bind(this) }, redisPool);

    // map: user => jobId. Will be used for draining jobs.
    this.workInProgressJobs = {};
}
util.inherits(Batch, EventEmitter);

module.exports = Batch;

Batch.prototype.start = function () {
    var self = this;

    this.jobSubscriber.subscribe(
        function onJobHandler(user, host) {
            debug('[%s] onJobHandler(%s, %s)', self.name, user, host);
            self.hostScheduler.add(host, user, function(err) {
                if (err) {
                    return debug(
                        'Could not schedule host=%s user=%s from %s. Reason: %s',
                        host, self.name, user, err.message
                    );
                }
            });
        },
        function onJobSubscriberReady(err) {
            if (err) {
                return self.emit('error', err);
            }

            self.emit('ready');
        }
    );
};

Batch.prototype.processJob = function (user, callback) {
    var self = this;
    self.jobQueue.dequeue(user, function (err, jobId) {
        if (err) {
            return callback(new Error('Could not get job from "' + user + '". Reason: ' + err.message), !EMPTY_QUEUE);
        }

        if (!jobId) {
            debug('Queue empty user=%s', user);
            return callback(null, EMPTY_QUEUE);
        }

        self._processWorkInProgressJob(user, jobId, function (err, job) {
            if (err) {
                debug(err);
                if (err.name === 'JobNotRunnable') {
                    return callback(null, !EMPTY_QUEUE);
                }
                return callback(err, !EMPTY_QUEUE);
            }

            debug(
                '[%s] Job=%s status=%s user=%s (failed_reason=%s)',
                self.name, jobId, job.data.status, user, job.failed_reason
            );

            self.logger.log(job);

            return callback(null, !EMPTY_QUEUE);
        });
    });
};

Batch.prototype._processWorkInProgressJob = function (user, jobId, callback) {
    var self = this;

    self.setWorkInProgressJob(user, jobId, function (errSet) {
        if (errSet) {
            return callback(new Error('Could not add job to work-in-progress list. Reason: ' + errSet.message));
        }

        self.jobRunner.run(jobId, function (err, job) {
            self.clearWorkInProgressJob(user, jobId, function (errClear) {
                if (errClear) {
                    return callback(
                        new Error('Could not clear job from work-in-progress list. Reason: ' + errClear.message)
                    );
                }

                return callback(err, job);
            });
        });
    });
}

Batch.prototype.drain = function (callback) {
    var self = this;
    var workingUsers = this.getWorkInProgressUsers();
    var batchQueues = queue(workingUsers.length);

    workingUsers.forEach(function (user) {
        batchQueues.defer(self._drainJob.bind(self), user);
    });

    batchQueues.awaitAll(function (err) {
        if (err) {
            debug('Something went wrong draining', err);
        } else {
            debug('Drain complete');
        }

        callback();
    });
};

Batch.prototype._drainJob = function (user, callback) {
    var self = this;
{'fraction_non_alphanumeric': 0.12577423840222476, 'fraction_numerical': 0.0, 'mean_word_length': 1.8625180897250362, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 5, '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': '566547', 'n_tokens_mistral': 2274, 'n_tokens_neox': 2193, 'n_words': 508}