Empresas
Empleos
  • Sobre nosotros
  • Soluciones
    • Publicación de vacantes
      Publica tu vacante y recibe candidatos calificados en 48h.
    • Evaluación de candidatos
      500+ pruebas técnicas y psicológicas, más anti-fraude.
    • Headhunting
      Búsqueda ejecutiva a la medida de principio a fin.
    • Nómina + EOR
      Dispersión de nómina y EOR en más de 15 países de LATAM.
  • Precios
  • Empleos

0

114
Vistas
Make ReadableStream sequential

I use the fetchAPI to retrieve my data from the backend as a stream. I decrypt the data chunk by chunk and the concat the content back together for the original file.

I have found that the stream seems to provide data differently each time makling the chunnks different. How can I force the stream to the chunks in the original sequence.

   fetch(myRequest, myInit).then(response => {
    var tmpResult = new Uint8Array();
    const reader = response.body.getReader();
    return new ReadableStream({
      start(controller) {
        return pump();
        function pump() {
          return reader.read().then(({ done, value }) => {
            // When no more data needs to be consumed, close the stream

            if (value) {
                //values here are different in order every time 
                //making my concatenated values different every time

                controller.enqueue(value);

                var decrypted = cryptor.decrypt(value);
                var arrayResponse = decrypted.toArrayBuffer();

                if (arrayResponse) {
                   tmpResult = arrayBufferConcat(tmpResult, arrayResponse);
                }
            }
            // Enqueue the next data chunk into our target stream

            if (done) {
                    if (counter == length) { 
                        callback(obj);        
                    }
              controller.close();
              return;
            }    
            return pump();
          });
        }
      }
    })
  })
about 4 years ago · Juan Pablo Isaza
3 Respuestas
Responde la pregunta

0

The documentation tells us that:

Each chunk is read sequentially and output to the UI, until the stream has finished being read, at which point we return out of the recursive function and print the entire stream to another part of the UI.

I made a test program with node, using node-fetch:

import fetch from 'node-fetch';

const testStreamChunkOrder = async () => {
    return new Promise(async (resolve) => {
        let response = await fetch('https://jsonplaceholder.typicode.com/todos/');
        let stream = response.body;
        let data = '';

        stream.on('readable', () => {
            let chunk;

            while (null !== (chunk = stream.read())) {
                data += chunk;
            }
        })

        stream.on('end', () => {
            resolve(JSON.parse(data).splice(0, 5).map((x) => x.title));
        })
    });
}

(async () => {
    let results = await Promise.all(new Array(10).fill(testStreamChunkOrder()))
    let joined = results.map((r) => r.join(''));
    console.log(`Is every result same: ${joined.every((j) => j.localeCompare(joined[0]) === 0)}`)
})()

This one fetches some random todo-list json and streams it chunk-by-chunk, accumulating the chunks into data. When the stream is done, we parse the full json and take the first 5 elements of the todo-list and keep only the titles, after which we then return the result asynchronously.

This whole process is done 10 times. When we have 10 streamed title-lists, we go through each title-list and join the title names together to form a string. Finally we use .every to see if each of the 10 strings are the same, which means that each json was fetched and streamed in the same order.

So I believe the problems lies somewhere else - the streaming itself is working correctly. While I did use node-fetch instead of the actual Fetch API, I think it is safe to say that the actual Fetch API works as it should.

Also I noticed that you are directly calling response.body.getReader(), but when I looked at the documentation, the body.getReader call is done inside another then statement:

fetch('./tortoise.png')
.then(response => response.body)
.then(body => {
  const reader = body.getReader();

This might not matter, but considering everything else in your code, such as the excessive wrapping and returning of functions, I think your problems could go away just by reading a couple of tutorials on streams and cleaning up the code a bit. And if not, you will still be in a better position to figure out if the problem is in one of your many functions you are unwilling to expose. Asynchronous code's behavior is inherently difficult to debug and lacking information around such code makes it even harder.

about 4 years ago · Juan Pablo Isaza Denunciar

0

I'm assuming you're using the cipher/decipher family of methods in node's crypto library. We can simplify this using streams by first piping the ReadableStream into a decipher TransformStream (a stream that is both readable and writable) via ReadableStream#pipe().

const { createDecipherIv } = require('crypto');
const { createWriteStream } = require('fs');
const { pipeline } = require('stream');

// change these to match your encryption scheme and key retrieval
const algo = 'aes-256-cbc';
const key = 'my5up3r53cr3t';

// put your initialization vector you've determined here
// leave null if you are not (or the algo doesn't support) using an iv
const iv = null;

// creates the decipher TransformStream
const decipher = createDecipherIv(algo, key, iv);

// write plaintext file here
const destFile = createWriteStream('/path/to/destination.ext');

fetch(myRequest, myInit)
    .then(response => response.body)
    .then(body => body.pipe(decipher).pipe(destFile))
    .then(stream => stream.on('end', console.log('done writing file')));

You may also pipe this to be read out in a buffer, pipe to the browser, etc, just be sure to match your algorithm, key, and iv wherever you're defining your cipher/decipher functions.

about 4 years ago · Juan Pablo Isaza Denunciar

0

If we take the pattern in that MDN example seriously, we should use the controller to enqueue the decrypted data (not the still encrypted value), and aggregate the results with the stream returned by the first promise. In other words...

return fetch(myRequest, myInit)
  // Retrieve its body as ReadableStream
  .then(response => {
    const reader = response.body.getReader();
    return new ReadableStream({
      start(controller) {
        return pump();
        function pump() {
          return reader.read().then(({ done, value }) => {
            // When no more data needs to be consumed, close the stream
            if (done) {
              controller.close();
              return;
            }

            // do the computational work on each chunk here and enqueue
            // *the result of that work* on the controller stream...

            const decrypted = cryptor.decrypt(value);
            controller.enqueue(decrypted);
            return pump();
          });
        }
      }
    })
  })
  // Create a new response out of the stream
  .then(stream => new Response(stream))
  // Create an object URL for the response
  .then(response => response.blob())
  .then(blob => {
    const arrayResponse = blob.toArrayBuffer();
    // arrayResponse is the properly sequenced result
    // if the caller wants a promise to resolve to this, just return it
    return arrayResponse;

    // OR... the OP code makes reference to a callback. if that's real, 
    // call the callback with this result
    // callback(arrayResponse);
  })
  .catch(err => console.error(err));
about 4 years ago · Juan Pablo Isaza Denunciar
Responde la pregunta
Encuentra empleos remotos

¡Descubre la nueva forma de encontrar empleo!

Top de empleos
Top categorías de empleo
Empresas
Publicar vacante Precios Comercial
Legal
Términos y condiciones Política de privacidad
© 2026 PeakU Inc. All Rights Reserved.
Andres GPT
Recomiéndame algunas ofertas
Necesito ayuda