Drop support for native Node Readable stream: require manual conversion to Node's Web Streams

karma.conf.cjs

@@ -22,15 +22,15 @@ module.exports = function(config) {
         ],
 
         // list of files / patterns to load in the browser
-        files: [{ pattern: 'test/browser.test.js', watched: false }],
+        files: [{ pattern: 'test/{browser,common}.test.js', watched: false }],
 
         // list of files / patterns to exclude
         exclude: [],
 
         // preprocess matching files before serving them to the browser
         // available preprocessors: https://www.npmjs.com/search?q=keywords:karma-preprocessor
         preprocessors: {
-            'test/browser.test.js': 'webpack'
+            'test/{browser,common}.test.js': 'webpack'
         },
 
         webpack: {},

lib/index.d.ts

@@ -1,16 +1,11 @@
 /// <reference lib="dom" />
-
+import { ReadableStream as NodeWebReadableStream } from 'node:stream/web';
 type Data = Uint8Array | string;
 
-export type WebStream<T extends Data> = ReadableStream<T>;
-
-interface NodeStream<T extends Data> extends AsyncIterable<T> { // copied+simplified version of ReadableStream from @types/node/index.d.ts, which does no support generics
-  readable: boolean; pipe: Function; unpipe: Function; wrap: Function; setEncoding(encoding: string): this; pause(): this; resume(): this;
-  isPaused(): boolean; unshift(chunk: string | Uint8Array): void;
-  read(size?: number): T;
-}
+export type WebStream<T extends Data> = ReadableStream<T>
+export type NodeWebStream<T extends Data> = NodeWebReadableStream<T>;
 
-type Stream<T extends Data> = WebStream<T> | NodeStream<T>;
+type Stream<T extends Data> = WebStream<T> | NodeWebStream<T>;
 type MaybeStream<T extends Data> = T | Stream<T>;
 
 export function readToEnd<T extends Data, R extends any = T>(input: MaybeStream<T>, join?: (chunks: T[]) => R): Promise<R>;

lib/index.js

@@ -1,5 +1,3 @@
 export * from './streams.js';
 
-export { webToNode, nodeToWeb } from './node-conversions/index.js';
-
 export { isStream, isArrayStream, isUint8Array, concatUint8Array } from './util.js';

lib/reader.js

@@ -1,8 +1,12 @@
 import { isUint8Array, isStream, isArrayStream } from './util.js';
-import { nodeToWeb } from './node-conversions/index.js';
 import * as streams from './streams.js';
 
 const doneReadingSet = new WeakSet();
