node.js - 如何正确计算已处理的node.js流的字节数？

Question

我有一个通过线路发送的流，需要一些时间才能完全发送，所以我想显示它的传输距离。我知道您可以监听流的“数据”事件，但在较新版本的 Node 中，它还会将流置于“流动模式”。我想确保我这样做是正确的。

目前我有以下内容:

deploymentPackageStream.pause() // to prevent it from entering "flowing mode"
var bytesSent = 0
deploymentPackageStream.on('data', function(data) {
    bytesSent+=data.length
    process.stdout.write('\r                   ')
    process.stdout.write('\r'+(bytesSent/1000)+'kb sent')
})
deploymentPackageStream.resume()

// copy over the deployment package
execute(conn, 'cat > deploymentPackage.sh', deploymentPackageStream).wait()     

这给了我正确的 bytesSent 输出，但生成的包似乎缺少前面的一些数据。如果我在执行复制行(最后一行)后放置“恢复”行，则它不会复制任何内容。如果我不继续，它也不会复制任何内容。发生了什么事以及如何在不中断流且不进入流动模式(我想要背压)的情况下正确执行此操作？

我应该提到，我仍在使用 Node v0.10.x

Answer 1

好吧，我做了一些本质上是直通的东西，但在数据传入时调用回调:

// creates a stream that can view all the data in a stream and passes the data through
// parameters:
    // stream - the stream to peek at
    // callback - called when there's data sent from the passed stream
var StreamPeeker = exports.StreamPeeker = function(stream, callback) {
    Readable.call(this)
    this.stream = stream

    stream.on('readable', function() {
        var data = stream.read()
        if(data !== null) {
            if(!this.push(data)) stream.pause()
            callback(data)
        }
    }.bind(this))

    stream.on('end', function() {
        this.push(null)
    }.bind(this))
}
util.inherits(StreamPeeker, Readable)
StreamPeeker.prototype._read = function() {
    this.stream.resume()
}

如果我正确理解流，这应该适本地处理背压。

使用这个，我可以在回调中计算 data.length，如下所示:

var peeker = new StreamPeeker(stream, function(data) {
   // use data.length
})
peeker.pipe(destination)