如何正确实现 nodejs Stream API？答案

【问题标题】：How can I properly implement the nodejs Stream API?如何正确实现 nodejs Stream API？
【发布时间】：2011-08-06 00:08:42
【问题描述】：

我有一个应用程序，它有时从文件读取输入，有时通过套接字读取输入，有时从同一程序的另一部分读取输入，生成缓冲区或字符串。

作为数据源的套接字和文件都可以使用节点的 Stream API 进行处理，我一直在尝试为缓冲区和字符串提供某种包装器来模仿它；我不想为了处理字符串作为输入而编写两个版本的“消费者”代码，当数据已经在内存中时，我不想将它写入磁盘并再次读回。

大多数事情似乎都可以正常工作，但有些东西破坏了我对“管道”的实现：

MemoryStream.prototype = new process.EventEmitter;

MemoryStream.prototype.pipe = function(dest,opts){
    var that=this;
    function pipe_mem(data){
        if(!dest.write(data)){
            that.pause();
            dest.once('drain',function(){
                if(pipe_mem(data)){that.resume();}
            });
            return false;
        }
        return true;
    }
    this.on('data',pipe_mem);
    if(!opts || opts.end){this.on('end',dest.end);}
    this.resume();
};

MemoryStream.prototype.resume = function(){
    var i,l,encoder;
    if(this.active) return;
    this.active = true;
    l = this.buffer.length-1;
    if(l>=0){
        encoder = this.encoding?emit_string:emit_buffer;
        for(i=0;i<l;i++){encoder(this,this.buffer[i],this.encoding);}
        if(this.buffer[i]===''){
            this.emit('end');
            this.destroy();
        }else{encoder(this,this.buffer[i],encoding);}
        this.buffer = [];
    }
};

每当我调用“管道”时，我都会收到这个奇怪的错误：

TypeError: Object #<EventEmitter> has no method '_implicitHeader'
    at EventEmitter.<anonymous> (http.js:651:10)
    at EventEmitter.emit (events.js:61:17)
    at EventEmitter.resume (/MemoryStream.js:36:9)

/MemoryStream.js 第 36 行是this.emit('end');

知道发生了什么吗？我该如何解决这个问题，或者，有没有更好的方法来做我想做的事？

【问题讨论】：

标签： javascript node.js stream

【解决方案1】：

答案是：在某个地方你可以这样称呼：

var end = response.end; response.end = function() { end() }

而不是这样的：

var end = response.end; response.end = function() { response.end = end; response.end() }

你看到了吗？上下文不同：在第一种情况下，您使用 'this === global' 调用它，但没有 global._implicitHeader 函数，而在第二种情况下，您使用 'this === response' 调用它，并且有response._implicitHeader 函数。

【讨论】：