Node - wie kann ich einen neuen READABLE-Stream bereitstellen?

Question

Meine App soll Web Audio vom Client übertragen, als MP3 kodieren, bevor sie über WebSocket an die Clients gesendet wird.

kann ich zur Zeit kodieren und Rohr in einer Datei wie folgt aus:

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

Und wenn ich bereits eine Datei auf dem Server kann ich dies tun:

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

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

Aber ich möchte in Echtzeit auf die codierten Chunks zugreifen und diese an Clients senden - nicht in eine Datei schreiben.

Was ich will, ist dies effektiv:

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

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

ich Duplex geschaut haben und Bäche Trans, aber ehrlich gesagt bin ich immer noch das Lernen und das Prototyping machte meinen Kopf drehen.

Wie mache ich das? Vielen Dank.

UPDATE

Die Lösung unterhalb von @Nazar Sakharenko sicherlich tut, was ich wollte, aber der Aufwand für die Live-Codierung scheint diese inpossible zu machen, so die codierten MP3 Schreiben und Pre-Pufferung es das einzige zu sein scheint Weg (Danke an verschiedene Leute für den Vorschlag.)

Allerdings habe ich immer noch Probleme mit diesem Ansatz. Neue Frage hier:

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

Answer 1

Laut Dokumentation readable.pipe(destination[, options]) das Ziel sollte sein stream.Writable.

Was können Sie tun, ist Ihre eigene Beschreibbare Strom zu implementieren:

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

das ist alles.