<commit_before>var coalesce = require('extant')

var descendent = require('foremost')('descendent')

var Evaluator = require('prolific.evaluator')
var Chunk = require('prolific.chunk')
var Queue = require('prolific.queue')

var LEVEL = require('prolific.level')

var createUncaughtExceptionHandler = require('./uncaught')

var assert = require('assert')

function Shuttle () {
    this._state = 'created'
}

Shuttle.prototype.start = function (options) {
    if (this._state !== 'created') {
        return
    }
    options || (options = {})
    if (descendent.process.env.PROLIFIC_SUPERVISOR_PROCESS_ID != null) {
        this._state = 'started'
        this._listen(descendent, options)
    } else {
        this._state = 'closed'
    }
}

Shuttle.prototype._listen = function (descendent, options) {
    var now = coalesce(options.Date, Date).now()
    var monitorProcessId = +descendent.process.env.PROLIFIC_SUPERVISOR_PROCESS_ID
    var headerId = 'H/' + descendent.process.pid + '/' +  now
    var streamId = 'S/' + descendent.process.pid + '/' +  now

    descendent.increment()

    var path = descendent.path.splice(descendent.path.indexOf(monitorProcessId))

    if (options.uncaughtException != null) {
        var uncaughtException = this.uncaughtException(options.uncaughtException)
        descendent.process.on('uncaughtException', uncaughtException)
    }

    var queue = new Queue(streamId, descendent.process.stderr)
    this._queue = queue

    // All filtering will be performed by the monitor initially. Until
    // we get a configuration we send everything.
    var sink = require('prolific.resolver').sink
    sink.queue = queue
    sink.json = function (level, qualifier, label, body) {
        this.queue.push({
            when: this.Date.now(),
            level: level,
            qualifier: qualifier,
            label: label,
            body: body,
            system: this.properties
        })
    }

    this._handlers = { pipe: null, accept: null }

    descendent.once('prolific:pipe', this._handlers.pipe = function (message, handle) {
        queue.setPipe(handle)
    })
    descendent.on('prolific:accept', this._handlers.accept = function (message) {
        assert(message.body.triage)
        var triage = Evaluator.create(message.body.triage)
        assert(triage)
        sink.json = function (level, qualifier, label, body) {
            var header = {
                when: this.Date.now(),
                level: level,
                qualifier: qualifier,
                label: label,
                qualified: qualifier + '#' + label
            }
            if (triage(LEVEL[level], header, body, this.properties)) {
                for (var key in this.properties) {
                    header[key] = this.properties[key]
                }
                for (var key in body) {
                    header[key] = body[key]
                }
                this.queue.push(header)
            }
        }
        sink.queue.push([{ version: message.body.version }])
    })

    var chunks = []
    chunks.push(new Chunk(headerId, 0, Buffer.from(''), 1))
    var buffer = Buffer.from(JSON.stringify({<commit_msg>Add `require` to invocation of triage.<commit_after>var coalesce = require('extant')

var descendent = require('foremost')('descendent')

var Evaluator = require('prolific.evaluator')
var Chunk = require('prolific.chunk')
var Queue = require('prolific.queue')

var LEVEL = require('prolific.level')

var createUncaughtExceptionHandler = require('./uncaught')

var assert = require('assert')

function Shuttle () {
    this._state = 'created'
}

Shuttle.prototype.start = function (options) {
    if (this._state !== 'created') {
        return
    }
    options || (options = {})
    if (descendent.process.env.PROLIFIC_SUPERVISOR_PROCESS_ID != null) {
        this._state = 'started'
        this._listen(descendent, options)
    } else {
        this._state = 'closed'
    }
}

Shuttle.prototype._listen = function (descendent, options) {
    var now = coalesce(options.Date, Date).now()
    var monitorProcessId = +descendent.process.env.PROLIFIC_SUPERVISOR_PROCESS_ID
    var headerId = 'H/' + descendent.process.pid + '/' +  now
    var streamId = 'S/' + descendent.process.pid + '/' +  now

    descendent.increment()

    var path = descendent.path.splice(descendent.path.indexOf(monitorProcessId))

    if (options.uncaughtException != null) {
        var uncaughtException = this.uncaughtException(options.uncaughtException)
        descendent.process.on('uncaughtException', uncaughtException)
    }

    var queue = new Queue(streamId, descendent.process.stderr)
    this._queue = queue

    // All filtering will be performed by the monitor initially. Until
    // we get a configuration we send everything.
    var sink = require('prolific.resolver').sink
    sink.queue = queue
    sink.json = function (level, qualifier, label, body) {
        this.queue.push({
            when: this.Date.now(),
            level: level,
            qualifier: qualifier,
            label: label,
            body: body,
            system: this.properties
        })
    }

    this._handlers = { pipe: null, accept: null }

    descendent.once('prolific:pipe', this._handlers.pipe = function (message, handle) {
        queue.setPipe(handle)
    })
    descendent.on('prolific:accept', this._handlers.accept = function (message) {
        assert(message.body.triage)
        var triage = Evaluator.create(message.body.triage, require)
        assert(triage)
        sink.json = function (level, qualifier, label, body) {
            var header = {
                when: this.Date.now(),
                level: level,
                qualifier: qualifier,
                label: label,
                qualified: qualifier + '#' + label
            }
            if (triage(LEVEL[level], header, body, this.properties)) {
                for (var key in this.properties) {
                    header[key] = this.properties[key]
                }
                for (var key in body) {
                    header[key] = body[key]
                }
                this.queue.push(header)
            }
        }
        sink.queue.push([{ version: message.body.version }])
    })

    var chunks = []
    chunks.push(new Chunk(headerId, 0, Buffer.from(''), 1))
    var buffer = Buffer.from(JSON.stringify({
{'fraction_non_alphanumeric': 0.12190535882168599, 'fraction_numerical': 0.0006267627702914447, 'mean_word_length': 2.3210197710718004, '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': '1178847', 'n_tokens_mistral': 1898, 'n_tokens_neox': 1848, 'n_words': 405}