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,
+  }
+);