如何过早地完成 JS 转换中的管道答案

【问题标题】：How to prematurely finish a pipeline within a JS transform如何过早地完成 JS 转换中的管道
【发布时间】：2021-05-07 09:07:38
【问题描述】：

问题：我正在处理大文件 (>10GB)。有时我想处理完整的文件，有时我只想采样几行。

处理设置是一个管道：

        pipeline(
            inStream,
            split2(),
            sc,
            err => {
                 ...
            }
        );

sc 是一种本质上计算文件中一些标志的转换。

代码在处理完整文件时工作正常，但如果我想在inStream 完成之前退出转换，则永远不会在...中产生输出。

        _transform(chunk,encoding,done) {
        let strChunk = decoder.write(chunk);
        if(strChunk === '\u0003') {
            this.push('\u0003');
            process.exit(0);
        }
        if(strChunk.startsWith("@")) {
            done();
        } else {
            if(this.sampleMax === 0) {
                this.push('\u0003');
                //process.exit(0);
            } else 
                if(this.sampleMax > 0)
                    this.sampleMax--;
            let dta = strChunk.split("\t");
            let flag = dta[1].trim();
            this.flagCount[flag]++;
            done();
        }

如果我使用//process.exit(0)，则无法到达sc 之后的管道中的代码。如果我只使用this.push('\u0003');，则会处理完整的 inStream。

问题是如何正确终止转换并继续下游管道而不完全读取inStream。

【问题讨论】：

我现在找到了两个选项。 1) 达到限制时抛出错误并在管道中检查该错误。 2) 在达到限制时调用 this.destroy() 并检查管道的 err 部分中的相关错误（错误 [ERR_STREAM_PREMATURE_CLOSE]: Premature close）。这两个选项都有效，但我希望有更好的解决方案。

标签： javascript stream transform pipeline

【解决方案1】：

一种解决方案是通过销毁流来引发错误或隐式创建错误。这两个选项都显示在下面的代码中。

_transform(chunk,encoding,done) {
    let strChunk = decoder.write(chunk);
    if(strChunk === '\u0003') {
        this.push('\u0003');
        process.exit(0);
    }
    if(strChunk.startsWith("@")) {
        done();
    } else {
        if(this.sampleMax === 0) {
            //Either of the following lines can solve the problem.
            throw("PLANNED_PREMATURE_TERMINATION");
            this.destroy(); //=> Error [ERR_STREAM_PREMATURE_CLOSE]: Premature close
        } else 
            if(this.sampleMax > 0)
                this.sampleMax--;
        let dta = strChunk.split("\t");
        let flag = dta[1].trim();
        this.flagCount[flag]++;
        done();
    }

下一步是在管道中对上述代码中产生的错误做出反应。

    pipeline(
        inStream,
        split2(),
        sc,
        err => {
             if(err && (err === "PLANNED_PREMATURE_TERMINATION") || 
               (err === "Error [ERR_STREAM_PREMATURE_CLOSE]: Premature close") {
                 //do whatever should happen in this case
             } else {
                //stream was completely processed
                //do whatever should happen in this case
             }
        }
    );

由于err 包含提前终止期间抛出的消息，我们可以对此做出具体反应，并在sc 中显示部分聚合的数据。这解决了最初的问题，并且似乎是这种情况下唯一明显的途径。因此，我发布了这个解决方案。拥有其他（更优雅的）解决方案会很棒。

【讨论】：