How can I properly implement the nodejs Stream API?

I've got a piece of an application that reads input sometimes from a file, sometimes over a socket, and sometimes from another part of the same program which produces a buffer or a 开发者_StackOverflow社区string.

Both sockets and files as datasources can be handled with node's Stream API, and I've been trying to come up with some kind of wrapper for buffer and strings to mimic that; I don't want to have to write two versions of the 'consumer' code just to handle strings as input, and when the data's already in memory I don't want to have to write it to disk and read it back again.

Most things seem to work just fine, but something is breaking my implementation of 'pipe':

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 = [];
    }
};

Whenever I call 'pipe', I get this odd error:

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)

Where /MemoryStream.js line 36 is this.emit('end');

Any idea what's going on? How can I fix this, or, is there a better way to do what I want?

The answer is: somewhere you call something like this:

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

Instead of something like this:

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

You see? The context is different: in the first case you call it with 'this === global', but there is no global._implicitHeader function, and in the second case you call it with 'this === response', and there is response._implicitHeader function.