@lage-run/worker-threads-pool - npm Package Compare versions

Comparing version 0.1.4 to 0.1.5

CHANGELOG.json

@@ -5,3 +5,18 @@ {
     {
-      "date": "Sat, 01 Oct 2022 05:25:19 GMT",
+      "date": "Mon, 03 Oct 2022 20:41:09 GMT",
+      "tag": "@lage-run/worker-threads-pool_v0.1.5",
+      "version": "0.1.5",
+      "comments": {
+        "patch": [
+          {
+            "author": "[e-mail]",
+            "package": "@lage-run/worker-threads-pool",
+            "commit": "68c7fda6d0c1757b353d406b21223295c9830767",
+            "comment": "adds support for abort signal"
+          }
+        ]
+      }
+    },
+    {
+      "date": "Sat, 01 Oct 2022 05:25:29 GMT",
       "tag": "@lage-run/worker-threads-pool_v0.1.4",
@@ -8,0 +23,0 @@ "version": "0.1.4",

CHANGELOG.md

 # Change Log - @lage-run/worker-threads-pool
 
-This log was last generated on Sat, 01 Oct 2022 05:25:19 GMT and should not be manually modified.
+This log was last generated on Mon, 03 Oct 2022 20:41:09 GMT and should not be manually modified.
 
 <!-- Start content -->
 
+## 0.1.5
+
+Mon, 03 Oct 2022 20:41:09 GMT
+
+### Patches
+
+- adds support for abort signal ([e-mail])
+
 ## 0.1.4
 
-Sat, 01 Oct 2022 05:25:19 GMT
+Sat, 01 Oct 2022 05:25:29 GMT
 
@@ -11,0 +19,0 @@ ### Patches

lib/Pool.d.ts

 /// <reference types="node" />
 import type { Worker } from "worker_threads";
 import type { Readable } from "stream";
+import type { AbortSignal } from "abort-controller";
 export interface Pool {
-    exec(data: unknown, setup?: (worker?: Worker, stdout?: Readable, stderr?: Readable) => void, cleanup?: (args: any) => void): Promise<unknown>;
+    exec(data: unknown, setup?: (worker: Worker, stdout: Readable, stderr: Readable) => void, cleanup?: (args: any) => void, abortSignal?: AbortSignal): Promise<unknown>;
     close(): Promise<unknown>;
 }

lib/registerWorker.d.ts

@@ -1,1 +0,2 @@
-export declare function registerWorker(fn: (data: any) => Promise<void> | void): void;
+import type { AbortSignal } from "abort-controller";
+export declare function registerWorker(fn: (data: any, abortSignal?: AbortSignal) => Promise<void> | void): void;

lib/registerWorker.js

@@ -5,9 +5,20 @@ "use strict";
 const worker_threads_1 = require("worker_threads");
+const abort_controller_1 = require("abort-controller");
 const stdioStreamMarkers_1 = require("./stdioStreamMarkers");
 function registerWorker(fn) {
-    worker_threads_1.parentPort === null || worker_threads_1.parentPort === void 0 ? void 0 : worker_threads_1.parentPort.on("message", async (task) => {
+    worker_threads_1.parentPort === null || worker_threads_1.parentPort === void 0 ? void 0 : worker_threads_1.parentPort.on("message", async (message) => {
+        let abortController;
+        switch (message.type) {
+            case "start":
+                abortController = new abort_controller_1.AbortController();
+                return message.task && (await start(message.task, abortController.signal));
+            case "abort":
+                return abortController === null || abortController === void 0 ? void 0 : abortController.abort();
+        }
+    });
+    async function start(task, abortSignal) {
         try {
             process.stdout.write(`${stdioStreamMarkers_1.START_WORKER_STREAM_MARKER}\n`);
             process.stderr.write(`${stdioStreamMarkers_1.START_WORKER_STREAM_MARKER}\n`);
-            const results = await fn(task);
+            const results = await fn(task, abortSignal);
             worker_threads_1.parentPort === null || worker_threads_1.parentPort === void 0 ? void 0 : worker_threads_1.parentPort.postMessage({ err: undefined, results });
@@ -22,5 +33,5 @@ }
         }
-    });
+    }
 }
 exports.registerWorker = registerWorker;
 //# sourceMappingURL=registerWorker.js.map

lib/WorkerPool.d.ts

@@ -11,2 +11,3 @@ /**
 import type { WorkerOptions } from "worker_threads";
+import type { AbortSignal } from "abort-controller";
 interface WorkerPoolOptions {
@@ -33,6 +34,6 @@ maxWorkers?: number;
     addNewWorker(): void;
-    exec(task: unknown, setup?: (worker?: Worker, stdout?: Readable, stderr?: Readable) => void, cleanup?: (worker: Worker) => void): Promise<unknown>;
-    _exec(): void;
+    exec(task: unknown, setup?: (worker: Worker, stdout: Readable, stderr: Readable) => void, cleanup?: (worker: Worker) => void, abortSignal?: AbortSignal): Promise<unknown>;
+    _exec(abortSignal?: AbortSignal): void;
     close(): Promise<void>;
 }
 export {};

lib/WorkerPool.js

@@ -144,9 +144,9 @@ "use strict";
     }
-    exec(task, setup, cleanup) {
+    exec(task, setup, cleanup, abortSignal) {
         return new Promise((resolve, reject) => {
             this.queue.push({ task, resolve, reject, cleanup, setup });
-            this._exec();
+            this._exec(abortSignal);
         });
     }
-    _exec() {
+    _exec(abortSignal) {
         if (this.freeWorkers.length > 0) {
@@ -157,2 +157,5 @@ const worker = this.freeWorkers.pop();
             if (worker) {
+                abortSignal === null || abortSignal === void 0 ? void 0 : abortSignal.addEventListener("abort", () => {
+                    worker.postMessage({ type: "abort" });
+                });
                 worker[kTaskInfo] = new WorkerPoolTaskInfo({ cleanup, resolve, reject, worker, setup });
@@ -162,3 +165,3 @@ worker[kWorkerCapturedStreamPromise] = new Promise((onResolve) => {
                 });
-                worker.postMessage(task);
+                worker.postMessage({ type: "start", task });
             }
@@ -165,0 +168,0 @@ }

package.json

 {
   "name": "@lage-run/worker-threads-pool",
-  "version": "0.1.4",
+  "version": "0.1.5",
   "description": "A worker_threads pool implementation based on the official Node.js async_hooks documentation",
@@ -5,0 +5,0 @@ "repository": {

New alerts

License Policy Violation

License

This package is not allowed per your license policy. Review the package's license to ensure compliance.

Found 1 instance in 1 package

Fixed alerts

License Policy Violation

License

This package is not allowed per your license policy. Review the package's license to ensure compliance.

Found 1 instance in 1 package

Improved metrics

Total package byte prevSize

28170

increased by8.36%

Lines of code

459

increased by7.49%

No dependency changes