How do you catch stream transform errors?

Question

This is the readable stream native definition

// This is the part where you do stuff!
// override this function in implementation classes.
// 'chunk' is an input chunk.
//
// Call `push(newChunk)` to pass along transformed output
// to the readable side.  You may call 'push' zero or more times.
//
// Call `cb(err)` when you are done with this chunk.  If you pass
// an error, then that'll put the hurt on the whole operation.  If you
// never call cb(), then you'll never get another chunk.
Transform.prototype._transform = function (chunk, encoding, cb) {
  throw new Error('_transform() is not implemented');
};

So in your own definition when you want to pass an error you call cb(new Error('...'))

But when I do that, how can I catch these if the stream is piped ?

I mean catch them the normal way, without using the process.on('uncaughtException') event

Answer 1

Error handling in streams should be handled individually for each readable/writable by adding a .on("error", cb) handler.

However, node also provide a utility function that can promisify error handling in streams. The utility function takes care of destroying the streams too, if an error is thrown:

import util from "util";
import stream from "stream";

await util.promisify(stream.pipeline)(
   new stream.Readable(), // a source
   new stream.Transform({
     transform: (chunk, encoding, callback) => {}
   }),
   new stream.Writable() // a sink
);