+/**
+ * The external buffer is used to store values that have been peeked or unshifted from the original stream.
+ * Because of how streams are implemented, such values cannot be "put back" in the original stream,
+ * but they need to be returned first when reading from the input again.
+ */
 const externalBuffer = Symbol('externalBuffer');
 
 /**
@@ -25,9 +29,6 @@ function Reader(input) {
     return;
   }
   let streamType = isStream(input);
-  if (streamType === 'node') {
-    input = nodeToWeb(input);
-  }
   if (streamType) {
     const reader = input.getReader();
     this._read = reader.read.bind(reader);

lib/streams.js

@@ -1,5 +1,4 @@
 import { isStream, isArrayStream, isUint8Array, concatUint8Array } from './util.js';
-import { NodeBuffer, nodeToWeb } from './node-conversions/index.js';
 import { Reader, externalBuffer } from './reader.js';
 import { ArrayStream, Writer } from './writer.js';
 
@@ -10,9 +9,6 @@ import { ArrayStream, Writer } from './writer.js';
  */
 function toStream(input) {
   let streamType = isStream(input);
-  if (streamType === 'node') {
-    return nodeToWeb(input);
-  }
   if (streamType) {
     return input;
   }
@@ -25,7 +21,7 @@ function toStream(input) {
 }
 
 /**
- * Convert data to ArrayStream
+ * Convert non-streamed data to ArrayStream; this is a noop if `input` is already a stream.
  * @param {Object} input  data to convert
  * @returns {ArrayStream} Converted data
  */
@@ -58,9 +54,6 @@ function concat(list) {
   if (typeof list[0] === 'string') {
     return list.join('');
   }
-  if (NodeBuffer && NodeBuffer.isBuffer(list[0])) {
-    return NodeBuffer.concat(list);
-  }
   return concatUint8Array(list);
 }
 
@@ -469,9 +462,8 @@ function slice(input, begin=0, end=Infinity) {
   if (input[externalBuffer]) {
     input = concat(input[externalBuffer].concat([input]));
   }
-  if (isUint8Array(input) && !(NodeBuffer && NodeBuffer.isBuffer(input))) {
-    if (end === Infinity) end = input.length;
-    return input.subarray(begin, end);
+  if (isUint8Array(input)) {
+    return input.subarray(begin, end === Infinity ? input.length : end);
   }
   return input.slice(begin, end);
 }

lib/util.js

@@ -1,6 +1,5 @@
 /* eslint-disable no-prototype-builtins */
 import { isArrayStream } from './writer.js';
-import { NodeReadableStream } from './node-conversions/index.js'
 const isNode = typeof globalThis.process === 'object' &&
   typeof globalThis.process.versions === 'object';
 
@@ -16,8 +15,9 @@ function isStream(input) {
   if (globalThis.ReadableStream && globalThis.ReadableStream.prototype.isPrototypeOf(input)) {
     return 'web';
   }
-  if (NodeReadableStream && NodeReadableStream.prototype.isPrototypeOf(input)) {
-    return 'node';
+  // try and detect a node native stream without having to import its class
+  if (input && !(input instanceof globalThis.ReadableStream) && typeof input._read === 'function' && typeof input._readableState === 'object') {
+    throw new Error('Native Node streams are no longer supported: please manually convert the stream to a WebStream, using e.g. `stream.Readable.toWeb`');
   }
   if (input && input.getReader) {
     return 'web-like';

package.json

@@ -13,9 +13,8 @@
   },
   "types": "./lib/index.d.ts",
   "main": "./lib/index.js",
-  "browser": {
-    "buffer": false,
-    "stream": false
+  "engines": {
+    "node": ">= 16.5.0"
   },
   "repository": {
     "type": "git",
@@ -29,7 +28,7 @@
   "homepage": "https://github.com/openpgpjs/web-stream-tools#readme",
   "devDependencies": {
     "@openpgp/jsdoc": "^3.6.11",
-    "@types/node": "^18.11.9",
+    "@types/node": "^20.11.4",
     "@typescript-eslint/eslint-plugin": "^5.59.6",
     "@typescript-eslint/parser": "^5.59.6",
     "chai": "^4.3.7",
@@ -56,7 +55,7 @@
   "scripts": {
     "test-type-definitions": "tsx test/typescript.test.ts && tsx --tsconfig test/tsconfig.es5.json test/typescript.test.ts",
     "test-browser": "karma start karma.conf.cjs",
-    "test-node": "mocha ./test/node.test.js",
+    "test-node": "mocha ./test/node.test.js ./test/common.test.js",
     "lint": "eslint lib test",
     "docs": "jsdoc --configure .jsdocrc.cjs --destination docs --readme README.md lib && printf '%s' 'web-stream-tools.openpgpjs.org' > docs/CNAME",
     "preversion": "rm -rf docs",

test/browser.test.js

@@ -1,14 +1,16 @@
 import { expect } from 'chai';
-import { webToNode, toStream, readToEnd } from '../lib/index.js';
+import { toStream, readToEnd } from '@openpgp/web-stream-tools';
 
 describe('Browser integration tests', () => {
-  it('Node.js specific utils are not defined', () => {
-    expect(webToNode).to.be.null;
-  })
-
-  it('toStream/readToEnd', async () => {
+  it('accepts readable stream', async () => {
     const input = 'chunk';
-    const streamedData = toStream('chunk');
+    const stream = new ReadableStream({
+      start(controller) {
+        controller.enqueue(input);
+        controller.close();
+      }
+    });
+    const streamedData = toStream(stream);
     expect(await readToEnd(streamedData)).to.equal(input);
-  })
+  });
 })

test/common.test.js

@@ -0,0 +1,17 @@
+import { expect } from 'chai';
+import { toStream, readToEnd, slice } from  '@openpgp/web-stream-tools';
+
+describe('Common integration tests', () => {
+  it('toStream/readToEnd', async () => {
+    const input = 'chunk';
+    const streamedData = toStream('chunk');
+    expect(await readToEnd(streamedData)).to.equal(input);
+  });
+
+  it('slice', async () => {
+    const input = 'another chunk';
+    const streamedData = toStream(input);
+    const slicedStream = slice(streamedData, 8);
+    expect(await readToEnd(slicedStream)).to.equal('chunk');
+  })
+})