@sounisi5011/stream-transform-from/dist/index.js

"use strict";
var _a;
Object.defineProperty(exports, "__esModule", { value: true });
exports.transformFrom = exports.TransformFromAsyncIterable = void 0;
const process_1 = require("process");
const stream_1 = require("stream");
/**
 * Prior to Node.js v14, there is a bug where the `finish` event is fired before the callback passed to the `transform._flush()` method is called.
 * This bug has been fixed in Node.js v15.
 * @see https://github.com/nodejs/node/issues/34274
 * @see https://github.com/nodejs/node/pull/34314
 */
const HAS_FLUSH_BUG = !(Number((_a = process_1.version.match(/(?<=^v)\d+/)) === null || _a === void 0 ? void 0 : _a[0]) >= 15);
const DISALLOW_OPTION_NAMES = [
    'construct',
    'read',
    'write',
    'writev',
    'final',
    'destroy',
    'transform',
    'flush',
];
function removeProp(obj, props) {
    if (obj === null || obj === undefined)
        return obj;
    return Object.fromEntries(Object.entries(obj)
        .filter(([p]) => !props.includes(p)));
}
class TransformFromAsyncIterable extends stream_1.Transform {
    constructor(transformFn, opts) {
        super(removeProp(opts, DISALLOW_OPTION_NAMES));
        Object.defineProperty(this, "transformCallback", {
            enumerable: true,
            configurable: true,
            writable: true,
            value: void 0
        });
        Object.defineProperty(this, "isFinished", {
            enumerable: true,
            configurable: true,
            writable: true,
            value: false
        });
        Object.defineProperty(this, "receiveData", {
            enumerable: true,
            configurable: true,
            writable: true,
            value: void 0
        });
        Object.defineProperty(this, "receivedDataList", {
            enumerable: true,
            configurable: true,
            writable: true,
            value: []
        });
        const source = this.createSource();
        const result = transformFn(source);
        (async () => {
            for await (const chunk of result) {
                this.push(chunk);
            }
        })()
            .then(() => this.finish())
            .catch(error => this.finish(error));
    }
    _transform(chunk, encoding, callback) {
        if (this.isFinished) {
            callback();
        }
        else {
            this.transformCallback = callback;
            this.emitToSource({ chunk, encoding });
        }
    }
    _flush(callback) {
        if (this.isFinished) {
            callback();
        }
        else {
            this.transformCallback = HAS_FLUSH_BUG
                ? this.emitFinishEventAfterCallback(callback)
                : callback;
            this.emitToSource({ done: true });
        }
    }
    finish(error) {
        this.isFinished = true;
        if (error) {
            if (!this.callTransformCallback(error)) {
                this.destroy(error);
            }
        }
        else {
            this.push(null);
            this.callTransformCallback();
        }
    }
    callTransformCallback(...args) {
        const { transformCallback } = this;
        if (transformCallback) {
            this.transformCallback = undefined;
            transformCallback(...args);
            return true;
        }
        return false;
    }
    async *createSource() {
        var _a;
        while (true) {
            const data = ((_a = this.receivedDataList.shift()) !== null && _a !== void 0 ? _a : await new Promise(resolve => {
                this.receiveData = resolve;
                this.callTransformCallback();
            }));
            if (data.done)
                break;
            const { done: _, ...chunkData } = data;
            yield chunkData;
        }
    }
    emitToSource(data) {
        const { receiveData } = this;
        if (receiveData) {
            this.receiveData = undefined;
            receiveData(data);
        }
        else {
            this.receivedDataList.push(data);
        }
    }
    emitFinishEventAfterCallback(flushCallback) {
        const finishEventName = 'finish';
        const finishEventList = [];
        this.emit = (event, ...args) => {
            if (finishEventName === event) {
                finishEventList.push(args);
                return false;
            }
            return super.emit(event, ...args);
        };
        return (...args) => {
            flushCallback(...args);
            // @ts-expect-error TS2790: The operand of a 'delete' operator must be optional.
            delete this.emit;
            const [error] = args;
            if (!error) {
                for (const args of finishEventList) {
                    this.emit(finishEventName, ...args);
                }
            }
        };
    }
}
exports.TransformFromAsyncIterable = TransformFromAsyncIterable;
function transformFrom(transformFn, options) {
    return new TransformFromAsyncIterable(transformFn, options);
}
exports.transformFrom = transformFrom;
//# sourceMappingURL=index.js.map