feat: drop the stream dependency from browser builds

docs/getting_started.md

@@ -63,6 +63,12 @@ a document unless the document is created with `{ font: null }`.
       iframe.src = url;
     });
 
+The browser build does not depend on Node's `stream` module. A document is still a
+readable stream, but only the parts a PDF document needs are implemented: `on`,
+`once`, `off`, `emit`, `pipe` and async iteration. `read`, `setEncoding`,
+`destroy`, `stream.pipeline` and the `readable`, `error` and `close` events are
+available in Node only.
+
 You can see an interactive in-browser demo of PDFKit [here](http://pdfkit.org/demo/browser.html).
 
 ## Document options

lib/document.js

@@ -3,7 +3,7 @@ PDFDocument - represents an entire PDF document
 By Devon Govett
 */
 
-import stream from 'stream';
+import Readable from '#stream';
 import PDFObject from './object';
 import PDFReference from './reference';
 import PDFPage from './page';
@@ -25,7 +25,7 @@ import TableMixin from './mixins/table';
 import MetadataMixin from './mixins/metadata';
 import { fromBinaryString } from './binary';
 
-class PDFDocument extends stream.Readable {
+class PDFDocument extends Readable {
   constructor(options = {}) {
     super(options);
     this.options = options;

lib/event_emitter.js

@@ -0,0 +1,36 @@
+class EventEmitter {
+  constructor() {
+    this._listeners = Object.create(null);
+  }
+
+  on(event, listener) {
+    (this._listeners[event] || (this._listeners[event] = [])).push(listener);
+    return this;
+  }
+
+  once(event, listener) {
+    const wrapper = (...args) => {
+      this.off(event, wrapper);
+      listener(...args);
+    };
+    return this.on(event, wrapper);
+  }
+
+  off(event, listener) {
+    const listeners = this._listeners[event];
+    if (listeners) {
+      const index = listeners.indexOf(listener);
+      if (index !== -1) listeners.splice(index, 1);
+    }
+    return this;
+  }
+
+  emit(event, ...args) {
+    const listeners = this._listeners[event];
+    if (!listeners) return false;
+    for (const listener of listeners.slice()) listener(...args);
+    return listeners.length > 0;
+  }
+}
+
+export default EventEmitter;

lib/line_wrapper.js

@@ -1,12 +1,13 @@
 import LineBreaker from 'linebreak';
+import EventEmitter from './event_emitter';
 import { PDFNumber } from './utils';
 
 const SOFT_HYPHEN = '\u00AD';
 const HYPHEN = '-';
 
-class LineWrapper {
+class LineWrapper extends EventEmitter {
   constructor(document, options) {
-    this._listeners = Object.create(null);
+    super();
     this.document = document;
     this.horizontalScaling = options.horizontalScaling || 100;
     this.indent = ((options.indent || 0) * this.horizontalScaling) / 100;
@@ -84,25 +85,6 @@ class LineWrapper {
     });
   }
 
-  on(event, listener) {
-    (this._listeners[event] || (this._listeners[event] = [])).push(listener);
-  }
-
-  once(event, listener) {
-    const wrapper = (...args) => {
-      const listeners = this._listeners[event];
-      listeners.splice(listeners.indexOf(wrapper), 1);
-      listener(...args);
-    };
-    this.on(event, wrapper);
-  }
-
-  emit(event, ...args) {
-    const listeners = this._listeners[event];
-    if (!listeners) return;
-    for (const listener of listeners.slice()) listener(...args);
-  }
-
   wordWidth(word) {
     return PDFNumber(
       this.document.widthOfString(word, this) +

lib/stream/browser.js

@@ -0,0 +1,111 @@
+import EventEmitter from '../event_emitter';
+
+/**
+ * A readable stream with just enough of the Node API for a PDF document:
+ * `on('data')`, `on('end')`, `pipe()` and async iteration.
+ *
+ * ponytail: no `read()`, `setEncoding()`, `destroy()` or `error`/`close`
+ * events. The document is built in memory and never applies backpressure to
+ * its own producer, so the only backpressure that matters is the `drain` of
+ * the pipe destination.
+ */
+class Readable extends EventEmitter {
+  constructor() {
+    super();
+    this._buffer = [];
+    this._flowing = false;
+    this._scheduled = false;
+    this._finished = false;
+    this._endEmitted = false;
+  }
+
+  on(event, listener) {
+    super.on(event, listener);
+    if (event === 'data') this.resume();
+    return this;
+  }
+
+  push(chunk) {
+    if (chunk === null) {
+      this._finished = true;
+    } else {
+      this._buffer.push(chunk);
+    }
+    this._schedule();
+    return true;
+  }
+
+  resume() {
+    this._flowing = true;
+    this._schedule();
+    return this;
+  }
+
+  pause() {
+    this._flowing = false;
+    return this;
+  }
+
+  pipe(destination) {
+    this.on('data', (chunk) => {
+      if (destination.write(chunk) === false) {
+        this.pause();
+        destination.once('drain', () => this.resume());
+      }
+    });
+    this.on('end', () => destination.end());
+    return destination;
+  }
+
+  // Chunks pushed before a listener is attached are replayed on a microtask so
+  // that `on('data')` followed by `on('end')` sees both events.
+  _schedule() {
+    if (this._scheduled || !this._flowing) return;
+    this._scheduled = true;
+    queueMicrotask(() => {
+      this._scheduled = false;
+      this._drain();
+    });
+  }
+
+  _drain() {
+    while (this._flowing && this._buffer.length > 0) {
+      this.emit('data', this._buffer.shift());
+    }
+
+    if (this._flowing && this._finished && !this._endEmitted) {
+      this._endEmitted = true;
+      this.emit('end');
+    }
+  }
+
+  async *[Symbol.asyncIterator]() {
+    const chunks = [];
+    let ended = false;
+    let notify = null;
+    const wake = () => {
+      const resolve = notify;
+      notify = null;
+      if (resolve) resolve();
+    };
+
+    this.on('data', (chunk) => {
+      chunks.push(chunk);
+      wake();
+    });
+    this.on('end', () => {
+      ended = true;
+      wake();
+    });
+
+    for (;;) {
+      while (chunks.length > 0) yield chunks.shift();
+      if (ended) return;
+      await new Promise((resolve) => {
+        notify = resolve;
+      });
+    }
+  }
+}
+
+export default Readable;

lib/stream/node.js

@@ -0,0 +1,3 @@
+import stream from 'stream';
+
+export default stream.Readable;

package.json

@@ -146,6 +146,10 @@
       "node": "./lib/zlib/node.js",
       "default": "./lib/zlib/browser.js"
     },
+    "#stream": {
+      "node": "./lib/stream/node.js",
+      "default": "./lib/stream/browser.js"
+    },
     "#standard-fonts/*": {
       "require": "./js/standard-fonts/*.cjs",
       "default": "./js/standard-fonts/*.mjs"

tests/unit/stream.spec.js

@@ -0,0 +1,115 @@
+import { vi } from 'vitest';
+import Readable from '../../lib/stream/browser';
+
+const collect = (stream) =>
+  new Promise((resolve) => {
+    const chunks = [];
+    stream.on('data', (chunk) => chunks.push(chunk));
+    stream.on('end', () => resolve(chunks));
+  });
+
+describe('browser stream', function () {
+  test('replays chunks pushed before a listener is attached', async function () {
+    const stream = new Readable();
+    stream.push('a');
+    stream.push('b');
+    stream.push(null);
+
+    expect(await collect(stream)).toEqual(['a', 'b']);
+  });
+
+  test('emits chunks pushed after a listener is attached', async function () {
+    const stream = new Readable();
+    const collected = collect(stream);
+
+    stream.push('a');
+    stream.push(null);
+
+    expect(await collected).toEqual(['a']);
+  });
+
+  test('off removes a listener', function () {
+    const stream = new Readable();
+    const listener = vi.fn();
+
+    stream.on('pageAdded', listener);
+    stream.off('pageAdded', listener);
+    stream.emit('pageAdded');
+
+    expect(listener).not.toHaveBeenCalled();
+  });
+
+  test('once fires a single time', function () {
+    const stream = new Readable();
+    const listener = vi.fn();
+
+    stream.once('pageAdded', listener);
+    stream.emit('pageAdded');
+    stream.emit('pageAdded');
+
+    expect(listener).toHaveBeenCalledTimes(1);
+  });
+
+  test('pipe writes every chunk and ends the destination', async function () {
+    const stream = new Readable();
+    const written = [];
+    const destination = {
+      write: (chunk) => written.push(chunk),
+      end: vi.fn(),
+    };
+
+    stream.pipe(destination);
+    stream.push('a');
+    stream.push('b');
+    stream.push(null);
+
+    await new Promise((resolve) => stream.on('end', resolve));
+
+    expect(written).toEqual(['a', 'b']);
+    expect(destination.end).toHaveBeenCalled();
+  });
+
+  test('pipe waits for drain when the destination is full', async function () {
+    const stream = new Readable();
+    const written = [];
+    let drain;
+    const destination = {
+      write: (chunk) => {
+        written.push(chunk);
+        return written.length !== 2;
+      },
+      end: vi.fn(),
+      once: (event, listener) => {
+        if (event === 'drain') drain = listener;
+      },
+    };
+
+    stream.pipe(destination);
+    stream.push('a');
+    stream.push('b');
+    stream.push('c');
+    stream.push(null);
+
+    await Promise.resolve();
+    expect(written).toEqual(['a', 'b']);
+    expect(destination.end).not.toHaveBeenCalled();
+
+    drain();
+    await new Promise((resolve) => stream.on('end', resolve));
+
+    expect(written).toEqual(['a', 'b', 'c']);
+    expect(destination.end).toHaveBeenCalled();
+  });
+
+  test('supports async iteration', async function () {
+    const stream = new Readable();
+    stream.push('a');
+    stream.push('b');
+    stream.push(null);
+
+    const chunks = [];
+    for await (const chunk of stream) chunks.push(chunk);
+
+    expect(chunks).toEqual(['a', 'b']);
+  });
+});