How to handle errors inside a NodeJS stream?

Question

I'm trying to find a way to send a NestJS HTTP Error within a readable stream when the data is not valid:

stream
   .pipe(csv())
   .on('data', (data) => {
     try {
       spreadsheetValidationSchema.parse(data);
       spreadsheet.push(data);
     } catch (err) {
       throw new UnprocessableEntityException();;
     }
   })
   .on('end', () => {
     console.log(spreadsheet);
   });

The issue here is that, when throwing that error inside the stream itself, it just get warned by the terminal, instead of actually returning the errors.

throw new UnprocessableEntityException();
                ^

Any ideas are appreciated.

Answer 1

I just got a workaround. I wrapped the stream pipe in a Promise.

await new Promise((resolve, reject) => {
  stream
    .pipe(csv())
    .on('data', (data) => {
      try {
        spreadsheetValidationSchema.parse(data);
        spreadsheet.push(data);
      } catch (err) {
        reject(new UnprocessableEntityException());
        stream.destroy();
      }
    })
    .on('end', () => {
      resolve('success');
    });
});