adds worker-side on(message) events

src/Pool.js

@@ -204,6 +204,9 @@ Pool.prototype._next = function () {
       // get the first task from the queue
       var me = this;
       var task = this.tasks.shift();
+
+      // assign worker to promise so it can be referred to by the handler
+      task.resolver.promise.worker = worker;
 
       // check if the task is still pending (and not cancelled -> promise rejected)
       if (task.resolver.promise.pending) {

src/Promise.js

@@ -22,6 +22,9 @@ function Promise(handler, parent) {
   var _onSuccess = [];
   var _onFail = [];
 
+  // empty worker
+  this.worker = null;
+
   // status
   this.resolved = false;
   this.rejected = false;
@@ -141,6 +144,22 @@ function Promise(handler, parent) {
     return me;
   };
 
+/**
+   * Calls a defined worker method with optional parameters
+   * @param {string} methodName Name of the method to be called as defined in the workerpool.worker() function
+   * @param {Array} [methodParameters] Array with optional parameters
+   */
+this.emit = function (methodName, methodParameters) {
+  if (this.worker?.worker && methodName) {
+    const payload = {
+      eventName: methodName,
+      eventParams: methodParameters,
+    };
+    this.worker.worker.postMessage(payload);
+  }
+  return this;
+};
+
   // attach handler passing the resolve and reject functions
   handler(function (result) {
     _resolve(result);

src/index.js

@@ -16,9 +16,10 @@ exports.pool = function pool(script, options) {
  * Create a worker and optionally register a set of methods to the worker.
  * @param {Object} [methods]
  */
-exports.worker = function worker(methods) {
+exports.worker = function worker(methods, events) {
   var worker = require('./worker');
   worker.add(methods);
+  if (events) worker.addOnEvents(events);
 };
 
 /**
@@ -38,4 +39,4 @@ exports.Promise = require('./Promise');
 
 exports.platform = environment.platform;
 exports.isMainThread = environment.isMainThread;
-exports.cpus = environment.cpus;
+exports.cpus = environment.cpus;

src/worker.js

@@ -120,7 +120,7 @@ worker.on('message', function (request) {
 
     if (method) {
       currentRequestId = request.id;
-      
+
       // execute the function
       var result = method.apply(method, request.params);
 
@@ -196,7 +196,21 @@ worker.emit = function (payload) {
   }
 };
 
+worker.onEvents = function(events) {
+  for (const eventKey of Object.keys(events)) {
+    worker.on('message', function({eventName, eventParams}) {
+      if (eventName === eventKey) {
+        if (eventParams && !Array.isArray(eventParams)) {
+          eventParams = [eventParams];
+        }
+        events[eventKey].apply(worker, eventParams);
+      }
+    });
+  }
+}
+
 if (typeof exports !== 'undefined') {
   exports.add = worker.register;
   exports.emit = worker.emit;
+  exports.addOnEvents = worker.onEvents;
 }

test/CallMethod.test.js

@@ -0,0 +1,42 @@
+var assert = require('assert');
+var Pool = require('../src/Pool');
+
+it('creates a worker pool with event callbacks', function (done) {
+  const pool = new Pool(__dirname + "/workers/waitForMessage.js");
+
+  const handler = pool.exec('fibonacci', [15]);
+
+  handler
+  .then(function (result) {
+    assert.strictEqual(result, 610);
+  })
+  .catch(function (err) {
+    console.error(err);
+  })
+  .then(function () {
+    pool.terminate(); // terminate all workers when done
+    done();
+  })
+  .catch(done);
+});
+
+it('creates a worker pool and sends a message', function (done) {
+  const pool = new Pool(__dirname + "/workers/waitForMessage.js");
+
+  const handler = pool.exec('fibonacci', [15]);
+
+  handler
+  .then(function (result) {
+    assert.strictEqual(result, 610);
+  })
+  .catch(function (err) {
+    assert.strictEqual(err.toString().includes('exitCode: `99`'), true);
+  })
+  .then(function () {
+    pool.terminate(); // terminate all workers when done
+    done();
+  })
+  .catch(done);
+
+  handler.emit('killme', 99); // forces the worker to exit with exitcode 99
+});

test/workers/waitForMessage.js

@@ -0,0 +1,18 @@
+// a simple worker that just sits there and receives messages
+
+// load workerpool
+var workerpool = require('../../');
+
+function fibonacci(n) {
+  if (n < 2) return n;
+  return fibonacci(n - 2) + fibonacci(n - 1);
+}
+
+// create a worker and register public functions
+workerpool.worker({
+  fibonacci: fibonacci,
+}, {
+  killme: function (payload) {
+    process.exit(payload);
+  }
+});