NodeJS：如何使用管道读取两个文件并写入单个输出文件？

Question

上下文

我正在使用事件流模块来帮助我读取和写入这些本地文件，我希望 return 生成一个结果文件。长话短说，我预计 2 个输入文件（通过 express API 作为 multipart/form-data 发送）的大小可能超过 200MB，其中包含一个条目列表（每行 1 个）。我想做的是将这些条目组合成以下格式 <entry1>:<entry2>，其中 entry1 是第一个文件的条目，entry2 来自第二个文件。我这样做的方式较早，我可以在内存中存储 return inputs/outputs，但是由于我的应用程序服务器上的内存 space 非常有限，所以我运行堆内存不足。我读到我可以使用事件流和管道逐行读取每个文件并输出到文件，而不是使用读取流输出到内存中的大字符串。问题是我似乎无法正确解决 way/time 以便生成的输出文件准备好发送回调用者。

到目前为止我有什么

到目前为止我所做的工作是我得到了我期望的正确文件输出，然而，这似乎是一个异步问题，因为我在文件实际完成之前解决了承诺writing/saving.请在下面查看我的代码...

const fs = require('fs');
const es = require('event-stream');
const uuid = require('uuid');

const buildFile = async (fileOne, fileTwo) =>
    await new Promise((resolve, reject) => {
        try {
            // Output stream
            let fileID = uuid.v4();
            let outStream = fs
                .createWriteStream(`files/outputFile-${fileID}.txt`, {
                    flags    : 'a',
                    encoding : 'utf-8'
                });

            let fileOneRS = fs
                .createReadStream(fileOne.path, {
                    flags    : 'r',
                    encoding : 'utf-8'
                })
                .pipe(es.split())
                .pipe(
                    es.mapSync((lineOne) => {
                        fileOneRS.pause();

                        let fileTwoRS = fs
                            .createReadStream(fileTwo.path, {
                                flags    : 'r',
                                encoding : 'utf-8'
                            })
                            .pipe(es.split())
                            .pipe(
                                es.mapSync((lineTwo) => {
                                    fileTwoRS.pause();

                                    // Write combo to file
                                    outStream.write(`${lineOne}:${lineTwo}\n`);

                                    fileTwoRS.resume();
                                })
                            );

                        fileOneRS.resume();
                    })
                ); // This is where I have tried doing .on('end', () => resolve), but it also does not work :(
        } catch (err) {
            reject(err);
        }
    });

注意：这个函数是从另一个服务函数调用的，如下：

buildFile(fileOne, fileTwo)
    .then((result) => {
        resolve(result);
    })
    .catch((err) => {
        console.log(err);
        reject(err);
    });

作为新手 Javascript 开发人员，甚至是 NodeJS 的新手，我已经坚持尝试自己解决这个问题超过 2 周了。如果有人能够提供帮助，我将不胜感激这里的一些智慧！

谢谢

Answer 1

编辑：更新代码以符合 OP 的预期输出。

promise' resolve() 函数应在写入流完成后调用。 OP 代码段中提供的注释表明，解析函数可能在耗尽 fileOneRS（在 pipe() 链的末尾）时被调用。

与其为第一个文件中的每一行创建一个新的读取流，代码应该只实例化一次读取流。

以下示例说明如何重构此代码流以仅读取每行一次，并逐行连接文件 A 和 B 中的行：

import stream from "stream";
import util from "util";
import readline from "readline";
import fs from "fs";
import os from "os";

/** Returns a readable stream as an async iterable over text lines */
function lineIteratorFromFile( fileStream ){
  return readline.createInterface({
    input: fileStream,
    crlfDelay: Infinity
  })
}

// Use stream.pipeline to handle errors and to stream the combined output
// to a Writable stream. The promise will resolve once the data has finished
// writing to the output stream.
await util
  .promisify(stream.pipeline)(
    async function*(){
      for await ( const lineA of lineIteratorFromFile(fs.createReadStream( "./in1.txt" ))){
        for await (const lineB of lineIteratorFromFile(fs.createReadStream( "./in2.txt" ))){
          yield `${lineA}: ${lineB}${os.EOL}`
        }
      }
    },
    fs.createWriteStream( outputFile )
  );

下面折叠的代码段中提供了一个使用 NodeJS v13+ 的可运行示例：

// in1.txt:
foo1
foo2

// in2.txt:
bar1
bar2

// out.txt (the file created by this script, with expected output):
foo1: bar1
foo1: bar2
foo2: bar1
foo2: bar2

// main.mjs:
import stream from "stream";
import util from "util";
import readline from "readline";
import fs from "fs";
import os from "os";

/** Returns a readable stream as an async iterable over text lines */
function lineIteratorFromFile( fileStream ){
  return readline.createInterface({
input: fileStream,
crlfDelay: Infinity
  })
}

(async ()=>{
  await util
.promisify(stream.pipeline)(
  async function*(){
    for await ( const lineA of lineIteratorFromFile(fs.createReadStream( "./in1.txt" ))){
      for await (const lineB of lineIteratorFromFile(fs.createReadStream( "./in2.txt" ))){
        yield `${lineA}: ${lineB}${os.EOL}`
      }
    }
  },
  fs.createWriteStream( "./out.txt" )
);
})()
  .catch(console.error);

NodeJS：如何使用管道读取两个文件并写入单个输出文件？

NodeJS: How to read from two files and write to single output file using pipes?

javascript

pipe

event-stream

node.js

express

上下文

到目前为止我有什么