@naturalcycles/nodejs-lib/dist/stream/transform/transformLogProgress.js

"use strict";
Object.defineProperty(exports, "__esModule", { value: true });
exports.transformLogProgress = void 0;
const js_lib_1 = require("@naturalcycles/js-lib");
const time_lib_1 = require("@naturalcycles/time-lib");
const stream_1 = require("stream");
const util_1 = require("util");
const colors_1 = require("../../colors");
const inspectOpt = {
    colors: true,
    breakLength: 200,
};
/**
 * Pass-through transform that optionally logs progress.
 */
function transformLogProgress(opt = {}) {
    const { metric = 'progress', heapTotal: logHeapTotal = false, heapUsed: logHeapUsed = true, rss: logRss = true, logRPS = true, logEvery = 1000, extra, } = opt;
    const logProgress = opt.logProgress !== false && logEvery !== 0; // true by default
    const logEvery10 = logEvery * 10;
    const started = Date.now();
    let lastSecondStarted = Date.now();
    const sma = new js_lib_1.SimpleMovingAverage(10); // over last 10 seconds
    let processedLastSecond = 0;
    let progress = 0;
    logStats(); // initial
    return new stream_1.Transform({
        objectMode: true,
        ...opt,
        transform(chunk, _encoding, cb) {
            progress++;
            processedLastSecond++;
            if (logProgress && progress % logEvery === 0) {
                logStats(chunk, false, progress % logEvery10 === 0);
            }
            cb(null, chunk); // pass-through
        },
        final(cb) {
            logStats(undefined, true);
            cb();
        },
    });
    function logStats(chunk, final = false, tenx = false) {
        if (!logProgress)
            return;
        const mem = process.memoryUsage();
        const now = Date.now();
        const lastRPS = processedLastSecond / ((now - lastSecondStarted) / 1000) || 0;
        const rpsTotal = Math.round(progress / ((now - started) / 1000)) || 0;
        lastSecondStarted = now;
        processedLastSecond = 0;
        const rps10 = Math.round(sma.push(lastRPS));
        console.log(util_1.inspect({
            [final ? `${metric}_final` : metric]: progress,
            ...(extra ? extra(chunk, progress) : {}),
            ...(logHeapUsed ? { heapUsed: js_lib_1._mb(mem.heapUsed) } : {}),
            ...(logHeapTotal ? { heapTotal: js_lib_1._mb(mem.heapTotal) } : {}),
            ...(logRss ? { rss: js_lib_1._mb(mem.rss) } : {}),
            ...(opt.rssMinusHeap ? { rssMinusHeap: js_lib_1._mb(mem.rss - mem.heapTotal) } : {}),
            ...(opt.external ? { external: js_lib_1._mb(mem.external) } : {}),
            ...(opt.arrayBuffers ? { arrayBuffers: js_lib_1._mb(mem.arrayBuffers || 0) } : {}),
            ...(logRPS
                ? {
                    rps10,
                    rpsTotal,
                }
                : {}),
        }, inspectOpt));
        if (tenx) {
            let perHour = Math.round((progress * 1000 * 60 * 60) / (now - started)) || 0;
            if (perHour > 900) {
                perHour = Math.round(perHour / 1000) + 'K';
            }
            console.log(`${colors_1.dimGrey(time_lib_1.dayjs().toPretty())} ${colors_1.white(metric)} took ${colors_1.yellow(js_lib_1._since(started))} so far to process ${colors_1.yellow(progress)} rows, ~${colors_1.yellow(perHour)}/hour`);
        }
        else if (final) {
            console.log(`${colors_1.boldWhite(metric)} took ${colors_1.yellow(js_lib_1._since(started))} to process ${colors_1.yellow(progress)} rows with total RPS of ${colors_1.yellow(rpsTotal)}`);
        }
    }
}
exports.transformLogProgress = transformLogProgress;