javascript - express .js : stream result from Mongo and serialize non-blockingly

Question

我想做的是更新 this answer不依赖 JSONStream library ，不再维护。我们有:

Comment.find()
  .cursor()
  .pipe(JSONStream.stringify())
  .pipe(res.type('json'))

它使用 Mongoose .cursor()，返回 Node.js-compatible readable stream ，但我愿意使用 native mongo 驱动程序。

现在，我的第一个问题:是否还有人在使用 Node.js 流，或者您现在应该使用 JavaScript 迭代器和生成器？

如果是这样，我想我可以convert the cursor to an iterator并将每个 block 分别转换为 JSON。 (虽然进行错误处理等的图书馆建议是受欢迎的，即使这里离题而不是这个问题的核心)。

但是如何将迭代器流式传输到 express.js 结果中呢？

我找不到任何相关文档(尽管也找不到关于 res 是可写流的文档，尽管它有效。)我在这里的想法是否正确？ ？

编辑:

与此同时，我做了更多研究并发现了以下库:

• pull-stream – 没有 TypeScript 类型，但有 JSONStream 的创建者作为贡献者
• ts-stream – 与 no built-in node stream conversion
• scramjet – 看起来很有趣……也许有点矫枉过正，但至少为他们的流实现了 map 和 filter 等......？

Answer 1

编辑:添加了自定义字符串化步骤。

ExpressJS 中的res 对象是http.ServerResponse 的可写子类。 , 并且可以是管道数据。

我倾向于使用 NodeJS 的内置支持将迭代器转换为可读，并使用 stream.pipeline 进行异常处理来连接此数据流。

Note that it's no longer necessary to convert the cursor to a readable in NodeJS v13+, as stream.pipeline now accepts async iterators in place of a stream.

Note that it is redundant to use stringify() if it is possible to use Mongoose's lean() directly. Lean will emit JSON data.

import stream from "stream";
import util from "util";

function handler(req, res, next){
  try {
    // init the cursor
    const cursor = Comment.find().lean(); // "lean" will emit json data
    const readable = stream.Readable.from( cursor );
    // promisifying the pipeline will make it throw on errors
    await util.promisify(stream.pipeline)( readable, res.type('json') );
    next();
  }
  catch( error ){
    next( error );
  }
}

在 NodeJS v13+ 中使用自定义字符串化:

import stream from "stream";
import util from "util";

function handler(req, res, next){
  try {
    // init the cursor
    const cursor = Comment.find().lean(); // "lean" will emit json data
    const readable = stream.Readable.from( cursor );
    // promisifying the pipeline will make it throw on errors
    await util.promisify(stream.pipeline)( 
      readable,
      // Custom "stringifying" using an async iterator
      async function*( source ){
        // Add some output before the result from mongodb. Typically, this could be information about 
        // the number of results in a REST API.
        yield "Appended"

        for await (const comment of source ){
          // Emit a "projection" of the data retrieved from MongoDB 
          yield {
            text: comment.text,
            // Add a new property:
            newProperty: true
          }
        }
        // Add some final data to the response. In a REST API, this might be the closing bracket of an array "]".
        yield "Prended"
      },
      // the stringified data is then piped to express' res object
      res.type('json') 
    );
    next();
  }
  catch( error ){
    next( error );
  }
}