Node.js:为什么转换流的转换函数从未被调用?

Question

我正在尝试在 Node.js 12 中编写自定义转换器流。具体来说，我在流(数据库驱动程序(中接收 json 对象并返回转换的对象。但是我的转换器函数从未被调用。 我也通过覆盖流来尝试这一点。转换类。

我想使自定义转换通用，因此我将其包含在闭包中，以便传入泛型函数：

// transformStream.js
var through2, transformStream;
through2 = require('through2');
transformStream = (handler) => {
// Through2 in Object Mode
_transformStream = through2.obj((data, encoding, callback) => {
console.log(data); // Never called
this.push(handler(data));
return callback();
// also tried:
// return callback(null, handler(data));
});
return _transformStream;
};
module.exports = transformStream;

以下是试用的测试台：

// transformStream.test.js
var jsonStream, through2, transformFunc, transformStream, transformer;
through2 = require('through2');
transformStream = require('./transformStream.js');
// Convert back to a string buffer for console output.
jsonStream = through2.obj(function(chunk, encoding, callback) {
return callback(null, JSON.stringify(chunk, null, 2) + 'n');
});
transformFunc = function(data) {
console.log("called with data", data); // Never called!
data.c = data.a * data.b;
return data;
};
// deviceStream.pipe(process.stdout)
transformer = transformStream(transformFunc);
transformer.on("error", function(error) {
return console.error(`Error in Transform: ${error.message}`);
});
transformer.pipe(jsonStream).pipe(process.stdout);
transformer.push({
a: 1,
b: 2
});

该流似乎有效，从不调用实际的转换代码，并且始终只返回原始 json：

{
A: 1,
b: 2
}

在控制台中。

我希望看到：{ a: 1, b:2, c:2 }

编辑：我还有另一个使用类的版本(绕过2(，具有相同的确切问题：

module.exports = TransformStream = class TransformStream extends Transform {
constructor(handler, {debug, highWaterMark, ...options}) {
super({
highWaterMark: highWaterMark || 10,
autoDestroy: true,
emitClose: true,
objectMode: true,
debug: true
});
this._transform = this._transform.bind(this);
this.handler = handler;
this.debug = debug;
this.options = options;
}
};
TransformStream.prototype._transform = (data, encoding, callback) => {
if (this.debug) {
console.log(data);
}
return callback(null, this.handler(data));
};

Answer 1

显然，执行TransformClass.push会调用流的内部输出缓冲区。 流实际上需要一个.write({})方法，并适当地调用_transform函数。

在测试台中，最终测试应该是：

transformer.write({
a: 1,
b: 2
});