From 33eddeac3b97036848c20bf4b4fc12402e9ebaa1 Mon Sep 17 00:00:00 2001
From: David Whittington <djwhitt@gmail.com>
Date: Tue, 25 Jul 2023 14:57:30 -0500
Subject: [PATCH] fix(bundles ans-104): use pipeline to improve error handling
 PE-4212

Switching from 'pipe' to 'pipeline' because 'pipeline' has more
consistent and comprehensive error handling. With 'pipe' there was some
unhandled case that was causing promises to not get resolved/rejected
reliably when streams errored. No doubt that's resolvable by setting up
the correct handlers, but 'pipeline' is more fool proof.
---
 src/lib/ans-104.ts | 34 ++++++++++++++++++++--------------
 1 file changed, 20 insertions(+), 14 deletions(-)

diff --git a/src/lib/ans-104.ts b/src/lib/ans-104.ts
index fcc33128..ca3856a9 100644
--- a/src/lib/ans-104.ts
+++ b/src/lib/ans-104.ts
@@ -20,6 +20,7 @@ import * as EventEmitter from 'node:events';
 import fs from 'node:fs';
 import fsPromises from 'node:fs/promises';
 import path from 'node:path';
+import { pipeline } from 'node:stream';
 import {
   Worker,
   isMainThread,
@@ -205,20 +206,25 @@ export class Ans104Parser {
             `${parentId}`,
           );
           const writeStream = fs.createWriteStream(bundlePath);
-          data.stream.pipe(writeStream);
-          writeStream.on('error', (error) => {
-            log.error('Error writing ANS-104 bundle stream', error);
-            reject(error);
-          });
-          writeStream.on('finish', async () => {
-            log.info('Parsing ANS-104 bundle stream...');
-            this.worker.postMessage({
-              rootTxId,
-              parentId,
-              parentIndex,
-              bundlePath,
-            });
-          });
+          pipeline(
+            data.stream,
+            writeStream,
+            (error) => {
+              if (error !== undefined) {
+                this.unbundlePromiseReject?.(error);
+                this.resetUnbundlePromise();
+                log.error('Error writing ANS-104 bundle stream', error);
+              } else {
+                log.info('Parsing ANS-104 bundle stream...');
+                this.worker.postMessage({
+                  rootTxId,
+                  parentId,
+                  parentIndex,
+                  bundlePath,
+                });
+              }
+            }
+          );
         } catch (error) {
           reject(error);
         }