节点-如何通过管道传输到新的READABLE流? [英] Node - how can i pipe to a new READABLE stream?

问题描述

我的应用应该从客户端流式传输Web音频，将其编码为MP3，然后再通过WebSocket发送回客户端.

My app is supposed to take Web Audio streamed from the client, encode it as MP3, before sending it back out to clients via WebSocket.

我目前可以像这样对文件进行编码和管道传输:

I can currently encode and pipe to a file like this:

inbound_stream.pipe(encoder).pipe(fs.createWriteStream('audio.mp3'));

如果我已经在服务器上拥有文件，则可以执行以下操作:

And if I already have a file on the server I can do this:

var mp3File = fs.createReadStream('audio.mp3');

            mp3File.on('data', function(buffer){
                io.sockets.emit('audio', { buffer: buffer });
            });

但是，我想实时访问编码的块，然后将其发送给客户端-不是写入文件.

However, I want to access the encoded chunks in real time, and send those out to clients - not write to a file.

我想要的是有效的:

inbound_stream.pipe(encoder).pipe(newReadStream);

        newReadStream.on('data', function(buffer){
            io.sockets.emit('audio', { buffer: buffer });
        });

我已经看过Duplex和Transform流，但是坦率地说，我仍在学习，原型制作使我旋转.

I've looked at Duplex and Transform streams, but frankly I am still learning and the Prototyping made my head spin.

我该怎么做?谢谢.

下面来自@Nazar Sakharenko的解决方案当然可以满足我的要求，但是实时编码的开销似乎无法实现这一点，因此编写编码的MP3并进行预缓冲似乎是唯一的方法(这要感谢各个人)的建议.)

The solution below from @Nazar Sakharenko certainly does what I wanted, but the overhead of live encoding seems to make this inpossible, so writing the encoded MP3, and pre-buffering it seems to be the only way (thanks to various people for the suggestion.)

但是我仍然对这种方法有疑问.这里有新问题:

However I still have problems with this approach. New question here:

推荐答案

根据文档 ready.pipe (目的地[，选项])目的地应为流.可写.

您可以做的是实现自己的可写流:

What you can do is to implement your own Writable stream:

const Writable = require('stream').Writable;

var buffer = [];
//in bytes
const CHUNK_SIZE = 102400; //100kb

const myWritable = new Writable({
  write(chunk, encoding, callback) {
    buffer += chunk;
    if(buffer.length >= CHUNK_SIZE) {
       io.sockets.emit('audio', { buffer: buffer});
       buffer = [];
    }

    callback();
  }
});

myWritable.on('finish', () => {
   //emit final part if there is data to emit
   if(buffer.length) {
       io.sockets.emit('audio', { buffer: buffer});
   }
});


inbound_stream.pipe(encoder).pipe(myWritable);

仅此而已.

这篇关于节点-如何通过管道传输到新的READABLE流?的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！