mongodb 以 through2.spy 结束的流查询未被调用

Question

我有一个来自 mongo 的流查询，我正在将其通过管道传输到一个 through2 "spy" 可写流。它完全可以工作，包括带有 5 个文档的小集合的 "end" 回调。然而，对于包含 344 个文档的更大集合，只有前 15 个通过，然后它永远挂起，并且 "end" 事件永远不会触发。这是一个 MCVE：

var spy = require("through2-spy").obj;
var MongoClient = require("mongodb").MongoClient;

function getStream() {
  var stream = spy(function() {
    console.log("@bug counting", stream.total++);
  });
  stream.total = 0;
  return stream;
}

function onEnd() {
  console.log("ended");
}

MongoClient.connect(process.argv[2], function(error, db) {
  if (error) {
    console.error(error);
    return;
  }
  var stream = db.collection(process.argv[3]).find().stream();
  stream
    // behavior is the same with the follow line commented out or not
    .on("end", db.close.bind(db))
    .on("error", console.error)
    .on("end", onEnd)
    .pipe(getStream());
});

Answer 1

问题出在 through2-spy 默认使用 highWaterMark 16。为了处理流量控制，流维护一个内部缓冲区，当从中使用数据时该缓冲区会被清除。因为没有可读流使用 getStream 返回的转换流中的数据，内部缓冲区被填满并到达 highWaterMark。增加 highWaterMark 应该可以解决这个问题：

var stream = spy({highWaterMark: 350}, function() {
  console.log("@bug counting", stream.total++);
});

另一种非标准的替代方法是重置转换流的可读状态：

var stream = spy(function() {
    console.log("@bug counting", stream.total++);
    this._readableState.length = 0;
});

Answer 2

解决这个问题的另一种方法是确保下游有一些东西可以完全读取上游源代码直至完成。我最终在流的末尾添加了一个额外的 .pipe(terminus.devnull({objectMode: true});，这也起到了作用。

var MongoClient = require("mongodb").MongoClient;
var spy = require("through2-spy").obj;
var terminus = require("terminus");

function getStream() {
  var stream = spy(function() {
    console.log("@bug counting", stream.total++);
  });
  stream.total = 0;
  return stream;
}

function onEnd() {
  console.log("ended");
}

MongoClient.connect(process.argv[2], function(error, db) {
  if (error) {
    console.error(error);
    return;
  }
  var stream = db.collection(process.argv[3]).find().stream();
  stream
    // behavior is the same with the follow line commented out or not
    .on("end", db.close.bind(db))
    .on("error", console.error)
    .on("end", onEnd)
    .pipe(getStream())
    .pipe(terminus.devnull({objectMode: true}));
});

mongodb 以 through2.spy 结束的流查询未被调用

mongodb stream query with through2.spy end is not called

mongodb

node.js

node.js-stream