如何正确实现 nodejs Stream API? [英] 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 string.

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

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

每当我调用pipe"时，我都会收到这个奇怪的错误:

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)

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

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() }

而不是像这样:

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

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

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.

这篇关于如何正确实现 nodejs Stream API?的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！