Node - comment puis-je rediriger vers un nouveau flux READABLE?

Question

Mon application est censée prendre l'audio Web en streaming du client, l'encoder en MP3, avant de la renvoyer aux clients via WebSocket.

Je peux actuellement coder et conduit à un fichier comme celui-ci:

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

Et si je déjà un fichier sur le serveur que je peux faire ceci:

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

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

Cependant, je voulez accéder aux blocs codés en temps réel, et les envoyer aux clients - pas écrire dans un fichier.

Ce que je veux est-ce, efficacement:

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

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

Je l'ai regardé Duplex et transformer les flux, mais franchement, je suis encore à apprendre et le Prototypage fait tourner la tête.

Comment faire? Merci.

MISE À JOUR

La solution ci-dessous de @Nazar Sakharenko fait certainement ce que je voulais, mais la surcharge de l'encodage direct semble faire ce inpossible, écrit ainsi le MP3 encodée, et pré-tampon, il semble être le seul façon (merci à diverses personnes pour la suggestion.)

Cependant, j'ai toujours des problèmes avec cette approche. Nouvelle question ici:

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

Answer 1

Selon la documentation readable.pipe(destination[, options]) la destination doit être stream.Writable.

Ce que vous pouvez faire est de mettre en œuvre votre propre flux inscriptibles:

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

c'est tout.