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

"use strict";
Object.defineProperty(exports, "__esModule", { value: true });
exports.transformMapSync = exports.TransformMapSync = void 0;
const js_lib_1 = require("@naturalcycles/js-lib");
const colors_1 = require("../../colors");
const pipeline_1 = require("../pipeline/pipeline");
const stream_util_1 = require("../stream.util");
class TransformMapSync extends pipeline_1.AbortableTransform {
}
exports.TransformMapSync = TransformMapSync;
/**
 * Sync (not async) version of transformMap.
 * Supposedly faster, for cases when async is not needed.
 */
function transformMapSync(mapper, opt = {}) {
    let index = -1;
    const { predicate, // defaults to "no predicate" (pass everything)
    errorMode = js_lib_1.ErrorMode.THROW_IMMEDIATELY, flattenArrayOutput = false, onError, metric = 'stream', objectMode = true, logger = console, } = opt;
    let isSettled = false;
    let errors = 0;
    const collectedErrors = []; // only used if errorMode == THROW_AGGREGATED
    return new TransformMapSync({
        objectMode,
        ...opt,
        transform(chunk, _, cb) {
            // Stop processing if isSettled
            if (isSettled)
                return cb();
            const currentIndex = ++index;
            try {
                // map and pass through
                const v = mapper(chunk, currentIndex);
                const passedResults = (flattenArrayOutput && Array.isArray(v) ? v : [v]).filter(r => {
                    if (r === js_lib_1.END) {
                        isSettled = true; // will be checked later
                        return false;
                    }
                    return r !== js_lib_1.SKIP && (!predicate || predicate(r, currentIndex));
                });
                passedResults.forEach(r => this.push(r));
                if (isSettled) {
                    logger.log(`transformMapSync END received at index ${currentIndex}`);
                    (0, stream_util_1.pipelineClose)('transformMapSync', this, this.sourceReadable, this.streamDone, logger);
                }
                cb(); // done processing
            }
            catch (err) {
                logger.error(err);
                errors++;
                logErrorStats();
                if (onError) {
                    try {
                        onError(err, chunk);
                    }
                    catch { }
                }
                if (errorMode === js_lib_1.ErrorMode.THROW_IMMEDIATELY) {
                    isSettled = 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;
        logger.log(`${metric} ${final ? 'final ' : ''}errors: ${(0, colors_1.yellow)(errors)}`);
    }
}
exports.transformMapSync = transformMapSync;