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

"use strict";
Object.defineProperty(exports, "__esModule", { value: true });
exports.transformMap = exports.notNullishPredicate = void 0;
const js_lib_1 = require("@naturalcycles/js-lib");
const through2Concurrent = require("through2-concurrent");
const colors_1 = require("../../colors");
function notNullishPredicate(item) {
    return item !== undefined && item !== null;
}
exports.notNullishPredicate = notNullishPredicate;
/**
 * Like pMap, but for streams.
 * Inspired by `through2`.
 * Main feature is concurrency control (implemented via `through2-concurrent`) and convenient options.
 * Using this allows native stream .pipe() to work and use backpressure.
 *
 * Only works in objectMode (due to through2Concurrent).
 *
 * Concurrency defaults to 16.
 *
 * If an Array is returned by `mapper` - it will be flattened and multiple results will be emitted from it. Tested by Array.isArray().
 */
function transformMap(mapper, opt = {}) {
    const { concurrency = 16, predicate = notNullishPredicate, errorMode = js_lib_1.ErrorMode.THROW_IMMEDIATELY, flattenArrayOutput, onError, beforeFinal, metric = 'stream', } = opt;
    let index = -1;
    let isRejected = false;
    let errors = 0;
    const collectedErrors = []; // only used if errorMode == THROW_AGGREGATED
    return through2Concurrent.obj({
        maxConcurrency: concurrency,
        // autoDestroy: true,
        async final(cb) {
            // console.log('transformMap final')
            logErrorStats(true);
            await beforeFinal?.(); // call beforeFinal if defined
            if (collectedErrors.length) {
                // emit Aggregated error
                cb(new js_lib_1.AggregatedError(collectedErrors));
            }
            else {
                // emit no error
                cb();
            }
        },
    }, async function transformMapFn(chunk, _encoding, cb) {
        index++;
        // console.log({chunk, _encoding})
        // Stop processing if THROW_IMMEDIATELY mode is used
        if (isRejected && errorMode === js_lib_1.ErrorMode.THROW_IMMEDIATELY)
            return cb();
        try {
            const currentIndex = index; // because we need to pass it to 2 functions - mapper and predicate. Refers to INPUT index (since it may return multiple outputs)
            const res = await mapper(chunk, currentIndex);
            const passedResults = await (0, js_lib_1.pFilter)(flattenArrayOutput && Array.isArray(res) ? res : [res], async (r) => await predicate(r, currentIndex));
            if (passedResults.length === 0) {
                cb(); // 0 results
            }
            else {
                passedResults.forEach(r => {
                    this.push(r);
                    // cb(null, r)
                });
                cb(); // done processing
            }
        }
        catch (err) {
            console.error(err);
            errors++;
            logErrorStats();
            if (onError) {
                try {
                    onError(err, chunk);
                }
                catch { }
            }
            if (errorMode === js_lib_1.ErrorMode.THROW_IMMEDIATELY) {
                isRejected = true;
                // Emit error immediately
                return cb(err);
            }
            if (errorMode === js_lib_1.ErrorMode.THROW_AGGREGATED) {
                collectedErrors.push(err);
            }
            // Tell input stream that we're done processing, but emit nothing to output - not error nor result
            cb();
        }
    });
    function logErrorStats(final = false) {
        if (!errors)
            return;
        console.log(`${metric} ${final ? 'final ' : ''}errors: ${(0, colors_1.yellow)(errors)}`);
    }
}
exports.transformMap = transformMap;