Fix _destroy implementation (#19)

Also, add tests for Node.js stream conversions

Co-authored-by: larabr <[email protected]>

.github/workflows/node.js.yml

@@ -0,0 +1,29 @@
+# This workflow will do a clean install of node dependencies, build the source code and run tests across different versions of node
+# For more information see: https://docs.github.com/en/actions/automating-builds-and-tests/building-and-testing-nodejs
+
+name: Test on Node.js
+
+on:
+  push:
+    branches: [openpgpjs_v5]
+  pull_request:
+    branches: [openpgpjs_v5]
+
+jobs:
+  build:
+    name: ${{ matrix.node-version }}
+
+    runs-on: ubuntu-latest
+
+    strategy:
+      matrix:
+        node-version: [14.x, 16.x, 18.x]
+        # See supported Node.js release schedule at https://nodejs.org/en/about/releases/
+
+    steps:
+      - uses: actions/checkout@v3
+      - uses: actions/setup-node@v3
+        with:
+          node-version: ${{ matrix.node-version }}
+      - run: npm ci
+      - run: npm test

.github/workflows/test-type-definitions.yml

@@ -2,9 +2,9 @@ name: Types
 
 on:
   push:
-    branches: [main]
+    branches: [main, openpgpjs_v5]
   pull_request:
-    branches: [main]
+    branches: [main, openpgpjs_v5]
 
 jobs:
   lint:
@@ -13,9 +13,9 @@ jobs:
     runs-on: ubuntu-latest
 
     steps:
-    - uses: actions/checkout@v2
-    - uses: actions/setup-node@v1
+    - uses: actions/checkout@v3
+    - uses: actions/setup-node@v3
       with:
-        node-version: '15'
+        node-version: '16'
     - run: npm ci
     - run: npm run test-type-definitions

lib/node-conversions.js

@@ -62,23 +62,22 @@ if (NodeReadableStream) {
     async _read(size) {
       try {
         while (true) {
-          const { done, value } = await this._reader.read()
+          const { done, value } = await this._reader.read();
           if (done) {
             this.push(null);
             break;
           }
-          if (!this.push(value) || this._cancelling) {
-            this._reading = false;
+          if (!this.push(value)) {
             break;
           }
         }
-      } catch(e) {
-        this.emit('error', e);
+      } catch (e) {
+        this.destroy(e);
       }
     }
 
-    _destroy(reason) {
-      this._reader.cancel(reason);
+    async _destroy(error, callback) {
+      this._reader.cancel(error).then(callback, callback);
     }
   }
 

lib/reader.js

@@ -20,7 +20,7 @@ function Reader(input) {
     const reader = input.getReader();
     this._read = reader.read.bind(reader);
     this._releaseLock = () => {};
-    this._cancel = () => {};
+    this._cancel = async () => {};
     return;
   }
   let streamType = streams.isStream(input);