readable-stream v3, nodejs v16

.github/workflows/build-test.yml

@@ -11,7 +11,7 @@ jobs:
     runs-on: ubuntu-20.04
     strategy:
       matrix:
-        node-version: [12.x, 14.x, 16.x]
+        node-version: [16.x, 18.x, 20.x]
     steps:
       - uses: actions/checkout@v2
       - name: Use Node.js ${{ matrix.node-version }}

.nvmrc

@@ -1 +1 @@
-v12
+v18

README.md

@@ -27,7 +27,7 @@ streamB.pipe(evilAiBrain).pipe(streamB);
 
 ### Setup
 
-- Install [Node.js](https://nodejs.org) version 12
+- Install [Node.js](https://nodejs.org) version 18
   - If you are using [nvm](https://github.com/creationix/nvm#installation) (recommended) running `nvm use` will automatically choose the right node version for you.
 - Install [Yarn v1](https://yarnpkg.com/en/docs/install)
 - Run `yarn setup` to install dependencies and run any requried post-install scripts

package.json

@@ -9,7 +9,7 @@
     "url": "https://github.com/MetaMask/object-multiplex.git"
   },
   "engines": {
-    "node": ">=12.0.0"
+    "node": "^16.20 || ^18.16 || >=20"
   },
   "files": [
     "dist/"
@@ -28,20 +28,18 @@
   "author": "",
   "license": "ISC",
   "dependencies": {
-    "end-of-stream": "^1.4.4",
     "once": "^1.4.0",
-    "readable-stream": "^2.3.3"
+    "readable-stream": "^3.6.2"
   },
   "devDependencies": {
     "@lavamoat/allow-scripts": "^1.0.6",
     "@metamask/auto-changelog": "^2.3.0",
     "@metamask/eslint-config": "^6.0.0",
     "@metamask/eslint-config-nodejs": "^6.0.0",
     "@metamask/eslint-config-typescript": "^6.0.0",
-    "@types/end-of-stream": "^1.4.0",
-    "@types/node": "^14.14.9",
+    "@types/node": "^16",
     "@types/once": "^1.4.0",
-    "@types/readable-stream": "^2.3.9",
+    "@types/readable-stream": "4.0.0",
     "@typescript-eslint/eslint-plugin": "^4.26.0",
     "@typescript-eslint/parser": "^4.26.0",
     "eslint": "^7.27.0",
@@ -50,7 +48,6 @@
     "eslint-plugin-node": "^11.1.0",
     "eslint-plugin-prettier": "^3.4.0",
     "prettier": "^2.3.0",
-    "pump": "^1.0.2",
     "rimraf": "^3.0.2",
     "tape": "^4.8.0",
     "typescript": "^4.1.2"

src/ObjectMultiplex.ts

@@ -1,5 +1,4 @@
-import { Duplex } from 'readable-stream';
-import eos from 'end-of-stream';
+import { Duplex, finished } from 'readable-stream';
 import once from 'once';
 import { Substream } from './Substream';
 
@@ -112,6 +111,6 @@ function anyStreamEnd(
   _cb: (error?: Error | null) => void,
 ) {
   const cb = once(_cb);
-  eos(stream, { readable: false }, cb);
-  eos(stream, { writable: false }, cb);
+  finished(stream, { readable: false }, cb);
+  finished(stream, { writable: false }, cb);
 }

src/readable-stream.d.ts

@@ -0,0 +1,19 @@
+// import indirectly used for module augmentation to extend @types/readable-stream with missing exports
+// eslint-disable-next-line @typescript-eslint/no-unused-vars
+import * as readableStream from 'readable-stream';
+
+interface Options {
+  readable?: boolean | undefined;
+  writable?: boolean | undefined;
+  error?: boolean | undefined;
+}
+type Stream = NodeJS.ReadableStream | NodeJS.WritableStream;
+type Callback = (error?: Error | null) => void;
+
+declare module 'readable-stream' {
+  function finished(
+    stream: Stream,
+    options: Options,
+    callback?: Callback,
+  ): () => void;
+}

test/index.js

@@ -1,7 +1,5 @@
+const { PassThrough, Transform, pipeline } = require('readable-stream');
 const test = require('tape');
-const pump = require('pump');
-const { PassThrough, Transform } = require('readable-stream');
-const endOfStream = require('end-of-stream');
 // eslint-disable-next-line import/no-unresolved
 const ObjMultiplex = require('../dist');
 
@@ -20,7 +18,7 @@ test('basic - string', (t) => {
   inStream.write('wuurl');
 
   // simulate disconnect
-  setTimeout(() => inTransport.destroy());
+  setImmediate(() => inTransport.end(null, () => undefined));
 });
 
 test('basic - obj', (t) => {
@@ -41,7 +39,7 @@ test('basic - obj', (t) => {
   inStream.write({ message: 'wuurl' });
 
   // simulate disconnect
-  setTimeout(() => inTransport.destroy());
+  setImmediate(() => inTransport.end(null, () => undefined));
 });
 
 test('roundtrip', (t) => {
@@ -56,7 +54,7 @@ test('roundtrip', (t) => {
     },
   });
 
-  pump(outStream, doubler, outStream);
+  pipeline(outStream, doubler, outStream, () => undefined);
 
   bufferToEnd(inStream, (err, results) => {
     t.error(err, 'should not error');
@@ -68,7 +66,7 @@ test('roundtrip', (t) => {
   inStream.write(12);
 
   // simulate disconnect
-  setTimeout(() => outTransport.destroy(), 100);
+  setTimeout(() => outTransport.end(), 100);
 });
 
 test('error on createStream if destroyed', (t) => {
@@ -106,7 +104,7 @@ function basicTestSetup() {
   const inStream = inMux.createStream('hello');
   const outStream = outMux.createStream('hello');
 
-  pump(inMux, inTransport, outMux, outTransport, inMux);
+  pipeline(inMux, inTransport, outMux, outTransport, inMux, () => undefined);
 
   return {
     inTransport,
@@ -120,6 +118,18 @@ function basicTestSetup() {
 
 function bufferToEnd(stream, callback) {
   const results = [];
-  endOfStream(stream, (err) => callback(err, results));
-  stream.on('data', (chunk) => results.push(chunk));
+  let flushed = false;
+  function onFinish(err) {
+    if (flushed) {
+      return;
+    }
+    flushed = true;
+    callback(err, results);
+  }
+  // cleanup of stream should be called at end of each stream
+  // this ensures that
+  stream.prependListener('close', onFinish);
+  stream.on('data', (chunk) => {
+    results.push(chunk);
+  });
 }