如何编写 Node.js 模块来处理传入的管道流

Question

我正在尝试编写一个接受传入管道二进制（或 base-64 编码）流的节点模块，但坦率地说，我什至不知道从哪里开始。我在 Node 文档中看不到任何关于 handling 传入流的示例；我只看到使用它们的示例？

比如说我希望能够做到这一点：

var asset = new ProjectAsset('myFile', __dirname + '/image.jpg')
var stream = fs.createReadStream(__dirname + '/image.jpg', { encoding: 'base64' }).pipe(asset)
stream.on('finish', function() {
    done()
})

ProjectAsset 看起来像这样，但我不知道下一步该去哪里：

'use strict'

var stream = require('stream'),
    util = require('util')

var ProjectAsset = function() {
    var self = this

    Object.defineProperty(self, 'binaryData', {
        configurable: true,
        writable: true
    })

    stream.Stream.call(self)

    self.on('pipe', function(src) {
        // does it happen here? how do I set self.binaryData?
    })

    return self
}

util.inherits(ProjectAsset, stream.Stream)

module.exports = ProjectAsset
module.exports.DEFAULT_FILE_NAME = 'file'

Answer 1

可以从stream.Stream 继承并使其工作，但是根据documentation 中可用的内容，我建议从stream.Writable 继承。通过管道连接到 stream.Writable，您需要定义 _write(chunk, encoding, done) 来处理管道。这是一个例子：

var asset = new ProjectAsset('myFile', __dirname + '/image.jpg')
var stream = fs.createReadStream(__dirname + '/image.jpg', { encoding: 'base64' }).pipe(asset)
stream.on('finish', function() {
    console.log(asset.binaryData);
})

项目资产

'use strict'

var stream = require('stream'),
    util = require('util')

var ProjectAsset = function() {
    var self = this

    self.data
    self.binaryData = [];

    stream.Writable.call(self)

    self._write = function(chunk, encoding, done) {
        // Can handle this data however you want
        self.binaryData.push(chunk.toString())
        // Call after processing data
        done()
    }
    self.on('finish', function() {
        self.data = Buffer.concat(self.binaryData)
    })

    return self
}

util.inherits(ProjectAsset, stream.Writable)

module.exports = ProjectAsset
module.exports.DEFAULT_FILE_NAME = 'file'

如果您还想从stream 中读取数据，请查看从stream.Duplex 继承并包括_read(size) 方法。

还有simplified constructors api，如果你做的事情更简单的话。

Answer 2

我不确定这是否正是您要查找的内容，但我认为您可以在缓冲区数组上使用带有Buffer.concat 的缓冲区api 处理它，这些缓冲区可以从chunk 在流data 侦听器上检索

'use strict'

var stream = require('stream'),
    util = require('util');

var ProjectAsset = function() {
    var self = this

    Object.defineProperty(self, 'binaryData', {
        configurable: true,
        writable: true
    })

    stream.Stream.call(self)
    var data;
    var dataBuffer=[];
    self.on('data', function(chunk) {
        dataBuffer.push(chunk);
    }).on('end',function(){
        data=Buffer.concat(dataBuffer);
    });
    self.binaryData=data.toString('binary');
    return self
}

util.inherits(ProjectAsset, stream.Stream)

module.exports = ProjectAsset
module.exports.DEFAULT_FILE_NAME = 'file'

Answer 3

由于您使用var asset = new ProjectAsset('myFile', __dirname + '/image.jpg')，我想您的 ProjectAsset 职责是对一些输入流进行一些转换并将其写入文件。您可以实现转换流，因为您从流中接收一些输入并生成一些输出，这些输出可以保存到文件或其他写入流中。

您当然可以通过从 node.js 继承 Transform Stream 来实现转换流，但是继承非常麻烦，所以我的实现使用 through2 来实现转换流：

module.exports = through2(function (chunk, enc, callback) {
  // This function is called whenever a piece of data from the incoming stream is read
  // Transform the chunk or buffer the chunk in case you need more data to transform

  // Emit a data package to the next stream in the pipe or omit this call if you need more data from the input stream to be read
  this.push(chunk);

  // Signal through2 that you processed the incoming data package
  callback();
 }))

用法

var stream = fs.createReadStream(__dirname + '/image.jpg', { encoding: 'base64' })
               .pipe(projectAsset)
               .pipe(fs.createWriteStream(__dirname + '/image.jpg'));

正如您在此示例中所见，实现流管道将数据转换和数据保存完全解耦。

工厂函数

如果您喜欢在项目资产模块中使用类似构造函数的方法，因为您需要传递一些值或东西，您可以轻松导出构造函数，如下所示

var through2 = require('through2');

module.exports = function(someData) {

  // New stream is returned that can use someData argument for doing things
  return through2(function (chunk, enc, callback) {
    // This function is called whenever a piece of data from the incoming stream is read
    // Transform the chunk or buffer the chunk in case you need more data to transform

    // Emit a data package to the next stream in the pipe or omit this call if you need more data from the input stream to be read
    this.push(chunk);

    // Signal through2 that you processed the incoming data package
    callback();
  });
}

用法

var stream = fs.createReadStream(__dirname + '/image.jpg', { encoding: 'base64' })
               .pipe(projectAsset({ foo: 'bar' }))
               .pipe(fs.createWriteStream(__dirname + '/image.jpg'));