javascript - framework - Node.js Canalizando la misma secuencia legible en múltiples destinos(modificables)

node js vs php

Question

Necesito ejecutar dos comandos en serie que necesitan leer datos de la misma transmisión. Después de conectar una secuencia a otra, el almacenamiento intermedio se vacía, por lo que no puedo volver a leer datos de esa secuencia, así que esto no funciona:

var spawn = require(''child_process'').spawn;
var fs = require(''fs'');
var request = require(''request'');

var inputStream = request(''http://placehold.it/640x360'');
var identify = spawn(''identify'',[''-'']);

inputStream.pipe(identify.stdin);

var chunks = [];
identify.stdout.on(''data'',function(chunk) {
  chunks.push(chunk);
});

identify.stdout.on(''end'',function() {
  var size = getSize(Buffer.concat(chunks)); //width
  var convert = spawn(''convert'',[''-'',''-scale'',size * 0.5,''png:-'']);
  inputStream.pipe(convert.stdin);
  convert.stdout.pipe(fs.createWriteStream(''half.png''));
});

function getSize(buffer){
  return parseInt(buffer.toString().split('' '')[2].split(''x'')[0]);
}

Solicitar quejas sobre esto

Error: You cannot pipe after data has been emitted from the response.

y cambiar el inputStream a fs.createWriteStream produce el mismo problema, por supuesto. No quiero escribir en un archivo pero reutilizar de alguna manera el flujo que produce la solicitud (o cualquier otra cosa para el caso).

¿Hay alguna forma de reutilizar una secuencia legible una vez que finaliza la canalización? ¿Cuál sería la mejor manera de lograr algo como el ejemplo anterior?

Answer 1

¿Qué hay sobre conectar dos o más corrientes al mismo tiempo?

Por ejemplo :

var PassThrough = require(''stream'').PassThrough;
var mybiraryStream = stream.start(); //never ending audio stream
var file1 = fs.createWriteStream(''file1.wav'',{encoding:''binary''})
var file2 = fs.createWriteStream(''file2.wav'',{encoding:''binary''})
var mypass = PassThrough
mybinaryStream.pipe(mypass)
mypass.pipe(file1)
setTimeout(function(){
   mypass.pipe(file2);
},2000)

El código anterior no produce ningún error, pero el archivo2 está vacío

Answer 2

Debe crear un duplicado de la secuencia conectándolo a dos flujos. Puede crear una secuencia simple con una transmisión PassThrough, simplemente pasa la entrada a la salida.

const spawn = require(''child_process'').spawn;
const PassThrough = require(''stream'').PassThrough;

const a = spawn(''echo'', [''hi user'']);
const b = new PassThrough();
const c = new PassThrough();

a.stdout.pipe(b);
a.stdout.pipe(c);

let count = 0;
b.on(''data'', function (chunk) {
  count += chunk.length;
});
b.on(''end'', function () {
  console.log(count);
  c.pipe(process.stdout);
});

Salida:

8
hi user

Answer 3

La primera respuesta solo funciona si las transmisiones toman aproximadamente la misma cantidad de tiempo para procesar los datos. Si uno tarda mucho más tiempo, el más rápido solicitará nuevos datos, sobrescribiendo los datos que aún están siendo utilizados por el más lento (tuve este problema después de tratar de resolverlo usando un flujo duplicado).

El siguiente patrón funcionó muy bien para mí. Utiliza una biblioteca basada en streams Stream2, Streamz y Promises para sincronizar transmisiones asincrónicas a través de una devolución de llamada. Usando el ejemplo familiar de la primera respuesta:

spawn = require(''child_process'').spawn;
pass = require(''stream'').PassThrough;
streamz = require(''streamz'').PassThrough;
var Promise = require(''bluebird'');

a = spawn(''echo'', [''hi user'']);
b = new pass;
c = new pass;   

a.stdout.pipe(streamz(combineStreamOperations)); 

function combineStreamOperations(data, next){
  Promise.join(b, c, function(b, c){ //perform n operations on the same data
  next(); //request more
}

count = 0;
b.on(''data'', function(chunk) { count += chunk.length; });
b.on(''end'', function() { console.log(count); c.pipe(process.stdout); });

Answer 4

Para un problema general, el siguiente código funciona bien

var PassThrough = require(''stream'').PassThrough
a=PassThrough()
b1=PassThrough()
b2=PassThrough()
a.pipe(b1)
a.pipe(b2)
b1.on(''data'', function(data) {
  console.log(''b1:'', data.toString())
})
b2.on(''data'', function(data) {
  console.log(''b2:'', data.toString())
})
a.write(''text'')

Answer 5

Tengo una solución diferente para escribir en dos transmisiones simultáneamente, naturalmente, el tiempo para escribir será la suma de las dos veces, pero lo uso para responder a una solicitud de descarga, donde quiero mantener una copia del archivo descargado en mi servidor (en realidad utilizo una copia de seguridad S3, así que guardo en caché los archivos más usados ​​localmente para evitar transferencias múltiples de archivos)

/**
 * A utility class made to write to a file while answering a file download request
 */
class TwoOutputStreams {
  constructor(streamOne, streamTwo) {
    this.streamOne = streamOne
    this.streamTwo = streamTwo
  }

  setHeader(header, value) {
    if (this.streamOne.setHeader)
      this.streamOne.setHeader(header, value)
    if (this.streamTwo.setHeader)
      this.streamTwo.setHeader(header, value)
  }

  write(chunk) {
    this.streamOne.write(chunk)
    this.streamTwo.write(chunk)
  }

  end() {
    this.streamOne.end()
    this.streamTwo.end()
  }
}

Puede usar esto como un OutputStream regular

const twoStreamsOut = new TwoOutputStreams(fileOut, responseStream)

y pásalo a tu método como si fuera una respuesta o un fileOutputStream