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

"use strict";
Object.defineProperty(exports, "__esModule", { value: true });
exports.transformLogProgress = void 0;
const stream_1 = require("stream");
const util_1 = require("util");
const js_lib_1 = require("@naturalcycles/js-lib");
const time_lib_1 = require("@naturalcycles/time-lib");
const colors_1 = require("../../colors");
const inspectOpt = {
    colors: true,
    breakLength: 300,
};
/**
 * Pass-through transform that optionally logs progress.
 */
function transformLogProgress(opt = {}) {
    const { metric = 'progress', heapTotal: logHeapTotal = false, heapUsed: logHeapUsed = true, rss: logRss = true, peakRSS: logPeakRSS = false, 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;
    let peakRSS = 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));
        if (mem.rss > peakRSS)
            peakRSS = mem.rss;
        console.log((0, util_1.inspect)({
            [final ? `${metric}_final` : metric]: progress,
            ...(extra ? extra(chunk, progress) : {}),
            ...(logHeapUsed ? { heapUsed: (0, js_lib_1._mb)(mem.heapUsed) } : {}),
            ...(logHeapTotal ? { heapTotal: (0, js_lib_1._mb)(mem.heapTotal) } : {}),
            ...(logRss ? { rss: (0, js_lib_1._mb)(mem.rss) } : {}),
            ...(logPeakRSS ? { peakRSS: (0, js_lib_1._mb)(peakRSS) } : {}),
            ...(opt.rssMinusHeap ? { rssMinusHeap: (0, js_lib_1._mb)(mem.rss - mem.heapTotal) } : {}),
            ...(opt.external ? { external: (0, js_lib_1._mb)(mem.external) } : {}),
            ...(opt.arrayBuffers ? { arrayBuffers: (0, 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(`${(0, colors_1.dimGrey)((0, time_lib_1.dayjs)().toPretty())} ${(0, colors_1.white)(metric)} took ${(0, colors_1.yellow)((0, js_lib_1._since)(started))} so far to process ${(0, colors_1.yellow)(progress)} rows, ~${(0, colors_1.yellow)(perHour)}/hour`);
        }
        else if (final) {
            console.log(`${(0, colors_1.boldWhite)(metric)} took ${(0, colors_1.yellow)((0, js_lib_1._since)(started))} to process ${(0, colors_1.yellow)(progress)} rows with total RPS of ${(0, colors_1.yellow)(rpsTotal)}`);
        }
    }
}
exports.transformLogProgress = transformLogProgress;