Message Port

Worker threads can receive MessagePort objects, enabling direct communication channels with the main thread. This feature is useful for scenarios that require continuous communication or transfer of large data sets between threads.

The example below shows how to use MessagePort with Piscina.

Javascript

index.js

'use strict';

const Piscina = require('piscina');
const { resolve } = require('path');
const { MessageChannel } = require('worker_threads');

const piscina = new Piscina({
  filename: resolve(__dirname, 'worker.js')
});

(async function () {
  const channel = new MessageChannel();
  channel.port2.on('message', (message) => {
    console.log(message);
    channel.port2.close();
  });
  await piscina.run({ port: channel.port1 }, { transferList: [channel.port1] });
})();

Typescript

index.ts

import Piscina from "piscina";
import { resolve } from "path";
import { filename } from "./worker";
import { TransferListItem } from "piscina/dist/types";

const piscina = new Piscina({
  filename: resolve(__dirname, "./workerWrapper.js"),
  workerData: { fullpath: filename },
});

interface CustomMessagePort extends MessagePort {
  on(event: string, listener: (message: any) => void): void;
}
(async function () {
  const channel = new MessageChannel();
  (channel.port2 as CustomMessagePort).on('message', (message: string) => {
    console.log(message);
    channel.port2.close();
  });
  await piscina.run({ port: channel.port1 }, { transferList:[channel.port1] as TransferListItem});
})();

workerWrapper.js

const { workerData } = require('worker_threads');

if (workerData.fullpath.endsWith(".ts")) {
require("ts-node").register();
}
module.exports = require(workerData.fullpath);

The worker file receives the MessagePort and uses it to send a message back to the main thread:

Javascript

worker.js

'use strict';

module.exports = ({ port }) => {
  port.postMessage('hello from the worker pool');
};

Typescript

worker.ts

import { MessagePort } from 'worker_threads';

interface WorkerInput {
  port: MessagePort;
}

export default ({ port }: WorkerInput): void => {
  port.postMessage('hello from the worker pool');
};

You can also check out this example on github.