1

我有一个来自 mongo 的流式查询，我将它传送到一个 through2“间谍”可写流。它完全包括带有 5 个文档的小集合的“结束”回调。但是，对于 344 个文档的更大集合，只有前 15 个通过，然后它永远挂起，并且“结束”事件永远不会触发。这是一个 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());
});
4

2 回答 2

1

问题在于through2-spy默认使用highWaterMark16 的 a。为了处理流量控制，流维护一个内部缓冲区，当从它们中消耗数据时该缓冲区被清除。因为没有可读流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;
});
于 2015-03-22T16:19:36.900 回答
0

解决这个问题的另一种方法是确保下游有一些东西可以完全读取上游源以完成。我最终在流的末尾添加了一个额外.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}));
});
于 2015-03-23T01:24:34.747 回答