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

"use strict";
Object.defineProperty(exports, "__esModule", { value: true });
exports.transformMapSync = void 0;
const js_lib_1 = require("@naturalcycles/js-lib");
const stream_1 = require("stream");
const colors_1 = require("../../colors");
const transformMap_1 = require("./transformMap");
/**
 * Sync (not async) version of transformMap.
 * Supposedly faster, for cases when async is not needed.
 */
function transformMapSync(mapper, opt = {}) {
    let index = -1;
    const { predicate = transformMap_1.notNullishPredicate, errorMode = js_lib_1.ErrorMode.THROW_IMMEDIATELY, flattenArrayOutput = false, onError, metric = 'stream', objectMode = true, } = opt;
    let isRejected = false;
    let errors = 0;
    const collectedErrors = []; // only used if errorMode == THROW_AGGREGATED
    return new stream_1.Transform({
        objectMode,
        ...opt,
        transform(chunk, _encoding, cb) {
            // Stop processing if THROW_IMMEDIATELY mode is used
            if (isRejected && errorMode === js_lib_1.ErrorMode.THROW_IMMEDIATELY) {
                return cb();
            }
            try {
                if (!predicate(chunk, ++index)) {
                    cb(); // signal that we've finished processing, but emit no output here
                    return;
                }
                // map and pass through
                const v = mapper(chunk, index);
                if (flattenArrayOutput && Array.isArray(v)) {
                    // Pass each item individually
                    v.forEach(item => this.push(item));
                }
                else {
                    cb(null, v);
                }
            }
            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);
                }
                cb();
            }
        },
        final(cb) {
            // console.log('transformMap final')
            logErrorStats(true);
            if (collectedErrors.length) {
                // emit Aggregated error
                cb(new js_lib_1.AggregatedError(collectedErrors));
            }
            else {
                // emit no error
                cb();
            }
        },
    });
    function logErrorStats(final = false) {
        if (!errors)
            return;
        console.log(`${metric} ${final ? 'final ' : ''}errors: ${(0, colors_1.yellow)(errors)}`);
    }
}
exports.transformMapSync = transformMapSync;