From 88658b9fd8a97a6745bf5a8bb6680d7f8d8de9ea Mon Sep 17 00:00:00 2001
From: Le Roux Bodenstein <lerouxb@gmail.com>
Date: Thu, 20 Jul 2023 16:54:20 +0100
Subject: [PATCH] pass errors in streams to the callback (#201)

---
 src/stream.ts | 34 +++++++++++++++++++++-------------
 1 file changed, 21 insertions(+), 13 deletions(-)

diff --git a/src/stream.ts b/src/stream.ts
index 10a7a6d..5b47741 100644
--- a/src/stream.ts
+++ b/src/stream.ts
@@ -23,24 +23,32 @@ export class ParseStream extends Duplex {
     this.analyzer = new SchemaAnalyzer(options);
   }
 
-  _write(obj: Document, enc: unknown, cb: () => void) {
-    this.analyzer.analyzeDoc(obj);
-    this.emit('progress', obj);
-    cb();
+  _write(obj: Document, enc: unknown, cb: (err?: any) => void) {
+    try {
+      this.analyzer.analyzeDoc(obj);
+      this.emit('progress', obj);
+      cb();
+    } catch (err: any) {
+      cb(err);
+    }
   }
 
   _read() {}
 
-  _final(cb: () => void) {
-    if (this.options.schemaPaths) {
-      this.push(this.analyzer.getSchemaPaths());
-    } else if (this.options.simplifiedSchema) {
-      this.push(this.analyzer.getSimplifiedSchema());
-    } else {
-      this.push(this.analyzer.getResult());
+  _final(cb: (err?: any) => void) {
+    try {
+      if (this.options.schemaPaths) {
+        this.push(this.analyzer.getSchemaPaths());
+      } else if (this.options.simplifiedSchema) {
+        this.push(this.analyzer.getSimplifiedSchema());
+      } else {
+        this.push(this.analyzer.getResult());
+      }
+      this.push(null);
+      cb();
+    } catch (err: any) {
+      cb(err);
     }
-    this.push(null);
-    cb();
   }
 }