Search code examples
node.jswebsocketstreampipeduplex

Node - how can i pipe to a new READABLE stream?


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.

What I want is this, effectively:

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

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

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

How do I do this? Thanks.

UPDATE

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:

node.js - create a new ReadStream for a new file, when that file reaches a certain size


Solution

  • According to documentation readable.pipe(destination[, options]) the destination should be stream.Writable.

    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);
    

    thats all.