<commit_before>/*
    ___ usage ___ en_US ___
    usage: node monitor.bin.js

        -c, --configuration <string>    json configuration path
        -s, --supervisor <string>       pid of supervisor
            --help                      display this message

    ___ . ___
*/
require('arguable')(module, require('cadence')(function (async, program) {
    program.required('configuration', 'supervisor')

    // Node.js API.
    var assert = require('assert')

    // Route messages through a process hierarchy using Node.js IPC.
    var Descendent = require('descendent')

    // Construct a prolific pipeline from a configuration.
    var Pipeline = require('prolific.pipeline')

    // Controlled demolition of objects.
    var Destructible = require('destructible')

    // Consolidate chunks from async and sync streams.
    var Asynchronous = require('prolific.consolidator/asynchronous')

    var Signal = require('signal')
    var cadence = require('cadence')

    var destructible = new Destructible(15000, 'prolific.monitor')
    program.on('shutdown', destructible.destroy.bind(destructible))

    program.on('SIGTERM', function () { console.log('MONITOR GOT SIGTERM') })
    program.on('SIGINT', function () { console.log('MONITOR GOT SIGINT') })

    destructible.completed.wait(async())

    // TODO Remove this debugging stuff and the debugging stuff above.
    destructible.destruct.wait(function () {
        var stackTraceLimit = Error.stackTraceLimit
        Error.stackTraceLimit = Infinity
        console.log(new Error().stack)
        Error.stackTraceLimit = stackTraceLimit
    })

    var descendent = new Descendent(program)
    destructible.destruct.wait(descendent, 'decrement')

    var Processor = require('./processor')

    var Turnstile = require('turnstile')
    Turnstile.Queue = require('turnstile/queue')

    var Watcher = require('./watcher')

    async(function () {
        destructible.completed.wait(async())
    }, function () {
        return 0
    })

    var reader = require('./stdin')(destructible.destroy.bind(destructible))

    destructible.monitor('main', cadence(function (async) {
        async(function () {
            setImmediate(async()) // allows test to get handle
        }, function () {
            var reloaded = descendent.up.bind(descendent, +program.ultimate.supervisor, 'prolific:accept')
            destructible.monitor('processor', Processor, program.ultimate.configuration, reloaded, async())
        }, function (processor) {
            // Drain all chunks immediately into a turnstile.
            var turnstile = new Turnstile
            var queue = new Turnstile.Queue(processor, 'process', turnstile)
            turnstile.listen(destructible.monitor('turnstile'))
            destructible.destruct.wait(turnstile, 'close')

            // Create our asynchronous listener that reads directly from the
            // monitored process.
            var asynchronous = new Asynchronous(queue)

            // Copy any final messages written to standard error into the
            // asynchronous listener so it can eliminate any duplicates that
            // where already written to our primary asynchronous pipe.
            reader(program.stdin, asynchronous, destructible.monitor('stdin'))
            program.stdin.resume()

            // Listen to our asynchronous pipe.
            var socket = new program.attributes.net.Socket({ fd: 3 })
            destructible.destruct.wait(socket, 'destroy')
            asynchronous.listen(socket, destructible.monitor('asynchronous'))

            // Let the supervisor know that we're ready. It will send our
            // asynchronous pipe down to the monitored process.
            descendent.up(+program.ultimate.supervisor, 'prolific:pipe', true)

            // Reload immediately because the reload will cause the first
            // round of filtering to get pushed into the monitored process.
            var watcher = new Watcher(program.ultimate.configuration, processor)
            destructible.destruct.wait(watcher, 'destroy')
            watcher.monitor(destructible.monitor('watch'))

            // Ready for testing.
            program.ready.unlatch()
        })
    }), program.ready.unlatch.bind(program.ready))<commit_msg>Tidy `monitor.bin.js`.<commit_after>/*
    ___ usage ___ en_US ___
    usage: node monitor.bin.js

        -c, --configuration <string>    json configuration path
        -s, --supervisor <string>       pid of supervisor
            --help                      display this message

    ___ . ___
*/
require('arguable')(module, require('cadence')(function (async, program) {
    program.required('configuration', 'supervisor')

    // Node.js API.
    var assert = require('assert')

    // Route messages through a process hierarchy using Node.js IPC.
    var Descendent = require('descendent')

    // Construct a prolific pipeline from a configuration.
    var Pipeline = require('prolific.pipeline')

    // Controlled demolition of objects.
    var Destructible = require('destructible')

    // Consolidate chunks from async and sync streams.
    var Asynchronous = require('prolific.consolidator/asynchronous')

    var Signal = require('signal')
    var cadence = require('cadence')

    var destructible = new Destructible(15000, 'prolific.monitor')
    program.on('shutdown', destructible.destroy.bind(destructible))

    var descendent = new Descendent(program)
    destructible.destruct.wait(descendent, 'decrement')

    var Processor = require('./processor')

    var Turnstile = require('turnstile')
    Turnstile.Queue = require('turnstile/queue')

    var Watcher = require('./watcher')

    async(function () {
        destructible.completed.wait(async())
    }, function () {
        return 0
    })

    var reader = require('./stdin')(destructible.destroy.bind(destructible))

    destructible.monitor('main', cadence(function (async) {
        async(function () {
            setImmediate(async()) // allows test to get handle
        }, function () {
            var reloaded = descendent.up.bind(descendent, +program.ultimate.supervisor, 'prolific:accept')
            destructible.monitor('processor', Processor, program.ultimate.configuration, reloaded, async())
        }, function (processor) {
            // Drain all chunks immediately into a turnstile.
            var turnstile = new Turnstile
            var queue = new Turnstile.Queue(processor, 'process', turnstile)
            turnstile.listen(destructible.monitor('turnstile'))
            destructible.destruct.wait(turnstile, 'close')

            // Create our asynchronous listener that reads directly from the
            // monitored process.
            var asynchronous = new Asynchronous(queue)

            // Copy any final messages written to standard error into the
            // asynchronous listener so it can eliminate any duplicates that
            // where already written to our primary asynchronous pipe.
            reader(program.stdin, asynchronous, destructible.monitor('stdin'))
            program.stdin.resume()

            // Listen to our asynchronous pipe.
            var socket = new program.attributes.net.Socket({ fd: 3 })
            destructible.destruct.wait(socket, 'destroy')
            asynchronous.listen(socket, destructible.monitor('asynchronous'))

            // Let the supervisor know that we're ready. It will send our
            // asynchronous pipe down to the monitored process.
            descendent.up(+program.ultimate.supervisor, 'prolific:pipe', true)

            // Reload immediately because the reload will cause the first
            // round of filtering to get pushed into the monitored process.
            var watcher = new Watcher(program.ultimate.configuration, processor)
            destructible.destruct.wait(watcher, 'destroy')
            watcher.monitor(destructible.monitor('watch'))
        })
    }), program.ready.unlatch.bind(program.ready))
{'fraction_non_alphanumeric': 0.10421618916551983, 'fraction_numerical': 0.00175153259101714, 'mean_word_length': 2.61229100768188, 'pattern_counts': {'":': 0, '<': 7, '<?xml version=': 0, '>': 7, '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': '2251303', 'n_tokens_mistral': 2131, 'n_tokens_neox': 1983, 'n_words': 561}