Feat/abort listener

src/index.js

@@ -55,6 +55,17 @@ exports.Promise = Promise;
 
 exports.Transfer = require('./transfer');
 
+/**
+ * Registers a handler for worker timeout / cancelation
+ * which will be run to cleanup a given worker.
+ * Allowing cleanup to resolve with out worker termination
+ * @param {() => Promise<void>} listener 
+ */
+exports.addAbortListener = function (listener) {
+ var worker = require('./worker');
+ worker.addAbortListener(listener);
+}
+
 exports.platform = platform;
 exports.isMainThread = isMainThread;
 exports.cpus = cpus;

src/worker.js

@@ -112,6 +112,11 @@ worker.methods.methods = function methods() {
  */
 worker.terminationHandler = undefined;
 
+/**
+ * abort listener for handling worker timeout / canceling
+ */
+worker.onAbort = undefined;
+
 /**
  * Cleanup and exit the worker.
  * @param {Number} code 
@@ -121,16 +126,28 @@ worker.cleanupAndExit = function(code) {
  var _exit = function() {
  worker.exit(code);
  }
-
- if(!worker.terminationHandler) {
- return _exit();
+ var _abort = function() {
+ worker.onAbort = undefined;
  }
 
- var result = worker.terminationHandler(code);
- if (isPromise(result)) {
- result.then(_exit, _exit);
+ if (worker.onAbort) {
+ let handler = worker.onAbort();
+ if(isPromise(handler)) {
+ handler.then(_abort);
+ } else {
+ _abort();
+ }
  } else {
- _exit();
+ if(!worker.terminationHandler) {
+ return _exit();
+ }
+
+ var result = worker.terminationHandler(code);
+ if (isPromise(result)) {
+ result.then(_exit, _exit);
+ } else {
+ _exit();
+ }
  }
 }
 
@@ -253,4 +270,7 @@ worker.emit = function (payload) {
 if (typeof exports !== 'undefined') {
  exports.add = worker.register;
  exports.emit = worker.emit;
+ exports.addAbortListener = function(listener) {
+ worker.onAbort = listener;
+ };
 }

test/Pool.test.js

@@ -1315,6 +1315,22 @@ describe('Pool', function () {
  });
  });
 
+ it('should not terminate worker if abort listener is defined', function () {
+ var pool = createPool(__dirname + '/workers/cleanup-abort.js');
+
+ pool.exec('asyncTimeout', [])
+ .timeout(200)
+ .catch((err) => {
+ assert(err instanceof Promise.TimeoutError);
+ // We should still have a busy worker since the worker should still be
+ // active and the promise chain for the timeout has yet to resolve
+ // which keeps the task active.
+ assert(pool.stats().busyWorkers === 1);
+ pool.terminate();
+ done();
+ });
+ });
+
  describe('validate', () => {
  it('should not allow unknown properties in forkOpts', function() {
  var pool = createPool({

test/workers/cleanup-abort.js

@@ -0,0 +1,21 @@
+var workerpool = require("../..");
+
+function asyncTimeout() {
+ return new Promise(function (resolve) {
+
+ let timeout = setTimeout(function () {
+ resolve();
+ }, 5000);
+ workerpool.addAbortListener(function () {
+ clearTimeout(timeout);
+ resolve();
+ });
+ });
+}
+
+// create a worker and register public functions
+workerpool.worker(
+ {
+ asyncTimeout: asyncTimeout,
+ }
+);