如何在node.js中读取整个文本流？

Question

RingoJS中有一个名为read的函数，它允许您在流达到末尾时读取整个流。当您制作命令行应用程序时，这非常有用。例如，您可以编写一个tac程序，如下所示：

#!/usr/bin/env ringo

var string = system.stdin.read(); // read the entire input stream
var lines = string.split("\n");   // split the lines

lines.reverse();                  // reverse the lines

var reversed = lines.join("\n");  // join the reversed lines
system.stdout.write(reversed);    // write the reversed lines

这使您可以启动一个 shell 并运行 tac 命令。然后，您可以输入任意数量的行，完成后按下 Ctrl+D（在 Windows 上是 Ctrl+Z）来表示传输结束。

我想在 node.js 中做同样的事情，但我找不到能够执行此操作的任何函数。我想过使用 fs库中的 readSync 函数 来模仿，但无济于事：

fs.readSync(0, buffer, 0, buffer.length, null);

标准输入流的文件描述符（第一个参数）是0。因此，它应该从键盘读取数据，但是它给我以下错误：

Error: ESPIPE, invalid seek
    at Object.fs.readSync (fs.js:381:19)
    at repl:1:4
    at REPLServer.self.eval (repl.js:109:21)
    at rli.on.self.bufferedCmd (repl.js:258:20)
    at REPLServer.self.eval (repl.js:116:5)
    at Interface.<anonymous> (repl.js:248:12)
    at Interface.EventEmitter.emit (events.js:96:17)
    at Interface._onLine (readline.js:200:10)
    at Interface._line (readline.js:518:8)
    at Interface._ttyWrite (readline.js:736:14)

你如何在node.js中同步收集输入文本流中的所有数据并作为字符串返回？一个代码示例将非常有帮助。

- Aadit M Shah

---

你不能在异步流中同步读取。反正你为什么要这样做呢？ - beatgammit

我正在尝试做同样的事情。原因是在我的程序中创建一个交互式选项，这对很多方面都很有用。异步读取器并不能帮太大的忙。 - ton

这里有一个方法 https://www.npmjs.com/package/readline-sync：https://dev59.com/questions/Wl7Va4cB1Zd3GeqPGyXq#27931290 - ton

Answer 1

由于node.js是事件和流导向的，没有API可以等待stdin结束并缓冲结果，但可以手动轻松实现该功能。

var content = '';
process.stdin.resume();
process.stdin.on('data', function(buf) { content += buf.toString(); });
process.stdin.on('end', function() {
    // your code here
    console.log(content.split('').reverse().join(''));
});

在大多数情况下，最好不缓存数据并在数据到达时立即处理（使用已有的流解析器链，例如xml或zlib，或您自己的FSM解析器）。

- Andrey Sidorov

---

在恢复后，您可以执行 process.stdin.setEncoding('utf-8'); ，回调中的 bug 将变为字符串。 - Mitar

类似，但使用 Buffer.concat()：https://dev59.com/iWgv5IYBdhLWcg3wTvPL - joeytwiddle

@Mitar：是“buf”，不是“bug”。 - evandrix

为什么要反转字符串？ - Paul

那只是一个处理数据的例子。 - Andrey Sidorov

注意：如果没有传递任何内容到标准输入(stdin)，你的程序将会挂起。你可以使用process.stdin.isTTY来检测你的脚本是否在管道中被调用，具体细节请参考stackoverflow上的另一个答案。 - Brad Parks

Answer 2

关键是要使用这两个流事件：

Event: 'data'
Event: 'end'

使用 stream.on('data', ...) 时，应将数据收集到缓冲区（如果是二进制）或字符串中。

对于 on('end', ...)，您应该通过回调函数调用已完成的缓冲区，或者如果可以将其内联并使用 Promises 库进行返回。

- Nathan

Answer 3

让我来解释一下StreetStrider的答案。

以下是使用concat-stream的方法：

var concat = require('concat-stream');

yourStream.pipe(concat(function(buf){
    // buf is a Node Buffer instance which contains the entire data in stream
    // if your stream sends textual data, use buf.toString() to get entire stream as string
    var streamContent = buf.toString();
    doSomething(streamContent);
}));

// error handling is still on stream
yourStream.on('error',function(err){
   console.error(err);
});

请注意，process.stdin是一个流对象。

- Ali Ok

Answer 4

有一个用于特定任务的模块，叫做concat-stream。

- Strider

---

该模块允许您将块与另一个字符串交错。可能仅用于调试：https://www.npmjs.org/package/join-stream - joeytwiddle

Answer 5

如果您处于async上下文并且拥有最新版的Node.js，则这里有一个快速的建议：

const chunks = []
for await (let chunk of readable) {
  chunks.push(chunk)
}
console.log(Buffer.concat(chunks))

- famzah

---

精彩的答案，简短明了，非常适合我的使用情况。好东西！ - Aaron Meese

Answer 6

这是一个老问题，但值得一提的是Node.js有一些新的流辅助工具，其中之一是toArray：

require('http')
    .createServer(async (req, res) => {
        const str = (await req.toArray()).toString().toUpperCase();
        res.end(str);
    })
    .listen(4000);

请注意：该 API 目前被标记为实验性的，因此可能更适合用于测试/非生产代码。

- Lee Houghton

Answer 7

在Windows上，我遇到了一些其他解决方案发布的问题 - 当没有输入时，程序会无限运行。

这里是一个TypeScript实现，适用于现代NodeJS，使用异步生成器和for await - 比使用旧的基于回调的API更简单、更健壮，而且在Windows上也可以工作：

import process from "process";

/**
 * Read everything from standard input and return a string.
 * 
 * (If there is no data available, the Promise is rejected.)
 */
export async function readInput(): Promise<string> {  
  const { stdin } = process;

  const chunks: Uint8Array[] = [];

  if (stdin.isTTY) {
    throw new Error("No input available");
  }

  for await (const chunk of stdin) {
    chunks.push(chunk);
  }

  return Buffer.concat(chunks).toString('utf8');
}

例子：

(async () => {
  const input = await readInput();

  console.log(input);
})();

（如果您想处理 Promise 拒绝并在没有输入时显示更用户友好的错误消息，请考虑添加 try/catch。）

- mindplay.dk