Should transform streams with async generators return return all values concatenated

[debadree25]

Take the example of the following code:

import { pipeline, Transform, Readable } from 'stream';

const values = [];
const rs = new Readable({
  start() {},
});

const t = new Transform({
  transform(chunk, encoding, callback) {
    callback(null, chunk?.toString().toUpperCase());
  },
});

pipeline(rs, t, async function(source) {
  for await (const chunk of source) {
    values.push(chunk?.toString());
  }
}, (err) => {
  if (err) throw err;
  console.log(values);
});

rs.push('a');
rs.push('b');
rs.push('c');
rs.push(null);

The output of the following is

[ 'ABC' ]

But I expect the output to be ['A','B','C'], I not sure if this is an issue or flaw in my understanding hence asking here if anyone can guide it would be very helpful!

[arpit9616]

Your understanding is correct, and the behavior you're seeing is due to the way you've implemented the stream processing in your code.

In your code, you're using the pipeline function to process the data from the rs (readable) stream through the t (transform) stream and then into the final destination. However, the way you've implemented the transform stream t is causing the result you're observing.

Here's the key part of your transform stream implementation:

const t = new Transform({
  transform(chunk, encoding, callback) {
    callback(null, chunk?.toString().toUpperCase());
  },
});

In this implementation, you're transforming each chunk to its uppercase representation and then passing it to the callback. This effectively combines the chunks together, as you're pushing the combined, uppercase chunk downstream.

To achieve the output you expect (['A','B','C']), you should modify your transform stream to emit each uppercase character separately, like this:

const t = new Transform({
  transform(chunk, encoding, callback) {
    const upperCaseChunk = chunk?.toString().toUpperCase();
    for (const char of upperCaseChunk) {
      callback(null, char);
    }
  },
});

With this change, each character will be emitted individually by the transform stream, and your output will be ['A','B','C'] as you expected.