Add stream example

examples/stream/index.js

@@ -0,0 +1,31 @@
+'use strict';
+
+const Piscina = require('../../dist/src');
+const { resolve } = require('path');
+const { MessageChannel } = require('worker_threads');
+const PortDuplex = require('./port_duplex');
+const { pipeline, Writable } = require('stream');
+
+const piscina = new Piscina({
+  filename: resolve(__dirname, 'worker.js')
+});
+
+class W extends Writable {
+  length = 0;
+  _write (chunk, encoding, callback) {
+    this.length += chunk.length;
+    callback();
+  }
+};
+
+(async function () {
+  const channel = new MessageChannel();
+  const duplex = new PortDuplex(channel.port2, { writable: false });
+  const w = new W();
+
+  duplex.on('close', () => channel.port2.close());
+
+  pipeline(duplex, w, () => console.log(w.length));
+
+  await piscina.runTask({ port: channel.port1 }, [channel.port1]);
+})();

examples/stream/port_duplex.js

@@ -0,0 +1,58 @@
+'use strict';
+
+const { Duplex } = require('stream');
+
+class PortDuplex extends Duplex {
+  #port = undefined;
+
+  constructor (port, options) {
+    const {
+      readable = true,
+      writable = true
+    } = { ...options };
+    if (typeof readable !== 'boolean') {
+      throw new TypeError('readable must be a boolean');
+    }
+    if (typeof writable !== 'boolean') {
+      throw new TypeError('writable must be a boolean');
+    }
+    super({ autoDestroy: true, readable, writable });
+    this.#port = port;
+    this.#port.onmessage = PortDuplex.#onmessage.bind(this);
+  }
+
+  _write (chunk, encoding, callback) {
+    // Chunk should always be a buffer here
+    const temp = new Uint8Array(chunk);
+    // TODO(@jasnell): This will need backpressure implemented
+    this.#port.postMessage(temp, [temp.buffer]);
+    callback();
+  }
+
+  _read () {
+    // Do nothing here. A more complete example would
+    // implement proper read/pause behavior.
+  }
+
+  _destroy (err, callback) {
+    if (err) {
+      this.#port.close();
+      callback(err);
+      return;
+    }
+    if (this.writableEnded) {
+      this.#port.postMessage(null);
+    }
+    callback();
+  }
+
+  static #onmessage = function ({ data }) {
+    if (data != null) {
+      this.push(data);
+    } else {
+      this.push(null);
+    }
+  };
+}
+
+module.exports = PortDuplex;

examples/stream/worker.js

@@ -0,0 +1,20 @@
+'use strict';
+
+const PortDuplex = require('./port_duplex');
+const { createReadStream } = require('fs');
+const { pipeline } = require('stream');
+const { createGzip } = require('zlib');
+
+module.exports = ({ port }) => {
+  return new Promise((resolve, reject) => {
+    const input = createReadStream(__filename);
+    const stream = new PortDuplex(port, { readable: false });
+    pipeline(input, createGzip(), stream, (err) => {
+      if (err) {
+        reject(err);
+        return;
+      }
+      resolve();
+    });
+  });
+};