如何在levelup(node.js)中使用管道？

Question

LevelUP文档说明pipe()可以使用(https://github.com/rvagg/node-levelup/#pipes-and-node-stream-compatibility).

我试过以下代码:

db.createValueStream().pipe(response)

但我不能这样做,我有一个错误:

events.js:72
        throw er; // Unhandled 'error' event
              ^
TypeError: Invalid non-string/buffer chunk
    at validChunk (_stream_writable.js:150:14)
    at Writable.write (_stream_writable.js:179:12)
    at write (_stream_readable.js:573:24)
    at flow (_stream_readable.js:582:7)
    at ReadStream.pipeOnReadable (_stream_readable.js:614:5)
    at ReadStream.EventEmitter.emit (events.js:92:17)
    at emitReadable_ (_stream_readable.js:408:10)
    at emitReadable (_stream_readable.js:404:5)
    at readableAddChunk (_stream_readable.js:165:9)
    at ReadStream.Readable.push (_stream_readable.js:127:10)

实际问题是关于使用事件'data'()时的内存使用情况.然后我试着做一个流.转换和使用pipe()来做我需要的东西.事件发生器中的内存泄漏是一个问题:在Node.js中使用流时内存泄漏？

UPDATE

我试过@ paul-mougel没有成功.不调用错误事件的功能,并且它崩溃了.这是一段代码:

    var rs = db.createValueStream();

    request.on('close', function(){
        rs.destroy();
        response.end();
    });

    rs.on('end', function(){
        response.end();
    });
    rs.on('error', function(err){
        console.err('READ STREAM ERROR:',err.message);
        response.end();
        rs.destroy();
    });

    response.on('error', function(err){
        console.log('RESPONSE ERROR:',err);
        rs.destroy();
    });

    rs.pipe(stringifier).pipe(response);

Answer 1

有很多事情需要考虑.

首先,你得到这个例外,因为你不听这个error事件.在流的情况下,总是听它,i)将允许您记录问题ii)不会使程序崩溃.

var valueStream = db.createValueStream()
valueStream.on('error', function (err) {
  console.error('valueStream.on error ' + err.message);
});
valueStream.pipe(response);
response('error', function (err) {
  console.error('response error ' + err.message);
});

其次,db.createValueStream()在对象模式下创建可读流(参见源代码):它将输出javascript对象.另一方面,您response是字节模式下的可写流:它只需要字节作为输入,因此error事件.你可以做的是创建一个转换流,它将javascript对象作为输入并输出其字符串化版本:

var stream = require('stream')
var stringifier = new stream.Transform();
stringifier._writableState.objectMode = true;
stringifier._transform = function (data, encoding, done) {
    this.push(JSON.stringify(data));
    this.push('\n');
    done();
}

valueStream.pipe(stringifier).pipe(response);

请注意,我们创建一个转换流,将对象作为输入和输出字节.有关更多信息,请参阅文档.

但是,你必须告诉我们更多关于你试图解决的具体问题,方法是将水平流加工成一个请求:上述解决方案不是一个很好的解决方案.

第三,使用时没有遇到内存泄漏.on('data').添加此侦听器会将流转换为流模式,这意味着它将尽快输出数据.您始终可以使用.pause()和.resume()方法来停止和重新启动流.但是,使用新v0.10流接口(又名streams2)可以帮助你处理这个问题,因为Readable,Writable和Transform班照顾所有这一切为您服务.